fix(server): recover Teams attempts from saved claim deadlines (#17592)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Teams file delivery saves a claim before each provider operation.
> - A crashed operation can leave that claim unfinished.
> - Periodic maintenance scans for abandoned claims even when none
exist.
> - This PR wakes recovery from committed changes and the saved claim
deadline.
> - Empty work becomes quiet while unknown effects remain safe to
review.

## Linked Issues or Issue Description

**What existing behavior does this improve?**
Teams abandoned file-transfer recovery. This is slice 27, stacked on
consent expiry #17590.

**Current behavior**
The remaining Teams maintenance callback scans every 30 seconds for
expired transport claims. It runs from shared publication maintenance.

**Proposed behavior**
The Teams owner reconstructs existing claim deadlines at startup and
reloads a transfer after its committed update. At expiry it clears the
claim and preserves the existing unknown-outcome or conflict state.
Normal receipt settlement removes the deadline.

**Reason and benefit**
Remove the last periodic Teams maintenance scan without resending a
possibly delivered card or file.

**Breaking changes**
No schema or public API change. Explicit diagnostic recovery methods
remain compatible. Consent expiry stays with its separate parent worker.

## What Changed

- Add a Teams-owned worker for consent_sending, uploading,
file_info_sending and conflict rows with a saved attempt expiry.
- Notify from the existing versioned update before writes. Preserve
transaction ownership and shared uncertain-write reconciliation.
- Expose a narrow per-transfer recovery method that uses the existing
lock order, integrity checks and atomic projection.
- Remove the periodic Teams maintenance callback and its
publication-maintenance invocation.
- Add real claim, receipt, conflict, backlog, restart, precision,
failure, drain and shutdown tests. Wire the owner through existing app
admission gates.
- Retain the existing signoff cancelled-run selection test and GitHub
inline retry-window race assertion.
- Document supported writers, remaining sleep limits and the failure
ledger in `doc/idle-teams-attempt-recovery.md` and its linked audit.

## Verification

Published head: `24c54399909105984e2a10f5869064984f69c472`, targeting
master after consent #17590 merged at
`4a89456bb3f9aef87cfe0e922e8b9d1a777a6c7e`. A normal ancestry merge
reconciles the landed squash with the previously verified child. The
complete 13-path binary patch is identical before and after
reconciliation, and all 30 unrelated master paths are preserved. This
adds no new worker behavior; the 239-case local evidence below belongs
to the verified unchanged patch. All 58 current-head checks are now
clean, including the complete required CI run. Greptile is 5/5 at this
exact head. The current Ponytail turn is paused without a verdict, which
is an explicitly authorized nonblocker; no failing review is waived.

The complete 13-path slice preserves the reviewed attempt implementation
and inherits the parent's report migration 0397, bounded consent worker
and retention CI test repair. No new schema, queue table, shared
scheduler contract, workflow or lockfile change is added by this duty.

- **239 distinct selected cases pass:** 221 across five complete
transfer/publication/consent/runtime/app-wiring suites, 16 selected real
Board/Teams-authority/GitHub-retry integration cases, and two delegated
Slack policy cases. Other chat integration cases (1,579) and
connection-management cases (81) were skipped, not passed. Each database
owner uses a new isolated PostgreSQL database.
- Real Board publication saves an active attempt deadline; its confirmed
receipt retires recovery, and SQL stays quiet afterward. The inherited
consent producer and authority cases still pass. Lifecycle coverage
exercises claim stages, unknown-outcome quarantine without repeating
provider I/O, conflicts, rollback, restart, precision, backlog/fairness,
failure recovery, drain and shutdown. The changed GitHub metadata/retry
race passes without duplicate assessments.
- The old server10 CI failure, run `40557802358` job `122390238707`, was
a delegated Slack invitation rate-window assertion (2,122 passed/one
failed/15 skipped in that shard). The verified parent already contains
its Date-only fixture fix. Both linked and invitation cases now pass
their full caller/target/shared-policy scenarios on a fresh server
fixture; no duplicate policy change or new owned file was added. That
old failure is preserved rather than called a pass.
- Server production typecheck, build and module-boundary checks pass.
Test files were executed; production typechecking does not statically
check them. The existing test-only signoff cancelled-run selection fix
is retained but was not executed locally in this acceptance; fresh
remote E2E gates remain required.
- [Current-parent acceptance and preserved failure
ledger](https://github.com/paperclipai/paperclip/blob/24c54399909105984e2a10f5869064984f69c472/doc/plans/2026-10-25-teams-attempt-recovery-sleep-audit.md#current-consent-parent-preparation-and-acceptance-october-28)
records current scope, commands, counts and limits. Earlier 235-case and
combined-stack counts belong to their stated revisions and are not added
to 239. The [stack
ledger](https://github.com/paperclipai/paperclip/blob/24c54399909105984e2a10f5869064984f69c472/doc/plans/2026-10-24-recurring-server-sleep-stack.md)
retains inherited failure history. No complete local repository suite,
live Teams/Slack tenant, external wake, physical suspension or savings
claim is made.
- **Exact-head Greptile 5/5 and all 58 checks are clean.** Actual
failing Ponytail findings must be repaired; a missing/stopped/paused
turn without a verdict is an authorized nonblocker. An active no-verdict
turn waits at most 30 minutes from its original request/start (25-minute
monitor cutoff). Older reviews and parent checks are not treated as
current passes.

## Risks

- Recovery must not infer that an expired claim means a provider effect
did not happen. It retains unknown outcomes and does not resend.
- Late receipts must still match current claim identity. Current-state
rechecks and existing protocol tests preserve this fence.
- Supported deployment uses one active writer process through its owning
DB client. Independent SQL and additional processes need a separate wake
contract.
- Cloud external alarm coverage and sleep eligibility are unchanged.
Other chat duties still run periodically until their own migrations.
- Voice-report parent review repairs are propagated: explicit
single-active-writer scope and an indexed report window. Thirty
report-worker/startup tests pass on the combined child. Fresh remote
checks, older stack findings and final master reconciliation remain
required.

## 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 10:51:16 -05:00
1 parent 2e5bbde8cc
commit 7d75ae0569
13 files changed
+620 -56

No files matched your search

+82
View File
@@ -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.
@@ -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.
@@ -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.
@@ -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", () => {
@@ -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<void>(resolve => { release = resolve; }), ready = new Promise<void>(resolve => { entered = resolve; });
context.controls.beforeConsentReceipt = async () => { entered(); await hold; };
let send: ReturnType<typeof context.service.publishBoardMessage> | 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;
+1
View File
@@ -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 () => {
+2 -2
View File
@@ -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,
@@ -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<boolean>) {
async function load(id?: string, signal?: AbortSignal): Promise<Entry[]> {
try {
const entries: Entry[] = []; let cursor: string | undefined;
for (;;) {
signal?.throwIfAborted();
const rows = await db.select({ id: chatTeamsFileTransfers.id, companyId: chatTeamsFileTransfers.companyId,
dueAt: sql<string>`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<Entry | null> {
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<typeof createDeliveryWorkCoordinator>, 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 };
}
@@ -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<typeof createDeliveryWorkCoordinator>, 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<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
return teamsAttemptRecoveryWork(db, (companyId, id) => teamsTransferProtocol().recoverAbandonedAttempt(companyId, id))
.register(coordinator, () => !isShuttingDown() && canRun());
}
return { registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, teamsTransferProtocol, handleTeamsFileConsent, tryProcessTeamsFilePublication };
}
@@ -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<typeof createDeliveryWorkCoordinator> | 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<ReturnType<typeof transfer>>, 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<boolean>) {
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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<void>(resolve => { release = resolve; }), ready = new Promise<void>(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<typeof createDeliveryWorkCoordinator> | undefined;
const transferIds: string[] = [], errors: unknown[] = [];
@@ -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<typeof chatTeamsFileTransfers.$inferInsert>,
): Promise<Row> {
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,
@@ -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",
+38 -2
View File
@@ -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;