fix(server): invalidate decision targets from committed changes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

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.

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<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("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<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);
+  });
+});

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");

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();
     }

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");
         }));

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));

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<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(),
+  });
+}

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<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().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 };
+}

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<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";
+    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 };
 }

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",

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))

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);