mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
fix(server): wake decision target checks from committed changes (#17486)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Decisions can depend on strict references to other tasks. > - A cancelled, deleted, or completed target can make a decision obsolete. > - The server currently scans open decisions on every maintenance tick to find those changes. > - The owning issue writes already have transaction boundaries that can signal the change. > - This pull request reconciles affected decisions after commit and reconstructs open targets at startup. > - Valid open decisions and completed history stop causing periodic SQL. ## Linked Issues or Issue Description **What existing behavior does this improve?** Decision target invalidation. This is slice 12 of the recurring-server sleep stack, based on #17482. TTL expiry is #17479 and continuation delivery is #17478. Retention remains separate work. **Subsystem affected** Server decision policy, issue mutation and deletion, and startup wiring. **Current behavior** The aggregate sweep repeatedly reads open decisions and their strict targets. It shares a maintenance tick with unrelated jobs even when every target remains valid. **Proposed behavior** Committed decision creation and terminal issue updates request a scoped check. Deletion requests a bounded full scan because its target links can cascade away. Startup reconstructs open target state. Once reconciliation settles, no target scan remains scheduled. **Reason and benefit** Remove normal empty polling without losing target invalidation, transactional delivery, or restart recovery. Reuse existing decision rows and target links instead of adding a table. **Breaking changes** No API or schema change. The main tick stops calling the combined decision sweep after its four duties have separate owners. The public combined method retains its default behavior. **Additional context** The initial decision implementation stored secondary links as well as primary links. This migration does not assume a new legacy backfill. Independent SQL writers still require an explicit signal integration. ## What Changed - Replace the periodic target scan with committed-change activation and bounded startup reconstruction. Keep the existing decision policy and expiry/activity owner. - Require production notifications to identify an issue or decision. Before deletion, lock the issue and capture its linked open decisions in the same company; emit their IDs before links cascade. This avoids resetting full scans and fences concurrent proposal creation. - Reuse that owner helper for issue removal and routine failure cleanup. No linked decisions means no target signal. Remove company deletion's ineffective global signal; its pre-existing restrictive decision foreign keys remain unchanged. - Retain PostgreSQL timestamp text in paging cursors so rows within one millisecond cannot repeat indefinitely. - Remove the unused combined internal decision sweep after its final duty is split; exercise each owner directly in the existing behavioral tests. - Keep query policy and notification discovery in `decision-targets.ts`. The coordinator registration contains lifecycle wiring only. Add no queue table, per-route recovery wrapper or external-wake exemption. ## Verification Current candidate normally integrates landed effect recovery `e9e84bb8a1`. Its runtime and acceptance tests are byte-identical to tested target head `85ce6545bf`; the only new source files are four independently merged Runner E2E accounting files, preserved exactly. The 248-case/type/build evidence comes from test revision `54936a919f36c2ea421e7ce8e15912b1105b8dcc`, followed by documentation-only changes. Fresh current-head CI and reviews are required. This remains the target-invalidation duty only. - **248 cases pass** in all seven complete suites: 25 target, 80 routine, 54 decision, 26 effect-recovery, 20 expiry, 42 startup and one scoped-deadline case. Database suites use separate migrated PostgreSQL fixtures. Server production typecheck and build pass. - **Four regressions failed before repair:** paging across microsecond timestamps, deletion behind a full valid startup page, unnecessary global work from unrelated deletion, and unnecessary work after deleting an empty company. All four pass. - A real concurrent proposal is observed waiting on the issue-row lock between pre-delete discovery and deletion commit. Removing that lock made the test fail because the proposal committed instead of waiting. Restoring the implementation passes. The mutation was restored before commit. - Existing real producer tests cover commit/rollback, deletion with scoped hints in one outer transaction, secondary targets, re-opened tasks, valid/lenient history quietness, startup paging, failure retry, drain/resume and shutdown. The routine failure case now requires no signal when its cleanup has no linked decision. - The earlier dropped-hint and drain-fixture failures remain in [the behavior audit](doc/idle-decision-targets.md). Extra upstream embedded-only tests attempted on the parent could not initialize locally; CI must cover that environment. No new schema migration or table is added. The final review cleanup removes the combined internal `sweepExpired` method, cursor, flags and unused target-policy export. A repository/package/SDK audit found no remaining production caller or public exposure. Behavioral tests now call the dedicated workers; isolation tests prove TTL does not recover stale effects and effect recovery does not expire open decisions. HTTP and plugin SDK contracts are unchanged. The full 248-case selection passes after this cleanup and parent reconciliation; types/build pass with the updated Runner vendor. Current-head CI run 40602159553 (attempt 1) failed in the workspace CLI group: company import/export startup exceeded its 30-second health deadline, and the dense ZIP case exceeded its unchanged five-second timeout. The preceding UI group passed all 4,169 cases. Both CLI files and the transfer implementation are unchanged from the landed parent. The complete ZIP-transfer file passes all 17 cases locally on this exact head; the CLI startup E2E is skipped because the host's embedded PostgreSQL bootstrap fails. This does not prove Linux startup locally. The failure logs remain recorded; no timeout or assertion is changed. One bounded retry of the failed jobs is requested after the workflow completed. Its result and the current-head gates remain required before merge. Commands: server typecheck/build and Vitest for `decision-target-work`, `routines-service`, `decisions-service`, `decision-effect-recovery-work`, `decision-expiry-work`, `server-startup-feedback-export` and `scoped-deadline-work`. Local targeted acceptance does not establish deployed suspension or fleet savings. Fresh exact-head CI and Greptile/Ponytail remain merge gates. ## Risks - Deletion now holds the issue row lock while discovering dependent decisions and completing the existing removal transaction. This is needed to serialize new FK references. Provider calls are not placed inside that transaction. - Company deletion with decision-origin references still has the pre-existing restrictive FK limitation. Removing the futile notification does not implement a new company-data cleanup policy. - Supported writers share the application DB owner and maintain existing target links. Independent SQL writers, damaged links or multiple active processes need their own wake contract; startup still reconstructs all open targets. - Cloud sleep blockers and durable external alarms are unchanged. Worker quiet is not proof of deployed host/database suspension. ## 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
e9e84bb8a1
commit
8dfed893d8
15 files changed
+852
-76
No files matched your search
@@ -316,12 +316,15 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
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 });
|
||||
it("does not expire overdue open decisions during effect recovery", async () => {
|
||||
const row = await create();
|
||||
await db.update(decisions).set({ expiresAt: new Date(Date.now() - 1000) }).where(eq(decisions.id, row.id));
|
||||
expect(await service().recoverEffects()).toEqual({ resumed: 0 });
|
||||
expect(await service().get(row.id)).toMatchObject({ status: "open" });
|
||||
expect(await comments()).toHaveLength(0);
|
||||
expect(await service().sweepExpired()).toEqual({ expired: 0, resumed: 1 }); expect(await comments()).toHaveLength(1);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
expect(await service().sweepTtlExpiry()).toEqual({ expired: 1 });
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each(["missing_decider", "missing_option"] as const)("settles %s as durable failure and then becomes quiet", async reason => {
|
||||
|
||||
@@ -62,8 +62,10 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
async function start() {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
await registerDecisionContinuationWork(coordinator, service(), () => !draining).ready;
|
||||
await registerDecisionExpiryWork(coordinator, service(), () => !draining).ready;
|
||||
const continuation = service(), expiry = service();
|
||||
await registerDecisionContinuationWork(coordinator, continuation, () => !draining).ready;
|
||||
await registerDecisionExpiryWork(coordinator, expiry, () => !draining).ready;
|
||||
return { continuation, expiry };
|
||||
}
|
||||
async function quiet() {
|
||||
await vi.waitFor(async () => {
|
||||
@@ -218,12 +220,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
});
|
||||
|
||||
it("keeps a failed expiry retry through drain and preserves pending state", async () => {
|
||||
const row = await create(new Date(Date.now() + 30_000)); await start();
|
||||
const row = await create(new Date(Date.now() + 30_000)); const workers = await start();
|
||||
const sweeps = [vi.spyOn(workers.expiry, "sweepTtlExpiry"), vi.spyOn(workers.continuation, "sweepContinuations")];
|
||||
await db.execute(sql`create function decision_expiry_fail() returns trigger language plpgsql as $$ begin
|
||||
if NEW.status = 'expired' then raise exception 'expiry unavailable'; end if; return NEW; end $$`);
|
||||
await db.execute(sql`create trigger decision_expiry_fail before update on decisions for each row execute function decision_expiry_fail()`);
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); });
|
||||
draining = true; expect((await service().get(row.id))?.status).toBe("open");
|
||||
await Promise.allSettled(sweeps.flatMap(sweep => sweep.mock.results.filter(result => result.type === "return").map(result => result.value)));
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await db.execute(sql`drop trigger decision_expiry_fail on decisions`);
|
||||
@@ -247,12 +251,15 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
|
||||
expect(wake).not.toHaveBeenCalled(); expect(await service().nextExpiryAt()).toBeNull();
|
||||
});
|
||||
|
||||
it("keeps TTL expiry out of the aggregate sweep when its worker owns the duty", async () => {
|
||||
it("does not recover stale running effects during TTL expiry", async () => {
|
||||
const row = await create();
|
||||
await db.update(decisions).set({ expiresAt: new Date(Date.now() - 1000) }).where(eq(decisions.id, row.id));
|
||||
expect(await service().sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false })).toEqual({ expired: 0, resumed: 0 });
|
||||
expect((await service().get(row.id))?.status).toBe("open");
|
||||
expect(await service().sweepTtlExpiry()).toEqual({ expired: 1 }); expect(wake).toHaveBeenCalledTimes(1);
|
||||
await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes",
|
||||
decidedByUserId: null, updatedAt: new Date(Date.now() - 120_000), expiresAt: new Date(Date.now() - 1000) })
|
||||
.where(eq(decisions.id, row.id));
|
||||
expect(await service().sweepTtlExpiry()).toEqual({ expired: 0 });
|
||||
expect(await service().get(row.id)).toMatchObject({ status: "decided", executionStatus: "running" });
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
expect(await service().nextExpiryAt()).toBeNull();
|
||||
});
|
||||
|
||||
it("pages past a failed candidate and retries it without repeating completed expirations", async () => {
|
||||
|
||||
@@ -0,0 +1,366 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agents, authUsers, companies, companyMemberships, createDb, subscribeDatabaseWork, 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 { registerDecisionTargetWork } from "../services/decision-target-work.js";
|
||||
import { registerDecisionContinuationWork } from "../services/decision-continuation-work.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
|
||||
import { deleteCompany } from "../services/company-deletion.js";
|
||||
import { notifyDecisionTargetIssueRemoval } from "../services/decision-targets.js";
|
||||
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISION_TARGETS_TEST_DATABASE_URL");
|
||||
const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("decision target 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) temporary = await startEmbeddedPostgresTestDatabase("decision-targets-");
|
||||
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_target_expire_fail on decisions`);
|
||||
await db.execute(sql`drop function if exists decision_target_expire_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 = (staleness: "strict" | "lenient" = "strict") => 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, bodyMarkdown: "hello" },
|
||||
] }] });
|
||||
async function start() {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
|
||||
onError: error => errors.push(error) });
|
||||
await registerDecisionContinuationWork(coordinator, service(), () => !draining).ready;
|
||||
recovery = service();
|
||||
await registerDecisionTargetWork(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 expired(id: string, expiredReason = "target_gone") {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await service().get(id)).toMatchObject({ status: "expired", metadata: { continuationPending: false, expiredReason } });
|
||||
}, { timeout: 5000 });
|
||||
}
|
||||
async function failExpiry() {
|
||||
await db.execute(sql`create function decision_target_expire_fail() returns trigger language plpgsql as $$ begin
|
||||
if NEW.status = 'expired' then raise exception 'expiry unavailable'; end if;
|
||||
return NEW; end $$`);
|
||||
await db.execute(sql`create trigger decision_target_expire_fail before update on decisions for each row execute function decision_target_expire_fail()`);
|
||||
}
|
||||
|
||||
const change = (status: "done" | "cancelled") => issueService(db).update(targetId, { status, actorUserId: userId });
|
||||
|
||||
it("stays quiet with valid open decisions and wakes from the real terminal update", async () => {
|
||||
await start(); await quiet(); const row = await create(); await quiet();
|
||||
await change("done"); await expired(row.id, "target_completed"); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("finishes startup paging across submillisecond expiry values", async () => {
|
||||
const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions);
|
||||
await db.insert(decisions).values(Array.from({ length: 101 }, () => ({ ...seed, id: randomUUID() })));
|
||||
await db.execute(sql`update decisions set expires_at = date_trunc('milliseconds', now() + interval '1 day') + interval '456 microseconds'`);
|
||||
const owner = service();
|
||||
await owner.sweepInvalidTargets();
|
||||
expect(owner.hasDecisionTargetWork()).toBe(true);
|
||||
await owner.sweepInvalidTargets();
|
||||
expect(owner.hasDecisionTargetWork()).toBe(false);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("reaches a deleted target behind a full page of valid startup decisions", async () => {
|
||||
const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions);
|
||||
const invalidId = "ffffffff-ffff-4fff-afff-ffffffffffff";
|
||||
const valid = Array.from({ length: 100 }, (_, index) => ({ ...seed,
|
||||
id: `01000000-0000-4000-8000-${String(index).padStart(12, "0")}`,
|
||||
options: [{ id: "yes", label: "Yes", effects: [{ type: "comment_on_issue" as const, targetIssueId: originId, staleness: "strict" as const, bodyMarkdown: "hello" }] }],
|
||||
}));
|
||||
await db.insert(decisions).values([...valid, { ...seed, id: invalidId }]);
|
||||
await db.insert(decisionTargetIssues).values({ companyId, decisionId: invalidId, issueId: targetId });
|
||||
const owner = service();
|
||||
await owner.sweepInvalidTargets();
|
||||
expect(owner.hasDecisionTargetWork()).toBe(true);
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, (event, hints) => {
|
||||
if (event === "settled") owner.decisionTargetsChanged(hints);
|
||||
});
|
||||
try {
|
||||
await issueService(db).remove(targetId);
|
||||
expect(await owner.sweepInvalidTargets()).toEqual({ expired: 1 });
|
||||
expect((await owner.get(invalidId))?.metadata).toMatchObject({ expiredReason: "target_gone" });
|
||||
expect(owner.hasDecisionTargetWork()).toBe(false);
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
} finally { unsubscribe(); }
|
||||
});
|
||||
|
||||
it("fences new target references between deletion discovery and commit", async () => {
|
||||
let entered!: () => void, release!: () => void;
|
||||
const discovered = new Promise<void>(resolve => { entered = resolve; });
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const removing = db.transaction(async tx => {
|
||||
await notifyDecisionTargetIssueRemoval(tx, targetId);
|
||||
entered(); await gate;
|
||||
await tx.delete(issues).where(eq(issues.id, targetId));
|
||||
});
|
||||
let proposing: Promise<{ accepted: boolean }> | undefined;
|
||||
try {
|
||||
await discovered;
|
||||
proposing = create().then(() => ({ accepted: true }), () => ({ accepted: false }));
|
||||
await vi.waitFor(async () => {
|
||||
const blocked = await db.execute(sql`select pid from pg_stat_activity where datname = current_database()
|
||||
and pid <> pg_backend_pid() and wait_event_type = 'Lock' and query like '%decision_target_issues%'`);
|
||||
expect(blocked).toHaveLength(1);
|
||||
});
|
||||
release(); await removing;
|
||||
await expect(proposing).resolves.toEqual({ accepted: false });
|
||||
expect(await db.select().from(decisions)).toEqual([]);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
} finally { release(); await Promise.allSettled([removing, ...(proposing ? [proposing] : [])]); }
|
||||
});
|
||||
|
||||
it("does not signal target work for removal with no linked decisions", async () => {
|
||||
await create();
|
||||
const unrelatedId = randomUUID();
|
||||
await db.insert(issues).values({ id: unrelatedId, companyId, title: "Unrelated", status: "todo", responsibleUserId: userId });
|
||||
const settled = vi.fn();
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, event => { if (event === "settled") settled(); });
|
||||
try { await issueService(db).remove(unrelatedId); expect(settled).not.toHaveBeenCalled(); }
|
||||
finally { unsubscribe(); }
|
||||
});
|
||||
|
||||
it("does not launch global target work after deleting an empty company", async () => {
|
||||
const emptyCompanyId = randomUUID();
|
||||
await db.insert(companies).values({ id: emptyCompanyId, name: "Empty", issuePrefix: `E${emptyCompanyId.slice(0, 6)}` });
|
||||
const home = await mkdtemp(join(tmpdir(), "decision-target-company-"));
|
||||
vi.stubEnv("PAPERCLIP_HOME", home);
|
||||
const settled = vi.fn();
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, event => { if (event === "settled") settled(); });
|
||||
try {
|
||||
expect(await deleteCompany(db, emptyCompanyId)).toMatchObject({ id: emptyCompanyId });
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
} finally { unsubscribe(); await rm(home, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
it("keeps lenient and intentional already-done proposals open and quiet", async () => {
|
||||
const lenient = await create("lenient"); await change("done"); const intentional = await create();
|
||||
await start(); await quiet();
|
||||
expect((await service().get(lenient.id))?.status).toBe("open");
|
||||
expect((await service().get(intentional.id))?.status).toBe("open"); expect(wake).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("expires a decision created after its target was cancelled", async () => {
|
||||
await change("cancelled"); await start(); await quiet(); const row = await create();
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("reacts to the real removal producer after target links cascade away", async () => {
|
||||
const row = await create(); await start(); await quiet(); await issueService(db).remove(targetId);
|
||||
expect(await db.select().from(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, row.id))).toHaveLength(0);
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each(["cancelled", "deleted"] as const)("reconstructs %s target invalidation at startup", async action => {
|
||||
const row = await create();
|
||||
if (action === "deleted") await db.delete(issues).where(eq(issues.id, targetId));
|
||||
else await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, targetId));
|
||||
await start(); await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("includes a strict secondary effect reference in notification scope", async () => {
|
||||
const blockerId = randomUUID();
|
||||
await db.insert(issues).values({ id: blockerId, companyId, title: "Blocker", status: "todo", responsibleUserId: userId });
|
||||
const row = await service().create({ companyId, actor: actor(), agentId, runId, title: "Follow up?", body: "Body",
|
||||
continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [
|
||||
{ type: "create_issue", targetIssueId: targetId, staleness: "strict", draft: { title: "Next", blockedByIssueIds: [blockerId] } },
|
||||
] }] });
|
||||
await start(); await quiet(); await issueService(db).update(blockerId, { status: "cancelled", actorUserId: userId });
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("waits for the last strict target to complete", async () => {
|
||||
const blockerId = randomUUID();
|
||||
await db.insert(issues).values({ id: blockerId, companyId, title: "Blocker", status: "todo", responsibleUserId: userId });
|
||||
const row = await service().create({ companyId, actor: actor(), agentId, runId, title: "Follow up?", body: "Body",
|
||||
continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [
|
||||
{ type: "create_issue", targetIssueId: targetId, staleness: "strict", draft: { title: "Next", blockedByIssueIds: [blockerId] } },
|
||||
] }] });
|
||||
await start(); await quiet(); await change("done"); await quiet();
|
||||
expect((await service().get(row.id))?.status).toBe("open");
|
||||
await issueService(db).update(blockerId, { status: "done", actorUserId: userId });
|
||||
await expired(row.id, "target_completed"); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("waits for the actual issue update's outer transaction and ignores rollback", async () => {
|
||||
const row = await create(); await start(); await quiet();
|
||||
await expect(db.transaction(async tx => {
|
||||
await issueService(db).update(targetId, { status: "cancelled", actorUserId: userId }, tx);
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(wake).not.toHaveBeenCalled(); throw new Error("rollback");
|
||||
})).rejects.toThrow("rollback");
|
||||
await quiet(); expect((await service().get(row.id))?.status).toBe("open");
|
||||
await db.transaction(async tx => {
|
||||
await issueService(db).update(targetId, { status: "cancelled", actorUserId: userId }, tx);
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(wake).not.toHaveBeenCalled();
|
||||
});
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each([true, false])("preserves a deletion mixed with a scoped hint in one transaction, deleteFirst=%s", async deleteFirst => {
|
||||
const row = await create(); const otherId = randomUUID();
|
||||
await db.insert(issues).values({ id: otherId, companyId, title: "Unrelated", status: "todo", responsibleUserId: userId });
|
||||
await start(); await quiet();
|
||||
await db.transaction(async tx => {
|
||||
const scoped = issueService(tx as unknown as typeof db);
|
||||
if (deleteFirst) await scoped.remove(targetId);
|
||||
await scoped.update(otherId, { status: "done", actorUserId: userId });
|
||||
if (!deleteFirst) await scoped.remove(targetId);
|
||||
});
|
||||
expect(await db.select().from(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, row.id))).toHaveLength(0);
|
||||
await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("retains a first target change through drain without passive SQL", async () => {
|
||||
const row = await create(); await start(); await quiet(); draining = true; await change("cancelled");
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("checks current state if a cancelled target reopened before reconciliation", async () => {
|
||||
const row = await create(); await start(); await quiet(); draining = true; await change("cancelled");
|
||||
await issueService(db).update(targetId, { status: "todo", actorUserId: userId });
|
||||
draining = false; await quiet(); expect((await service().get(row.id))?.status).toBe("open"); expect(wake).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("retains failed expiry through drain and retries after the database recovers", async () => {
|
||||
const row = await create(); await start(); await quiet(); await failExpiry(); await change("cancelled");
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); });
|
||||
draining = true;
|
||||
const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
|
||||
await db.execute(sql`drop trigger decision_target_expire_fail on decisions`);
|
||||
draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
for (const error of errors) expect(error).toMatchObject({ cause: { message: "expiry unavailable" } }); errors.length = 0;
|
||||
});
|
||||
|
||||
it("does not inspect unrelated open decisions on an ordinary scoped change", async () => {
|
||||
const unrelated = await create();
|
||||
const otherId = randomUUID(); await db.insert(issues).values({ id: otherId, companyId, title: "Other", status: "todo", responsibleUserId: userId });
|
||||
const affected = await service().create({ companyId, actor: actor(), agentId, runId, title: "Other?", body: "Body",
|
||||
continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [
|
||||
{ type: "comment_on_issue", targetIssueId: otherId, staleness: "strict", bodyMarkdown: "hello" },
|
||||
] }] });
|
||||
await start(); await quiet();
|
||||
await db.update(decisions).set({ options: sql`'null'::jsonb` }).where(eq(decisions.id, unrelated.id));
|
||||
await issueService(db).update(otherId, { status: "cancelled", actorUserId: userId });
|
||||
await expired(affected.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("bounds startup pages and continues after more than one page of valid decisions", async () => {
|
||||
const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions);
|
||||
await db.insert(decisions).values(Array.from({ length: 101 }, () => ({ ...seed, id: randomUUID() })));
|
||||
await start(); expect(recovery.hasDecisionTargetWork()).toBe(true);
|
||||
await quiet(); expect(await db.select().from(decisions).where(eq(decisions.status, "open"))).toHaveLength(101);
|
||||
expect(wake).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("continues past a failing first page and retains its failed candidate", async () => {
|
||||
const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions);
|
||||
const ids = Array.from({ length: 101 }, (_, i) => `12000000-0000-4000-8000-${String(i).padStart(12, "0")}`);
|
||||
await db.insert(decisions).values(ids.map(id => ({ ...seed, id })));
|
||||
await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, targetId));
|
||||
await db.execute(sql`create function decision_target_expire_fail() returns trigger language plpgsql as $$ begin
|
||||
if NEW.id = '12000000-0000-4000-8000-000000000000' and NEW.status = 'expired' then raise exception 'expiry unavailable'; end if;
|
||||
return NEW; end $$`);
|
||||
await db.execute(sql`create trigger decision_target_expire_fail before update on decisions for each row execute function decision_target_expire_fail()`);
|
||||
await start(); expect(wake).toHaveBeenCalledTimes(99); expect((await service().get(ids.at(-1)!))?.status).toBe("open");
|
||||
await expired(ids.at(-1)!); expect((await service().get(ids[0]))?.status).toBe("open");
|
||||
await db.execute(sql`drop trigger decision_target_expire_fail on decisions`);
|
||||
await expired(ids[0]); await quiet(); expect(wake).toHaveBeenCalledTimes(101);
|
||||
for (const error of errors) expect(error).toMatchObject({ cause: { message: "expiry unavailable" } }); errors.length = 0;
|
||||
});
|
||||
|
||||
it("retains a change to the same issue scope while its earlier pass is still running", async () => {
|
||||
await change("done"); const intentional = await create();
|
||||
await db.update(decisions).set({ expiresAt: new Date(Date.now() + 24 * 60 * 60_000) }).where(eq(decisions.id, intentional.id));
|
||||
await issueService(db).update(targetId, { status: "todo", actorUserId: userId });
|
||||
const ordinary = await create(); await start(); await quiet();
|
||||
wake.mockImplementationOnce(async () => { await change("cancelled"); });
|
||||
await change("done"); await expired(ordinary.id, "target_completed"); await expired(intentional.id); await quiet();
|
||||
expect(wake).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("learns a decision created during invalidation delivery", async () => {
|
||||
const first = await create(); await change("cancelled"); let nextId = "";
|
||||
wake.mockImplementationOnce(async () => { nextId = (await create()).id; });
|
||||
await start(); await expired(first.id); await vi.waitFor(() => expect(nextId).not.toBe(""));
|
||||
await expired(nextId); await quiet(); expect(wake).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("finishes admitted invalidation and leaves later rows for restart during shutdown", async () => {
|
||||
const first = await create(), second = await create(); await change("cancelled");
|
||||
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);
|
||||
const states = await Promise.all([service().get(first.id), service().get(second.id)]);
|
||||
expect(states.map(row => row!.status).sort()).toEqual(["expired", "open"]);
|
||||
await start(); await quiet(); expect(wake).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
});
|
||||
@@ -225,12 +225,12 @@ describePg("decisionService", () => {
|
||||
...original(source, options), archiveReviewedManifest: async () => { throw new Error("archive_failure"); },
|
||||
})) : undefined;
|
||||
try {
|
||||
expect(await service().sweepExpired()).toMatchObject({ resumed: 1 });
|
||||
expect(await service().recoverEffects()).toMatchObject({ resumed: 1 });
|
||||
const saved = await service().get(row.id);
|
||||
expect(saved).toMatchObject({ executionStatus: "failed", metadata: { archiveProposalError: failure, continuationPending: false } });
|
||||
expect(saved!.metadata.continuationDeliveredAt).toEqual(expect.any(String));
|
||||
expect(wakes).toHaveLength(1);
|
||||
await service().sweepExpired();
|
||||
await service().recoverEffects();
|
||||
expect(wakes).toHaveLength(1);
|
||||
} finally { factory?.mockRestore(); }
|
||||
});
|
||||
@@ -512,7 +512,7 @@ describePg("decisionService", () => {
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("recovers stale running decisions from the bounded server sweep", async () => {
|
||||
it("recovers stale running decisions from the effect worker", async () => {
|
||||
process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS = "0";
|
||||
const created = await createCommentDecision();
|
||||
await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes", decidedByUserId,
|
||||
@@ -520,7 +520,7 @@ describePg("decisionService", () => {
|
||||
await db.insert(decisionEffectExecutions).values({ decisionId: created.id, effectIndex: 0, effectType: "comment_on_issue",
|
||||
targetIssueId, status: "claimed" });
|
||||
|
||||
expect(await service().sweepExpired()).toEqual({ expired: 0, resumed: 1 });
|
||||
expect(await service().recoverEffects()).toEqual({ resumed: 1 });
|
||||
expect((await service().get(created.id))?.executionStatus).toBe("succeeded");
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1);
|
||||
expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]);
|
||||
@@ -538,7 +538,7 @@ describePg("decisionService", () => {
|
||||
expect((await service().get(created.id))?.metadata).toMatchObject({ continuationPending: true });
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1);
|
||||
|
||||
await service().sweepExpired();
|
||||
await service().sweepContinuations();
|
||||
|
||||
expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]);
|
||||
expect((await service().get(created.id))?.metadata).toMatchObject({ continuationPending: false });
|
||||
@@ -613,8 +613,8 @@ describePg("decisionService", () => {
|
||||
const second = await createCommentDecision("lenient", { idempotencyKey: "batch-2", expiresAt: nearFutureExpiry() });
|
||||
await expireDecisionNow(first.id);
|
||||
await expireDecisionNow(second.id);
|
||||
expect((await service().sweepExpired()).expired).toBe(1);
|
||||
expect((await service().sweepExpired()).expired).toBe(1);
|
||||
expect((await service().sweepTtlExpiry()).expired).toBe(1);
|
||||
expect((await service().sweepTtlExpiry()).expired).toBe(1);
|
||||
});
|
||||
|
||||
it("falls back to the default sweep batch size for invalid configuration", async () => {
|
||||
@@ -624,7 +624,7 @@ describePg("decisionService", () => {
|
||||
await expireDecisionNow(first.id);
|
||||
await expireDecisionNow(second.id);
|
||||
|
||||
await expect(service().sweepExpired()).resolves.toMatchObject({ expired: 2 });
|
||||
await expect(service().sweepTtlExpiry()).resolves.toMatchObject({ expired: 2 });
|
||||
});
|
||||
|
||||
it("expires TTL and target-gone decisions and wakes the origin agent", async () => {
|
||||
@@ -633,7 +633,8 @@ describePg("decisionService", () => {
|
||||
await db.update(decisions).set({ expiresAt: new Date(0) }).where(eq(decisions.id, ttl.id));
|
||||
await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, targetIssueId));
|
||||
await expireDecisionNow(ttl.id);
|
||||
expect((await service().sweepExpired()).expired).toBe(2);
|
||||
expect((await service().sweepTtlExpiry()).expired).toBe(1);
|
||||
expect((await service().sweepInvalidTargets()).expired).toBe(1);
|
||||
const rows = await db.select().from(decisions);
|
||||
expect(rows.find((row) => row.id === ttl.id)?.metadata).toMatchObject({ expiredReason: "ttl" });
|
||||
expect(rows.find((row) => row.id === gone.id)?.metadata).toMatchObject({ expiredReason: "target_gone" });
|
||||
@@ -645,7 +646,7 @@ describePg("decisionService", () => {
|
||||
const lenient = await createCommentDecision("lenient", { idempotencyKey: "lenient-survives" });
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, targetIssueId));
|
||||
|
||||
expect((await service().sweepExpired()).expired).toBe(1);
|
||||
expect((await service().sweepInvalidTargets()).expired).toBe(1);
|
||||
|
||||
expect((await service().get(completed.id))?.metadata).toMatchObject({ expiredReason: "target_completed" });
|
||||
expect((await service().get(lenient.id))?.status).toBe("open");
|
||||
@@ -656,7 +657,7 @@ describePg("decisionService", () => {
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, targetIssueId));
|
||||
const reopen = await createCommentDecision("strict", { idempotencyKey: "already-done" });
|
||||
|
||||
expect((await service().sweepExpired()).expired).toBe(0);
|
||||
expect((await service().sweepInvalidTargets()).expired).toBe(0);
|
||||
expect((await service().get(reopen.id))?.status).toBe("open");
|
||||
});
|
||||
|
||||
@@ -691,7 +692,7 @@ describePg("decisionService", () => {
|
||||
});
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, blockerId));
|
||||
|
||||
expect((await service().sweepExpired()).expired).toBe(1);
|
||||
expect((await service().sweepInvalidTargets()).expired).toBe(1);
|
||||
expect((await service().get(created.id))?.metadata).toMatchObject({ expiredReason: "target_completed" });
|
||||
});
|
||||
|
||||
@@ -716,7 +717,7 @@ describePg("decisionService", () => {
|
||||
await service().decide({ id: acceptedAgain.id, optionId: "assign", decidedByUserId, userActor: boardActor() });
|
||||
await service().dismiss(rejected.id, decidedByUserId, boardActor(), "Not this time");
|
||||
await expireDecisionNow(stale.id);
|
||||
await service().sweepExpired();
|
||||
await service().sweepTtlExpiry();
|
||||
|
||||
const stats = await service().stats(companyId, { originAgentId: agentId });
|
||||
expect(stats.filters).toEqual({ originAgentId: agentId, since: null });
|
||||
@@ -825,7 +826,7 @@ describePg("decisionService", () => {
|
||||
if (action === "decide") await svc.decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() });
|
||||
if (action === "cancel") await svc.cancel(row.id, { actorType: "agent", actorId: agentId });
|
||||
if (action === "dismiss") await svc.dismiss(row.id, decidedByUserId, boardActor());
|
||||
if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepExpired(new Date(), { includeContinuations: false }); }
|
||||
if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepTtlExpiry(); }
|
||||
await quiet(); await delivered(row.id);
|
||||
expect(wake).toHaveBeenCalledTimes(1);
|
||||
expect(wake.mock.calls[0][0]).toMatchObject({ decisionId: row.id, companyId, agentId, issueId: originIssueId });
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { subscribeDatabaseWork } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
|
||||
import { createHmac, randomUUID } from "node:crypto";
|
||||
@@ -38,7 +39,8 @@ import * as providerRegistry from "../secrets/provider-registry.ts";
|
||||
import { routineService } from "../services/routines.ts";
|
||||
import { secretService } from "../services/secrets.ts";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ROUTINES_TEST_DATABASE_URL");
|
||||
const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
const originalSecretsProviderEnv = process.env.PAPERCLIP_SECRETS_PROVIDER;
|
||||
const originalPaperclipApiUrlEnv = process.env.PAPERCLIP_API_URL;
|
||||
@@ -55,8 +57,8 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
||||
|
||||
beforeAll(async () => {
|
||||
process.env.PAPERCLIP_API_URL = "http://localhost:3100";
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-routines-service-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-routines-service-");
|
||||
db = createDb(externalUrl ?? tempDb!.connectionString);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
@@ -2167,7 +2169,10 @@ describeEmbeddedPostgres("routine service live-execution coalescing", () => {
|
||||
},
|
||||
});
|
||||
|
||||
const run = await svc.runRoutine(routine.id, { source: "manual" });
|
||||
const events: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, event => { events.push(event); });
|
||||
const run = await svc.runRoutine(routine.id, { source: "manual" }).finally(unsubscribe);
|
||||
expect(events).toEqual([]);
|
||||
|
||||
expect(run.status).toBe("failed");
|
||||
expect(run.failureReason).toContain("queue unavailable");
|
||||
|
||||
@@ -348,7 +348,9 @@ vi.mock("../services/index.js", () => ({
|
||||
list: vi.fn(async () => ({ items: [], nextCursor: null })),
|
||||
})),
|
||||
decisionService: vi.fn(() => ({
|
||||
sweepExpired: vi.fn(async () => ({ expired: 0 })),
|
||||
sweepInvalidTargets: vi.fn(async () => ({ expired: 0 })),
|
||||
decisionTargetsChanged: vi.fn(),
|
||||
hasDecisionTargetWork: vi.fn(() => false),
|
||||
recoverEffects: vi.fn(async () => ({ resumed: 0 })),
|
||||
nextEffectRecoveryAt: vi.fn(async () => null),
|
||||
sweepTtlExpiry: vi.fn(async () => ({ expired: 0 })),
|
||||
@@ -926,7 +928,8 @@ describe("startServer feedback export wiring", () => {
|
||||
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, includeEffects: false });
|
||||
expect(decisions.sweepInvalidTargets).toHaveBeenCalledTimes(1);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionTargets)).toBe(true);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
@@ -966,6 +969,7 @@ describe("startServer feedback export wiring", () => {
|
||||
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);
|
||||
expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionTargets)).toBe(false);
|
||||
} finally {
|
||||
setIntervalSpy.mockRestore();
|
||||
}
|
||||
|
||||
+3
-4
@@ -1,3 +1,4 @@
|
||||
import { registerDecisionTargetWork } from "./services/decision-target-work.js";
|
||||
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";
|
||||
@@ -1845,7 +1846,8 @@ async function startServerWithDatabaseTeardown(
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
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 }));
|
||||
await step("decision_target_worker", () => registerDecisionTargetWork(deliveryWork, decisionExecutor,
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
|
||||
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 },
|
||||
@@ -1866,9 +1868,6 @@ async function startServerWithDatabaseTeardown(
|
||||
trackHeartbeatSchedulerWork((async () => {
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
scheduleRoutineStatusReconcile();
|
||||
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) => {
|
||||
logger.error({ err }, "decision retention sweep failed");
|
||||
}));
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import type { decisionService } from "./decisions.js";
|
||||
|
||||
export function registerDecisionTargetWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
service: Pick<ReturnType<typeof decisionService>, "sweepInvalidTargets" | "decisionTargetsChanged" | "hasDecisionTargetWork">,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
return coordinator.register(DELIVERY_QUEUES.decisionTargets, {
|
||||
retryMs: 30_000,
|
||||
pausePassiveMaintenance: true,
|
||||
onSettled: service.decisionTargetsChanged,
|
||||
run: signal => canRun() ? service.sweepInvalidTargets(new Date(), signal) : Promise.resolve(),
|
||||
hasPending: async () => service.hasDecisionTargetWork(),
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
import { and, asc, eq, getTableColumns, inArray, sql } from "drizzle-orm";
|
||||
import { decisions, decisionTargetIssues, issues, type Db } from "@paperclipai/db";
|
||||
import { decisionEffectTargetIssueIds } from "@paperclipai/shared";
|
||||
import type { decisionExpiryService } from "./decision-expiry.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
type Decision = typeof decisions.$inferSelect;
|
||||
type Cursor = { expiresAt: string; id: string };
|
||||
type Scope = { kind: "all" | "issue" | "decision"; id?: string; cursor?: Cursor };
|
||||
const open = sql`${decisions.status} = 'open'`;
|
||||
|
||||
export function notifyDecisionTargetWork(transaction: object, hint: `issue:${string}` | `decision:${string}`) {
|
||||
return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionTargets, hint);
|
||||
}
|
||||
|
||||
export async function notifyDecisionTargetIssueRemoval(transaction: Pick<Db, "select" | "selectDistinct">, issueId: string) {
|
||||
const [issue] = await transaction.select({ companyId: issues.companyId }).from(issues)
|
||||
.where(eq(issues.id, issueId)).for("update");
|
||||
if (!issue) return;
|
||||
const linked = await transaction.selectDistinct({ decisionId: decisionTargetIssues.decisionId }).from(decisionTargetIssues)
|
||||
.innerJoin(decisions, eq(decisions.id, decisionTargetIssues.decisionId))
|
||||
.where(and(eq(decisionTargetIssues.companyId, issue.companyId), eq(decisionTargetIssues.issueId, issueId), eq(decisions.companyId, issue.companyId), open));
|
||||
for (const row of linked) await notifyDecisionTargetWork(transaction, `decision:${row.decisionId}`);
|
||||
}
|
||||
|
||||
export function decisionTargetService(db: Db, expire: ReturnType<typeof decisionExpiryService>["expire"]) {
|
||||
const scopes = new Map<string, Scope>([["all", { kind: "all" }]]);
|
||||
|
||||
function changed(hints?: ReadonlySet<string>) {
|
||||
if (!hints?.size) { scopes.set("all", { kind: "all" }); return; }
|
||||
for (const hint of hints) {
|
||||
const [kind, id] = hint.split(":");
|
||||
if ((kind === "issue" || kind === "decision") && id) scopes.set(hint, { kind, id });
|
||||
else scopes.set("all", { kind: "all" });
|
||||
}
|
||||
}
|
||||
|
||||
async function reason(decision: Decision): Promise<"target_gone" | "target_completed" | null> {
|
||||
const strictIds = new Set(decision.options.flatMap(option => option.effects
|
||||
.filter(effect => effect.staleness === "strict").flatMap(decisionEffectTargetIssueIds)));
|
||||
if (strictIds.size === 0) return null;
|
||||
const targets = await db.select({ id: issues.id, status: issues.status }).from(issues)
|
||||
.where(and(eq(issues.companyId, decision.companyId), inArray(issues.id, [...strictIds])));
|
||||
if (targets.length !== strictIds.size || targets.some(target => target.status === "cancelled")) return "target_gone";
|
||||
const snapshots = decision.targetSnapshots as Record<string, { status?: string }>;
|
||||
return targets.every(target => target.status === "done") && targets.some(target => snapshots[target.id]?.status !== "done")
|
||||
? "target_completed" : null;
|
||||
}
|
||||
|
||||
async function sweep(now = new Date(), signal?: AbortSignal) {
|
||||
let expired = 0, inspected = 0;
|
||||
const errors: unknown[] = [];
|
||||
for (const [key, scope] of [...scopes]) {
|
||||
if (signal?.aborted || inspected >= 100) break;
|
||||
if (scopes.get(key) !== scope) continue;
|
||||
scopes.delete(key);
|
||||
const limit = 100 - inspected;
|
||||
try {
|
||||
const rows = await db.select({ ...getTableColumns(decisions), cursorExpiresAt: sql<string>`${decisions.expiresAt}::text` }).from(decisions).where(and(open,
|
||||
scope.kind === "decision" ? eq(decisions.id, scope.id!) : undefined,
|
||||
scope.kind === "issue" ? inArray(decisions.id, db.select({ id: decisionTargetIssues.decisionId })
|
||||
.from(decisionTargetIssues).where(eq(decisionTargetIssues.issueId, scope.id!))) : undefined,
|
||||
scope.cursor ? sql`(${decisions.expiresAt}, ${decisions.id}) > (${scope.cursor.expiresAt}::timestamptz, ${scope.cursor.id}::uuid)` : undefined,
|
||||
)).orderBy(asc(decisions.expiresAt), asc(decisions.id)).limit(limit);
|
||||
inspected += Math.max(1, rows.length);
|
||||
for (const row of rows) {
|
||||
if (signal?.aborted) {
|
||||
if (!scopes.has(key)) scopes.set(key, scope);
|
||||
return { expired };
|
||||
}
|
||||
try {
|
||||
const invalid = await reason(row);
|
||||
if (invalid) expired += await expire(row, invalid, now);
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
const retry = `decision:${row.id}`;
|
||||
if (!scopes.has(retry)) scopes.set(retry, { kind: "decision", id: row.id });
|
||||
}
|
||||
scope.cursor = { expiresAt: row.cursorExpiresAt, id: row.id };
|
||||
}
|
||||
if (rows.length === limit && !scopes.has(key)) scopes.set(key, scope);
|
||||
} catch (error) {
|
||||
errors.push(error); inspected++;
|
||||
if (!scopes.has(key)) scopes.set(key, scope);
|
||||
}
|
||||
}
|
||||
if (errors.length === 1) throw errors[0];
|
||||
if (errors.length) throw new AggregateError(errors, "Decision target reconciliation failed");
|
||||
return { expired };
|
||||
}
|
||||
|
||||
return { sweep, changed, hasPending: () => scopes.size > 0 };
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { decisionTargetService, notifyDecisionTargetWork } from "./decision-targets.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, asc, count, desc, eq, gt, gte, inArray, sql } from "drizzle-orm";
|
||||
import { and, asc, count, desc, eq, 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,7 +16,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 { decisionEffectRecoveryService } from "./decision-effect-recovery.js";
|
||||
import { decisionExpiryService } from "./decision-expiry.js";
|
||||
import { notifyDecisionCreated, writeDecisionChange } from "./decision-work-notifications.js";
|
||||
|
||||
@@ -128,13 +129,13 @@ 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 targets = decisionTargetService(db, expiry.expire);
|
||||
const effectRecovery = decisionEffectRecoveryService(db, recoverDecisionEffects);
|
||||
const scope = databaseWorkScopeKey(db);
|
||||
let flights = effectFlights.get(scope);
|
||||
if (!flights) effectFlights.set(scope, flights = new Map());
|
||||
const executing = flights;
|
||||
const authz = authorizationService(db);
|
||||
let targetSweepCursor: string | null = null;
|
||||
type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null;
|
||||
ruleKey?: string | null; title: string; body: string; options: DecisionOption[]; inputs?: DecisionInput[] | null; expiresAt?: Date | null;
|
||||
idempotencyKey?: string | null; continuationPolicy?: "none" | "wake_origin_agent"; metadata?: Record<string, unknown> };
|
||||
@@ -239,6 +240,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
};
|
||||
const id = randomUUID();
|
||||
await notifyDecisionCreated(dbOrTx);
|
||||
await notifyDecisionTargetWork(dbOrTx, `decision:${id}`);
|
||||
const [created] = await dbOrTx.insert(decisions).values({ id, companyId: input.companyId, bundleId: input.bundleId ?? null,
|
||||
originAgentId: input.agentId, originIssueId: provenance.issueId, originRunId: input.runId, ruleKey: input.ruleKey ?? null,
|
||||
title: input.title, body: input.body, options: input.options, inputs: input.inputs ?? null, expiresAt,
|
||||
@@ -760,40 +762,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
|
||||
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;
|
||||
const targetRows = remaining > 0
|
||||
? await db.select().from(decisions)
|
||||
.where(and(eq(decisions.status, "open"), ...(targetSweepCursor ? [gt(decisions.id, targetSweepCursor)] : [])))
|
||||
.orderBy(asc(decisions.id)).limit(remaining)
|
||||
: [];
|
||||
targetSweepCursor = targetRows.length === remaining && targetRows.length > 0 ? targetRows[targetRows.length - 1]!.id : null;
|
||||
const rows = [...new Map([...ttlRows, ...targetRows].map((row) => [row.id, row])).values()]; let expired = 0;
|
||||
for (const decision of rows) { const strictTargetIds = new Set(decision.options.flatMap((option) => option.effects.filter((effect) => effect.staleness === "strict").flatMap(decisionEffectTargetIssueIds)));
|
||||
const targets = strictTargetIds.size > 0
|
||||
? await db.select({ id: issues.id, status: issues.status }).from(issues).where(and(eq(issues.companyId, decision.companyId), inArray(issues.id, [...strictTargetIds])))
|
||||
: [];
|
||||
const targetGone = targets.length !== strictTargetIds.size || targets.some((target) => target.status === "cancelled");
|
||||
// A decision whose strict targets all completed after it was proposed is moot:
|
||||
// the strict guard would skip its effects anyway, so retire it instead of
|
||||
// leaving it pending until TTL. Targets that were already done at proposal
|
||||
// time don't count — those decisions intentionally act on a done issue.
|
||||
const snapshots = decision.targetSnapshots as Record<string, Snapshot>;
|
||||
const targetsCompleted = !targetGone && strictTargetIds.size > 0 &&
|
||||
targets.every((target) => target.status === "done") &&
|
||||
targets.some((target) => snapshots[target.id]?.status !== "done");
|
||||
if (!targetGone && !targetsCompleted && (options.includeExpiry === false || decision.expiresAt > now)) continue;
|
||||
const reason = targetGone ? "target_gone" : targetsCompleted ? "target_completed" : "ttl";
|
||||
expired += await expiry.expire(decision, reason, now);
|
||||
}
|
||||
return { expired, resumed };
|
||||
}
|
||||
|
||||
return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep,
|
||||
return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepContinuations: continuations.sweep,
|
||||
sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next,
|
||||
recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next };
|
||||
recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next,
|
||||
sweepInvalidTargets: targets.sweep, decisionTargetsChanged: targets.changed, hasDecisionTargetWork: targets.hasPending };
|
||||
}
|
||||
@@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = {
|
||||
executionStatus: "execution-status",
|
||||
executionDisposition: "execution-disposition",
|
||||
environmentCleanup: "environment-cleanup",
|
||||
decisionTargets: "decision-targets",
|
||||
decisionEffectRecovery: "decision-effect-recovery",
|
||||
decisionExpiry: "decision-expiry",
|
||||
decisionContinuation: "decision-continuation",
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import { notifyDecisionTargetWork, notifyDecisionTargetIssueRemoval } from "./decision-targets.js";
|
||||
import { assertOnboardingFirstTaskCreationAllowed } from "./onboarding-first-task-admission.js";
|
||||
|
||||
import { resolveConnectionIntentContinuationSource, resolveItemVerdictContinuationWake } from "./chat-interaction-continuation.js";
|
||||
import { insertIssueComment } from "./issue-comment-writes.js";
|
||||
import { measureRunTransaction } from "./execution-transaction-timing.js";
|
||||
@@ -11384,6 +11386,9 @@ export function issueService(db: Db) {
|
||||
(patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined))) {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
|
||||
}
|
||||
if (patch.status === "done" || patch.status === "cancelled") {
|
||||
await notifyDecisionTargetWork(tx, `issue:${id}`);
|
||||
}
|
||||
const updated = await tx
|
||||
.update(issues)
|
||||
.set(patch)
|
||||
@@ -11791,6 +11796,7 @@ export function issueService(db: Db) {
|
||||
try {
|
||||
await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
|
||||
await requestRestorePointDeletion(tx, { sourceIssueId: id });
|
||||
await notifyDecisionTargetIssueRemoval(tx, id);
|
||||
removedIssue = await tx
|
||||
.delete(issues)
|
||||
.where(eq(issues.id, id))
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyDecisionTargetIssueRemoval } from "./decision-targets.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { verifyAppWebhook } from "./app-webhook.js";
|
||||
import crypto from "node:crypto";
|
||||
@@ -1964,6 +1965,7 @@ export function routineService(
|
||||
return updated ?? createdRun;
|
||||
} catch (error) {
|
||||
if (createdIssue) {
|
||||
await notifyDecisionTargetIssueRemoval(txDb, createdIssue.id);
|
||||
await txDb.delete(issues).where(eq(issues.id, createdIssue.id));
|
||||
}
|
||||
const failureReason = error instanceof Error ? error.message : String(error);
|
||||
|
||||
Reference in new issue
Block a user