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)); +}