diff --git a/doc/idle-decision-targets.md b/doc/idle-decision-targets.md new file mode 100644 index 00000000000..3ceb5153bd2 --- /dev/null +++ b/doc/idle-decision-targets.md @@ -0,0 +1,164 @@ +# 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. Startup +registers the target worker when scheduling is enabled. The combined internal +`sweepExpired` method is removed after the final duty is split. 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 | Lock the issue and capture linked open decision IDs in the existing transaction, before deletion | Affected decision IDs; no signal when none | +| Routine dispatch failure deleting its created issue | Same locked pre-delete lookup in the existing dispatch transaction | Affected decision IDs; usually no signal | + +The existing target link table supports ordinary issue scopes. Before a target +is deleted, the decision owner locks its issue row, reads distinct linked open +decisions in the same company, and signals their IDs. The links may then cascade +without losing notification scope. The row lock closes the race in which a new +proposal could add its target reference after discovery but before deletion. +This remains part of the caller's existing transaction; no provider operation +is wrapped in a new transaction. Production hints require an issue or decision +ID, so routine cleanup and repeated removals do not restart full startup paging. + +Company deletion no longer signals target work. With no decision rows it cannot +create a same-company invalidation obligation. Pre-existing restrictive origin +run/issue foreign keys still prevent deleting some companies with decisions; +that separate limitation is unchanged. This migration does not claim to implement +complete decision-data deletion for those companies. + +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. The cursor retains PostgreSQL timestamp text, including +microseconds, rather than rounding through a JavaScript Date. 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. + +## Current verification + +All 248 cases in seven complete suites pass: 25 target, 80 routine, 54 decision, +26 effect-recovery, 20 expiry, 42 startup and one scoped-deadline case. PostgreSQL +suites use separate migrated external fixtures. Server typecheck and build pass. +No new migration or table is added. The normal parent merge preserves landed +expiry and all 21 independent master files. Server types and build use the +updated Runner vendor (1,524 inputs and 1,563 outputs). + +Four real regressions failed before repair: sub-millisecond paging, deletion +behind a full valid startup page, unrelated deletion notifications and successful +empty-company deletion notifications. They all pass. The new concurrency test +observes the proposal's actual PostgreSQL lock wait while deletion is held. +Temporarily removing the issue-row lock made the test fail because the proposal +was accepted instead of waiting; restoring it passes. The mutation was restored +before commit. No timeout or production hold was weakened. + +The combined internal method had no production caller after this split. The +server package exposes only its startup module; neither its exports nor the +plugin SDK exposes the decision service. Its tests now call the dedicated TTL, +target, effect and continuation workers, preserving their outcome assertions. +The obsolete cursor, scan options and exported target-policy method are removed. +HTTP and plugin SDK contracts are unchanged. + +Fresh current-head CI and both automated reviews remain required. Historical +fixture and CI failures below remain part of the verification record. Extra +upstream embedded-only tests attempted during parent integration could not +initialize locally; CI must cover that environment. + +## Earlier 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 original routine +failure test observed intent and settlement around its cleanup transaction. The +current test requires no target signal when that created issue has no linked +decisions. +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. + +## Earlier review and CI repair + +Review found that an absent deletion hint was dropped when the outer transaction +also carried a scoped hint. Both delete-before-update and update-before-delete +regressions reproduced the missing expiry after the target links cascaded away. +That revision sent the explicit `all` hint alongside scoped hints; the current +pre-delete decision-ID contract below supersedes the global hint. +The repaired target suite passes 20 cases; the TTL suite passes 14 cases. + +The initial CI run failed the TTL worker's drain assertion. It measured SQL +before already-admitted sweeps had finished. The test now joins those specific +sweep promises before measuring passive SQL; unresolved-intent holds remain +intact. Waiting for the entire idle count to reach zero was tested and rejected: +that count correctly includes the retained recovery holds. No production drain +or retry behavior changed for this test repair. 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 c4fd7944142..f81d138e2d9 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -16,7 +16,8 @@ Do not mark an item done from a timer removal or a green check alone. | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged after exact-head 5/5 reviews and passing current CI | -| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482), and targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) await current-head review/CI gates | +| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) await current-head review/CI gates | + | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | @@ -630,3 +631,140 @@ integrates that landed parent and all 21 independent master files. All 143 cases in its five complete suites pass. Its preceding CI shard interruption remains recorded in the owning effect-recovery audit; it is not a passing receipt. Fresh CI/reviews are required on the reconciled head. + + +Slice 11 is [#17482](https://github.com/paperclipai/paperclip/pull/17482), head +`a5b5994c1d`, based on #17479. It is attached and Greptile review was requested. +Its CI is active with no failure in the latest snapshot. Slice 10's one +serialized-job retry passed; all checks now pass at `63b58cc26f`, with Greptile +5/5 and no unresolved findings. The recorded original timeout remains part of +its verification history. + +## Next owner audit: decision target invalidation + +Branch `codex/sleep-12-decision-target-invalidation` starts from #17482. No target +migration is implemented or ready yet. This is the final responsibility inside +the current combined decision sweep, separate from retention and notifications. + +Preserve its exact policy: only strict effect references count. A missing target +or any cancelled target expires with `target_gone`. If all strict targets are +done and at least one was not done in the saved proposal snapshot, expire with +`target_completed`. Lenient decisions and proposals intentionally targeting +already-done tasks remain open. Include secondary effect references through the +existing `decisionEffectTargetIssueIds` helper. Do not duplicate provider/effect +payload interpretation inside the scheduler or broaden invalidation to every +updated timestamp. TTL validity remains separately enforced on use. + +The direct-writer audit found the main `issueService.update` boundary can enter +done/cancelled and already owns an outer transaction. Release preserves an +existing terminal status; checkout/adoption enter in-progress; conversation +writers enter todo/in-progress/in-review; recovery writers enter blocked. +Tree restoration explicitly excludes terminal snapshots. Administrative release +only changes locks/assignment. Route aliases `issueRows` write source-trust or +owner-notification metadata. Raw SQL updates found in issue/project privacy +propagation change privacy only. These are audit findings, not a substitute for +producer integration tests or final inspection. + +Three direct deletion paths need accounting: issue removal, routine dispatch +failure cleanup, and company deletion. Routine creation is visible through its +service while dispatch holds a separate transaction, so do not assume its failed +issue could never have become a target. Company deletion currently encounters +restricting decision/run references if decisions exist; do not silently fix or +relax those foreign keys in this slice. Single-target links cascade on issue +deletion, so an after-commit lookup through those links cannot identify deleted +targets. Use pre-write intent and either preserve the affected decision scope or +request a bounded startup-style scan for this rare path. + +A candidate design uses the existing coordinator's scoped hints: decision +creation signals its decision ID; terminal issue writes signal their issue ID; +deletion or unknown hints request a bounded full scan. Existing decision-target +links select affected open rows; the domain helper computes invalidation from +current targets and saved snapshots. Stable valid open decisions must become +quiet after their finite reconciliation, even with long retention or no near TTL. +No per-decision timer or new queue table is required. + +Keep scoped scan state in the owner, with bounded pages and fair rotation. +Changes arriving during a scan must reset/requeue that scope without being lost +when the admitted pass completes. Failed candidates retain retry debt while +allowing later scopes to progress. Startup reconstructs all open targets from +durable state; rollback and uncertain writes use the existing transaction +protocol. `hasPending` must not perform passive SQL during drain. Shutdown must +finish admitted expiry and leave later candidates for restart. A changed target +can reopen before reconciliation; inspect current state rather than blindly +expiring from the notification itself. + +Required tests include the actual issue update/removal producers, creation after +cancellation, intentional done-target proposals, secondary targets, lenient +history, no recurring SQL with valid open decisions, outer transactions, +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. + +Slice 11 #17482 now has all checks passing at `a5b5994c1d`, Greptile 5/5, +no unresolved findings, and no merge conflict. Its one browser-shard retry passed +after both affected cases passed locally on the exact head (2/2). The original +fixture-cleanup lock timeout remains documented in the PR. + +Slice 12 is #17486 at initial head `5062efb368`. Review found that deletion's +absent hint could be dropped when the same outer transaction included a scoped +hint. Both operation orders failed new regressions before the explicit `all` +hint repair. The repaired target/TTL selection passes 34 cases. CI also exposed +a TTL test race: passive-SQL measurement now joins already-admitted sweeps, +without dropping unresolved recovery holds. Fresh CI/review remain required. + +The fresh master audit at `991c47bc6e` identified an additional recurring-server +owner: the Grok routine dispatcher starts a one-second settings/dispatch scan +when public MCP is configured, even when Grok is disabled. Include this in the +full stack as slice 13, separately from decision retention. The disabled baseline +issued three settings queries in 3.2 seconds. Its durable prepared dispatches, +atomic claim, and existing uncertain-outcome policy must remain authoritative. + +Slice 12 review repair: deletion captures affected open decision IDs under an +issue-row lock before links cascade. Ordinary cleanup without linked decisions +and empty-company deletion no longer signal global work. PostgreSQL text cursors +preserve microsecond precision. Four real regressions failed before repair; +removing the new lock also failed a controlled concurrent-proposal test. All +246 target/routine/decision/effect/expiry/startup/deadline cases and server +types/build pass. The existing restrictive company-deletion FKs and explicit +combined-service compatibility path remain documented. Fresh head gates required. + + +Slice 12 final combined-sweep cleanup: the repository/package/SDK audit found no +production consumer or public exposure of `decisionService.sweepExpired` after +the final duty split. Removed that method, its cursor/flags/imports and the +unneeded target-policy export. Its behavioral tests now use the four dedicated +workers. Real isolation tests show TTL does not recover effects and effect +recovery does not expire open decisions. All 248 cases in seven complete suites, +server types and build pass after integrating parent `c304273b75`. This supersedes +the earlier manual-caller compatibility claim. Both creation signals remain: +expiry-only creation and scoped target invalidation. Fresh head gates required. + + +Slice 11 merged at `c304273b75` → `e9e84bb8a1` after exact-head Greptile/Ponytail +5/5 and all 58 checks passed or were neutral/skipped. Slice 12 normally merges +that landed dependency and preserves four independent Runner E2E accounting +files. All runtime and acceptance test sources are byte-identical to its tested +`85ce6545bf` tree (248 cases, server types/build). Fresh final-head CI and reviews +remain required after retargeting to master. diff --git a/server/src/__tests__/decision-effect-recovery-work.test.ts b/server/src/__tests__/decision-effect-recovery-work.test.ts index 4358b8e3215..6c1a17d2b39 100644 --- a/server/src/__tests__/decision-effect-recovery-work.test.ts +++ b/server/src/__tests__/decision-effect-recovery-work.test.ts @@ -316,12 +316,15 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes expect(errors.every(error => String(error).includes("wake unavailable"))).toBe(true); errors.length = 0; }); - it("keeps aggregate maintenance from recovering effects when the worker owns them", async () => { - const row = await create(); await claim(row.id, new Date(Date.now() - 120_000)); - expect(await service().sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false })) - .toEqual({ expired: 0, resumed: 0 }); + it("does not expire overdue open decisions during effect recovery", async () => { + const row = await create(); + await db.update(decisions).set({ expiresAt: new Date(Date.now() - 1000) }).where(eq(decisions.id, row.id)); + expect(await service().recoverEffects()).toEqual({ resumed: 0 }); + expect(await service().get(row.id)).toMatchObject({ status: "open" }); expect(await comments()).toHaveLength(0); - expect(await service().sweepExpired()).toEqual({ expired: 0, resumed: 1 }); expect(await comments()).toHaveLength(1); + expect(wake).not.toHaveBeenCalled(); + expect(await service().sweepTtlExpiry()).toEqual({ expired: 1 }); + expect(wake).toHaveBeenCalledTimes(1); }); it.each(["missing_decider", "missing_option"] as const)("settles %s as durable failure and then becomes quiet", async reason => { diff --git a/server/src/__tests__/decision-expiry-work.test.ts b/server/src/__tests__/decision-expiry-work.test.ts index 51c5c7c9673..7b35a9dde6a 100644 --- a/server/src/__tests__/decision-expiry-work.test.ts +++ b/server/src/__tests__/decision-expiry-work.test.ts @@ -62,8 +62,10 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes async function start() { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); - await registerDecisionContinuationWork(coordinator, service(), () => !draining).ready; - await registerDecisionExpiryWork(coordinator, service(), () => !draining).ready; + const continuation = service(), expiry = service(); + await registerDecisionContinuationWork(coordinator, continuation, () => !draining).ready; + await registerDecisionExpiryWork(coordinator, expiry, () => !draining).ready; + return { continuation, expiry }; } async function quiet() { await vi.waitFor(async () => { @@ -218,12 +220,14 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes }); it("keeps a failed expiry retry through drain and preserves pending state", async () => { - const row = await create(new Date(Date.now() + 30_000)); await start(); + const row = await create(new Date(Date.now() + 30_000)); const workers = await start(); + const sweeps = [vi.spyOn(workers.expiry, "sweepTtlExpiry"), vi.spyOn(workers.continuation, "sweepContinuations")]; await db.execute(sql`create function decision_expiry_fail() returns trigger language plpgsql as $$ begin if NEW.status = 'expired' then raise exception 'expiry unavailable'; end if; return NEW; end $$`); await db.execute(sql`create trigger decision_expiry_fail before update on decisions for each row execute function decision_expiry_fail()`); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); }); draining = true; expect((await service().get(row.id))?.status).toBe("open"); + await Promise.allSettled(sweeps.flatMap(sweep => sweep.mock.results.filter(result => result.type === "return").map(result => result.value))); const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); await db.execute(sql`drop trigger decision_expiry_fail on decisions`); @@ -247,12 +251,15 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes expect(wake).not.toHaveBeenCalled(); expect(await service().nextExpiryAt()).toBeNull(); }); - it("keeps TTL expiry out of the aggregate sweep when its worker owns the duty", async () => { + it("does not recover stale running effects during TTL expiry", async () => { const row = await create(); - await db.update(decisions).set({ expiresAt: new Date(Date.now() - 1000) }).where(eq(decisions.id, row.id)); - expect(await service().sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false })).toEqual({ expired: 0, resumed: 0 }); - expect((await service().get(row.id))?.status).toBe("open"); - expect(await service().sweepTtlExpiry()).toEqual({ expired: 1 }); expect(wake).toHaveBeenCalledTimes(1); + await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes", + decidedByUserId: null, updatedAt: new Date(Date.now() - 120_000), expiresAt: new Date(Date.now() - 1000) }) + .where(eq(decisions.id, row.id)); + expect(await service().sweepTtlExpiry()).toEqual({ expired: 0 }); + expect(await service().get(row.id)).toMatchObject({ status: "decided", executionStatus: "running" }); + expect(wake).not.toHaveBeenCalled(); + expect(await service().nextExpiryAt()).toBeNull(); }); it("pages past a failed candidate and retries it without repeating completed expirations", async () => { 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..f9ec92d1068 --- /dev/null +++ b/server/src/__tests__/decision-target-work.test.ts @@ -0,0 +1,366 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, authUsers, companies, companyMemberships, createDb, subscribeDatabaseWork, decisionEffectExecutions, + decisions, decisionRetention, decisionTargetIssues, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { decisionService } from "../services/decisions.js"; +import { registerDecisionTargetWork } from "../services/decision-target-work.js"; +import { registerDecisionContinuationWork } from "../services/decision-continuation-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { issueService } from "../services/issues.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { deleteCompany } from "../services/company-deletion.js"; +import { notifyDecisionTargetIssueRemoval } from "../services/decision-targets.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISION_TARGETS_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("decision target worker", () => { + let temporary: Awaited> | 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("finishes startup paging across submillisecond expiry values", async () => { + const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions); + await db.insert(decisions).values(Array.from({ length: 101 }, () => ({ ...seed, id: randomUUID() }))); + await db.execute(sql`update decisions set expires_at = date_trunc('milliseconds', now() + interval '1 day') + interval '456 microseconds'`); + const owner = service(); + await owner.sweepInvalidTargets(); + expect(owner.hasDecisionTargetWork()).toBe(true); + await owner.sweepInvalidTargets(); + expect(owner.hasDecisionTargetWork()).toBe(false); + expect(wake).not.toHaveBeenCalled(); + }); + + it("reaches a deleted target behind a full page of valid startup decisions", async () => { + const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions); + const invalidId = "ffffffff-ffff-4fff-afff-ffffffffffff"; + const valid = Array.from({ length: 100 }, (_, index) => ({ ...seed, + id: `01000000-0000-4000-8000-${String(index).padStart(12, "0")}`, + options: [{ id: "yes", label: "Yes", effects: [{ type: "comment_on_issue" as const, targetIssueId: originId, staleness: "strict" as const, bodyMarkdown: "hello" }] }], + })); + await db.insert(decisions).values([...valid, { ...seed, id: invalidId }]); + await db.insert(decisionTargetIssues).values({ companyId, decisionId: invalidId, issueId: targetId }); + const owner = service(); + await owner.sweepInvalidTargets(); + expect(owner.hasDecisionTargetWork()).toBe(true); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, (event, hints) => { + if (event === "settled") owner.decisionTargetsChanged(hints); + }); + try { + await issueService(db).remove(targetId); + expect(await owner.sweepInvalidTargets()).toEqual({ expired: 1 }); + expect((await owner.get(invalidId))?.metadata).toMatchObject({ expiredReason: "target_gone" }); + expect(owner.hasDecisionTargetWork()).toBe(false); + expect(wake).toHaveBeenCalledTimes(1); + } finally { unsubscribe(); } + }); + + it("fences new target references between deletion discovery and commit", async () => { + let entered!: () => void, release!: () => void; + const discovered = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const removing = db.transaction(async tx => { + await notifyDecisionTargetIssueRemoval(tx, targetId); + entered(); await gate; + await tx.delete(issues).where(eq(issues.id, targetId)); + }); + let proposing: Promise<{ accepted: boolean }> | undefined; + try { + await discovered; + proposing = create().then(() => ({ accepted: true }), () => ({ accepted: false })); + await vi.waitFor(async () => { + const blocked = await db.execute(sql`select pid from pg_stat_activity where datname = current_database() + and pid <> pg_backend_pid() and wait_event_type = 'Lock' and query like '%decision_target_issues%'`); + expect(blocked).toHaveLength(1); + }); + release(); await removing; + await expect(proposing).resolves.toEqual({ accepted: false }); + expect(await db.select().from(decisions)).toEqual([]); + expect(wake).not.toHaveBeenCalled(); + } finally { release(); await Promise.allSettled([removing, ...(proposing ? [proposing] : [])]); } + }); + + it("does not signal target work for removal with no linked decisions", async () => { + await create(); + const unrelatedId = randomUUID(); + await db.insert(issues).values({ id: unrelatedId, companyId, title: "Unrelated", status: "todo", responsibleUserId: userId }); + const settled = vi.fn(); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, event => { if (event === "settled") settled(); }); + try { await issueService(db).remove(unrelatedId); expect(settled).not.toHaveBeenCalled(); } + finally { unsubscribe(); } + }); + + it("does not launch global target work after deleting an empty company", async () => { + const emptyCompanyId = randomUUID(); + await db.insert(companies).values({ id: emptyCompanyId, name: "Empty", issuePrefix: `E${emptyCompanyId.slice(0, 6)}` }); + const home = await mkdtemp(join(tmpdir(), "decision-target-company-")); + vi.stubEnv("PAPERCLIP_HOME", home); + const settled = vi.fn(); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.decisionTargets, event => { if (event === "settled") settled(); }); + try { + expect(await deleteCompany(db, emptyCompanyId)).toMatchObject({ id: emptyCompanyId }); + expect(settled).not.toHaveBeenCalled(); + } finally { unsubscribe(); await rm(home, { recursive: true, force: true }); } + }); + + it("keeps lenient and intentional already-done proposals open and quiet", async () => { + const lenient = await create("lenient"); await change("done"); const intentional = await create(); + await start(); await quiet(); + expect((await service().get(lenient.id))?.status).toBe("open"); + expect((await service().get(intentional.id))?.status).toBe("open"); expect(wake).not.toHaveBeenCalled(); + }); + + it("expires a decision created after its target was cancelled", async () => { + await change("cancelled"); await start(); await quiet(); const row = await create(); + await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("reacts to the real removal producer after target links cascade away", async () => { + const row = await create(); await start(); await quiet(); await issueService(db).remove(targetId); + expect(await db.select().from(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, row.id))).toHaveLength(0); + await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it.each(["cancelled", "deleted"] as const)("reconstructs %s target invalidation at startup", async action => { + const row = await create(); + if (action === "deleted") await db.delete(issues).where(eq(issues.id, targetId)); + else await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, targetId)); + await start(); await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("includes a strict secondary effect reference in notification scope", async () => { + const blockerId = randomUUID(); + await db.insert(issues).values({ id: blockerId, companyId, title: "Blocker", status: "todo", responsibleUserId: userId }); + const row = await service().create({ companyId, actor: actor(), agentId, runId, title: "Follow up?", body: "Body", + continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [ + { type: "create_issue", targetIssueId: targetId, staleness: "strict", draft: { title: "Next", blockedByIssueIds: [blockerId] } }, + ] }] }); + await start(); await quiet(); await issueService(db).update(blockerId, { status: "cancelled", actorUserId: userId }); + await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("waits for the last strict target to complete", async () => { + const blockerId = randomUUID(); + await db.insert(issues).values({ id: blockerId, companyId, title: "Blocker", status: "todo", responsibleUserId: userId }); + const row = await service().create({ companyId, actor: actor(), agentId, runId, title: "Follow up?", body: "Body", + continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [ + { type: "create_issue", targetIssueId: targetId, staleness: "strict", draft: { title: "Next", blockedByIssueIds: [blockerId] } }, + ] }] }); + await start(); await quiet(); await change("done"); await quiet(); + expect((await service().get(row.id))?.status).toBe("open"); + await issueService(db).update(blockerId, { status: "done", actorUserId: userId }); + await expired(row.id, "target_completed"); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("waits for the actual issue update's outer transaction and ignores rollback", async () => { + const row = await create(); await start(); await quiet(); + await expect(db.transaction(async tx => { + await issueService(db).update(targetId, { status: "cancelled", actorUserId: userId }, tx); + await vi.advanceTimersByTimeAsync(120_000); expect(wake).not.toHaveBeenCalled(); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect((await service().get(row.id))?.status).toBe("open"); + await db.transaction(async tx => { + await issueService(db).update(targetId, { status: "cancelled", actorUserId: userId }, tx); + await vi.advanceTimersByTimeAsync(120_000); expect(wake).not.toHaveBeenCalled(); + }); + await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it.each([true, false])("preserves a deletion mixed with a scoped hint in one transaction, deleteFirst=%s", async deleteFirst => { + const row = await create(); const otherId = randomUUID(); + await db.insert(issues).values({ id: otherId, companyId, title: "Unrelated", status: "todo", responsibleUserId: userId }); + await start(); await quiet(); + await db.transaction(async tx => { + const scoped = issueService(tx as unknown as typeof db); + if (deleteFirst) await scoped.remove(targetId); + await scoped.update(otherId, { status: "done", actorUserId: userId }); + if (!deleteFirst) await scoped.remove(targetId); + }); + expect(await db.select().from(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, row.id))).toHaveLength(0); + await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("retains a first target change through drain without passive SQL", async () => { + const row = await create(); await start(); await quiet(); draining = true; await change("cancelled"); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("checks current state if a cancelled target reopened before reconciliation", async () => { + const row = await create(); await start(); await quiet(); draining = true; await change("cancelled"); + await issueService(db).update(targetId, { status: "todo", actorUserId: userId }); + draining = false; await quiet(); expect((await service().get(row.id))?.status).toBe("open"); expect(wake).not.toHaveBeenCalled(); + }); + + it("retains failed expiry through drain and retries after the database recovers", async () => { + const row = await create(); await start(); await quiet(); await failExpiry(); await change("cancelled"); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors.length).toBeGreaterThan(0); }); + draining = true; + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await db.execute(sql`drop trigger decision_target_expire_fail on decisions`); + draining = false; await expired(row.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + for (const error of errors) expect(error).toMatchObject({ cause: { message: "expiry unavailable" } }); errors.length = 0; + }); + + it("does not inspect unrelated open decisions on an ordinary scoped change", async () => { + const unrelated = await create(); + const otherId = randomUUID(); await db.insert(issues).values({ id: otherId, companyId, title: "Other", status: "todo", responsibleUserId: userId }); + const affected = await service().create({ companyId, actor: actor(), agentId, runId, title: "Other?", body: "Body", + continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [ + { type: "comment_on_issue", targetIssueId: otherId, staleness: "strict", bodyMarkdown: "hello" }, + ] }] }); + await start(); await quiet(); + await db.update(decisions).set({ options: sql`'null'::jsonb` }).where(eq(decisions.id, unrelated.id)); + await issueService(db).update(otherId, { status: "cancelled", actorUserId: userId }); + await expired(affected.id); await quiet(); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("bounds startup pages and continues after more than one page of valid decisions", async () => { + const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions); + await db.insert(decisions).values(Array.from({ length: 101 }, () => ({ ...seed, id: randomUUID() }))); + await start(); expect(recovery.hasDecisionTargetWork()).toBe(true); + await quiet(); expect(await db.select().from(decisions).where(eq(decisions.status, "open"))).toHaveLength(101); + expect(wake).not.toHaveBeenCalled(); + }); + + it("continues past a failing first page and retains its failed candidate", async () => { + const seed = await create(); await db.delete(decisionTargetIssues); await db.delete(decisions); + const ids = Array.from({ length: 101 }, (_, i) => `12000000-0000-4000-8000-${String(i).padStart(12, "0")}`); + await db.insert(decisions).values(ids.map(id => ({ ...seed, id }))); + await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, targetId)); + await db.execute(sql`create function decision_target_expire_fail() returns trigger language plpgsql as $$ begin + if NEW.id = '12000000-0000-4000-8000-000000000000' and NEW.status = 'expired' then raise exception 'expiry unavailable'; end if; + return NEW; end $$`); + await db.execute(sql`create trigger decision_target_expire_fail before update on decisions for each row execute function decision_target_expire_fail()`); + await start(); expect(wake).toHaveBeenCalledTimes(99); expect((await service().get(ids.at(-1)!))?.status).toBe("open"); + await expired(ids.at(-1)!); expect((await service().get(ids[0]))?.status).toBe("open"); + await db.execute(sql`drop trigger decision_target_expire_fail on decisions`); + await expired(ids[0]); await quiet(); expect(wake).toHaveBeenCalledTimes(101); + for (const error of errors) expect(error).toMatchObject({ cause: { message: "expiry unavailable" } }); errors.length = 0; + }); + + it("retains a change to the same issue scope while its earlier pass is still running", async () => { + await change("done"); const intentional = await create(); + await db.update(decisions).set({ expiresAt: new Date(Date.now() + 24 * 60 * 60_000) }).where(eq(decisions.id, intentional.id)); + await issueService(db).update(targetId, { status: "todo", actorUserId: userId }); + const ordinary = await create(); await start(); await quiet(); + wake.mockImplementationOnce(async () => { await change("cancelled"); }); + await change("done"); await expired(ordinary.id, "target_completed"); await expired(intentional.id); await quiet(); + expect(wake).toHaveBeenCalledTimes(2); + }); + + it("learns a decision created during invalidation delivery", async () => { + const first = await create(); await change("cancelled"); let nextId = ""; + wake.mockImplementationOnce(async () => { nextId = (await create()).id; }); + await start(); await expired(first.id); await vi.waitFor(() => expect(nextId).not.toBe("")); + await expired(nextId); await quiet(); expect(wake).toHaveBeenCalledTimes(2); + }); + + it("finishes admitted invalidation and leaves later rows for restart during shutdown", async () => { + const first = await create(), second = await create(); await change("cancelled"); + let release!: () => void; const gate = new Promise(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__/decisions-service.test.ts b/server/src/__tests__/decisions-service.test.ts index 63719ba1ea7..7fc2cc50c0d 100644 --- a/server/src/__tests__/decisions-service.test.ts +++ b/server/src/__tests__/decisions-service.test.ts @@ -225,12 +225,12 @@ describePg("decisionService", () => { ...original(source, options), archiveReviewedManifest: async () => { throw new Error("archive_failure"); }, })) : undefined; try { - expect(await service().sweepExpired()).toMatchObject({ resumed: 1 }); + expect(await service().recoverEffects()).toMatchObject({ resumed: 1 }); const saved = await service().get(row.id); expect(saved).toMatchObject({ executionStatus: "failed", metadata: { archiveProposalError: failure, continuationPending: false } }); expect(saved!.metadata.continuationDeliveredAt).toEqual(expect.any(String)); expect(wakes).toHaveLength(1); - await service().sweepExpired(); + await service().recoverEffects(); expect(wakes).toHaveLength(1); } finally { factory?.mockRestore(); } }); @@ -512,7 +512,7 @@ describePg("decisionService", () => { expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); }); - it("recovers stale running decisions from the bounded server sweep", async () => { + it("recovers stale running decisions from the effect worker", async () => { process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS = "0"; const created = await createCommentDecision(); await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes", decidedByUserId, @@ -520,7 +520,7 @@ describePg("decisionService", () => { await db.insert(decisionEffectExecutions).values({ decisionId: created.id, effectIndex: 0, effectType: "comment_on_issue", targetIssueId, status: "claimed" }); - expect(await service().sweepExpired()).toEqual({ expired: 0, resumed: 1 }); + expect(await service().recoverEffects()).toEqual({ resumed: 1 }); expect((await service().get(created.id))?.executionStatus).toBe("succeeded"); expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]); @@ -538,7 +538,7 @@ describePg("decisionService", () => { expect((await service().get(created.id))?.metadata).toMatchObject({ continuationPending: true }); expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); - await service().sweepExpired(); + await service().sweepContinuations(); expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]); expect((await service().get(created.id))?.metadata).toMatchObject({ continuationPending: false }); @@ -613,8 +613,8 @@ describePg("decisionService", () => { const second = await createCommentDecision("lenient", { idempotencyKey: "batch-2", expiresAt: nearFutureExpiry() }); await expireDecisionNow(first.id); await expireDecisionNow(second.id); - expect((await service().sweepExpired()).expired).toBe(1); - expect((await service().sweepExpired()).expired).toBe(1); + expect((await service().sweepTtlExpiry()).expired).toBe(1); + expect((await service().sweepTtlExpiry()).expired).toBe(1); }); it("falls back to the default sweep batch size for invalid configuration", async () => { @@ -624,7 +624,7 @@ describePg("decisionService", () => { await expireDecisionNow(first.id); await expireDecisionNow(second.id); - await expect(service().sweepExpired()).resolves.toMatchObject({ expired: 2 }); + await expect(service().sweepTtlExpiry()).resolves.toMatchObject({ expired: 2 }); }); it("expires TTL and target-gone decisions and wakes the origin agent", async () => { @@ -633,7 +633,8 @@ describePg("decisionService", () => { await db.update(decisions).set({ expiresAt: new Date(0) }).where(eq(decisions.id, ttl.id)); await db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, targetIssueId)); await expireDecisionNow(ttl.id); - expect((await service().sweepExpired()).expired).toBe(2); + expect((await service().sweepTtlExpiry()).expired).toBe(1); + expect((await service().sweepInvalidTargets()).expired).toBe(1); const rows = await db.select().from(decisions); expect(rows.find((row) => row.id === ttl.id)?.metadata).toMatchObject({ expiredReason: "ttl" }); expect(rows.find((row) => row.id === gone.id)?.metadata).toMatchObject({ expiredReason: "target_gone" }); @@ -645,7 +646,7 @@ describePg("decisionService", () => { const lenient = await createCommentDecision("lenient", { idempotencyKey: "lenient-survives" }); await db.update(issues).set({ status: "done" }).where(eq(issues.id, targetIssueId)); - expect((await service().sweepExpired()).expired).toBe(1); + expect((await service().sweepInvalidTargets()).expired).toBe(1); expect((await service().get(completed.id))?.metadata).toMatchObject({ expiredReason: "target_completed" }); expect((await service().get(lenient.id))?.status).toBe("open"); @@ -656,7 +657,7 @@ describePg("decisionService", () => { await db.update(issues).set({ status: "done" }).where(eq(issues.id, targetIssueId)); const reopen = await createCommentDecision("strict", { idempotencyKey: "already-done" }); - expect((await service().sweepExpired()).expired).toBe(0); + expect((await service().sweepInvalidTargets()).expired).toBe(0); expect((await service().get(reopen.id))?.status).toBe("open"); }); @@ -691,7 +692,7 @@ describePg("decisionService", () => { }); await db.update(issues).set({ status: "done" }).where(eq(issues.id, blockerId)); - expect((await service().sweepExpired()).expired).toBe(1); + expect((await service().sweepInvalidTargets()).expired).toBe(1); expect((await service().get(created.id))?.metadata).toMatchObject({ expiredReason: "target_completed" }); }); @@ -716,7 +717,7 @@ describePg("decisionService", () => { await service().decide({ id: acceptedAgain.id, optionId: "assign", decidedByUserId, userActor: boardActor() }); await service().dismiss(rejected.id, decidedByUserId, boardActor(), "Not this time"); await expireDecisionNow(stale.id); - await service().sweepExpired(); + await service().sweepTtlExpiry(); const stats = await service().stats(companyId, { originAgentId: agentId }); expect(stats.filters).toEqual({ originAgentId: agentId, since: null }); @@ -825,7 +826,7 @@ describePg("decisionService", () => { if (action === "decide") await svc.decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() }); if (action === "cancel") await svc.cancel(row.id, { actorType: "agent", actorId: agentId }); if (action === "dismiss") await svc.dismiss(row.id, decidedByUserId, boardActor()); - if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepExpired(new Date(), { includeContinuations: false }); } + if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepTtlExpiry(); } await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); expect(wake.mock.calls[0][0]).toMatchObject({ decisionId: row.id, companyId, agentId, issueId: originIssueId }); diff --git a/server/src/__tests__/routines-service.test.ts b/server/src/__tests__/routines-service.test.ts index 22c40d25aa1..399a8bb5400 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([]); 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 3c814c6dcfe..12f46927634 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -348,7 +348,9 @@ vi.mock("../services/index.js", () => ({ list: vi.fn(async () => ({ items: [], nextCursor: null })), })), decisionService: vi.fn(() => ({ - sweepExpired: vi.fn(async () => ({ expired: 0 })), + sweepInvalidTargets: vi.fn(async () => ({ expired: 0 })), + decisionTargetsChanged: vi.fn(), + hasDecisionTargetWork: vi.fn(() => false), recoverEffects: vi.fn(async () => ({ resumed: 0 })), nextEffectRecoveryAt: vi.fn(async () => null), sweepTtlExpiry: vi.fn(async () => ({ expired: 0 })), @@ -926,7 +928,8 @@ describe("startServer feedback export wiring", () => { expect(decisions.recoverEffects).toHaveBeenCalledTimes(1); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionEffectRecovery)).toBe(true); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionExpiry)).toBe(true); - expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false, includeExpiry: false, includeEffects: false }); + expect(decisions.sweepInvalidTargets).toHaveBeenCalledTimes(1); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionTargets)).toBe(true); } finally { setIntervalSpy.mockRestore(); } @@ -966,6 +969,7 @@ describe("startServer feedback export wiring", () => { expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(false); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionExpiry)).toBe(false); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionEffectRecovery)).toBe(false); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionTargets)).toBe(false); } finally { setIntervalSpy.mockRestore(); } diff --git a/server/src/index.ts b/server/src/index.ts index 0d27d6f5b5e..4ea740f7113 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"; @@ -1845,7 +1846,8 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("decision_effect_recovery_worker", () => registerDecisionEffectRecoveryWork(deliveryWork, decisionExecutor, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); - await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false })); + await step("decision_target_worker", () => registerDecisionTargetWork(deliveryWork, decisionExecutor, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); const loginSchedule = loginCleanupSchedule(db, config.heartbeatSchedulerIntervalMs); await step("login_cleanup_workers", () => registerLoginCleanupWork(deliveryWork, [ { queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult), next: loginSchedule.adapter }, @@ -1866,9 +1868,6 @@ async function startServerWithDatabaseTeardown( trackHeartbeatSchedulerWork((async () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleRoutineStatusReconcile(); - trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false }).catch((err: unknown) => { - logger.error({ err }, "decision expiry sweep failed"); - })); trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => { logger.error({ err }, "decision retention sweep failed"); })); diff --git a/server/src/services/decision-target-work.ts b/server/src/services/decision-target-work.ts new file mode 100644 index 00000000000..219c9ad6f2b --- /dev/null +++ b/server/src/services/decision-target-work.ts @@ -0,0 +1,17 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { decisionService } from "./decisions.js"; + +export function registerDecisionTargetWork( + coordinator: ReturnType, + 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..8334dbaa2a7 --- /dev/null +++ b/server/src/services/decision-targets.ts @@ -0,0 +1,93 @@ +import { and, asc, eq, getTableColumns, inArray, sql } from "drizzle-orm"; +import { decisions, decisionTargetIssues, issues, type Db } from "@paperclipai/db"; +import { decisionEffectTargetIssueIds } from "@paperclipai/shared"; +import type { decisionExpiryService } from "./decision-expiry.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +type Decision = typeof decisions.$inferSelect; +type Cursor = { expiresAt: string; id: string }; +type Scope = { kind: "all" | "issue" | "decision"; id?: string; cursor?: Cursor }; +const open = sql`${decisions.status} = 'open'`; + +export function notifyDecisionTargetWork(transaction: object, hint: `issue:${string}` | `decision:${string}`) { + return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionTargets, hint); +} + +export async function notifyDecisionTargetIssueRemoval(transaction: Pick, issueId: string) { + const [issue] = await transaction.select({ companyId: issues.companyId }).from(issues) + .where(eq(issues.id, issueId)).for("update"); + if (!issue) return; + const linked = await transaction.selectDistinct({ decisionId: decisionTargetIssues.decisionId }).from(decisionTargetIssues) + .innerJoin(decisions, eq(decisions.id, decisionTargetIssues.decisionId)) + .where(and(eq(decisionTargetIssues.companyId, issue.companyId), eq(decisionTargetIssues.issueId, issueId), eq(decisions.companyId, issue.companyId), open)); + for (const row of linked) await notifyDecisionTargetWork(transaction, `decision:${row.decisionId}`); +} + +export function decisionTargetService(db: Db, expire: ReturnType["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({ ...getTableColumns(decisions), cursorExpiresAt: sql`${decisions.expiresAt}::text` }).from(decisions).where(and(open, + scope.kind === "decision" ? eq(decisions.id, scope.id!) : undefined, + scope.kind === "issue" ? inArray(decisions.id, db.select({ id: decisionTargetIssues.decisionId }) + .from(decisionTargetIssues).where(eq(decisionTargetIssues.issueId, scope.id!))) : undefined, + scope.cursor ? sql`(${decisions.expiresAt}, ${decisions.id}) > (${scope.cursor.expiresAt}::timestamptz, ${scope.cursor.id}::uuid)` : undefined, + )).orderBy(asc(decisions.expiresAt), asc(decisions.id)).limit(limit); + inspected += Math.max(1, rows.length); + for (const row of rows) { + if (signal?.aborted) { + if (!scopes.has(key)) scopes.set(key, scope); + return { expired }; + } + try { + const invalid = await reason(row); + if (invalid) expired += await expire(row, invalid, now); + } catch (error) { + errors.push(error); + const retry = `decision:${row.id}`; + if (!scopes.has(retry)) scopes.set(retry, { kind: "decision", id: row.id }); + } + scope.cursor = { expiresAt: row.cursorExpiresAt, id: row.id }; + } + if (rows.length === limit && !scopes.has(key)) scopes.set(key, scope); + } catch (error) { + errors.push(error); inspected++; + if (!scopes.has(key)) scopes.set(key, scope); + } + } + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Decision target reconciliation failed"); + return { expired }; + } + + return { sweep, changed, hasPending: () => scopes.size > 0 }; +} diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index 9d915c7955d..282327d433f 100644 --- a/server/src/services/decisions.ts +++ b/server/src/services/decisions.ts @@ -1,5 +1,6 @@ +import { decisionTargetService, notifyDecisionTargetWork } from "./decision-targets.js"; import { randomUUID } from "node:crypto"; -import { and, asc, count, desc, eq, gt, gte, inArray, sql } from "drizzle-orm"; +import { and, asc, count, desc, eq, gte, inArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { databaseWorkScopeKey, companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db"; import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared"; @@ -15,7 +16,7 @@ import { import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js"; import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js"; -import { decisionEffectRecoveryService, decisionSweepBatchSize } from "./decision-effect-recovery.js"; +import { decisionEffectRecoveryService } from "./decision-effect-recovery.js"; import { decisionExpiryService } from "./decision-expiry.js"; import { notifyDecisionCreated, writeDecisionChange } from "./decision-work-notifications.js"; @@ -128,13 +129,13 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) { export function decisionService(db: Db, options: DecisionServiceOptions) { const continuations = decisionContinuationService(db, options.wakeOriginAgent); const expiry = decisionExpiryService(db, continuations.deliver); + const targets = decisionTargetService(db, expiry.expire); const effectRecovery = decisionEffectRecoveryService(db, recoverDecisionEffects); const scope = databaseWorkScopeKey(db); let flights = effectFlights.get(scope); if (!flights) effectFlights.set(scope, flights = new Map()); const executing = flights; const authz = authorizationService(db); - let targetSweepCursor: string | null = null; type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null; ruleKey?: string | null; title: string; body: string; options: DecisionOption[]; inputs?: DecisionInput[] | null; expiresAt?: Date | null; idempotencyKey?: string | null; continuationPolicy?: "none" | "wake_origin_agent"; metadata?: Record }; @@ -239,6 +240,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }; const id = randomUUID(); await notifyDecisionCreated(dbOrTx); + await notifyDecisionTargetWork(dbOrTx, `decision:${id}`); const [created] = await dbOrTx.insert(decisions).values({ id, companyId: input.companyId, bundleId: input.bundleId ?? null, originAgentId: input.agentId, originIssueId: provenance.issueId, originRunId: input.runId, ruleKey: input.ruleKey ?? null, title: input.title, body: input.body, options: input.options, inputs: input.inputs ?? null, expiresAt, @@ -760,40 +762,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return true; } - async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean; includeEffects?: boolean } = {}) { - const batchSize = decisionSweepBatchSize(); - const { resumed } = options.includeEffects === false ? { resumed: 0 } : await effectRecovery.sweep(now, batchSize); - if (options.includeContinuations !== false) await continuations.sweep(batchSize); - const ttlRows = options.includeExpiry === false ? [] : await expiry.due(now, batchSize); - const remaining = batchSize - ttlRows.length; - const targetRows = remaining > 0 - ? await db.select().from(decisions) - .where(and(eq(decisions.status, "open"), ...(targetSweepCursor ? [gt(decisions.id, targetSweepCursor)] : []))) - .orderBy(asc(decisions.id)).limit(remaining) - : []; - targetSweepCursor = targetRows.length === remaining && targetRows.length > 0 ? targetRows[targetRows.length - 1]!.id : null; - const rows = [...new Map([...ttlRows, ...targetRows].map((row) => [row.id, row])).values()]; let expired = 0; - for (const decision of rows) { const strictTargetIds = new Set(decision.options.flatMap((option) => option.effects.filter((effect) => effect.staleness === "strict").flatMap(decisionEffectTargetIssueIds))); - const targets = strictTargetIds.size > 0 - ? await db.select({ id: issues.id, status: issues.status }).from(issues).where(and(eq(issues.companyId, decision.companyId), inArray(issues.id, [...strictTargetIds]))) - : []; - const targetGone = targets.length !== strictTargetIds.size || targets.some((target) => target.status === "cancelled"); - // A decision whose strict targets all completed after it was proposed is moot: - // the strict guard would skip its effects anyway, so retire it instead of - // leaving it pending until TTL. Targets that were already done at proposal - // time don't count — those decisions intentionally act on a done issue. - const snapshots = decision.targetSnapshots as Record; - const targetsCompleted = !targetGone && strictTargetIds.size > 0 && - targets.every((target) => target.status === "done") && - targets.some((target) => snapshots[target.id]?.status !== "done"); - if (!targetGone && !targetsCompleted && (options.includeExpiry === false || decision.expiresAt > now)) continue; - const reason = targetGone ? "target_gone" : targetsCompleted ? "target_completed" : "ttl"; - expired += await expiry.expire(decision, reason, now); - } - return { expired, resumed }; - } - - return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep, + return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepContinuations: continuations.sweep, sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next, - recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next }; + recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next, + sweepInvalidTargets: targets.sweep, decisionTargetsChanged: targets.changed, hasDecisionTargetWork: targets.hasPending }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 908be12ff8d..899ca920654 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + decisionTargets: "decision-targets", decisionEffectRecovery: "decision-effect-recovery", decisionExpiry: "decision-expiry", decisionContinuation: "decision-continuation", diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 391d08c660f..1e77167e85d 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,4 +1,6 @@ +import { notifyDecisionTargetWork, notifyDecisionTargetIssueRemoval } from "./decision-targets.js"; import { assertOnboardingFirstTaskCreationAllowed } from "./onboarding-first-task-admission.js"; + import { resolveConnectionIntentContinuationSource, resolveItemVerdictContinuationWake } from "./chat-interaction-continuation.js"; import { insertIssueComment } from "./issue-comment-writes.js"; import { measureRunTransaction } from "./execution-transaction-timing.js"; @@ -11384,6 +11386,9 @@ export function issueService(db: Db) { (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined))) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); } + if (patch.status === "done" || patch.status === "cancelled") { + await notifyDecisionTargetWork(tx, `issue:${id}`); + } const updated = await tx .update(issues) .set(patch) @@ -11791,6 +11796,7 @@ export function issueService(db: Db) { try { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await requestRestorePointDeletion(tx, { sourceIssueId: id }); + await notifyDecisionTargetIssueRemoval(tx, id); removedIssue = await tx .delete(issues) .where(eq(issues.id, id)) diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index a5384e1fc71..50c200490b7 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1,3 +1,4 @@ +import { notifyDecisionTargetIssueRemoval } from "./decision-targets.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { verifyAppWebhook } from "./app-webhook.js"; import crypto from "node:crypto"; @@ -1964,6 +1965,7 @@ export function routineService( return updated ?? createdRun; } catch (error) { if (createdIssue) { + await notifyDecisionTargetIssueRemoval(txDb, createdIssue.id); await txDb.delete(issues).where(eq(issues.id, createdIssue.id)); } const failureReason = error instanceof Error ? error.message : String(error);