diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a9119cabd7b..21eb4e09409 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -496,6 +496,8 @@ Normal issue-thread feedback can still wake the current assignee according to th Restart recovery revisits accepted deferred messages through the latest issue run, including historical mention queues for another agent. Ordinary messages for a former owner retain their assignment handoff. Native sources require their matching committed finalization, settled environment leases, and released finalizer and copyback ownership before recovery clears old issue claims. A successful retained warm release can satisfy the lease check. Released local bookkeeping without a provider allocation and a matching SSH bookkeeping release need no provider cleanup receipt. Plugin releases record their successful acknowledgement. Pending or failed cleanup and durable owners still hold work. A successful fenced cleanup completion clears its in-flight claim in the same database write, so a restart before retry bookkeeping cannot strand saved input. The normal release checks preserve cleanup holds, pauses, and operator Stops. Recovery checks for a competing turn under the issue lock before it promotes saved input. +A paused recipient does not fail an accepted deferred request. Release and periodic recovery preserve its pending receipt and may process an eligible request behind it. Resuming the agent allows normal admission to reconsider the saved input. Missing, foreign, terminated, and unapproved recipients retain their rejection rules. Explicit task holds and operator Stops still prevent unauthorized execution. This does not replay historical failed requests. + Accepted agent feedback must survive a child changing to `done` before its active run exits. Deferred wake promotion may reopen that completed child only for its current assignee, with explicit agent resume intent and live tracked comments from another author. Claim promotion before reopening. Cancelled tasks, deleted comments, self-authored comments, empty continuations, and agent continuations without explicit intent keep their existing suppression rules. Normal pause, ownership, authorization, and budget gates still apply. Pause and tree-control previews should make the same distinction visible. They should report whether the affected subtree contains live running work, queued wakes, agent-owned work, or only human-owned/static issues, so a pause after a handoff does not look like it interrupted agent execution when no agent execution path existed. diff --git a/maps/agent-access-map/steering.md b/maps/agent-access-map/steering.md index f5f10a1f49d..496ae859659 100644 --- a/maps/agent-access-map/steering.md +++ b/maps/agent-access-map/steering.md @@ -29,6 +29,10 @@ Compact task run reads retain `wakeupRequestId` and `queuedCommentSourceWakeIds` to identify consumed input queues, including coalesced requests. These receipt fields follow the existing task/run read checks and grant no queue mutation authority. +Accepted deferred requests remain pending while their recipient agent is paused. +Resume uses the normal admission checks. This grants no new pause, resume, queue +mutation, or execution authority to an agent. + Evidence: `source-only` — [implementation](../../server/src/routes/issues.ts#L16166) at the reviewed commit above. diff --git a/maps/feature-map/steering.md b/maps/feature-map/steering.md index 96c81b373a7..e35c184eb0a 100644 --- a/maps/feature-map/steering.md +++ b/maps/feature-map/steering.md @@ -100,6 +100,11 @@ the task (and separately its ancestor), preserve a draft, then resume the actual hold; no hidden send should occur while paused. Check both task presentations when available; the named component tests do not prove every host integration. +Pause an agent after a follow-up is accepted but before the prior turn finishes. +Let release and periodic recovery run. Confirm the saved request remains pending, +then resume the agent and verify one delivery. A paused request must not prevent +another eligible recipient's queued work from proceeding. + ### `agent-chat` Automated: [agent chat sessions](../../tests/e2e/agent-chat-sessions.spec.ts) covers diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 471b5cf18d0..24230989937 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -177,6 +177,41 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { return id; } + it.each(["assignee", "mentioned"])("preserves a paused recipient's accepted message until resume (%s)", async kind => { + const companyId = await seedCompany(); + const ownerId = await seedAgent({ companyId }); + const recipientId = kind === "assignee" ? ownerId : await seedAgent({ companyId }); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, recipientId)); + const issueId = await seedIssue({ companyId, status: "blocked", assigneeAgentId: ownerId }); + const runId = await seedRun({ companyId, agentId: ownerId, status: "succeeded", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user", + authorUserId: "responsible-user", body: "Accepted input must survive an agent pause" }).returning(); + const wakeId = await seedDeferredWake({ companyId, agentId: recipientId, issueId, requestedByActorId: "responsible-user", + payload: { _paperclipWakeContext: { issueId, wakeReason: kind === "mentioned" ? "issue_comment_mentioned" : "issue_commented", + wakeCommentIds: [comment.id] } }, + }); + const [accepted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + const input = { companyId, runId, now: new Date(), suppressImmediateRecovery: true, resumeStrandedQueue: true }; + for (let pass = 0; pass < 2; pass++) { + expect((await release(input)).outcome.kind).toBe("released"); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0]).toEqual(accepted); + expect(await db.select().from(heartbeatRuns)).toHaveLength(1); + } + await db.update(agents).set({ status: "active" }).where(eq(agents.id, recipientId)); + expect((await release(input)).outcome.kind).toBe("promoted"); + const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)); + expect(delivered).toMatchObject({ status: "queued", runId: expect.any(String), payload: accepted.payload }); + const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!)); + expect(successor).toMatchObject({ agentId: recipientId, contextSnapshot: { wakeCommentIds: [comment.id] } }); + await release(input); + expect(await db.select().from(heartbeatRuns)).toHaveLength(2); + expect((await db.select().from(issueComments).where(eq(issueComments.id, comment.id)))[0].body).toBe(comment.body); + }); + it.each(["queued", "running", "scheduled_retry", "cancelled"])("preserves a stranded queue when another agent has a newer %s turn", async status => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 3767c09740a..d147c3042f5 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -213,7 +213,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru .then((rows) => rows[0] ?? null); if (!agent) return null; const invokability = await evaluateAgentInvokabilityFromDb(tx, agent); - return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable }; + return { id: agent.id, companyId: agent.companyId, name: agent.name, invokable: invokability.invokable, + paused: !invokability.invokable && invokability.reason === "paused" }; }, async findNextDeferredWake({ companyId, issueId, excludedWakeIds }) { diff --git a/server/src/modules/wake-queue/application/types.ts b/server/src/modules/wake-queue/application/types.ts index 080a705bc80..5d793c0b477 100644 --- a/server/src/modules/wake-queue/application/types.ts +++ b/server/src/modules/wake-queue/application/types.ts @@ -58,6 +58,7 @@ export type InvokableAgentSnapshot = { companyId: string; name: string | null; invokable: boolean; + paused?: boolean; }; /** A new heartbeat run reached the queued state and should be published and dispatched. */ diff --git a/server/src/modules/wake-queue/application/use-cases.test.ts b/server/src/modules/wake-queue/application/use-cases.test.ts index 8f2cff02ab0..c05248a85cb 100644 --- a/server/src/modules/wake-queue/application/use-cases.test.ts +++ b/server/src/modules/wake-queue/application/use-cases.test.ts @@ -146,6 +146,28 @@ function createFakeRecovery(): RecoveryEscalationPort { } describe("releaseIssueExecution", () => { + it.each([false, true])("preserves a paused head without blocking an eligible follower (%s)", async withFollower => { + const paused = wakeCandidate({ id: "paused-wake", wakeReason: "issue_comment_mentioned", queuedCommentIds: ["paused-comment"] }); + const ready = wakeCandidate({ id: "ready-wake", agentId: "ready-agent", queuedCommentIds: ["ready-comment"] }); + const pending = withFollower ? [paused, ready] : [paused]; + const transaction = createFakeTransaction({ + findNextDeferredWake: vi.fn(async input => pending.find(wake => !input.excludedWakeIds?.includes(wake.id)) ?? null), + findInvokableAgent: vi.fn(async input => input.agentId === paused.agentId + ? { ...AGENT, invokable: false, paused: true } : { ...AGENT, id: ready.agentId }), + getQueuedCommentLiveness: vi.fn(async input => ({ liveNonSelfCommentIds: input.queuedCommentIds, containedSelfAuthoredComment: false })), + }); + const release = createReleaseIssueExecution({ issueLock: createFakeIssueLock(createFakeHost(), transaction, + { ...ISSUE, status: "blocked", assigneeAgentId: ready.agentId }, { ...RUN, status: "succeeded" }), recovery: createFakeRecovery() }); + const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date(), resumeStrandedQueue: true }); + expect(result.outcome.kind).toBe(withFollower ? "promoted" : "released"); + expect(transaction.failDeferredWake).not.toHaveBeenCalled(); + expect(transaction.cancelDeferredWake).not.toHaveBeenCalled(); + expect(transaction.normalizeDeferredWakeCommentIds).not.toHaveBeenCalled(); + expect(transaction.findNextDeferredWake).toHaveBeenNthCalledWith(2, expect.objectContaining({ excludedWakeIds: [paused.id] })); + if (withFollower) expect(transaction.finalizePromotedWake).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ wakeId: ready.id })); + else expect(transaction.finalizePromotedWake).not.toHaveBeenCalled(); + }); + it.each(["closing", "mixed_comments", "explicit_resume", "interaction", "other_source_task", "parent_open"])( "checks late completed-delegation mentions without dropping independent input (%s)", async (scenario) => { const originalCommentIds = ["closing-comment", "second-comment"]; diff --git a/server/src/modules/wake-queue/application/use-cases.ts b/server/src/modules/wake-queue/application/use-cases.ts index afe739e327f..d1b0b326f45 100644 --- a/server/src/modules/wake-queue/application/use-cases.ts +++ b/server/src/modules/wake-queue/application/use-cases.ts @@ -152,12 +152,12 @@ async function runReleaseDrain( // The `processedWakeIds` guard below makes a break of the invariant // fail loudly, instead of holding this transaction open forever. const processedWakeIds = new Set(); - const handoffWakeIds: string[] = []; + const preservedWakeIds: string[] = []; while (true) { const candidate = await ports.transaction.findNextDeferredWake({ companyId: run.companyId, issueId: issue.id, - ...(handoffWakeIds.length ? { excludedWakeIds: handoffWakeIds } : {}), + ...(preservedWakeIds.length ? { excludedWakeIds: preservedWakeIds } : {}), }); if (!candidate) break; if (processedWakeIds.has(candidate.id)) { @@ -205,7 +205,7 @@ async function runReleaseDrain( // The old owner can release before assignment admission adopts these // exact IDs. Leave its receipt intact, skip it for this drain, and let // a current-assignee wake behind it proceed. - handoffWakeIds.push(candidate.id); + preservedWakeIds.push(candidate.id); } else { // The current owner has finished. An obsolete assignment cannot // launch another former-owner run or reopen its completed task. @@ -283,7 +283,7 @@ async function runReleaseDrain( // A comment-id rewrite cannot change the agent or pause-hold facts already fetched above, so one decision covers both the rewritten and un-rewritten cases. const wakeOutcome = decideWakeOutcome({ - agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false }, + agent: { agentFound: deferredAgent !== null, invokable: deferredAgent?.invokable ?? false, paused: deferredAgent?.paused === true }, pauseHold: { activePauseHold: pauseHold.activePauseHold, treeHoldInteractionWake: pauseHold.treeHoldInteractionWake }, }); @@ -292,6 +292,11 @@ async function runReleaseDrain( continue; } + if (wakeOutcome.kind === "defer_agent_pause") { + preservedWakeIds.push(workingCandidate.id); + continue; + } + if (wakeOutcome.kind === "cancel_pause_hold") { await ports.transaction.cancelDeferredWake({ companyId: run.companyId, diff --git a/server/src/modules/wake-queue/domain/policy.test.ts b/server/src/modules/wake-queue/domain/policy.test.ts index 1d09806c3c0..1d73e887ecc 100644 --- a/server/src/modules/wake-queue/domain/policy.test.ts +++ b/server/src/modules/wake-queue/domain/policy.test.ts @@ -247,6 +247,28 @@ describe("decideWakeOutcome", () => { facts: { ...baseWakeOutcomeFacts, agent: { agentFound: false, invokable: false } }, expected: { kind: "fail_not_invokable" }, }, + { + name: "defer_agent_pause: a paused recipient keeps its accepted request", + facts: { ...baseWakeOutcomeFacts, agent: { agentFound: true, invokable: false, paused: true } }, + expected: { kind: "defer_agent_pause" }, + }, + { + name: "fail_not_invokable: a pause cannot authorize a missing recipient", + facts: { ...baseWakeOutcomeFacts, agent: { agentFound: false, invokable: false, paused: true } }, + expected: { kind: "fail_not_invokable" }, + }, + { + name: "cancel_pause_hold: an explicit task hold still suppresses a paused recipient", + facts: { agent: { agentFound: true, invokable: false, paused: true }, + pauseHold: { activePauseHold: true, treeHoldInteractionWake: false } }, + expected: { kind: "cancel_pause_hold" }, + }, + { + name: "defer_agent_pause: a permitted tree interaction cannot invoke a paused agent", + facts: { agent: { agentFound: true, invokable: false, paused: true }, + pauseHold: { activePauseHold: true, treeHoldInteractionWake: true } }, + expected: { kind: "defer_agent_pause" }, + }, { name: "fail_not_invokable: the agent is found but not invokable", facts: { ...baseWakeOutcomeFacts, agent: { agentFound: true, invokable: false } }, diff --git a/server/src/modules/wake-queue/domain/policy.ts b/server/src/modules/wake-queue/domain/policy.ts index 8ef20f07afd..11b904bdc61 100644 --- a/server/src/modules/wake-queue/domain/policy.ts +++ b/server/src/modules/wake-queue/domain/policy.ts @@ -142,6 +142,7 @@ export type DeferredWakeAgentFacts = { agentFound: boolean; /** True when the agent is invokable (status, org chain). Meaningless when agentFound is false. */ invokable: boolean; + paused?: boolean; }; export type DeferredWakePauseHoldFacts = { @@ -158,6 +159,7 @@ export type DeferredWakeOutcomeFacts = { export type DeferredWakeOutcomeDecision = | { kind: "fail_not_invokable" } + | { kind: "defer_agent_pause" } | { kind: "cancel_pause_hold" } | { kind: "promote" }; @@ -180,11 +182,10 @@ export function decideQueuedCommentAction( return { kind: "proceed" }; } -/** Decides the outcome for a deferred wake whose queued-comment action resolved to "proceed" (or finished its rewrite): fail, cancel, or promote. */ export function decideWakeOutcome(facts: DeferredWakeOutcomeFacts): DeferredWakeOutcomeDecision { const { agent, pauseHold } = facts; - if (!agent.agentFound || !agent.invokable) { + if (!agent.agentFound || (!agent.invokable && !agent.paused)) { return { kind: "fail_not_invokable" }; } @@ -192,6 +193,8 @@ export function decideWakeOutcome(facts: DeferredWakeOutcomeFacts): DeferredWake return { kind: "cancel_pause_hold" }; } + if (agent.paused) return { kind: "defer_agent_pause" }; + return { kind: "promote" }; } diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 00186fd2299..30bfc166ecb 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -6,6 +6,8 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vites import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { createRunDispatch } from "../../modules/run-dispatch/index.js"; import type { createWakeQueue } from "../../modules/wake-queue/index.js"; +import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; +import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; import { createHeartbeatRunState } from "./run-state.js"; import { createHeartbeatRunPreparation } from "./run-preparation.js"; import { createHeartbeatQueue, type HeartbeatQueueDependencies } from "./queue.js"; @@ -51,7 +53,7 @@ function callbacks(db: Db) { canStartNewExternalBotWork: vi.fn(async () => true), executeRun: vi.fn(async () => undefined), applyWakeQueuePostCommitEffects: vi.fn(async () => undefined), - releaseIssueExecutionAndPromote: vi.fn(async () => undefined), + releaseIssueExecutionAndPromote: vi.fn(async () => undefined), publishRunLifecyclePluginEvent: vi.fn(), cancelRunInternal: vi.fn(async () => undefined), cancelActiveForAgentInternal: vi.fn(async () => undefined), @@ -165,6 +167,29 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { return { ...seeded, recipient, pending }; } + it.each(["legacy", "native"])("periodic recovery preserves a paused recipient through the real %s release adapter", async runtimeMode => { + const { company, issue, recipient, pending } = await strandedMention(runtimeMode); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, recipient.id)); + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issue.id)); + const deps = callbacks(db); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, { + resolveResponsibleUserId: async () => "responsible-user", + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + }), recovery: { escalateStrandedAssignedIssue: async () => undefined, escalateStrandedRecoveryIssueInPlace: async () => undefined } }); + deps.releaseIssueExecutionAndPromote.mockImplementation(async (run, options) => { + await release({ companyId: run.companyId, runId: run.id, now: new Date(), ...options }); + }); + for (let pass = 0; pass < 2; pass++) { + await createHeartbeatQueue(db, deps).resumeQueuedRuns(); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, pending.id)))[0]) + .toMatchObject({ status: "deferred_issue_execution", runId: null, finishedAt: null, error: null, payload: pending.payload }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, company.id))).toHaveLength(1); + } + expect(deps.releaseIssueExecutionAndPromote).toHaveBeenCalledTimes(2); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + it.each(["legacy", "native"])("resumes an accepted non-assignee queue through its %s owner", async runtimeMode => { const { run, pending } = await strandedMention(runtimeMode); const deps = callbacks(db);