diff --git a/doc/idle-secret-proposal-expiry.md b/doc/idle-secret-proposal-expiry.md new file mode 100644 index 00000000000..a9ca93c11bb --- /dev/null +++ b/doc/idle-secret-proposal-expiry.md @@ -0,0 +1,73 @@ +# 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. `nextExpiryAt()` returns the earliest pending proposal expiry. +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. + +The remaining proposal write only attaches an interaction ID and does not change +expiry eligibility. Company deletion removes obligations. 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 reconciles committed changes and registers the earliest saved expiry. +A failed pass or an overdue batch retries after 30 seconds. Each pass retains the +existing bounded 100-row sweep. Completed batches publish their changes; further +overdue rows retain a deadline until drained. Own writes are coalesced to avoid +an immediate self-triggering scan loop. No pending proposal, unresolved write, or +failed reconciliation means no recurring query. + +`deadline-work.ts` shares only generation tracking, bounded re-entry, and scheduler +registration with custom-image expiry. Each service supplies its own operation and +next deadline. Neither shared code nor startup wiring interprets secret proposals +or provider protocols. Custom-image expiry retains its existing behavior. + +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 +custom-image lifecycle suite checks the other consumer of the shared helper. + +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..a7126d42071 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), CI/review running | | 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__/secret-proposals-routes.test.ts b/server/src/__tests__/secret-proposals-routes.test.ts index 3354345e4f4..47cb66c6a29 100644 --- a/server/src/__tests__/secret-proposals-routes.test.ts +++ b/server/src/__tests__/secret-proposals-routes.test.ts @@ -1,3 +1,6 @@ +import { registerSecretProposalExpiryWork } from "../services/secret-proposal-expiry-work.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 +8,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 +41,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabase = process.env.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 +54,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,6 +79,7 @@ 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; @@ -221,6 +227,151 @@ 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", + }); + + 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 settle(); + expect(await proposals.nextExpiryAt()).toBeNull(); + } finally { release(); } + await writing; await settle(); + expect(await proposals.nextExpiryAt()).toEqual(rollback ? null : expect.any(Number)); + expect(coordinator!.nextWakeAt()).toBe(await proposals.nextExpiryAt()); + }); + + it("reconstructs a deadline after restart and disarms when the proposer withdraws", async () => { + const secret = await create(); + await start(); + 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, "sweepExpired").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/index.ts b/server/src/index.ts index 380ec1b6411..2666d05c814 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 { registerEnvironmentCleanupWork } from "./services/environment-cleanup-work.js"; import { loginCleanupSchedule } from "./services/login-cleanup-state.js"; @@ -1757,6 +1758,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); @@ -1853,14 +1856,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/custom-image-cleanup-work.ts b/server/src/services/custom-image-cleanup-work.ts index e409bc45d42..feee4ebefa7 100644 --- a/server/src/services/custom-image-cleanup-work.ts +++ b/server/src/services/custom-image-cleanup-work.ts @@ -1,34 +1,22 @@ 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"; -/** Existing setup rows own expiry; provider operations remain in their service. */ export function registerCustomImageCleanupWork( coordinator: ReturnType, service: Pick, "cleanupExpiredSetupSessions" | "nextSetupCleanupAt">, 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(); - 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..2e39dc4c4c3 --- /dev/null +++ b/server/src/services/deadline-work.ts @@ -0,0 +1,34 @@ +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(); + next = deadline === null ? null : Math.max(earliest, deadline); + reconciled = generation; + }, + hasPending: async () => requested !== reconciled, + nextRunAt: () => next, + }); +} diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index b1d47f72982..98ab73c8de2 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -8,6 +8,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 4e25c890615..45041525c07 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); const [proposal] = await db .update(companySecretProposals) .set({ @@ -3154,6 +3155,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); await tx .update(companySecretProposals) .set({ 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..f4bf8d0f073 --- /dev/null +++ b/server/src/services/secret-proposal-expiry-work.ts @@ -0,0 +1,21 @@ +import { logger } from "../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { createSecretProposalsService } from "./secret-proposals.js"; + +export function registerSecretProposalExpiryWork( + coordinator: ReturnType, + service: Pick, "sweepExpired" | "nextExpiryAt">, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.secretProposalExpiry, { + retryMs: 30_000, + canRun, + run: async () => { + const expired = await service.sweepExpired(); + if (expired > 0) logger.warn({ expired }, "secret proposal expiry scrubbed proposals"); + }, + next: () => service.nextExpiryAt(), + }); +} diff --git a/server/src/services/secret-proposals.ts b/server/src/services/secret-proposals.ts index fbefd53d5fd..e5efbbf7124 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); 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); 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); const updated = await tx.update(companySecretProposals).set({ status, resolvedByUserId: input.resolvedByUserId ?? null, @@ -887,5 +891,11 @@ export function createSecretProposalsService(db: Db) { return expiredCount; } - return { getById, view: enrich, createSecret, createBinding, listForAgent, listForBoard, assertBindingSnapshotCurrent, approve, transition, sweepExpired }; + 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, nextExpiryAt }; }