From f53db0c70fe13db4471f253a7ddfee138cb6e393 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 30 Oct 2026 15:21:05 -0500 Subject: [PATCH] perf(server): overlap OpenCode setup and trace reply recovery (#18367) Overlap eligible OpenCode GitHub and assigned MCP preparation. Trace deferred reply discovery and recovery, and forward the native GitHub transport recorder. Co-Authored-By: Paperclip --- doc/observability.md | 14 ++++- .../execution-preparation-timing.test.ts | 18 ++++--- .../services/execution-preparation-timing.ts | 8 +++ .../native-github-access.test.ts | 13 +++-- .../native-runtime/native-github-access.ts | 2 + .../native-response-presentation-work.ts | 19 +++++-- .../native-runtime/native-run-finalizer.ts | 16 +++++- .../native-session-executor.test.ts | 13 +++-- .../native-runtime/native-session-executor.ts | 3 +- .../presentation-ownership-read.test.ts | 54 ++++++++++++++++++- ui/src/lib/honeycomb-run-link.test.ts | 2 +- ui/src/lib/honeycomb-run-link.ts | 2 +- 12 files changed, 138 insertions(+), 26 deletions(-) diff --git a/doc/observability.md b/doc/observability.md index 5065e511ecc..e2bb4065caf 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -254,6 +254,17 @@ It does not extend an earlier execution attempt or include time waiting for the background worker to run. Retries have separate roots. A check that finds no eligible work can end without a restore child. +`task.response_presentation` owns recovery of an already committed native reply. +It starts a separate linked root because deferred presentation can run after the +execution attempt and workspace finalization have ended. Worker discovery for +a specific run and worker checks include candidate reads under +`response.resolve`. Each discovery or check owns its root; direct recovery and +its worker check share one root when nested. `response.transaction` +includes the recovery transaction; `response.transaction.entry` measures entry +into its callback, including connection acquisition. These spans retain only the +same hashed run ID and never capture reply content. A successful task status is +separate from the time the reply is saved. + These spans exist before provider dispatch, including setup failures. Preparation phases have only the existing `paperclip.task.run.run_id` attribute: the first 12 hexadecimal characters of the run UUID's SHA-256 hash. Failed operations set error status @@ -641,7 +652,8 @@ Without an OTLP endpoint, tracing remains silent and does not export data. The run link in Honeycomb matches `task.provider_session` for native runs, `task.run` for legacy adapters, `task.execution_attempt` for server preparation and settlement, and -`task.workspace_finalization` for deferred native copyback. It retains the exact hashed run-ID and service filters and groups by trace ID. The URL builder uses Honeycomb’s supported `in` operator for the four root names; see the [query specification](https://github.com/honeycombio/honeycomb-mcp#query-specification-for-run_query). This changes only the query link. The operator endpoint gate and span attribute allowlist remain unchanged. +`task.workspace_finalization` for deferred native copyback and `task.response_presentation` +for committed native reply recovery. It retains the exact hashed run-ID and service filters and groups by trace ID. The URL builder uses Honeycomb’s supported `in` operator for the five root names; see the [query specification](https://github.com/honeycombio/honeycomb-mcp#query-specification-for-run_query). This changes only the query link. The operator endpoint gate and span attribute allowlist remain unchanged. The canonical lifecycle is: diff --git a/server/src/services/execution-preparation-timing.test.ts b/server/src/services/execution-preparation-timing.test.ts index c267d1ee68d..7b1703ae7a9 100644 --- a/server/src/services/execution-preparation-timing.test.ts +++ b/server/src/services/execution-preparation-timing.test.ts @@ -1,6 +1,6 @@ import { getActiveStepContext, runWithRuntimeParent } from "@paperclipai/adapter-utils/acpx-engine/startup-timing"; import { afterEach, beforeEach, expect, it, vi } from "vitest"; -import { createRunPreparationTiming, measureRunPhase, recordRepositoryObjectCacheDetail, withRunExecutionTiming, withRunWorkspaceFinalizationTiming } from "./execution-preparation-timing.js"; +import { createRunPreparationTiming, measureRunPhase, recordRepositoryObjectCacheDetail, withRunExecutionTiming, withRunWorkspaceFinalizationTiming, withRunResponsePresentationTiming } from "./execution-preparation-timing.js"; import { ROOT_CONTEXT, trace, SpanStatusCode, type Context } from "@opentelemetry/api"; import { createHash } from "node:crypto"; @@ -45,6 +45,7 @@ it.each([undefined, "", " "])("does not consult tracing for endpoint %s", async try { await expect(createRunPreparationTiming("run", "native.controller")("backend_setup", () => 42)).resolves.toBe(42); await expect(withRunWorkspaceFinalizationTiming("run", () => 42)).resolves.toBe(42); + await expect(withRunResponsePresentationTiming("run", () => 42)).resolves.toBe(42); for (const phase of ["execution.usage_startup", "execution.mcp_handoff", "execution.dispatch_admission", "execution.budget_reservation", "skills.inventory", "skills.inventory.wait", "skills.cache.acquire", "skills.version.materialize", "skills.consume", "workspace_sync.seed_read_copy", "workspace_sync.seed_read_hash", "workspace_sync.seed_read_touch"] as const) { @@ -69,7 +70,10 @@ it.each([undefined, "", " "])("does not consult tracing for endpoint %s", async } }); -it("parents concurrent deferred copyback phases to their own finalization roots and preserves failure", async () => { +it.each([ + [withRunWorkspaceFinalizationTiming, "task.workspace_finalization", "workspace_sync.restore"], + [withRunResponsePresentationTiming, "task.response_presentation", "response.transaction"], +] as const)("parents concurrent deferred phases to their own %s root and preserves failure", async (withTiming, rootName, phase) => { const spans: Array<{ name: string; attributes: Record; context: Context; span: { end: ReturnType; setStatus: ReturnType } }> = []; handles.startActiveSpan.mockImplementation((name, options, context, callback) => { @@ -77,16 +81,16 @@ it("parents concurrent deferred copyback phases to their own finalization roots spans.push({ name, attributes: options.attributes, context, span }); return callback(span); }); - await Promise.all(["first", "second"].map(runId => withRunWorkspaceFinalizationTiming(runId, async () => { + await Promise.all(["first", "second"].map(runId => withTiming(runId, async () => { await Promise.resolve(); - await measureRunPhase("workspace_sync.restore", async () => {}); + await measureRunPhase(phase, async () => {}); }))); for (const runId of ["first", "second"]) { const hash = createHash("sha256").update(runId).digest("hex").slice(0, 12); const owned = spans.filter(span => span.attributes["paperclip.task.run.run_id"] === hash); expect(owned).toHaveLength(2); - const root = owned.find(span => span.name === "task.workspace_finalization")!; - const child = owned.find(span => span.name === "workspace_sync.restore")!; + const root = owned.find(span => span.name === rootName)!; + const child = owned.find(span => span.name === phase)!; expect(root.context).toBe(ROOT_CONTEXT); expect(trace.getSpan(child.context)).toBe(root.span); expect(root.span.end).toHaveBeenCalledOnce(); @@ -94,7 +98,7 @@ it("parents concurrent deferred copyback phases to their own finalization roots } const original = new Error("private finalization failure"); const work = vi.fn(() => { throw original; }); - await expect(withRunWorkspaceFinalizationTiming("failed", work)).rejects.toBe(original); + await expect(withTiming("failed", work)).rejects.toBe(original); expect(work).toHaveBeenCalledOnce(); expect(spans.at(-1)!.span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR }); expect(spans.at(-1)!.span.end).toHaveBeenCalledOnce(); diff --git a/server/src/services/execution-preparation-timing.ts b/server/src/services/execution-preparation-timing.ts index 9041b76f8f3..9ee7f8a189e 100644 --- a/server/src/services/execution-preparation-timing.ts +++ b/server/src/services/execution-preparation-timing.ts @@ -199,6 +199,14 @@ export function withRunWorkspaceFinalizationTiming(runId: string, work: () => return measure(runId, "task.workspace_finalization", work, true); } +export function withRunResponsePresentationTiming(runId: string, work: () => T | Promise): Promise { + const owned = ownership.getStore(); + if (owned?.runId === runId && owned.phase === "task.response_presentation" && !owned.ended) { + return (async () => work())(); + } + return measure(runId, "task.response_presentation", work, true); +} + export async function measureRunPhase(phase: ExecutionPhase, work: () => T | Promise, counts?: { fileCount?: number; bytes?: number }): Promise { const owned = ownership.getStore(); if (!endpointEnabled() || !owned || owned.ended) return work(); diff --git a/server/src/services/native-runtime/native-github-access.test.ts b/server/src/services/native-runtime/native-github-access.test.ts index 444715a1b14..1f8b07172b1 100644 --- a/server/src/services/native-runtime/native-github-access.test.ts +++ b/server/src/services/native-runtime/native-github-access.test.ts @@ -3,6 +3,7 @@ import { spawn, execFile } from "node:child_process"; import { mkdtemp, mkdir, readFile, rm, writeFile, access } from "node:fs/promises"; import { promisify } from "node:util"; import type { CommandManagedDuplexChannel, CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import type { DuplexObservabilityRecorder } from "@paperclipai/adapter-utils/duplex-observability"; import { startAdapterExecutionTargetPaperclipBridge } from "@paperclipai/adapter-utils/execution-target"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -39,7 +40,7 @@ function localDuplexChannel(command: readonly string[], cwd: string): CommandMan async function broker(resolveCredentials = vi.fn(async (binding: ReturnType) => ({ status: "available", env: { GH_TOKEN: `fixture-${binding.runId}` }, })), remote = false, bridgeUnavailable = false, onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise, -duplex?: { enabled?: boolean; capability: boolean; unavailable?: boolean }) { +duplex?: { enabled?: boolean; capability: boolean; unavailable?: boolean; recorder?: DuplexObservabilityRecorder }) { const root = await mkdtemp(path.join(tmpdir(), "native-github-reuse-")); cleanups.push(() => rm(root, { recursive: true, force: true })); const bin = path.join(root, "real-bin"); @@ -61,7 +62,7 @@ duplex?: { enabled?: boolean; capability: boolean; unavailable?: boolean }) { }); const target = remote ? { kind: "remote" as const, transport: "sandbox" as const, providerKey: "test", remoteCwd: root, runner: { execute, ...(duplex ? { openDuplexChannel } : {}) }, streamRunLogs: false, - ...(duplex ? { enableSandboxDuplexBridge: duplex.enabled, effectiveCapabilities: { + ...(duplex ? { enableSandboxDuplexBridge: duplex.enabled, duplexObservabilityRecorder: duplex.recorder, effectiveCapabilities: { reusableLeases: true, nativeSyncIn: false, nativeSyncOut: false, persistentProcessSessions: false, independentControlCommands: true, incrementalSessionOutput: false, concurrentSyncOperations: false, duplexCommandStream: duplex.capability, runnerWebSocketIngress: false, @@ -122,9 +123,13 @@ it.each([false, true])("resolves managed GitHub credentials from OpenCode shell expect(childEnvironment.githubTokenPresent).toBe(false); }, 30000); it("uses the selected duplex transport across credential rotation without file polling", async () => { - const b = await broker(undefined, true, false, undefined, { enabled: true, capability: true }); + const recorder = { recordSpan: vi.fn(), incrementCounter: vi.fn(), emitEvent: vi.fn() }; + const b = await broker(undefined, true, false, undefined, { enabled: true, capability: true, recorder }); expect(b.bridgeModes).toEqual(["http2_v1"]); expect(b.openDuplexChannel).toHaveBeenCalledOnce(); + expect(recorder.recordSpan).toHaveBeenCalledWith(expect.objectContaining({ name: "sandbox.duplex.channel_open", dimensions: { + provider: "other", transport: "http2", outcome: "ok", + } })); const commandsAfterStartup = b.execute.mock.calls.length; expect((await request(b)).status).toBe(403); const releaseA = b.activate(run("a")); @@ -135,6 +140,8 @@ it("uses the selected duplex transport across credential rotation without file p releaseA(); expect(await (await request(b)).json()).toMatchObject({ env: { GH_TOKEN: "fixture-b" } }); expect(b.resolveCredentials.mock.calls.map(([binding]) => binding.runId)).toEqual(["a", "b"]); + expect(recorder.recordSpan).toHaveBeenCalledWith(expect.objectContaining({ name: "sandbox.duplex.request" })); + expect(recorder.emitEvent).not.toHaveBeenCalledWith(expect.objectContaining({ dimensions: expect.objectContaining({ transport: "file" }) })); expect(b.execute.mock.calls).toHaveLength(commandsAfterStartup); releaseB(); await b.stop(); diff --git a/server/src/services/native-runtime/native-github-access.ts b/server/src/services/native-runtime/native-github-access.ts index 87233394538..870dbbbbac9 100644 --- a/server/src/services/native-runtime/native-github-access.ts +++ b/server/src/services/native-runtime/native-github-access.ts @@ -3,6 +3,7 @@ import { createServer } from "node:http"; import { cleanupGitHubOperationLaunchers, adapterExecutionTargetEnablesSandboxDuplexBridge, + adapterExecutionTargetDuplexObservabilityRecorder, githubOperationLauncherDirectory, prepareGitHubOperationLaunchers, startAdapterExecutionTargetPaperclipBridge, @@ -107,6 +108,7 @@ export async function createNativeGitHubAccess(input: { hostApiToken: token, hostApiUrl: url, enableSandboxDuplexBridge: duplexRequested, + duplexObservabilityRecorder: adapterExecutionTargetDuplexObservabilityRecorder(input.target), onLog: input.onLog, }); if (input.target?.kind === "remote" && !bridge) { diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index aac11b54e3f..cad63e0f53c 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -7,6 +7,7 @@ import type { NativeResponsePresentationScope } from "../native-response-present import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wakeup.js"; import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; +import { measureRunPhase, withRunResponsePresentationTiming } from "../execution-preparation-timing.js"; export function registerNativeResponsePresentationWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { @@ -20,27 +21,35 @@ export function nativeResponsePresentationWork(db: Db) { const query = nativeResponsePresentationCandidates(scope, after, runId).as("pending_native_response"); return db.select().from(query).orderBy(asc(query.id)); }; - async function load(scope?: NativeResponsePresentationScope, signal?: AbortSignal, canRun = () => true): Promise { + function load(scope?: NativeResponsePresentationScope, signal?: AbortSignal, canRun = () => true): Promise { + const work = () => loadWithinTrace(scope, signal, canRun); + return scope?.runId ? withRunResponsePresentationTiming(scope.runId, work) : work(); + } + async function loadWithinTrace(scope: NativeResponsePresentationScope | undefined, signal: AbortSignal | undefined, + canRun: () => boolean): Promise { const entries: Entry[] = []; let after: string | undefined; while (canRun() && !signal?.aborted) { - const rows = await candidates(scope, after); + const rows = await measureRunPhase("response.resolve", () => candidates(scope, after)); for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); if (rows.length < 100) return entries; after = rows[rows.length - 1].id; } throw new Error("Native response presentation discovery stopped"); } - async function check(entry: Entry, signal?: AbortSignal): Promise { + function check(entry: Entry, signal?: AbortSignal): Promise { + return withRunResponsePresentationTiming(entry.id, () => checkWithinTrace(entry, signal)); + } + async function checkWithinTrace(entry: Entry, signal?: AbortSignal): Promise { if (signal?.aborted) throw new Error("Native response presentation recovery stopped"); const scope = { companyId: entry.companyId, issueId: entry.issueId }; - const [current] = await candidates(scope, undefined, entry.id); + const [current] = await measureRunPhase("response.resolve", () => candidates(scope, undefined, entry.id)); if (!current) return null; if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() }; if (signal?.aborted) throw new Error("Native response presentation recovery stopped"); const recoveryStartedAt = Date.now(); await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true }); - const [remaining] = await candidates(scope, undefined, entry.id); + const [remaining] = await measureRunPhase("response.resolve", () => candidates(scope, undefined, entry.id)); return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null; } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index ba2017eb9a8..202d2b4eac3 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -67,6 +67,7 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueService } from "../issues.js"; import { publishChatPublicationCommitSignal } from "../chat-publication-reconciliation.js"; import { nativeSha256 } from "./canonical.js"; +import { beginRunPhase, measureRunPhase, withRunResponsePresentationTiming } from "../execution-preparation-timing.js"; import { readNativeBoardResponseWaitOrigin, readNativeBoardResponseWaitSource, @@ -736,11 +737,20 @@ export async function repairCommittedNativeChatResponse( export async function recoverCommittedNativeChatResponse( db: Db, input: { companyId: string; issueId: string; runId: string; onlyWhenDue?: boolean }, +): Promise { + return withRunResponsePresentationTiming(input.runId, () => recoverCommittedNativeChatResponseInScope(db, input)); +} + +async function recoverCommittedNativeChatResponseInScope( + db: Db, + input: { companyId: string; issueId: string; runId: string; onlyWhenDue?: boolean }, ): Promise { let agentId: string | null = null; let materialized = false; + const entered = beginRunPhase("response.transaction.entry"); try { - materialized = await db.transaction(async (transaction) => { + materialized = await measureRunPhase("response.transaction", () => db.transaction(async (transaction) => { + entered(); const tx = transaction as unknown as Db; @@ -1022,10 +1032,12 @@ export async function recoverCommittedNativeChatResponse( await projectCommittedRun({ db: tx, run, coordinator }); agentId = run.agentId; return true; - }); + })); } catch (error) { if (error instanceof CommittedChatResponseAuthorizationError) return false; throw error; + } finally { + entered(); } if (materialized && agentId) publishChatPublicationCommitSignal({ diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index e65220098a7..30d94805b98 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -12376,7 +12376,14 @@ describe("runnerd provider runtime wiring", () => { } }); - it.each(["github", "assigned"] as const)("cleans run-owned access after overlapping executor failure: %s", async failure => { + it.each((["codex", "opencode"] as const).flatMap(provider => + (["github", "assigned"] as const).map(failure => ({ provider, failure })), + ))("cleans run-owned access after overlapping $provider executor failure: $failure", async ({ provider, failure }) => { + const current = { + ...execution, + provider: provider === "codex" ? execution.provider : { kind: "opencode", model: null }, + session: { ...execution.session, driverKind: provider === "codex" ? "codex_app_server" : "opencode_server" }, + } as NativeExecutionInputV1; const issueService = vi.spyOn(issueServiceModule, "issueService").mockReturnValue({ update: vi.fn(async () => ({ status: "blocked", statusVersion: 1 })), } as unknown as ReturnType); @@ -12402,7 +12409,7 @@ describe("runnerd provider runtime wiring", () => { state.createBackend.mockClear(); let settled = false; const running = executePaperclipNativeSession({ - db: leaseDb(execution), execution, runnerInstanceId: "runner-overlap-cleanup", + db: leaseDb(current), execution: current, runnerInstanceId: "runner-overlap-cleanup", useRunnerd: true, managedGitHub: true, runnerEnvironment: { PAPERCLIP_NATIVE_MCP_NAME: "paperclip-assigned", @@ -12428,7 +12435,7 @@ describe("runnerd provider runtime wiring", () => { expect(state.execute).not.toHaveBeenCalled(); expect(state.createBackend).not.toHaveBeenCalled(); if (failure === "assigned") { - expect(activate).toHaveBeenCalledWith(execution.binding); + expect(activate).toHaveBeenCalledWith(current.binding); expect(deactivate).toHaveBeenCalledOnce(); expect(stop).toHaveBeenCalledOnce(); } diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6778fd46f2c..66b77f875a8 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9186,7 +9186,8 @@ async function executePaperclipNativeSessionWithinScope( return input.runnerEnvironment; }; const overlapGitHubPreparation = needsCoreGitHubAccess(input) && input.useRunnerd && input.backend === undefined - && input.runnerExecutionTarget?.kind === "remote" && input.execution.provider.kind === "codex" + && input.runnerExecutionTarget?.kind === "remote" + && (input.execution.provider.kind === "codex" || input.execution.provider.kind === "opencode") && input.execution.schema !== "paperclip.native-execution-input.v6" && input.execution.schema !== "paperclip.native-execution-input.v8"; if (!overlapGitHubPreparation) await prepareRunnerEnvironment(); diff --git a/server/src/services/presentation-ownership-read.test.ts b/server/src/services/presentation-ownership-read.test.ts index 427eb1edb86..c08037e78af 100644 --- a/server/src/services/presentation-ownership-read.test.ts +++ b/server/src/services/presentation-ownership-read.test.ts @@ -1,12 +1,15 @@ -import { randomBytes, randomUUID } from "node:crypto"; +import { createHash, randomBytes, randomUUID } from "node:crypto"; import { agents, companies, completionContracts, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; import { eq, type SQL } from "drizzle-orm"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { trace } from "@opentelemetry/api"; +import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { NativeRunCoordinatorStore } from "./native-runtime/native-run-coordinator-store.js"; import { getConversationOwnershipBlocker } from "./conversation-continuation.js"; import { isExternalChatPresentationContext } from "./heartbeat-run-summary.js"; import { finalizeNativeRun, recoverCommittedNativeChatResponse, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js"; +import { nativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; @@ -210,6 +213,53 @@ describe.skipIf(!support.supported)("presentation and ownership PostgreSQL read expect(outcomes[1]).toEqual(outcomes[0]); }); + it.each(["direct", "worker", "discovery and worker"])("traces %s reply recovery independently and leaves reply content out of spans", async lane => { + const f = await unpublishedAcceptedReply(); + const exporter = new InMemorySpanExporter(); + const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] }); + trace.setGlobalTracerProvider(provider); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector.test"); + try { + if (lane !== "direct") { + const worker = nativeResponsePresentationWork(db); + if (lane === "discovery and worker") { + expect(await worker.load({ companyId: f.companyId, issueId: f.issueId, runId: f.run.id })) + .toEqual([expect.objectContaining({ id: f.run.id })]); + } + expect(await worker.check({ id: f.run.id, companyId: f.companyId, + issueId: f.issueId, agentId: f.agentId, dueAt: 0 })).toBeNull(); + } else { + expect(await repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: f.issueId, runId: f.run.id })).toBe(true); + } + const spans = exporter.getFinishedSpans(); + const roots = spans.filter(span => span.name === "task.response_presentation"); + expect(roots).toHaveLength(lane === "discovery and worker" ? 2 : 1); + for (const root of roots) { + expect(root.parentSpanContext).toBeUndefined(); + expect(root.attributes).toEqual({ "paperclip.task.run.run_id": createHash("sha256").update(f.run.id).digest("hex").slice(0, 12) }); + } + const root = roots.at(-1)!; + for (const name of ["response.transaction", "response.transaction.entry"]) { + const child = spans.find(span => span.name === name)!; + expect(child).toBeDefined(); + expect(child.parentSpanContext?.spanId).toBe(root.spanContext().spanId); + } + if (lane !== "direct") expect(spans.filter(span => span.name === "response.resolve")) + .toHaveLength(lane === "discovery and worker" ? 3 : 2); + expect(JSON.stringify(spans.map(span => span.attributes))).not.toContain(f.reply); + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.run.id)); + expect(comments).toHaveLength(1); + expect(comments[0].body).toBe(f.reply); + vi.stubEnv("OTEL_EXPORTER_OTLP_ENDPOINT", ""); + expect(await repairCommittedNativeChatResponse(db, { companyId: f.companyId, issueId: f.issueId, runId: f.run.id })).toBe(false); + expect(exporter.getFinishedSpans()).toHaveLength(spans.length); + } finally { + vi.unstubAllEnvs(); + await provider.shutdown(); + trace.disable(); + } + }); + it.each([null, [], "scalar", 42, { source: 42, skipIssueComment: "true", paperclipWake: [] }, { source: "issue.comment.created", externalChatContinuation: true }, { source: "chat:fixture", skipIssueComment: true, paperclipWake: { externalInteractionContinuation: "true" } }, diff --git a/ui/src/lib/honeycomb-run-link.test.ts b/ui/src/lib/honeycomb-run-link.test.ts index 15dbcc0ecf6..a9531346d39 100644 --- a/ui/src/lib/honeycomb-run-link.test.ts +++ b/ui/src/lib/honeycomb-run-link.test.ts @@ -31,7 +31,7 @@ describe("Honeycomb run links", () => { expect(query.filter_combination).toBe("AND"); expect(query.filters).toEqual([ { column: "service.name", op: "=", value: "paperclip" }, - { column: "name", op: "in", value: ["task.run", "task.provider_session", "task.execution_attempt", "task.workspace_finalization"] }, + { column: "name", op: "in", value: ["task.run", "task.provider_session", "task.execution_attempt", "task.workspace_finalization", "task.response_presentation"] }, { column: HONEYCOMB_RUN_HASH_ATTRIBUTE, op: "=", diff --git a/ui/src/lib/honeycomb-run-link.ts b/ui/src/lib/honeycomb-run-link.ts index dc6a9c96c31..54c9f5b41bc 100644 --- a/ui/src/lib/honeycomb-run-link.ts +++ b/ui/src/lib/honeycomb-run-link.ts @@ -26,7 +26,7 @@ export function buildHoneycombRunQueryUrl(runIdHash: string): string { breakdowns: ["trace.trace_id"], filters: [ { column: "service.name", op: "=", value: "paperclip" }, - { column: "name", op: "in", value: ["task.run", "task.provider_session", "task.execution_attempt", "task.workspace_finalization"] }, + { column: "name", op: "in", value: ["task.run", "task.provider_session", "task.execution_attempt", "task.workspace_finalization", "task.response_presentation"] }, { column: HONEYCOMB_RUN_HASH_ATTRIBUTE, op: "=", value: runIdHash }, ], filter_combination: "AND",