mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-12 14:19:00 +01:00
feat(observability): span the unnamed gaps in run preparation (#20396)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Before a task reaches the agent, the server prepares the run: it reads the task, the workspace and the configuration, and then hands the run to the native controller > - With an OTLP endpoint, each preparation step emits a `heartbeat.prepare.<phase>` or `native.controller.<phase>` span > - On a hosted test deployment, 0.3 to 0.5 s of each `task.prepare` sits between those spans, and about 0.5 s more sits between `execution.dispatch_admission` and `native.coordinator.claim` > - Nobody can say which step is slow in that time, so latency work in this area is guesswork > - This pull request adds spans around the existing awaited work in those gaps, with the existing timing helpers and naming conventions > - The benefit is that a trace of a slow preparation now names the step that used the time ## Linked Issues or Issue Description **What existing behavior does this improve?** The OpenTelemetry trace of task preparation. Today it has unnamed gaps between the named preparation spans. **Subsystem affected** Server: heartbeat run preparation and the native controller (`server/src/services/heartbeat.ts`, `server/src/services/native-runtime/native-session-executor.ts`). **Current behavior** On a hosted test deployment (8 runs per group), `task.prepare` takes 2.6 to 3.4 s. Between `heartbeat.prepare.runtime_state` and `heartbeat.prepare.run_identity`, about 0.25 s has no span. Between `execution.dispatch_admission` and `native.coordinator.claim`, about 0.5 s has no span, and in one pair of runs it was 1.4 s. Nobody can tell which step used that time. **Proposed behavior** Spans name the existing awaited reads and writes in those gaps. The change adds no reads and records no content. The operator endpoint gate and the no-operation behaviour stay. **Reason and benefit** Latency work in this area needs a measurement. With these spans, a trace of a slow preparation names the step that used the time, so a fix can be confirmed or rejected after deployment. **Breaking changes** None. The change adds span names to the closed list. It changes no API, schema or behaviour. **Additional context** The server uses postgres.js, which the auto-instrumentation does not cover, so database spans do not fill these gaps. ## What Changed - `server/src/services/execution-preparation-timing.ts`: add the new closed phase names. - `server/src/services/heartbeat.ts`: add `heartbeat.prepare.*` spans for `chat_retry_admission`, `workspace_selection`, `dependency_readiness`, `issue_checkout`, `instance_settings`, `workspace_context`, `project_context`, `plan_wake_routing`, `routine_environment`, `responsible_user` and `execution_start`. The re-reads of the issue after a checkout use the existing `issue_context` name. - `server/src/services/native-runtime/native-session-executor.ts`: add `native.controller.*` spans for `attachment_staging`, `warm_restore`, `span_persistence` (persistence of the preparation spans to the run log) and `state_migration` before the coordinator claim. - `doc/observability.md`: list the new spans. - `server/src/__tests__/heartbeat-preparation-gap-spans.test.ts`: run a real heartbeat through native preparation with an in-memory span exporter. Check that each new span is emitted and that the new preparation spans are children of `task.execution_attempt`. ## Verification - `pnpm --filter @paperclipai/server typecheck` - `npx vitest run src/__tests__/heartbeat-preparation-gap-spans.test.ts` (fails without the source change, passes with it) - `npx vitest run src/services/heartbeat/preparation-timing-context.test.ts src/services/execution-preparation-timing.test.ts src/services/native-runtime/native-session-executor.test.ts src/__tests__/heartbeat-context-compaction-execution.test.ts` ## Risks - Low risk. The change only wraps existing awaited calls. Values and errors pass through the timing helper unchanged, and a checkout conflict still reaches the same `catch`. - Phases of 50 ms or more inside `heartbeat.prepare.*` are also written to the run log, as for the existing phases. This adds a few run-log rows on slow runs. - Without an OTLP endpoint, the helpers keep their current no-operation behaviour. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, test runs) and extended thinking. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Nicky LeachandClaude Opus 5.5 authored and GitHub committed
1 parent
09241442ea
commit
d95fdd8d79
5 files changed
+234
-73
No files matched your search
@@ -0,0 +1,138 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { context, trace } from "@opentelemetry/api";
|
||||
import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks";
|
||||
import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, authUsers, companyMemberships, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import type { NativeRunIdentity, NativeSession, NativeSessionBackend, PrpEvent } from "@paperclipai/paperclip-runner";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { heartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { closeIdleWarmNativeSessionsForRestart } from "../services/native-runtime/native-session-executor.js";
|
||||
import { nativeRestartFixture, waitForCondition } from "./helpers/native-restart-fixture.js";
|
||||
import { getRecoveryPostgresSupport } from "./helpers/recovery-postgres-fixture.js";
|
||||
|
||||
const support = await getRecoveryPostgresSupport({});
|
||||
|
||||
function completingBackend(db: ReturnType<typeof createDb>): NonNullable<HeartbeatServiceOptions["nativeSessionBackendFactory"]> {
|
||||
return execution => {
|
||||
let identity: NativeRunIdentity;
|
||||
let sourceInstanceId = "";
|
||||
let turn = Promise.withResolvers<string>();
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
capabilities: async () => capabilities,
|
||||
attachRun: async input => { identity = input.identity; },
|
||||
startTurn: async () => {
|
||||
const turnId = randomUUID();
|
||||
turn.resolve(turnId);
|
||||
return { turnId };
|
||||
},
|
||||
async *events() {
|
||||
const turnId = await turn.promise;
|
||||
const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record<string, unknown>): PrpEvent => ({
|
||||
schema: "paperclip.prp.event.v1", sourceKind: "runner", sourceInstanceId,
|
||||
sourceEventId: `${sourceInstanceId}:${identity.runId}:${sourceSeq}`, sourceSeq,
|
||||
runId: identity.runId, normalizedSessionId: identity.sessionId, turnId,
|
||||
schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), eventType, payload,
|
||||
});
|
||||
yield event(1, "turn.started", {});
|
||||
yield event(2, "item.completed", { kind: "agentMessage", channel: "final", text: "Done." });
|
||||
turn = Promise.withResolvers<string>();
|
||||
yield event(3, "turn.completed", { status: "completed" });
|
||||
},
|
||||
result: async () => null,
|
||||
snapshot: async () => ({ backendKind: "mock", sessionId: "gap-spans-session", identity,
|
||||
providerSessionId: "gap-spans-provider-session", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }),
|
||||
close: async () => {},
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
descriptor: async () => {
|
||||
const [run] = await db.select({ runnerInstanceId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, execution.binding.runId));
|
||||
if (!run?.runnerInstanceId) throw new Error("Normal native admission did not persist a runner identity");
|
||||
sourceInstanceId = run.runnerInstanceId;
|
||||
return { kind: "mock", name: "preparation-gap-spans", version: "1", capabilities,
|
||||
runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } };
|
||||
},
|
||||
openSession: async input => { identity = input.identity; return session; },
|
||||
recoverSession: async snapshot => { identity = snapshot.identity; return { recovered: true, session }; },
|
||||
};
|
||||
return backend;
|
||||
};
|
||||
}
|
||||
|
||||
describe.skipIf(!support.supported)("heartbeat preparation gap spans", () => {
|
||||
const fixture = nativeRestartFixture({});
|
||||
|
||||
it("emits spans for the work between the named preparation phases", async () => {
|
||||
const { companyId, agentId, temporary } = fixture.current;
|
||||
const db = createDb(temporary.connectionString, { maxConnections: 4 });
|
||||
const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test");
|
||||
const manager = new AsyncLocalStorageContextManager().enable();
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] });
|
||||
expect(context.setGlobalContextManager(manager)).toBe(true);
|
||||
expect(trace.setGlobalTracerProvider(provider)).toBe(true);
|
||||
const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: completingBackend(db) });
|
||||
try {
|
||||
const userId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Task owner", email: `${userId}@example.invalid`,
|
||||
createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" });
|
||||
await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex", lifecycleMode: "warm", idleTimeoutMs: 60_000 } })
|
||||
.where(eq(agents.id, agentId));
|
||||
await instanceSettingsService(db).updateExperimental({ enableAgentChat: true, enableNativeRunner: true });
|
||||
const [issue] = await db.insert(issues).values({ companyId, title: "Trace the preparation gaps", status: "in_progress",
|
||||
assigneeAgentId: agentId, responsibleUserId: userId, conversationAgentId: agentId, conversationUserId: userId,
|
||||
conversationState: "active" }).returning();
|
||||
const [message] = await db.insert(issueComments).values({ companyId, issueId: issue!.id,
|
||||
authorType: "user", authorUserId: userId, body: "Start." }).returning();
|
||||
const run = await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
|
||||
requestedByActorType: "user", requestedByActorId: userId, manualUserWake: true,
|
||||
payload: { issueId: issue!.id, commentId: message!.id },
|
||||
contextSnapshot: { issueId: issue!.id, taskKey: issue!.id, wakeCommentId: message!.id } });
|
||||
expect(run).not.toBeNull();
|
||||
await waitForCondition("the run to settle", () => heartbeat.getTaskDrainStatus().quiescent);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id)))[0])
|
||||
.toMatchObject({ status: "succeeded" });
|
||||
|
||||
await provider.forceFlush();
|
||||
const spans = exporter.getFinishedSpans();
|
||||
const names = new Set(spans.map(span => span.name));
|
||||
const preparation = [
|
||||
"heartbeat.prepare.chat_retry_admission",
|
||||
"heartbeat.prepare.workspace_selection",
|
||||
"heartbeat.prepare.dependency_readiness",
|
||||
"heartbeat.prepare.instance_settings",
|
||||
"heartbeat.prepare.workspace_context",
|
||||
"heartbeat.prepare.plan_wake_routing",
|
||||
"heartbeat.prepare.routine_environment",
|
||||
"heartbeat.prepare.responsible_user",
|
||||
"heartbeat.prepare.execution_start",
|
||||
];
|
||||
const controller = [
|
||||
"native.controller.attachment_staging",
|
||||
"native.controller.warm_restore",
|
||||
"native.controller.span_persistence",
|
||||
"native.controller.state_migration",
|
||||
];
|
||||
for (const name of [...preparation, ...controller]) expect(names, name).toContain(name);
|
||||
const attempts = spans.filter(span => span.name === "task.execution_attempt");
|
||||
expect(attempts).toHaveLength(1);
|
||||
for (const span of spans.filter(span => preparation.includes(span.name))) {
|
||||
expect(span.parentSpanContext?.spanId, span.name).toBe(attempts[0]!.spanContext().spanId);
|
||||
}
|
||||
} finally {
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
await closeIdleWarmNativeSessionsForRestart();
|
||||
await db.$client.end();
|
||||
vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint);
|
||||
trace.disable();
|
||||
context.disable();
|
||||
await provider.shutdown();
|
||||
}
|
||||
}, 60_000);
|
||||
});
|
||||
@@ -7,6 +7,9 @@ import { isSpanContextValid, ROOT_CONTEXT, SpanStatusCode, trace, type Context }
|
||||
type PreparationPhase = "environment_selection" | "execution_configuration" | "session"
|
||||
| "agent_read" | "chat_completion" | "conversation" | "runtime_state" | "issue_context"
|
||||
| "run_identity" | "trust_preset" | "task_context"
|
||||
| "chat_retry_admission" | "workspace_selection" | "dependency_readiness" | "issue_checkout"
|
||||
| "instance_settings" | "workspace_context" | "project_context" | "plan_wake_routing"
|
||||
| "routine_environment" | "responsible_user" | "execution_start"
|
||||
| "workspace" | "github_environment" | "github_launchers" | "run_output"
|
||||
| "workspace_validation" | "runtime_services" | "instructions" | "runtime_selection"
|
||||
| "configuration_mentions" | "configuration_github_identity" | "configuration_secrets"
|
||||
@@ -34,7 +37,8 @@ type PreparationPhase = "environment_selection" | "execution_configuration" | "s
|
||||
| "github_access" | "wake_binding" | "backend_setup" | "connector_assignments"
|
||||
| "assigned_mcp" | "tool_authority" | "companion_selection" | "runner_binary"
|
||||
| "github_stop" | "coordinator_release" | "instruction_checkpoint"
|
||||
| "instruction_changes" | "warm_release" | "instruction_collection";
|
||||
| "instruction_changes" | "warm_release" | "instruction_collection"
|
||||
| "attachment_staging" | "warm_restore" | "span_persistence" | "state_migration";
|
||||
|
||||
export type ExecutionPhase =
|
||||
| "skills.inventory" | "skills.inventory.wait" | "skills.inventory.bundled" | "skills.inventory.releases"
|
||||
|
||||
@@ -2153,7 +2153,7 @@ export function heartbeatService(
|
||||
contextSnapshot: context,
|
||||
}));
|
||||
});
|
||||
const isFailedChatRunRetry = await authorizeFailedChatRetryExecution();
|
||||
const isFailedChatRunRetry = await timePreparation("chat_retry_admission", () => authorizeFailedChatRetryExecution());
|
||||
// Never adopt a chat-execution attestation supplied in a wake payload.
|
||||
// Reviewed chat turns rebuild it from the current durable owner below.
|
||||
delete context[PAPERCLIP_EXTERNAL_CHAT_EXECUTION_BOUND_KEY];
|
||||
@@ -2208,28 +2208,31 @@ export function heartbeatService(
|
||||
const taskKey = deriveTaskKeyWithHeartbeatFallback(context, null);
|
||||
const sessionCodec = getAdapterSessionCodec(agent.adapterType);
|
||||
const issueId = readNonEmptyString(context.issueId);
|
||||
const freshWorkspace = await readFreshWorkspaceReadiness(db, run);
|
||||
if (issueId && canApplyTaskWorkspaceSelectionAtAdmission({ admittedInput: parseObject(run.runnerProfileJson).nativeExecutionInput, restarting: Boolean(runOptions.nativeRestartRecovery), hasLeaseOwner: Boolean(runOptions.nativeLeaseOwner) })) {
|
||||
const changed = await executionWorkspacesSvc.applyPendingTaskWorkspaceSelection({ companyId: agent.companyId,
|
||||
issueId, runId: run.id, actor: { type: "agent", agentId: agent.id, companyId: agent.companyId, source: "agent_key" } });
|
||||
// A subtask an older release bound to its parent's workspace gets its own
|
||||
// here. A process-loss retry or a continuation that keeps the current
|
||||
// workspace still finishes the interrupted attempt there first.
|
||||
const released = run.processLossRetryCount > 0 || (freshWorkspace && freshWorkspace.mode !== "fresh_isolated_workspace") ? false
|
||||
: await executionWorkspacesSvc.releaseAncestorWorkspaceBindingAtAdmission({ companyId: agent.companyId, issueId, runId: run.id });
|
||||
if (changed || released) {
|
||||
delete context.resumeSessionParams;
|
||||
delete context.resumeSessionDisplayId;
|
||||
context.workspaceSelectionChanged = true;
|
||||
const freshWorkspace = await timePreparation("workspace_selection", async () => {
|
||||
const readiness = await readFreshWorkspaceReadiness(db, run);
|
||||
if (issueId && canApplyTaskWorkspaceSelectionAtAdmission({ admittedInput: parseObject(run.runnerProfileJson).nativeExecutionInput, restarting: Boolean(runOptions.nativeRestartRecovery), hasLeaseOwner: Boolean(runOptions.nativeLeaseOwner) })) {
|
||||
const changed = await executionWorkspacesSvc.applyPendingTaskWorkspaceSelection({ companyId: agent.companyId,
|
||||
issueId, runId: run.id, actor: { type: "agent", agentId: agent.id, companyId: agent.companyId, source: "agent_key" } });
|
||||
// A subtask an older release bound to its parent's workspace gets its own
|
||||
// here. A process-loss retry or a continuation that keeps the current
|
||||
// workspace still finishes the interrupted attempt there first.
|
||||
const released = run.processLossRetryCount > 0 || (readiness && readiness.mode !== "fresh_isolated_workspace") ? false
|
||||
: await executionWorkspacesSvc.releaseAncestorWorkspaceBindingAtAdmission({ companyId: agent.companyId, issueId, runId: run.id });
|
||||
if (changed || released) {
|
||||
delete context.resumeSessionParams;
|
||||
delete context.resumeSessionDisplayId;
|
||||
context.workspaceSelectionChanged = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
return readiness;
|
||||
});
|
||||
let issueContext = issueId
|
||||
? await timePreparation("issue_context", () => getIssueExecutionContext(agent.companyId, issueId))
|
||||
: null;
|
||||
const issueDependencyReadiness = issueId
|
||||
? await issuesSvc
|
||||
? await timePreparation("dependency_readiness", () => issuesSvc
|
||||
.listDependencyReadiness(agent.companyId, [issueId])
|
||||
.then((rows) => rows.get(issueId) ?? null)
|
||||
.then((rows) => rows.get(issueId) ?? null))
|
||||
: null;
|
||||
const dependencyResponse = issueId && issueContext?.status === "blocked"
|
||||
&& context.dependencyBlockedInteraction === true
|
||||
@@ -2247,14 +2250,17 @@ export function heartbeatService(
|
||||
// dispatch: an operator parking the issue after claim but before this
|
||||
// checkout must not be overwritten by the continuation. A todo task
|
||||
// that has never started has nothing parked, so it may start here.
|
||||
const startsUnstartedTask = issueContext.status === "todo" &&
|
||||
await isUnstartedTaskQuestionAnswer(db, { companyId: agent.companyId, issueId, interactionId: context.interactionId });
|
||||
await issuesSvc.checkout(
|
||||
issueId,
|
||||
agent.id,
|
||||
[...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : [])],
|
||||
run.id,
|
||||
);
|
||||
const issueStatus = issueContext.status;
|
||||
await timePreparation("issue_checkout", async () => {
|
||||
const startsUnstartedTask = issueStatus === "todo" &&
|
||||
await isUnstartedTaskQuestionAnswer(db, { companyId: agent.companyId, issueId, interactionId: context.interactionId });
|
||||
await issuesSvc.checkout(
|
||||
issueId,
|
||||
agent.id,
|
||||
[...resolvedInteractionCheckoutExpectedStatuses(), ...(startsUnstartedTask ? ["todo"] : [])],
|
||||
run.id,
|
||||
);
|
||||
});
|
||||
context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true;
|
||||
} catch (error) {
|
||||
if (!isCheckoutConflictError(error)) throw error;
|
||||
@@ -2269,7 +2275,7 @@ export function heartbeatService(
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
issueContext = await getIssueExecutionContext(agent.companyId, issueId);
|
||||
issueContext = await timePreparation("issue_context", () => getIssueExecutionContext(agent.companyId, issueId));
|
||||
}
|
||||
if (
|
||||
issueId &&
|
||||
@@ -2286,18 +2292,18 @@ export function heartbeatService(
|
||||
})
|
||||
) {
|
||||
try {
|
||||
await issuesSvc.checkout(
|
||||
await timePreparation("issue_checkout", () => issuesSvc.checkout(
|
||||
issueId,
|
||||
agent.id,
|
||||
typeof context.workspaceReconstructionActionId === "string" ? ["todo"] : ["todo", "backlog", "blocked"],
|
||||
run.id,
|
||||
);
|
||||
));
|
||||
context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true;
|
||||
} catch (error) {
|
||||
if (!isCheckoutConflictError(error)) throw error;
|
||||
context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = false;
|
||||
}
|
||||
issueContext = await getIssueExecutionContext(agent.companyId, issueId);
|
||||
issueContext = await timePreparation("issue_context", () => getIssueExecutionContext(agent.companyId, issueId));
|
||||
}
|
||||
if (
|
||||
issueId &&
|
||||
@@ -2338,7 +2344,7 @@ export function heartbeatService(
|
||||
)
|
||||
: null;
|
||||
const experimentalInstanceSettings =
|
||||
await instanceSettings.getExperimental();
|
||||
await timePreparation("instance_settings", () => instanceSettings.getExperimental());
|
||||
const isolatedWorkspacesEnabled =
|
||||
experimentalInstanceSettings.enableIsolatedWorkspaces;
|
||||
// Inert on its own: the operator default only reaches the resolver when
|
||||
@@ -2349,16 +2355,18 @@ export function heartbeatService(
|
||||
experimentalInstanceSettings.enableIsolatedWorkspacesByDefault;
|
||||
const providerApiPlacement = resolveProviderApiPlacement({ run, agent });
|
||||
const externalProvider = heartbeatExternalProvider({ run, agent }) ?? providerApiPlacement?.provider;
|
||||
const [taskWorkspaceIntentRow] = issueId && !externalProvider ? await db.select({ intent: issues.workspaceSelection }).from(issues)
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId))) : [];
|
||||
const [taskWorkspaceIntentRow] = issueId && !externalProvider ? await timePreparation("workspace_context", () => db.select({ intent: issues.workspaceSelection }).from(issues)
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, agent.companyId)))) : [];
|
||||
const contextProjectId = readNonEmptyString(context.projectId);
|
||||
const boundSourceWorkspace = !externalProvider && issueContext?.executionWorkspaceId
|
||||
? await executionWorkspacesSvc.getStoredById(agent.companyId, issueContext.executionWorkspaceId) : null;
|
||||
const [selectedWorkspaceSource] = !externalProvider && !boundSourceWorkspace && issueContext?.projectWorkspaceId
|
||||
? await db.select({ projectId: projectWorkspaces.projectId }).from(projectWorkspaces)
|
||||
.where(and(eq(projectWorkspaces.id, issueContext.projectWorkspaceId), eq(projectWorkspaces.companyId, agent.companyId))) : [];
|
||||
const executionWorkspaceId = issueContext?.executionWorkspaceId;
|
||||
const boundSourceWorkspace = !externalProvider && executionWorkspaceId
|
||||
? await timePreparation("workspace_context", () => executionWorkspacesSvc.getStoredById(agent.companyId, executionWorkspaceId)) : null;
|
||||
const projectWorkspaceId = issueContext?.projectWorkspaceId;
|
||||
const [selectedWorkspaceSource] = !externalProvider && !boundSourceWorkspace && projectWorkspaceId
|
||||
? await timePreparation("workspace_context", () => db.select({ projectId: projectWorkspaces.projectId }).from(projectWorkspaces)
|
||||
.where(and(eq(projectWorkspaces.id, projectWorkspaceId), eq(projectWorkspaces.companyId, agent.companyId)))) : [];
|
||||
const loadProjectContext = async (projectId: string | null | undefined) => projectId
|
||||
? await db
|
||||
? await timePreparation("project_context", () => db
|
||||
.select({
|
||||
id: projects.id,
|
||||
executionWorkspacePolicy: projects.executionWorkspacePolicy,
|
||||
@@ -2385,7 +2393,7 @@ export function heartbeatService(
|
||||
eq(projects.companyId, agent.companyId),
|
||||
),
|
||||
)
|
||||
.then((rows) => rows[0] ?? null)
|
||||
.then((rows) => rows[0] ?? null))
|
||||
: null;
|
||||
const taskProjectId = issueContext ? issueContext.projectId : contextProjectId;
|
||||
const taskProjectContext = await loadProjectContext(taskProjectId);
|
||||
@@ -2407,14 +2415,14 @@ export function heartbeatService(
|
||||
readNonEmptyString(context.interactionStatus) === "accepted")
|
||||
: false;
|
||||
const acceptedPlanWakeRoutingDecision = issueContext
|
||||
? await resolveAcceptedPlanWakeRoutingDecision({
|
||||
? await timePreparation("plan_wake_routing", () => resolveAcceptedPlanWakeRoutingDecision({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
agentId: agent.id,
|
||||
issueId,
|
||||
acceptedPlanContinuationWake,
|
||||
contextSnapshot: context,
|
||||
})
|
||||
}))
|
||||
: null;
|
||||
if (acceptedPlanWakeRoutingDecision) {
|
||||
context.forceFreshSession = true;
|
||||
@@ -2434,17 +2442,17 @@ export function heartbeatService(
|
||||
} else {
|
||||
delete context.acceptedPlanWakeRouting;
|
||||
}
|
||||
const routineEnvContext = await getRoutineEnvForExecutionIssue(
|
||||
const routineEnvContext = await timePreparation("routine_environment", () => getRoutineEnvForExecutionIssue(
|
||||
agent.companyId,
|
||||
issueContext,
|
||||
);
|
||||
));
|
||||
let responsibleUserId: string | null =
|
||||
await resolveResponsibleUserIdForRun({
|
||||
await timePreparation("responsible_user", () => resolveResponsibleUserIdForRun({
|
||||
run,
|
||||
contextSnapshot: context,
|
||||
issueContext,
|
||||
routineEnvContext,
|
||||
});
|
||||
}));
|
||||
const identityContext = await timePreparation("run_identity", () => initializeRunIdentity(db, {
|
||||
companyId: agent.companyId,
|
||||
runId: run.id,
|
||||
@@ -3231,7 +3239,7 @@ export function heartbeatService(
|
||||
} = { current: null };
|
||||
try {
|
||||
const startedAt = run.startedAt ?? new Date();
|
||||
const runningWithSession = await db
|
||||
const runningWithSession = await timePreparation("execution_start", () => db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
startedAt,
|
||||
@@ -3242,12 +3250,12 @@ export function heartbeatService(
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, run.id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
.then((rows) => rows[0] ?? null));
|
||||
if (runningWithSession) run = runningWithSession;
|
||||
|
||||
// Pause Durability: flip to "running" ONLY if the agent is still invokable.
|
||||
// Atomic conditional UPDATE is the sole gate (no read-then-write); 0 rows => abort.
|
||||
const runningAgent = await db
|
||||
const runningAgent = await timePreparation("execution_start", () => db
|
||||
.update(agents)
|
||||
.set({ status: "running", updatedAt: new Date() })
|
||||
.where(
|
||||
@@ -3258,7 +3266,7 @@ export function heartbeatService(
|
||||
),
|
||||
)
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
.then((rows) => rows[0] ?? null));
|
||||
|
||||
if (!runningAgent) {
|
||||
logger.warn(
|
||||
@@ -3300,12 +3308,12 @@ export function heartbeatService(
|
||||
});
|
||||
|
||||
const currentRun = run;
|
||||
await appendRunEvent(currentRun, {
|
||||
await timePreparation("execution_start", () => appendRunEvent(currentRun, {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "info",
|
||||
message: "run started",
|
||||
});
|
||||
}));
|
||||
|
||||
const { currentUserRedactionOptions, appendIdentityRedactedLog, onLog, onPersistedNativeLog, onAdapterMeta, onAdapterEvent } =
|
||||
await timePreparation("run_output", () => runOutput.begin(currentRun));
|
||||
|
||||
@@ -9533,9 +9533,10 @@ export async function executePaperclipNativeSession(input: {
|
||||
}
|
||||
return preparedInput;
|
||||
};
|
||||
const stageAttachments = () => createRunPreparationTiming(runId, "native.controller")("attachment_staging", prepareAttachments);
|
||||
return needsWarmRetirement
|
||||
? await executePaperclipNativeSessionWithinScope(input, prepareAttachments, false, ownDiagnosticLease)
|
||||
: await executePaperclipNativeSessionWithinScope(await prepareAttachments(), undefined, false, ownDiagnosticLease);
|
||||
? await executePaperclipNativeSessionWithinScope(input, stageAttachments, false, ownDiagnosticLease)
|
||||
: await executePaperclipNativeSessionWithinScope(await stageAttachments(), undefined, false, ownDiagnosticLease);
|
||||
} catch (error) {
|
||||
executionFailure = error;
|
||||
if (error instanceof NativeControllerRecoveryWaitingError) {
|
||||
@@ -9692,7 +9693,8 @@ async function executePaperclipNativeSessionAttempt(
|
||||
: null;
|
||||
const stagedRestart = input.restartRecovery?.kind === "reattach_remote_runner"
|
||||
? await verifyStagedAttachmentRestart(input) : null;
|
||||
if (input.useRunnerd) await restoreWarmRunnerdController(input);
|
||||
const timeController = createRunPreparationTiming(input.execution.binding.runId, "native.controller");
|
||||
if (input.useRunnerd) await timeController("warm_restore", () => restoreWarmRunnerdController(input));
|
||||
const identityRedactor = createAgentIdentityRedactor(input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY);
|
||||
const redactIdentityText = (text: string) => redactSensitiveText(
|
||||
identityRedactor.chunk("diagnostic", text) + identityRedactor.finish("diagnostic"),
|
||||
@@ -9718,7 +9720,6 @@ async function executePaperclipNativeSessionAttempt(
|
||||
"paperclip_runner_provider_unsupported: ACPX candidate requires exact host qualification authorization",
|
||||
);
|
||||
}
|
||||
const timeController = createRunPreparationTiming(input.execution.binding.runId, "native.controller");
|
||||
const preparationSpans = input.preparationSpans ?? [];
|
||||
const preparationStarts = nativeRunPreparationStarts(
|
||||
preparationSpans,
|
||||
@@ -9772,20 +9773,22 @@ async function executePaperclipNativeSessionAttempt(
|
||||
startedAtMs: environmentStartedAtMs,
|
||||
})
|
||||
: null;
|
||||
for (const span of preparationSpans) {
|
||||
const rootMilestone = isNativeRunRootHistoricalSpan(span.name);
|
||||
await trace.record({
|
||||
...span,
|
||||
parentName: rootMilestone
|
||||
? "task.provider_session"
|
||||
: environmentSpans.includes(span)
|
||||
? "environment.startup"
|
||||
: "task.prepare",
|
||||
});
|
||||
}
|
||||
if (environmentScope) {
|
||||
await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs });
|
||||
}
|
||||
await timeController("span_persistence", async () => {
|
||||
for (const span of preparationSpans) {
|
||||
const rootMilestone = isNativeRunRootHistoricalSpan(span.name);
|
||||
await trace.record({
|
||||
...span,
|
||||
parentName: rootMilestone
|
||||
? "task.provider_session"
|
||||
: environmentSpans.includes(span)
|
||||
? "environment.startup"
|
||||
: "task.prepare",
|
||||
});
|
||||
}
|
||||
if (environmentScope) {
|
||||
await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs });
|
||||
}
|
||||
});
|
||||
let retainedTransition: VerifiedWarmTransitionBinding | undefined;
|
||||
let remoteWarmRecovery: RemoteWarmSessionRecovery | undefined;
|
||||
const cachedReleasedOwner = warmNativeSessions.get(nativeSessionScopeKey(input.execution));
|
||||
@@ -9870,7 +9873,7 @@ async function executePaperclipNativeSessionAttempt(
|
||||
}
|
||||
}
|
||||
if (input.useRunnerd) {
|
||||
retainedTransition = await migrateRunnerdStateRootForExecution({
|
||||
retainedTransition = await timeController("state_migration", () => migrateRunnerdStateRootForExecution({
|
||||
db: input.db,
|
||||
execution: input.execution,
|
||||
preparedExecutionAdmission: input.preparedExecutionAdmission,
|
||||
@@ -9881,7 +9884,7 @@ async function executePaperclipNativeSessionAttempt(
|
||||
restartRecovery: input.restartRecovery,
|
||||
runnerExecutionTarget: input.runnerExecutionTarget,
|
||||
bootstrapRetryLeaseOwner: input.leaseOwner,
|
||||
});
|
||||
}));
|
||||
}
|
||||
const durableRunnerBinding = input.useRunnerd
|
||||
? loadRunnerdDurableBinding(input.execution, retainedTransition)
|
||||
|
||||
Reference in new issue
Block a user