diff --git a/server/src/__tests__/adapter-registry.test.ts b/server/src/__tests__/adapter-registry.test.ts index b7cfa075d3..0cfbb0e66c 100644 --- a/server/src/__tests__/adapter-registry.test.ts +++ b/server/src/__tests__/adapter-registry.test.ts @@ -233,6 +233,8 @@ describe("server adapter registry", () => { it("rejects an unsupported persisted runner provider before probing Codex", async () => { const adapter = requireServerAdapter("paperclip_runner"); + expect(adapter.supportsInstructionsBundle).toBe(true); + expect(adapter.instructionsPathKey).toBe("instructionsFilePath"); const result = await adapter.testEnvironment({ companyId: "company-1", adapterType: "paperclip_runner", diff --git a/server/src/__tests__/adapter-routes.test.ts b/server/src/__tests__/adapter-routes.test.ts index 649f030e89..d71fc2d493 100644 --- a/server/src/__tests__/adapter-routes.test.ts +++ b/server/src/__tests__/adapter-routes.test.ts @@ -163,7 +163,7 @@ describe("adapter routes", () => { .toMatchObject({ disabled: false, capabilities: { - supportsInstructionsBundle: false, + supportsInstructionsBundle: true, }, }); }); diff --git a/server/src/__tests__/agent-adapter-validation-routes.test.ts b/server/src/__tests__/agent-adapter-validation-routes.test.ts index 4c0f8f64c7..737ad8c24b 100644 --- a/server/src/__tests__/agent-adapter-validation-routes.test.ts +++ b/server/src/__tests__/agent-adapter-validation-routes.test.ts @@ -596,6 +596,11 @@ describe("agent routes adapter validation", () => { expect(res.status, JSON.stringify(res.body)).toBe(201); expect(mockAgentService.create).toHaveBeenCalledOnce(); + expect(mockAgentInstructionsService.materializeManagedBundle).toHaveBeenCalledWith( + expect.objectContaining({ adapterType: "paperclip_runner" }), + expect.any(Object), + expect.objectContaining({ entryFile: "AGENTS.md", replaceExisting: false }), + ); }); it("rejects non-Codex providers on fresh paperclip_runner agents and hires", async () => { diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 7fc7555627..5031361c6e 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -412,6 +412,170 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { expect(discard.status).toBe(403); }); + it("leaves the selected row queued when no native steering session is attached", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(initial.body.steeringDisposition).toBe("temporarily_unavailable"); + + const steered = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision }); + + expect(steered.status).toBe(409); + expect(steered.body.details).toMatchObject({ + code: "steering_temporarily_unavailable", + retryable: true, + }); + const queueAfterFailure = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id)).toEqual(seeded.commentIds); + }); + + it("returns the persisted acknowledgement when the final steering response is retried", async () => { + const seeded = await seedQueue(); + await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1])); + await db + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: new Date("2026-08-22T15:04:00.000Z"), + payload: { + issueId: seeded.issueId, + commentId: seeded.commentIds[0], + _paperclipWakeContext: { + commentId: seeded.commentIds[0], + wakeCommentId: seeded.commentIds[0], + wakeCommentIds: [seeded.commentIds[0]], + }, + }, + }) + .where(eq(agentWakeupRequests.id, seeded.wakeId)); + await db + .update(heartbeatRuns) + .set({ + resultJson: { + queuedSteeringAcknowledgements: { + [seeded.commentIds[0]]: { + status: "acknowledged", + queueId: seeded.wakeId, + turnId: "turn-acknowledged", + acknowledgedAt: "2026-08-22T15:04:00.000Z", + }, + }, + }, + }) + .where(eq(heartbeatRuns.id, seeded.runId)); + + const retried = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ + queueId: seeded.wakeId, + targetRunId: seeded.runId, + revision: "response-was-lost-before-the-client-stored-the-revision", + }); + + expect(retried.status, JSON.stringify(retried.body)).toBe(200); + expect(retried.body).toMatchObject({ + issueId: seeded.issueId, + queueId: null, + state: null, + entries: [], + }); + const activity = await db + .select({ details: activityLog.details }) + .from(activityLog) + .where(eq(activityLog.action, "issue.queued_comment_steered")) + .then((rows) => rows[0]); + expect(activity?.details).toMatchObject({ + commentId: seeded.commentIds[0], + targetRunId: seeded.runId, + turnId: "turn-acknowledged", + duplicate: true, + }); + }); + + it("does not reuse an acknowledgement from a different queue", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + await db + .update(heartbeatRuns) + .set({ + resultJson: { + queuedSteeringAcknowledgements: { + [seeded.commentIds[0]]: { + status: "acknowledged", + queueId: randomUUID(), + turnId: "turn-from-another-queue", + acknowledgedAt: "2026-08-22T15:04:00.000Z", + }, + }, + }, + }) + .where(eq(heartbeatRuns.id, seeded.runId)); + + const steered = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ + queueId: seeded.wakeId, + targetRunId: seeded.runId, + revision: initial.body.revision, + }); + + expect(steered.status).toBe(409); + expect(steered.body.details).toMatchObject({ + code: "steering_temporarily_unavailable", + retryable: true, + }); + const queueAfterFailure = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id)) + .toEqual(seeded.commentIds); + }); + + it("keeps queue edits available during handoff but rejects stale same-turn steering", async () => { + const seeded = await seedQueue(); + const initial = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + await db + .update(heartbeatRuns) + .set({ status: "succeeded", finishedAt: new Date("2026-08-22T15:05:00.000Z") }) + .where(eq(heartbeatRuns.id, seeded.runId)); + + const edit = await request(app(seeded.companyId)) + .patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`) + .send({ + queueId: seeded.wakeId, + revision: initial.body.revision, + body: "edited during handoff", + }); + + expect(edit.status, JSON.stringify(edit.body)).toBe(200); + const stored = await db + .select({ body: issueComments.body }) + .from(issueComments) + .where(eq(issueComments.id, seeded.commentIds[0])) + .then((rows) => rows[0]); + expect(stored?.body).toBe("edited during handoff"); + const steer = await request(app(seeded.companyId)) + .post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`) + .send({ + queueId: seeded.wakeId, + targetRunId: seeded.runId, + revision: edit.body.revision, + }); + expect(steer.status).toBe(409); + expect(steer.body.details?.code).toBe("queued_comment_stale_target"); + const wake = await db + .select({ status: agentWakeupRequests.status, payload: agentWakeupRequests.payload }) + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, seeded.wakeId)) + .then((rows) => rows[0]); + expect(wake?.status).toBe("deferred_issue_execution"); + expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual(seeded.commentIds); + }); + it("cancels a queued continuation whose comments disappeared before claim", async () => { const seeded = await seedQueue(); const queueRunId = await promoteQueue(seeded); @@ -437,6 +601,27 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { expect(wake?.status).toBe("cancelled"); }); + it("keeps a persisted legacy queue on the legacy protocol after the agent changes adapters", async () => { + const seeded = await seedQueue(); + const queueRunId = await promoteQueue(seeded); + await db + .update(heartbeatRuns) + .set({ runtimeMode: "legacy" }) + .where(eq(heartbeatRuns.id, queueRunId)); + + const queued = await request(app(seeded.companyId)) + .get(`/api/issues/${seeded.issueId}/queued-comments`); + + expect(queued.status, JSON.stringify(queued.body)).toBe(200); + expect(queued.body).toMatchObject({ + queueId: seeded.wakeId, + state: "queued", + targetRunId: null, + protocol: "legacy", + steeringDisposition: "unsupported", + }); + }); + it("serializes discard against queued-run claim", async () => { const seeded = await seedQueue(); await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1])); diff --git a/server/src/adapters/registry.ts b/server/src/adapters/registry.ts index 46fce45e96..f548c64a77 100644 --- a/server/src/adapters/registry.ts +++ b/server/src/adapters/registry.ts @@ -419,7 +419,8 @@ const paperclipRunnerAdapter: ServerAdapterModule = { listModels: listCodexModels, refreshModels: refreshCodexModels, supportsLocalAgentJwt: false, - supportsInstructionsBundle: false, + supportsInstructionsBundle: true, + instructionsPathKey: "instructionsFilePath", requiresMaterializedRuntimeSkills: false, getRuntimeCommandSpec: (config) => buildNpmRuntimeCommandSpec(config, "codex", "@openai/codex"), agentConfigurationDoc: diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 0640ea1877..1928360045 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -263,6 +263,11 @@ import { observeCrossIssueInfluence, type CrossIssueInfluenceKind, } from "../services/cross-issue-influence-limit.js"; +import { + getNativeSessionSteeringState, + NativeSessionSteeringError, + steerNativeSession, +} from "../services/native-runtime/native-session-executor.js"; import { queuedCommentIdsFromWakePayload, withQueuedCommentIdsInRunContext, @@ -277,6 +282,9 @@ const queuedCommentMutationTargetSchema = z.object({ queueId: z.string().min(1), revision: z.string().min(1), }); +const queuedCommentSteeringTargetSchema = queuedCommentMutationTargetSchema.extend({ + targetRunId: z.string().min(1), +}); const editQueuedCommentSchema = queuedCommentMutationTargetSchema.extend({ body: z .string() @@ -5511,12 +5519,33 @@ export function issueRoutes( : input.queueState; const wake = queueState?.wake ?? null; const comments = await queueCommentsForWake(input.executor, input.issue.id, wake); - const protocol = input.activeRun?.runtimeMode === "native" - || queueState?.queueRun?.runtimeMode === "native" + const assignedAgent = input.issue.assigneeAgentId + ? await input.executor + .select({ adapterType: agents.adapterType }) + .from(agents) + .where(and( + eq(agents.id, input.issue.assigneeAgentId), + eq(agents.companyId, input.issue.companyId), + )) + .limit(1) + .then((rows) => rows[0] ?? null) + : null; + const persistedRuntimeMode = queueState?.state === "queued" && queueState.queueRun + ? queueState.queueRun.runtimeMode + : queueState?.state === "deferred" && input.activeRun + ? input.activeRun.runtimeMode + : null; + const protocol = persistedRuntimeMode === "native" + || (persistedRuntimeMode === null && assignedAgent?.adapterType === "paperclip_runner") ? "paperclip_runner_v1" as const : "legacy" as const; const steeringRun = queueState?.state === "deferred" ? input.activeRun : null; - let steeringDisposition = input.steeringDisposition ?? "unsupported" as const; + let steeringDisposition = input.steeringDisposition + ?? (protocol === "paperclip_runner_v1" && steeringRun + ? await getNativeSessionSteeringState(steeringRun.id) + .then((state) => state.disposition) + .catch(() => "temporarily_unavailable" as const) + : "unsupported" as const); if (protocol === "paperclip_runner_v1" && (!steeringRun || comments.length === 0)) { steeringDisposition = "temporarily_unavailable"; } @@ -11851,6 +11880,210 @@ export function issueRoutes( }, ); + router.post( + "/issues/:id/queued-comments/:commentId/steer", + validate(queuedCommentSteeringTargetSchema), + async (req, res) => { + assertBoard(req); + if (!req.actor.userId) throw forbidden("Board user context required"); + const id = req.params.id as string; + const commentId = req.params.commentId as string; + const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found"); + if (!issue) return; + const actor = getActorInfo(req); + let acknowledgedTurnId: string | null = null; + let duplicate = false; + let queue: IssueQueuedCommentQueue; + try { + queue = await db.transaction(async (tx) => { + // A client can lose the successful response after the final queued + // message cancels its wake. Lock the original queue and target run + // first so that the persisted acknowledgement remains a durable + // idempotency record even when no pending queue remains. + await tx + .select({ id: issueRows.id }) + .from(issueRows) + .where(and(eq(issueRows.id, issue.id), eq(issueRows.companyId, issue.companyId))) + .for("update"); + const retryWake = await tx + .select() + .from(agentWakeupRequests) + .where(and( + eq(agentWakeupRequests.id, req.body.queueId), + eq(agentWakeupRequests.companyId, issue.companyId), + issue.assigneeAgentId + ? eq(agentWakeupRequests.agentId, issue.assigneeAgentId) + : undefined, + )) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null); + const retryRun = retryWake && readObject(retryWake.payload).issueId === issue.id + ? await tx + .select() + .from(heartbeatRuns) + .where(and( + eq(heartbeatRuns.id, req.body.targetRunId), + eq(heartbeatRuns.companyId, issue.companyId), + eq(heartbeatRuns.agentId, retryWake.agentId), + )) + .for("update") + .limit(1) + .then((rows) => rows[0] ?? null) + : null; + const retryRunContext = readObject(retryRun?.contextSnapshot); + const retryRunResult = readObject(retryRun?.resultJson); + const retryAcknowledgements = readObject( + retryRunResult.queuedSteeringAcknowledgements, + ); + const retryAcknowledgement = readObject(retryAcknowledgements[commentId]); + if ( + retryRun + && (retryRunContext.issueId === issue.id || retryRunContext.taskId === issue.id) + && retryAcknowledgement.status === "acknowledged" + && retryAcknowledgement.queueId === req.body.queueId + ) { + duplicate = true; + acknowledgedTurnId = typeof retryAcknowledgement.turnId === "string" + ? retryAcknowledgement.turnId + : null; + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: retryRun.status === "running" ? retryRun : null, + actor, + }); + } + + const locked = await lockQueuedCommentState({ + tx, + issue, + actor, + queueId: req.body.queueId, + targetRunId: req.body.targetRunId, + }); + if (!locked.activeRun) { + throw conflict("The queued message targets a stale run", { + code: "queued_comment_stale_target", + }); + } + const runResult = readObject(locked.activeRun.resultJson); + const acknowledgements = readObject(runResult.queuedSteeringAcknowledgements); + const priorAcknowledgement = readObject(acknowledgements[commentId]); + if ( + priorAcknowledgement.status === "acknowledged" + && priorAcknowledgement.queueId === req.body.queueId + ) { + duplicate = true; + acknowledgedTurnId = typeof priorAcknowledgement.turnId === "string" + ? priorAcknowledgement.turnId + : null; + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: locked.activeRun, + actor, + queueState: locked.queueState, + }); + } + assertQueueMutationTarget({ + queue: locked.queue, + queueId: req.body.queueId, + revision: req.body.revision, + }); + if (locked.queue.protocol !== "paperclip_runner_v1") { + throw conflict("This runner does not support same-turn steering", { + code: "steering_unsupported", + }); + } + const entry = locked.queue.entries.find((candidate) => candidate.comment.id === commentId); + if (!entry) { + throw conflict("The queued message is no longer pending", { + code: "queued_comment_not_pending", + }); + } + + const acknowledgement = await steerNativeSession({ + runId: locked.activeRun.id, + message: entry.comment.body, + correlationId: commentId, + }); + acknowledgedTurnId = acknowledgement.turnId; + const remainingIds = locked.queue.entries + .map((candidate) => candidate.comment.id) + .filter((candidateId) => candidateId !== commentId); + const now = new Date(); + const nextWake = remainingIds.length === 0 + ? await tx + .update(agentWakeupRequests) + .set({ status: "cancelled", finishedAt: now, updatedAt: now }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then(() => null) + : await tx + .update(agentWakeupRequests) + .set({ + payload: withQueuedCommentIdsInWakePayload(locked.wake.payload, remainingIds), + updatedAt: now, + }) + .where(eq(agentWakeupRequests.id, locked.wake.id)) + .returning() + .then((rows) => rows[0] ?? locked.wake); + await tx + .update(heartbeatRuns) + .set({ + resultJson: { + ...runResult, + queuedSteeringAcknowledgements: { + ...acknowledgements, + [commentId]: { + status: "acknowledged", + queueId: req.body.queueId, + turnId: acknowledgement.turnId, + acknowledgedAt: now.toISOString(), + }, + }, + }, + updatedAt: now, + }) + .where(eq(heartbeatRuns.id, locked.activeRun.id)); + return buildQueuedCommentQueue({ + executor: tx, + issue, + activeRun: locked.activeRun, + actor, + queueState: nextWake + ? { wake: nextWake, state: "deferred", queueRun: null } + : null, + }); + }); + } catch (error) { + if (error instanceof NativeSessionSteeringError) { + throw conflict(error.message, { code: error.code, retryable: true }); + } + throw error; + } + await logActivity(db, { + companyId: issue.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + agentApiKeyId: actor.agentApiKeyId, + action: "issue.queued_comment_steered", + entityType: "issue", + entityId: issue.id, + details: { + commentId, + targetRunId: req.body.targetRunId, + turnId: acknowledgedTurnId, + duplicate, + }, + }); + res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); + }, + ); + router.delete( "/issues/:id/queued-comments/:commentId", diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 563f064275..f73b47f6a1 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -4702,6 +4702,22 @@ registry.registerPath({ responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 404: r.notFound, 409: r.conflict }, }); +registry.registerPath({ + method: "post", + path: "/api/issues/{id}/queued-comments/{commentId}/steer", + tags: ["issues"], + summary: "Steer a queued issue comment into the active native run", + request: { + params: z.object({ id: z.string(), commentId: z.string() }), + body: jsonBody(z.object({ + queueId: z.string().min(1), + revision: z.string().min(1), + targetRunId: z.string().min(1), + })), + }, + responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict }, +}); + registry.registerPath({ method: "post", path: "/api/heartbeat-runs/{runId}/runtime-requests/{requestId}/resolve", diff --git a/ui/src/adapters/use-adapter-capabilities.ts b/ui/src/adapters/use-adapter-capabilities.ts index 58e6a1f1a4..a3da3d9e83 100644 --- a/ui/src/adapters/use-adapter-capabilities.ts +++ b/ui/src/adapters/use-adapter-capabilities.ts @@ -22,7 +22,7 @@ const ALL_FALSE: AdapterCapabilities = { const KNOWN_DEFAULTS: Record = { claude_local: { supportsInstructionsBundle: true, supportsSkills: true, supportsLocalAgentJwt: true, requiresMaterializedRuntimeSkills: false, supportsAcp: true, login: { panelMode: "submitted_browser_code", timeoutPolicy: "fixed" } }, codex_local: { supportsInstructionsBundle: true, supportsSkills: true, supportsLocalAgentJwt: true, requiresMaterializedRuntimeSkills: false, supportsAcp: true, login: { panelMode: "displayed_code", timeoutPolicy: "caller_bounded" } }, - paperclip_runner: { supportsInstructionsBundle: false, supportsSkills: true, supportsLocalAgentJwt: false, requiresMaterializedRuntimeSkills: false, supportsAcp: false }, + paperclip_runner: { supportsInstructionsBundle: true, supportsSkills: true, supportsLocalAgentJwt: false, requiresMaterializedRuntimeSkills: false, supportsAcp: false }, cursor: { supportsInstructionsBundle: true, supportsSkills: true, supportsLocalAgentJwt: true, requiresMaterializedRuntimeSkills: true, supportsAcp: false }, gemini_local: { supportsInstructionsBundle: true, supportsSkills: true, supportsLocalAgentJwt: true, requiresMaterializedRuntimeSkills: true, supportsAcp: true }, grok_local: { supportsInstructionsBundle: true, supportsSkills: true, supportsLocalAgentJwt: true, requiresMaterializedRuntimeSkills: true, supportsAcp: false, login: { panelMode: "displayed_code", timeoutPolicy: "caller_bounded" } }, diff --git a/ui/src/components/TaskChatThread.test.tsx b/ui/src/components/TaskChatThread.test.tsx index 0d37f90ca7..f0abe80280 100644 --- a/ui/src/components/TaskChatThread.test.tsx +++ b/ui/src/components/TaskChatThread.test.tsx @@ -14,8 +14,14 @@ import type { } from "@paperclipai/shared"; import { heartbeatsApi } from "@/api/heartbeats"; -const transcriptState = vi.hoisted(() => ({ transcriptByRun: new Map() })); -const nativeTranscriptState = vi.hoisted(() => ({ transcriptByRun: new Map() })); +const transcriptState = vi.hoisted(() => ({ + transcriptByRun: new Map(), + isInitialHydrating: false, +})); +const nativeTranscriptState = vi.hoisted(() => ({ + transcriptByRun: new Map(), + errorsByRun: new Map(), +})); const transcriptHookRuns = vi.hoisted(() => ({ legacy: [] as unknown[][], native: [] as unknown[][] })); const sidebarState = vi.hoisted(() => ({ isMobile: false })); const planState = vi.hoisted(() => ({ data: null as IssueDocument | null })); @@ -31,13 +37,19 @@ const DIRECT_ADAPTER_TYPES = [ vi.mock("@/components/transcript/useLiveRunTranscripts", () => ({ useLiveRunTranscripts: ({ runs }: { runs: unknown[] }) => { transcriptHookRuns.legacy.push(runs); - return { transcriptByRun: new Map(transcriptState.transcriptByRun) }; + return { + transcriptByRun: new Map(transcriptState.transcriptByRun), + isInitialHydrating: transcriptState.isInitialHydrating, + }; }, })); vi.mock("@/components/transcript/useNativeRunTranscripts", () => ({ useNativeRunTranscripts: (runs: unknown[]) => { transcriptHookRuns.native.push(runs); - return { transcriptByRun: new Map(nativeTranscriptState.transcriptByRun) }; + return { + transcriptByRun: new Map(nativeTranscriptState.transcriptByRun), + errorsByRun: new Map(nativeTranscriptState.errorsByRun), + }; }, })); vi.mock("@/context/SidebarContext", () => ({ @@ -79,7 +91,9 @@ let root: Root | null = null; beforeEach(() => { localStorage.clear(); transcriptState.transcriptByRun.clear(); + transcriptState.isInitialHydrating = false; nativeTranscriptState.transcriptByRun.clear(); + nativeTranscriptState.errorsByRun.clear(); transcriptHookRuns.legacy.length = 0; transcriptHookRuns.native.length = 0; sidebarState.isMobile = false; @@ -349,7 +363,7 @@ describe("TaskChatThread draft pass-through", () => { }); describe("TaskChatThread runtime transcript selection", () => { - it("selects persisted runtime facts while leaving direct adapters on the legacy parser", () => { + it("selects persisted runtime facts while retaining the log parser as native fallback", () => { render( { const legacyRuns = transcriptHookRuns.legacy.at(-1) as Array<{ id: string }>; const nativeRuns = transcriptHookRuns.native.at(-1) as Array<{ id: string }>; - expect(legacyRuns.map((run) => run.id)).toEqual( - DIRECT_ADAPTER_TYPES.map((_, index) => `legacy-run-${index}`), - ); + expect(legacyRuns.map((run) => run.id)).toEqual([ + "native-run", + ...DIRECT_ADAPTER_TYPES.map((_, index) => `legacy-run-${index}`), + ]); expect(nativeRuns.map((run) => run.id)).toEqual(["native-run"]); }); @@ -451,6 +466,88 @@ describe("TaskChatThread runtime transcript selection", () => { expect(container.querySelector('[data-testid="task-chat-runner-turn"]')).toBeNull(); }); + it("uses the live log when a native run has no persisted event transcript", () => { + transcriptState.transcriptByRun.set("native-run", [ + { + kind: "assistant", + ts: "2026-08-25T18:00:01.000Z", + text: "Visible from the runner log fallback.", + channel: "progress", + }, + ]); + + render( + {}} + issueStatus="in_progress" + activeRun={{ + id: "native-run", + runtimeMode: "native", + status: "running", + invocationSource: "issue", + triggerDetail: null, + startedAt: "2026-08-25T18:00:00.000Z", + finishedAt: null, + createdAt: "2026-08-25T18:00:00.000Z", + agentId: "agent-1", + agentName: "Runner", + adapterType: "paperclip_runner", + }} + />, + ); + + expect(container.textContent).toContain("Visible from the runner log fallback."); + }); + + it("uses a fresher live log when native event polling fails after earlier events", () => { + nativeTranscriptState.transcriptByRun.set("native-run", [ + { + kind: "assistant", + ts: "2026-08-25T18:00:01.000Z", + text: "Stale native activity.", + channel: "progress", + }, + ]); + nativeTranscriptState.errorsByRun.set("native-run", { + message: "event endpoint unavailable", + failedAt: "2026-08-25T18:00:02.000Z", + }); + transcriptState.transcriptByRun.set("native-run", [ + { + kind: "assistant", + ts: "2026-08-25T18:00:03.000Z", + text: "Fresh activity from the runner log.", + channel: "progress", + }, + ]); + + render( + {}} + issueStatus="in_progress" + activeRun={{ + id: "native-run", + runtimeMode: "native", + status: "running", + invocationSource: "issue", + triggerDetail: null, + startedAt: "2026-08-25T18:00:00.000Z", + finishedAt: null, + createdAt: "2026-08-25T18:00:00.000Z", + agentId: "agent-1", + agentName: "Runner", + adapterType: "paperclip_runner", + }} + />, + ); + + expect(container.textContent).toContain("Fresh activity from the runner log."); + expect(container.textContent).not.toContain("Stale native activity."); + expect(container.textContent).not.toContain("temporarily unavailable"); + }); + it("keeps legacy channel-less native messages readable across settlement", () => { nativeTranscriptState.transcriptByRun.set("native-run", [ { diff --git a/ui/src/components/TaskChatThread.tsx b/ui/src/components/TaskChatThread.tsx index 25f27dcfef..6e6aeb1443 100644 --- a/ui/src/components/TaskChatThread.tsx +++ b/ui/src/components/TaskChatThread.tsx @@ -706,23 +706,44 @@ export function TaskChatThread(props: TaskChatThreadProps) { return [...map.values()]; }, [linkedRuns, liveRuns, activeRun]); - const legacyRuns = useMemo( - () => runs.filter((run) => run.runtimeMode !== "native"), - [runs], - ); const nativeRuns = useMemo( () => runs.filter((run) => run.runtimeMode === "native"), [runs], ); - const { transcriptByRun: legacyTranscriptByRun } = useLiveRunTranscripts({ - runs: legacyRuns, + const { + transcriptByRun: logTranscriptByRun, + isInitialHydrating: logsAreInitiallyHydrating, + } = useLiveRunTranscripts({ + // Native events are authoritative, but the persisted/live log remains a + // compatibility source when an upgraded server has no event history or + // the native event endpoint is temporarily unavailable. + runs, companyId, }); - const { transcriptByRun: nativeTranscriptByRun } = useNativeRunTranscripts(nativeRuns); - const transcriptByRun = useMemo( - () => new Map([...legacyTranscriptByRun, ...nativeTranscriptByRun]), - [legacyTranscriptByRun, nativeTranscriptByRun], - ); + const { + transcriptByRun: nativeTranscriptByRun, + errorsByRun: nativeTranscriptErrorsByRun, + } = useNativeRunTranscripts(nativeRuns); + const transcriptByRun = useMemo(() => { + const next = new Map(logTranscriptByRun); + for (const run of nativeRuns) { + const logTranscript = logTranscriptByRun.get(run.id) ?? []; + const nativeTranscript = nativeTranscriptByRun.get(run.id) ?? []; + const nativeEventsUnavailable = nativeTranscriptErrorsByRun.has(run.id); + if ( + nativeTranscript.length > 0 + && (!nativeEventsUnavailable || logTranscript.length === 0) + ) { + next.set(run.id, nativeTranscript); + } + } + return next; + }, [ + logTranscriptByRun, + nativeRuns, + nativeTranscriptByRun, + nativeTranscriptErrorsByRun, + ]); // The single in-flight run whose turn we stream live (non-terminal). const liveRun = useMemo(() => { @@ -1652,6 +1673,12 @@ export function TaskChatThread(props: TaskChatThreadProps) { const tailAllEntries = tailRunId ? (transcriptByRun.get(tailRunId) ?? []) : []; + const tailActivityUnavailable = Boolean( + tailRunId + && nativeTranscriptErrorsByRun.has(tailRunId) + && (logTranscriptByRun.get(tailRunId)?.length ?? 0) === 0 + && !logsAreInitiallyHydrating, + ); const tailTimelineAnchors = tailRunId ? paperclipRunnerTail ? (steeringAnchorsByRun.get(tailRunId) ?? []) @@ -2205,6 +2232,7 @@ export function TaskChatThread(props: TaskChatThreadProps) { } startedAtMs={tailStartedAtMs} finishedAtMs={tailFinishedAtMs} + activityUnavailable={tailActivityUnavailable} onRuntimeRequestDecision={ handleRuntimeRequestDecision } diff --git a/ui/src/components/task-chat/TaskChatRunnerTurn.test.tsx b/ui/src/components/task-chat/TaskChatRunnerTurn.test.tsx index 2e79e5c8fd..fbb0a642ba 100644 --- a/ui/src/components/task-chat/TaskChatRunnerTurn.test.tsx +++ b/ui/src/components/task-chat/TaskChatRunnerTurn.test.tsx @@ -314,6 +314,47 @@ describe("TaskChatRunnerTurn", () => { ).toContain("Reasoning"); }); + it("keeps the latest provider-authored reasoning line visible while activity is folded", () => { + render([ + { + id: "reasoning", + kind: "thinking", + lines: ["Inspecting the task state.", "Checking the steering path."], + streaming: true, + channel: "summary", + transcriptIndex: 2, + }, + ]); + + const ticker = container.querySelector('[data-testid="task-chat-reasoning-ticker"]'); + expect(ticker?.textContent).toContain("Checking the steering path."); + expect(container.querySelector('[data-testid="task-chat-thinking"]')).toBeNull(); + }); + + it("surfaces native activity transport failure while retrying", () => { + act(() => + root.render( + + + + + , + ), + ); + + expect( + container.querySelector('[data-testid="task-chat-activity-unavailable"]') + ?.textContent, + ).toContain("temporarily unavailable"); + }); + it("starts a separate activity group at every commentary boundary", () => { render([ { diff --git a/ui/src/components/task-chat/TaskChatRunnerTurn.tsx b/ui/src/components/task-chat/TaskChatRunnerTurn.tsx index fdaf7b3c5b..5393a5e573 100644 --- a/ui/src/components/task-chat/TaskChatRunnerTurn.tsx +++ b/ui/src/components/task-chat/TaskChatRunnerTurn.tsx @@ -1,5 +1,5 @@ -import { useRef, type ComponentType, type SVGProps } from "react"; -import { OctagonX } from "lucide-react"; +import { useRef, useState, type ComponentType, type SVGProps } from "react"; +import { Brain, OctagonX } from "lucide-react"; import { MarkdownBody } from "@/components/MarkdownBody"; import { useSecondTick } from "@/hooks/useSecondTick"; import { cn } from "@/lib/utils"; @@ -75,6 +75,118 @@ function currentActivityStatusItems( return items.slice(boundaryIndex + 1); } +type FoldedNarration = + | { kind: "commentary"; item: TaskChatMessageItem; order: number } + | { + kind: "reasoning"; + item: TaskChatThinkingItem; + line: string | null; + lineIndex: number; + order: number; + }; + +function latestFoldedNarration(items: readonly TaskChatItem[]): FoldedNarration | null { + let latest: FoldedNarration | null = null; + for (const [index, item] of items.entries()) { + const order = item.kind === "message" || item.kind === "thinking" + ? item.transcriptIndex ?? index + : -1; + if (item.kind === "message" && item.interstitial && item.text.trim()) { + if (!latest || order >= latest.order) latest = { kind: "commentary", item, order }; + continue; + } + if (item.kind !== "thinking") continue; + let lineIndex = -1; + for (let candidate = item.lines.length - 1; candidate >= 0; candidate -= 1) { + if (item.lines[candidate]?.trim()) { + lineIndex = candidate; + break; + } + } + if (!latest || order >= latest.order) { + latest = { + kind: "reasoning", + item, + line: lineIndex < 0 ? null : item.lines[lineIndex]!.trim(), + lineIndex, + order, + }; + } + } + return latest; +} + +function FoldedReasoningTicker({ logicalKey, text }: { logicalKey: string; text: string }) { + const [ticker, setTicker] = useState({ + logicalKey, + motionKey: 0, + current: text, + exiting: null as string | null, + }); + if (ticker.logicalKey !== logicalKey) { + setTicker({ + logicalKey, + motionKey: ticker.motionKey + 1, + current: text, + exiting: ticker.current, + }); + } else if (ticker.current !== text) { + // Token fragments update the mounted line. Only a new logical line moves + // the ticker, so streaming text does not restart the animation per token. + setTicker({ ...ticker, current: text }); + } + + return ( +
+
+ +
+
+ {ticker.exiting !== null ? ( + setTicker((current) => ({ ...current, exiting: null }))} + > + {ticker.exiting} + + ) : null} + 0 && "cot-line-enter", + )} + aria-live="polite" + aria-atomic="true" + > + {ticker.current} + +
+
+ ); +} + +function FoldedLiveNarration({ narration }: { narration: FoldedNarration }) { + if (narration.kind === "reasoning") { + if (!narration.line) return null; + return ( + + ); + } + return ( +
+ {narration.item.text} +
+ ); +} + function formatCompactDuration(ms: number | null): string | null { if (ms == null || !Number.isFinite(ms)) return null; const totalSeconds = Math.max(0, Math.floor(ms / 1000)); @@ -296,6 +408,7 @@ export function TaskChatRunnerTurn({ status, startedAtMs, finishedAtMs, + activityUnavailable = false, onRuntimeRequestDecision, }: { /** Stable identity used to clear replay-latched final text for the next turn. */ @@ -306,12 +419,14 @@ export function TaskChatRunnerTurn({ status: string; startedAtMs: number | null; finishedAtMs?: number | null; + activityUnavailable?: boolean; onRuntimeRequestDecision?: ( item: TaskChatRuntimeRequestItem, decision: TaskChatRuntimeRequestDecision, ) => void | Promise; }) { const terminal = isTerminalRunStatus(status); + const narration = latestFoldedNarration(items); const timelineRows = buildTurnTimelineRows( paperclipRunnerTimelineItems(items), !terminal, @@ -373,6 +488,20 @@ export function TaskChatRunnerTurn({ finishedAtMs={finishedAtMs} /> + {!terminal && narration && !final ? ( +
+ +
+ ) : null} + {activityUnavailable ? ( +
+ Live runner activity is temporarily unavailable. Retrying… +
+ ) : null} {timelineRows.length > 0 ? (
vi.fn()); + +vi.mock("@/api/heartbeats", () => ({ + heartbeatsApi: { events: eventsMock }, +})); + +function Probe() { + const { errorsByRun } = useNativeRunTranscripts([ + { id: "native-run", status: "succeeded", runtimeMode: "native" }, + ]); + return ( +
+ {[...errorsByRun.keys()].join(",")} +
+ ); +} + +function MultiRunProbe() { + useNativeRunTranscripts([ + { id: "failed-run", status: "succeeded", runtimeMode: "native" }, + { id: "healthy-run", status: "succeeded", runtimeMode: "native" }, + ]); + return null; +} + +describe("useNativeRunTranscripts", () => { + let container: HTMLDivElement; + let root: Root; + + beforeEach(() => { + vi.useFakeTimers(); + eventsMock.mockReset(); + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + }); + + afterEach(() => { + act(() => root.unmount()); + container.remove(); + vi.useRealTimers(); + }); + + it("exposes event transport failures and retries terminal runs until recovery", async () => { + eventsMock + .mockRejectedValueOnce(new Error("event endpoint unavailable")) + .mockResolvedValue([]); + + await act(async () => { + root.render(); + await Promise.resolve(); + }); + expect(container.textContent).toBe("native-run"); + + await act(async () => { + await vi.advanceTimersByTimeAsync(2_000); + }); + expect(eventsMock).toHaveBeenCalledTimes(2); + expect(container.textContent).toBe(""); + }); + + it("retries only terminal runs whose event request failed", async () => { + eventsMock.mockImplementation((runId: string) => ( + runId === "failed-run" + ? Promise.reject(new Error("event endpoint unavailable")) + : Promise.resolve([]) + )); + + await act(async () => { + root.render(); + await Promise.resolve(); + }); + expect(eventsMock).toHaveBeenCalledTimes(2); + + await act(async () => { + await vi.advanceTimersByTimeAsync(2_000); + }); + expect(eventsMock).toHaveBeenCalledTimes(3); + expect(eventsMock.mock.calls.at(-1)?.[0]).toBe("failed-run"); + }); +}); diff --git a/ui/src/components/transcript/useNativeRunTranscripts.ts b/ui/src/components/transcript/useNativeRunTranscripts.ts index 6f225247f4..0f1247131c 100644 --- a/ui/src/components/transcript/useNativeRunTranscripts.ts +++ b/ui/src/components/transcript/useNativeRunTranscripts.ts @@ -13,6 +13,11 @@ export interface NativeRunTranscriptSource { runtimeMode?: "legacy" | "native"; } +export interface NativeRunTranscriptError { + message: string; + failedAt: string; +} + function isLive(status: string): boolean { return status === "queued" || status === "running"; } @@ -30,15 +35,17 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource [nativeRunsKey], ); const [eventsByRun, setEventsByRun] = useState>(new Map()); + const [errorsByRun, setErrorsByRun] = useState>(new Map()); const cursorByRunRef = useRef(new Map()); useEffect(() => { let cancelled = false; let timer: number | null = null; - const refresh = async () => { + const refresh = async (runsToRefresh: readonly NativeRunTranscriptSource[]) => { const updates = new Map(); - await Promise.all(nativeRuns.map(async (run) => { + const errors = new Map(); + await Promise.all(runsToRefresh.map(async (run) => { try { let cursor = cursorByRunRef.current.get(run.id) ?? 0; const incoming: HeartbeatRunEvent[] = []; @@ -56,8 +63,12 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource } if (incoming.length > 0) updates.set(run.id, incoming); cursorByRunRef.current.set(run.id, cursor); - } catch { + } catch (error) { // Keep the last durable cursor; the next poll retries this run only. + errors.set(run.id, { + message: error instanceof Error ? error.message : "Native run activity could not be loaded", + failedAt: new Date().toISOString(), + }); } })); @@ -75,13 +86,27 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource } return next; }); + setErrorsByRun((previous) => { + const next = new Map(); + for (const runId of retainedIds) { + const error = errors.get(runId); + if (error) next.set(runId, previous.get(runId) ?? error); + } + return next; + }); - if (nativeRuns.some((run) => isLive(run.status))) { - timer = window.setTimeout(refresh, EVENT_POLL_INTERVAL_MS); + if (nativeRuns.some((run) => isLive(run.status)) || errors.size > 0) { + const retryRuns = nativeRuns.filter( + (run) => isLive(run.status) || errors.has(run.id), + ); + timer = window.setTimeout( + () => void refresh(retryRuns), + EVENT_POLL_INTERVAL_MS, + ); } }; - void refresh(); + void refresh(nativeRuns); return () => { cancelled = true; if (timer !== null) window.clearTimeout(timer); @@ -96,5 +121,5 @@ export function useNativeRunTranscripts(runs: readonly NativeRunTranscriptSource return transcripts; }, [eventsByRun, nativeRuns]); - return { transcriptByRun }; + return { transcriptByRun, errorsByRun }; }