diff --git a/doc/idle-decision-targets.md b/doc/idle-decision-targets.md new file mode 100644 index 00000000000..5883ed0f9b4 --- /dev/null +++ b/doc/idle-decision-targets.md @@ -0,0 +1,109 @@ +# Idle decision target invalidation + +Open decisions no longer require a periodic target scan. The target worker +reconciles committed issue changes and decision creation. Startup reconstructs +open decisions in bounded pages. After reconciliation, valid open decisions and +completed history cause no recurring SQL. TTL expiry, continuation delivery, +interrupted effects, and retention remain separate responsibilities. + +## Domain policy + +Only strict effect references participate. The existing shared +`decisionEffectTargetIssueIds` helper includes primary and secondary references. +A missing or cancelled target expires a decision with `target_gone`. If all +strict targets are done and at least one was not done in the proposal snapshot, +it expires with `target_completed`. Lenient decisions and proposals intentionally +acting on already-done targets remain open. A target that reopened before +reconciliation is checked in its current state; the notification is not authority +to expire it. The existing open-status compare-and-set arbitrates expiry against +other decision transitions. + +`decision-targets.ts` owns this policy and scoped reconciliation. It delegates +expiry, activity, and continuation to the existing expiry owner. The public +combined `sweepExpired` keeps its default behavior for compatibility. Startup +registers the target worker when scheduling is enabled. The main maintenance tick +no longer calls the combined decision sweep at all. No new table or migration is +added. + +## Producer audit + +| Producer | Intent before write | Scope | +| --- | --- | --- | +| Decision creation, including bundles and caller-owned transactions | Existing creation transaction, before insertion | Decision ID | +| `issueService.update` entering/reasserting done or cancelled | Existing issue update transaction, before the patch | Issue ID | +| Issue removal | Existing removal transaction, before deletion | Full open scan | +| Routine dispatch failure deleting its created issue | Existing dispatch transaction, before deletion | Full open scan | +| Company deletion | Existing deletion transaction, before issue deletion | Full open scan | + +The existing target link table supports ordinary issue scopes. Deletion can +cascade away those links before notification delivery, so it requests a full +bounded scan. This rare path must not depend on after-commit links. Company +removal retains all existing restrictive foreign keys; the notification does not +make an otherwise invalid deletion succeed. + +Other direct issue writers were inspected: checkout/adoption enter in-progress; +conversation writers enter todo/in-progress/in-review; recovery enters blocked; +release preserves existing terminal status; administrative release changes locks +and assignment. Tree restoration excludes terminal snapshots. Route aliases and +raw privacy SQL do not write terminal status. These paths do not create new +invalidation obligations. They can restore a target before the worker runs; the +worker reads its current status. + +The initial decision feature (0197 / commit `9c1f8e7887`) already stored primary +and secondary target links in the creation transaction. The later shared-helper +extraction `ef33c1d9ed` preserved that behavior. There is no newly assumed legacy +secondary-link backfill. Direct SQL writers or damaged/incomplete link data are +outside the supported live writer contract; startup still examines all open rows. + +## Scheduling and lifecycle + +The owning service holds pending scopes and cursors, not a second durable queue. +The database decisions, options, snapshots, and target rows remain truth. Each +pass examines at most 100 rows/scopes, ordered by expiry and ID using the existing +open-decision index. An empty scope consumes one unit of the pass budget. It +rotates unfinished scopes so a long startup scan does not starve new changes. + +A settled notification replaces its scope with a fresh cursor. An admitted pass +cannot delete that replacement when it completes. Failed rows receive individual +retry scopes while later rows continue; query failures retain their scope. +Retries use the coordinator's existing 30-second cadence only while reconciliation +is unfinished. Unknown hints request a bounded full scan. Once all scopes settle, +`hasPending` returns from memory and no future target scan is scheduled. + +Intent registration is awaited before writes. Nested writes belong to the outer +transaction. Rollback does not publish a change; ambiguous outcomes retain the +existing XID reconciliation contract. Notifications cover the application's DB +owner and child pools, not arbitrary SQL writers or independent active processes. + +Drain preserves pending scopes without passive target SQL. Resume processes them. +Shutdown checks cancellation between rows and awaits admitted expiry/delivery. +Startup reconstructs anything left after process exit. Cloud's conservative +sleep guards are unchanged; this establishes worker quiet, not permission to +sleep with unsupported obligations or proof of host suspension. + +## Verification + +The initial 17-case real-PostgreSQL target suite passed. The combined selection +passed 233 tests: those 17, 17 effect-recovery cases, 14 TTL cases, 50 decision +service cases, 39 startup cases, 80 routine-service cases, and 16 routine-route +cases. The final target run passed all 18 cases, including a change to the same scope +during reconciliation. In total, 234 selected cases pass across these runs. + +Coverage includes actual issue update/removal, decision creation after cancellation, +secondary targets, all-target completion, intentional done targets, lenient history, +outer commit/rollback, startup after deleted links, first change and failed expiry +through drain, reopened targets, scope isolation, more than one page of valid +open rows, failure fairness, creation during delivery, and shutdown. The routine +failure test observes intent and settlement around its real cleanup transaction. +Quiet assertions observe SQL through one hour of fake time after settlement. + +Workspace typecheck/build passed before the final test-only addition. No runtime +test failure was suppressed or timeout increased. Existing warnings from startup +fixture signal listeners remain visible. The routine suite now supports the +validated disposable external test database path, so local embedded PostgreSQL +unavailability does not silently skip its 80 cases. + +A local query probe with 40,000 completed decisions, one open decision, and 40,001 +target links used existing indexes: scoped lookup read five buffers (0.012 ms), +full open-page lookup two (0.003 ms). The probe was rolled back. No new schema +index is needed for this change. Full PR CI and fresh review remain required. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index eea85a50269..856149cc62d 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -581,3 +581,24 @@ rollback, restart with deleted target links, failed expiry, concurrent scope changes, fair paging, drain/resume, and shutdown. Exercise relevant route/service consumers when adding notification calls to their existing transaction boundary. Do not remove the periodic target sweep before this coverage exists. + + +## Slice 12 implementation: decision targets + +The target owner, scoped notification adapter, and audited producer calls are +implemented. Startup reconstructs open targets; ordinary terminal writes select +linked decisions; deletion requests a bounded full scan after cascading links. +The aggregate decision scan is removed now that its four separate owners exist. +See [target invalidation](../idle-decision-targets.md) for scope and policy. + +The combined real-database/consumer selection passes 233 tests. Workspace +typecheck and build pass. The final target run also passes all 18 cases, including the additional +same-scope concurrent-change regression (234 unique selected cases in total). No new table or migration is introduced. Existing +indexes support both lookup shapes in the 40,000-row completed-history probe. + +Slice 11 #17482 has Greptile 5/5 at `a5b5994c1d` and no unresolved findings. +Its first browser shard 3/8 failed in agent-chat fixture cleanup: cancelling a +run hit a PostgreSQL lock timeout, and the following test saw the experimental +flag left enabled. No CI retry has been used. Reproduce the two tests on the +exact parent head before classifying a retry or repair. This parent is not yet +claimed green. The full original inventory remains in scope. diff --git a/server/src/__tests__/decision-target-work.test.ts b/server/src/__tests__/decision-target-work.test.ts new file mode 100644 index 00000000000..d6df1c41b2b --- /dev/null +++ b/server/src/__tests__/decision-target-work.test.ts @@ -0,0 +1,262 @@ +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, 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 { 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"; + +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> | undefined; + let db: ReturnType; + let companyId: string, agentId: string, originId: string, targetId: string, runId: string, userId: string; + let coordinator: ReturnType | undefined; + let draining: boolean; + let recovery: ReturnType; + 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("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("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(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => gate); + const starting = start(); let stopped = false; let stopping: Promise; + 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); + }); +}); diff --git a/server/src/__tests__/routines-service.test.ts b/server/src/__tests__/routines-service.test.ts index 22c40d25aa1..d84a9aa1ea3 100644 --- a/server/src/__tests__/routines-service.test.ts +++ b/server/src/__tests__/routines-service.test.ts @@ -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(["intent", "settled"]); expect(run.status).toBe("failed"); expect(run.failureReason).toContain("queue unavailable"); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 46ffb44aa1c..20c1d510286 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -340,6 +340,9 @@ vi.mock("../services/index.js", () => ({ })), 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 })), @@ -867,7 +870,9 @@ 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); + expect(decisions.sweepExpired).not.toHaveBeenCalled(); } finally { setIntervalSpy.mockRestore(); } @@ -907,6 +912,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(); } diff --git a/server/src/index.ts b/server/src/index.ts index 7b5e361a84f..99b57f15e8a 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -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"; @@ -1809,7 +1810,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 }, @@ -1830,9 +1832,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"); })); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 432cad76e7a..4060eda0896 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { agentObservabilityService } from "./agent-observability.js"; import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server"; @@ -106,6 +107,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(routines).where(eq(routines.companyId, id)); await tx.delete(issueReadStates).where(eq(issueReadStates.companyId, id)); await tx.delete(documents).where(eq(documents.companyId, id)); + await notifyDecisionTargetWork(tx); await tx.delete(issues).where(eq(issues.companyId, id)); await tx.delete(companyLogos).where(eq(companyLogos.companyId, id)); await tx.delete(assets).where(eq(assets.companyId, id)); diff --git a/server/src/services/decision-target-work.ts b/server/src/services/decision-target-work.ts new file mode 100644 index 00000000000..e3ae6fee108 --- /dev/null +++ b/server/src/services/decision-target-work.ts @@ -0,0 +1,21 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import type { decisionService } from "./decisions.js"; + +export function notifyDecisionTargetWork(transaction: object, hint?: `issue:${string}` | `decision:${string}`) { + return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionTargets, hint); +} + +export function registerDecisionTargetWork( + coordinator: ReturnType, + service: Pick, "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(), + }); +} diff --git a/server/src/services/decision-targets.ts b/server/src/services/decision-targets.ts new file mode 100644 index 00000000000..bbebfd080d3 --- /dev/null +++ b/server/src/services/decision-targets.ts @@ -0,0 +1,78 @@ +import { and, asc, eq, 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"; + +type Decision = typeof decisions.$inferSelect; +type Cursor = { expiresAt: Date; id: string }; +type Scope = { kind: "all" | "issue" | "decision"; id?: string; cursor?: Cursor }; +const open = sql`${decisions.status} = 'open'`; + +export function decisionTargetService(db: Db, expire: ReturnType["expire"]) { + const scopes = new Map([["all", { kind: "all" }]]); + + function changed(hints?: ReadonlySet) { + 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; + 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().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.toISOString()}::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.expiresAt, 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 { reason, sweep, changed, hasPending: () => scopes.size > 0 }; +} diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index ea85e24e8c9..2e935fadd4a 100644 --- a/server/src/services/decisions.ts +++ b/server/src/services/decisions.ts @@ -1,3 +1,5 @@ +import { decisionTargetService } from "./decision-targets.js"; +import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { randomUUID } from "node:crypto"; import { and, asc, count, desc, eq, gt, gte, inArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -127,6 +129,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 targets = decisionTargetService(db, expiry.expire); const effectRecovery = decisionEffectRecoveryService(db, recoverDecisionEffects); const scope = databaseWorkScopeKey(db); let flights = effectFlights.get(scope); @@ -238,6 +241,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }; const id = randomUUID(); await notifyDecisionChange(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, @@ -727,34 +731,23 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return true; } - async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean; includeEffects?: boolean } = {}) { + async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean; includeEffects?: boolean; includeTargets?: 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 + const targetRows = options.includeTargets !== false && 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; - 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"; + for (const decision of rows) { + const invalid = options.includeTargets === false ? null : await targets.reason(decision); + if (!invalid && (options.includeExpiry === false || decision.expiresAt > now)) continue; + const reason = invalid ?? "ttl"; expired += await expiry.expire(decision, reason, now); } return { expired, resumed }; @@ -762,5 +755,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, - recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next }; + recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next, + sweepInvalidTargets: targets.sweep, decisionTargetsChanged: targets.changed, hasDecisionTargetWork: targets.hasPending }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 8d69fe02f3d..3c922e87299 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + decisionTargets: "decision-targets", decisionEffectRecovery: "decision-effect-recovery", decisionExpiry: "decision-expiry", decisionContinuation: "decision-continuation", diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index b1cbdaa2fdb..e10ade72260 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { insertIssueComment } from "./issue-comment-writes.js"; import { measureRunTransaction } from "./execution-transaction-timing.js"; import { readNativeMonitorSource } from "./native-runtime/native-monitor-delivery.js"; @@ -11296,6 +11297,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) @@ -11703,6 +11707,7 @@ export function issueService(db: Db) { try { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await requestRestorePointDeletion(tx, { sourceIssueId: id }); + await notifyDecisionTargetWork(tx); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index a5384e1fc71..d9295c8322f 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1,3 +1,4 @@ +import { notifyDecisionTargetWork } from "./decision-target-work.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 notifyDecisionTargetWork(txDb); await txDb.delete(issues).where(eq(issues.id, createdIssue.id)); } const failureReason = error instanceof Error ? error.message : String(error);