From 623f1dc289a14a699ebbe3f09b94179d38fc2ae2 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:36:28 -0500 Subject: [PATCH] fix(server): schedule secret-proposal expiry from committed changes (#17446) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agents can propose secrets and bindings for approval. > - Pending proposals already contain their expiry and encrypted value. > - The server scans for expired proposals on every maintenance tick, even when none exist. > - Committed proposal changes and saved expiry times provide the required triggers. > - This PR uses those triggers and preserves scrubbing, approval checks, and recovery. > - Empty and completed proposal history no longer causes recurring expiry queries. ## Linked Issues or Issue Description Refs #17443. This is the second PR in the sleep migration stack and targets its branch. **What problem does this solve?** The periodic proposal scan contacts PostgreSQL when no proposal needs maintenance. That obstructs idle database behavior across tenant instances. **What is the proposed solution?** Use transaction-aware notifications for creation and resolution. Reconstruct the earliest pending expiry at startup. Run cleanup at that deadline, or retry while reconciliation remains unfinished. **Why is this useful?** It removes an idle scan while retaining security checks and physical ciphertext cleanup. No queue table or migration is required. ## What Changed - Added intent notifications before secret/binding creation, approval, terminal transitions, and linked interaction resolution. The existing outer transaction owns settlement. - Kept expiry selection, ciphertext scrubbing, dependent rejection, and authorization in the secret-proposal service. - Registered a startup and deadline worker in place of the aggregate maintenance sweep. Empty/terminal states disarm; overdue batches and failures retain retries. - Reused shared scoped deadline dispatch for secret expiry. The service owns company-filtered SQL and expiry policy. Custom-image expiry continues to use `registerDeadlineWork`. - Added real PostgreSQL producer, outer-transaction, concurrent-write, expiry, restart, interaction resolution, and drain/resume tests. Updated startup wiring fixtures and documented the supported writer/lifecycle contract. ## Verification - Landing reconciliation at `c10eb1bcf752b058713e05d3345e5cf38dd3fa72`: 160 tests pass across the complete secret-proposal, scoped deadline, company batch, coordinator, startup, custom-image service and custom-image worker suites; no cases skipped in the accepted runs. Server typecheck passes. This integrates master after #17443 landed; fresh remote CI and Ponytail review are required before merge. Current head: `5bbe17ad9ebc4573b08ed1a9323644ef9608ca09`. This remains a single-duty PR stacked on #17443. - Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5` with normal merges. Every branch passed generated-schema checks and its target-branch migration-order gate. Existing SQL is unchanged; the 25 stack migrations follow master’s new migration. - Duty-specific producer coverage, lifecycle tests, and earlier failure records: [2026-10-08-instance-sleep-and-scheduler-inventory.md](https://github.com/paperclipai/paperclip/blob/5bbe17ad9ebc4573b08ed1a9323644ef9608ca09/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md), [2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/5bbe17ad9ebc4573b08ed1a9323644ef9608ca09/doc/plans/2026-10-24-recurring-server-sleep-stack.md). - Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355 startup/cancellation/recovery/dispatch/preservation tests**, **75 connection-management tests**, and production server typechecking pass. These combined checks do not mean every intermediate PR independently ran the full repository suite. - Real master-to-stack database upgrade passes: **379 unchanged prior migration receipts plus 25 new migrations**, with the gateway responsibility fixture preserved. - Owning-branch checks: 84 runner installation/settlement cases plus runner typechecking and generated TypeScript/Rust profile agreement; 253 GitHub review workflow cases. Stop-attribution and held-worker readiness regressions failed before their fixes. - Failure accounting: the initial combined run passed 349 cases and failed six idle assertions in a reused preservation database containing old cancelled runs. Repeating the combined command with a fresh preservation database passed. Earlier invalid project selection, merge import/type errors, and the interrupted full local suite are not counted as passing checks. - **Latest-head CI and review remain required.** Prior CI findings include GitHub idle observation, route teardown, attention cleanup, touch-picker E2E, workspace health, and voice-cleanup timing; local passes do not erase those failures. No full local repository-suite, deployed sleep, fleet cost, or live Copilot qualification claim is made. ## Risks - Notifications cover supported writers sharing the database signal owner. Independent SQL writers and other process roots require a separate ingress contract; startup recovery alone is not live replica support. - The shared coordinator and scoped worker also serve later stack duties. Their defaults remain intact; existing voice, chat, native, and startup consumers are included in integrated verification. - Startup now checks overdue proposals immediately. During drain, physical cleanup waits; approval still rejects expired proposals synchronously. - Each due company pass remains bounded to 100 rows. Failed scopes back off from 30 seconds to five minutes; healthy overdue work keeps draining. Self-generated write hints coalesce and do not erase a failed sweep's backoff. - Heartbeat-disabled behavior and Cloud sleep-safety gates remain unchanged. This does not establish deployed suspension or external wake coverage. Tool review and gateway token cleanup remain separate slices. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## 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 (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) 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 - [ ] All Paperclip CI gates are green - [ ] 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-secret-proposal-expiry.md | 90 +++++ ...-instance-sleep-and-scheduler-inventory.md | 6 +- ...2026-10-24-recurring-server-sleep-stack.md | 4 +- .../src/__tests__/company-work-batch.test.ts | 29 ++ .../delivery-work-coordinator.test.ts | 52 +++ .../__tests__/scoped-deadline-work.test.ts | 39 +++ .../__tests__/secret-proposals-routes.test.ts | 314 +++++++++++++++++- .../server-startup-feedback-export.test.ts | 25 +- server/src/index.ts | 11 +- server/src/services/company-deletion.ts | 1 + server/src/services/company-work-batch.ts | 28 ++ .../src/services/custom-image-cleanup-work.ts | 21 +- server/src/services/deadline-work.ts | 36 ++ .../src/services/delivery-work-coordinator.ts | 16 +- .../services/delivery-work-notifications.ts | 1 + .../src/services/issue-thread-interactions.ts | 2 + server/src/services/scoped-deadline-work.ts | 99 ++++++ .../services/secret-proposal-expiry-work.ts | 29 ++ server/src/services/secret-proposals.ts | 30 +- 19 files changed, 792 insertions(+), 41 deletions(-) create mode 100644 doc/idle-secret-proposal-expiry.md create mode 100644 server/src/__tests__/company-work-batch.test.ts create mode 100644 server/src/__tests__/scoped-deadline-work.test.ts create mode 100644 server/src/services/company-work-batch.ts create mode 100644 server/src/services/deadline-work.ts create mode 100644 server/src/services/scoped-deadline-work.ts create mode 100644 server/src/services/secret-proposal-expiry-work.ts diff --git a/doc/idle-secret-proposal-expiry.md b/doc/idle-secret-proposal-expiry.md new file mode 100644 index 00000000000..a66747421fb --- /dev/null +++ b/doc/idle-secret-proposal-expiry.md @@ -0,0 +1,90 @@ +# Secret-proposal expiry and idle behavior + +Secret proposals already store `status`, `expiresAt`, and encrypted pending +values. This worker uses those records; it adds no queue table or migration. + +## Ownership and activation + +`secret-proposals.ts` still owns quotas, authority checks, approval, expiry, +ciphertext scrubbing, dependent binding rejection, audit records, and interaction +reflection. `expiryDeadlines(companyId?)` returns one saved deadline per pending +company. Normal write notifications reload only the affected company. Startup +reconstruction loads all companies once. The manual `nextExpiryAt()` query remains +available to existing callers. +Terminal history does not create a deadline. Approval still rejects an expired +proposal synchronously, even when maintenance is paused. + +The supported producers register `secretProposalExpiry` intent before their +writes in the owning transaction: + +- `createWithinQuota`: both secret and binding creation, after quota admission. +- `markApproved`: ordinary approval and prerequisite approval during a cascade. +- `transition`: rejection, withdrawal, expiry, and dependent binding rejection. +- Interaction resolution: rejection, withdrawal/cancellation, expiry, and failed + acceptance execution that resolves a linked pending binding. +- Company deletion: a company hint removes its cached deadline after deletion commits. + +Each producer includes its company ID. The remaining proposal write only attaches +an interaction ID and does not change expiry eligibility. Independent SQL writers +or another DB root are not supported live producers of these process-local signals; +startup reconstructs their persisted rows, but it is not a live replica contract. + +The shared database signal boundary defers notification until the outer transaction +settles. Nested savepoints, rollback, and uncertain commit outcomes retain the +existing protocol. Callers do not pass recovery counters or reconnect the database. + +## Scheduling and lifecycle + +`registerSecretProposalExpiryWork` runs once at startup while heartbeat scheduling +is enabled. It replaces the secret-proposal sweep in the aggregate maintenance +tick. Disabled heartbeat scheduling keeps its existing behavior. + +The worker reloads the affected company's saved minimum expiry after a committed +change. It does not sweep future work. At a due deadline it sweeps at most 100 +expired proposals for that company, then reads that company's next deadline. +Companies rotate fairly between bounded passes. Retained history creates no work. +No pending proposal, unresolved write, or failed reconciliation means no recurring +query. Unscoped legacy hints still reconstruct all companies. + +`scoped-deadline-work.ts` owns scope invalidation, cached deadlines, fair dispatch, +and failure backoff. It owns no SQL or proposal policy. Failed loads and failed +company sweeps back off independently from 30 seconds to five minutes. A new +write reloads that company without erasing its failed sweep's backoff. Other +companies keep their own deadlines, including already-due work. The coordinator's +optional `nextRunAtOnError` contract permits immediate eligible work only after +the worker has recorded per-item backoffs. Failed transaction reconciliation and +failed pending-state inspection retain the coordinator's normal retry. + +Secret expiry opts out of the shared helper's default one-second pass spacing so +newly learned short deadlines are not clamped. Other scoped consumers retain that +default. Custom-image expiry continues to use `deadline-work.ts` unchanged. No +shared scheduler or startup wiring interprets secret-proposal policy. + +A request stays pending until reconciliation succeeds. Changes arriving during a +pass, first writes during drain, and failures followed by drain retain a request. +Passive drain performs no proposal SQL. Resume reconciles retained requests; +restart reconstructs deadlines. Coordinator shutdown unsubscribes and waits for +active work. Deferred physical cleanup does not extend approval authorization. + +## Verification and limits + +The PostgreSQL route suite exercises actual service producers, nested outer commit +and rollback, quiet empty and terminal states, quiet future deadlines, automatic +ciphertext scrubbing and dependent-card resolution, startup reconstruction, +withdrawal, interaction rejection/failed execution, failure retry, and drain/resume. +It also verifies that expired approval fails while cleanup is paused. The existing +coordinator suite verifies fallback retry and uncertainty boundaries. New cases +record actual SQL to prove normal writes and due expiry stay company-scoped; they +also cover five-second deadlines, earlier replacement deadlines, company deletion, +and another company making progress while one sweep fails. A deterministic +shared-worker test covers a 250ms replacement deadline. + +The worker tests use a controlled clock and a real PostgreSQL database. Existing +route tests cover quotas, isolation, authorization, concurrency, and lifecycle +semantics. The test-only database URL must refer to a disposable migrated database: +the suite deletes its fixture tables. Ordinary CI uses its isolated embedded DB. + +This is worker quiet, not proof of host or database suspension. No Cloud alarm or +sleep-eligibility exemption is added. Existing sleep-safety checks remain in force; +a local expiry timer cannot wake a stopped host. Tool review expiry and gateway +session cleanup are separate migrations. diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index b92d7520ec0..9141878caa8 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -647,8 +647,10 @@ Longer polling intervals alone do not complete an item. visible within the agreed freshness contract. Source: [index.ts](../../server/src/index.ts), `tools.sweepConnectionHealth`. -- [ ] **JOB-21 — Secret proposals and expired action tokens.** Current: **30s** - expiry/cleanup queries. Change: enforce expiry synchronously on use and +- [ ] **JOB-21 — Secret proposals and expired action tokens (partially migrated).** + Secret-proposal expiry now uses transaction signals and saved expiry deadlines; + see [the lifecycle contract](../idle-secret-proposal-expiry.md). Tool review + recovery and gateway token cleanup still use **30s** queries. Change: enforce expiry synchronously on use and register required scrub/cleanup deadlines. **Complete when:** idle cleanup does not extend authorization or secret-retention periods. Sources: [secret-proposals.ts](../../server/src/services/secret-proposals.ts), 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 50f73222265..20221274e70 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -12,13 +12,13 @@ Do not mark an item done from a timer removal or a green check alone. | ID | Original checklist duty | Required slices / disposition | PR / evidence | | --- | --- | --- | --- | -| 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | In progress | +| 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | [#17443](https://github.com/paperclipai/paperclip/pull/17443), merged after CI and passing reviews | | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | -| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Pending | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret-proposal expiry in progress; tool reviews and gateway tokens pending | | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | diff --git a/server/src/__tests__/company-work-batch.test.ts b/server/src/__tests__/company-work-batch.test.ts new file mode 100644 index 00000000000..9fe1abc5a1d --- /dev/null +++ b/server/src/__tests__/company-work-batch.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import { takeCompanyWorkBatch } from "../services/company-work-batch.js"; + +describe("company work batch", () => { + it("shares capacity while preserving the owner's order within each company", () => { + const entries = [ + { companyId: "a", id: "a-oldest" }, { companyId: "a", id: "a-next" }, + { companyId: "a", id: "a-later" }, { companyId: "b", id: "b-due" }, + ]; + expect(takeCompanyWorkBatch(entries, 3)).toEqual({ + entries: [entries[0], entries[3], entries[1]], lastCompany: "a", + }); + }); + it("rotates between batches when more companies are due than fit in one batch", () => { + const entries = ["a", "b", "c", "d"].flatMap(companyId => [ + { companyId, id: `${companyId}-first` }, { companyId, id: `${companyId}-second` }, + ]); + const first = takeCompanyWorkBatch(entries, 2); + const remaining = entries.filter(entry => !first.entries.includes(entry)); + const second = takeCompanyWorkBatch(remaining, 2, first.lastCompany); + expect([...first.entries, ...second.entries].map(entry => entry.companyId)).toEqual(["a", "b", "c", "d"]); + expect(takeCompanyWorkBatch(remaining, 1, "z").entries[0].companyId).toBe("a"); + }); + it("finishes an empty or exhausted batch without inventing work", () => { + expect(takeCompanyWorkBatch([], 25, "gone")).toEqual({ entries: [], lastCompany: "gone" }); + const entry = { companyId: "only" }; + expect(takeCompanyWorkBatch([entry], 25).entries).toEqual([entry]); + }); +}); diff --git a/server/src/__tests__/delivery-work-coordinator.test.ts b/server/src/__tests__/delivery-work-coordinator.test.ts index 805aa3db4c4..2ec8054336f 100644 --- a/server/src/__tests__/delivery-work-coordinator.test.ts +++ b/server/src/__tests__/delivery-work-coordinator.test.ts @@ -59,6 +59,58 @@ describe("delivery work coordinator", () => { expect(s.onError).not.toHaveBeenCalled(); } finally { clock.mockRestore(); } }); + it("honors a future backlog deadline after failure while preserving the fallback retry", async () => { + const s = setup(), t = task(); + let next: number | null = Date.now() + 1000; + t.run.mockRejectedValueOnce(new Error("One item failed")); + await s.coordinator.register(DELIVERY_QUEUES.feedback, { ...t, nextRunAt: () => next }).ready; + expect(s.coordinator.nextWakeAt()).toBe(next); + expect(s.onError).toHaveBeenCalledOnce(); + next = null; + await vi.advanceTimersByTimeAsync(1000); + expect(t.run).toHaveBeenCalledTimes(2); + expect(s.coordinator.nextWakeAt()).toBeNull(); + }); + it("dispatches another due item after a worker records scoped failure backoffs", async () => { + const s = setup(), t = task(), now = Date.now(); + t.run.mockRejectedValueOnce(new Error("One scoped item failed")); + const nextRunAtOnError = vi.fn(() => now); + await s.coordinator.register(DELIVERY_QUEUES.feedback, { ...t, nextRunAtOnError }).ready; + expect(nextRunAtOnError).toHaveBeenCalledOnce(); + expect(s.coordinator.nextWakeAt()).toBe(now); + await vi.advanceTimersByTimeAsync(1); + expect(t.run).toHaveBeenCalledTimes(2); + expect(s.coordinator.nextWakeAt()).toBeNull(); + }); + it.each(["reconcile", "pending"])("does not consume scoped retry deadlines for a %s failure", async mode => { + const s = setup(), t = task(), now = Date.now(); + if (mode === "reconcile") { + s.failCommit(); + await expect(s.enqueue()).rejects.toThrow("lost commit reply"); + s.execute.mockRejectedValueOnce(new Error("Database unavailable")); + } else t.hasPending.mockRejectedValueOnce(new Error("Pending state unavailable")); + const nextRunAtOnError = vi.fn(() => now); + await s.coordinator.register(DELIVERY_QUEUES.feedback, { ...t, nextRunAtOnError }).ready; + expect(nextRunAtOnError).not.toHaveBeenCalled(); + expect(s.coordinator.nextWakeAt()).toBe(now + 5000); + expect(s.onError).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(4999); + expect(s.onError).toHaveBeenCalledOnce(); + }); + it.each(["overdue", "later", "unreadable", "draining"])("keeps failure backoff for a %s deadline", async mode => { + const s = setup(), t = task(), now = Date.now(); + t.run.mockImplementation(async () => { + if (mode === "draining") { s.canRun.mockReturnValue(false); s.canReconcile.mockReturnValue(true); } + throw new Error("Unavailable"); + }); + await s.coordinator.register(DELIVERY_QUEUES.feedback, { ...t, nextRunAt: () => { + if (mode === "unreadable") throw new Error("Deadline unavailable"); + return mode === "overdue" ? now : now + (mode === "later" ? 60_000 : 1000); + } }).ready; + expect(s.coordinator.nextWakeAt()).toBe(now + 5000); + await vi.advanceTimersByTimeAsync(4999); + expect(t.run).toHaveBeenCalledOnce(); + }); it("schedules maintenance without holding idle intent, and lets new work preempt it", async () => { const s = setup(), t = task(); let next: number | null = Date.now() + 60_000; diff --git a/server/src/__tests__/scoped-deadline-work.test.ts b/server/src/__tests__/scoped-deadline-work.test.ts new file mode 100644 index 00000000000..fda914dd13e --- /dev/null +++ b/server/src/__tests__/scoped-deadline-work.test.ts @@ -0,0 +1,39 @@ +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { afterEach, expect, it, vi } from "vitest"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../services/scoped-deadline-work.js"; + +let coordinator: ReturnType | undefined; +afterEach(async () => { await coordinator?.stop(); vi.useRealTimers(); }); + +it("preserves a subsecond replacement deadline when the owner does not require pass spacing", async () => { + vi.useFakeTimers(); + const onError = vi.fn(); + const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) }); + coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError }); + const now = Date.now(); + let dueAt = now + 60_000; + const check = vi.fn(async () => null); + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.secretProposalExpiry, + minimumPassIntervalMs: 0, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: (entry, companyId?: string) => companyId === undefined || entry.companyId === companyId, + load: async () => [{ id: "company", companyId: "company", dueAt }], + check, + }, () => true); + await worker.ready; + dueAt = now + 250; + worker.invalidate("company"); + await vi.advanceTimersByTimeAsync(1); + expect(coordinator.nextWakeAt()).toBe(dueAt); + await vi.advanceTimersByTimeAsync(248); + expect(check).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1); + expect(check).toHaveBeenCalledOnce(); + expect(coordinator.nextWakeAt()).toBeNull(); + await vi.advanceTimersByTimeAsync(60_000); + expect(check).toHaveBeenCalledOnce(); + expect(onError).not.toHaveBeenCalled(); +}); diff --git a/server/src/__tests__/secret-proposals-routes.test.ts b/server/src/__tests__/secret-proposals-routes.test.ts index 3354345e4f4..935fdb84f89 100644 --- a/server/src/__tests__/secret-proposals-routes.test.ts +++ b/server/src/__tests__/secret-proposals-routes.test.ts @@ -1,3 +1,8 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { registerSecretProposalExpiryWork } from "../services/secret-proposal-expiry-work.js"; +import { deleteCompany } from "../services/company-deletion.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { mkdirSync, rmSync } from "node:fs"; import os from "node:os"; @@ -5,7 +10,7 @@ import path from "node:path"; import express from "express"; import request from "supertest"; import { and, eq, sql } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { activityLog, agents, @@ -38,7 +43,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabase = resolveExternalTestDatabaseUrl("PAPERCLIP_SECRET_PROPOSAL_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabase ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("secret proposal routes", () => { @@ -50,9 +56,10 @@ describeEmbeddedPostgres("secret proposal routes", () => { beforeAll(async () => { mkdirSync(secretsTmpDir, { recursive: true }); process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = path.join(secretsTmpDir, "master.key"); + if (externalTestDatabase) { db = createDb(externalTestDatabase, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); return; } const started = await startEmbeddedPostgresTestDatabase("secret-proposal-routes"); stopDb = started.cleanup; - db = createDb(started.connectionString); + db = createDb(started.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }); afterEach(async () => { @@ -74,14 +81,14 @@ describeEmbeddedPostgres("secret proposal routes", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await stopDb?.(); if (previousKeyFile === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; else process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = previousKeyFile; rmSync(secretsTmpDir, { recursive: true, force: true }); }); - async function seedRun() { - const companyId = randomUUID(); + async function seedRun(identifier = "SEC-1", companyId = randomUUID()) { const agentId = randomUUID(); const heartbeatRunId = randomUUID(); const issueId = randomUUID(); @@ -113,7 +120,7 @@ describeEmbeddedPostgres("secret proposal routes", () => { id: issueId, companyId, title: "Needs credential", - identifier: "SEC-1", + identifier, status: "in_progress", responsibleUserId: "user-1", executionRunId: heartbeatRunId, @@ -221,6 +228,301 @@ describeEmbeddedPostgres("secret proposal routes", () => { return app; } + describe("expiry scheduling", () => { + let coordinator: ReturnType | undefined; + let proposals: ReturnType; + let fixture: Awaited>; + let draining: boolean; + const errors: unknown[] = []; + const expiryMs = 14 * 24 * 60 * 60_000; + + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + fixture = await seedRun(); + proposals = createSecretProposalsService(db); + draining = false; + errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); + coordinator = undefined; + vi.useRealTimers(); + expect(errors).toEqual([]); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerSecretProposalExpiryWork(coordinator, proposals, () => !draining).ready; + } + async function settle(ms = 60_000) { + await vi.advanceTimersByTimeAsync(ms); + if (!draining) await vi.waitFor(() => expect(idleWorkSnapshot().active === 0).toBe(true), { timeout: 5000 }); + } + async function quiet() { + await settle(); + expect(coordinator!.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"); + await settle(60 * 60_000); + expect(reads).not.toHaveBeenCalled(); + reads.mockRestore(); + } + const create = (service = proposals) => service.createSecret({ ...fixture, registerForRedaction: () => {} }, { + name: "dev/scheduled/secret", value: "expiry-test-only", justification: "Test durable expiry", + }); + const bind = (secretProposalId: string) => proposals.createBinding(fixture, { + secretProposalId, configPath: "access.SCHEDULED", justification: "Test dependent expiry", + bindingTargetPolicy: "self_and_reports", + }); + + async function createWithExpiry(expiresAt: Date) { + return db.transaction(async tx => { + const proposal = await create(createSecretProposalsService(tx as unknown as typeof db)); + const [updated] = await tx.update(companySecretProposals).set({ expiresAt }) + .where(eq(companySecretProposals.id, proposal.id)).returning(); + return updated; + }); + } + + it("honors a saved five-second expiry without clamping it to recovery spacing", async () => { + const delay = 5000; + await start(); await quiet(); + const expiresAt = new Date(Date.now() + delay); + const proposal = await createWithExpiry(expiresAt); await settle(1); + expect(coordinator!.nextWakeAt()).toBe(expiresAt.getTime()); + await settle(delay / 2); + expect(await proposals.getById(fixture.companyId, proposal.id)).toMatchObject({ status: "pending" }); + await settle(delay); + expect(await proposals.getById(fixture.companyId, proposal.id)).toMatchObject({ status: "expired", valueCiphertext: null }); + await quiet(); + }); + + it("moves to an earlier committed proposal deadline without losing the later one", async () => { + await start(); await quiet(); + const later = await createWithExpiry(new Date(Date.now() + 120_000)); await settle(1); + expect(coordinator!.nextWakeAt()).toBe(later.expiresAt.getTime()); + const earlier = await createWithExpiry(new Date(Date.now() + 5000)); await settle(1); + expect(coordinator!.nextWakeAt()).toBe(earlier.expiresAt.getTime()); + await settle(6000); + expect(await proposals.getById(fixture.companyId, earlier.id)).toMatchObject({ status: "expired", valueCiphertext: null }); + expect(await proposals.getById(fixture.companyId, later.id)).toMatchObject({ status: "pending" }); + expect(coordinator!.nextWakeAt()).toBe(later.expiresAt.getTime()); + await settle(120_000); await quiet(); + }); + + it("keeps sequential tenant writes from scanning another company's proposals", async () => { + const other = await seedRun("OTHER-1"); + const foreign = await proposals.createSecret({ ...other, registerForRedaction: () => {} }, { + name: "dev/other/secret", value: "other-expiry-only", justification: "Another company's deadline", + }); + await start(); await settle(1); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query, params) => { + if (/^select/i.test(query) && query.includes('"company_secret_proposals"') && + (query.includes('min(') || query.includes('"expires_at" <='))) reads.push({ query, params: [...params] }); + }; + try { + for (let i = 0; i < 3; i++) { await create(); await settle(1000); } + } finally { db.$client.options.debug = debug; } + expect(reads.length).toBeGreaterThan(0); + for (const read of reads) { + expect(read.query).toContain('"company_secret_proposals"."company_id" ='); + expect(read.params).toContain(fixture.companyId); + expect(read.params).not.toContain(other.companyId); + } + expect(await proposals.getById(other.companyId, foreign.id)).toMatchObject({ status: "pending", valueCiphertext: foreign.valueCiphertext }); + }); + + it("expires only the company whose saved deadline is due", async () => { + const other = await seedRun("OTHER-1"); + const foreign = await proposals.createSecret({ ...other, registerForRedaction: () => {} }, { + name: "dev/other/secret", value: "other-expiry-only", justification: "Another company's deadline", + }); + const due = await createWithExpiry(new Date(Date.now() + 5000)); + await start(); await settle(1); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query, params) => { + if (/^select/i.test(query) && query.includes('"company_secret_proposals"') && + (query.includes('min(') || query.includes('"expires_at" <='))) reads.push({ query, params: [...params] }); + }; + try { await settle(6000); } finally { db.$client.options.debug = debug; } + expect(reads.some(read => read.query.includes('"expires_at" <='))).toBe(true); + for (const read of reads) { + expect(read.query).toContain('"company_secret_proposals"."company_id" ='); + expect(read.params).toContain(fixture.companyId); + expect(read.params).not.toContain(other.companyId); + } + expect(await proposals.getById(fixture.companyId, due.id)).toMatchObject({ status: "expired", valueCiphertext: null }); + expect(await proposals.getById(other.companyId, foreign.id)).toMatchObject({ status: "pending", valueCiphertext: foreign.valueCiphertext }); + expect(coordinator!.nextWakeAt()).toBe(foreign.expiresAt.getTime()); + }); + + it.each([0, 3000])("keeps a failed company on backoff while another company expires on time (offset %s)", async offset => { + // Make the failing company first in the fair dispatch order. + const other = await seedRun("OTHER-1", "ffffffff-ffff-ffff-ffff-ffffffffffff"); + const failing = await createWithExpiry(new Date(Date.now() + 5000)); + const healthy = await db.transaction(async tx => { + const proposal = await createSecretProposalsService(tx as unknown as typeof db).createSecret( + { ...other, registerForRedaction: () => {} }, + { name: "dev/other/secret", value: "other-expiry-only", justification: "Independent expiry" }); + const [saved] = await tx.update(companySecretProposals).set({ expiresAt: new Date(failing.expiresAt.getTime() + offset) }) + .where(eq(companySecretProposals.id, proposal.id)).returning(); + return saved; + }); + const sweep = proposals.sweepExpiredForCompany; + let unavailable = true; + const attempts = vi.spyOn(proposals, "sweepExpiredForCompany").mockImplementation(async companyId => { + if (unavailable && companyId === fixture.companyId) throw new Error("company expiry unavailable"); + return sweep(companyId); + }); + await start(); await settle(10_000); + expect(errors).toHaveLength(1); errors.length = 0; + expect(await proposals.getById(other.companyId, healthy.id)).toMatchObject({ status: "expired", valueCiphertext: null }); + expect(await proposals.getById(fixture.companyId, failing.id)).toMatchObject({ status: "pending" }); + // A new write must refresh this company without erasing its failed sweep's backoff. + await create(); await settle(1000); + expect(attempts.mock.calls.filter(([id]) => id === fixture.companyId)).toHaveLength(1); + unavailable = false; await settle(30_000); + expect(await proposals.getById(fixture.companyId, failing.id)).toMatchObject({ status: "expired", valueCiphertext: null }); + await settle(expiryMs); await quiet(); + }); + + it("removes the cached expiry when the company deletion owner commits", async () => { + await create(); await start(); await settle(1); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.heartbeatRunId)); + await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, fixture.issueId)); + // The deletion fixture has no live resources; its agent has already + // finished termination before entering the company deletion transaction. + await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }) + .where(eq(agents.id, fixture.agentId)); + expect(await deleteCompany(db, fixture.companyId)).toMatchObject({ id: fixture.companyId }); + await quiet(); + expect(await proposals.expiryDeadlines()).toEqual([]); + }); + + it("stays quiet until the real proposal deadline, scrubs ciphertext, and leaves terminal history quiet", async () => { + await start(); await quiet(); + const secret = await create(); + const binding = await bind(secret.id); + expect(secret.valueCiphertext).not.toBeNull(); + await settle(); + expect(coordinator!.nextWakeAt()).toBe(secret.expiresAt.getTime()); + const reads = vi.spyOn(db, "select"); + await settle(24 * 60 * 60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await settle(expiryMs); + expect(await proposals.getById(fixture.companyId, secret.id)).toMatchObject({ + status: "expired", valueCiphertext: null, ciphertextScrubbedAt: expect.any(Date), + }); + expect(await proposals.getById(fixture.companyId, binding.id)).toMatchObject({ status: "rejected" }); + expect(await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, binding.interactionId!))) + .toEqual([expect.objectContaining({ status: "rejected" })]); + await quiet(); + }); + + it.each([false, true])("only observes the producer after outer settlement (rollback %s)", async rollback => { + await start(); await quiet(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const staged = new Promise(resolve => { written = resolve; }); + const rollbackError = new Error("rollback fixture"); + const writing = db.transaction(async tx => { + await create(createSecretProposalsService(tx as unknown as typeof db)); + written(); await gate; + if (rollback) throw rollbackError; + }).catch(error => { if (error !== rollbackError) throw error; }); + await staged; + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(await proposals.nextExpiryAt()).toBeNull(); + } finally { release(); } + await writing; await settle(); + expect(await proposals.nextExpiryAt()).toEqual(rollback ? null : expect.any(Number)); + if (rollback) await quiet(); + else expect(coordinator!.nextWakeAt()).toBe(await proposals.nextExpiryAt()); + }); + + it("retains a proposal committed while an empty deadline read is finishing", async () => { + let release!: () => void, reading!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const entered = new Promise(resolve => { reading = resolve; }); + const readDeadline = proposals.expiryDeadlines; + vi.spyOn(proposals, "expiryDeadlines").mockImplementationOnce(async () => { + const stale = await readDeadline(); + reading(); await gate; + return stale; + }); + const starting = start(); + await entered; + let secret!: Awaited>; + try { secret = await create(); } finally { release(); } + await starting; await settle(); + expect(coordinator!.nextWakeAt()).toBe(secret.expiresAt.getTime()); + await proposals.transition(fixture.companyId, secret.id, "withdrawn", { proposerAgentId: fixture.agentId }); + await quiet(); + }); + + it("reconstructs a deadline after restart and disarms when the proposer withdraws", async () => { + const secret = await create(); + await start(); await settle(); + expect(coordinator!.nextWakeAt()).toBe(secret.expiresAt.getTime()); + await proposals.transition(fixture.companyId, secret.id, "withdrawn", { proposerAgentId: fixture.agentId }); + await quiet(); + expect(await proposals.getById(fixture.companyId, secret.id)).toMatchObject({ valueCiphertext: null }); + }); + + it.each(["startup", "first_write", "failed_pass"])("retains %s work through drain and still rejects expired approval on use", async mode => { + if (mode === "startup") draining = true; + await start(); + if (mode === "first_write") { await quiet(); draining = true; } + const secret = await create(); + if (mode === "failed_pass") { + vi.spyOn(proposals, "expiryDeadlines").mockRejectedValueOnce(new Error("database unavailable")); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(errors).toHaveLength(1)); errors.length = 0; + draining = true; + } + await settle(); + vi.setSystemTime(Date.now() + expiryMs); + const reads = vi.spyOn(db, "select"); + await settle(2 * 60 * 60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(await proposals.getById(fixture.companyId, secret.id)).toMatchObject({ status: "pending" }); + await expect(proposals.approve(fixture.companyId, secret.id, { resolvedByUserId: "board-user" })) + .rejects.toThrow("Expired proposals cannot be approved"); + draining = false; await settle(); + expect(await proposals.getById(fixture.companyId, secret.id)).toMatchObject({ status: "expired", valueCiphertext: null }); + await quiet(); + }); + + it.each(["rejected", "execution_failed"])("reschedules when the interaction owner resolves a binding: %s", async mode => { + const secret = await create(); + await start(); + const approved = await proposals.approve(fixture.companyId, secret.id, { resolvedByUserId: "board-user" }); + await quiet(); + const remaining = await proposals.createBinding(fixture, { secretId: approved.createdSecretId, + configPath: "access.SECOND", justification: "Independent binding", bindingTargetPolicy: "self_and_reports" }); + await settle(); + expect(coordinator!.nextWakeAt()).toBe(remaining.expiresAt.getTime()); + const interactions = issueThreadInteractionService(db); + if (mode === "rejected") { + await interactions.rejectInteraction({ id: fixture.issueId, companyId: fixture.companyId, status: "in_progress" }, + remaining.interactionId!, { reason: "No longer required" }, { userId: "board-user" }); + } else { + await db.update(issueThreadInteractions).set({ status: "accepted", result: { version: 1, outcome: "accepted" }, + resolvedByUserId: "board-user", resolvedAt: new Date() }).where(eq(issueThreadInteractions.id, remaining.interactionId!)); + await interactions.recordSecretProposalExecutionResult({ id: fixture.issueId, companyId: fixture.companyId }, + remaining.interactionId!, remaining.id, { status: "failed", errorCode: "permission_denied" }); + } + expect(await proposals.getById(fixture.companyId, remaining.id)).toMatchObject({ status: "rejected" }); + await quiet(); + }); + }); + it("requires company admin access to reject secret proposals", async () => { const fixture = await seedRun(); const proposed = await request(createAgentApp(fixture)) diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 81c689366c3..f3d7171c03a 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,4 @@ +import { createSecretProposalsService } from "../services/secret-proposals.js"; import { logger } from "../middleware/logger.js"; import { getStartupRecoveryState } from "../startup-recovery-state.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -29,6 +30,7 @@ const { detectPortMock, deriveAuthTrustedOriginsMock, deliveryWorkRegisterMock, + deliveryWorkErrors, environmentCustomImagesServiceMock, environmentCustomImagesServiceFactoryMock, executionWorkspaceServiceFactoryMock, @@ -53,11 +55,12 @@ const { }; const caseHandoffDeliveryServiceFactoryMock = vi.fn(() => caseHandoffDeliveryServiceMock); const completionSweepMock = vi.fn(async () => undefined); - const deliveryWorkRegisterMock = vi.fn((_queue: string, task: { + const deliveryWorkErrors: Array<{ queue: string; error: unknown }> = []; + const deliveryWorkRegisterMock = vi.fn((queue: string, task: { retryMs?: number; run: (signal?: AbortSignal) => Promise; hasPending?: () => Promise; - }) => ({ ready: task.run(new AbortController().signal).catch(() => undefined), wake: vi.fn() })); + }) => ({ ready: task.run(new AbortController().signal).catch(error => { deliveryWorkErrors.push({ queue, error }); }), wake: vi.fn() })); const createAppMock = vi.fn(async () => Object.assign((_: unknown, __: unknown) => {}, { locals: { fastResponses: { hasPending: vi.fn(async () => false), sweepPending: vi.fn(async () => undefined) }, @@ -181,6 +184,7 @@ const { detectPortMock, deriveAuthTrustedOriginsMock, deliveryWorkRegisterMock, + deliveryWorkErrors, environmentCustomImagesServiceMock, environmentCustomImagesServiceFactoryMock, executionWorkspaceServiceFactoryMock, @@ -442,6 +446,9 @@ vi.mock("../services/native-runtime/native-question-bridge.js", () => ({ vi.mock("../services/secret-proposals.js", () => ({ createSecretProposalsService: vi.fn(() => ({ + expiryDeadlines: vi.fn(async () => []), + sweepExpiredForCompany: vi.fn(async () => 0), + nextExpiryAt: vi.fn(async () => null), sweepExpired: vi.fn(async () => 0), })), })); @@ -490,6 +497,7 @@ async function settleStartupHousekeeping() { describe("startServer feedback export wiring", () => { beforeEach(() => { vi.clearAllMocks(); + deliveryWorkErrors.length = 0; process.env.PAPERCLIP_DECISION_SIGNING_SECRET = "fedcba9876543210fedcba9876543210"; process.env.PAPERCLIP_AGENT_JWT_SECRET = "0123456789abcdef0123456789abcdef"; loadConfigMock.mockReturnValue(buildTestConfig()); @@ -791,6 +799,19 @@ describe("startServer feedback export wiring", () => { } }); + it("loads secret expiry deadlines successfully during enabled startup", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); + await startServer(); + await settleStartupHousekeeping(); + const index = deliveryWorkRegisterMock.mock.calls.findIndex(([queue]) => queue === DELIVERY_QUEUES.secretProposalExpiry); + expect(index).toBeGreaterThanOrEqual(0); + await deliveryWorkRegisterMock.mock.results[index].value.ready; + expect(deliveryWorkErrors.filter(({ queue }) => queue === DELIVERY_QUEUES.secretProposalExpiry)).toEqual([]); + const service = vi.mocked(createSecretProposalsService).mock.results[0].value; + expect(service.expiryDeadlines).toHaveBeenCalledExactlyOnceWith(undefined); + expect(service.sweepExpiredForCompany).not.toHaveBeenCalled(); + }); + it("registers native replacement recovery without an execution-control interval", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: true })); const setIntervalSpy = vi.spyOn(globalThis, "setInterval"); diff --git a/server/src/index.ts b/server/src/index.ts index 411dda3f87f..ca51efe4345 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerSecretProposalExpiryWork } from "./services/secret-proposal-expiry-work.js"; import { registerCustomImageCleanupWork } from "./services/custom-image-cleanup-work.js"; import { registerWorkspaceRecoveryWork } from "./services/workspace-recovery-work.js"; import { registerEnvironmentCleanupWork } from "./services/environment-cleanup-work.js"; @@ -1791,6 +1792,8 @@ async function startServerWithDatabaseTeardown( work().catch((err: unknown) => { logger.error({ err, startupStep: name }, "startup housekeeping step failed"); }); + await step("secret_proposal_expiry", () => registerSecretProposalExpiryWork(deliveryWork, secretProposals, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready); await step("custom_image_setup_cleanup", () => registerCustomImageCleanupWork(deliveryWork, environmentCustomImages, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, config.heartbeatSchedulerIntervalMs).ready); @@ -1887,14 +1890,6 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "periodic tool connection health sweep failed"); })); - trackHeartbeatSchedulerWork(secretProposals.sweepExpired() - .then((expired) => { - if (expired > 0) logger.warn({ expired }, "periodic secret proposal expiry scrubbed proposals"); - }) - .catch((err) => { - logger.error({ err }, "periodic secret proposal expiry sweep failed"); - })); - trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); trackHeartbeatSchedulerWork(app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "gateway token cleanup failed"))); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 432cad76e7a..7e4a41a95d7 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -61,6 +61,7 @@ export async function deleteCompany(db: Db, id: string) { .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id); await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id); // Finance can reference costs and both can reference runs. Incidents // reference policies and approvals; delete these dependents first. diff --git a/server/src/services/company-work-batch.ts b/server/src/services/company-work-batch.ts new file mode 100644 index 00000000000..f83d2dcd5a5 --- /dev/null +++ b/server/src/services/company-work-batch.ts @@ -0,0 +1,28 @@ +export function takeCompanyWorkBatch( + ordered: readonly T[], + limit: number, + afterCompany: string | null = null, +): { entries: T[]; lastCompany: string | null } { + const grouped = new Map(); + for (const entry of ordered) { + const group = grouped.get(entry.companyId); + if (group) group.push(entry); else grouped.set(entry.companyId, [entry]); + } + const companies = [...grouped.keys()].sort(); + const next = afterCompany === null ? 0 : companies.findIndex(id => id > afterCompany); + const start = next < 0 ? 0 : next; + const rotation = [...companies.slice(start), ...companies.slice(0, start)]; + const entries: T[] = []; + let lastCompany = afterCompany; + for (let round = 0; entries.length < limit; round++) { + let added = false; + for (const companyId of rotation) { + const entry = grouped.get(companyId)![round]; + if (!entry) continue; + entries.push(entry); lastCompany = companyId; added = true; + if (entries.length === limit) break; + } + if (!added) break; + } + return { entries, lastCompany }; +} diff --git a/server/src/services/custom-image-cleanup-work.ts b/server/src/services/custom-image-cleanup-work.ts index 9ecd4e73612..feee4ebefa7 100644 --- a/server/src/services/custom-image-cleanup-work.ts +++ b/server/src/services/custom-image-cleanup-work.ts @@ -1,5 +1,6 @@ import { logger } from "../middleware/logger.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerDeadlineWork } from "./deadline-work.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import type { environmentCustomImageService } from "./environment-custom-images.js"; @@ -9,27 +10,13 @@ export function registerCustomImageCleanupWork( canRun: () => boolean, retryMs = 30_000, ) { - let requested = 1, reconciled = 0; - let next: number | null = null; - let earliest = 0; - return coordinator.register(DELIVERY_QUEUES.customImageExpiry, { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.customImageExpiry, { retryMs, - pausePassiveMaintenance: true, - onSettled: () => { requested++; }, + canRun, run: async () => { - if (!canRun()) { requested++; return; } - if (Date.now() < earliest) { next = earliest; return; } - const generation = ++requested; - earliest = Date.now() + retryMs; const result = await service.cleanupExpiredSetupSessions(); if (result.timedOut > 0 || result.failed > 0) logger.warn(result, "environment customImage setup cleanup changed sessions"); - const deadline = await service.nextSetupCleanupAt(); - // Back off failed or still-overdue work, not a successfully read future deadline. - if (deadline === null || deadline > Date.now()) earliest = 0; - next = deadline === null ? null : Math.max(earliest, deadline); - reconciled = generation; }, - hasPending: async () => requested !== reconciled, - nextRunAt: () => next, + next: () => service.nextSetupCleanupAt(), }); } diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts new file mode 100644 index 00000000000..9f1de1b2a24 --- /dev/null +++ b/server/src/services/deadline-work.ts @@ -0,0 +1,36 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import type { DeliveryQueue } from "./delivery-work-notifications.js"; + +export function registerDeadlineWork( + coordinator: ReturnType, + queue: DeliveryQueue, + task: { + run: () => Promise; + next: () => Promise; + canRun: () => boolean; + retryMs: number; + }, +) { + let requested = 1, reconciled = 0; + let next: number | null = null; + let earliest = 0; + return coordinator.register(queue, { + retryMs: task.retryMs, + pausePassiveMaintenance: true, + onSettled: () => { requested++; }, + run: async () => { + if (!task.canRun()) { requested++; return; } + if (Date.now() < earliest) { next = earliest; return; } + const generation = ++requested; + earliest = Date.now() + task.retryMs; + await task.run(); + const deadline = await task.next(); + // Back off failed or still-overdue work, not a successfully read future deadline. + if (deadline === null || deadline > Date.now()) earliest = 0; + next = deadline === null ? null : Math.max(earliest, deadline); + reconciled = generation; + }, + hasPending: async () => requested !== reconciled, + nextRunAt: () => next, + }); +} diff --git a/server/src/services/delivery-work-coordinator.ts b/server/src/services/delivery-work-coordinator.ts index 0284e06aba2..0d1a88c9f59 100644 --- a/server/src/services/delivery-work-coordinator.ts +++ b/server/src/services/delivery-work-coordinator.ts @@ -26,6 +26,9 @@ export function createDeliveryWorkCoordinator(input: { onSettled?: (hints?: ReadonlySet) => void; // Optional maintenance/retry deadline restored from durable state. nextRunAt?: () => number | null; + // Opt in only when run has recorded per-item failure backoffs. Other + // eligible items may already be due, unlike an unchanged failed deadline. + nextRunAtOnError?: () => number | null; // An empty queue's maintenance may wait until dispatch admission reopens. // Notifications and unresolved producer writes still reconcile in drain. pausePassiveMaintenance?: boolean; @@ -65,9 +68,10 @@ export function createDeliveryWorkCoordinator(input: { dirty = false; const finish = beginIdleTrackedWork(); const attempt = new AbortController(); + let runFailed = false; controller = attempt; running = Promise.resolve().then(() => reconcileDatabaseWork(input.owner, queue)) - .then(() => workerStopped || !(input.canRun() || (task.runDuringDrain && input.canReconcile?.())) ? undefined : task.run(attempt.signal)).then(() => workerStopped ? false : task.hasPending()).then(pending => { + .then(() => workerStopped || !(input.canRun() || (task.runDuringDrain && input.canReconcile?.())) ? undefined : task.run(attempt.signal).catch(error => { runFailed = true; throw error; })).then(() => workerStopped ? false : task.hasPending()).then(pending => { pendingWork = pending || databaseWorkPending(input.owner, queue); if (pendingWork) schedule(task.retryMs); else { finishIntent?.(); finishIntent = undefined; } @@ -78,9 +82,17 @@ export function createDeliveryWorkCoordinator(input: { } }).catch(error => { schedule(task.retryMs); + let failure = error; + try { + const scopedRetry = runFailed && task.nextRunAtOnError; + const next = input.canRun() ? (scopedRetry ? scopedRetry() : task.nextRunAt?.()) : null; + if (next != null && (scopedRetry || next > Date.now())) schedule(0, Math.max(Date.now(), next)); + } catch (deadlineError) { + failure = new AggregateError([error, deadlineError], "Delivery failure deadline could not be read"); + } // Logging must never turn a recoverable sweep failure into an // unhandled rejection or prevent other queues from running. - try { if (!workerStopped) input.onError(error, queue); } catch { /* Recovery remains scheduled. */ } + try { if (!workerStopped) input.onError(failure, queue); } catch { /* Recovery remains scheduled. */ } }).finally(() => { running = null; controller = null; diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index f6d50c71930..b6bed1d0f25 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + secretProposalExpiry: "secret-proposal-expiry", customImageExpiry: "custom-image-expiry", historicalCleanup: "historical-cleanup", adapterLoginCleanup: "adapter-login-cleanup", diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index bcfbe2882a9..74e8d8216db 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1159,6 +1159,7 @@ async function resolveLinkedSecretProposal( ) { const proposalId = linkedSecretProposalId(interaction); if (!proposalId) return; + await notifyDeliveryWork(db, DELIVERY_QUEUES.secretProposalExpiry, interaction.companyId); const [proposal] = await db .update(companySecretProposals) .set({ @@ -3181,6 +3182,7 @@ export function issueThreadInteractionService( : execution.status; if (executionStatus === "failed" && proposal.status === "pending") { const resolutionReason = `Interaction acceptance failed: ${execution.errorCode ?? "secret_proposal_execution_failed"}`; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, proposal.companyId); await tx .update(companySecretProposals) .set({ diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts new file mode 100644 index 00000000000..6de9d3895da --- /dev/null +++ b/server/src/services/scoped-deadline-work.ts @@ -0,0 +1,99 @@ +import { takeCompanyWorkBatch } from "./company-work-batch.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import type { DeliveryQueue } from "./delivery-work-notifications.js"; + +type ScheduledEntry = { id: string; companyId: string; dueAt: number }; + +/** Coalesced scope reconstruction and fair deadline dispatch. Domain owners + * supply persistence, eligibility, and execution; this helper owns no SQL. */ +export function registerScopedDeadlineWork( + coordinator: ReturnType, + options: { + queue: DeliveryQueue; + batchSize?: number; + minimumPassIntervalMs?: number; + decodeScope(hint: string): Scope | undefined; + matches(entry: Entry, scope?: Scope): boolean; + load(scope?: Scope, signal?: AbortSignal): Promise; + check(entry: Entry, signal?: AbortSignal): Promise; + }, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const entries = new Map(); + const failures = new Map(); + const scopeFailures = new Map(); + let earliest = 0; + let lastCompany: string | null = null; + function failed(map: typeof failures, id: string) { + const count = Math.min((map.get(id)?.count ?? 0) + 1, 5); + map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }); + } + function dirtyScopes() { + return [...dirty].map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } }); + } + function next() { + if (!canRun()) return dirty.size > 0 || entries.size > 0 ? Date.now() + 30_000 : null; + let nextAt: number | null = null; + const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); }; + for (const hint of dirty.has("*") ? ["*"] : dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0)); + const unavailable = dirtyScopes(); + for (const entry of entries.values()) if (!unavailable.some(scope => options.matches(entry, scope))) + take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0)); + return nextAt; + } + const worker = coordinator.register(options.queue, { + retryMs: 30_000, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + (options.minimumPassIntervalMs ?? 1000); + const errors: unknown[] = []; + const requested = (dirty.has("*") ? ["*"] : [...dirty]).filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now()); + const covered = requested.includes("*") ? [...dirty] : requested; + for (const hint of covered) dirty.delete(hint); + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { for (const hint of requested.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; } + const hint = requested[index]; + let scope: Scope | undefined; + let decoded = false; + try { + scope = options.decodeScope(hint); decoded = true; + const rows = await options.load(scope, signal), retained = new Set(rows.map(row => row.id)); + for (const entry of entries.values()) if (options.matches(entry, scope) && !retained.has(entry.id)) { + entries.delete(entry.id); failures.delete(entry.id); + } + for (const entry of rows) entries.set(entry.id, entry); + if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint); + } catch (error) { + const retryHint = decoded ? hint : "*"; + dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error); + } + } + const unavailable = dirtyScopes(); + const due = [...entries.values()].filter(entry => !unavailable.some(scope => options.matches(entry, scope)) + && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)); + const batch = takeCompanyWorkBatch(due, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany; + for (const entry of batch.entries) { + if (signal.aborted || !canRun()) break; + try { + const current = await options.check(entry, signal); + if (current) entries.set(entry.id, current); else entries.delete(entry.id); + failures.delete(entry.id); + } catch (error) { failed(failures, entry.id); errors.push(error); } + } + if (errors.length) throw new AggregateError(errors, `${options.queue} could not settle`); + }, + // Retain scopes/deadlines for resume without retaining a settled write's + // idle hold during drain. The coordinator still fences unresolved XIDs. + hasPending: async () => canRun() && dirty.size > 0, + nextRunAt: next, + nextRunAtOnError: next, + }); + return { ...worker, invalidate(hint = "*") { + dirty.add(hint); + worker.wake(); + } }; +} diff --git a/server/src/services/secret-proposal-expiry-work.ts b/server/src/services/secret-proposal-expiry-work.ts new file mode 100644 index 00000000000..1a019047c92 --- /dev/null +++ b/server/src/services/secret-proposal-expiry-work.ts @@ -0,0 +1,29 @@ +import { logger } from "../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; +import type { createSecretProposalsService } from "./secret-proposals.js"; + +export function registerSecretProposalExpiryWork( + coordinator: ReturnType, + service: Pick, "sweepExpiredForCompany" | "expiryDeadlines">, + canRun: () => boolean, +) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.secretProposalExpiry, + batchSize: 1, + // Saved expiry deadlines are authoritative, including deadlines less than + // a second away. Only actual failures require a recovery backoff. + minimumPassIntervalMs: 0, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: (entry, companyId?: string) => companyId === undefined || entry.companyId === companyId, + load: async (companyId?: string) => (await service.expiryDeadlines(companyId)) + .map(row => ({ ...row, id: row.companyId })), + check: async entry => { + const expired = await service.sweepExpiredForCompany(entry.companyId); + if (expired > 0) logger.warn({ companyId: entry.companyId, expired }, "secret proposal expiry scrubbed proposals"); + const [next] = await service.expiryDeadlines(entry.companyId); + return next ? { ...next, id: next.companyId } : null; + }, + }, canRun); +} diff --git a/server/src/services/secret-proposals.ts b/server/src/services/secret-proposals.ts index fbefd53d5fd..05f5af71d91 100644 --- a/server/src/services/secret-proposals.ts +++ b/server/src/services/secret-proposals.ts @@ -1,3 +1,4 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { updateAgentConfigurationInTransaction } from "./agent-configuration-transaction.js"; import { withAgentAppearance } from "@paperclipai/shared"; import { and, count, desc, eq, gte, inArray, lte, or, sql } from "drizzle-orm"; @@ -144,6 +145,7 @@ export function createSecretProposalsService(db: Db) { }); return { denial, value: null }; } + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.secretProposalExpiry, input.companyId); return { denial: null, value: await create(txDb) }; }); if (result.denial) throw unprocessable(result.denial.message); @@ -619,6 +621,7 @@ export function createSecretProposalsService(db: Db) { appliedBindingConfigPath?: string | null; }) { const now = new Date(); + await notifyDeliveryWork(txDb, DELIVERY_QUEUES.secretProposalExpiry, proposal.companyId); const updated = await txDb.update(companySecretProposals).set({ status: "approved", resolvedByUserId: input.resolvedByUserId, @@ -794,6 +797,7 @@ export function createSecretProposalsService(db: Db) { const now = new Date(); return db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, companyId); const updated = await tx.update(companySecretProposals).set({ status, resolvedByUserId: input.resolvedByUserId ?? null, @@ -868,10 +872,12 @@ export function createSecretProposalsService(db: Db) { limit = DEFAULT_EXPIRY_SWEEP_LIMIT, expireProposal: (companyId: string, proposalId: string) => Promise = (companyId, proposalId) => transition(companyId, proposalId, "expired", { reason: "Pending proposal expired" }), + companyId?: string, ) { const expired = await db.select({ id: companySecretProposals.id, companyId: companySecretProposals.companyId }) .from(companySecretProposals) - .where(and(eq(companySecretProposals.status, "pending"), lte(companySecretProposals.expiresAt, now))) + .where(and(eq(companySecretProposals.status, "pending"), lte(companySecretProposals.expiresAt, now), + companyId === undefined ? undefined : eq(companySecretProposals.companyId, companyId))) .orderBy(companySecretProposals.expiresAt) .limit(limit); let expiredCount = 0; @@ -887,5 +893,25 @@ export function createSecretProposalsService(db: Db) { return expiredCount; } - return { getById, view: enrich, createSecret, createBinding, listForAgent, listForBoard, assertBindingSnapshotCurrent, approve, transition, sweepExpired }; + async function sweepExpiredForCompany(companyId: string) { + return sweepExpired(new Date(), DEFAULT_EXPIRY_SWEEP_LIMIT, undefined, companyId); + } + + async function expiryDeadlines(companyId?: string) { + const rows = await db.select({ companyId: companySecretProposals.companyId, + at: sql`min(${companySecretProposals.expiresAt})` }) + .from(companySecretProposals) + .where(and(eq(companySecretProposals.status, "pending"), + companyId === undefined ? undefined : eq(companySecretProposals.companyId, companyId))) + .groupBy(companySecretProposals.companyId); + return rows.map(row => ({ companyId: row.companyId, dueAt: new Date(row.at).getTime() })); + } + + async function nextExpiryAt(): Promise { + const [row] = await db.select({ at: sql`min(${companySecretProposals.expiresAt})` }) + .from(companySecretProposals).where(eq(companySecretProposals.status, "pending")); + return row?.at == null ? null : new Date(row.at).getTime(); + } + + return { getById, view: enrich, createSecret, createBinding, listForAgent, listForBoard, assertBindingSnapshotCurrent, approve, transition, sweepExpired, sweepExpiredForCompany, expiryDeadlines, nextExpiryAt }; }