diff --git a/doc/idle-teams-attempt-recovery.md b/doc/idle-teams-attempt-recovery.md new file mode 100644 index 00000000000..dea26a0c5c7 --- /dev/null +++ b/doc/idle-teams-attempt-recovery.md @@ -0,0 +1,82 @@ +# Teams abandoned attempts and idle work + +Teams transport-attempt recovery now runs from committed transfer changes and +saved claim deadlines. The shared coordinator and scheduler own its lifecycle. +Together with consent expiry, this replaces the periodic Teams maintenance +callback. It adds no table, schema or independent timer. + +## Durable obligation + +A transfer in `consent_sending`, `uploading`, `file_info_sending`, or `conflict` +with `attemptExpiresAt` has an outstanding claim. The existing protocol writes +an attempt ID and a 90-second deadline before provider I/O. A normal receipt +clears both. A conflicting callback retains the original in-flight claim until +its deadline; the conflict is not proof that the original operation stopped. + +The protocol's central versioned update registers an awaited notification before +the write, within the existing transaction. This includes normal claims, +settlement, callbacks and caller-owned operator resolution. The DB work-signal +boundary owns outer commit, rollback and uncertain transaction settlement. +Inserts create unused consent, so they do not create an attempt obligation. + +Startup reconstructs eligible attempts in 100-row UUID pages. A committed hint +reloads its transfer ID. PostgreSQL timestamps round upward to a millisecond. +An inspection does not move the saved deadline. A changed claim is reloaded +before recovery; settled or deleted rows remove the cached obligation. + +## Recovery preserves uncertainty + +At the deadline, the existing Teams owner locks publication then transfer, +checks integrity, and rechecks the current phase and deadline. It clears the +expired attempt and changes a sending phase to its corresponding unknown outcome. +Conflict stays conflict with its original reason and private evidence. Publication +projection and transfer mutation remain atomic. Lock waits remain bounded. + +Recovery does not call a provider, resend a card or file, manufacture a receipt, +or assert that an unknown effect never occurred. A delayed foreground receipt +still has to match its current attempt ID and phase. Existing operator resolution +and on-use authorization rules remain unchanged. + +Unknown outcomes with no live attempt are passive review evidence. Unused consent +is owned by the separate consent-expiry worker. Neither keeps attempt recovery +querying. Failed loads, locks or projections retain bounded retry work; fair +batches allow healthy rows and other companies to progress. Worker errors redact +SQL parameters and private transfer envelopes. + +## Lifecycle and ownership + +The Teams module selects phases and deadlines. The existing protocol performs +recovery. Shared startup only registers the worker and admission gates. The +shared scoped worker coalesces writes, preserves dirty scopes through drain, +reconstructs after restart, and joins in-flight work during shutdown. An unresolved +write may still need transaction-status reconciliation while dispatch is paused; +this is not an empty maintenance scan. + +The explicit combined `expireAndRecover` and attempts-only diagnostic methods +remain available. Production `processPublicationMaintenance` no longer invokes +Teams recovery. Other chat maintenance reasons are separate pending migrations. + +Supported writes use this process's owning DB client. Another active server or +independent SQL writer requires a separate notification contract. Cloud external +alarm coverage and conservative sleep eligibility are unchanged. A registered +local deadline cannot wake a stopped host. This migration proves worker quiet, +not measured provider suspension or fleet savings. + +## Verification + +Real PostgreSQL tests exercise all sending stages, conflict during a live send, +normal settlement, interrupted receipt persistence, saved deadlines across restart, +quiet history, drain/resume, rollback, precision, atomic projection failure, +102-entry paging/fairness and shutdown. The real Board producer integration +registers the worker before sending, observes the saved claim while the provider +receipt is held, and verifies that the committed receipt disarms recovery. Existing +Teams authority, consent and publication cases remain covered. See the +[audit and failure ledger](plans/2026-10-25-teams-attempt-recovery-sleep-audit.md). + + +Current parent acceptance is recorded in the audit's +[current-parent section](plans/2026-10-25-teams-attempt-recovery-sleep-audit.md#current-consent-parent-preparation-and-acceptance-october-28): +239 selected cases pass, with the old inherited rate-window CI failure and its +already-landed fixture fix preserved. Actual landed-parent reconciliation and +fresh remote checks remain merge requirements; historical counts are not current +full-workspace verification. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 9f5cdbb750b..4dc61f402f6 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1822,3 +1822,34 @@ failures. Publication and fresh remote checks/review remain required. Parent #17586 Actions/security/Greptile pass at 250fca6cff; Paperclip Review is still running. The original full inventory, older E2E failure, deletion-policy finding and master reconciliation remain outstanding. No stack PR is merged. + + +Slice 26 is published as [#17590](https://github.com/paperclipai/paperclip/pull/17590) +at 5ddb4543ed, based on #17586. It is attached; remote gates remain required. +Slice 27 begins the separate Teams abandoned transport-attempt audit. The full +original inventory, older findings and final master reconciliation remain open. + + +## Slice 27: Teams abandoned transport attempts + +Attempt recovery now follows committed claims and saved 90-second deadlines. +It preserves unknown outcomes and performs no provider I/O. This removes the last +Teams periodic maintenance callback; unused consent remains the separate parent +owner. The [audit](2026-10-25-teams-attempt-recovery-sleep-audit.md) records 235 +unique selected passing cases and preserved initial test failures. Workspace +typecheck and build pass; publication and remote acceptance remain open. +#17590 has Greptile 5/5; CI is running. #17586 has new review findings for writer +support and report-query indexing. Fix at that parent and propagate before +acceptance. All original remaining work and previous unresolved findings remain. + + +Teams attempts are published as [#17592](https://github.com/paperclipai/paperclip/pull/17592). +Voice report review repair 94c60683b8 adds the bounded report-window index and +precise keyset pagination, with the single-active-writer contract explicit. The +20,001-old-row cost regression failed before repair, then all 14 report tests, +138 full voice/wiring cases and workspace types/build passed. Both review threads +are resolved. Consent #17590 incorporates it at e501d6c99d; both combined Teams +children pass 30 report/startup tests after propagation. Fresh remote gates still +apply. The next new reason is chat run-milestone projection, separate from wake +notices, terminal confirmations, interaction continuations and GitHub recovery. +The entire original inventory and older unresolved findings remain open. diff --git a/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md b/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md new file mode 100644 index 00000000000..0d6f8f6f339 --- /dev/null +++ b/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md @@ -0,0 +1,154 @@ +# Teams abandoned transport-attempt recovery sleep audit + +Baseline: consent expiry PR #17590 at 5ddb4543ed. +Branch: codex/sleep-27-teams-attempt-recovery. +Status: implemented; targeted local acceptance and workspace types/build pass; publication and remote gates pending. + +## Reason and source of truth + +This slice owns only abandoned Teams transport claims. Existing transfer phases +consent_sending, uploading, file_info_sending and conflict with a non-null +attemptExpiresAt carry the obligation. claim() saves attemptId and a 90-second +attempt expiry before provider I/O. Normal settlement clears the attempt. A +conflicting callback can retain an in-flight attempt until it expires. + +Recovery rechecks the row under the publication/transfer lock order, clears the +expired claim, and moves a sending phase to its matching unknown outcome. Conflict +stays conflict. It must never resend, manufacture a receipt or treat time as proof +that provider I/O did not occur. Unknown outcome rows without a deadline are +passive review evidence. Unused consent remains owned by #17590. + +## Producers and boundary + +teamsFileTransferService central update() owns claims, settlement, cancellation, +conflict and operator resolution. Add an awaited attempt signal there, within the +existing transaction. Insert creates consent_pending, not an attempt. Caller-owned +resolution must retain outer transaction settlement. Direct SQL/other processes +remain outside the local signal contract. Startup reconstructs saved attempts. + +Use the existing scoped deadline helper in the Teams folder. Expose only a narrow +per-transfer recovery method and worker registration. Shared app wiring supplies +admission gates; provider policy stays in Teams. Remove scheduleTeamsFileMaintenance +and its call from processPublicationMaintenance after both duties have owners. +Keep the explicit combined diagnostic recovery method compatible. + +## Verification required + +Prove real claim activation and settlement, three sending phases and conflict, +quiet unused/unknown/completed history, saved deadline across restart, no early +SQL, precision, projection rollback, lock contention and fair backlog, first work +during drain, nested rollback/outer commit, changes before deadline, and shutdown. +Exercise production wiring and ensure no periodic Teams scan remains. Keep +provider calls outside transactions and prove recovery makes no new provider call. +No change to external alarms, on-use authority or Cloud sleep eligibility. + +## Implementation and local evidence + +The new Teams-owned worker registers existing attempt deadlines. The central +versioned update signals both consent and attempt owners before writes. A narrow +recoverAbandonedAttempt method delegates to the original locked recovery logic. +The periodic scheduleTeamsFileMaintenance callback and its orchestration call are +removed. No other chat duty, schema or external wake policy is changed. + +- Five real-PostgreSQL/wiring suites pass 220 cases: 77 transfer protocol cases + and 143 publication, consent, consent-runtime and app cases. +- Fifteen selected production integration cases pass, including actual Board + claim/receipt activation and the parent expiry case plus existing Teams current + authority cases. The other 1,411 cases are skipped, not passed. +- These are 235 unique selected passing cases. The production claim test measures + 120 seconds with no select/transaction calls after the receipt settles. The + 102-entry recovery case drains healthy work from two companies while corrupt + evidence retains recovery. No recovery test causes another provider request. +- Server typecheck passes; workspace typecheck passes. Workspace build passes. Typecheck excludes test files. +- Full local test:run remains unclaimed due to the earlier embedded-PostgreSQL + host limitation. Full remote gates are required. No live Teams/suspension claim. + +Failure ledger: the first five-suite run passed 217 and failed two new assertions. +The drain fixture simulated an interrupted receipt transaction, then incorrectly +counted its pg_xact_status reconciliation as forbidden idle SQL. The rollback +fixture incorrectly rejected any future wake, including the legitimate shared +transaction-resolution deadline. It now reads committed eligibility and verifies +that no provider call occurred. Its asynchronous rejection is joined in finally. +The next run passed 219 and failed the drain fixture's wait for all idle holds to +clear while dispatch was deliberately paused. The final drain fixture uses a real +committed conflicting callback, retains its live claim through a quiet paused +window, and verifies retirement after resume. The interrupted receipt and failed +projection cases remain independently covered. Production behavior was not +relaxed to satisfy these assertions. + +Logs: /tmp/sleep27-teams-tests.log (initial failures), +/tmp/sleep27-teams-fixed-tests.log (drain hold assertion), +/tmp/sleep27-teams-final-tests.log (220 pass), +/tmp/sleep27-production-tests.log (15 pass), +/tmp/sleep27-server-types.log, /tmp/sleep27-workspace-types.log and +/tmp/sleep27-workspace-build.log (pass). + +## Stack gates + +Consent parent #17590 at 5ddb4543ed has Greptile 5/5 with no inline findings; +CI is still running. Voice-report #17586 now has two new Paperclip findings: +explicit single-active-process writer support and indexed report reconstruction. +They require owning-parent review/repair and propagation before stack acceptance. +Older #17490 E2E and #17580 deletion-policy findings and final master reconciliation +remain open. No stack PR is merged or the full objective complete. + + +Publication: [#17592](https://github.com/paperclipai/paperclip/pull/17592), initial +head 6f2f3951ca. Parent voice-report findings are repaired at 94c60683b8 and both +review threads resolved. The partial index and precise pagination are propagated +through consent parent e501d6c99d. Thirty report-worker/startup cases pass on each +combined Teams child; logs /tmp/sleep26-parent-review-tests.log and +/tmp/sleep27-parent-review-tests.log. Snapshot retention remains five. Fresh +remote gates on the propagated heads remain required; older stack findings remain. + + +## Current consent-parent preparation and acceptance — October 28 + +This existing attempt-recovery slice is prepared on consent candidate +`9e84cfab1f795279c8cec413a946631f1d781bd2`. Its complete 13-path patch is unchanged +while inheriting the parent's bounded consent reconstruction and retention +acceptance repair. The normal merge keeps the complete parent stack history +and the attempt appendix. The initial explicit-base composition was recorded +at `163a81d1`; the parent follow-up is included at `642095e2`. This preparation may be published on that exact pending parent to run CI in +parallel. Merge still requires the consent PR to have landed, an ordinary clean +Git composition onto actual master, and proof that only the complete owned +attempt patch is added while all master changes survive. A normal ancestry +merge remains necessary if that composition is not clean. No additional worker +refactor or new background duty is introduced. + +Current local acceptance passes 239 distinct selected cases: 221 across the +complete transfer/publication/consent/consent-runtime/app-wiring suites, 16 +selected production integration cases, and two delegated Slack policy cases. +The production cases include the real Board attempt claim and confirmed receipt, +retired recovery deadline and quiet SQL, consent expiry and current authority, +and the existing timed-out GitHub metadata/retry race. The other 1,579 chat +integration cases were skipped. The Slack probe covers linked and invitation +recipients; its other 81 cases were skipped. Each database-owning suite uses a +new isolated PostgreSQL database. Server production typecheck, build and module +boundaries pass. Test files were executed; production typechecking does not +statically check them. The unchanged signoff E2E test was not run locally in this +acceptance; fresh remote E2E gates remain required. + +The old CI run `40557802358`, server shard 10/12 job `122390238707`, failed the +delegated Slack invitation target-rate assertion: the second call was admitted +at a real minute boundary. That shard had 2,122 passing tests, one failed and 15 +skipped. The actual prepared parent already contains the Date-only rate-window +fixture fix, so it is preserved rather than duplicated in this PR. Both linked +and invitation policy cases now pass through their full caller/target/shared +rate scenarios on a fresh server database. No production policy, timeout, test +expectation or new owned file was changed for this inherited failure; current +remote CI must pass independently. The old failure is not erased by the probe. + +Evidence: `/tmp/sleep27-old-server10-job.log`, +`/tmp/sleep27-current-parent-protocols.log` (221 pass), +`/tmp/sleep27-current-parent-producers.log` (16 pass), +`/tmp/sleep27-inherited-rate-window-probe.log` (two pass), +`/tmp/sleep27-current-parent-types.log`, +`/tmp/sleep27-current-parent-build.log`, and +`/tmp/sleep27-current-parent-boundaries.log`. +Earlier 235-case or combined-stack evidence belongs to its stated revisions and +is not added to 239. Latest-head Greptile 5/5 and required CI still gate merge; +missing or paused Ponytail turns without a verdict follow the user's existing +merge authorization, and actual failing findings must be fixed. No full local +repository suite, live Teams/Slack tenant, external wake, physical suspension or +cost-saving claim is made. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 4c798afba0a..2aabbe2f370 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -81,10 +81,11 @@ describe("createChatReconciliationCoordinator", () => { expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,"); const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8"); - const remainingMaintenance = teams.slice(teams.indexOf("function scheduleTeamsFileMaintenance"), teams.indexOf("function registerTeamsConsentExpiryWork")); - expect(remainingMaintenance).toContain("protocol.recoverAbandonedAttempts("); - expect(remainingMaintenance).not.toContain("protocol.expireAndRecover("); - expect(remainingMaintenance).not.toContain("chatTeamsFileTransfers.expiresAt"); + expect(source).toContain("chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork,"); + expect(teams).not.toContain("scheduleTeamsFileMaintenance"); + expect(teams).not.toContain("selectDistinct"); + const chat = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8"); + expect(chat).not.toContain("scheduleTeamsFileMaintenance"); }); it("separates live voice recovery from the remaining periodic maintenance", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 5e8c692955b..56fc1a628f0 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -38,6 +38,7 @@ import { secretService } from "../services/secrets.js"; import { githubChatWizardService } from "../services/chat-github-wizard.js"; import { dispatchGitHubBotCloudEvent, registerGitHubBotCloudSetupPoller } from "../services/chat-github-cloud-ingress.js"; import { githubChatManagementService } from "../services/chat-github-management.js"; +import { GITHUB_INLINE_RETRY_WINDOW_MS } from "../services/chat-channels/github/retry.js"; import { githubChatReviewService } from "../services/chat-github-reviews.js"; import { resolveGitHubReviewInstructions } from "../services/chat-channels/github/review-instructions.js"; import { githubResponseCommentService } from "../services/chat-github-response-comments.js"; @@ -1218,6 +1219,47 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { companyId, assignedAgentId, replacementAgentId }; } + it("tracks the actual Board Teams claim and retires recovery when its receipt commits", async () => { + const context = await teamsFileAuthorityFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + context.controls.beforeConsentReceipt = async () => { entered(); await hold; }; + let send: ReturnType | undefined; + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await context.service.registerTeamsAttemptRecoveryWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + const attachment = await context.createFile(); + send = context.service.publishBoardMessage(context.endpoint.id, context.conversation.id, + "File with an active claim", `teams-attempt-${randomUUID()}`, "owner-user", [attachment.id]); + await ready; + const [claim] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.endpointId, context.endpoint.id)); + expect(claim.phase).toBe("consent_sending"); expect(context.consentCards).toHaveLength(1); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect(coordinator.nextWakeAt()).toBe(claim.attemptExpiresAt!.getTime()); + }, { timeout: 5000 }); + release(); await send; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const [settled] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.id, claim.id)); + expect(settled.phase).toBe("awaiting_consent"); expect(settled.attemptId).toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + expect(context.consentCards).toHaveLength(1); expect(context.fileCards).toEqual([]); + expect(context.uploadRequest).not.toHaveBeenCalled(); expect(errors).toEqual([]); + } finally { + release(); await send; + await coordinator.stop(); vi.useRealTimers(); + await retirePublicationFixture(context.service, context.endpoint.id); + // This test intentionally leaves unused consent; do not feed the next global expiry fixture. + await db.delete(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.endpointId, context.endpoint.id)); + } + }); + it("expires Teams consent from the committed Board transfer without periodic maintenance", async () => { const context = await teamsFileAuthorityFixture(); const errors: unknown[] = []; @@ -3895,7 +3937,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const retry = gateway.executeTool(input); release(); expect(await retry).toMatchObject({ result: { data: { actionId: expect.any(String) } } }); - await expect.poll(async () => (await f.action())?.status).toBe("processed"); + // A replay may return the durable receipt while the timed-out request + // still publishes. Observe the owner's bounded retry window, not the + // assertion library's one-second default. + await expect.poll(async () => (await f.action())?.status, + { timeout: GITHUB_INLINE_RETRY_WINDOW_MS }).toBe("processed"); expect(await gateway.executeTool(input)).toMatchObject({ result: { data: { status: "processed" } } }); const actions = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"))); @@ -3903,7 +3949,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(actions[0].status).toBe("processed"); expect(f.responseComments.size).toBe(1); expect(f.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]); - }); + }, 60_000); it("rejects changed requests and revoked access before repeating an unstaged review", async () => { const f = await publicationRetryFixture(); let metadataReads = 0; diff --git a/server/src/app.ts b/server/src/app.ts index 849985244af..6da6cb70474 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1302,6 +1302,7 @@ export async function createApp( chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index ef96c0788df..009193f7cfc 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1251,7 +1251,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { stageTerminalReceiptReactionRemovals, runtimeContextForRecord, scheduleMessageProcessing, processReceiptReaction, }); const { stageTeamsBoardFileIntent, deriveTeamsFileTransferAuthority } = createTeamsFileAuthority({ runtimeGeneration, credentialFingerprint, runtime, runtimeContexts, localRuntimeEpoch, runtimeVersions, lockCurrentPrincipalAuthorization, lifecycleRuntimeFence, assertFailedRetryReceipt, inboundWakeScope, failedChatRetrySource, isExplicitOperatorPublication, retryCommentIds }); - const { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance } = createTeamsFileDelivery(db, { + const { registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication } = createTeamsFileDelivery(db, { endpointRecord, runtimeContextForRecord, runtimeFor, runtimeContexts, localRuntimeEpoch, runtime, deriveTeamsFileTransferAuthority, publicationFiles, scheduleMessageProcessing, processPendingPublications, withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown, @@ -3342,7 +3342,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); async function processPublicationMaintenance(limit = 25) { - scheduleTeamsFileMaintenance(); scheduleGitHubMaintenance(); await enqueueInboundWakeupPublications(); await reconcileTerminalConfirmationActions(limit); @@ -3498,6 +3497,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, registerCommandWork, registerReceiverWork, diff --git a/server/src/services/chat-channels/microsoft-teams/attempt-recovery-work.ts b/server/src/services/chat-channels/microsoft-teams/attempt-recovery-work.ts new file mode 100644 index 00000000000..d32aff5594b --- /dev/null +++ b/server/src/services/chat-channels/microsoft-teams/attempt-recovery-work.ts @@ -0,0 +1,54 @@ +import { and, eq, gt, inArray, isNotNull, sql } from "drizzle-orm"; +import { chatTeamsFileTransfers, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; + +export const abandonedTeamsAttemptPhases = ["consent_sending", "uploading", "file_info_sending", "conflict"]; +type Entry = { id: string; companyId: string; dueAt: number }; + +export function notifyTeamsAttemptRecovery(tx: object, id: string) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTeamsAttemptRecovery, id); +} + +/** Retires expired transport claims without retrying an unknown provider effect. */ +export function teamsAttemptRecoveryWork(db: Db, recover: (companyId: string, id: string) => Promise) { + async function load(id?: string, signal?: AbortSignal): Promise { + try { + const entries: Entry[] = []; let cursor: string | undefined; + for (;;) { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId, + dueAt: sql`ceil(extract(epoch from ${chatTeamsFileTransfers.attemptExpiresAt}) * 1000)`, + }).from(chatTeamsFileTransfers).where(and(inArray(chatTeamsFileTransfers.phase, abandonedTeamsAttemptPhases), + isNotNull(chatTeamsFileTransfers.attemptExpiresAt), + id ? eq(chatTeamsFileTransfers.id, id) : undefined, + cursor ? gt(chatTeamsFileTransfers.id, cursor) : undefined)).orderBy(chatTeamsFileTransfers.id).limit(100); + entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) }))); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + } catch { + // Never expose private transfer envelopes or SQL parameter errors. + throw new Error("Teams attempt recovery selection is temporarily unavailable"); + } + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + const current = (await load(entry.id, signal))[0]; + if (!current) return null; + if (current.dueAt > Date.now()) return current; + signal?.throwIfAborted(); + try { await recover(current.companyId, current.id); } + catch { throw new Error("Teams attempt recovery could not settle"); } + return (await load(entry.id, signal))[0] ?? null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTeamsAttemptRecovery, + decodeScope: (hint: string) => hint === "*" ? undefined : hint, + matches: (entry: Entry, id?: string) => !id || entry.id === id, + load, check, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/chat-channels/microsoft-teams/file-delivery.ts b/server/src/services/chat-channels/microsoft-teams/file-delivery.ts index 396662a4e49..c87c045eb10 100644 --- a/server/src/services/chat-channels/microsoft-teams/file-delivery.ts +++ b/server/src/services/chat-channels/microsoft-teams/file-delivery.ts @@ -1,3 +1,4 @@ +import { teamsAttemptRecoveryWork } from "./attempt-recovery-work.js"; import { teamsConsentExpiryWork } from "./consent-expiry-work.js"; import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; import { createOutboundAttachments } from "../outbound-attachments.js"; @@ -14,8 +15,6 @@ import { chatCredentialMutationLease, type CredentialMutationLeaseGuard } from " import { and, eq, - inArray, - lte, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -37,7 +36,6 @@ import { HttpError, conflict, } from "../../../errors.js"; -import { logger } from "../../../middleware/logger.js"; import { type ChatSdkCallbackEvent, @@ -376,51 +374,13 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption } } - let teamsMaintenancePending = false; - let teamsMaintenanceAfter = 0; - function scheduleTeamsFileMaintenance() { - if ( - teamsMaintenancePending || - Date.now() < teamsMaintenanceAfter || - isShuttingDown() - ) - return; - teamsMaintenancePending = true; - teamsMaintenanceAfter = Date.now() + 30_000; - scheduleMessageProcessing(async () => { - try { - const scopes = await db - .selectDistinct({ companyId: chatTeamsFileTransfers.companyId }) - .from(chatTeamsFileTransfers) - .where( - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_sending", "uploading", "file_info_sending", "conflict", - ]), - lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()), - ), - ); - const protocol = teamsTransferProtocol(); - // Separate from the ordinary message queue. Each company sweep bounds - // its row count and lock waits and performs no provider I/O. - for (const scope of scopes) { - if (isShuttingDown()) break; - await protocol.recoverAbandonedAttempts(scope.companyId, 100); - } - } catch { - // Selection errors are closed here too; private transfer envelopes and - // SQL parameters must not escape through the generic background logger. - logger.warn( - "Teams file maintenance will retry after a temporary failure", - ); - } finally { - teamsMaintenancePending = false; - } - }); - } function registerTeamsConsentExpiryWork(coordinator: ReturnType, canRun: () => boolean) { return teamsConsentExpiryWork(db, (companyId, id) => teamsTransferProtocol().expireUnusedConsent(companyId, id)) .register(coordinator, () => !isShuttingDown() && canRun()); } - return { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance }; + function registerTeamsAttemptRecoveryWork(coordinator: ReturnType, canRun: () => boolean) { + return teamsAttemptRecoveryWork(db, (companyId, id) => teamsTransferProtocol().recoverAbandonedAttempt(companyId, id)) + .register(coordinator, () => !isShuttingDown() && canRun()); + } + return { registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication }; } diff --git a/server/src/services/chat-teams-file-transfers.test.ts b/server/src/services/chat-teams-file-transfers.test.ts index 0dd72d0f387..93a6909d354 100644 --- a/server/src/services/chat-teams-file-transfers.test.ts +++ b/server/src/services/chat-teams-file-transfers.test.ts @@ -1,3 +1,4 @@ +import { teamsAttemptRecoveryWork } from "./chat-channels/microsoft-teams/attempt-recovery-work.js"; import { teamsConsentExpiryWork } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { idleWorkSnapshot } from "./task-admission.js"; @@ -332,6 +333,200 @@ suite( }; } + describe("scheduled abandoned attempt recovery", () => { + let coordinator: ReturnType | undefined; + const transferIds: string[] = [], errors: unknown[] = []; + let enabled = true; + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + if (transferIds.length) await db.delete(chatTeamsFileTransfers).where(inArray(chatTeamsFileTransfers.id, transferIds.splice(0))); + errors.length = 0; enabled = true; + }); + async function transfer() { + const f = await fixture(600_000); transferIds.push(f.created.id); + f.opts.now = () => new Date(); + return f; + } + async function abandon(f: Awaited>, stage = "consent") { + if (stage !== "consent") { + await f.service().process(f.a.companyId, f.created.id); + await f.service().recordConsent(await f.event()); + if (stage === "file_info") await f.service().process(f.a.companyId, f.created.id); + } + const outcome = stage === "consent" ? "awaiting_consent" : stage === "upload" ? "file_info_pending" : "delivered"; + f.opts.project = async (_tx, row) => { if (row.phase === outcome) throw new Error("receipt projection interrupted"); }; + await expect(f.service().process(f.a.companyId, f.created.id)).rejects.toThrow("receipt projection interrupted"); + delete f.opts.project; + return (await f.read()).attemptExpiresAt!.getTime(); + } + function worker(recover: (companyId: string, id: string) => Promise) { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + const work = teamsAttemptRecoveryWork(db, recover); + return { work, ready: work.register(coordinator, () => enabled).ready }; + } + async function settled() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 }); + } + async function quiet(ms = 10_000) { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + try { await vi.advanceTimersByTimeAsync(ms); expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); } + finally { select.mockRestore(); transaction.mockRestore(); execute.mockRestore(); } + } + it("leaves unused consent and retained terminal history quiet", async () => { + const f = await transfer(), cancelled = await transfer(); + await cancelled.service().cancel(cancelled.a.companyId, cancelled.created.id); + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect((await f.read()).phase).toBe("consent_pending"); expect(errors).toEqual([]); + }); + it.each(["consent", "upload", "file_info"])("restores the actual %s claim deadline and quarantines without repeating provider I/O", async stage => { + const f = await transfer(); const due = await abandon(f, stage); + const calls = [f.opts.postConsent, f.opts.uploadRequest!, f.opts.postFileInfo].map(fn => vi.mocked(fn).mock.calls.length); + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await coordinator!.stop(); await worker(f.service().recoverAbandonedAttempt).ready; + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await settled(); + expect(await f.read()).toMatchObject({ phase: `${stage}_unknown`, attemptId: null, attemptExpiresAt: null, reason: "worker_outcome_unknown" }); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect([f.opts.postConsent, f.opts.uploadRequest!, f.opts.postFileInfo].map(fn => vi.mocked(fn).mock.calls.length)).toEqual(calls); + expect(errors).toEqual([]); + }); + it("wakes from the real first claim and removes the deadline after normal settlement", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.postConsent = vi.fn(async () => { entered(); await hold; return { id: "card-1" }; }); + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); + const send = f.service().process(f.a.companyId, f.created.id); + try { + await ready; await settled(); const row = await f.read(); + expect(coordinator!.nextWakeAt()).toBe(row.attemptExpiresAt!.getTime()); await quiet(); + } finally { release(); } + expect((await send).phase).toBe("awaiting_consent"); await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(f.opts.postConsent).toHaveBeenCalledTimes(1); + }); + it("retains a first committed attempt across drain and recovers on resume", async () => { + const f = await transfer(); enabled = false; + await worker(f.service().recoverAbandonedAttempt).ready; + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.postConsent = vi.fn(async () => { entered(); await hold; return { id: "late-card" }; }); + const send = f.service().process(f.a.companyId, f.created.id); + try { + await ready; await f.service().recordConsent(await f.event()); + await f.service().recordConsent(await f.event("decline")); + } finally { release(); await send; } + await quiet(120_000); + expect((await f.read()).phase).toBe("conflict"); expect((await f.read()).attemptId).not.toBeNull(); + enabled = true; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("conflict"); expect((await f.read()).attemptId).toBeNull(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); expect(errors).toEqual([]); + }); + it("retains atomic recovery after projection failure without leaking its error", async () => { + const f = await transfer(); const due = await abandon(f); let fail = true; + f.opts.project = async tx => { + await tx.update(chatPublications).set({ state: "failed" }).where(eq(chatPublications.id, f.a.publicationId)); + if (fail) throw new Error("PRIVATE-TEAMS-UPLOAD-CANARY"); + }; + await worker(f.service().recoverAbandonedAttempt).ready; + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await settled(); + expect((await f.read()).phase).toBe("consent_sending"); + expect((await db.select().from(chatPublications).where(eq(chatPublications.id, f.a.publicationId)))[0].state).toBe("pending"); + expect(errors).toHaveLength(1); expect(String((errors[0] as AggregateError).errors[0])).not.toContain("CANARY"); + fail = false; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("consent_unknown"); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); expect(f.opts.postConsent).toHaveBeenCalledTimes(1); + }); + it("preserves conflict evidence and clears only its expired ownership", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.postConsent = vi.fn(async () => { entered(); await hold; return { id: "late-card" }; }); + await worker(f.service().recoverAbandonedAttempt).ready; + const send = f.service().process(f.a.companyId, f.created.id); + try { + await ready; await f.service().recordConsent(await f.event()); + await f.service().recordConsent(await f.event("decline")); + } finally { release(); } + await send; const before = await f.read(); expect(before.phase).toBe("conflict"); + await settled(); expect(coordinator!.nextWakeAt()).toBe(before.attemptExpiresAt!.getTime()); + await vi.advanceTimersByTimeAsync(before.attemptExpiresAt!.getTime() - Date.now() + 1000); await settled(); + const after = await f.read(); expect(after).toMatchObject({ phase: "conflict", attemptId: null, attemptExpiresAt: null, reason: before.reason, privateState: before.privateState }); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(f.opts.postConsent).toHaveBeenCalledTimes(1); expect(errors).toEqual([]); + }); + it("does not observe a rolled-back claim or contact the provider before commit", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.project = async (_tx, row) => { if (row.phase === "consent_sending") { entered(); await hold; throw new Error("claim rolled back"); } }; + await worker(f.service().recoverAbandonedAttempt).ready; await settled(); + const send = f.service().process(f.a.companyId, f.created.id); + const failure = expect(send).rejects.toThrow("claim rolled back"); + try { + await ready; await vi.advanceTimersByTimeAsync(10_000); + expect(await teamsAttemptRecoveryWork(db, f.service().recoverAbandonedAttempt).load()).toEqual([]); + expect((await f.read()).phase).toBe("consent_pending"); expect(f.opts.postConsent).not.toHaveBeenCalled(); + } finally { release(); await failure; } + await settled(); + expect((await f.read()).phase).toBe("consent_pending"); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("rounds submillisecond attempts up and rechecks a changed claim", async () => { + const f = await transfer(); await abandon(f); + const whole = Date.now() + 60_000; + await db.update(chatTeamsFileTransfers).set({ attemptExpiresAt: sql`${new Date(whole).toISOString()}::timestamptz + interval '0.0005 second'` }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + vi.useFakeTimers({ toFake: ["Date"] }); + const work = teamsAttemptRecoveryWork(db, f.service().recoverAbandonedAttempt), [entry] = await work.load(); + expect(entry.dueAt).toBe(whole + 1); vi.setSystemTime(whole); await work.check(entry); + expect((await f.read()).phase).toBe("consent_sending"); + const changed = whole + 90_000; + await db.update(chatTeamsFileTransfers).set({ attemptId: randomUUID(), attemptExpiresAt: new Date(changed) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + vi.setSystemTime(whole + 1); expect(await work.check(entry)).toMatchObject({ dueAt: changed }); + expect((await f.read()).phase).toBe("consent_sending"); + vi.setSystemTime(changed); expect(await work.check(entry)).toBeNull(); expect((await f.read()).phase).toBe("consent_unknown"); + }); + it("pages and drains healthy attempts across companies while a corrupt claim retains recovery", async () => { + const f = await transfer(), other = await transfer(); await abandon(f); await abandon(other); + for (let index = 0; index < 100; index++) { + const publicationId = randomUUID(); + await db.insert(chatPublications).values({ id: publicationId, companyId: f.a.companyId, + endpointId: f.a.endpointId, conversationId: f.a.conversationId, issueId: f.a.issueId, + commentId: f.a.commentId, idempotencyKey: `attempt-page:${publicationId}`, + payload: { text: "Same authorized file", attachmentIds: [f.a.attachmentId] } }); + const a = { ...f.a, publicationId }; + const service = teamsFileTransferService(db, { ...f.opts, authorize: async () => a, + project: async (_tx, row) => { if (row.phase === "awaiting_consent") throw new Error("receipt interrupted"); } }); + const created = await service.issue(a, new Date(Date.now() + 600_000)); transferIds.push(created.id); + await expect(service.process(a.companyId, created.id)).rejects.toThrow("receipt interrupted"); + } + await db.update(chatTeamsFileTransfers).set({ authorityDigest: "0".repeat(64) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + const { work, ready } = worker(f.service().recoverAbandonedAttempt); await ready; + const entries = await work.load(); expect(entries).toHaveLength(102); + const calls = vi.mocked(f.opts.postConsent).mock.calls.length; + await vi.advanceTimersByTimeAsync(Math.max(...entries.map(row => row.dueAt)) - Date.now() + 1000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect((await other.read()).phase).toBe("consent_unknown"); + expect((await work.load()).map(row => row.id)).toEqual([f.created.id]); + }, { timeout: 10_000 }); + expect((await f.read()).phase).toBe("consent_sending"); expect(errors.length).toBeGreaterThan(0); + expect(f.opts.postConsent).toHaveBeenCalledTimes(calls); + }, 30_000); + it("joins recovery projection before shutdown completes", async () => { + const f = await transfer(); const due = await abandon(f); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { entered = resolve; }); + f.opts.project = async () => { entered(); await hold; }; + await worker(f.service().recoverAbandonedAttempt).ready; + try { + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await ready; + let stopped = false; const stop = coordinator!.stop().then(() => { stopped = true; }); + await Promise.resolve(); expect(stopped).toBe(false); release(); await stop; + expect((await f.read()).phase).toBe("consent_unknown"); + } finally { release(); } + }); + }); + describe("scheduled unused consent expiry", () => { let coordinator: ReturnType | undefined; const transferIds: string[] = [], errors: unknown[] = []; diff --git a/server/src/services/chat-teams-file-transfers.ts b/server/src/services/chat-teams-file-transfers.ts index 0f82870da06..ef4a17f8c5d 100644 --- a/server/src/services/chat-teams-file-transfers.ts +++ b/server/src/services/chat-teams-file-transfers.ts @@ -1,3 +1,4 @@ +import { notifyTeamsAttemptRecovery } from "./chat-channels/microsoft-teams/attempt-recovery-work.js"; import { notifyTeamsConsentExpiry, unusedTeamsConsentPhases } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; import { createHash, randomUUID } from "node:crypto"; import { and, eq, gt, inArray, lte, or, sql } from "drizzle-orm"; @@ -390,6 +391,7 @@ export function teamsFileTransferService( patch: Partial, ): Promise { await notifyTeamsConsentExpiry(tx, row.id); + await notifyTeamsAttemptRecovery(tx, row.id); const [next] = await tx .update(chatTeamsFileTransfers) .set({ ...patch, version: row.version + 1, updatedAt: now() }) @@ -1159,6 +1161,7 @@ export function teamsFileTransferService( get, expireAndRecover: (companyId: string, limit = 25) => expireAndRecover(companyId, limit), recoverAbandonedAttempts: (companyId: string, limit = 25) => expireAndRecover(companyId, limit, "attempts"), + recoverAbandonedAttempt: (companyId: string, id: string) => recoverTransfer(companyId, id, "attempts"), expireUnusedConsent: (companyId: string, id: string) => recoverTransfer(companyId, id, "consent"), cancel, canCancelConflict, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 4bc877804d3..16586747cde 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -50,6 +50,7 @@ export const DELIVERY_QUEUES = { voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", chatTeamsConsentExpiry: "chat-teams-consent-expiry", + chatTeamsAttemptRecovery: "chat-teams-attempt-recovery", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs", diff --git a/tests/e2e/signoff-policy.spec.ts b/tests/e2e/signoff-policy.spec.ts index 670304fb2d8..31276258904 100644 --- a/tests/e2e/signoff-policy.spec.ts +++ b/tests/e2e/signoff-policy.spec.ts @@ -239,9 +239,13 @@ async function invokeHeartbeat( if ( typeof recentRun.id === "string" && recentRun.agentId === agentId && - (context.issueId === issueId || context.taskId === issueId) + (context.issueId === issueId || context.taskId === issueId) && + ["queued", "running", "succeeded"].includes(recentRun.status) ) { - return recentRun.id; + // A prior signoff stage can leave a cancelled run in recent history. + // Re-read a viable candidate: it may have settled since the list read. + const verified = await fetchVerifiedRun(board, recentRun.id, agentId, issueId); + if (verified && ["queued", "running", "succeeded"].includes(verified.status)) return verified.id; } } } @@ -550,6 +554,38 @@ async function createIssueWithPolicy(ctx: TestContext, title: string, stages?: u return issue; } +test("a deferred signoff wake ignores cancelled history from an earlier stage", async () => { + const actingAgent = "executor"; + const issueId = "returning-task"; + const json = (body: unknown) => ({ ok: () => true, json: async () => body }); + const reads: string[] = []; + let wakes = 0; + const board = { + post: async (url: string) => { + expect(url).toBe(`${BASE_URL}/api/agents/${actingAgent}/wakeup`); + wakes++; + return json({ reason: "issue_execution_deferred", executionRunId: "reviewer-turn", executionAgentId: "reviewer" }); + }, + get: async (url: string) => { + reads.push(url); + if (url === `${BASE_URL}/api/heartbeat-runs/reviewer-turn`) return json({ status: "succeeded" }); + if (url === `${BASE_URL}/api/issues/${issueId}`) return json({ companyId: "company", assigneeAgentId: actingAgent }); + if (url === `${BASE_URL}/api/companies/company/heartbeat-runs?agentId=${actingAgent}&limit=20`) return json([ + { id: "cancelled-earlier-stage", agentId: actingAgent, status: "cancelled", errorCode: "issue_assignee_changed", contextSnapshot: { issueId } }, + { id: "current-executor-turn", agentId: actingAgent, status: "running", contextSnapshot: { issueId } }, + ]); + if (url === `${BASE_URL}/api/heartbeat-runs/current-executor-turn`) return json({ + id: "current-executor-turn", agentId: actingAgent, status: "running", contextSnapshot: { issueId }, + }); + throw new Error(`Unexpected fixture read: ${url}`); + }, + } as unknown as APIRequestContext; + expect(await invokeHeartbeat(board, actingAgent, issueId)).toBe("current-executor-turn"); + expect(wakes).toBe(1); + expect(reads).toContain(`${BASE_URL}/api/heartbeat-runs/current-executor-turn`); + expect(reads).toContain(`${BASE_URL}/api/companies/company/heartbeat-runs?agentId=${actingAgent}&limit=20`); +}); + test.describe("Signoff execution policy", () => { let ctx: TestContext;