From b54317d3fb12531b990e615b262ca05ac2e2b686 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 22:44:18 -0600 Subject: [PATCH] fix(native): give a preparing run's first turn the user's queued messages instead of interrupting it (#20168) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A Paperclip Runner (native) run prepares its workspace and environment before it starts its first provider turn. This can take about a minute. > - A user's message that arrives during that time waits as a deferred wake behind the run. The Interrupt control then cancelled the run before any provider turn. > - The cancel threw away the whole preparation. A new run had to prepare again, so the user waited longer for the first token. > - This pull request gives the run's first turn the queued messages. The run seals its first-turn input immediately before `turn.start`, so the messages are part of the prompt that the turn starts with. > - The benefit is that the user sees the message handled in the same run, with no cancel and no second preparation. ## Linked Issues or Issue Description **What happened?** A Paperclip Runner run took about 54 s to prepare (environment startup, workspace, runtime selection). The user sent a second message 32 s into that time and clicked Interrupt 21 s later. Interrupt cancelled the run as `operator_interrupted` ("Interrupted to send queued messages") before any provider turn. A new run then prepared again from the start. The runtime-selection transaction that froze the run's input completed about 5 s before the click, and the coordinator claim came 4 s after it, so the message could not be given to the first turn at any point in the old code. **Expected behavior** When a user message arrives for a task whose current run has not started its first provider turn, the run does not stop. The message goes into that run's first turn. A run that is in a turn is interrupted as before. **Steps to reproduce** 1. Assign a task to a Paperclip Runner agent with a sandbox environment, so preparation takes a long time. 2. While the run prepares, post a second comment on the task. 3. Click Interrupt on the queued message before the run starts its provider turn. 4. See that the run is cancelled and a new run prepares again. **Paperclip version or commit** Staging revision b43cccbb0. ## What Changed - `server/src/services/native-runtime/first-turn-queued-input.ts` (new): `claimFirstTurnQueuedInput` seals a run's first-turn input in one transaction. It locks the task, the deferred wakes, and the run in the same order as the queue routes. It takes the responsible user's plain queued comments in queue order (not interaction answers, interrupt intents, independent continuations, conversations, or other users' messages). It renders them with source-trust sanitizing and secret redaction, then writes a `firstTurnQueuedInput` receipt (`paperclip.first-turn-queued-input.v1`) into the run context. A second call for the same run returns the same messages. `runAcceptsFirstTurnQueuedInput` is true for a running native (or preparing native) run that has no receipt and no accepted provider turn. - `packages/paperclip-runner/src/native-session-runtime.ts`: new `getFirstTurnQueuedInput` option. The runner calls it immediately before the first `session.startTurn` and appends the result to `task.prompt` (or to `events` for a resumed-session continuation envelope). It is not called for the disposition-only recovery turn. - `native-session-executor.ts`: passes the claim to the runner. A failed seal commits nothing, so the executor logs it and starts the turn without the messages. The messages stay queued. - `queued-comment-delivery.ts`: a message in a run's receipt counts as delivered only after the provider accepted that run's turn (`turn.accepted` or the provider's `turn.started`), and only if the message was not edited after the seal. Release promotion, Send, and later first turns use this rule. `explicit-native-continuation.ts` uses the same rule for stopped native runs. - `routes/issues.ts` Interrupt: when the target run has not sealed its first turn and has no accepted turn, the route does not record interrupt intent and does not cancel the run. The response has `executionWait.reason = "first_turn_pending"`. Other runs are interrupted as before. - Queue views (`routes/issues.ts` and the wake-queue adapter) hide messages that the active run took into its first turn. Queue edits keep those ids on the wake, so a run that fails before acceptance still leaves them queued. - `run-state.ts`: the issue-control run projection includes the receipt, so the queue poll can hide taken messages. - `doc/execution-semantics.md`: new section "Queued messages before a native run's first turn". ## Verification - Red before the fix, green after: - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/issue-queued-comments-routes.test.ts -t "first turn|already started"`: Interrupt on a preparing run and on a claimed native run before `turn.start` does not stop the run, and the first turn gets both messages in order. Before the fix, both cases cancelled the run. - `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/native-session-runtime.test.ts -t "first-turn queued input"`: the claim runs immediately before `turn.start`, and the first turn's prompt ends with the queued messages. Before the fix, the claim was never called. - `pnpm --filter @paperclipai/server exec vitest run src/services/native-runtime/first-turn-queued-input.test.ts`: a run that fails before `turn.start` leaves the message queued. The next run delivers it, and after that no later run or copied retry context takes it again. Before the delivery rule, the "exactly once" and "accepted then failed" cases failed. - Safety cases that pass before and after: Interrupt still cancels a run whose provider accepted a turn (including a run saved by the previous release with no receipt) and a run whose first-turn input is already sealed. The existing "explicit interrupt fallback" cases for preparing and native runs without a responsible user still pass. - Speed: the seal is one short transaction before `turn.start`. In the embedded-Postgres test it took 5.5 ms. It removes a full second preparation (54 s on staging) from the message-to-first-token time. - Related suites: `src/modules/wake-queue`, `explicit-native-continuation`, `issue-queued-comment-queue`, `issue-queued-comments-routes`, `queued-comment-consumption`, `heartbeat/run-state` (888 tests), `native-session-executor.test.ts` (1306 tests), and the full `native-session-runtime.test.ts` (238 tests) pass. `pnpm --filter @paperclipai/server typecheck` passes. - Not done: no live run against a real provider or sandbox. Staging was read only. ## Risks - Previous-release state: a deferred wake saved by the previous release has the same shape and is taken normally. A run saved by the previous release has no receipt. If it is resumed before its first turn, it seals on the new release. If its provider already accepted a turn, Interrupt stops it as before. A run saved by the previous release while it was in `turn.start` but before acceptance counts as unsealed, so in that short window an Interrupt does not stop it and the message is delivered at the turn boundary. - Rollback: if the server rolls back after a run sealed and its turn was accepted, the old release does not read the receipt and can deliver those messages a second time. - A first turn that does not use the fresh-turn path (restart continuation, disposition-only recovery) does not call the claim. Its messages are delivered at the turn boundary as before. - A message that arrives after the seal waits for the turn boundary. Interrupt stops the run as before. - No schema migration. The receipt is a key in the run context. ## Model Used - Claude Opus 5.5 (Anthropic), model ID `claude-opus-5-5`, used through Claude Code with tool use (shell, file edits, test execution, read-only browser). ## 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 --- doc/execution-semantics.md | 26 ++ .../src/native-session-runtime.test.ts | 50 ++++ .../src/native-session-runtime.ts | 14 + .../issue-queued-comments-routes.test.ts | 102 +++++++ .../adapters/queued-comment-postgres.ts | 23 +- server/src/routes/issues.ts | 32 ++- .../services/explicit-native-continuation.ts | 5 +- server/src/services/heartbeat/run-state.ts | 5 +- .../first-turn-queued-input.test.ts | 189 ++++++++++++ .../native-runtime/first-turn-queued-input.ts | 269 ++++++++++++++++++ .../native-runtime/native-session-executor.ts | 7 + .../src/services/queued-comment-delivery.ts | 40 ++- 12 files changed, 749 insertions(+), 13 deletions(-) create mode 100644 server/src/services/native-runtime/first-turn-queued-input.test.ts create mode 100644 server/src/services/native-runtime/first-turn-queued-input.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 0f5c7d37023..762d41bf647 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3627,6 +3627,32 @@ after cancellation. The old task's Stop receipt cannot settle cleanup of a new replacement task. A lost reply with no known resource identity remains unresolved until provider inspection or operator reconciliation establishes its outcome. +### Queued messages before a native run's first turn + +A user message that arrives while a Paperclip Runner run is still preparing, +or after its coordinator claim but before `turn.start`, is not a reason to stop +that run. Immediately before the first `turn.start`, the runner seals the +turn's input: in one transaction under the task, queue, and run locks it takes +the responsible user's plain queued comments for the task, in their queued +order, adds them to that turn's prompt, and records a +`firstTurnQueuedInput` receipt (`paperclip.first-turn-queued-input.v1`) in the +run context. Interaction answers, interrupt intents, independent +continuations, conversations, and other users' messages keep their own +delivery. Taken messages leave the visible queue. + +The receipt does not consume the messages. They stay on their deferred wake +until the provider accepts that run's turn (`turn.accepted` or the provider's +`turn.started`); only then do release promotion, Send, and later first turns +treat them as delivered, whatever the run's later outcome. A run that ends +before acceptance leaves them queued for the next run, and a restart before +acceptance repeats the same sealed input. An edit after the seal is unread. + +**Interrupt** on a run that has neither sealed its first turn nor had a turn +accepted records no intent and does not stop the run; the response's +`executionWait.reason` is `first_turn_pending`. Once the input is sealed or a +turn was accepted, Interrupt stops the run as described below. A run saved by +a release without the seal counts as started once its provider accepted a turn. + ### Delivering queued messages after a legacy run stops The legacy queued-message Interrupt action accepts a null `targetRunId` when diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 9d40f756c99..fd5d7939cab 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -10352,3 +10352,53 @@ it("fails preconfigured recovery without replacing a missing provider session", expect(openSession).not.toHaveBeenCalled(); expect(openReplacementSession).not.toHaveBeenCalled(); }); + +describe("first-turn queued input", () => { + async function startFirstTurn(executionInput: NativeExecutionInputV1 | NativeExecutionInputV5, queued: string | null) { + const order: string[] = []; + const sent: string[] = []; + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() {}, + async startTurn(turn) { + order.push("turn.start"); + sent.push(turn.message.text); + throw new Error("stop after turn.start"); + }, + async result() { return null; }, + async snapshot() { + return { backendKind: "mock", sessionId: "driver-first-turn", identity, providerSessionId: "provider-first-turn", + cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; + }, + close: vi.fn(async () => undefined), + }; + const getFirstTurnQueuedInput = vi.fn(async () => { order.push("claim"); return queued; }); + await expect(executeNativeSession({ + input: executionInput, + getFirstTurnQueuedInput, + backend: { async descriptor() { return { kind: "mock", name: "first-turn", version: "1", capabilities: await session.capabilities(), + runtimeContextCapabilities: { instructions: "native" as const, skills: "native" as const, mcp: "native" as const } }; }, async openSession() { return session; } }, + controlPlane: { async openRun() {}, async checkpointSession() {}, async appendEvent() { return { cursor: 1, highestContiguousSourceSeq: 1, disposition: "committed" as const }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, async completeRun() {} }, + runnerInstanceId: "runner-first-turn", controlPlaneInstanceId: "control-first-turn", + })).rejects.toThrow("stop after turn.start"); + return { order, sent, getFirstTurnQueuedInput }; + } + + it("adds messages queued during preparation to the first turn, sealed just before turn.start", async () => { + const queued = "New messages from the user arrived while this run was starting.\n\n[Message 1 of 2]\nfirst\n\n[Message 2 of 2]\nsecond"; + const { order, sent, getFirstTurnQueuedInput } = await startFirstTurn(preparedInput(), queued); + expect(getFirstTurnQueuedInput).toHaveBeenCalledOnce(); + expect(order).toEqual(["claim", "turn.start"]); + const envelope = JSON.parse(sent[0]!) as { task: { prompt: string }; completionContract: { criteria: Array<{ source?: unknown }> } }; + expect(envelope.task.prompt).toBe(`${preparedInput().task.prompt}\n\n${queued}`); + // Appending input keeps the prompt-bound completion sources. + expect(envelope.completionContract.criteria[0]!.source).toMatchObject({ location: "task.prompt" }); + }); + + it("starts the first turn unchanged when nothing is queued", async () => { + const { sent } = await startFirstTurn(input, null); + expect((JSON.parse(sent[0]!) as { task: { prompt: string } }).task.prompt).toBe(input.task.prompt); + }); +}); diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 2e8bcb37074..6246376bee9 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -305,6 +305,12 @@ export interface ExecuteNativeSessionOptions { resumeInterruptedTurn?: boolean; /** Read bounded task history only after recovery requires a fresh conversation. */ getFreshSessionHandoff?: () => Promise; + /** + * Seals this run's first-turn input immediately before `turn.start` and + * returns user messages that arrived while the run was preparing. Repeated + * calls for the same run return the same messages. + */ + getFirstTurnQueuedInput?: () => Promise; /** * Steered messages the provider acknowledged but the model never read, read * when a restart continuation starts. The provider lost them with the @@ -3048,6 +3054,14 @@ export async function executeNativeSession( if ("workspace" in modelEnvelope && modelEnvelope.workspace && recoveredSnapshot.workingDirectory) { modelEnvelope.workspace.cwd = recoveredSnapshot.workingDirectory; } + if (!(dispositionOnlyRecovery && !effectFreeInitialAcpxTurn) && options.getFirstTurnQueuedInput) { + const queuedInput = await options.getFirstTurnQueuedInput(); + if (queuedInput && "task" in modelEnvelope) { + modelEnvelope.task.prompt = `${modelEnvelope.task.prompt}\n\n${queuedInput}`; + } else if (queuedInput && "events" in modelEnvelope) { + modelEnvelope.events = `${modelEnvelope.events}\n\n${queuedInput}`; + } + } await session.startTurn({ message: { role: "user", text: JSON.stringify(modelEnvelope) }, ...(recovered && modelEnvelope.schema === "paperclip.native-continuation.v1" diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index c1a7e3e2401..1aa33edb2c1 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -36,6 +36,7 @@ import { hasQueuedInteractionResponse } from "../services/queued-interaction-res import { remoteTerminationReceipt } from "../services/remote-execution-termination.js"; import { initializeRunIdentity, reconcileSteeredIdentity } from "../services/run-identity.js"; import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { claimFirstTurnQueuedInput } from "../services/native-runtime/first-turn-queued-input.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, @@ -901,6 +902,107 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }, ); + async function seedFirstTurnQueue(runtime: "preparing" | "native") { + const seeded = await seedQueue(); + // A user's follow-ups, as the comment route saves them behind that user's run. + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + await db.update(agentWakeupRequests).set({ payload: { ...wake!.payload, + _paperclipWakeContext: { ...(wake!.payload?._paperclipWakeContext as Record), + issueId: seeded.issueId, wakeReason: "issue_commented" } } }).where(eq(agentWakeupRequests.id, seeded.wakeId)); + await db.update(heartbeatRuns).set(runtime === "preparing" + ? { responsibleUserId: "queue-owner", runtimeMode: "legacy", runtimeModeResolvedAt: null, executionStage: "preparing", + runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } } } + : { responsibleUserId: "queue-owner", nativeIssueId: seeded.issueId }).where(eq(heartbeatRuns.id, seeded.runId)); + if (runtime === "native") { + await db.insert(nativeRunFinalizations).values({ companyId: seeded.companyId, + issueId: seeded.issueId, runId: seeded.runId, phase: "observed" }); + } + // Hold any successor in the ordinary queue without launching a provider. + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }) + .where(eq(agents.id, seeded.agentId)); + await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId, + status: "running", contextSnapshot: { issueId: randomUUID() } }); + return seeded; + } + + it.each(["preparing", "native"] as const)( + "keeps a %s run that has not started its first turn and gives it the queued messages", + async (runtime) => { + const seeded = await seedFirstTurnQueue(runtime); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + const response = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({ + queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision, + }); + expect(response.status, JSON.stringify(response.body)).toBe(200); + expect(response.body.executionWait).toMatchObject({ reason: "first_turn_pending" }); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(run!.status).toBe("running"); + expect(run!.errorCode).toBeNull(); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)); + expect(wake!.status).toBe("deferred_issue_execution"); + expect(wake!.payload?.queuedCommentInterrupt).toBeUndefined(); + + // Preparation finishes and the runner seals its first turn's input. + if (runtime === "preparing") { + await db.update(heartbeatRuns).set({ runtimeMode: "native", runtimeModeResolvedAt: new Date(), + nativeIssueId: seeded.issueId }).where(eq(heartbeatRuns.id, seeded.runId)); + } + const firstTurn = await claimFirstTurnQueuedInput(db, { companyId: seeded.companyId, runId: seeded.runId, + issueId: seeded.issueId, agentId: seeded.agentId }); + expect(firstTurn).toContain("First queued message"); + expect(firstTurn!.indexOf("First queued message")).toBeLessThan(firstTurn!.indexOf("Second queued message")); + // The taken messages leave the visible queue; nothing else can claim them. + const after = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(after.body.entries).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))) + .filter(row => row.status === "queued")).toEqual([]); + }, + ); + + it("still interrupts a preparing run when its queue holds another user's message", async () => { + const seeded = await seedFirstTurnQueue("native"); + // Ordinary comments from two users merge into one wake that keeps the + // first author as its requester; the first turn would take only that user's. + await db.update(issueComments).set({ authorUserId: "other-operator" }).where(eq(issueComments.id, seeded.commentIds[1]!)); + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + const response = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({ + queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision, + }); + expect(response.status, JSON.stringify(response.body)).toBe(200); + expect(response.body.executionWait?.reason).not.toBe("first_turn_pending"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(run!.status).toBe("cancelled"); + expect(run!.errorCode).toBe("operator_interrupted"); + }); + + it.each(["accepted turn", "sealed first turn"] as const)( + "still interrupts a run whose %s already started", + async (state) => { + const seeded = await seedFirstTurnQueue("native"); + if (state === "accepted turn") { + // A run saved by the previous release carries no seal; its accepted turn decides. + await appendHeartbeatRunEvent(db, { companyId: seeded.companyId, runId: seeded.runId, agentId: seeded.agentId, + eventType: "turn.accepted", stream: "system", level: "info", message: "turn accepted" }); + } else { + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: seeded.issueId, firstTurnQueuedInput: { + schema: "paperclip.first-turn-queued-input.v1", runId: seeded.runId, wakeupRequestIds: [], commentIds: [], + claimedAt: new Date().toISOString() } } }).where(eq(heartbeatRuns.id, seeded.runId)); + } + const client = app(seeded.companyId); + const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries).toHaveLength(2); + const response = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({ + queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision, + }); + expect(response.status, JSON.stringify(response.body)).toBe(200); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + expect(run!.status).toBe("cancelled"); + expect(run!.errorCode).toBe("operator_interrupted"); + }, + ); + it("does not accept interruption authority from an agent wake payload", async () => { const seeded = await seedQueue(); await db.update(agents).set({ adapterType: "claude_local", diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 4ab29038938..8b4d6ebbd0c 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -21,6 +21,7 @@ import { } from "../../../services/issue-queued-comment-queue.js"; import { logActivity as persistActivityLogRow, type ActivityPublication } from "../../../services/activity-log.js"; import { decideQueuedCommentWakeLookup } from "../domain/policy.js"; +import { firstTurnClaimedCommentIds } from "../../../services/native-runtime/first-turn-queued-input.js"; import { parseObject, readNonEmptyString } from "../domain/values.js"; import { QueuedCommentMutationError } from "../application/queued-comment-use-cases.js"; import type { @@ -56,6 +57,10 @@ export type QueuedCommentQueuePostgresAdapterDeps = { }; function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps, wakeAgentId: string): QueuedCommentQueueTransaction { + // Messages the active run took into its first turn leave the visible queue + // but stay on the wake until that turn is accepted, so a run that fails + // before then still leaves them queued. Queue edits keep them. + let heldByFirstTurn: string[] = []; return { async updateCommentBody({ issueId, commentId, body, updatedAt }) { await notifyStatusCardIssueChange(tx, companyId); @@ -74,9 +79,10 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async updateWakeQueuedCommentIds({ wakeId, payload, ids, updatedAt }) { + const kept = [...heldByFirstTurn.filter((id) => !ids.includes(id)), ...ids]; const row = await tx .update(agentWakeupRequests) - .set({ payload: withQueuedCommentIdsInWakePayload(payload, ids), updatedAt }) + .set({ payload: withQueuedCommentIdsInWakePayload(payload, kept), updatedAt }) .where(and(eq(agentWakeupRequests.id, wakeId), eq(agentWakeupRequests.companyId, companyId))) .returning() .then((rows) => rows[0]); @@ -105,6 +111,16 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async cancelWake({ wakeId, reason, now }) { + if (heldByFirstTurn.length > 0) { + const [held] = await tx.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.id, wakeId), eq(agentWakeupRequests.companyId, companyId))); + if (held) { + await tx.update(agentWakeupRequests) + .set({ payload: withQueuedCommentIdsInWakePayload(held.payload, heldByFirstTurn), updatedAt: now }) + .where(and(eq(agentWakeupRequests.id, wakeId), eq(agentWakeupRequests.companyId, companyId))); + return; + } + } await notifyStrandedAgentStatus(tx, { companyId, agentId: wakeAgentId }); await notifyChatInboundWakeNotices(tx, { companyId, issueId }); await tx @@ -139,7 +155,10 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async buildQueueSnapshot({ issue, actor, wake, state, queueRun, activeRun }): Promise { - const commentIds = queuedCommentIdsFromWakePayload(wake?.payload ?? null); + const wakeCommentIds = queuedCommentIdsFromWakePayload(wake?.payload ?? null); + const claimed = state === "deferred" ? firstTurnClaimedCommentIds(activeRun) : new Set(); + if (wake) heldByFirstTurn = wakeCommentIds.filter((id) => claimed.has(id)); + const commentIds = wakeCommentIds.filter((id) => !claimed.has(id)); const rows = commentIds.length > 0 ? await tx diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index d87ca4fb3a7..7c191590925 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -44,6 +44,7 @@ import { confirmNativeControllerStopped, NATIVE_CONTROLLER_HOST_UNVERIFIED } fro import { confirmOwnerGone, OWNER_UNPROVABLE_CAUSE } from "../services/recovery/owner-gone-attestation.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; import { retryNativeWorkspaceExport } from "../services/native-runtime/native-workspace-export-retry.js"; +import { deferredWakeJoinsFirstTurn, firstTurnClaimedCommentIds, firstTurnTakesAllQueuedComments, runAcceptsFirstTurnQueuedInput } from "../services/native-runtime/first-turn-queued-input.js"; import { reconstructNativeWorkspace, unreconciledNativeExportPredicate } from "../services/native-runtime/native-workspace-reconstruction.js"; import { reconstructWorkspaceSchema } from "@paperclipai/shared"; import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse, type QueuedInteractionResponse } from "../services/queued-interaction-response.js"; @@ -7637,12 +7638,14 @@ export function issueRoutes( ? queueState.response : await readQueuedInteractionResponse(input.executor as Db, input.issue.companyId, input.issue.id, wake.payload) : null; - const comments = await queueCommentsForWake( + // Messages the active run took into its first turn are no longer queued. + const claimedByFirstTurn = queueState?.state === "deferred" ? firstTurnClaimedCommentIds(input.activeRun) : new Set(); + const comments = (await queueCommentsForWake( input.executor, input.issue.id, wake, response, - ); + )).filter((comment) => !claimedByFirstTurn.has(comment.id)); const assignedAgent = input.issue.assigneeAgentId ? await input.executor .select({ adapterType: agents.adapterType }) @@ -17266,7 +17269,7 @@ export function issueRoutes( } } const actor = getActorInfo(req); - const runToInterrupt = await db.transaction(async (tx) => { + const interruptPlan = await db.transaction(async (tx): Promise<{ firstTurnRunId: string } | { runToInterrupt: string | null }> => { const locked = await lockQueuedCommentState({ tx, issue, actor, queueId: req.body.queueId, targetRunId: req.body.targetRunId ?? undefined, allowStoppedTarget: true, @@ -17283,6 +17286,16 @@ export function issueRoutes( if (locked.activeRun?.status === "running" && locked.activeRun.id !== req.body.targetRunId) { throw conflict("The queued message targets a stale run", { code: "queued_comment_stale_target" }); } + // A run that has not started its first provider turn takes these + // messages into that turn. Stopping it would only discard its setup. + if (locked.activeRun && !issue.conversationAgentId && issue.executionRunId === locked.activeRun.id && + deferredWakeJoinsFirstTurn(locked.wake, locked.activeRun) && + await firstTurnTakesAllQueuedComments(tx, { companyId: issue.companyId, issueId: issue.id, + responsibleUserId: locked.activeRun.responsibleUserId, + commentIds: queuedCommentIdsFromWakePayload(locked.wake.payload) }) && + await runAcceptsFirstTurnQueuedInput(tx, locked.activeRun)) { + return { firstTurnRunId: locked.activeRun.id }; + } // The click is durable fresh user intent, including when the message // predates a failed run's stop. Keep its content and original attribution. await notifyDeferredExecutionWork(tx, { companyId: issue.companyId, requestId: locked.wake.id }); @@ -17294,8 +17307,10 @@ export function issueRoutes( } }, updatedAt: new Date(), }).where(eq(agentWakeupRequests.id, locked.wake.id)); - return locked.activeRun?.status === "running" ? locked.activeRun.id : null; + return { runToInterrupt: locked.activeRun?.status === "running" ? locked.activeRun.id : null }; }); + const firstTurnRunId = "firstTurnRunId" in interruptPlan ? interruptPlan.firstTurnRunId : null; + const runToInterrupt = "runToInterrupt" in interruptPlan ? interruptPlan.runToInterrupt : null; // Never hold the issue lock while joining the adapter. Queue edits and // discards stay authoritative until the dispatcher claims the successor. const options = operatorInterruptCancelOptions({ issueId: issue.id, actor }); @@ -17304,18 +17319,23 @@ export function issueRoutes( suppressImmediateRecovery: true, resultJson: { ...options.resultJson, queuedCommentInterruptQueueId: req.body.queueId }, }); - await heartbeat.resumeQueuedCommentInterrupt(issue.companyId, req.body.queueId, { retryCleanup: true }); + if (!firstTurnRunId) await heartbeat.resumeQueuedCommentInterrupt(issue.companyId, req.body.queueId, { retryCleanup: true }); await logActivity(db, { companyId: issue.companyId, actorType: actor.actorType, actorId: actor.actorId, agentId: actor.agentId, runId: actor.runId, agentApiKeyId: actor.agentApiKeyId, action: "issue.queued_comments_interrupted", entityType: "issue", entityId: issue.id, - details: { queueId: req.body.queueId, targetRunId: req.body.targetRunId }, + details: { queueId: req.body.queueId, targetRunId: req.body.targetRunId, + ...(firstTurnRunId ? { deliveredWithFirstTurnOfRunId: firstTurnRunId } : {}) }, }); const currentIssue = await svc.getById(issue.id); const queue = await buildQueuedCommentQueue({ executor: db, issue: currentIssue ?? issue, activeRun: await resolveActiveIssueRun(currentIssue ?? issue), actor, }); + if (firstTurnRunId && queue.state === "deferred" && queue.entries.length > 0) { + queue.executionWait = { reason: "first_turn_pending", + message: "The agent is still starting. These messages will be included in its first turn." }; + } res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue)); }, ); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index c42208d44fe..7ea79bd92b0 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -34,6 +34,7 @@ import { activeIssueInteractionCondition } from "./issue-question-context.js"; import { historicalAdapterType, isConversationAdapter } from "./conversation-continuation.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; +import { firstTurnDeliveredCommentIds } from "./queued-comment-delivery.js"; type Run = Pick valid.has(id)); + // An accepted first turn that took a queued message consumed it too. + const firstTurn = await firstTurnDeliveredCommentIds(db, { companyId, issueId, agentId, commentIds: [...valid] }); + return commentIds.filter(id => valid.has(id) && !firstTurn.has(id)); } /** Called under the issue lock, in the transaction that creates the new turn. diff --git a/server/src/services/heartbeat/run-state.ts b/server/src/services/heartbeat/run-state.ts index 4584ea7fd8a..3150bff2e5e 100644 --- a/server/src/services/heartbeat/run-state.ts +++ b/server/src/services/heartbeat/run-state.ts @@ -372,7 +372,10 @@ const heartbeatRunIssueControlColumns = { runtimeModeResolvedAt: heartbeatRuns.runtimeModeResolvedAt, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, contextSnapshot: sql | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' - then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId') else null end`, + then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId') + || case when ${heartbeatRuns.contextSnapshot} ? 'firstTurnQueuedInput' + then jsonb_build_object('firstTurnQueuedInput', ${heartbeatRuns.contextSnapshot}->'firstTurnQueuedInput') else '{}'::jsonb end + else null end`, runnerProfileJson: sql | null>`case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then jsonb_build_object('adapterDispatch', jsonb_build_object('adapterType', ${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->'adapterType')) else null end`, diff --git a/server/src/services/native-runtime/first-turn-queued-input.test.ts b/server/src/services/native-runtime/first-turn-queued-input.test.ts new file mode 100644 index 00000000000..b86b869b85c --- /dev/null +++ b/server/src/services/native-runtime/first-turn-queued-input.test.ts @@ -0,0 +1,189 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { + agents, agentWakeupRequests, assets, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issueAttachments, + issueComments, issues, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; +import { undeliveredLegacyUserCommentIds } from "../explicit-native-continuation.js"; +import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; +import { + claimFirstTurnQueuedInput, firstTurnClaimedCommentIds, firstTurnTakesAllQueuedComments, runAcceptsFirstTurnQueuedInput, +} from "./first-turn-queued-input.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const USER = "first-turn-user"; + +describe.skipIf(!support.supported)("first-turn queued input", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-first-turn-queued-input-"); + db = createDb(database.connectionString); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function seed() { + const [company] = await db.insert(companies).values({ name: "First turn", issuePrefix: `F${randomUUID().slice(0, 4)}` }).returning(); + const [agent] = await db.insert(agents).values({ companyId: company!.id, name: "Codie", adapterType: "paperclip_runner" }).returning(); + const issueId = randomUUID(); + const runId = await startRun(company!.id, agent!.id, issueId); + await db.insert(issues).values({ id: issueId, companyId: company!.id, title: "Fix the reports", status: "in_progress", + assigneeAgentId: agent!.id, executionRunId: runId }); + return { companyId: company!.id, agentId: agent!.id, issueId, runId }; + } + + async function startRun(companyId: string, agentId: string, issueId: string) { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running", runtimeMode: "native", + runtimeModeResolvedAt: new Date(), nativeIssueId: issueId, responsibleUserId: USER, startedAt: new Date(), + contextSnapshot: { issueId } }).returning(); + return run!.id; + } + + /** A user's comment saved behind the active run, in the shape the previous release writes. */ + async function queueMessage(f: Awaited>, body: string) { + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: USER, body }).returning(); + const [existing] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.status, "deferred_issue_execution")) + .then(rows => rows.filter(row => row.companyId === f.companyId)); + const ids = [...((existing?.payload?._paperclipWakeContext as { wakeCommentIds?: string[] } | undefined)?.wakeCommentIds ?? []), comment!.id]; + const payload = { issueId: f.issueId, commentId: comment!.id, _paperclipWakeContext: { issueId: f.issueId, taskId: f.issueId, + wakeReason: "issue_commented", source: "issue.comment", wakeCommentId: comment!.id, wakeCommentIds: ids } }; + if (existing) await db.update(agentWakeupRequests).set({ payload }).where(eq(agentWakeupRequests.id, existing.id)); + else await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, source: "automation", + triggerDetail: "system", reason: "issue_execution_deferred", status: "deferred_issue_execution", + requestedByActorType: "user", requestedByActorId: USER, payload }); + return comment!.id; + } + + const binding = (f: { companyId: string; issueId: string; agentId: string }, runId: string) => + ({ companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, runId }); + const accept = (f: { companyId: string; agentId: string }, runId: string) => appendHeartbeatRunEvent(db, { + companyId: f.companyId, runId, agentId: f.agentId, eventType: "turn.accepted", stream: "system", level: "info", message: "turn accepted" }); + const readRun = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + + it("gives a message that arrived during preparation to the first turn, without a stop", async () => { + const f = await seed(); + const message = await queueMessage(f, "oh i mean each of the ones that appeared within the last 24 hours"); + expect(await runAcceptsFirstTurnQueuedInput(db, await readRun(f.runId))).toBe(true); + const startedAt = performance.now(); + const firstTurn = await claimFirstTurnQueuedInput(db, binding(f, f.runId)); + const sealMs = performance.now() - startedAt; + console.info(`first-turn queued input seal took ${sealMs.toFixed(1)}ms before turn.start`); + expect(firstTurn).toContain("oh i mean each of the ones that appeared within the last 24 hours"); + expect(firstTurn).toContain(`comment ${message}`); + const run = await readRun(f.runId); + expect(run.status).toBe("running"); + expect([...firstTurnClaimedCommentIds(run)]).toEqual([message]); + // Sealed: a later message waits for the turn boundary and Interrupt applies. + expect(await runAcceptsFirstTurnQueuedInput(db, run)).toBe(false); + const later = await queueMessage(f, "one more thing"); + expect(await claimFirstTurnQueuedInput(db, binding(f, f.runId))).not.toContain("one more thing"); + expect(firstTurnClaimedCommentIds(await readRun(f.runId)).has(later)).toBe(false); + expect(sealMs).toBeLessThan(1_000); + }); + + it("delivers two messages from preparation in their queued order", async () => { + const f = await seed(); + await queueMessage(f, "FIRST follow-up"); + await queueMessage(f, "SECOND follow-up"); + const firstTurn = await claimFirstTurnQueuedInput(db, binding(f, f.runId)); + expect(firstTurn).toMatch(/Message 1 of 2[^\n]*\nFIRST follow-up[\s\S]*Message 2 of 2[^\n]*\nSECOND follow-up/); + }); + + it("keeps the message queued when the run fails before turn.start, then delivers it exactly once", async () => { + const f = await seed(); + const message = await queueMessage(f, "please include the staging run"); + expect(await claimFirstTurnQueuedInput(db, binding(f, f.runId))).toContain("please include the staging run"); + // A restart before the provider accepted the turn repeats the same input. + expect(await claimFirstTurnQueuedInput(db, binding(f, f.runId))).toContain("please include the staging run"); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + // Not consumed: release promotion and Send both still deliver it. + const ids = { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, commentIds: [message] }; + expect([...await deliveredQueuedCommentIds(db, ids)]).toEqual([]); + expect(await undeliveredLegacyUserCommentIds(db, f.companyId, f.issueId, f.agentId, [message])).toEqual([message]); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(wake!.status).toBe("deferred_issue_execution"); + + // The next run takes it into its own first turn; the provider accepts. + const nextRunId = await startRun(f.companyId, f.agentId, f.issueId); + await db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, f.issueId)); + expect(await claimFirstTurnQueuedInput(db, binding(f, nextRunId))).toContain("please include the staging run"); + await accept(f, nextRunId); + expect([...await deliveredQueuedCommentIds(db, ids)]).toEqual([message]); + expect(await undeliveredLegacyUserCommentIds(db, f.companyId, f.issueId, f.agentId, [message])).toEqual([]); + + // Exactly once: a later run, and a retry that copied the context, take nothing. + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, nextRunId)); + const thirdRunId = await startRun(f.companyId, f.agentId, f.issueId); + await db.update(heartbeatRuns).set({ contextSnapshot: (await readRun(nextRunId)).contextSnapshot }) + .where(eq(heartbeatRuns.id, thirdRunId)); + await db.update(issues).set({ executionRunId: thirdRunId }).where(eq(issues.id, f.issueId)); + expect(await claimFirstTurnQueuedInput(db, binding(f, thirdRunId))).toBeNull(); + }); + + it("counts a message as delivered once the provider accepted the turn, even if the run later fails", async () => { + const f = await seed(); + const message = await queueMessage(f, "ship it"); + await claimFirstTurnQueuedInput(db, binding(f, f.runId)); + await accept(f, f.runId); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + expect([...await deliveredQueuedCommentIds(db, { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + commentIds: [message] })]).toEqual([message]); + }); + + it("treats a run saved by the previous release as started once its provider accepted a turn", async () => { + const f = await seed(); + await queueMessage(f, "mid-turn message"); + await accept(f, f.runId); + expect(await runAcceptsFirstTurnQueuedInput(db, await readRun(f.runId))).toBe(false); + }); + + it("leaves another user's message and a run without a responsible user to their own delivery", async () => { + const f = await seed(); + await queueMessage(f, "mine"); + await db.update(agentWakeupRequests).set({ requestedByActorId: "someone-else" }).where(eq(agentWakeupRequests.companyId, f.companyId)); + expect(await claimFirstTurnQueuedInput(db, binding(f, f.runId))).toBeNull(); + }); + it("seals on a one-connection pool without waiting for a second connection", async () => { + const f = await seed(); + await queueMessage(f, "secret-free follow-up"); + const single = createDb(database.connectionString, { maxConnections: 1 }); + try { + const sealed = await Promise.race([ + claimFirstTurnQueuedInput(single, binding(f, f.runId)), + new Promise<"stuck">((resolve) => setTimeout(() => resolve("stuck"), 5_000)), + ]); + expect(sealed).toContain("secret-free follow-up"); + } finally { + await single.$client.end({ timeout: 1 }); + } + }); + + it("leaves a message with files on its own delivery path so the files reach the agent", async () => { + const f = await seed(); + await queueMessage(f, "text only"); + const withFile = await queueMessage(f, "see the attached screenshot"); + const [asset] = await db.insert(assets).values({ companyId: f.companyId, provider: "local_disk", + objectKey: `${f.companyId}/screenshot`, originalFilename: "screenshot.png", contentType: "image/png", + byteSize: 1, sha256: "0".repeat(64) }).returning(); + await db.insert(issueAttachments).values({ companyId: f.companyId, issueId: f.issueId, assetId: asset!.id, + issueCommentId: withFile }); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId)); + const queuedIds = (wake!.payload?._paperclipWakeContext as { wakeCommentIds: string[] }).wakeCommentIds; + // Interrupt keeps its meaning for this queue: the first turn cannot take all of it. + expect(await firstTurnTakesAllQueuedComments(db, { companyId: f.companyId, issueId: f.issueId, responsibleUserId: USER, + commentIds: queuedIds })).toBe(false); + const firstTurn = await claimFirstTurnQueuedInput(db, binding(f, f.runId)); + expect(firstTurn).toContain("text only"); + expect(firstTurn).not.toContain("see the attached screenshot"); + const run = await readRun(f.runId); + expect(firstTurnClaimedCommentIds(run).has(withFile)).toBe(false); + await accept(f, f.runId); + // Still owed after the turn: release promotion delivers it with its file. + expect([...await deliveredQueuedCommentIds(db, { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, + commentIds: [withFile] })]).toEqual([]); + }); +}); diff --git a/server/src/services/native-runtime/first-turn-queued-input.ts b/server/src/services/native-runtime/first-turn-queued-input.ts new file mode 100644 index 00000000000..216b173c276 --- /dev/null +++ b/server/src/services/native-runtime/first-turn-queued-input.ts @@ -0,0 +1,269 @@ +import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { agentWakeupRequests, heartbeatRunEvents, heartbeatRuns, issueAttachments, issueComments, issues } from "@paperclipai/db"; +import type { SourceTrustMetadata } from "@paperclipai/shared"; + +import { queuedCommentIdsFromWakePayload } from "../issue-queued-comment-queue.js"; +import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; +import { createRunSecretRedactionRegistry } from "../run-secret-redaction.js"; +import { sanitizeQuarantinedCommentForHigherTrust } from "../source-trust.js"; +import { PROVIDER_TURN_ACCEPTED_EVENT_TYPES } from "./provider-attempt-outcome.js"; + +/** + * Queued input for a run's first provider turn. + * + * A user's message that arrives while a run is still preparing waits as a + * deferred wake behind that run. Interrupting the run for it throws away the + * whole preparation. Instead, the run takes the waiting messages into the + * turn it is about to start: the runner calls `claimFirstTurnQueuedInput` + * immediately before `turn.start`, and the claim and the first-turn input are + * sealed in one transaction on the run row. + * + * The claim does not remove the messages from their deferred wake. It records + * a receipt in the run's context. A message counts as delivered only after the + * provider accepted that run's turn (`deliveredQueuedCommentIds`). If the run + * ends before that, the receipt proves nothing and the deferred wake still + * holds the message, so the next run delivers it exactly once. A restart + * before the turn was accepted repeats the same claim from the receipt. + */ +export const FIRST_TURN_QUEUED_INPUT_KEY = "firstTurnQueuedInput"; +export const FIRST_TURN_QUEUED_INPUT_SCHEMA = "paperclip.first-turn-queued-input.v1"; + +const DEFERRED_WAKE_STATUS = "deferred_issue_execution"; +const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext"; +const COMMENT_WAKE_REASONS = new Set(["issue_commented", "issue_reopened_via_comment", "issue_comment_mentioned"]); +const MAX_COMMENT_CHARS = 16_000; +const MAX_TOTAL_CHARS = 48_000; +const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +export type FirstTurnQueuedInputReceipt = { + schema: typeof FIRST_TURN_QUEUED_INPUT_SCHEMA; + runId: string; + wakeupRequestIds: string[]; + commentIds: string[]; + claimedAt: string; +}; + +function record(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record : {}; +} + +function nonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +function stringIds(value: unknown): string[] { + return Array.isArray(value) ? [...new Set(value.filter((id): id is string => typeof id === "string" && UUID.test(id)))] : []; +} + +/** + * The receipt this exact run sealed. A retry copies its predecessor's context, + * so a receipt naming another run is inert. + */ +export function readFirstTurnQueuedInputReceipt(run: { id: string; contextSnapshot?: unknown }): FirstTurnQueuedInputReceipt | null { + const receipt = record(record(run.contextSnapshot)[FIRST_TURN_QUEUED_INPUT_KEY]); + if (receipt.schema !== FIRST_TURN_QUEUED_INPUT_SCHEMA || receipt.runId !== run.id) return null; + const claimedAt = nonEmptyString(receipt.claimedAt); + if (!claimedAt) return null; + return { + schema: FIRST_TURN_QUEUED_INPUT_SCHEMA, + runId: run.id, + wakeupRequestIds: stringIds(receipt.wakeupRequestIds), + commentIds: stringIds(receipt.commentIds), + claimedAt, + }; +} + +/** Message ids a live run already took into its first turn; queue views hide them. */ +export function firstTurnClaimedCommentIds(run: { id: string; status?: string | null; contextSnapshot?: unknown } | null | undefined): Set { + if (!run || (run.status !== undefined && run.status !== "running")) return new Set(); + return new Set(readFirstTurnQueuedInputReceipt(run)?.commentIds ?? []); +} + +/** + * Whether a deferred wake is plain user input that the run's first turn may + * take. Interaction answers, interrupt intents, independent continuations and + * chat-conversation wakes keep their own delivery contracts. Only the run's + * responsible user's own messages qualify, so taking them cannot change the + * run's execution authority. + */ +export function deferredWakeJoinsFirstTurn( + wake: { agentId: string; status: string; requestedByActorType: string | null; requestedByActorId: string | null; reason: string | null; payload: unknown }, + run: { agentId: string; responsibleUserId: string | null }, +): boolean { + const payload = record(wake.payload); + const context = record(payload[DEFERRED_WAKE_CONTEXT_KEY]); + const reason = nonEmptyString(context.wakeReason) ?? nonEmptyString(wake.reason); + return wake.status === DEFERRED_WAKE_STATUS && + wake.agentId === run.agentId && + wake.requestedByActorType === "user" && + Boolean(run.responsibleUserId) && wake.requestedByActorId === run.responsibleUserId && + reason !== null && COMMENT_WAKE_REASONS.has(reason) && + !payload.interactionId && !context.interactionId && !context.interactionStatus && + !payload.queuedCommentInterrupt && context.resumeIntent !== true && + !nonEmptyString(context.chatConversationId) && !nonEmptyString(payload.externalChatProvider) && + queuedCommentIdsFromWakePayload(payload).length > 0; +} + +type QueuedMessage = { id: string; body: string; createdAt: Date }; + +export function renderFirstTurnQueuedInput(messages: readonly QueuedMessage[]): string | null { + if (messages.length === 0) return null; + let remaining = MAX_TOTAL_CHARS; + const sections = messages.map((message, index) => { + const limit = Math.max(0, Math.min(MAX_COMMENT_CHARS, remaining)); + const truncated = message.body.length > limit; + const body = truncated ? message.body.slice(0, limit) : message.body; + remaining -= body.length; + return [ + `[Message ${index + 1} of ${messages.length} · comment ${message.id} · ${message.createdAt.toISOString()}]`, + body, + ...(truncated ? [`(Truncated here. Read comment ${message.id} on the task for the rest.)`] : []), + ].join("\n"); + }); + return [ + "New messages from the user arrived while this run was starting. They were not in the request above. Handle each of them in this turn, oldest first, together with that request:", + ...sections, + ].join("\n\n"); +} + +/** + * The one rule for which queued comment a first turn may take: a live, plain + * comment written directly by the run's responsible user, with no files. The + * prompt carries text only; a message with files keeps its own delivery, which + * stages those files for the turn that reads it. + */ +function firstTurnEligibleComment(input: { companyId: string; issueId: string; responsibleUserId: string }) { + return and( + eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId), + eq(issueComments.authorUserId, input.responsibleUserId), + isNull(issueComments.authorAgentId), isNull(issueComments.createdByRunId), isNull(issueComments.deletedAt), + sql`not exists (select 1 from ${issueAttachments} where ${issueAttachments.companyId} = ${input.companyId} + and ${issueAttachments.issueCommentId} = ${issueComments.id})`); +} + +async function readQueuedMessages( + tx: Db, + input: { companyId: string; issueId: string; agentId: string; responsibleUserId: string; commentIds: string[]; excludeDelivered: boolean }, +): Promise { + if (input.commentIds.length === 0) return []; + const rows = await tx.select({ + id: issueComments.id, body: issueComments.body, createdAt: issueComments.createdAt, + sourceTrust: issueComments.sourceTrust, + }).from(issueComments).where(and(inArray(issueComments.id, input.commentIds), firstTurnEligibleComment(input))); + const delivered = input.excludeDelivered + ? await deliveredQueuedCommentIds(tx, { companyId: input.companyId, issueId: input.issueId, agentId: input.agentId, + commentIds: rows.map(row => row.id) }) + : new Set(); + const byId = new Map(rows.map(row => [row.id, row])); + return input.commentIds.flatMap((id) => { + const row = byId.get(id); + if (!row || delivered.has(id)) return []; + const safe = sanitizeQuarantinedCommentForHigherTrust({ ...row, sourceTrust: row.sourceTrust as SourceTrustMetadata | null }); + return [{ id, body: safe.body, createdAt: row.createdAt }]; + }); +} + +/** + * Seals the first provider turn's input. Returns the queued messages to add to + * that turn, or null. Called once per turn start, immediately before + * `turn.start`; repeated calls for the same run return the same messages. + */ +export async function claimFirstTurnQueuedInput( + db: Db, + binding: { companyId: string; runId: string; issueId: string; agentId: string }, +): Promise { + return db.transaction(async (tx) => { + // Redact on this transaction's own connection: a second pool connection + // while the seal holds its locks can wait forever on a one-connection pool. + const render = async (messages: QueuedMessage[]) => messages.length === 0 ? null : renderFirstTurnQueuedInput( + await createRunSecretRedactionRegistry(tx as unknown as Db).redactForIssue(binding.companyId, binding.issueId, messages)); + // Lock in the queue routes' order (task, queue, run) so a concurrent + // Interrupt and this seal serialize instead of deadlocking. + const [issue] = await tx.select({ conversationAgentId: issues.conversationAgentId, executionRunId: issues.executionRunId }) + .from(issues).where(and(eq(issues.id, binding.issueId), eq(issues.companyId, binding.companyId))).for("update").limit(1); + const wakes = !issue ? [] : await tx.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, binding.companyId), eq(agentWakeupRequests.agentId, binding.agentId), + eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS), + sql`${agentWakeupRequests.payload} ->> 'issueId' = ${binding.issueId}`, + )).orderBy(asc(agentWakeupRequests.requestedAt), asc(agentWakeupRequests.id)).for("update"); + const [run] = await tx.select({ + id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, status: heartbeatRuns.status, + runtimeMode: heartbeatRuns.runtimeMode, responsibleUserId: heartbeatRuns.responsibleUserId, + contextSnapshot: heartbeatRuns.contextSnapshot, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId), + eq(heartbeatRuns.agentId, binding.agentId))).for("update").limit(1); + if (!run || run.status !== "running" || run.runtimeMode !== "native" || !run.responsibleUserId) return null; + const receipt = readFirstTurnQueuedInputReceipt(run); + // A restart before the provider accepted the turn repeats the sealed input. + if (receipt) { + return render(await readQueuedMessages(tx as unknown as Db, { ...binding, responsibleUserId: run.responsibleUserId, + commentIds: receipt.commentIds, excludeDelivered: false })); + } + // Only the run that holds the task takes its queue. Conversations keep + // their own turn-boundary contract. + const joining = !issue || issue.conversationAgentId || issue.executionRunId !== run.id + ? [] : wakes.filter(wake => deferredWakeJoinsFirstTurn(wake, run)); + const commentIds = [...new Set(joining.flatMap(wake => queuedCommentIdsFromWakePayload(wake.payload)))]; + const taken = await readQueuedMessages(tx as unknown as Db, { ...binding, responsibleUserId: run.responsibleUserId, + commentIds, excludeDelivered: true }); + // Render before sealing: once the receipt commits, nothing can fail + // between it and the turn input that the receipt describes. + const text = await render(taken); + // Seal even when nothing joined: later messages wait for the turn boundary. + const sealed: FirstTurnQueuedInputReceipt = { + schema: FIRST_TURN_QUEUED_INPUT_SCHEMA, + runId: run.id, + wakeupRequestIds: joining.filter(wake => queuedCommentIdsFromWakePayload(wake.payload) + .some(id => taken.some(message => message.id === id))).map(wake => wake.id), + commentIds: taken.map(message => message.id), + claimedAt: new Date().toISOString(), + }; + await tx.update(heartbeatRuns).set({ + contextSnapshot: sql`coalesce(${heartbeatRuns.contextSnapshot}, '{}'::jsonb) || ${JSON.stringify({ [FIRST_TURN_QUEUED_INPUT_KEY]: sealed })}::jsonb`, + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + return text; + }); +} + +/** + * True while a running native run can still take queued input into its first + * turn: it has not sealed that input and no provider turn was accepted. A run + * saved by a release without the seal counts as started once its provider + * accepted a turn, so a mid-turn Interrupt keeps its old meaning. + */ +export async function runAcceptsFirstTurnQueuedInput( + tx: Pick, + run: { id: string; companyId: string; status: string; runtimeMode: string | null; runtimeModeResolvedAt?: Date | null; + runnerProfileJson?: unknown; contextSnapshot?: unknown }, +): Promise { + if (run.status !== "running") return false; + const preparingNative = run.runtimeMode === "legacy" && !run.runtimeModeResolvedAt && + record(record(run.runnerProfileJson).adapterDispatch).adapterType === "paperclip_runner"; + if (run.runtimeMode !== "native" && !preparingNative) return false; + if (readFirstTurnQueuedInputReceipt(run)) return false; + const [accepted] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [...PROVIDER_TURN_ACCEPTED_EVENT_TYPES]))).limit(1); + return !accepted; +} + +/** + * True when the first turn would take every pending message in this queue, + * under the same rule as the claim. Another user's message, an agent's, or one + * with files keeps its own delivery, so Interrupt must still apply. + */ +export async function firstTurnTakesAllQueuedComments( + tx: Pick, + input: { companyId: string; issueId: string; responsibleUserId: string | null; commentIds: string[] }, +): Promise { + if (input.commentIds.length === 0 || !input.responsibleUserId) return false; + const rows = await tx.select({ + id: issueComments.id, + eligible: sql`coalesce((${firstTurnEligibleComment({ ...input, responsibleUserId: input.responsibleUserId })}), false)`, + }).from(issueComments).where(and(eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId), + inArray(issueComments.id, input.commentIds), isNull(issueComments.deletedAt))); + return rows.length > 0 && rows.every(row => row.eligible === true); +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c2d683a6cbe..a7b8873c3a0 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -253,6 +253,7 @@ import { NATIVE_CURRENT_WAKE_COMMENTS_UNREAD, resolveCurrentWakeCommentsBinding, } from "./current-wake-comments.js"; +import { claimFirstTurnQueuedInput } from "./first-turn-queued-input.js"; import { renderNativeRunnerStagedAttachmentPrompt, stageNativeRunnerWakeAttachments, @@ -11062,6 +11063,12 @@ async function executePaperclipNativeSessionWithinScope( executeNativeSession({ resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner", getFreshSessionHandoff: input.getFreshSessionHandoff, + // A failed seal commits nothing: the messages stay queued for the + // turn boundary instead of failing this run before its turn. + getFirstTurnQueuedInput: () => claimFirstTurnQueuedInput(input.db, input.execution.binding).catch((error) => { + logger.warn({ err: error, runId: input.execution.binding.runId }, "first-turn queued input was not sealed"); + return null; + }), getUnreadSteeredMessages: () => readUnreadSteeredMessages(input.db, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, diff --git a/server/src/services/queued-comment-delivery.ts b/server/src/services/queued-comment-delivery.ts index 557a8b0b283..6ef16eab4fd 100644 --- a/server/src/services/queued-comment-delivery.ts +++ b/server/src/services/queued-comment-delivery.ts @@ -1,5 +1,5 @@ import { and, eq, inArray, sql } from "drizzle-orm"; -import { heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; +import { heartbeatRunEvents, heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; /** * Per-comment consumption (work-intent outbox, invariant 4). A succeeded run @@ -11,6 +11,11 @@ import { heartbeatRuns, issueComments, type Db } from "@paperclipai/db"; * counts: its context can be refreshed after the fact, and an edit made after * it started is still unread. Every path that replays a saved message reads * this before delivering it. + * + * A run that took the message into its first provider turn + * (`firstTurnQueuedInput`, sealed by that run before `turn.start`) consumed + * it once the provider accepted that turn, whatever the run's later outcome. + * Before acceptance the receipt proves nothing and the message stays queued. */ export async function deliveredQueuedCommentIds(tx: Pick, input: { companyId: string; issueId: string; agentId: string; commentIds: string[]; @@ -20,7 +25,7 @@ export async function deliveredQueuedCommentIds(tx: Pick, input: { const rows = await tx.select({ id: issueComments.id }).from(issueComments).where(and( eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, input.commentIds), - sql`exists (select 1 from ${heartbeatRuns} delivered_run + sql`(exists (select 1 from ${heartbeatRuns} delivered_run where delivered_run.company_id = ${companyId} and delivered_run.agent_id = ${input.agentId} and delivered_run.status = 'succeeded' and (delivered_run.native_issue_id = ${issueId} or delivered_run.context_snapshot->>'issueId' = ${issueId}) @@ -30,7 +35,36 @@ export async function deliveredQueuedCommentIds(tx: Pick, input: { or (delivered_run.retry_of_run_id is not null and jsonb_path_exists(delivered_run.context_snapshot, '$.executionContinuation.messages[*] ? (@.id == $id)', jsonb_build_object('id', ${issueComments.id}::text)) or jsonb_path_exists(delivered_run.context_snapshot, '$.paperclipWake.messages[*] ? (@.id == $id)', - jsonb_build_object('id', ${issueComments.id}::text)))))`, + jsonb_build_object('id', ${issueComments.id}::text))))) + or ${firstTurnDeliveredSql(companyId, issueId, input.agentId)})`, )); return new Set(rows.map(row => row.id)); } + +/** + * The comment row was taken into an accepted first provider turn of a run of + * this agent on this task, and not edited after that run sealed its input. + */ +function firstTurnDeliveredSql(companyId: string, issueId: string, agentId: string) { + return sql`exists (select 1 from ${heartbeatRuns} first_turn_run + where first_turn_run.company_id = ${companyId} and first_turn_run.agent_id = ${agentId} + and (first_turn_run.native_issue_id = ${issueId} or first_turn_run.context_snapshot->>'issueId' = ${issueId}) + and first_turn_run.context_snapshot->'firstTurnQueuedInput'->>'schema' = 'paperclip.first-turn-queued-input.v1' + and first_turn_run.context_snapshot->'firstTurnQueuedInput'->>'runId' = first_turn_run.id::text + and first_turn_run.context_snapshot->'firstTurnQueuedInput'->'commentIds' ? ${issueComments.id}::text + and (first_turn_run.context_snapshot->'firstTurnQueuedInput'->>'claimedAt')::timestamptz >= date_trunc('milliseconds', ${issueComments.updatedAt}) + and exists (select 1 from ${heartbeatRunEvents} accepted + where accepted.company_id = ${companyId} and accepted.run_id = first_turn_run.id + and accepted.event_type in ('turn.accepted', 'turn.started')))`; +} + +/** Ids that an accepted first provider turn already received (see `deliveredQueuedCommentIds`). */ +export async function firstTurnDeliveredCommentIds(tx: Pick, input: { + companyId: string; issueId: string; agentId: string; commentIds: string[]; +}): Promise> { + if (input.commentIds.length === 0) return new Set(); + const rows = await tx.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, input.companyId), eq(issueComments.issueId, input.issueId), + inArray(issueComments.id, input.commentIds), firstTurnDeliveredSql(input.companyId, input.issueId, input.agentId))); + return new Set(rows.map(row => row.id)); +}