mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
fix(server): schedule secret-proposal expiry from committed changes (#17446)
## 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
5982246771
commit
623f1dc289
19 files changed
+792
-41
No files matched your search
@@ -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]);
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator> | 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();
|
||||
});
|
||||
@@ -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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let proposals: ReturnType<typeof createSecretProposalsService>;
|
||||
let fixture: Awaited<ReturnType<typeof seedRun>>;
|
||||
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<void>(resolve => { release = resolve; });
|
||||
const staged = new Promise<void>(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<void>(resolve => { release = resolve; });
|
||||
const entered = new Promise<void>(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<ReturnType<typeof create>>;
|
||||
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))
|
||||
|
||||
@@ -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<unknown>;
|
||||
hasPending?: () => Promise<boolean>;
|
||||
}) => ({ 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");
|
||||
|
||||
+3
-8
@@ -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")));
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
export function takeCompanyWorkBatch<T extends { companyId: string }>(
|
||||
ordered: readonly T[],
|
||||
limit: number,
|
||||
afterCompany: string | null = null,
|
||||
): { entries: T[]; lastCompany: string | null } {
|
||||
const grouped = new Map<string, T[]>();
|
||||
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 };
|
||||
}
|
||||
@@ -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(),
|
||||
});
|
||||
}
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
queue: DeliveryQueue,
|
||||
task: {
|
||||
run: () => Promise<unknown>;
|
||||
next: () => Promise<number | null>;
|
||||
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,
|
||||
});
|
||||
}
|
||||
@@ -26,6 +26,9 @@ export function createDeliveryWorkCoordinator(input: {
|
||||
onSettled?: (hints?: ReadonlySet<string>) => 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;
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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<Entry extends ScheduledEntry, Scope>(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
options: {
|
||||
queue: DeliveryQueue;
|
||||
batchSize?: number;
|
||||
minimumPassIntervalMs?: number;
|
||||
decodeScope(hint: string): Scope | undefined;
|
||||
matches(entry: Entry, scope?: Scope): boolean;
|
||||
load(scope?: Scope, signal?: AbortSignal): Promise<Entry[]>;
|
||||
check(entry: Entry, signal?: AbortSignal): Promise<Entry | null>;
|
||||
},
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
const dirty = new Set<string>(["*"]);
|
||||
const entries = new Map<string, Entry>();
|
||||
const failures = new Map<string, { count: number; retryAt: number }>();
|
||||
const scopeFailures = new Map<string, { count: number; retryAt: number }>();
|
||||
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();
|
||||
} };
|
||||
}
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
service: Pick<ReturnType<typeof createSecretProposalsService>, "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);
|
||||
}
|
||||
@@ -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<unknown> =
|
||||
(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<string | Date>`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<number | null> {
|
||||
const [row] = await db.select({ at: sql<string | Date | null>`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 };
|
||||
}
|
||||
Reference in new issue
Block a user