diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 3cc2067d727..fdfc680c06e 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1816,6 +1816,12 @@ dispatch with cancellation and execution, and re-exports the existing public helpers and configuration-incomplete error class. Keep preparation policy changes in this module and its tests. +Queued-comment display reads select only wake identity, queue state, typed comment +and interaction bindings, the displayed wait reason, and fresh-session flags. +They read only the queued run's runtime mode. Mutation locks retain complete +wake and run rows, so edits, ordering, interruption, and saved execution inputs +keep their existing write contract. + Run retrieval and session state are in `server/src/services/heartbeat/run-state.ts`. It owns bounded run projections, database encoding checks, task session reads and writes, explicit resumes, session compaction, and usage/billing helpers. diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 60985954bea..7b0af0e8a3d 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -220,6 +220,38 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { } finally { capture.flush(); await observed.$client.end(); } }); + it.each(["deferred", "queued"])("reads only queue display inputs from large %s wake payloads", async state => { + const f = await seedQueue(); + if (state === "queued") await promoteQueue(f); + const client = app(f.companyId); + const path = `/api/issues/${f.issueId}/queued-comments`; + const expected = await request(client).get(path).expect(200); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + const payload = "saved-wake-input🙂".repeat(30_000); + const context = wake.payload?._paperclipWakeContext as Record; + await db.update(agentWakeupRequests).set({ payload: { + ...wake.payload, prompt: payload, + _paperclipWakeContext: { ...context, prompt: payload, nativeExecutionInput: { instructions: payload } }, + } }).where(eq(agentWakeupRequests.id, f.wakeId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, prompt: payload }, + runnerProfileJson: { instructions: payload }, resultJson: { output: payload }, + }).where(eq(heartbeatRuns.companyId, f.companyId)); + const [before] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)); + const observed = createDb(tempDb!.connectionString), events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(observed, { agentWakeupRequests, heartbeatRuns }, capture); + try { + const response = await request(app(f.companyId, "queue-owner", undefined, observed)).get(path).expect(200); + expect(response.body).toEqual(expected.body); + capture.flush(); + const reads = events.filter(event => ["agent_wakeup_requests", "heartbeat_runs"].includes(String(event.table)) && event.operation === "select"); + expect(reads.some(event => event.table === "agent_wakeup_requests")).toBe(true); + expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(4096); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0]).toEqual(before); + await request(app(randomUUID(), "queue-owner", undefined, observed)).get(path).expect(404); + } finally { capture.flush(); await observed.$client.end(); } + }); + it.each(["cancelled", "running"])("rejects a late Done from an interrupted %s task run at the write boundary", async status => { const seeded = await seedQueue(); await db.update(heartbeatRuns).set({ status, resultJson: { executionCancellation: { state: "requested" } } }) diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 6aa65dfb774..bb92497ea3b 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -7287,9 +7287,9 @@ export function issueRoutes( type IssueQueueWake = typeof agentWakeupRequests.$inferSelect; type IssueQueueRun = typeof heartbeatRuns.$inferSelect; type IssueQueueState = { - wake: IssueQueueWake; + wake: Pick; state: "deferred" | "queued"; - queueRun: IssueQueueRun | null; + queueRun: Pick | null; }; async function findQueuedCommentWake( @@ -7298,7 +7298,23 @@ export function issueRoutes( ): Promise { if (!issue.assigneeAgentId) return null; const rows = await executor - .select() + .select({ + id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, + status: agentWakeupRequests.status, runId: agentWakeupRequests.runId, + payload: sql>`jsonb_build_object( + 'issueId', ${agentWakeupRequests.payload}->'issueId', + 'mutation', ${agentWakeupRequests.payload}->'mutation', + 'interactionId', ${agentWakeupRequests.payload}->'interactionId', + 'interactionStatus', ${agentWakeupRequests.payload}->'interactionStatus', + 'externalChatContinuation', ${agentWakeupRequests.payload}->'externalChatContinuation', + 'executionWait', jsonb_build_object( + 'reason', ${agentWakeupRequests.payload}->'executionWait'->'reason', + 'message', ${agentWakeupRequests.payload}->'executionWait'->'message'), + '_paperclipWakeContext', jsonb_build_object( + 'wakeCommentIds', ${agentWakeupRequests.payload}->'_paperclipWakeContext'->'wakeCommentIds', + 'forceFreshSession', ${agentWakeupRequests.payload}->'_paperclipWakeContext'->'forceFreshSession') + )`, + }) .from(agentWakeupRequests) .where( and( @@ -7323,7 +7339,7 @@ export function issueRoutes( } if (!wake.runId) continue; const queueRun = await executor - .select() + .select({ runtimeMode: heartbeatRuns.runtimeMode }) .from(heartbeatRuns) .where( and( @@ -7344,7 +7360,7 @@ export function issueRoutes( async function queueCommentsForWake( executor: IssueQueueDb, issueId: string, - wake: IssueQueueWake | null, + wake: Pick | null, ) { if (wake && queuedInteractionId(wake.payload)) { const response = await readQueuedInteractionResponse(executor as Db, wake.companyId, issueId, wake.payload);