From 4a89456bb3f9aef87cfe0e922e8b9d1a777a6c7e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 10:32:30 -0500 Subject: [PATCH] fix(server): schedule Teams consent expiry from saved deadlines (#17590) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Teams file delivery saves consent and transfer state in the database. > - Unused consent must expire even when no new message arrives. > - The maintenance loop currently checks this state every 30 seconds. > - This PR schedules each existing consent deadline and wakes on committed changes. > - Empty or settled consent work stops querying while expiry remains recoverable. ## Linked Issues or Issue Description **What existing behavior does this improve?** Teams unused file-consent expiry. This is slice 26 of the recurring-server migration stack, following merged #17586 and targeting master. **Current behavior** The Teams maintenance callback scans for expired unused consent and abandoned transport attempts together, even when consent work is empty. **Proposed behavior** The Teams owner reconstructs unused consent at startup and reloads one transfer after a committed write. The shared scheduler runs its saved expiry. Abandoned attempts keep their current recovery until a separate slice migrates them. **Reason and benefit** Remove unnecessary consent queries without losing expiry, atomic publication projection, or recovery after restart. **Breaking changes** No schema or API change. The explicit combined recovery method remains available. Cloud sleep eligibility and external alarms remain unchanged. ## What Changed - Add a Teams-owned scoped deadline worker for the three unused consent phases. Registered reconstruction yields after one 100-row page per scope refresh and resets cursors on registration; the explicit diagnostic inventory still returns every eligible row. - Signal before insertion and the existing versioned update inside the owning transaction. Nested writes wait for outer settlement. - Extract the existing locked per-transfer recovery operation. Preserve authority checks, atomic projection, bounded lock waits and unknown-outcome quarantine. - Remove unused consent from periodic maintenance selection and execution. Keep abandoned attempts as a separate obligation. - Register the worker through existing app lifecycle gates. Add lifecycle, producer, production wiring and precision regressions. - Document supported writers, remaining sleep limits and the failure ledger in `doc/idle-teams-consent-expiry.md` and its linked audit. ## Verification Published head: `9e84cfab1f795279c8cec413a946631f1d781bd2`. This follow-up changes only the inherited voice-retention test and its CI failure ledger; all production behavior is unchanged from `36f43922`. It normally merges actual landed master/report parent `8049fbf51d0701ca6b242e4b1198e5449f7661b2`. The verified 12-path Teams patch was byte-identical across the actual-parent ancestry integration. This test-only CI follow-up adds one existing retention test path, bringing the final scope to 13 paths. Every unrelated master change, report documentation fix, and migration 0397 is preserved. No database, shared-worker, workflow or lockfile change belongs to this slice. - **223 distinct selected cases pass:** 209 across five complete transfer/publication/consent/runtime/app-wiring suites, plus 14 selected real Board-producer and Teams-authority integration cases. The other 1,580 integration cases were skipped, not passed. The protocol and production consumers use separate new PostgreSQL databases migrated through the application migrator. - A real 105-transfer regression failed against the original worker because one registered startup pass selected 105 rows instead of yielding at 100. The corrected worker uses the existing paged shared contract. Restarting the same service midway through discovery selects the first 100 again, restores all 105 saved deadlines and expires them at the original deadline. SQL is quiet before the deadline and after settlement, with no consent-card or upload requests. The one-shot inventory still returns all 105 rows. - A guarded mutation removing only cursor reset fails at five rows instead of 100 after restart. The original production source was restored before the complete suites. The initial regression fixture incorrectly faked network timers before its first SQL connection and timed out; that setup failure is retained separately from the corrected 105-versus-100 regression. No assertion or timeout was weakened. - The real Board producer registers the worker before publication, commits the transfer wake, preserves the saved expiry, remains SQL-quiet until due and atomically projects expiry to a cancelled publication without another provider request. Existing authority and consent transitions remain covered. - Complete lifecycle coverage retains all three consent phases, precision rounding, rollback and held outer commit, exact-row hints, expiry recheck, failed projection retry, corrupt-row fairness, drain/resume, shutdown and separation from abandoned-attempt maintenance. Production server typecheck, build and module boundaries pass. Server production typechecking does not statically check test files; those files were executed. - [Current bounded reconstruction and failure ledger](https://github.com/paperclipai/paperclip/blob/36f43922b0a408f76adb7de31280812e9b752019/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md#bounded-reconstruction-and-current-parent-acceptance-october-28) records exact scope, counts, guarded mutation, corrected timer setup, earlier fixture failures and limitations. Earlier 222-case and combined-stack results belong to their stated historical heads; they are not added to 223 or claimed rerun here. The [stack ledger](https://github.com/paperclipai/paperclip/blob/36f43922b0a408f76adb7de31280812e9b752019/doc/plans/2026-10-24-recurring-server-sleep-stack.md) retains inherited failure history. No complete local repository-suite, live Teams tenant, Cloud wake, host/database suspension or savings claim is made. - **Fresh exact-head Greptile 5/5 and all CI are required.** Actual failing Ponytail findings must be repaired. A missing/stopped/paused turn without a verdict is an authorized nonblocker; an active no-verdict review waits at most 30 minutes from its original request/start (25-minute monitor cutoff). Older scores and parent checks are not accepted as current passes. - **Preserved inherited CI failure:** prior head `36f43922` had exact-head Greptile/Ponytail 5/5 but CI run `40708566195` server6 job `122887395220` failed the retention fairness assertion (2,291 passed/one failed/one skipped in that shard). A fixed final UUID reproduces the wrong expectation that an undiscovered page-two row is already deleted. The corrected test controls time, expects that row to remain before discovery and requires its deletion in the next fair batch, then all remaining history and idle SQL. The existing five-second wait budget and global timeout are unchanged. All 13 retention lifecycle cases pass; bypassing company rotation fails the corrected assertion, and the original shared helper is restored byte-for-byte. Earlier 223-case Teams acceptance remains valid for unchanged Teams source and is not relabeled as a new complete repository run. [Full CI repair ledger](https://github.com/paperclipai/paperclip/blob/9e84cfab1f795279c8cec413a946631f1d781bd2/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md#inherited-retention-ci-fairness-repair-october-28). Fresh CI and exact-head reviews are required for this test-only follow-up; old scores are stale. - **Current merge gates:** all 58 checks are now clean for this exact head and Greptile is 5/5 with no actionable findings. The current Ponytail turn is paused without a verdict, an authorized nonblocking state. Workflow `40712464720` attempt 1 independently hit npm `ECONNRESET` during the serialized8 job's tool self-installer, before its tests ran. Exactly one targeted job retry was requested after the original workflow finished; attempt 2 completed successfully on the unchanged head. No code, dependencies, workflow, timeout or review timer changed for that retry. The original setup failure is retained in its [job log](https://github.com/paperclipai/paperclip/actions/runs/40712464720/job/122901243926); the successful replacement is [job 122910350074](https://github.com/paperclipai/paperclip/actions/runs/40712464720/job/122910350074). ## Risks - Missed supported producer signals could delay expiry until restart. Real insertion, cancellation, consent transitions, caller-owned transactions and Board publication are covered. - Corrupt or locked rows must retain retry without starving healthy work. Paging, fairness, atomic rollback, redaction, drain/resume and shutdown tests cover these paths. - Independent SQL writers and other processes need their own notification contract. No persistent DB listener is added. - Attempt recovery still scans periodically until the next slice. A local deadline is not a durable Cloud alarm. This PR does not claim whole-instance sleep. - This slice is reconciled with its landed parent; current-head remote checks/review and live-master composition proof still gate merge. Later stack duties remain separate PRs. ## Model Used OpenAI GPT-6 through Codex, with repository inspection, tool use and code execution. The exact serving model ID and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either linked existing issues or described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/idle-teams-consent-expiry.md | 72 ++++++ ...2026-10-24-recurring-server-sleep-stack.md | 13 + ...-10-25-teams-consent-expiry-sleep-audit.md | 219 ++++++++++++++++ .../app-private-hostname-gate.test.ts | 6 + .../chat-channels.integration.test.ts | 34 +++ .../__tests__/voice-inbound-retention.test.ts | 20 +- server/src/app.ts | 1 + server/src/services/chat-channels.ts | 3 +- .../microsoft-teams/consent-expiry-work.ts | 64 +++++ .../microsoft-teams/file-delivery.ts | 34 +-- .../chat-teams-file-transfers.test.ts | 238 ++++++++++++++++++ .../src/services/chat-teams-file-transfers.ts | 81 +++--- .../services/delivery-work-notifications.ts | 1 + 13 files changed, 714 insertions(+), 72 deletions(-) create mode 100644 doc/idle-teams-consent-expiry.md create mode 100644 doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md create mode 100644 server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts diff --git a/doc/idle-teams-consent-expiry.md b/doc/idle-teams-consent-expiry.md new file mode 100644 index 00000000000..6faa09fca42 --- /dev/null +++ b/doc/idle-teams-consent-expiry.md @@ -0,0 +1,72 @@ +# Teams consent expiry and idle work + +Unused Teams file consent now uses the shared delivery coordinator and scheduler. +Existing `chat_teams_file_transfers` rows own its state. There is no new table, +independent timer, provider request or external-wake capability. + +## Obligation and trigger + +Only `consent_pending`, `awaiting_consent` and `upload_pending` rows are eligible. +Each uses its saved `expiresAt`. Registered startup reconstruction yields after +one 100-row UUID page per scope refresh and resumes through the shared scheduler. +Registration resets the owner’s cursors so a partial scan restarts safely. The +explicit one-shot inventory method still returns the complete inventory. Committed writes reload the affected transfer ID. The service registers +intent before insertion and its central versioned update, within the existing +transaction. Nested savepoints wait for outer settlement. The shared work-signals +protocol retains ambiguous outcomes; callers do not manage recovery wrappers. + +Configured endpoints, completed history, unknown outcomes and active transport +attempts do not keep this worker scanning. Pending consent retains its actual +future deadline. PostgreSQL submillisecond timestamps round up to the next +millisecond; checking a deadline does not extend it. + +## Expiry and recovery + +At the deadline, the worker reloads the current row and delegates to the existing +Teams protocol owner. That owner locks publication then transfer, verifies the +current authority digest, rechecks phase and expiry, and updates the transfer +and publication projection in the same transaction. It does not send a consent +card, upload data or send a file-info card. Lock waits remain bounded. A changed +phase is re-evaluated; a deleted or settled row removes the cached obligation. + +A failed load, lock or projection retains retry work. Corrupt rows do not starve +other transfers or companies. Errors emitted by this worker are generic and do +not contain SQL parameters, private transfer envelopes or provider capabilities. +The shared worker owns coalescing, batching, dirty hints, drain/resume and shutdown. +Shutdown joins an in-flight expiry; restart reconstructs the saved deadline. + +The explicit `expireAndRecover` diagnostic method retains combined behavior. +Production periodic maintenance now selects and executes abandoned attempts only. +Attempt recovery is a separate remaining migration. Unknown provider outcomes +remain quarantined; this change does not authorize a resend. + +## Ownership and limitations + +Provider phase selection and expiry live in the Teams module and existing Teams +transfer service. Shared app wiring only registers the owner and lifecycle gate. +The queue topic is a notification namespace, not a new durable queue. Other +providers and the shared scheduler receive no Teams protocol policy. + +Supported producers use the process's owning DB client. Independent SQL writers +or another process require their own wake contract. On-use authorization remains +with the protocol. Cloud eligibility and external alarm coverage are unchanged: +a future in-process deadline cannot wake stopped compute. This proves the changed +worker can become quiet, not fleet savings or complete instance suspension. + +## Verification + +Real PostgreSQL protocol cases cover all three phases, producer activation, +rollback and held outer commit, restart, pre-deadline quiet, timestamp precision, +projection rollback, fairness, drain/resume and shutdown. Production integration +registers the worker before Board publication and proves expiry cancels the +publication without another provider call. Existing authority, publication and +consent suites remain covered. See the [audit and failure ledger](plans/2026-10-25-teams-consent-expiry-sleep-audit.md). + + +Current local acceptance covers 223 distinct cases: 209 across five complete +protocol/publication/consent/runtime/wiring suites, plus 14 selected real Board +producer and Teams-authority integration cases. The other 1,580 integration +cases were skipped. A real 105-row regression proves bounded startup, same-owner +restart, complete saved-deadline expiry, and quiet SQL. A guarded missing-reset +mutation fails that regression. See the audit for exact revisions, earlier +fixture failures, current-parent integration status and deployment limits. 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 e51747d7775..9f5cdbb750b 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1809,3 +1809,16 @@ notifications and startup; those saves select zero rows and preserve the deadlin The [report notification ledger](2026-10-25-voice-call-report-sleep-audit.md#configuration-scope-review-repair-october-28) records 303 passing cases, types/build, the narrower contract and limits. Fresh exact-head review and CI are required after the repair. + + +## Slice 26: Teams unused-consent expiry + +The Teams owner now selects existing unused-consent phases and their saved expiry +through scoped deadline work. Production maintenance retains only abandoned +attempts; these are the next distinct reason. No new table or Cloud alarm policy +is included. The [audit](2026-10-25-teams-consent-expiry-sleep-audit.md) records +222 unique selected passing cases, workspace types/build, and initial fixture +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. diff --git a/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md b/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md new file mode 100644 index 00000000000..5a1e969678a --- /dev/null +++ b/doc/plans/2026-10-25-teams-consent-expiry-sleep-audit.md @@ -0,0 +1,219 @@ +# Teams unused-consent expiry sleep audit + +Baseline: voice call-report PR #17586 at 250fca6cff. +Branch: codex/sleep-26-teams-consent-expiry. +Status: implemented and locally verified; publication and remote gates pending. + +## Separate reasons + +The Teams maintenance callback in microsoft-teams/file-delivery.ts runs from +processPublicationMaintenance and repeats a database selection every 30 seconds. +It combines two obligations. This slice owns only unused consent expiry: +consent_pending, awaiting_consent and upload_pending rows become expired at their +existing expiresAt. Abandoned consent_sending/uploading/file_info_sending attempts, +and conflict rows with a retained attempt expiry, belong to the next slice. +Unknown outcomes without a live deadline are passive operator-review evidence; +expiry must not convert them into retries or manufacture delivery receipts. + +## Owner and persistence + +chatTeamsFileTransfers already owns the phase, expiry, company, endpoint, +publication, version, authority digest and encrypted private state. No schema or +new queue table is needed. teamsFileTransferService is the only production insert +and update owner found by the initial writer search. Its issue() insertion and +central update() helper are the two notification boundaries. Caller-owned +resolveInTransaction uses that helper; its signal must wait for outer commit. +Deletion should simply remove a cached entry on the next owned deadline; it must +not retain a perpetual scan. + +The existing lock order is publication -> transfer -> projection policy/source +locks. Expiry rechecks eligibility after those locks and performs no provider +I/O. Integrity checks and projection in the same transaction must remain intact. +Maintenance lock waits are bounded to 250ms, with no-wait publication/transfer +locking. A bad or locked row must not starve other companies or later rows. +Never log private envelopes, SQL parameter errors, provider capabilities or tokens. + +## Intended migration + +- Reuse scoped deadline work, with paged startup reconstruction and transfer-ID + hints from committed writes. Register the worker through the Teams owner; + app wiring only supplies shared lifecycle and admission gates. +- Load only the three unused phases, with ceil(expiresAt milliseconds) so a + PostgreSQL submillisecond expiry cannot be executed early or spin. +- Recheck and expire one transfer through its owning protocol method, preserving + atomic publication projection and no-provider-I/O semantics. Result loading + removes settled/changed rows; failed reconstruction or expiry retains retry. +- Keep explicit manual expireAndRecover compatibility for tests/diagnostics. + Production aggregate maintenance must select and execute only abandoned + attempts after this slice. It must not keep expiring unused consent as a hidden + second owner. That remaining attempt scan is intentionally still pending. +- Preserve configured idle, completed history, unknown outcome and active-attempt + quiet for this worker. Real outstanding consent remains a future obligation. +- Do not change authority-on-use checks, transport execution, Cloud wake coverage + or whole-instance sleep eligibility. + +## Verification plan + +Use the existing real-PostgreSQL Teams protocol fixture, and exercise real issue, +process/consent and cancellation/resolution writers. Cover saved expiry and +restart, first committed transfer, nested commit/rollback, no early reads, +changed phase at claim, provider-call absence, atomic projection rollback, +private-error redaction, lock contention/fair backlog, drain/resume, shutdown, +and submillisecond deadlines. Existing protocol, publication and consent tests +must continue to pass. Exercise production registration and ensure old maintenance +cannot expire consent. Full stack checks and master reconciliation remain required. + +## Other outstanding gates + +#17586 is published, attached, and locally verified; remote CI/reviews are pending. +#17580's company deletion policy finding and #17490's E2E cancellation failure are +still unresolved. No original recurring-server obligation is removed from scope. + +## Implementation and acceptance + +The Teams owner now registers a scoped deadline worker. The existing service +signals before issue/update writes, including caller-owned resolution transactions. +Startup pages eligible rows; transfer hints reload only that primary key. The +owner rechecks expiry with the original publication/transfer lock order and +atomic projection. The old periodic selection and execution now cover attempts +only. No schema or external wake contract changes are included. + +Local verification at the slice-26 implementation: + +- 65 transfer protocol cases pass on a fresh external PostgreSQL database. + These include 102-entry paging/fairness with one corrupt row, shutdown joining + a held projection, the three consent phases, precision, rollback, restart, + drain/resume, private-error redaction and the remaining maintenance boundary. +- The earlier five-suite run passed 206 cases: then-63 transfer cases and 143 + publication, consent, consent-runtime and app-wiring cases. The final transfer + run adds two cases. The latest app-wiring assertions pass all 16 cases again. +- 14 selected real production integration cases pass. This includes registration + before an actual Board publication, post-commit wake, saved deadline, pre-expiry + SQL quiet, atomic expiry/publication cancellation, no extra provider requests, + and the existing Teams current-authority/Board-transfer group. The other 1,411 + integration cases were skipped, not passed. +- Together these are 222 unique selected cases. Workspace `pnpm -r typecheck` + and `pnpm build` pass. Server typechecking excludes test files; it is not proof + that the test files were statically checked. +- Local full `pnpm test:run` remains unclaimed because this host's embedded + PostgreSQL setup failed earlier in the stack. Full remote gates are required. + No live Teams tenant or deployed host-suspension qualification is claimed. + +Failure ledger: the first protocol run had two invalid fixtures. A direct phase +mutation set attempt expiry without its required paired attempt ID; another +assertion treated the public ISO expiry string as a Date. Both were corrected, +then all five suites passed on a fresh database. The first production expiry +assertion expected a failed publication; the existing correct expiry projection +is cancelled. The corrected test and the authority group pass. These were test +expectation/setup errors, not weakened production contracts or hidden retries. + +Evidence logs: `/tmp/sleep26-teams-tests.log` (initial two fixture failures), +`/tmp/sleep26-teams-fixed-tests.log` (206 pass), +`/tmp/sleep26-teams-final-tests.log` (65 pass), +`/tmp/sleep26-producer-tests.log` (initial projection-state assertion), +`/tmp/sleep26-production-acceptance.log` (14 pass), +`/tmp/sleep26-final-wiring.log` (16 pass), and +`/tmp/sleep26-workspace-{types,build}.log` (pass). + +Parent #17586 at 250fca6cff has passing Actions, security and Greptile checks; +Paperclip Review is still running at the latest inspection. This does not clear +older stack findings or imply merge authority. Refreshed master is b7dc594bfa; +final whole-stack reconciliation is still required. + + +## Bounded reconstruction and current-parent acceptance — October 28 + +The published Teams head `aa082e4679` was inspected after the report parent was +prepared on landed master. Its worker's registered load still looped through +all pages in one scheduler pass. This acceptance repairs that existing slice; +it does not introduce another queue, scheduler, database change or duty. + +A local normal merge prepares the slice on report candidate `c28de398`. The +explicit-base merge had one stack-plan append conflict. The complete report +plan and complete Teams appendix are both preserved; all source blobs match +the clean explicit-base composition. An initial literal patch comparison +rejected shifted queue line numbers (47 to 49) and blob IDs, then an overly +broad merge whitespace check flagged an inherited runner report-lock patch. +The actual patch content and owned-scope whitespace check pass; inherited +upstream files were not altered. Actual landed-parent reconciliation and fresh +remote gates are required before publication and merge. + +The new 105-transfer PostgreSQL regression failed against the original reader: +startup selected 105 rows instead of yielding after 100. The worker now splits +one-page selection from complete explicit inventory reads. Registered refresh +returns the existing shared `{rows, complete}` contract and retains per-scope +cursors; registration clears them. Restarting the same service halfway through +a scan reconstructs all 105 saved deadlines, expires them at the original time, +and leaves no recurring SQL before the deadline or after settlement. No consent +card or upload is sent. Removing only cursor reset makes the restart read five +rows instead of 100; original source was restored before final acceptance. + +The first regression fixture faked all timers before establishing its first +database connection and timed out, including teardown. That was not the intended +row-budget failure and is not accepted as regression evidence. Seeding now uses +real network timers with only Date frozen, then controls scheduler timers after +seeding. The corrected unchanged-reader run fails at 105 versus 100, and the +fixed worker passes. Both the failed setup and actual regression remain recorded. + +Current acceptance: all 209 cases in the complete transfer (66), publication, +consent, consent-runtime and app-wiring suites pass on a new migrated fixture; +14 selected real Board-producer and Teams-authority integration cases pass on a +separate new migrated fixture (1,580 others skipped). Total 223 distinct cases; +the targeted red/green case is not added again. Server production typecheck, +build and module-boundary checks pass. The new source remains within the +existing Teams owner and existing test file. No schema, shared-worker contract, +workflow or lockfile change is introduced. The old 222-case evidence remains +historical. This does not claim a complete local repository suite, a live Teams +tenant, external Cloud wake, physical suspension or cost savings. + +Logs: `/tmp/sleep26-bounded-reconstruction-red.log` (invalid timer setup), +`/tmp/sleep26-bounded-reconstruction-actual-red.log` (105 vs 100), +`/tmp/sleep26-bounded-reconstruction-green.log`, +`/tmp/sleep26-bounded-cursor-mutant-red.log` (5 vs 100 after restart), +`/tmp/sleep26-bounded-final-protocols.log`, +`/tmp/sleep26-bounded-final-producers.log`, +`/tmp/sleep26-bounded-final-types.log`, `/tmp/sleep26-bounded-final-build.log`, +and `/tmp/sleep26-bounded-final-boundaries.log`. + +Current stack state supersedes the historical outstanding-gates paragraph: +cleanup and retention have landed, including the repaired legacy ownership case +and the explicitly accepted company-deletion ownership policy. The report parent +has a documentation-only follow-up after `c28de398` and still needs its own latest +CI/reviews. This local Teams preparation must preserve that actual landed parent +before publication; it is not authority to merge into an unmerged feature branch. + + +## Inherited retention CI fairness repair — October 28 + +Current Teams head `36f43922` received exact-head Greptile 5/5 and Ponytail 5/5, +but CI run `40708566195`, server shard 6/12 job `122887395220`, failed the existing +voice-retention case `pages through more than one batch of history across +companies`. The shard had 2,291 passed tests, one failed and one skipped. Its +initial assertion expected the other company's row already deleted after one +startup batch, even when its random ID put it beyond the first 100-row inventory +page. This CI failure is not accepted as green evidence. + +Placing the other company at a fixed final UUID reproduces the exact assertion. +The fixture now uses a supplied ID for both insertion and its transaction hint. +The test controls scheduler time, requires that row to remain before discovery, +then requires its deletion in the next fair batch after page-two discovery. It +also checks the first 25 deletions leave exactly 80 eligible rows, drains all +remaining history, and verifies SQL quiet. The existing five-second assertion +wait budget and global test timeout are unchanged. This corrects the acceptance +expectation without changing retention, Teams, or shared scheduler behavior. + +The complete 13-case retention suite passes. Its run spent 96.66 seconds in +module transforms but all actual cases completed successfully; no timeout was +increased and this was not counted as a failed test or silently retried. A +guarded mutation bypassing company rotation fails the new post-discovery fair +batch assertion. The original shared helper was restored byte-for-byte. The +previous 223-case Teams acceptance remains evidence for unchanged Teams source; +these 13 retention cases are a separate CI-repair acceptance group. Earlier +current-head Greptile/Ponytail scores become stale after publishing this test-only +follow-up, and fresh CI must pass independently. No old-job retry or waiver is +claimed. + +Logs: `/tmp/sleep26-36f-server6-job.log`, +`/tmp/sleep26-retention-fairness-red.log`, +`/tmp/sleep26-retention-fairness-green.log`, and +`/tmp/sleep26-retention-fairness-mutant-red.log`. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index bc4f3461e03..4c798afba0a 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -79,6 +79,12 @@ describe("createChatReconciliationCoordinator", () => { expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); expect(maintenance).not.toContain("chatChannels.voice."); 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"); }); 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 ceecf407f5e..5e8c692955b 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1218,6 +1218,40 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { companyId, assignedAgentId, replacementAgentId }; } + it("expires Teams consent from the committed Board transfer without periodic maintenance", async () => { + const context = await teamsFileAuthorityFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const settle = () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + // Register before the ordinary Board producer, with no transfer rows yet. + await context.service.registerTeamsConsentExpiryWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + const attachment = await context.createFile(); + const publication = await context.service.publishBoardMessage(context.endpoint.id, context.conversation.id, + "A file awaiting consent", `teams-expiry-${randomUUID()}`, "owner-user", [attachment.id]); + const [transfer] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.publicationId, publication.id)); + expect(transfer.phase).toBe("awaiting_consent"); expect(context.consentCards).toHaveLength(1); + await settle(); expect(coordinator.nextWakeAt()).toBe(transfer.expiresAt.getTime()); + const reads = vi.spyOn(db, "select"); + try { await vi.advanceTimersByTimeAsync(10_000); expect(reads).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); } + await vi.advanceTimersByTimeAsync(transfer.expiresAt.getTime() - Date.now() + 1000); await settle(); + const [expired] = await db.select().from(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.id, transfer.id)); + const [projected] = await db.select().from(chatPublications).where(eq(chatPublications.id, publication.id)); + expect(expired.phase).toBe("expired"); expect(projected.state).toBe("cancelled"); + expect(coordinator.nextWakeAt()).toBeNull(); + expect(context.consentCards).toHaveLength(1); expect(context.fileCards).toEqual([]); + expect(context.uploadRequest).not.toHaveBeenCalled(); expect(errors).toEqual([]); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await retirePublicationFixture(context.service, context.endpoint.id); + } + }); + it("retries a colliding company fixture prefix and creates a distinct complete fixture", async () => { const existing = await seedCompany(); const [original] = await db.select().from(companies).where(eq(companies.id, existing.companyId)); diff --git a/server/src/__tests__/voice-inbound-retention.test.ts b/server/src/__tests__/voice-inbound-retention.test.ts index 306e43e9408..6bdedd256d5 100644 --- a/server/src/__tests__/voice-inbound-retention.test.ts +++ b/server/src/__tests__/voice-inbound-retention.test.ts @@ -56,7 +56,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); async function admission(patch: Partial = {}, writer = db) { - const id = randomUUID(); + const id = patch.id ?? randomUUID(); return writer.transaction(async tx => { await notifyVoiceInbound(tx, { kind: "admission", id }); return (await tx.insert(chatVoiceInboundCalls).values({ id, companyId, endpointId, providerSessionId: randomUUID(), state: "ended", cleanupConfirmedAt: new Date(), updatedAt: new Date(Date.now() - week - 1000), approvalCode: "123456", generation: 1, credentialFingerprint: "test", toolTokenHash: "a".repeat(64), requestFingerprint: id, expiresAt: new Date(Date.now() - week - 1000), ...patch }).returning())[0]; @@ -211,12 +211,22 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); }); it("pages through more than one batch of history across companies", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); for (let i = 0; i < 104; i++) await admission(); - const other = await fixture(); const last = await admission({ companyId: other.companyId, endpointId: other.endpointId }); + const other = await fixture(); const last = await admission({ id: "ffffffff-ffff-4fff-bfff-ffffffffffff", companyId: other.companyId, endpointId: other.endpointId }); const service = await start(); - expect(await row(last.id)).toBeUndefined(); - await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await service.load()).toHaveLength(0); }, { timeout: 5000 }); - await flush(); await quiet(); + expect(await row(last.id)).toBeDefined(); + expect(await service.load()).toHaveLength(80); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(async () => expect(await row(last.id)).toBeUndefined(), { interval: 0, timeout: 5000 }); + for (let pass = 0; pass < 20; pass++) { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt() !== null || idleWorkSnapshot().active === 0).toBe(true), { interval: 0, timeout: 5000 }); + } + await flush(); + expect(await service.load()).toHaveLength(0); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId))).toHaveLength(0); + expect(await row(last.id)).toBeUndefined(); await quiet(); }); it("retains failed deletion for a bounded retry and then becomes quiet", async () => { const call = await admission(); diff --git a/server/src/app.ts b/server/src/app.ts index 169cc9634f4..849985244af 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1301,6 +1301,7 @@ export async function createApp( chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerTeamsConsentExpiryWork(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 ab9dd9263e3..ef96c0788df 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 { teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance } = createTeamsFileDelivery(db, { + const { registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance } = createTeamsFileDelivery(db, { endpointRecord, runtimeContextForRecord, runtimeFor, runtimeContexts, localRuntimeEpoch, runtime, deriveTeamsFileTransferAuthority, publicationFiles, scheduleMessageProcessing, processPendingPublications, withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown, @@ -3498,6 +3498,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 { + registerTeamsConsentExpiryWork, registerCommandWork, registerReceiverWork, voice, diff --git a/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts b/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts new file mode 100644 index 00000000000..7a2a9f8e97d --- /dev/null +++ b/server/src/services/chat-channels/microsoft-teams/consent-expiry-work.ts @@ -0,0 +1,64 @@ +import { and, eq, gt, inArray, 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 unusedTeamsConsentPhases = ["consent_pending", "awaiting_consent", "upload_pending"]; +type Entry = { id: string; companyId: string; dueAt: number }; + +export function notifyTeamsConsentExpiry(tx: object, id: string) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTeamsConsentExpiry, id); +} + +/** Owns only unused consent deadlines, not transport attempts or unknown outcomes. */ +export function teamsConsentExpiryWork(db: Db, expire: (companyId: string, id: string) => Promise) { + const cursors = new Map(); + async function page(id?: string, cursor?: string, signal?: AbortSignal): Promise { + try { + signal?.throwIfAborted(); + const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId, + dueAt: sql`ceil(extract(epoch from ${chatTeamsFileTransfers.expiresAt}) * 1000)`, + }).from(chatTeamsFileTransfers).where(and(inArray(chatTeamsFileTransfers.phase, unusedTeamsConsentPhases), + id ? eq(chatTeamsFileTransfers.id, id) : undefined, + cursor ? gt(chatTeamsFileTransfers.id, cursor) : undefined)).orderBy(chatTeamsFileTransfers.id).limit(100); + signal?.throwIfAborted(); + return rows.map(row => ({ ...row, dueAt: Number(row.dueAt) })); + } catch { + throw new Error("Teams consent expiry selection is temporarily unavailable"); + } + } + async function load(id?: string, signal?: AbortSignal): Promise { + const entries: Entry[] = []; let cursor: string | undefined; + for (;;) { + const rows = await page(id, cursor, signal); + entries.push(...rows); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + } + async function refresh(id?: string, signal?: AbortSignal) { + const key = id ?? "*", rows = await page(id, cursors.get(key), signal), complete = rows.length < 100; + if (complete) cursors.delete(key); else cursors.set(key, rows[rows.length - 1].id); + return { rows, complete }; + } + 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 expire(current.companyId, current.id); } + catch { throw new Error("Teams consent expiry could not settle"); } + return (await load(entry.id, signal))[0] ?? null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + cursors.clear(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTeamsConsentExpiry, + decodeScope: (hint: string) => hint === "*" ? undefined : hint, + matches: (entry: Entry, id?: string) => !id || entry.id === id, + load: refresh, 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 9848222c9b9..396662a4e49 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,5 @@ +import { teamsConsentExpiryWork } from "./consent-expiry-work.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; import { createOutboundAttachments } from "../outbound-attachments.js"; import { createTeamsFileAuthority } from "./file-transfer-authority.js"; @@ -14,7 +16,6 @@ import { eq, inArray, lte, - or, } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -392,24 +393,11 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption .selectDistinct({ companyId: chatTeamsFileTransfers.companyId }) .from(chatTeamsFileTransfers) .where( - or( - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_pending", - "awaiting_consent", - "upload_pending", - ]), - lte(chatTeamsFileTransfers.expiresAt, new Date()), - ), - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_sending", - "uploading", - "file_info_sending", - "conflict", - ]), - lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()), - ), + and( + inArray(chatTeamsFileTransfers.phase, [ + "consent_sending", "uploading", "file_info_sending", "conflict", + ]), + lte(chatTeamsFileTransfers.attemptExpiresAt, new Date()), ), ); const protocol = teamsTransferProtocol(); @@ -417,7 +405,7 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption // its row count and lock waits and performs no provider I/O. for (const scope of scopes) { if (isShuttingDown()) break; - await protocol.expireAndRecover(scope.companyId, 100); + await protocol.recoverAbandonedAttempts(scope.companyId, 100); } } catch { // Selection errors are closed here too; private transfer envelopes and @@ -430,5 +418,9 @@ export function createTeamsFileDelivery(db: Db, options: TeamsFileDeliveryOption } }); } - return { teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication, scheduleTeamsFileMaintenance }; + 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 }; } diff --git a/server/src/services/chat-teams-file-transfers.test.ts b/server/src/services/chat-teams-file-transfers.test.ts index ad6b5084c17..0dd72d0f387 100644 --- a/server/src/services/chat-teams-file-transfers.test.ts +++ b/server/src/services/chat-teams-file-transfers.test.ts @@ -1,3 +1,6 @@ +import { teamsConsentExpiryWork } from "./chat-channels/microsoft-teams/consent-expiry-work.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "./task-admission.js"; import { createHash, randomUUID } from "node:crypto"; import { afterAll, @@ -329,6 +332,241 @@ suite( }; } + describe("scheduled unused consent expiry", () => { + 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(); transferIds.push(f.created.id); + f.opts.now = () => new Date(); + return f; + } + function worker(expire: (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 = teamsConsentExpiryWork(db, expire); + 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("bounds reconstruction and resets its cursor on same-owner restart", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + const f = await transfer(), due = Date.now() + 300_000; + await db.update(chatTeamsFileTransfers).set({ expiresAt: new Date(due) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + for (let index = 0; index < 104; 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: `expiry-restart:${publicationId}`, + payload: { text: "Same authorized file", attachmentIds: [f.a.attachmentId] } }); + const a = { ...f.a, publicationId }; + const created = await teamsFileTransferService(db, { ...f.opts, authorize: async () => a }).issue(a, new Date(due)); + transferIds.push(created.id); + } + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: false }); + const counts: number[] = [], select = db.select.bind(db); + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + const selection = select(...args), from = selection.from.bind(selection); + vi.spyOn(selection, "from").mockImplementation(table => { + const query = from(table); + if (table === chatTeamsFileTransfers) { + const execute = query.execute.bind(query); + vi.spyOn(query, "execute").mockImplementation(async () => { + const rows = await execute(); counts.push(rows.length); return rows; + }); + } + return query; + }); + return selection; + }); + const expire = vi.fn(f.service().expireUnusedConsent), work = teamsConsentExpiryWork(db, expire); + const start = () => { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + return work.register(coordinator, () => enabled).ready; + }; + try { + await start(); expect(counts.reduce((sum, count) => sum + count, 0)).toBe(100); + await coordinator!.stop(); coordinator = undefined; + const before = counts.length; + await start(); expect(counts.slice(before).reduce((sum, count) => sum + count, 0)).toBe(100); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(due), { interval: 0 }); + expect(Math.max(...counts)).toBeLessThanOrEqual(100); reads.mockRestore(); + expect(await work.load()).toHaveLength(105); + await quiet(1000); expect(expire).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); + expect(expire).toHaveBeenCalledTimes(105); + expect(await db.select().from(chatTeamsFileTransfers).where(inArray(chatTeamsFileTransfers.id, transferIds))) + .toEqual(expect.arrayContaining([expect.objectContaining({ id: f.created.id, phase: "expired" })])); + expect(await work.load()).toEqual([]); expect(errors).toEqual([]); + await quiet(); expect(f.opts.postConsent).not.toHaveBeenCalled(); expect(f.opts.uploadRequest).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); } + }, 30_000); + it("is quiet with configured endpoints and cancelled history", async () => { + const f = await transfer(); await f.service().cancel(f.a.companyId, f.created.id); + const expire = vi.fn(f.service().expireUnusedConsent); + await worker(expire).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(expire).not.toHaveBeenCalled(); + }); + it.each(["consent_pending", "awaiting_consent", "upload_pending"])("restores the saved %s expiry without an early scan or provider effect", async phase => { + const f = await transfer(); + if (phase !== "consent_pending") await f.service().process(f.a.companyId, f.created.id); + if (phase === "upload_pending") await f.service().recordConsent(await f.event()); + expect((await f.read()).phase).toBe(phase); + const providerCalls = vi.mocked(f.opts.postConsent).mock.calls.length; + const due = (await f.read()).expiresAt.getTime(); + await worker(f.service().expireUnusedConsent).ready; await settled(); + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await coordinator!.stop(); + await worker(f.service().expireUnusedConsent).ready; + expect(coordinator!.nextWakeAt()).toBe(due); await quiet(); + await vi.advanceTimersByTimeAsync(due - Date.now() + 1000); await settled(); + expect((await f.read()).phase).toBe("expired"); expect(coordinator!.nextWakeAt()).toBeNull(); + await quiet(180_000); expect(f.opts.postConsent).toHaveBeenCalledTimes(providerCalls); + expect(f.opts.uploadRequest).not.toHaveBeenCalled(); expect(f.opts.postFileInfo).not.toHaveBeenCalled(); expect(errors).toEqual([]); + }); + it("reactivates from real issuance and disarms on cancellation", async () => { + let f: Awaited>; + await worker((companyId, id) => f.service().expireUnusedConsent(companyId, id)).ready; + await settled(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + f = await transfer(); await settled(); + expect(coordinator!.nextWakeAt()).toBe((await f.read()).expiresAt.getTime()); + await f.service().cancel(f.a.companyId, f.created.id); await settled(); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(180_000); + expect(f.opts.postConsent).not.toHaveBeenCalled(); + }); + it("keeps first committed work through drain without passive SQL", async () => { + enabled = false; + let f: Awaited>; + await worker((companyId, id) => f.service().expireUnusedConsent(companyId, id)).ready; + f = await transfer(); await quiet(90_000); + expect((await f.read()).phase).toBe("consent_pending"); + enabled = true; await vi.advanceTimersByTimeAsync(30_000); await settled(); + expect((await f.read()).phase).toBe("expired"); expect(coordinator!.nextWakeAt()).toBeNull(); + await quiet(180_000); expect(errors).toEqual([]); + }); + it("retains expiry after atomic projection failure and redacts its private error", async () => { + const f = await transfer(); let fail = true; + f.opts.project = async tx => { + await tx.update(chatPublications).set({ state: "cancelled" }).where(eq(chatPublications.id, f.a.publicationId)); + if (fail) throw new Error("PRIVATE-TEAMS-UPLOAD-CANARY"); + }; + await worker(f.service().expireUnusedConsent).ready; + await vi.advanceTimersByTimeAsync((await f.read()).expiresAt.getTime() - Date.now() + 1000); await settled(); + expect((await f.read()).phase).toBe("consent_pending"); + 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])).not.toContain("CANARY"); + 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("expired"); + expect((await db.select().from(chatPublications).where(eq(chatPublications.id, f.a.publicationId)))[0].state).toBe("cancelled"); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + it("waits for outer commit and ignores nested rollback", async () => { + const f = await transfer(); await db.delete(chatTeamsFileTransfers).where(eq(chatTeamsFileTransfers.id, f.created.id)); + await worker(f.service().expireUnusedConsent).ready; await settled(); + const due = new Date(Date.now() + 300_000); + await expect(db.transaction(async tx => { + await tx.transaction(async nested => teamsFileTransferService(nested as unknown as typeof db, f.opts).issue(f.a, due)); + await vi.advanceTimersByTimeAsync(10_000); expect(coordinator!.nextWakeAt()).not.toBe(due.getTime()); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await settled(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + let release!: () => void, inserted!: () => void; + const hold = new Promise(resolve => { release = resolve; }), ready = new Promise(resolve => { inserted = resolve; }); + const write = db.transaction(async tx => { + const row = await tx.transaction(async nested => teamsFileTransferService(nested as unknown as typeof db, f.opts).issue(f.a, due)); + inserted(); await hold; return row; + }); + try { + await ready; await vi.advanceTimersByTimeAsync(10_000); + expect(coordinator!.nextWakeAt()).not.toBe(due.getTime()); + } finally { release(); } + const row = await write; transferIds.push(row.id); await settled(); + expect(coordinator!.nextWakeAt()).toBe(due.getTime()); await quiet(); + }); + it("rounds PostgreSQL expiry up and leaves active or unknown attempts alone", async () => { + const f = await transfer(); + const whole = Date.now() + 60_000; + await db.update(chatTeamsFileTransfers).set({ expiresAt: sql`${new Date(whole).toISOString()}::timestamptz + interval '0.0005 second'` }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + vi.useFakeTimers({ toFake: ["Date"] }); + const work = teamsConsentExpiryWork(db, f.service().expireUnusedConsent); + const [entry] = await work.load(); expect(entry.dueAt).toBe(whole + 1); + vi.setSystemTime(whole); await work.check(entry); + expect((await f.read()).phase).toBe("consent_pending"); + vi.setSystemTime(whole + 1); expect(await work.check(entry)).toBeNull(); + expect((await f.read()).phase).toBe("expired"); + for (const phase of ["consent_sending", "uploading", "file_info_sending", "consent_unknown", "upload_unknown", "file_info_unknown", "conflict"]) { + await db.update(chatTeamsFileTransfers).set({ phase, attemptId: randomUUID(), attemptExpiresAt: new Date(0) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + const work = teamsConsentExpiryWork(db, f.service().expireUnusedConsent); + expect(await work.load()).toEqual([]); + } + }); + it("pages and drains more than 100 obligations without starving another company behind a corrupt row", async () => { + const f = await transfer(), other = await transfer(); + const due = new Date(Date.now() + 60_000); + 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: `expiry-page:${publicationId}`, + payload: { text: "Same authorized file", attachmentIds: [f.a.attachmentId] } }); + const a = { ...f.a, publicationId }; + const created = await teamsFileTransferService(db, { ...f.opts, authorize: async () => a }).issue(a, due); + transferIds.push(created.id); + } + await db.update(chatTeamsFileTransfers).set({ authorityDigest: "0".repeat(64) }).where(eq(chatTeamsFileTransfers.id, f.created.id)); + const { work, ready } = worker(f.service().expireUnusedConsent); await ready; + expect(await work.load()).toHaveLength(102); + await vi.advanceTimersByTimeAsync(due.getTime() - Date.now() + 1000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await other.read()).phase).toBe("expired"); + const remaining = await work.load(); expect(remaining.map(row => row.id)).toEqual([f.created.id]); + }, { timeout: 10_000 }); + expect((await f.read()).phase).toBe("consent_pending"); + expect(errors.length).toBeGreaterThan(0); expect(f.opts.postConsent).not.toHaveBeenCalled(); + }, 30_000); + it("joins an in-flight expiry projection during shutdown", async () => { + const f = await transfer(); let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + f.opts.project = async () => { entered(); await hold; }; + await worker(f.service().expireUnusedConsent).ready; + try { + await vi.advanceTimersByTimeAsync((await f.read()).expiresAt.getTime() - 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("expired"); + } finally { release(); } + }); + it("keeps the compatibility attempt sweep from expiring unused consent", async () => { + const f = await transfer(); + // Restore fixture time so the explicit maintenance operation sees it due. + f.opts.now = () => new Date(Date.parse(f.created.expiresAt) + 1); + expect(await f.service().recoverAbandonedAttempts(f.a.companyId)).toMatchObject({ scanned: 0, recovered: 0 }); + expect((await f.read()).phase).toBe("consent_pending"); + expect(await f.service().expireUnusedConsent(f.a.companyId, f.created.id)).toBe(true); + expect((await f.read()).phase).toBe("expired"); + }); + }); + it("uses the exact attachment-derived filename for an unnamed source across issuance and restart", async () => { const f = await fixture(60_000, null); const expected = `attachment-${f.a.attachmentId}`; diff --git a/server/src/services/chat-teams-file-transfers.ts b/server/src/services/chat-teams-file-transfers.ts index 4f88fb88e74..0f82870da06 100644 --- a/server/src/services/chat-teams-file-transfers.ts +++ b/server/src/services/chat-teams-file-transfers.ts @@ -1,3 +1,4 @@ +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"; import { z } from "zod"; @@ -388,6 +389,7 @@ export function teamsFileTransferService( row: Row, patch: Partial, ): Promise { + await notifyTeamsConsentExpiry(tx, row.id); const [next] = await tx .update(chatTeamsFileTransfers) .set({ ...patch, version: row.version + 1, updatedAt: now() }) @@ -533,6 +535,7 @@ export function teamsFileTransferService( binding, ), }; + await notifyTeamsConsentExpiry(tx, id); const [row] = await tx .insert(chatTeamsFileTransfers) .values({ @@ -859,9 +862,32 @@ export function teamsFileTransferService( if (!row) fail(); return summary(row); } + async function recoverTransfer(companyId: string, id: string, reason: "all" | "consent" | "attempts") { + return db.transaction(async (tx) => { + // Bound maintenance waits, including locks taken by atomic projection. + await tx.execute(sql`set local lock_timeout = '250ms'`); + const row = await locked(tx, companyId, id, true); + if (reason !== "consent" && (sending.includes(row.phase) || row.phase === "conflict") && + row.attemptExpiresAt && row.attemptExpiresAt <= now()) { + await update(tx, row, { + phase: row.phase === "conflict" ? "conflict" : unknownPhase[row.phase], + attemptId: null, + attemptExpiresAt: null, + reason: row.phase === "conflict" ? row.reason : "worker_outcome_unknown", + }); + return true; + } + if (reason !== "attempts" && unusedTeamsConsentPhases.includes(row.phase) && row.expiresAt <= now()) { + await update(tx, row, { phase: "expired", reason: "unused_consent_expired" }); + return true; + } + return false; + }); + } async function expireAndRecover( companyId: string, limit = 25, + reason: "all" | "attempts" = "all", ): Promise { if (!z.uuid().safeParse(companyId).success) fail(); const size = Number.isSafeInteger(limit) @@ -872,7 +898,8 @@ export function teamsFileTransferService( cursors = new Map(); expiryRecoveryCursors.set(db, cursors); } - const cursor = cursors.get(companyId); + const cursorKey = `${companyId}:${reason}`; + const cursor = cursors.get(cursorKey); const cutoff = now(); const eligible = and( eq(chatTeamsFileTransfers.companyId, companyId), @@ -881,14 +908,10 @@ export function teamsFileTransferService( inArray(chatTeamsFileTransfers.phase, [...sending, "conflict"]), lte(chatTeamsFileTransfers.attemptExpiresAt, cutoff), ), - and( - inArray(chatTeamsFileTransfers.phase, [ - "consent_pending", - "awaiting_consent", - "upload_pending", - ]), + reason === "all" ? and( + inArray(chatTeamsFileTransfers.phase, unusedTeamsConsentPhases), lte(chatTeamsFileTransfers.expiresAt, cutoff), - ), + ) : undefined, ), ); let rows: Array<{ id: string }>; @@ -927,45 +950,11 @@ export function teamsFileTransferService( for (const candidate of rows) { // Progress over rejected rows too. Neither corrupt evidence nor a held // publication lock can pin every later scan to the same oldest page. - cursors.set(companyId, candidate.id); + cursors.set(cursorKey, candidate.id); result.cursor = candidate.id; result.scanned++; try { - const recovered = await db.transaction(async (tx) => { - // The projection may need endpoint/conversation rows too. Bound - // only maintenance lock waits, including locks inside that callback. - await tx.execute(sql`set local lock_timeout = '250ms'`); - const row = await locked(tx, companyId, candidate.id, true); - if ( - (sending.includes(row.phase) || row.phase === "conflict") && - row.attemptExpiresAt && - row.attemptExpiresAt <= now() - ) { - await update(tx, row, { - phase: - row.phase === "conflict" ? "conflict" : unknownPhase[row.phase], - attemptId: null, - attemptExpiresAt: null, - reason: - row.phase === "conflict" - ? row.reason - : "worker_outcome_unknown", - }); - return true; - } else if ( - ["consent_pending", "awaiting_consent", "upload_pending"].includes( - row.phase, - ) && - row.expiresAt <= now() - ) { - await update(tx, row, { - phase: "expired", - reason: "unused_consent_expired", - }); - return true; - } - return false; - }); + const recovered = await recoverTransfer(companyId, candidate.id, reason); if (recovered) result.recovered++; } catch { // Keep the exact evidence and rollback any paired projection. Only @@ -1168,7 +1157,9 @@ export function teamsFileTransferService( recordConsent, process, get, - expireAndRecover, + expireAndRecover: (companyId: string, limit = 25) => expireAndRecover(companyId, limit), + recoverAbandonedAttempts: (companyId: string, limit = 25) => expireAndRecover(companyId, limit, "attempts"), + expireUnusedConsent: (companyId: string, id: string) => recoverTransfer(companyId, id, "consent"), cancel, canCancelConflict, resolveInTransaction, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 99b6f396067..4bc877804d3 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -49,6 +49,7 @@ export const DELIVERY_QUEUES = { voiceInboundRetention: "voice-inbound-retention", voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", + chatTeamsConsentExpiry: "chat-teams-consent-expiry", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs",