mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-29 04:25:52 +01:00
fix(server): recover decision effects from saved claim deadlines (#17482)
## Thinking Path > - Paperclip manages agent work and the decisions that authorize it. > - A committed decision can leave effects unfinished after an interrupted execution. > - Recovery must preserve that debt without scanning settled history forever. > - Claims already store the state and time needed to schedule recovery. > - A failed claim must get another attempt even while newer claims arrive. > - Malformed claims need a durable failure outcome instead of an endless retry. > - This PR uses committed signals, saved deadlines, and bounded recovery rotations. ## Linked Issues or Issue Description **What happened?** Periodic decision maintenance queries for unfinished effects even when none remain. A cursor that only wraps after a short page can postpone an old failure indefinitely under continuous new work. Missing saved deciders or chosen options can also leave non-executable claims scanning forever. **Expected behavior** Open decisions and completed history produce no recurring effect-recovery SQL. Committed claims wake recovery, and restart reconstructs their deadlines. Genuine failures retain retries. Known malformed outcomes settle with an audit and preserve completed effects. Recovery must not invent authority, repeat committed effects, or lose continuation delivery. **Steps to reproduce** Create an old failing claim with a one-row recovery page, then keep adding newer due claims. Before the rotation fix, the old failure is never revisited. Separately, recover a claim with a missing decider or chosen option: the old implementation skips or throws on every attempt without settling it. Creating an ordinary open decision also caused an unnecessary effect scan. Stacked on #17479. Other decision expiry, continuation, target and retention duties have separate PRs in the stack. ## What Changed - Coalesce sequential committed notifications for up to one second without delaying already saved deadlines. This bounds instance-wide recovery reads and adds no idle polling. - Register effect recovery with the shared scheduler and transaction-aware notifications. Derive deadlines from the existing claim timestamp and recovery grace. - Remove effect recovery from recurring aggregate maintenance when the worker owns it. Preserve the explicit combined recovery entry point. - Page within a frozen final timestamp/ID and retain PostgreSQL precision at both boundaries. Failed claims and healthy later claims both make progress. - Load the current claim and derive recovery authority inside the shared execution flight. Preserve foreground joining, current authorization and per-effect locks. - Recheck known malformed claims under a row lock. Save a failed outcome, or partial when an executed effect receipt exists, with `effectRecoveryError` and an atomic `decision.recovery_failed` audit. Publish activity after commit and use normal continuation delivery. - Avoid effect-recovery notifications for open-decision creation. Claim and settlement writes retain their existing notification boundary. - Add generated migration `0390_salty_killraven.sql`, the concurrent, idempotent partial index for running-claim lookup. No queue table is added. - Cover quietness, transaction and lifecycle boundaries, retry fairness, malformed outcomes, and concurrent repair. Document the contract in `doc/idle-decision-effect-recovery.md`. ## Verification Current candidate normally integrates landed expiry `3421366903`, including expiry-only creation signals and all 21 independent master changes. The test revision is `dba8a79d1e1a5b97b8b5f6dac0f7c7837536531a`; the publication adds only this evidence documentation. - **143 current cases pass** across all 26 recovery, 54 decision, 20 expiry, 42 startup and one scoped-deadline cases. Server types/build pass with the rebuilt Runner vendor (1,524 inputs and 1,563 outputs). These results follow the landed dependency reconciliation; prior 69- and 141-case results remain historical evidence for their stated revisions. - Generated migration `0390_salty_killraven.sql` adds the existing running-claim partial index through the concurrent executor. All 388 parent journal entries are unchanged; five generated snapshots remain. Schema drift, numbering/safety and actual migration apply/replay pass. - Existing regressions exercise failed-claim rotation under sustained arrivals, malformed decider/option outcomes, atomic audit failure, row-lock repair races, current authority, foreground effect joining, coalesced tenant notifications, transaction commit/rollback, drain/resume, paging with PostgreSQL timestamp precision, and shutdown/restart. Completed effects are not repeated. - The first integrated run passed 140 cases and failed an overlap test that assumed the effect deadline was the scheduler's earliest deadline. Continuation coalescing now creates an earlier registered wake. The test services that earlier wake before asserting the original effect grace deadline; it still requires one shared effect flight, two lock attempts including the test gate, one committed comment and one continuation. No production timing was relaxed. - Earlier red/green fairness, malformed-claim, notification amplification and fixture failures remain recorded in [the owning audit](doc/idle-decision-effect-recovery.md). Extra upstream embedded-only tests attempted during parent integration could not initialize locally; CI must exercise that environment. - Previous candidate `f54a707d74` received both 5/5 reviews but its CI server shard 6 was interrupted (run 40597921712). The step was cancelled, with no failing assertion or suite summary, and the verify gate also failed. No retry or passing claim was made for that run. The necessary parent repair is included here; fresh CI and both reviews are required. Commands: server typecheck/build and Vitest for `decision-effect-recovery-work`, `decisions-service`, `decision-expiry-work`, `server-startup-feedback-export`, and `scoped-deadline-work`; DB migration checks, schema snapshot drift and real migration apply/replay. No full local repository-suite or deployed sleep/cost claim is made. Fresh current-head CI and Greptile/Ponytail are required before merge. ## Risks Missing deciders and missing chosen options now produce durable failed or partial outcomes instead of remaining running forever. Their audit and receipts remain available; no effect executes under invented authority. Only these known malformed states are settled this way. Database, audit, and other execution failures retain retries. Continuations retain their existing at-least-once delivery contract. The supported writer model remains process-local notifications plus restart reconstruction. Independent SQL writers and multiple replicas gain no new wake protocol. Cloud admission and durable external alarms are unchanged. These tests establish worker behavior, not deployed host suspension or fleet savings. ## 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
5787962454
commit
e9e84bb8a1
15 files changed
+1131
-40
No files matched your search
@@ -0,0 +1,500 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agents, applyPendingMigrations, authUsers, companies, companyMemberships, createDb, decisionEffectExecutions,
|
||||
decisions, decisionRetention, decisionTargetIssues, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { decisionService } from "../services/decisions.js";
|
||||
import { decisionEffectRecoveryService } from "../services/decision-effect-recovery.js";
|
||||
import { registerDecisionEffectRecoveryWork } from "../services/decision-effect-recovery-work.js";
|
||||
import { registerDecisionContinuationWork } from "../services/decision-continuation-work.js";
|
||||
import { writeDecisionChange } from "../services/decision-work-notifications.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { hashAttentionArchiveManifest } from "../services/decision-retention.js";
|
||||
import type { AttentionArchiveManifestEntry } from "@paperclipai/shared";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISION_EFFECTS_TEST_DATABASE_URL");
|
||||
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("decision effect recovery worker", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let companyId: string, agentId: string, originId: string, targetId: string, runId: string, userId: string;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let draining: boolean;
|
||||
let recovery: ReturnType<typeof decisionService>;
|
||||
const errors: unknown[] = [];
|
||||
const wake = vi.fn(async () => {});
|
||||
const service = () => decisionService(db, { wakeOriginAgent: wake });
|
||||
|
||||
beforeAll(async () => {
|
||||
if (externalUrl) await applyPendingMigrations(externalUrl);
|
||||
else temporary = await startEmbeddedPostgresTestDatabase("decision-effects-");
|
||||
db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
}, 30_000);
|
||||
beforeEach(async () => {
|
||||
vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", "0123456789abcdef0123456789abcdef");
|
||||
companyId = randomUUID(); agentId = randomUUID(); originId = randomUUID(); targetId = randomUUID(); runId = randomUUID(); userId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `R${companyId.slice(0, 6)}`, requireBoardApprovalForNewAgents: false });
|
||||
await db.insert(authUsers).values({ id: userId, name: "Decider", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Proposer", role: "engineer", status: "active", adapterType: "codex_local", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
|
||||
await db.insert(issues).values([
|
||||
{ id: originId, companyId, title: "Origin", status: "in_progress", priority: "medium", assigneeAgentId: agentId, responsibleUserId: userId },
|
||||
{ id: targetId, companyId, title: "Target", status: "todo", priority: "medium", responsibleUserId: userId },
|
||||
]);
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", responsibleUserId: userId, contextSnapshot: { issueId: originId } });
|
||||
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
|
||||
draining = false; errors.length = 0; wake.mockReset();
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
|
||||
});
|
||||
afterEach(async () => {
|
||||
await coordinator?.stop(); coordinator = undefined;
|
||||
vi.useRealTimers(); vi.restoreAllMocks(); vi.unstubAllEnvs();
|
||||
await db.execute(sql`drop trigger if exists decision_effect_finish_fail on decisions`);
|
||||
await db.execute(sql`drop function if exists decision_effect_finish_fail()`);
|
||||
await db.delete(decisionEffectExecutions); await db.delete(decisionTargetIssues); await db.delete(decisions);
|
||||
await db.delete(decisionRetention); await db.delete(activityLog); await db.delete(issueComments); await db.delete(heartbeatRuns);
|
||||
await db.delete(issues); await db.delete(agents); await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies);
|
||||
expect(errors).toEqual([]);
|
||||
});
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
|
||||
const actor = () => ({ type: "agent" as const, companyId, agentId, runId, source: "agent_jwt" as const,
|
||||
onBehalfOfUserId: userId, onBehalfOfMemberships: [{ companyId, membershipRole: "member", status: "active" }] });
|
||||
const board = () => ({ type: "board" as const, userId, companyIds: [companyId], source: "session" as const,
|
||||
memberships: [{ companyId, membershipRole: "member", status: "active" }] });
|
||||
const create = () => service().create({ companyId, agentId, runId, actor: actor(), title: "Comment?", body: "Body",
|
||||
continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [
|
||||
{ type: "comment_on_issue", targetIssueId: targetId, staleness: "lenient", bodyMarkdown: "hello" },
|
||||
] }] });
|
||||
const decide = (id: string) => service().decide({ id, optionId: "yes", decidedByUserId: userId, userActor: board() });
|
||||
const claim = (id: string, updatedAt = new Date(), owner = db) => writeDecisionChange(owner, async tx => tx.update(decisions).set({
|
||||
status: "decided", executionStatus: "running", chosenOptionId: "yes", decidedByUserId: userId,
|
||||
updatedAt, inputValues: {}, metadata: { continuationPending: true },
|
||||
}).where(eq(decisions.id, id)));
|
||||
async function start() {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
|
||||
onError: error => errors.push(error) });
|
||||
await registerDecisionContinuationWork(coordinator, service(), () => !draining).ready;
|
||||
recovery = service();
|
||||
vi.spyOn(recovery, "recoverEffects");
|
||||
await registerDecisionEffectRecoveryWork(coordinator, recovery, () => !draining).ready;
|
||||
}
|
||||
async function quiet() {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 5000 });
|
||||
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute");
|
||||
await vi.advanceTimersByTimeAsync(60 * 60_000);
|
||||
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
|
||||
reads.mockRestore(); writes.mockRestore(); executes.mockRestore();
|
||||
}
|
||||
async function settled(id: string, executionStatus = "succeeded") {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await service().get(id)).toMatchObject({ executionStatus, metadata: { continuationPending: false } });
|
||||
}, { timeout: 5000 });
|
||||
}
|
||||
const comments = () => db.select().from(issueComments).where(eq(issueComments.issueId, targetId));
|
||||
async function failSettlement() {
|
||||
await db.execute(sql`create function decision_effect_finish_fail() returns trigger language plpgsql as $$ begin
|
||||
if OLD.execution_status = 'running' and NEW.execution_status <> 'running' then raise exception 'settlement unavailable'; end if;
|
||||
return NEW; end $$`);
|
||||
await db.execute(sql`create trigger decision_effect_finish_fail before update on decisions for each row execute function decision_effect_finish_fail()`);
|
||||
}
|
||||
|
||||
it("stays quiet with open decisions and delivered history after a real decision", async () => {
|
||||
await start(); await quiet(); const row = await create(); await quiet();
|
||||
await decide(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("does not inspect effect recovery when an open decision is created", async () => {
|
||||
await start(); await quiet();
|
||||
const sweeps = vi.mocked(recovery.recoverEffects); sweeps.mockClear();
|
||||
await create(); await quiet();
|
||||
expect(sweeps).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("coalesces sequential committed cancellations without an idle recovery poll", async () => {
|
||||
const rows = [];
|
||||
for (let i = 0; i < 10; i++) rows.push(await create());
|
||||
// Freeze wall time while real SQL and transaction settlement continue.
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
|
||||
await start();
|
||||
const scans = vi.mocked(recovery.recoverEffects);
|
||||
expect(scans).toHaveBeenCalledTimes(1);
|
||||
for (const row of rows) {
|
||||
await service().cancel(row.id, { actorType: "user", actorId: userId });
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await db.execute(sql`select pg_sleep(0.01)`); // Allow each old scan to finish, rather than only testing overlap coalescing.
|
||||
}
|
||||
expect(scans).toHaveBeenCalledTimes(1);
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await vi.waitFor(() => expect(scans).toHaveBeenCalledTimes(2));
|
||||
await quiet();
|
||||
const count = scans.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(3_600_000);
|
||||
expect(scans).toHaveBeenCalledTimes(count);
|
||||
expect(wake).toHaveBeenCalledTimes(10);
|
||||
});
|
||||
|
||||
it("lets an already saved recovery deadline preempt notification coalescing", async () => {
|
||||
const due = await create(), cancelled = await create();
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false });
|
||||
const deadline = Date.now() + 500;
|
||||
await claim(due.id, new Date(deadline - 60_000));
|
||||
await start();
|
||||
expect(coordinator!.nextWakeAt()).toBe(deadline);
|
||||
await service().cancel(cancelled.id, { actorType: "user", actorId: userId });
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await db.execute(sql`select pg_sleep(0.01)`);
|
||||
expect(coordinator!.nextWakeAt()).toBe(deadline);
|
||||
await vi.advanceTimersByTimeAsync(499);
|
||||
await vi.waitFor(async () => expect(await comments()).toHaveLength(1));
|
||||
await settled(due.id); await quiet();
|
||||
expect(wake).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("retries an old failure while newer due claims keep filling pages", async () => {
|
||||
const failed = await create(), healthy = await create();
|
||||
const at = Date.now() - 600_000;
|
||||
await claim(failed.id, new Date(at)); await claim(healthy.id, new Date(at + 1));
|
||||
// Preserve PostgreSQL precision at both ends of the first rotation.
|
||||
await db.execute(sql`update decisions set updated_at = updated_at + interval '456 microseconds'
|
||||
where id in (${failed.id}, ${healthy.id})`);
|
||||
const failure = new Error("temporary recovery failure");
|
||||
const attempts: string[] = [];
|
||||
const scan = decisionEffectRecoveryService(db, async id => {
|
||||
attempts.push(id);
|
||||
if (id === failed.id) throw failure;
|
||||
await db.update(decisions).set({ executionStatus: "succeeded" }).where(eq(decisions.id, id));
|
||||
return true;
|
||||
});
|
||||
await expect(scan.sweep(new Date(), 1)).rejects.toBe(failure);
|
||||
const added: string[] = [];
|
||||
for (let index = 0; index < 7; index++) {
|
||||
const row = await create(); added.push(row.id); await claim(row.id, new Date(at + index + 2));
|
||||
try { await scan.sweep(new Date(), 1); } catch (error) { expect(error).toBe(failure); }
|
||||
}
|
||||
expect(attempts).toHaveLength(8);
|
||||
expect(attempts.slice(0, 6)).toEqual([failed.id, healthy.id, failed.id, added[0], added[1], failed.id]);
|
||||
expect(new Set(attempts.filter(id => id !== failed.id)).size).toBe(5);
|
||||
expect((await service().get(failed.id))?.executionStatus).toBe("running");
|
||||
});
|
||||
|
||||
it("reactivates from the actual claim and recovers a failed settlement without repeating the effect", async () => {
|
||||
await start(); await quiet(); const row = await create(); await failSettlement();
|
||||
await expect(decide(row.id)).rejects.toMatchObject({ cause: { message: "settlement unavailable" } });
|
||||
expect((await service().get(row.id))?.executionStatus).toBe("running"); expect(await comments()).toHaveLength(1);
|
||||
await db.execute(sql`drop trigger decision_effect_finish_fail on decisions`);
|
||||
await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("reconstructs the saved grace deadline at restart and recovers a claimed effect", async () => {
|
||||
const row = await create(); const claimedAt = new Date(); await claim(row.id, claimedAt);
|
||||
await db.insert(decisionEffectExecutions).values({ decisionId: row.id, effectIndex: 0, effectType: "comment_on_issue", targetIssueId: targetId, status: "claimed" });
|
||||
await start(); expect(coordinator!.nextWakeAt()).toBe(claimedAt.getTime() + 60_000);
|
||||
await vi.advanceTimersByTimeAsync(30_000); expect(await comments()).toHaveLength(0);
|
||||
await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("recovers at the exact grace boundary and preserves a zero grace override", async () => {
|
||||
const row = await create(), at = new Date(); await claim(row.id, at);
|
||||
expect(await service().recoverEffects(new Date(at.getTime() + 59_999))).toEqual({ resumed: 0 });
|
||||
expect(await service().recoverEffects(new Date(at.getTime() + 60_000))).toEqual({ resumed: 1 });
|
||||
vi.stubEnv("PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS", "0");
|
||||
const second = await create(); await claim(second.id, at);
|
||||
expect(await service().recoverEffects(at)).toEqual({ resumed: 1 }); expect(await comments()).toHaveLength(2);
|
||||
});
|
||||
|
||||
it("joins active foreground effects when their recovery deadline arrives", async () => {
|
||||
const row = await create(); await start();
|
||||
const effectKey = `decision-effect:${row.id}:0`;
|
||||
const previousDebug = db.$client.options.debug;
|
||||
let effectLockAttempts = 0;
|
||||
db.$client.options.debug = (_connection, query, parameters) => {
|
||||
if (query.includes("pg_advisory_xact_lock") && parameters.includes(effectKey)) effectLockAttempts++;
|
||||
};
|
||||
let release!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const lock = db.transaction(async tx => {
|
||||
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${effectKey}, 0))`);
|
||||
locked(); await gate;
|
||||
});
|
||||
let deciding: ReturnType<typeof decide> | undefined;
|
||||
let recovering: ReturnType<typeof recovery.recoverEffects> | undefined;
|
||||
try {
|
||||
await acquired;
|
||||
deciding = decide(row.id);
|
||||
await vi.waitFor(async () => expect((await service().get(row.id))?.executionStatus).toBe("running"));
|
||||
const claimed = (await service().get(row.id))!;
|
||||
const deadline = claimed.updatedAt.getTime() + 60_000;
|
||||
await vi.waitFor(async () => {
|
||||
const scheduled = coordinator!.nextWakeAt();
|
||||
if (scheduled !== null && scheduled < deadline) {
|
||||
await vi.advanceTimersByTimeAsync(Math.max(1, scheduled - Date.now()));
|
||||
}
|
||||
expect(coordinator!.nextWakeAt()).toBe(deadline);
|
||||
});
|
||||
await vi.waitFor(() => expect(effectLockAttempts).toBe(2)); // Test gate + foreground execution.
|
||||
const sweeps = vi.mocked(recovery.recoverEffects);
|
||||
const beforeDeadline = sweeps.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(Math.max(0, deadline - Date.now()));
|
||||
await vi.waitFor(() => expect(sweeps).toHaveBeenCalledTimes(beforeDeadline + 1));
|
||||
let recoverySettled = false;
|
||||
recovering = Promise.resolve(sweeps.mock.results[beforeDeadline].value).then(result => {
|
||||
recoverySettled = true;
|
||||
return result;
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(await comments()).toHaveLength(0); expect(wake).not.toHaveBeenCalled();
|
||||
expect(recoverySettled).toBe(false);
|
||||
expect(sweeps).toHaveBeenCalledTimes(beforeDeadline + 1);
|
||||
expect(effectLockAttempts).toBe(2);
|
||||
release();
|
||||
await lock; await deciding;
|
||||
await expect(recovering).resolves.toEqual({ resumed: 1 });
|
||||
await quiet();
|
||||
expect(effectLockAttempts).toBe(2);
|
||||
expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
release();
|
||||
await Promise.allSettled([lock, deciding, recovering]);
|
||||
db.$client.options.debug = previousDebug;
|
||||
}
|
||||
});
|
||||
|
||||
it("defers an outer transaction until commit and ignores rollback", async () => {
|
||||
const row = await create(); await start(); await quiet();
|
||||
await expect(db.transaction(async tx => {
|
||||
await claim(row.id, new Date(Date.now() - 120_000), tx as unknown as typeof db);
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(await comments()).toHaveLength(0); throw new Error("rollback");
|
||||
})).rejects.toThrow("rollback");
|
||||
await quiet(); expect((await service().get(row.id))?.status).toBe("open");
|
||||
await db.transaction(async tx => {
|
||||
await claim(row.id, new Date(Date.now() - 120_000), tx as unknown as typeof db);
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(await comments()).toHaveLength(0);
|
||||
});
|
||||
await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("retains a first claim through drain without passive SQL", async () => {
|
||||
const row = await create(); await start(); await quiet(); draining = true;
|
||||
await claim(row.id, new Date(Date.now() - 120_000));
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
draining = false; await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("retains a failed recovery through drain and does not replay the completed effect", async () => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000)); await failSettlement(); await start();
|
||||
await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); draining = true;
|
||||
expect(await comments()).toHaveLength(1);
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await db.execute(sql`drop trigger decision_effect_finish_fail on decisions`);
|
||||
draining = false; await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
|
||||
for (const error of errors) expect(error).toMatchObject({ cause: { message: "settlement unavailable" } }); errors.length = 0;
|
||||
});
|
||||
|
||||
it("rechecks current membership before executing a recovered effect", async () => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
|
||||
await db.update(companyMemberships).set({ status: "inactive" }).where(eq(companyMemberships.principalId, userId));
|
||||
await start(); await settled(row.id, "failed"); await quiet(); expect(await comments()).toHaveLength(0);
|
||||
const result = await service().outcome(row.id);
|
||||
expect(result.executions[0]?.error).toBe("deny_decision_intersection"); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("leaves continuation failures to their owner after effect recovery finishes", async () => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
|
||||
wake.mockRejectedValueOnce(new Error("wake unavailable")); await start(); await settled(row.id); await quiet();
|
||||
expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(2);
|
||||
expect(errors.every(error => String(error).includes("wake unavailable"))).toBe(true); errors.length = 0;
|
||||
});
|
||||
|
||||
it("keeps aggregate maintenance from recovering effects when the worker owns them", async () => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
|
||||
expect(await service().sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false }))
|
||||
.toEqual({ expired: 0, resumed: 0 });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(await service().sweepExpired()).toEqual({ expired: 0, resumed: 1 }); expect(await comments()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["missing_decider", "missing_option"] as const)("settles %s as durable failure and then becomes quiet", async reason => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
|
||||
await db.update(decisions).set(reason === "missing_decider" ? { decidedByUserId: null } : { chosenOptionId: "missing" })
|
||||
.where(eq(decisions.id, row.id));
|
||||
expect(await service().recoverEffects()).toEqual({ resumed: 1 });
|
||||
expect(await service().get(row.id)).toMatchObject({ executionStatus: "failed", metadata: {
|
||||
effectRecoveryError: reason, continuationPending: false,
|
||||
} });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(await db.select().from(decisionEffectExecutions)).toHaveLength(0);
|
||||
const audit = await db.select().from(activityLog).where(eq(activityLog.action, "decision.recovery_failed"));
|
||||
expect(audit).toHaveLength(1);
|
||||
expect(audit[0]).toMatchObject({ companyId, actorType: "system", actorId: "decision-executor",
|
||||
entityId: row.id, details: { reason } });
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
await start(); await quiet();
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("preserves completed effect receipts when a malformed outcome settles as partial", async () => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
|
||||
await db.update(decisions).set({ chosenOptionId: "missing" }).where(eq(decisions.id, row.id));
|
||||
const [receipt] = await db.insert(decisionEffectExecutions).values({ decisionId: row.id, effectIndex: 0,
|
||||
effectType: "comment_on_issue", targetIssueId: targetId, status: "executed", result: { preserved: true }, executedAt: new Date() }).returning();
|
||||
expect(await service().recoverEffects()).toEqual({ resumed: 1 });
|
||||
expect((await service().get(row.id))?.executionStatus).toBe("partial");
|
||||
expect(await db.select().from(decisionEffectExecutions)).toEqual([receipt]);
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
await start(); await quiet();
|
||||
});
|
||||
|
||||
it("keeps malformed claims pending when their failure audit cannot commit", async () => {
|
||||
const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
|
||||
await db.update(decisions).set({ decidedByUserId: null }).where(eq(decisions.id, row.id));
|
||||
await db.execute(sql`create function decision_recovery_audit_fail() returns trigger language plpgsql as $$ begin
|
||||
if NEW.action = 'decision.recovery_failed' then raise exception 'recovery audit unavailable'; end if;
|
||||
return NEW; end $$`);
|
||||
await db.execute(sql`create trigger decision_recovery_audit_fail before insert on activity_log
|
||||
for each row execute function decision_recovery_audit_fail()`);
|
||||
try {
|
||||
await expect(service().recoverEffects()).rejects.toMatchObject({ cause: { message: "recovery audit unavailable" } });
|
||||
expect((await service().get(row.id))?.executionStatus).toBe("running");
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.action, "decision.recovery_failed"))).toHaveLength(0);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
await db.execute(sql`drop trigger decision_recovery_audit_fail on activity_log`);
|
||||
await db.execute(sql`drop function decision_recovery_audit_fail()`);
|
||||
}
|
||||
expect(await service().recoverEffects()).toEqual({ resumed: 1 });
|
||||
expect((await service().get(row.id))?.executionStatus).toBe("failed");
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
await start(); await quiet();
|
||||
});
|
||||
|
||||
it("rechecks a malformed claim under lock before settling a concurrently repaired outcome", async () => {
|
||||
const row = await create(); const at = new Date(Date.now() - 120_000); await claim(row.id, at);
|
||||
await db.update(decisions).set({ chosenOptionId: "missing" }).where(eq(decisions.id, row.id));
|
||||
let release!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const repair = db.transaction(async tx => {
|
||||
await tx.select({ id: decisions.id }).from(decisions).where(eq(decisions.id, row.id)).for("update");
|
||||
locked(); await gate;
|
||||
await tx.update(decisions).set({ chosenOptionId: "yes", updatedAt: at }).where(eq(decisions.id, row.id));
|
||||
});
|
||||
const previousDebug = db.$client.options.debug;
|
||||
let attempts = 0;
|
||||
let pending: ReturnType<ReturnType<typeof decisionService>["recoverEffects"]> | undefined;
|
||||
try {
|
||||
await acquired;
|
||||
db.$client.options.debug = (_connection, query, parameters) => {
|
||||
if (query.includes("for update") && parameters.includes(row.id)) attempts++;
|
||||
};
|
||||
pending = service().recoverEffects();
|
||||
await vi.waitFor(() => expect(attempts).toBe(1));
|
||||
release(); await repair;
|
||||
expect(await pending).toEqual({ resumed: 0 });
|
||||
expect(await service().get(row.id)).toMatchObject({ executionStatus: "running", chosenOptionId: "yes" });
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.action, "decision.recovery_failed"))).toHaveLength(0);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
expect(await service().recoverEffects()).toEqual({ resumed: 1 });
|
||||
expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
release(); await Promise.allSettled([repair, pending]); db.$client.options.debug = previousDebug;
|
||||
}
|
||||
await start(); await quiet();
|
||||
});
|
||||
|
||||
it("settles invalid legacy claims without inventing authority or blocking later effects", async () => {
|
||||
vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1");
|
||||
const invalid = await create(), valid = await create();
|
||||
await claim(invalid.id, new Date(Date.now() - 180_000)); await claim(valid.id, new Date(Date.now() - 120_000));
|
||||
await db.update(decisions).set({ decidedByUserId: null }).where(eq(decisions.id, invalid.id));
|
||||
const recovery = service();
|
||||
expect(await recovery.recoverEffects()).toEqual({ resumed: 1 });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(await recovery.recoverEffects()).toEqual({ resumed: 1 }); expect(await comments()).toHaveLength(1);
|
||||
expect((await recovery.get(invalid.id))?.executionStatus).toBe("failed");
|
||||
expect(await recovery.nextEffectRecoveryAt()).toBeNull();
|
||||
});
|
||||
|
||||
it("pages past unresolved claims using database timestamp precision", async () => {
|
||||
const pending = await create(), valid = await create();
|
||||
await claim(pending.id); await claim(valid.id, new Date(Date.now() - 120_000));
|
||||
await db.execute(sql`update decisions set updated_at = date_trunc('milliseconds', now() - interval '3 minutes')
|
||||
+ interval '456 microseconds' where id = ${pending.id}`);
|
||||
const scan = decisionEffectRecoveryService(db, async id => {
|
||||
if (id === pending.id) return false;
|
||||
await decide(id); return true;
|
||||
});
|
||||
expect(await scan.sweep(new Date(), 1)).toEqual({ resumed: 0 });
|
||||
expect(await scan.sweep(new Date(), 1)).toEqual({ resumed: 1 });
|
||||
expect(await comments()).toHaveLength(1);
|
||||
expect((await service().get(pending.id))?.executionStatus).toBe("running");
|
||||
expect(await scan.next()).not.toBeNull();
|
||||
expect(await scan.sweep(new Date(), 1)).toEqual({ resumed: 0 });
|
||||
expect(await scan.sweep(new Date(), 1)).toEqual({ resumed: 0 });
|
||||
expect(await comments()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each([false, true])("preserves archive recovery semantics with prior committed archive: %s", async committed => {
|
||||
await db.update(issues).set({ status: "in_review", assigneeUserId: userId }).where(eq(issues.id, targetId));
|
||||
const activityAt = new Date("2026-04-01T00:00:00.000Z");
|
||||
await db.insert(decisionRetention).values({ companyId, sourceKind: "review", sourceId: targetId, sourceActivityAt: activityAt });
|
||||
const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetId,
|
||||
expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }];
|
||||
const row = await service().create({ companyId, actor: actor(), agentId, runId, title: "Archive?", body: "Exact set",
|
||||
continuationPolicy: "wake_origin_agent", options: [{ id: "archive", label: "Archive", effects: [] }],
|
||||
metadata: { kind: "attention_archive_proposal", manifestHash: hashAttentionArchiveManifest(manifest) },
|
||||
additionalTargetSnapshots: { [`attention:review:${targetId}`]: { status: "attention", assigneeAgentId: null,
|
||||
assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } });
|
||||
if (committed) {
|
||||
await failSettlement();
|
||||
await expect(service().decide({ id: row.id, optionId: "archive", decidedByUserId: userId, userActor: board() }))
|
||||
.rejects.toMatchObject({ cause: { message: "settlement unavailable" } });
|
||||
await db.execute(sql`drop trigger decision_effect_finish_fail on decisions`);
|
||||
} else {
|
||||
await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running",
|
||||
chosenOptionId: "archive", decidedByUserId: userId, updatedAt: new Date(Date.now() - 120_000),
|
||||
metadata: { ...row.metadata, continuationPending: true } }).where(eq(decisions.id, row.id)));
|
||||
}
|
||||
await start(); await settled(row.id, committed ? "failed" : "succeeded"); await quiet();
|
||||
const [retained] = await db.select().from(decisionRetention);
|
||||
expect(retained.archiveVersion).toBe(1); expect(retained.archivedAt).not.toBeNull();
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
if (committed) expect((await service().get(row.id))?.metadata).toHaveProperty("archiveProposalError");
|
||||
});
|
||||
|
||||
it("learns another claim written during delivery", async () => {
|
||||
const first = await create(), second = await create(); await claim(first.id, new Date(Date.now() - 120_000));
|
||||
wake.mockImplementationOnce(async () => { await claim(second.id); });
|
||||
await start(); await settled(first.id); await settled(second.id); await quiet();
|
||||
expect(await comments()).toHaveLength(2); expect(wake).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("finishes the admitted recovery and leaves later claims for restart during shutdown", async () => {
|
||||
const first = await create(), second = await create();
|
||||
await claim(first.id, new Date(Date.now() - 180_000)); await claim(second.id, new Date(Date.now() - 120_000));
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
wake.mockImplementationOnce(async () => gate);
|
||||
const starting = start(); let stopped = false; let stopping: Promise<void>;
|
||||
try {
|
||||
await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1));
|
||||
stopping = coordinator!.stop().then(() => { stopped = true; });
|
||||
await vi.advanceTimersByTimeAsync(60_000); expect(stopped).toBe(false);
|
||||
} finally { release(); }
|
||||
await starting; await stopping!; coordinator = undefined;
|
||||
expect(wake).toHaveBeenCalledTimes(1); expect(await comments()).toHaveLength(1);
|
||||
expect((await service().get(second.id))?.executionStatus).toBe("running");
|
||||
await start(); await quiet(); expect(await comments()).toHaveLength(2); expect(wake).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
@@ -349,6 +349,8 @@ vi.mock("../services/index.js", () => ({
|
||||
})),
|
||||
decisionService: vi.fn(() => ({
|
||||
sweepExpired: vi.fn(async () => ({ expired: 0 })),
|
||||
recoverEffects: vi.fn(async () => ({ resumed: 0 })),
|
||||
nextEffectRecoveryAt: vi.fn(async () => null),
|
||||
sweepTtlExpiry: vi.fn(async () => ({ expired: 0 })),
|
||||
nextExpiryAt: vi.fn(async () => null),
|
||||
sweepContinuations: vi.fn(async () => ({ retry: false })),
|
||||
@@ -921,8 +923,10 @@ describe("startServer feedback export wiring", () => {
|
||||
const decisions = vi.mocked(decisionService).mock.results.at(-1)!.value;
|
||||
expect(decisions.sweepContinuations).toHaveBeenCalledTimes(1);
|
||||
expect(decisions.sweepTtlExpiry).toHaveBeenCalledTimes(1);
|
||||
expect(decisions.recoverEffects).toHaveBeenCalledTimes(1);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionEffectRecovery)).toBe(true);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionExpiry)).toBe(true);
|
||||
expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false, includeExpiry: false });
|
||||
expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false, includeExpiry: false, includeEffects: false });
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
@@ -961,6 +965,7 @@ describe("startServer feedback export wiring", () => {
|
||||
expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled();
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(false);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionExpiry)).toBe(false);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionEffectRecovery)).toBe(false);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
|
||||
+5
-2
@@ -1,3 +1,4 @@
|
||||
import { registerDecisionEffectRecoveryWork } from "./services/decision-effect-recovery-work.js";
|
||||
import { registerDecisionExpiryWork } from "./services/decision-expiry-work.js";
|
||||
import { registerDecisionContinuationWork } from "./services/decision-continuation-work.js";
|
||||
|
||||
@@ -1842,7 +1843,9 @@ async function startServerWithDatabaseTeardown(
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
await step("decision_expiry_worker", () => registerDecisionExpiryWork(deliveryWork, decisionExecutor,
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false }));
|
||||
await step("decision_effect_recovery_worker", () => registerDecisionEffectRecoveryWork(deliveryWork, decisionExecutor,
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false }));
|
||||
const loginSchedule = loginCleanupSchedule(db, config.heartbeatSchedulerIntervalMs);
|
||||
await step("login_cleanup_workers", () => registerLoginCleanupWork(deliveryWork, [
|
||||
{ queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult), next: loginSchedule.adapter },
|
||||
@@ -1863,7 +1866,7 @@ async function startServerWithDatabaseTeardown(
|
||||
trackHeartbeatSchedulerWork((async () => {
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
scheduleRoutineStatusReconcile();
|
||||
trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false }).catch((err: unknown) => {
|
||||
trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false }).catch((err: unknown) => {
|
||||
logger.error({ err }, "decision expiry sweep failed");
|
||||
}));
|
||||
trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => {
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { registerDeadlineWork } from "./deadline-work.js";
|
||||
import type { decisionService } from "./decisions.js";
|
||||
|
||||
export function registerDecisionEffectRecoveryWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
service: Pick<ReturnType<typeof decisionService>, "recoverEffects" | "nextEffectRecoveryAt">,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionEffectRecovery, {
|
||||
retryMs: 30_000, canRun,
|
||||
notificationCoalesceMs: 1000,
|
||||
run: signal => service.recoverEffects(new Date(), undefined, signal),
|
||||
next: service.nextEffectRecoveryAt,
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import { and, asc, desc, lte, sql } from "drizzle-orm";
|
||||
import { decisions, type Db } from "@paperclipai/db";
|
||||
|
||||
const running = sql`${decisions.status} = 'decided' AND ${decisions.executionStatus} = 'running'`;
|
||||
|
||||
export function decisionSweepBatchSize() {
|
||||
const value = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100);
|
||||
return Number.isFinite(value) ? Math.max(1, Math.trunc(value)) : 100;
|
||||
}
|
||||
|
||||
function recoveryGraceMs() {
|
||||
const value = Number(process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS ?? 60_000);
|
||||
return Number.isFinite(value) && value >= 0 ? value : 60_000;
|
||||
}
|
||||
|
||||
export function decisionEffectRecoveryService(db: Db, resume: (id: string) => Promise<boolean>) {
|
||||
type Position = { updatedAt: string; id: string };
|
||||
let cursor: Position | undefined;
|
||||
let boundary: Position | undefined;
|
||||
|
||||
async function sweep(now = new Date(), limit = decisionSweepBatchSize(), signal?: AbortSignal) {
|
||||
let resumed = 0;
|
||||
if (signal?.aborted) return { resumed };
|
||||
const dueBefore = new Date(now.getTime() - recoveryGraceMs());
|
||||
const position = { id: decisions.id, updatedAt: sql<string>`${decisions.updatedAt}::text` };
|
||||
if (!boundary) {
|
||||
// Finish this rotation even if newer due claims arrive on every page.
|
||||
[boundary] = await db.select(position).from(decisions)
|
||||
.where(and(running, lte(decisions.updatedAt, dueBefore)))
|
||||
.orderBy(desc(decisions.updatedAt), desc(decisions.id)).limit(1);
|
||||
if (!boundary) return { resumed };
|
||||
}
|
||||
const rows = await db.select(position).from(decisions)
|
||||
.where(and(running, lte(decisions.updatedAt, dueBefore),
|
||||
sql`(${decisions.updatedAt}, ${decisions.id}) <= (${boundary.updatedAt}::timestamptz, ${boundary.id}::uuid)`, cursor
|
||||
? sql`(${decisions.updatedAt}, ${decisions.id}) > (${cursor.updatedAt}::timestamptz, ${cursor.id}::uuid)` : undefined))
|
||||
.orderBy(asc(decisions.updatedAt), asc(decisions.id)).limit(limit);
|
||||
const errors: unknown[] = [];
|
||||
for (const row of rows) {
|
||||
if (signal?.aborted) return { resumed };
|
||||
try { if (await resume(row.id)) resumed++; } catch (error) { errors.push(error); }
|
||||
cursor = row;
|
||||
}
|
||||
const last = rows.at(-1);
|
||||
if (rows.length < limit || (last?.id === boundary.id && last.updatedAt === boundary.updatedAt)) {
|
||||
cursor = undefined; boundary = undefined;
|
||||
}
|
||||
if (errors.length === 1) throw errors[0];
|
||||
if (errors.length) throw new AggregateError(errors, "Decision effect recovery failed");
|
||||
return { resumed };
|
||||
}
|
||||
|
||||
async function next(): Promise<number | null> {
|
||||
const [row] = await db.select({ at: decisions.updatedAt }).from(decisions).where(running)
|
||||
.orderBy(asc(decisions.updatedAt), asc(decisions.id)).limit(1);
|
||||
return row ? row.at.getTime() + recoveryGraceMs() : null;
|
||||
}
|
||||
|
||||
return { sweep, next };
|
||||
}
|
||||
@@ -13,6 +13,7 @@ export async function notifyDecisionChange(transaction: object) {
|
||||
export function writeDecisionChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
|
||||
return db.transaction(async transaction => {
|
||||
await notifyDecisionChange(transaction);
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionEffectRecovery);
|
||||
return write(transaction as unknown as Db);
|
||||
});
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, asc, count, desc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm";
|
||||
import { and, asc, count, desc, eq, gt, gte, inArray, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { databaseWorkScopeKey, companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db";
|
||||
import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared";
|
||||
@@ -15,6 +15,7 @@ import {
|
||||
import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js";
|
||||
|
||||
import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js";
|
||||
import { decisionEffectRecoveryService, decisionSweepBatchSize } from "./decision-effect-recovery.js";
|
||||
import { decisionExpiryService } from "./decision-expiry.js";
|
||||
import { notifyDecisionCreated, writeDecisionChange } from "./decision-work-notifications.js";
|
||||
|
||||
@@ -24,7 +25,8 @@ type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId
|
||||
export type DecisionServiceOptions = { wakeOriginAgent: DecisionContinuationWake };
|
||||
const DAY = 86_400_000;
|
||||
type DecisionOutcome = typeof decisions.$inferSelect & { executions: Array<typeof decisionEffectExecutions.$inferSelect> };
|
||||
const effectFlights = new WeakMap<object, Map<string, Promise<DecisionOutcome>>>();
|
||||
type EffectAttempt = { outcome: DecisionOutcome; resumed: boolean };
|
||||
const effectFlights = new WeakMap<object, Map<string, Promise<EffectAttempt | null>>>();
|
||||
|
||||
function targetIds(options: DecisionOption[]) {
|
||||
const result = new Set<string>();
|
||||
@@ -126,6 +128,7 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) {
|
||||
export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
const continuations = decisionContinuationService(db, options.wakeOriginAgent);
|
||||
const expiry = decisionExpiryService(db, continuations.deliver);
|
||||
const effectRecovery = decisionEffectRecoveryService(db, recoverDecisionEffects);
|
||||
const scope = databaseWorkScopeKey(db);
|
||||
let flights = effectFlights.get(scope);
|
||||
if (!flights) effectFlights.set(scope, flights = new Map());
|
||||
@@ -557,21 +560,53 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
}
|
||||
}
|
||||
|
||||
function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor): Promise<DecisionOutcome> {
|
||||
const existing = executing.get(decision.id);
|
||||
function runEffects(id: string, userActor?: AuthorizationActor): Promise<EffectAttempt | null> {
|
||||
const existing = executing.get(id);
|
||||
if (existing) return existing;
|
||||
const attempt = (async () => {
|
||||
const current = await get(decision.id);
|
||||
if (!current) throw notFound("Decision not found");
|
||||
return current.executionStatus === "running" ? executeDecisionEffects(current, userActor) : outcome(current.id);
|
||||
})().finally(() => { executing.delete(decision.id); });
|
||||
executing.set(decision.id, attempt);
|
||||
const current = await get(id);
|
||||
if (!current) return null;
|
||||
if (current.status !== "decided" || current.executionStatus !== "running") {
|
||||
return { outcome: await outcome(id), resumed: false };
|
||||
}
|
||||
// Recovery derives authority from the current claim inside the shared flight.
|
||||
const actor = userActor ?? (current.decidedByUserId
|
||||
? await recoveryActor(current.companyId, current.decidedByUserId)
|
||||
: { type: "none" as const, source: "none" as const });
|
||||
return { outcome: await executeDecisionEffects(current, actor), resumed: true };
|
||||
})().finally(() => { executing.delete(id); });
|
||||
executing.set(id, attempt);
|
||||
return attempt;
|
||||
}
|
||||
|
||||
async function settleInvalidEffectOutcome(id: string) {
|
||||
const publications: ActivityPublication[] = [];
|
||||
await writeDecisionChange(db, async tx => {
|
||||
const [current] = await tx.select().from(decisions).where(eq(decisions.id, id)).for("update");
|
||||
if (!current || current.status !== "decided" || current.executionStatus !== "running") return;
|
||||
const reason = !current.decidedByUserId ? "missing_decider"
|
||||
: !current.options.some(option => option.id === current.chosenOptionId) ? "missing_option" : null;
|
||||
if (!reason) return;
|
||||
const [executed] = await tx.select({ id: decisionEffectExecutions.id }).from(decisionEffectExecutions)
|
||||
.where(and(eq(decisionEffectExecutions.decisionId, id), eq(decisionEffectExecutions.status, "executed"))).limit(1);
|
||||
const executionStatus = executed ? "partial" : "failed";
|
||||
await tx.update(decisions).set({ executionStatus, updatedAt: new Date(), metadata: { ...current.metadata,
|
||||
effectRecoveryError: reason,
|
||||
...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}),
|
||||
} }).where(eq(decisions.id, id));
|
||||
await logActivity(tx, { companyId: current.companyId, actorType: "system", actorId: "decision-executor",
|
||||
agentId: current.originAgentId, runId: current.originRunId, responsibleUserIdOverride: current.decidedByUserId,
|
||||
action: "decision.recovery_failed", entityType: "decision", entityId: id,
|
||||
details: { reason, executionStatus, chosenOptionId: current.chosenOptionId, decidedByUserId: current.decidedByUserId },
|
||||
}, publications);
|
||||
});
|
||||
for (const publication of publications) publishActivity(publication);
|
||||
return outcome(id);
|
||||
}
|
||||
|
||||
async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
|
||||
const option = decision.options.find((item) => item.id === decision.chosenOptionId);
|
||||
if (!option || !decision.decidedByUserId) throw unprocessable("Stored decision outcome is invalid");
|
||||
if (!option || !decision.decidedByUserId) return settleInvalidEffectOutcome(decision.id);
|
||||
const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null);
|
||||
const metadata = decision.metadata as Record<string, unknown>;
|
||||
if (metadata.kind === "attention_archive_proposal") {
|
||||
@@ -635,7 +670,9 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
}
|
||||
|
||||
async function resumeDecision(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
|
||||
const result = decision.executionStatus === "running" ? await runEffects(decision, userActor) : await outcome(decision.id);
|
||||
const attempt = decision.executionStatus === "running" ? await runEffects(decision.id, userActor) : undefined;
|
||||
if (attempt === null) throw notFound("Decision not found");
|
||||
const result = attempt?.outcome ?? await outcome(decision.id);
|
||||
await continuations.deliver(result.id);
|
||||
return outcome(decision.id);
|
||||
}
|
||||
@@ -716,27 +753,16 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
return { ...bundle, decisions: created }; });
|
||||
}
|
||||
|
||||
async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean } = {}) {
|
||||
const configuredBatchSize = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100);
|
||||
const batchSize = Number.isFinite(configuredBatchSize)
|
||||
? Math.max(1, Math.trunc(configuredBatchSize))
|
||||
: 100;
|
||||
const configuredRecoveryGraceMs = Number(process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS ?? 60_000);
|
||||
const recoveryGraceMs = Number.isFinite(configuredRecoveryGraceMs) && configuredRecoveryGraceMs >= 0
|
||||
? configuredRecoveryGraceMs
|
||||
: 60_000;
|
||||
const runningRows = await db.select().from(decisions)
|
||||
.where(and(eq(decisions.status, "decided"), eq(decisions.executionStatus, "running"),
|
||||
lte(decisions.updatedAt, new Date(now.getTime() - recoveryGraceMs))))
|
||||
.orderBy(asc(decisions.updatedAt)).limit(batchSize);
|
||||
let resumed = 0;
|
||||
for (const decision of runningRows) {
|
||||
if (!decision.decidedByUserId) continue;
|
||||
const recovered = await runEffects(decision, await recoveryActor(decision.companyId, decision.decidedByUserId));
|
||||
if (recovered.executionStatus === "running") continue;
|
||||
resumed += 1;
|
||||
await continuations.deliver(recovered.id);
|
||||
}
|
||||
async function recoverDecisionEffects(id: string) {
|
||||
const attempt = await runEffects(id);
|
||||
if (!attempt?.resumed || attempt.outcome.executionStatus === "running") return false;
|
||||
await continuations.deliver(attempt.outcome.id);
|
||||
return true;
|
||||
}
|
||||
|
||||
async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean; includeEffects?: boolean } = {}) {
|
||||
const batchSize = decisionSweepBatchSize();
|
||||
const { resumed } = options.includeEffects === false ? { resumed: 0 } : await effectRecovery.sweep(now, batchSize);
|
||||
if (options.includeContinuations !== false) await continuations.sweep(batchSize);
|
||||
const ttlRows = options.includeExpiry === false ? [] : await expiry.due(now, batchSize);
|
||||
const remaining = batchSize - ttlRows.length;
|
||||
@@ -768,5 +794,6 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
}
|
||||
|
||||
return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep,
|
||||
sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next };
|
||||
sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next,
|
||||
recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next };
|
||||
}
|
||||
@@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = {
|
||||
executionStatus: "execution-status",
|
||||
executionDisposition: "execution-disposition",
|
||||
environmentCleanup: "environment-cleanup",
|
||||
decisionEffectRecovery: "decision-effect-recovery",
|
||||
decisionExpiry: "decision-expiry",
|
||||
decisionContinuation: "decision-continuation",
|
||||
|
||||
|
||||
Reference in new issue
Block a user