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 ec79f7b3cc0..54cbb1e16ec 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -3306,3 +3306,27 @@ stacked on its parent. Full-stack master propagation and fresh acceptance remain required. No merge or deployed host-sleep claim is made. See the PR for final verification and review evidence; a local test pass does not close earlier stack gate or suppression debts. + + +## Slice 56 gates; slice 57 Muse terminal settlement + +Slice 56 [#17772](https://github.com/paperclipai/paperclip/pull/17772), at +`4d1bc93840ae4bd238814d73249325246f69d76e`, has a successful authoritative Actions +run and Greptile 5/5 with zero unresolved threads. Its external Entire Gates +check still requires trail approval and draft assessment. The superseded run's +cancelled checks are not a new code failure. Do not call all gates green. + +Slice 57, `codex/sleep-57-muse-terminal-settlement`, extends the existing Muse +maintenance worker with original-binding terminal debt and scoped transaction +wakes. No new table or independent timer is added. The +[producer and lifecycle audit](2026-10-26-muse-terminal-settlement-sleep.md) +records 433 passing affected cases and workspace typechecking, including the +real finalization serialization regression and its fix. Periodic finalization +delegates only this duty; default live/manual and resource callbacks remain. +Build, publication and remote acceptance are still pending. + +Master was fetched again at `a67a18c171816b097f46eae78fc58c7e55c9c565`. +Whole-stack master integration, older failed gates and suppression audits, and +all remaining inventory reasons remain active. Nothing was merged. Existing +active-transport replay polling remains a separate reason to inventory and +migrate, rather than a false claim of whole-instance quiet. diff --git a/doc/plans/2026-10-26-external-conversation-settlement-sleep.md b/doc/plans/2026-10-26-external-conversation-settlement-sleep.md index e8f04ad2d43..6b76ee5e6e9 100644 --- a/doc/plans/2026-10-26-external-conversation-settlement-sleep.md +++ b/doc/plans/2026-10-26-external-conversation-settlement-sleep.md @@ -251,7 +251,8 @@ selection. Coverage includes the full settlement, queue, monitor scheduling, run-state, goal, decision-target, interaction, task Retry, projection, result commit, workspace, process recovery, and review consumers. The 38 selected test files restore to their exact committed hashes after temporary fixture imports. -Prepared fixtures use PostgreSQL 17 and the 413-migration schema; helpers create +Prepared fixtures use PostgreSQL 17 and 412 migration receipts through journal tag +0413_tired_kronos; helpers create additional fresh per-call fixtures where required. Failure accounting: the initial 35-file command passed 2,445 cases and timed out diff --git a/doc/plans/2026-10-26-muse-terminal-settlement-sleep.md b/doc/plans/2026-10-26-muse-terminal-settlement-sleep.md new file mode 100644 index 00000000000..2dbb802584f --- /dev/null +++ b/doc/plans/2026-10-26-muse-terminal-settlement-sleep.md @@ -0,0 +1,240 @@ +# Muse terminal assignment settlement: sleep migration + +Status: slice 57 in progress, stacked on external conversation settlement #17772 +at `4d1bc93840ae4bd238814d73249325246f69d76e`. This document is an implementation +and verification record, not a claim that the migration or deployed sleep is done. + +## Obligation and owner + +Muse's external-agent module owns assignments, operation receipts, binding +generations, and worker/native-effect holds. Terminal run projection does not +by itself establish that a private worker stopped or that an unknown effect +completed. Existing rows remain durable truth; no new table is required. + +The existing Muse maintenance worker also owns qualification deadlines, OAuth +authority reconciliation, and receiver lifecycle. Extend that worker through the +shared delivery coordinator. Do not add a competing timer or terminal worker. +The periodic native-finalizer path now delegates this duty to that worker. +Live heartbeat and default manual finalization retain their existing callbacks. + +## Initial evidence and changes + +Three regressions failed before implementation: + +1. A running accepted assignment armed the maintenance worker every five seconds. +2. A real committed Muse finish receipt emitted no maintenance notification. +3. Repeating terminal reconciliation rewrote settled assignment history, observed + through the PostgreSQL row version even though the visible values were equal. + +The owner now supplies one actionable-terminal predicate for discovery and locked +settlement. It includes a terminal live assignment requiring fencing, an accepted +settled result whose finalization timestamp differs from the run, and an +unreleased hold on which the owner still owes safe release or fencing. A fenced +private-worker hold awaiting external evidence is not automatic retry debt. +Null timestamps do not become invented completion times. New evidence can make +that row actionable later. + +Discovery joins the recorded binding ID/generation, company, agent, and run. +Revoked bindings are included when they still contain actionable work. New +pairing creates a new binding row/generation; current agent configuration does +not replace a prior assignment's execution contract. + +Settlement retains the existing admission-guard/binding lock order. Maintenance +processes at most 100 assignments per binding transaction; explicit manual +reconciliation retains its unbounded default. It does not rewrite unchanged +finalization timestamps. Terminal reconciliation fences the specific assignment; +explicit connection revocation keeps its binding-wide fence. This prevents a +historical result from fencing unrelated successor work. + +Committed native operation results notify the existing maintenance topic before +the receipt/assignment/hold writes. Replay of an already-saved identical outcome +remains a no-op. This is an owner-local protocol change, not provider parsing in +shared startup or heartbeat code. + +Maintenance retains scoped hints across failed sweeps and drain. Scoped terminal +hints inspect the affected run/binding/company; global startup and authority +signals retain their existing qualification/OAuth work. Concrete debt remaining +after a bounded batch earns another retry. Running assignments and completed +history do not. Passive pending checks use memory, and qualification deadlines +remain actual persisted deadlines. + +## Implementation and acceptance checklist + +- [x] Reproduce active-assignment polling, missing finish notification, and history + rewrite against the old implementation. +- [x] Add owner predicate, bounded settlement, unchanged-history avoidance, and + actual finish/result notification. +- [x] Extend the existing worker with scoped hints and concrete-debt retries. +- [x] Preserve and test current qualification, OAuth, startup, retry, drain/resume, + and idle maintenance behavior (14 lifecycle tests). +- [x] Complete the producer audit: native run terminal/projection writes, late + operation results, rejected or unknown sends, controller attach/revoke, + replacement/revocation, and hold/attestation transitions. +- [x] Wire terminal run/projection notifications through existing owner contracts. + Keep Muse tables and profiles out of generic heartbeat policy. +- [x] Wire the existing restore/worktree suppression check into terminal work, + preserving qualification and authority duties; reactivate on settings changes. +- [x] Add real worker/broker integration for both result/run orderings, original + revoked/replaced bindings, independent worker/native-effect holds, and scope + isolation. Do not assume that a late result can restore already-fenced authority. +- [x] Add a regression proving historical terminal reconciliation cannot fence + successor work, plus retained-history/no-configuration SQL quiet and first + relevant producer reactivation. +- [x] Verify outer commit/rollback, concurrent hints during a sweep, failed scoped + retry through drain, startup reconstruction, pagination, and joined shutdown. +- [x] Delegate both periodic native-finalizer callback branches while preserving + default live/manual callbacks and unrelated resource settlement. +- [ ] Run broader affected tests, required checks, PR review, and fresh-head CI. + +## Producer and boundary audit + +All notifications register intent before the write in the existing transaction. +No caller owns a new recovery wrapper or a table. The external-agent public +module supplies the notification contract; Muse owns its durable predicates. + +| Producer | Existing transaction and new wake | Scope / boundary | +| --- | --- | --- | +| Run lifecycle | Native terminal status and cancellation/failure helpers | Recorded company and run | +| Native committed projection | Actual repair CAS, after finding a repairable row | Recorded company and run; no hint for unchanged projection | +| Native finalization | Existing finalization fanout, live committed projection, exhausted/blocked finalization | Company/run/issue identifiers only | +| Native recovery/execution | Stranded failure, restart failure, executor terminal failure, cancelled continuation | Existing native ownership checks stay in force | +| Execution-control recovery | Terminal failure after its ownership checks | Recorded company and run | +| Stale-run recovery and watchdog | Eligible run terminalization / resolved-source fold | Existing CAS and runtime-settlement guards stay in force | +| Queue/run dispatch | Stale retry, discarded queued comment, superseded handoff, stopped recovery, and reassigned issue cancellation | Existing transaction; exact affected run | +| Muse native result | `port.settle` before receipt, assignment, and independent hold writes | Original binding; identical stored outcome returns before notification | +| Muse send failure | Rejected send gets a transaction; unknown-send uses the existing admission guard and binding lock | Exact original binding; actual CAS still decides whether to reopen an effect hold | +| New Muse assignment | Existing dispatch transaction | Binding-scoped, rather than all-binding scan | +| Qualification, pairing, revoke, OAuth | Existing authority transaction | Explicit `*` survives coalescing with scoped writes in the same outer transaction | +| Worktree execution setting | Existing settings transaction | Global reconstruction after reactivation | + +Controller attach runs under execution admission and the binding lock, with +controller ownership on a running run. Its pending-to-unknown transition keeps +an unresolved operation unresolved. A later terminal writer creates the +settlement wake; attach does not create a new terminal-work clock. Operator +attestation and revocation own their direct hold changes. A fenced worker waiting +for private stop evidence remains visible but is not an automatic retry debt. +Legacy-only status recovery and nonterminal progress/usage/error metadata writes +do not create this terminal obligation. + +The notification serializer explicitly copies four IDs. TypeScript structural +callers can pass a larger object, including a database handle; serializing the +whole object broke real finalization with a circular-JSON error. The actual +native-finalizer regression caught this integration defect before publication. + +## Verification record and limits + +At the integrated source checkpoint, 433 cases pass across six suites: 17 new +real broker/worker lifecycle cases, 256 Muse control-plane cases, 14 existing +maintenance cases, 28 native recovery-notice/finalizer cases, 82 startup cases, +and 36 native-finalization recovery cases. Workspace typechecking passes. +Build and remote CI/review acceptance remain pending. + +The 17 lifecycle cases cover configured-idle SQL silence, the actual terminal +writer and committed projection, outer commit/rollback, first wake during drain, +real worktree/restore suppression, settings activation, scoped isolation, mixed +global/scoped writes, failure/restart recovery, concurrent hints, joined shutdown, +late native-result versus private-worker uncertainty, revoked original binding, +historical failed work versus a successor, and 101 saved assignments. The last +case also proves explicit manual reconciliation retains its unbounded default. +Four real-finalizer cases verify both newly committed and already committed +branches with delegation on/off, and preserve resource settlement callbacks. + +Failure evidence is retained in local logs: + +- `sleep57-lifecycle-first.log`: the fixture left its independent active-transport + replay poller attached. Closing that transport before measuring this worker + made the eight original lifecycle cases pass. The transport poller is a + separate duty, not silently removed or excluded from SQL observation. +- `sleep57-lifecycle-expanded.log`: the failure/drain assertion checked the idle + hold before the coordinator's next reconciliation. Advancing that existing + recovery deadline correctly tests release and SQL silence during drain. +- `sleep57-delegation-red.log`: real finalization exposed the broad serializer. +- `sleep57-delegation-red-after-scope-fix.log`: the two delegated cases failed + because periodic callbacks still ran; the two default cases passed. +- `sleep57-affected-full.log`: five suites / 397 cases passed. The sixth suite + failed setup because a reused database contained its fixed IDs. A fresh, + migrated dedicated database passed all 36 cases (`sleep57-finalization-fresh.log`). +- `sleep57-workspace-types.log`: workspace typecheck passed. + +Tests use real PostgreSQL and transport authority with synthetic provider +receipts. They do not qualify a live Muse worker. The broader local full-suite +limitation from embedded PostgreSQL semaphore capacity remains; these runs use +separate databases on the already running test server. No additional IPC cleanup +was performed or authorized. + +Only the periodic external-assignment callback is delegated. Shared resource +callbacks, live heartbeat settlement, and default manual settlement remain. +The worker uses existing durable rows and process-local transaction signals; +startup reconstructs missed work. It does not add cross-replica SQL ingress. +Cloud's unsupported-work sleep blockers and external-alarm admission contract +remain unchanged. This slice establishes local worker quiet, not deployed host +suspension, database suspension, or measured cost savings. + +## Qualification deadline precedence + +When a bounded settlement pass leaves assignments outstanding, maintenance now +registers the earlier of its five-second settlement retry and the saved +qualification expiry. Neither obligation can hide the other. A regression with +real qualification and assignment rows first reproduced a one-second expiry +being delayed to the settlement retry. It now proves the exact expiry is +registered and enforced while assignment debt remains, then disarms after +settlement with no recurring SQL. All 15 maintenance lifecycle cases passed. +Integrated typecheck, build, and remote checks remain required. + +## Current master integration and quiet producers (2026-10-29) + +Production source `584015018ed4a9d57d2b0f192b75847503b54e53` contains reviewed parent `b256bb52647affd9f5279ea8c40f48bbe40f031f`. +The parent is merged as `4d0847c4ca2912af72506e3374e34536b4f39943`. A normal merge adopts its actual squash +with the complete accepted child tree unchanged. Final publication adds only +audit text, a parent migration-count clarification, and the proved Photon regression extension; production stays byte +for byte identical to the accepted source. + +The composition proof preserves 1,215 parent-only paths, 14 owned-only paths, +11,515 identical paths, and 16 automatic shared results. The only manual +resolution adds Muse settlement inside the existing ordinary issue/run guard +in native finalization notifications. Packages equal the parent, including +current Copilot profile 80 and preserved historical profile fixtures. + +A real Muse regression proves why the guard matters. Three routine company-only +finalization hints cause six settlement candidate SELECTs when only the new +Muse fanout is moved outside the guard. That controlled baseline fails the quiet +assertion. With the guard retained, the same case passes with zero candidate +SELECTs; a real terminal run change still finalizes the assignment and returns +to quiet. The test hash is identical across the red/green pair, and the temporary +production change restores exactly. Real company or binding changes retain +their existing global/scoped signals. No new table, poll, or producer wrappers +were added. + +The parent review also raised a Photon per-event settlement warning. Real +callbacks disprove that finding. The callback locks and rereads the endpoint +and accepts only attention status before notifying a recovery transition. Five +active assertions before and five after recovery emit zero settlement hints and +candidate SQL. Three concurrent attention assertions emit exactly one hint and +one candidate SELECT, then the endpoint is active. The complete 24-case Photon +file passes on production source `584015018ed4a9d57d2b0f192b75847503b54e53`. The relevant Photon and settlement +production files equal the parent's reviewed source byte for byte. This slice +keeps the fourth queue in the existing parameterized regression. The parent's +security service accepted the resolved finding and returned a clean exact-head +check before its protected merge; no security check was waived. + +Acceptance covers 3,715 distinct cases across 52 complete server test files, plus +20 targeted chat integration cases (3,735 selected cases total). The 49-file +main batch passes 3,008 cases in 474.27 seconds. Separate process recovery passes +410, review response passes 273, and the Photon file passes 24. The targeted chat +selection includes committed-response recovery and both passive/missing GitHub +idle cases from the earlier remote failure; 1,794 other chat cases are excluded. +The 52 original selected test files restore exactly after fixture imports; the +additional Photon test is bound to its passing proof by SHA-256. + +Database build, server production typecheck/build, module boundaries, migration +order, and profile identity pass. Fixtures use 34 prepared fresh PostgreSQL 17 +databases with 412 applied migration receipts through journal tag 0413_tired_kronos, plus fresh helper-owned databases. + +Failure records remain: the controlled unscoped Muse baseline failed as +expected; the earlier remote chat passive-idle failure is preserved and is +covered by the new targeted selection. The parent's original 15-second handoff +timeout and fresh unchanged 89-case pass remain in its audit. No assertion or +timeout changed. Fresh publication-head CI and exact-head Greptile 5/5 remain +required. This is not full repository or full chat-suite acceptance, live Muse or +Copilot qualification, deployed sleep, database suspension, external alarm +coverage, or a fleet cost claim. Existing Cloud sleep blockers remain. diff --git a/server/src/__tests__/muse-control.test.ts b/server/src/__tests__/muse-control.test.ts index c93fbf5b19e..08944e96deb 100644 --- a/server/src/__tests__/muse-control.test.ts +++ b/server/src/__tests__/muse-control.test.ts @@ -2134,6 +2134,37 @@ describe("personal Muse control plane",()=>{ }finally{await detach();} }); + it("notifies maintenance when a real Muse finish receipt commits", async () => { + const f = await active(); + const events: string[] = []; + let unsubscribe = () => {}; + try { + await f.broker.act(f.subject, { version: 1, command: "accept", assignmentId: f.assignment.id, requestId: randomUUID() }); + unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.museMaintenance, event => events.push(event)); + await f.broker.act(f.subject, { version: 1, command: "finish", assignmentId: f.assignment.id, requestId: randomUUID(), result: {} }); + expect(events).toContain("intent"); + expect(events.at(-1)).toBe("settled"); + expect((await db.select().from(assignments).where(eq(assignments.id, f.assignment.id)))[0].acceptedResultAt).not.toBeNull(); + } finally { unsubscribe(); await f.detach(); } + }); + + it("does not rewrite settled Muse terminal history", async () => { + const f = await active(); + try { + await f.broker.act(f.subject, { version: 1, command: "accept", assignmentId: f.assignment.id, requestId: randomUUID() }); + await f.broker.act(f.subject, { version: 1, command: "finish", assignmentId: f.assignment.id, requestId: randomUUID(), result: {} }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + await f.broker.reconcileTerminalAssignments(f.company.id, f.agent.id, f.binding.id); + const version = async () => (await db.select({ version: sql`${assignments}.xmin::text` }) + .from(assignments).where(eq(assignments.id, f.assignment.id)))[0].version; + const settled = await version(); + await f.broker.reconcileTerminalAssignments(f.company.id, f.agent.id, f.binding.id); + expect(await version()).toBe(settled); + const [hold] = await db.select().from(externalAgentHolds).where(eq(externalAgentHolds.assignmentId, f.assignment.id)); + expect(hold).toMatchObject({ workerUnknown: false, nativeEffectsUnknown: false, releasedAt: expect.any(Date) }); + } finally { await f.detach(); } + }); + it("makes identical native result replay idempotent after finish and finalization",async()=>{ const f=await active();try { const acceptId=randomUUID(),finishId=randomUUID(); diff --git a/server/src/__tests__/muse-terminal-settlement.test.ts b/server/src/__tests__/muse-terminal-settlement.test.ts new file mode 100644 index 00000000000..710c3746c07 --- /dev/null +++ b/server/src/__tests__/muse-terminal-settlement.test.ts @@ -0,0 +1,383 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, authUsers, companies, companyMemberships, createDb, closeRegisteredClients, + externalAgentHolds, heartbeatRuns, issues, completionContracts, nativeRunResults, nativeRunFinalizations, museAgentBindings as bindings, + museRunnerAssignments as assignments, museRunnerOperations as operations, type Db } from "@paperclipai/db"; +import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js"; +import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { museIdentity, museCredentialHash } from "../services/muse-identity.js"; +import { museRunnerBroker } from "../services/muse-runner-broker.js"; +import { startMuseMaintenance } from "../services/muse-maintenance.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { setTimeout as delay } from "node:timers/promises"; +import { externalOperationDigest, type ExternalProviderOperation } from "../vendor/paperclip-runner/index.js"; + + + +describe("Muse terminal settlement lifecycle", () => { + let db: ReturnType; + let temporary: Awaited> | undefined; + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MUSE_SETTLEMENT_TEST_DATABASE_URL"); + let coordinator: ReturnType | undefined; + let stop: ReturnType | undefined; + let admitted = true, recover = true; + const errors: unknown[] = [], detachers: Array<() => Promise> = []; + beforeEach(() => { + admitted = true; recover = true; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("muse-terminal-settlement-"); + db = createDb(external ?? temporary!.connectionString); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableMuse: true }); + }, 30_000); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + await stop?.(); stop = undefined; + for (const detach of detachers.splice(0)) await detach(); + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`TRUNCATE TABLE companies CASCADE`); + }); + async function start(canSettle: () => Promise = async () => recover, database: Db = db) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, + canReconcile: () => true, onError: error => errors.push(error) }); + stop = startMuseMaintenance(database, coordinator, canSettle); + await stop.ready; + } + async function idle() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(0); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + } + async function quiet() { + const previous = db.$client.options.debug, queries: string[] = []; + db.$client.options.debug = (_id, query) => queries.push(query); + try { await vi.advanceTimersByTimeAsync(600_000); await delay(10); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previous; } + } + const lifecycle = (writer: Db = db) => createHeartbeatLifecycle(writer, + { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies); + const readAssignment = async (id: string) => (await db.select().from(assignments).where(eq(assignments.id, id)))[0]; + const readHold = async (id: string) => (await db.select().from(externalAgentHolds).where(eq(externalAgentHolds.assignmentId, id)))[0]; + async function acceptedResult() { + const f = await active(); + await f.broker.act(f.subject, { version: 1, command: "accept", assignmentId: f.assignment.id, requestId: randomUUID() }); + await f.broker.act(f.subject, { version: 1, command: "finish", assignmentId: f.assignment.id, requestId: randomUUID(), result: {} }); + + + await f.detach(); + return f; + } + afterAll(async () => { await closeRegisteredClients(external ?? temporary!.connectionString); await temporary?.cleanup(); }); + async function fixture(ready=true) { + const operatorId=randomUUID();await db.insert(authUsers).values({id:operatorId,name:"Muse authorizer",email:operatorId+"@example.test",createdAt:new Date(),updatedAt:new Date()}); + const [company]=await db.insert(companies).values({name:"Muse control",issuePrefix:"MU"+operatorId.replaceAll("-","")}).returning(); + const [agent]=await db.insert(agents).values({companyId:company.id,name:"Muse",role:"engineer",status:"active",adapterType:"paperclip_runner",adapterConfig:{provider:"muse",allowUnmeteredProvider:true,lifecycle:"per_turn"},permissions:{canCreateTasks:true}}).returning(); + await db.insert(companyMemberships).values([{companyId:company.id,principalType:"user",principalId:operatorId,membershipRole:"owner",status:"active"},{companyId:company.id,principalType:"agent",principalId:agent.id,membershipRole:"member",status:"active"}]); + const ticket=randomBytes(32).toString("base64url");const [binding]=await db.insert(bindings).values({companyId:company.id,agentId:agent.id,operatorId,ticketHash:museCredentialHash(ticket),ticketExpiresAt:new Date(Date.now()+600000)}).returning(); + await db.update(agents).set({adapterConfig:{...agent.adapterConfig,museBindingId:binding.id}}).where(eq(agents.id,agent.id)); + const identity=museIdentity(db),credentials=await identity.pair(ticket,"2"),broker=museRunnerBroker(db,{publicOrigin:"https://paperclip.example",heartbeat:{wakeup:vi.fn().mockResolvedValue(null),cancelRun:vi.fn()}}); + if(ready)await db.update(bindings).set({status:"ready",receiverContactAt:new Date(),verifiedReplyAt:new Date()}).where(eq(bindings.id,binding.id)); + return {company,agent,operatorId,binding,ticket,identity,credentials,broker,subject:await identity.authenticate(credentials.accessToken)}; + } + async function active(provided?:Awaited>,afterSettle?: (command:ExternalProviderOperation)=>Promise) { + const f=provided??await fixture(),runId=randomUUID(),session=randomUUID(),turnId="turn_"+randomUUID(); + const [issue]=await db.insert(issues).values({companyId:f.company.id,title:"Muse task",status:"in_progress",assigneeAgentId:f.agent.id,responsibleUserId:f.operatorId}).returning(); + await db.insert(heartbeatRuns).values({id:runId,companyId:f.company.id,agentId:f.agent.id,status:"running",runtimeMode:"native",driverKind:"muse_external",nativeIssueId:issue.id,nativeSessionId:session,nativePhase:"turn_running",responsibleUserId:f.operatorId}); + await db.update(issues).set({checkoutRunId:runId,executionRunId:runId}).where(eq(issues.id,issue.id)); + await db.insert(nativeRunFinalizations).values({runId,companyId:f.company.id,issueId:issue.id,phase:"running",controllerGeneration:1,leaseExpiresAt:new Date(Date.now()+600000)}); + const execution={binding:{companyId:f.company.id,agentId:f.agent.id,runId},provider:{binding:await f.broker.snapshot(f.company.id,f.agent.id,f.binding.id)}}; + const port=f.broker.port(execution),sent:ExternalProviderOperation[]=[]; + const detach=await port.attach(async command=>{sent.push(command);await port.settle({sourceEventId:"settle_"+command.requestId,payload:{binding:ref,requestId:command.requestId,outcome:command.action==="consume_input"?{status:"consumed",requestId:command.input.requestId,inputDigest:command.input.inputDigest}:{status:command.action==="accept"?"accepted":"completed"}}});await afterSettle?.(command);}); + const ref={...execution.provider.binding,runId,normalizedSessionId:session,turnId,assignmentRevision:1}; + await port.dispatch({sourceEventId:"offer_"+randomUUID(),payload:{kind:"assignment",binding:ref,tools:[],acceptByUnixMs:Date.now()+600000,expiresAtUnixMs:Date.now()+3600000}}); + const [assignment]=await db.select().from(assignments).where(eq(assignments.runId,runId)); + detachers.push(detach); + return {...f,runId,issue,port,ref,assignment,sent,detach}; + } + it("ignores routine company-only finalization hints and still settles a committed Muse run", async () => { + const f = await acceptedResult(); + await start(); await idle(); await quiet(); + const previous = db.$client.options.debug, queries: string[] = []; + db.$client.options.debug = (_id, query) => queries.push(query); + try { + for (let attempt = 0; attempt < 3; attempt++) { + await db.transaction(tx => notifyNativeFinalizationWork(tx, { companyId: f.company.id })); + await idle(); + } + expect(queries.filter(query => query.includes('"muse_runner_assignments"'))).toEqual([]); + } finally { db.$client.options.debug = previous; } + expect((await readAssignment(f.assignment.id)).finalizedAt).toBeNull(); + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it("settles Muse from the real terminal run producer after becoming idle", async () => { + const f = await acceptedResult(); + await start(); await idle(); await quiet(); + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect(await readHold(f.assignment.id)).toMatchObject({ workerUnknown: false, nativeEffectsUnknown: false, releasedAt: expect.any(Date) }); + expect(errors).toEqual([]); await quiet(); + }); + + it.each(["commit", "rollback"] as const)("respects the terminal writer's outer transaction (%s)", async outcome => { + const f = await acceptedResult(); await start(); await idle(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { written = resolve; }); + const transaction = db.transaction(async tx => { + await tx.transaction(async nested => { await lifecycle(nested as unknown as Db) + .setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); }); + written(); await gate; + if (outcome === "rollback") throw new Error("rollback fixture"); + }); + try { + await ready; expect((await readAssignment(f.assignment.id)).finalizedAt).toBeNull(); + release(); + if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback fixture"); + else await transaction; + await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt !== null).toBe(outcome === "commit"); + expect(errors).toEqual([]); await quiet(); + } finally { release(); await transaction.catch(() => {}); } + }); + + it("retains the first terminal wake through drain without passive SQL", async () => { + const f = await acceptedResult(); await start(); await idle(); admitted = false; + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + await vi.advanceTimersByTimeAsync(0); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect((await readAssignment(f.assignment.id)).finalizedAt).toBeNull(); + await quiet(); admitted = true; + await vi.advanceTimersByTimeAsync(5000); await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it("keeps terminal work suppressed until a committed reactivation", async () => { + const f = await acceptedResult(); + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + recover = false; await start(); await idle(); await quiet(); + expect((await readAssignment(f.assignment.id)).finalizedAt).toBeNull(); + recover = true; + await instanceSettingsService(db).updateExperimental({ enableWorktreeRunExecution: true }); + await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it.each(["restore", "worktree"] as const)("honors the real %s suppression gate", async mode => { + const f = await acceptedResult(); + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" } + : { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "muse-settlement-test" }; + const settings = instanceSettingsService(db, { runtimeEnv }); + if (mode === "worktree") await settings.updateExperimental({ enableWorktreeRunExecution: false }); + const heartbeat = heartbeatService(db, { runtimeEnv }); + await start(async () => { + const state = await heartbeat.resolveSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }); + await idle(); await quiet(); + expect((await readAssignment(f.assignment.id)).finalizedAt).toBeNull(); + if (mode === "worktree") { + try { + await settings.updateExperimental({ enableWorktreeRunExecution: true }); + await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + await quiet(); + } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); } + } + expect(errors).toEqual([]); + }); + + it("limits a committed terminal hint to its original scope", async () => { + const first = await acceptedResult(), other = await acceptedResult(); + await start(); await idle(); + + + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, other.runId)); + await lifecycle().setRunStatus(first.runId, "succeeded", { finishedAt: new Date() }); + await idle(); + expect((await readAssignment(first.assignment.id)).finalizedAt).not.toBeNull(); + expect((await readAssignment(other.assignment.id)).finalizedAt).toBeNull(); + await quiet(); + await coordinator!.stop(); await stop!(); await start(); await idle(); + expect((await readAssignment(other.assignment.id)).finalizedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it("preserves a global settings wake alongside scoped writes in one outer transaction", async () => { + const first = await acceptedResult(), other = await acceptedResult(); + await start(); await idle(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, other.runId)); + await db.transaction(async tx => { + await instanceSettingsService(tx as unknown as Db).updateExperimental({ enableWorktreeRunExecution: true }); + await lifecycle(tx as unknown as Db).setRunStatus(first.runId, "succeeded", { finishedAt: new Date() }); + }); + await idle(); + expect((await readAssignment(first.assignment.id)).finalizedAt).not.toBeNull(); + expect((await readAssignment(other.assignment.id)).finalizedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it("retains a failed scoped sweep through drain and reconstructs it after restart", async () => { + const f = await acceptedResult(); let fail = false; + const observed = new Proxy(db, { get(target, key) { + if (key === "selectDistinct" && fail) return () => { throw new Error("settlement discovery unavailable"); }; + const value = Reflect.get(target, key); + return typeof value === "function" ? value.bind(target) : value; + } }); + await start(undefined, observed); await idle(); fail = true; + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(0); expect(errors.length).toBeGreaterThan(0); }); + admitted = false; + + await vi.advanceTimersByTimeAsync(5000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect((await readAssignment(f.assignment.id)).finalizedAt).toBeNull(); await quiet(); + await coordinator!.stop(); await stop!(); errors.length = 0; admitted = true; + await start(); await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it.each([false, true])("retains a concurrent hint while settlement is in flight (shutdown: %s)", async shutdown => { + const first = await acceptedResult(), second = await acceptedResult(); + let block = false, entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + await start(async () => { if (block) { entered(); await gate; } return true; }); await idle(); + block = true; + await lifecycle().setRunStatus(first.runId, "succeeded", { finishedAt: new Date() }); + await vi.advanceTimersByTimeAsync(0); await ready; + await lifecycle().setRunStatus(second.runId, "succeeded", { finishedAt: new Date() }); + try { + if (shutdown) { + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(10); expect(stopped).toBe(false); + release(); await stopping; await stop!(); await start(); + } else { block = false; release(); } + await idle(); + expect((await readAssignment(first.assignment.id)).finalizedAt).not.toBeNull(); + expect((await readAssignment(second.assignment.id)).finalizedAt).not.toBeNull(); + + if (!shutdown) expect(errors).toEqual([]); + await quiet(); + } finally { release(); } + }); + + it("wakes from the actual committed run projection owner", async () => { + const f = await acceptedResult(); + + + const contractId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: f.company.id, issueId: f.issue.id, + revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "settlement-test", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "settlement-test", objective: "Finish", criteria: [{ id: "objective", requirement: "Complete the task" }] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ completionContractId: contractId }).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await db.insert(nativeRunResults).values({ companyId: f.company.id, issueId: f.issue.id, + runId: f.runId, completionContractId: contractId, serverFingerprint: randomUUID(), canonicalSha256: randomUUID(), + schemaStatus: "accepted", resultJson: { terminal: { runTerminalState: "succeeded" } } }).returning(); + const [finalization] = await db.update(nativeRunFinalizations).set({ resultId: result.id, phase: "committed" }) + .where(eq(nativeRunFinalizations.runId, f.runId)).returning(); + await start(); await idle(); await quiet(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + await projectCommittedRun({ db, run, coordinator: finalization }); + await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect((await readHold(f.assignment.id)).releasedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it("drains more than one maintenance batch while preserving unbounded manual settlement", async () => { + const f = await acceptedResult(); + const [template] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const assignment = await readAssignment(f.assignment.id); + const historical = Array.from({ length: 101 }, () => ({ runId: randomUUID(), assignmentId: randomUUID(), sessionId: randomUUID() })); + await db.insert(heartbeatRuns).values(historical.map(row => ({ ...template, id: row.runId, + nativeSessionId: row.sessionId, status: "succeeded", finishedAt: new Date() }))); + await db.insert(assignments).values(historical.map(row => ({ ...assignment, id: row.assignmentId, + runId: row.runId, normalizedSessionId: row.sessionId, turnId: randomUUID(), finalizedAt: null }))); + await start(); + const missing = () => db.select().from(assignments).where(sql`${assignments.id} in (${sql.join(historical.map(row => sql`${row.assignmentId}::uuid`), sql`,`)}) and ${assignments.finalizedAt} is null`); + expect(await missing()).toHaveLength(1); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(5000); await idle(); + expect(await missing()).toHaveLength(0); await quiet(); + await db.update(assignments).set({ finalizedAt: null }).where(eq(assignments.bindingId, f.binding.id)); + await f.broker.reconcileTerminalAssignments(f.company.id, f.agent.id, f.binding.id); + expect(await missing()).toHaveLength(0); + expect(errors).toEqual([]); await quiet(); + }); + + it("keeps a fenced worker held when a late native result resolves only its effect", async () => { + const f = await active(); + await f.broker.act(f.subject, { version: 1, command: "accept", assignmentId: f.assignment.id, requestId: randomUUID() }); + const requestId = randomUUID(), input = { result: {} }; + + + await db.insert(operations).values({ companyId: f.company.id, assignmentId: f.assignment.id, requestId, + digest: externalOperationDigest("finish", input), command: { ...f.ref, requestId, action: "finish", input, + digest: externalOperationDigest("finish", input) }, status: "pending" }); + await f.detach(); + await start(); await idle(); + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + await idle(); + expect((await readAssignment(f.assignment.id)).status).toBe("fenced"); + expect(await readHold(f.assignment.id)).toMatchObject({ workerUnknown: true, nativeEffectsUnknown: true, releasedAt: null }); + await f.port.settle({ sourceEventId: randomUUID(), payload: { binding: f.ref, requestId, outcome: { status: "completed" } } }); + await idle(); + expect((await readAssignment(f.assignment.id)).status).toBe("fenced"); + expect(await readHold(f.assignment.id)).toMatchObject({ workerUnknown: true, nativeEffectsUnknown: false, releasedAt: null }); + expect(errors).toEqual([]); await quiet(); + }); + + it("reconstructs timestamp settlement on an original revoked binding", async () => { + const f = await acceptedResult(); + + + await db.update(externalAgentHolds).set({ workerUnknown: false, releasedAt: new Date() }) + .where(eq(externalAgentHolds.assignmentId, f.assignment.id)); + await f.broker.revoke(f.company.id, f.agent.id, f.operatorId); + await db.update(agents).set({ adapterConfig: { provider: "codex" } }).where(eq(agents.id, f.agent.id)); + await lifecycle().setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + await start(); await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect((await readAssignment(f.assignment.id)).status).toBe("settled"); + expect((await readHold(f.assignment.id)).releasedAt).not.toBeNull(); + expect(errors).toEqual([]); await quiet(); + }); + + it("does not fence successor work while repairing historical failed settlement", async () => { + const f = await acceptedResult(); + + await db.update(externalAgentHolds).set({ workerUnknown: false, releasedAt: new Date() }) + .where(eq(externalAgentHolds.assignmentId, f.assignment.id)); + await lifecycle().setRunStatus(f.runId, "failed", { finishedAt: new Date() }); + const successor = await active(f); + await successor.detach(); + await start(); await idle(); + expect((await readAssignment(f.assignment.id)).finalizedAt).not.toBeNull(); + expect((await readAssignment(successor.assignment.id)).status).toBe("offered"); + expect(errors).toEqual([]); await quiet(); + }); +}); diff --git a/server/src/__tests__/native-recovery-notice-work.test.ts b/server/src/__tests__/native-recovery-notice-work.test.ts index b7cfac1cef6..6cf670d56d2 100644 --- a/server/src/__tests__/native-recovery-notice-work.test.ts +++ b/server/src/__tests__/native-recovery-notice-work.test.ts @@ -1,3 +1,4 @@ +import * as museTerminal from "../services/native-runtime/muse-terminal-reconciliation.js"; import * as conversationLifecycle from "../services/slack-conversation-lifecycle.js"; import { projectCommittedRun } from "../services/native-runtime/native-run-projection.js"; import { heartbeatService } from "../services/heartbeat.js"; @@ -285,6 +286,22 @@ describe("committed native recovery notice lifecycle", () => { await lifecycle.setRunStatus(binding.runId, "succeeded", { nativePhase: "committed", errorCode: null }); await flush(); expect((await action(pending.id)).status).toBe("resolved"); await quiet(); }); + it.each([ + { delegated: false, alreadyCommitted: false }, { delegated: true, alreadyCommitted: false }, + { delegated: false, alreadyCommitted: true }, { delegated: true, alreadyCommitted: true }, + ])("preserves resource settlement while delegating external assignments: %j", async ({ delegated, alreadyCommitted }) => { + const binding = await (alreadyCommitted ? committed() : accepted()); + const settleAssignment = vi.spyOn(museTerminal, "reconcileMuseNativeTerminalRun"); + const settled = vi.fn(async () => undefined); + await reconcileNativeFinalizations(db, [binding.runId], { + skipExternalAgentSettlement: delegated, + skipExternalConversationSettlement: true, skipNativeRecoveryNotices: true, + skipNativeResponsePresentation: true, skipNativeReviewResponse: true, onWorkspaceSettled: settled, + }); + expect(settleAssignment).toHaveBeenCalledTimes(delegated ? 0 : 1); + if (!delegated) expect(settleAssignment).toHaveBeenCalledWith(db, expect.objectContaining({ id: binding.runId, companyId })); + expect(settled).toHaveBeenCalledWith(expect.objectContaining({ runId: binding.runId, succeeded: true })); + }); it.each([true, false])("preserves resource settlement while delegating conversations: %s", async delegated => { const binding = await committed(); const settleConversation = vi.spyOn(conversationLifecycle, "settleExternalConversation"); diff --git a/server/src/__tests__/photon/channel.integration.test.ts b/server/src/__tests__/photon/channel.integration.test.ts index 47e93a2d90a..1ee511e82c6 100644 --- a/server/src/__tests__/photon/channel.integration.test.ts +++ b/server/src/__tests__/photon/channel.integration.test.ts @@ -1,3 +1,4 @@ +import { externalConversationSettlementWork } from "../../services/external-conversation-settlement-work.js"; import { nativeStatusReassessmentWork } from "../../services/native-runtime/native-status-reassessment-work.js"; import { nativeReviewResponseWork } from "../../services/native-runtime/native-review-response-work.js"; import { nativeResponsePresentationWork } from "../../services/native-runtime/native-response-presentation-work.js"; @@ -378,7 +379,7 @@ describe("iMessage Photon channel control plane", () => { }, }; } - it.each([DELIVERY_QUEUES.nativeResponsePresentation, DELIVERY_QUEUES.nativeReviewResponse, DELIVERY_QUEUES.nativeStatusReassessment])( + it.each([DELIVERY_QUEUES.nativeResponsePresentation, DELIVERY_QUEUES.nativeReviewResponse, DELIVERY_QUEUES.nativeStatusReassessment, DELIVERY_QUEUES.externalConversationSettlement])( "keeps repeated active Photon ownership assertions out of %s reconstruction and wakes once on recovery", async queue => { const t = await setup(); await t.start(); @@ -386,7 +387,8 @@ describe("iMessage Photon channel control plane", () => { const errors: unknown[] = []; const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); coordinators.push(coordinator); - const work = queue === DELIVERY_QUEUES.nativeResponsePresentation ? nativeResponsePresentationWork(db) + const work = queue === DELIVERY_QUEUES.externalConversationSettlement ? externalConversationSettlementWork(db) + : queue === DELIVERY_QUEUES.nativeResponsePresentation ? nativeResponsePresentationWork(db) : queue === DELIVERY_QUEUES.nativeReviewResponse ? nativeReviewResponseWork(db) : nativeStatusReassessmentWork(db); const worker = work.register(coordinator, () => true); const debug = db.$client.options.debug; @@ -403,7 +405,8 @@ describe("iMessage Photon channel control plane", () => { await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) .toEqual({ active: 0, next: null }); db.$client.options.debug = (_connection, query) => { - if (query.startsWith("select") && query.includes("native_run_finalizations")) queries.push(query); + if (query.startsWith("select") && (queue === DELIVERY_QUEUES.externalConversationSettlement + ? query.includes("exists (select 1 from chat_conversations c") : query.includes("native_run_finalizations"))) queries.push(query); }; const assertOwned = t.callbacks().onPhotonAssertOwned!; for (let i = 0; i < 5; i++) await assertOwned(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 767771808a4..65e59ea8884 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1090,7 +1090,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/app.ts b/server/src/app.ts index f4efdb2165a..a2ab7b062c3 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1195,7 +1195,10 @@ export async function createApp( publicMcpEvents?.start(deliveryWork); dotMcpEvents?.start(deliveryWork); if (publicMcpOAuth) grokDispatcher.start(deliveryWork); - const stopMuseMaintenance = startMuseMaintenance(db, deliveryWork); + const stopMuseMaintenance = startMuseMaintenance(db, deliveryWork, async () => { + const state = await connectionIntentHeartbeat.resolveSchedulingSuppression(true); + return !state.suppressed || state.reason === "task_drain"; + }); const resourceLifecycleDelivery = registerPluginLifecycleDelivery(deliveryWork, db, workerManager, lifecycle); await resourceLifecycleDelivery.ready; if (opts.feedbackExportService) { diff --git a/server/src/index.ts b/server/src/index.ts index 66656c3fecb..377e777d805 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -2020,7 +2020,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 94eb3b154bc..fee9d5510ee 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; @@ -342,6 +343,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch const transactionResult = await db.transaction(async (tx) => { await notifyInstructionCollection(tx, { companyId, runId: input.run.id }); + await notifyExternalAgentSettlement(tx, { companyId, runId: input.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId: input.sourceIssue.id }); await notifyNativeStatusReassessment(tx, { companyId, issueId: input.sourceIssue.id }); diff --git a/server/src/modules/external-agents/index.ts b/server/src/modules/external-agents/index.ts index a77e0701e53..606d549b843 100644 --- a/server/src/modules/external-agents/index.ts +++ b/server/src/modules/external-agents/index.ts @@ -2,3 +2,5 @@ export { museExternalAgents } from "./muse.js"; export { withExternalAdmissionGuard, withExternalExecutionAdmission, assertNoExternalOverlap, type ExternalAdmissionTransaction } from "./admission.js"; export { dotRunnerBroker, createDotRunnerMcpTools } from "./dot.js"; + +export { notifyExternalAgentSettlement, pendingMuseTerminalBindings, type ExternalAgentSettlementScope } from "./muse-terminal-settlement.js"; diff --git a/server/src/modules/external-agents/muse-terminal-settlement.ts b/server/src/modules/external-agents/muse-terminal-settlement.ts new file mode 100644 index 00000000000..b13af2a8599 --- /dev/null +++ b/server/src/modules/external-agents/muse-terminal-settlement.ts @@ -0,0 +1,48 @@ +import { and, asc, eq, gt, inArray, isNotNull, or, sql } from "drizzle-orm"; +import { heartbeatRuns, museAgentBindings as bindings, museRunnerAssignments as assignments, + museRunnerOperations as operations, externalAgentHolds as holds, type Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../services/delivery-work-notifications.js"; + +export type ExternalAgentSettlementScope = { companyId: string; bindingId?: string; runId?: string; issueId?: string }; + + + +export function notifyExternalAgentSettlement(tx: object, scope?: ExternalAgentSettlementScope) { + + + return notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, scope ? JSON.stringify({ + companyId: scope.companyId, bindingId: scope.bindingId, runId: scope.runId, issueId: scope.issueId, + }) : "*"); +} + + + +export function pendingMuseTerminalAssignmentCondition() { + const unknown = sql`exists(select 1 from ${operations} where ${operations.assignmentId} = ${assignments.id} + and ${operations.status} in ('unknown', 'dispatched', 'pending'))`; + const unsettledHold = sql`exists(select 1 from ${holds} where ${holds.provider} = 'muse' + and ${holds.assignmentId} = ${assignments.id} and ${holds.companyId} = ${assignments.companyId} + and ${holds.agentId} = ${assignments.agentId} and ${holds.releasedAt} is null + and (${heartbeatRuns.status} <> 'succeeded' or ${unknown} or not ${holds.nativeEffectsUnknown}))`; + return and(inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]), + or(inArray(assignments.status, ["offered", "claimed", "accepted"]), + and(eq(assignments.status, "settled"), isNotNull(assignments.acceptedResultAt), + or(sql`${assignments.finalizedAt} is distinct from ${heartbeatRuns.finishedAt}`, unsettledHold)))); +} + +export function pendingMuseTerminalBindings(db: Pick, + scope?: ExternalAgentSettlementScope, after?: string) { + return db.selectDistinct({ id: bindings.id, companyId: bindings.companyId, agentId: bindings.agentId }) + .from(bindings) + .innerJoin(assignments, and(eq(assignments.bindingId, bindings.id), eq(assignments.bindingGeneration, bindings.generation), + eq(assignments.companyId, bindings.companyId), eq(assignments.agentId, bindings.agentId))) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, assignments.runId), eq(heartbeatRuns.companyId, assignments.companyId), + eq(heartbeatRuns.agentId, assignments.agentId))) + .where(and(pendingMuseTerminalAssignmentCondition(), + scope ? eq(bindings.companyId, scope.companyId) : undefined, + scope?.bindingId ? eq(bindings.id, scope.bindingId) : undefined, + scope?.runId ? eq(assignments.runId, scope.runId) : undefined, + scope?.issueId ? eq(heartbeatRuns.nativeIssueId, scope.issueId) : undefined, + after ? gt(bindings.id, after) : undefined)) + .orderBy(asc(bindings.id)).limit(100); +} diff --git a/server/src/modules/external-agents/muse.ts b/server/src/modules/external-agents/muse.ts index adcc20f2e72..b370d0f0981 100644 --- a/server/src/modules/external-agents/muse.ts +++ b/server/src/modules/external-agents/muse.ts @@ -1,3 +1,4 @@ +import { pendingMuseTerminalAssignmentCondition } from "./muse-terminal-settlement.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../services/delivery-work-notifications.js"; import { createHash, randomBytes, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm"; @@ -104,8 +105,8 @@ function createMuseBroker(db:Db, publicOrigin?:string) { ||(company.budgetMonthlyCents>0&&company.spentMonthlyCents>=company.budgetMonthlyCents)||(agent.budgetMonthlyCents>0&&agent.spentMonthlyCents>=agent.budgetMonthlyCents)) throw forbidden("Native run no longer owns execution authority."); return {b,a,run}; } - async function fence(tx:ExternalAdmissionTransaction,b:typeof bindings.$inferSelect,reason:string) { - const live=await tx.select().from(assignments).where(and(eq(assignments.bindingId,b.id),eq(assignments.bindingGeneration,b.generation),sql`(${assignments.status} in ('offered','claimed','accepted') or (${assignments.status}='settled' and exists(select 1 from ${holds} where ${holds.assignmentId}=${assignments.id} and ${holds.provider}='muse' and ${holds.releasedAt} is null)))`)); + async function fence(tx:ExternalAdmissionTransaction,b:typeof bindings.$inferSelect,reason:string,assignmentId?:string) { + const live=await tx.select().from(assignments).where(and(eq(assignments.bindingId,b.id),eq(assignments.bindingGeneration,b.generation),assignmentId?eq(assignments.id,assignmentId):undefined,sql`(${assignments.status} in ('offered','claimed','accepted') or (${assignments.status}='settled' and exists(select 1 from ${holds} where ${holds.assignmentId}=${assignments.id} and ${holds.provider}='muse' and ${holds.releasedAt} is null)))`)); for(const a of live) { const [last]=await tx.select({id:operations.requestId}).from(operations).where(eq(operations.assignmentId,a.id)).orderBy(desc(operations.createdAt)).limit(1); const boundary:MuseStopBoundary={bindingId:b.id,generation:b.generation,assignmentId:a.id,runId:a.runId,turnId:a.turnId,assignmentRevision:a.revision,stopNonce:randomUUID(),operationBoundary:last?.id??null}; @@ -122,7 +123,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { await tx.update(mcpOauthTokens).set({usedAt:new Date(),expiresAt:new Date()}).where(eq(mcpOauthTokens.grantId,b.grantId)); } async function revokeBinding(tx:ExternalAdmissionTransaction,b:typeof bindings.$inferSelect,operatorId:string,reason:string,actorType:"user"|"system"="user") { - await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, "*"); const runs=await fence(tx,b,reason); await tx.update(bindings).set({status:"revoked",revokedAt:new Date(),ticketHash:null,challengeHash:null,cleanupExpiresAt:new Date(Date.now()+MUSE_CLEANUP_TTL_MS),deadlineEnforcedAt:b.deadlineEnforcedAt??(b.qualificationExpiresAt&&b.qualificationExpiresAt<=new Date()?new Date():null),revision:b.revision+1,updatedAt:new Date()}).where(eq(bindings.id,b.id)); await tx.update(credentials).set({revokedAt:new Date()}).where(and(eq(credentials.bindingId,b.id),inArray(credentials.kind,["access","refresh","signal"]))); @@ -180,7 +181,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { await tx.update(credentials).set({expiresAt:new Date(Date.now()+MUSE_CLEANUP_TTL_MS)}).where(and(eq(credentials.bindingId,old.id),inArray(credentials.kind,["cleanup","detector_cleanup"]),isNull(credentials.revokedAt))); } else if(input.replaceBindingId)throw conflict("Invitation changed. Refresh before replacing its ticket."); const [latest]=await tx.select({generation:bindings.generation}).from(bindings).where(and(eq(bindings.companyId,input.companyId),eq(bindings.agentId,input.agentId))).orderBy(desc(bindings.generation)).limit(1); - await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, "*"); const [created]=await tx.insert(bindings).values({id:bindingId,companyId:input.companyId,agentId:input.agentId,operatorId:input.operatorId,generation:(latest?.generation??0)+1,ticketHash:museCredentialHash(ticket),ticketExpiresAt:new Date(Date.now()+MUSE_TICKET_TTL_MS)}).returning(); await logActivity(tx as unknown as Db,{companyId:input.companyId,actorType:"user",actorId:input.operatorId,action:"muse.pairing_created",entityType:"agent",entityId:input.agentId,details:{bindingId:created!.id,generation:created!.generation}}); await updateAgentConfigurationInTransaction(tx as unknown as Db,agent.id,{adapterConfig:{...agent.adapterConfig,museBindingId:created!.id}},{recordRevision:{createdByUserId:input.operatorId,source:"muse-pairing"}}); @@ -288,17 +289,23 @@ function createMuseBroker(db:Db, publicOrigin?:string) { for(const publication of publications){try{publishActivity(publication);}catch{}} return result; }, - async reconcileTerminalAssignments(companyId:string,agentId:string,bindingId:string) { + async reconcileTerminalAssignments(companyId:string,agentId:string,bindingId:string,options?:{limit:number}) { await db.transaction(tx=>withExternalAdmissionGuard(tx,companyId,agentId,async()=>{ const [b]=await tx.select().from(bindings).where(and(eq(bindings.id,bindingId),eq(bindings.companyId,companyId),eq(bindings.agentId,agentId))).for("update");if(!b)return; - const terminal=await tx.select({a:assignments,run:heartbeatRuns}).from(assignments).innerJoin(heartbeatRuns,eq(heartbeatRuns.id,assignments.runId)).where(and(eq(assignments.bindingId,b.id),inArray(heartbeatRuns.status,["succeeded","failed","cancelled","timed_out","interrupted"]))); + const candidates=tx.select({a:assignments,run:heartbeatRuns}).from(assignments) + .innerJoin(heartbeatRuns,and(eq(heartbeatRuns.id,assignments.runId),eq(heartbeatRuns.companyId,assignments.companyId),eq(heartbeatRuns.agentId,assignments.agentId))) + .where(and(eq(assignments.bindingId,b.id),eq(assignments.bindingGeneration,b.generation), + eq(assignments.companyId,b.companyId),eq(assignments.agentId,b.agentId),pendingMuseTerminalAssignmentCondition())) + .orderBy(asc(assignments.id)); + const terminal = await (options ? candidates.limit(options.limit) : candidates); for(const {a,run}of terminal) { if(a.status==="settled"&&a.acceptedResultAt) { - await tx.update(assignments).set({finalizedAt:run.finishedAt}).where(eq(assignments.id,a.id)); + if(a.finalizedAt?.getTime()!==run.finishedAt?.getTime()) + await tx.update(assignments).set({finalizedAt:run.finishedAt}).where(eq(assignments.id,a.id)); const [unknown]=await tx.select({id:operations.id}).from(operations).where(and(eq(operations.assignmentId,a.id),inArray(operations.status,["unknown","dispatched","pending"]))).limit(1); - if(run.status!=="succeeded"||unknown) {await fence(tx,b,"run_terminal_after_result");continue;} + if(run.status!=="succeeded"||unknown) {await fence(tx,b,"run_terminal_after_result",a.id);continue;} if(!unknown&&run.status==="succeeded")await tx.update(holds).set({workerUnknown:false,releasedAt:new Date(),updatedAt:new Date()}).where(and(eq(holds.provider,"muse"),eq(holds.assignmentId,a.id),eq(holds.nativeEffectsUnknown,false),isNull(holds.releasedAt))); - } else if(["offered","claimed","accepted"].includes(a.status)) await fence(tx,b,"run_terminal"); + } else if(["offered","claimed","accepted"].includes(a.status)) await fence(tx,b,"run_terminal",a.id); } })); }, @@ -316,7 +323,7 @@ function createMuseBroker(db:Db, publicOrigin?:string) { if(b.qualificationId)throw conflict("Another qualification already exists."); await assertMuseCurrentAuthority(tx,b); const startedAt=new Date(input.expiresAt.getTime()-24*60*60_000); - await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, "*"); await tx.update(bindings).set({qualificationId:input.qualificationId,qualificationStartedAt:startedAt,qualificationExpiresAt:input.expiresAt,revision:b.revision+1,updatedAt:new Date()}).where(eq(bindings.id,b.id)); await logActivity(tx as unknown as Db,{companyId:input.companyId,actorType:"user",actorId:input.operatorId,agentId:input.agentId,action:"muse.qualification_started",entityType:"agent",entityId:input.agentId,details:{bindingId:b.id,generation:b.generation,revision:b.revision+1,authorizingUserId:b.operatorId,qualificationId:input.qualificationId,startedAt:startedAt.toISOString(),expiresAt:input.expiresAt.toISOString()}},publications); return {qualificationId:input.qualificationId,startedAt:startedAt.toISOString(),expiresAt:input.expiresAt.toISOString(),revision:b.revision+1}; diff --git a/server/src/modules/external-agents/native-broker.ts b/server/src/modules/external-agents/native-broker.ts index f67181b347d..42cded192b1 100644 --- a/server/src/modules/external-agents/native-broker.ts +++ b/server/src/modules/external-agents/native-broker.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "./muse-terminal-settlement.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../services/delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { z } from "zod"; @@ -108,11 +109,19 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I await db.execute(sql`update ${operationTable} set status=${store.pending},updated_at=now() where id=${row.id}::uuid and status=${provider==="muse"?"dispatched":"reserved"}`); } catch(error) { const rejected=error&&typeof error==="object"&&("dotOperationRejected" in error||"externalOperationRejected" in error); - if(rejected)await db.execute(sql`update ${operationTable} set status='rejected',outcome=${JSON.stringify({status:"rejected",message:"Runner rejected this operation; inspect the assignment and arguments before proceeding."})}::jsonb,updated_at=now() where id=${row.id}::uuid and status=${provider==="muse"?"dispatched":"reserved"}`); + if(rejected) { + const reject = (writer: Database) => writer.execute(sql`update ${operationTable} set status='rejected',outcome=${JSON.stringify({status:"rejected",message:"Runner rejected this operation; inspect the assignment and arguments before proceeding."})}::jsonb,updated_at=now() where id=${row.id}::uuid and status=${provider==="muse"?"dispatched":"reserved"}`); + if (provider === "muse") await db.transaction(async tx => { + await notifyExternalAgentSettlement(tx, { companyId: a.companyId, bindingId: a.bindingId }); + await reject(tx); + }); + else await reject(db); + } else if(provider==="muse")await db.transaction(tx=>withExternalAdmissionGuard(tx,a.companyId,a.agentId,async()=>{ await tx.execute(sql`select id from ${bindingTable} where id=${a.bindingId}::uuid for update`); // A result can commit before the command acknowledgement is lost. // Only an actual unresolved-send transition adds an effect barrier. + await notifyExternalAgentSettlement(tx, { companyId: a.companyId, bindingId: a.bindingId }); const changed=await tx.execute(sql`update ${operationTable} set status='unknown',updated_at=now() where id=${row.id}::uuid and status in ('dispatched','pending') returning id`); if(changed.length)await tx.update(externalAgentHolds).set({nativeEffectsUnknown:true,releasedAt:null,updatedAt:new Date()}).where(and(eq(externalAgentHolds.provider,provider),eq(externalAgentHolds.assignmentId,a.id))); })); @@ -269,7 +278,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I const [old]=await tx.execute(sql`select ${assignmentColumns} from ${assignmentTable} where run_id=${runId}::uuid and turn_id=${b.turnId}`); const projection={...p,binding:b}; if(old&&externalOperationDigest("tool",old.projection)!==externalOperationDigest("tool",projection))throw conflict("Assignment event changed its immutable projection."); - if (!old && provider === "muse") await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + if (!old && provider === "muse") await notifyExternalAgentSettlement(tx, { companyId: b.companyId, bindingId: b.bindingId }); const [a]=old?[old]:await tx.execute(sql`insert into ${assignmentTable} (company_id,binding_id,binding_generation,run_id,agent_id,normalized_session_id,turn_id,revision,controller_generation,catalog_digest,status,projection,accept_by,expires_at) values (${b.companyId}::uuid,${b.bindingId}::uuid,${b.bindingGeneration},${runId}::uuid,${b.agentId}::uuid,${b.normalizedSessionId},${b.turnId},${b.assignmentRevision},${owner.controllerGeneration},${externalOperationDigest("tool",{tools:p.tools})},'offered',${JSON.stringify(projection)}::jsonb,${new Date(Number(p.acceptByUnixMs)).toISOString()}::timestamptz,${new Date(Number(p.expiresAtUnixMs)).toISOString()}::timestamptz) returning ${assignmentColumns}`); if(!a)throw conflict("Assignment creation failed."); if (provider === "openai_dot") await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${a.bindingId}`); @@ -298,6 +307,7 @@ export function createNativeExternalBroker(db:Db,provider:Provider,identity:I // the same fact must preserve later work and its original evidence. return; } + if (provider === "muse") await notifyExternalAgentSettlement(tx, { companyId: a.companyId, bindingId: a.bindingId }); if (provider === "openai_dot") await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `binding:${a.bindingId}`); await tx.execute(sql`update ${operationTable} set status=${provider==="muse"&&outcome.status==="rejected"?"rejected":store.completed},outcome=${JSON.stringify(outcome)}::jsonb,source_event_id=${event.sourceEventId},updated_at=now() where id=${op.id}::uuid`); const action=op.command.action; diff --git a/server/src/modules/run-dispatch/adapters/postgres.ts b/server/src/modules/run-dispatch/adapters/postgres.ts index 4c313517ef3..efa70ef5dab 100644 --- a/server/src/modules/run-dispatch/adapters/postgres.ts +++ b/server/src/modules/run-dispatch/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../../../services/instruction-collection-notifications.js"; @@ -716,6 +717,7 @@ export function createPostgresRunDispatchAdapter( input: CancelSuppressedRetryInput, ): Promise { await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); + await notifyExternalAgentSettlement(tx, { companyId: input.companyId, runId: input.runId }); await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId }); await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); @@ -926,6 +928,7 @@ export function createPostgresRunDispatchAdapter( providerDispatchNotStarted = false, ): Promise { await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: run.companyId, issueId }); diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index 02b02622968..184c799f82d 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../../../services/cancelled-workspace-preservation-notifications.js"; @@ -109,6 +110,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu async cancelQueueRun({ queueRunId, reason, now }) { await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId }); await notifyInstructionCollection(tx, { companyId, runId: queueRunId }); + await notifyExternalAgentSettlement(tx, { companyId, runId: queueRunId }); await notifyCancelledWorkspacePreservation(tx, { companyId, runId: queueRunId }); await notifyChatRunMilestones(tx, { companyId, issueId }); const row = await tx diff --git a/server/src/services/execution-control-reconciliation.ts b/server/src/services/execution-control-reconciliation.ts index 7f57bcbb8e1..ff09ee63367 100644 --- a/server/src/services/execution-control-reconciliation.ts +++ b/server/src/services/execution-control-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyNativeFinalizationWork } from "./native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "./unsafe-workspace-export-notifications.js"; @@ -190,6 +191,7 @@ export async function reconcileAbandonedExecutionControl( }) .where(eq(nativeRunFinalizations.runId, run.id)); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); await notifyExecutionRunWork(tx); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 2762436bc81..03e95a7f463 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery, notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -855,6 +856,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende : "Automatic continuation source could not be verified before provider admission. Review the task and send a fresh request; this attempt will not automatically retry."; await notifyChatRunMilestones(tx, { companyId: current.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id }); + await notifyExternalAgentSettlement(tx, { companyId: current.companyId, runId: current.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: current.companyId, runId: current.id }); await notifyUnsafeWorkspaceExportRunChange(tx, current); [terminal] = await tx @@ -980,6 +982,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende await notifyChatRunMilestones(tx, { companyId: run.companyId, runId: run.id, issueId: run.nativeIssueId ?? run.issueId ?? readNonEmptyString(parseObject(run.contextSnapshot).issueId) ?? undefined }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); const [row] = await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), @@ -1343,6 +1346,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende "Automatic continuation stopped by the committed chat conversation close. Send a fresh request in chat or on the Board."; await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); const [cancelled] = await tx @@ -1467,6 +1471,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const reason = "Queued messages were discarded before dispatch"; await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId }); await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id }); + await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, run); const [cancelled] = await tx @@ -3511,6 +3516,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende ? "Cancelled because the issue was cancelled before the scheduled retry became due" : "Cancelled because the issue was reassigned before the scheduled retry became due"; await notifyInstructionCollection(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); + await notifyExternalAgentSettlement(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); await notifyUnsafeWorkspaceExportRunChange(tx, scheduledRun); await notifyScheduledRetryCancellation(tx, { companyId: scheduledRun.companyId, runId: scheduledRun.id }); @@ -3652,6 +3658,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende activeExecutionRun.agentId !== issue.assigneeAgentId ) { await notifyInstructionCollection(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); + await notifyExternalAgentSettlement(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: activeExecutionRun.companyId, runId: activeExecutionRun.id }); await notifyUnsafeWorkspaceExportRunChange(tx, activeExecutionRun); await notifyChatRunMilestones(tx, { companyId: activeExecutionRun.companyId, issueId: issue.id }); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 0bb0a20cae5..f5e08ce8750 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1967,7 +1967,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1996,6 +1996,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipNativeResultCommit: opts?.skipNativeResultCommit, skipNativeRunProjection: opts?.skipNativeRunProjection, skipExternalConversationSettlement: opts?.skipExternalConversationSettlement, + skipExternalAgentSettlement: opts?.skipExternalAgentSettlement, skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index c9bfad459c1..ec632b793b0 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "../external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; @@ -1028,6 +1029,8 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl // commit cannot strand newly-actionable accounting after the live event. return db.transaction(async transaction => { const tx = transaction as unknown as Db; + if (previous?.runtimeMode === "native" && isHeartbeatRunTerminalStatus(status)) + await notifyExternalAgentSettlement(tx, { companyId: previous.companyId, runId: previous.id }); if (previous?.runtimeMode === "native") await notifyNativeRecoveryNotices(tx, { companyId: previous.companyId, runId: previous.id }); if (previous?.runtimeMode === "native") await notifyNativeRunProjection(tx, { companyId: previous.companyId, runId: previous.id }); if (previous && previous.status !== status) await notifyUnsafeWorkspaceExportRunChange(tx, previous); diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index b610a2f8cc1..e8d200a60f3 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -627,6 +628,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch); if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx); if (patch.enableWorktreeRunExecution !== undefined) { + await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); await notifyDirectoryCleanup(tx); await notifyInstructionCollection(tx); diff --git a/server/src/services/muse-maintenance.test.ts b/server/src/services/muse-maintenance.test.ts index be6f2cfa519..bb7477ab089 100644 --- a/server/src/services/muse-maintenance.test.ts +++ b/server/src/services/muse-maintenance.test.ts @@ -138,10 +138,25 @@ describeDatabase("Muse commit and deadline maintenance", () => { expect(effects.endQualification).not.toHaveBeenCalled(); expect(coordinator.nextWakeAt()).toBeNull(); }); - it("retains recovery only until an outstanding assignment settles", async () => { + it("does not poll a running Muse assignment waiting for a real event", async () => { const b = await fixture(); const [run] = await db.insert(heartbeatRuns).values({ companyId: b.companyId, agentId: b.agentId, invocationSource: "automation", status: "running" }).returning(); + await db.insert(museRunnerAssignments).values({ companyId: b.companyId, + agentId: b.agentId, bindingId: b.id, bindingGeneration: b.generation, runId: run.id, + normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, + catalogDigest: "test", status: "accepted", projection: {}, acceptBy: new Date(), expiresAt: new Date(Date.now() + 3_600_000) }); + await start(); + expect(coordinator.nextWakeAt()).toBeNull(); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(600_000); + expect(calls).toEqual(baseline); + }); + + it("retries a terminal assignment only until its concrete settlement completes", async () => { + const b = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: b.companyId, agentId: b.agentId, + invocationSource: "automation", status: "succeeded", finishedAt: new Date() }).returning(); const [assignment] = await db.insert(museRunnerAssignments).values({ companyId: b.companyId, agentId: b.agentId, bindingId: b.id, bindingGeneration: b.generation, runId: run.id, normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, @@ -149,7 +164,7 @@ describeDatabase("Muse commit and deadline maintenance", () => { await start(); expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); effects.reconcileTerminalAssignments.mockImplementationOnce(async () => { - await db.update(museRunnerAssignments).set({ status: "settled", acceptedResultAt: new Date(), finalizedAt: new Date() }) + await db.update(museRunnerAssignments).set({ status: "settled", acceptedResultAt: new Date(), finalizedAt: run.finishedAt }) .where(eq(museRunnerAssignments.id, assignment.id)); }); await vi.advanceTimersByTimeAsync(5000); await settle(); @@ -158,6 +173,35 @@ describeDatabase("Muse commit and deadline maintenance", () => { await vi.advanceTimersByTimeAsync(600_000); expect(calls).toEqual(baseline); }); + it("honors a nearer qualification deadline while terminal settlement remains", async () => { + const at = new Date(Date.now() + 1000); + const expiring = await fixture(at); + const b = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: b.companyId, agentId: b.agentId, + invocationSource: "automation", status: "succeeded", finishedAt: new Date() }).returning(); + const [assignment] = await db.insert(museRunnerAssignments).values({ companyId: b.companyId, + agentId: b.agentId, bindingId: b.id, bindingGeneration: b.generation, runId: run.id, + normalizedSessionId: randomUUID(), turnId: randomUUID(), controllerGeneration: 1, + catalogDigest: "test", status: "accepted", projection: {}, acceptBy: new Date(), + expiresAt: new Date(Date.now() + 60_000) }).returning(); + await start(); + expect(coordinator.nextWakeAt()).toBe(at.getTime()); + expect(effects.endQualification).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(Math.max(0, at.getTime() - Date.now())); await settle(); + expect(effects.endQualification).toHaveBeenCalledWith(expect.objectContaining({ bindingId: expiring.id })); + expect((await db.select().from(museRunnerAssignments).where(eq(museRunnerAssignments.id, assignment.id)))[0].finalizedAt).toBeNull(); + expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + effects.reconcileTerminalAssignments.mockImplementationOnce(async () => { + await db.update(museRunnerAssignments).set({ status: "settled", acceptedResultAt: new Date(), finalizedAt: run.finishedAt }) + .where(eq(museRunnerAssignments.id, assignment.id)); + }); + await vi.advanceTimersByTimeAsync(5000); await settle(); + expect(coordinator.nextWakeAt()).toBeNull(); + expect(errors).toEqual([]); + const baseline = [...calls]; + await vi.advanceTimersByTimeAsync(600_000); + expect(calls).toEqual(baseline); + }); it("moves a scheduled deadline earlier on a committed change", async () => { const binding = await fixture(new Date(Date.now() + 600_000)); await start(); const at = new Date(Date.now() + 60_000); @@ -180,7 +224,7 @@ describeDatabase("Muse commit and deadline maintenance", () => { expect(calls).toEqual([]); enabled = true; await vi.advanceTimersByTimeAsync(5000); await settle(); - expect(effects.reconcileTerminalAssignments).toHaveBeenCalledWith(binding.companyId, binding.agentId, binding.id); + expect(effects.reconcileTerminalAssignments).toHaveBeenCalledWith(binding.companyId, binding.agentId, binding.id, { limit: 100 }); expect(calls.filter(c => c === "update")).toHaveLength(1); }); it("pauses deadline maintenance during drain and catches up after release", async () => { diff --git a/server/src/services/muse-maintenance.ts b/server/src/services/muse-maintenance.ts index 08180b8ec54..9dc622d1463 100644 --- a/server/src/services/muse-maintenance.ts +++ b/server/src/services/muse-maintenance.ts @@ -1,6 +1,7 @@ +import { pendingMuseTerminalBindings, type ExternalAgentSettlementScope } from "../modules/external-agents/index.js"; import { logger } from "../middleware/logger.js"; -import { and, asc, eq, gt, inArray, isNull, lte, or } from "drizzle-orm"; -import { museAgentBindings as bindings, museRunnerAssignments as assignments, mcpOauthGrants, type Db } from "@paperclipai/db"; +import { and, asc, eq, gt, isNull, lte } from "drizzle-orm"; +import { museAgentBindings as bindings, mcpOauthGrants, type Db } from "@paperclipai/db"; import { museRunnerBroker } from "./muse-runner-broker.js"; import { museReceiver } from "./muse-receiver.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -9,90 +10,119 @@ import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; export function startMuseMaintenance( db: Db, coordinator: Pick, "register">, + canSettle: () => Promise = async () => true, ) { const broker = museRunnerBroker(db); const receiver = museReceiver(db); let startupMarked = false; - let requestedSweep = 0; - let completedSweep = -1; + const requested = new Set(["*"]); let nextDeadline: number | null = null; const worker = coordinator.register(DELIVERY_QUEUES.museMaintenance, { retryMs: 5_000, pausePassiveMaintenance: true, - onSettled: () => { requestedSweep++; }, + onSettled: hints => { for (const hint of hints ?? ["*"]) requested.add(hint); }, async run(signal) { - const sweep = requestedSweep; - if (!startupMarked) { - await db.update(bindings).set({ cadenceEvidenceIncompleteAt: new Date() }) - .where(and(isNull(bindings.revokedAt), gt(bindings.qualificationExpiresAt, new Date()))); - startupMarked = true; - } - let failed = false; - let after: string | undefined; - while (!signal.aborted) { - const expired = await db.select().from(bindings).where(and( - isNull(bindings.revokedAt), lte(bindings.qualificationExpiresAt, new Date()), - after ? gt(bindings.id, after) : undefined, - )).orderBy(asc(bindings.id)).limit(100); - for (const b of expired) { - if (signal.aborted) return; - if (!b.qualificationId) throw new Error("Muse qualification is missing its identity"); - try { - await broker.endQualification({ companyId: b.companyId, agentId: b.agentId, - bindingId: b.id, qualificationId: b.qualificationId, operatorId: b.operatorId }); - } catch { - failed = true; - logger.warn({ bindingId: b.id }, "Muse qualification deadline maintenance failed; will retry."); - } + if (nextDeadline !== null && nextDeadline <= Date.now()) requested.add("*"); + const hints = [...requested]; + for (const hint of hints) requested.delete(hint); + try { + const global = hints.includes("*"); + const settlementAllowed = await canSettle(); + const inspected = new Set(); + if (!startupMarked) { + await db.update(bindings).set({ cadenceEvidenceIncompleteAt: new Date() }) + .where(and(isNull(bindings.revokedAt), gt(bindings.qualificationExpiresAt, new Date()))); + startupMarked = true; } - if (expired.length < 100) break; - after = expired.at(-1)!.id; - } - after = undefined; - while (!signal.aborted) { - const active = await db.select().from(bindings).where(and( - isNull(bindings.revokedAt), after ? gt(bindings.id, after) : undefined, - )).orderBy(asc(bindings.id)).limit(100); - for (const b of active) { - if (signal.aborted) return; - try { - if (b.grantId) { - const [grant] = await db.select({ revokedAt: mcpOauthGrants.revokedAt }).from(mcpOauthGrants) - .where(and(eq(mcpOauthGrants.id, b.grantId), eq(mcpOauthGrants.companyId, b.companyId), - eq(mcpOauthGrants.agentId, b.agentId), eq(mcpOauthGrants.userId, b.operatorId))); - if (!grant || grant.revokedAt) { - await broker.revokeOAuthGrant({ grantId: b.grantId, companyId: b.companyId, - agentId: b.agentId, authorizingUserId: b.operatorId }); - continue; + let failed = false; + let after: string | undefined; + if (global) { + while (!signal.aborted) { + const expired = await db.select().from(bindings).where(and( + isNull(bindings.revokedAt), lte(bindings.qualificationExpiresAt, new Date()), + after ? gt(bindings.id, after) : undefined, + )).orderBy(asc(bindings.id)).limit(100); + for (const b of expired) { + if (signal.aborted) throw new Error("Muse maintenance stopped"); + if (!b.qualificationId) throw new Error("Muse qualification is missing its identity"); + try { + await broker.endQualification({ companyId: b.companyId, agentId: b.agentId, + bindingId: b.id, qualificationId: b.qualificationId, operatorId: b.operatorId }); + } catch { + failed = true; + logger.warn({ bindingId: b.id }, "Muse qualification deadline maintenance failed; will retry."); } } - await broker.reconcileTerminalAssignments(b.companyId, b.agentId, b.id); - } catch { - failed = true; - logger.warn({ bindingId: b.id }, "Muse authority or terminal reconciliation failed; will retry."); + if (expired.length < 100) break; + after = expired.at(-1)!.id; + } + after = undefined; + while (!signal.aborted) { + const active = await db.select().from(bindings).where(and( + isNull(bindings.revokedAt), after ? gt(bindings.id, after) : undefined, + )).orderBy(asc(bindings.id)).limit(100); + for (const b of active) { + if (signal.aborted) throw new Error("Muse maintenance stopped"); + try { + if (b.grantId) { + const [grant] = await db.select({ revokedAt: mcpOauthGrants.revokedAt }).from(mcpOauthGrants) + .where(and(eq(mcpOauthGrants.id, b.grantId), eq(mcpOauthGrants.companyId, b.companyId), + eq(mcpOauthGrants.agentId, b.agentId), eq(mcpOauthGrants.userId, b.operatorId))); + if (!grant || grant.revokedAt) { + await broker.revokeOAuthGrant({ grantId: b.grantId, companyId: b.companyId, + agentId: b.agentId, authorizingUserId: b.operatorId }); + continue; + } + } + if (settlementAllowed) { + await broker.reconcileTerminalAssignments(b.companyId, b.agentId, b.id, { limit: 100 }); + inspected.add(b.id); + } + } catch { + failed = true; + logger.warn({ bindingId: b.id }, "Muse authority or terminal reconciliation failed; will retry."); + } + } + if (active.length < 100) break; + after = active.at(-1)!.id; } } - if (active.length < 100) break; - after = active.at(-1)!.id; + if (settlementAllowed) { + for (const hint of global ? ["*"] : hints) { + const scope: ExternalAgentSettlementScope | undefined = hint === "*" ? undefined : JSON.parse(hint); + let after: string | undefined; + while (!signal.aborted) { + const pending = await pendingMuseTerminalBindings(db, scope, after); + for (const binding of pending) { + if (signal.aborted) throw new Error("Muse settlement stopped"); + if (inspected.has(binding.id)) continue; + try { await broker.reconcileTerminalAssignments(binding.companyId, binding.agentId, binding.id, { limit: 100 }); } + catch (error) { failed = true; logger.warn({ err: error, bindingId: binding.id }, "Muse settlement failed; will retry."); } + } + if (pending.length < 100) break; + after = pending.at(-1)!.id; + } + + + if ((await pendingMuseTerminalBindings(db, scope)).length) requested.add(hint); + } + } + if (signal.aborted) throw new Error("Muse maintenance stopped"); + if (failed) throw new Error("Muse maintenance has unfinished recovery"); + if (global) { + const [next] = await db.select({ at: bindings.qualificationExpiresAt }).from(bindings) + .where(isNull(bindings.revokedAt)) + .orderBy(asc(bindings.qualificationExpiresAt)).limit(1); + nextDeadline = next?.at?.getTime() ?? null; + } + } catch (error) { + for (const hint of hints) requested.add(hint); + throw error; } - if (failed) throw new Error("Muse maintenance has unfinished recovery"); - const [next] = await db.select({ at: bindings.qualificationExpiresAt }).from(bindings) - .where(isNull(bindings.revokedAt)) - .orderBy(asc(bindings.qualificationExpiresAt)).limit(1); - nextDeadline = next?.at?.getTime() ?? null; - completedSweep = sweep; }, - async hasPending() { - const [pending] = await db.select({ id: assignments.id }).from(assignments) - .innerJoin(bindings, and(eq(bindings.id, assignments.bindingId), - eq(bindings.companyId, assignments.companyId), eq(bindings.agentId, assignments.agentId), - eq(bindings.generation, assignments.bindingGeneration), isNull(bindings.revokedAt))) - .where(or(inArray(assignments.status, ["offered", "claimed", "accepted"]), - and(eq(assignments.status, "settled"), isNull(assignments.finalizedAt)))) - .limit(1); - return !!pending; - }, - nextRunAt: () => completedSweep !== requestedSweep ? Date.now() + 5_000 : nextDeadline, + + hasPending: async () => false, + nextRunAt: () => requested.size ? Math.min(Date.now() + 5_000, nextDeadline ?? Infinity) : nextDeadline, }); receiver.start(); return Object.assign(() => receiver.stop(), { ready: worker.ready }); diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 4d9799f5f69..05899ebf308 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -13,6 +14,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { + await notifyExternalAgentSettlement(transaction, scope); await notifyExternalConversationSettlement(transaction, scope); await notifyNativeRecoveryNotices(transaction, scope); await notifyNativeRunProjection(transaction, scope); diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 5032156c669..af828edfb98 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; import { resolveNativeReconciliationStatus } from "./native-reconciliation-status.js"; @@ -260,6 +261,7 @@ export async function claimNativeSessionResumptions(input: { nextAttemptAt: null, updatedAt: now, }).where(eq(nativeRunFinalizations.runId, row.run.id)); + await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, row.run); @@ -392,6 +394,7 @@ type NativeReconciliationOptions = { skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipExternalConversationSettlement?: boolean; + skipExternalAgentSettlement?: boolean; skipNativeRecoveryNotices?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; @@ -691,6 +694,7 @@ async function reconcileNativeFinalizationsPass( skipNativeReviewResponse: options.skipNativeReviewResponse, skipCommittedRunProjection: options.skipNativeRunProjection, skipExternalConversationSettlement: options.skipExternalConversationSettlement, + skipExternalAgentSettlement: options.skipExternalAgentSettlement, skipNativeRecoveryNotices: options.skipNativeRecoveryNotices, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( @@ -755,6 +759,7 @@ async function reconcileNativeFinalizationsPass( skipNativeReviewResponse: options.skipNativeReviewResponse, skipCommittedRunProjection: options.skipNativeRunProjection, skipExternalConversationSettlement: options.skipExternalConversationSettlement, + skipExternalAgentSettlement: options.skipExternalAgentSettlement, skipNativeRecoveryNotices: options.skipNativeRecoveryNotices, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 82a202c5ff7..e3052482467 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -923,6 +924,7 @@ export async function claimNativeRestartRecoveries(input: { eq(nativeRunFinalizations.phase, row.coordinator.phase), ), ); + await notifyExternalAgentSettlement(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyInstructionCollection(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: row.run.companyId, runId: row.run.id }); await notifyUnsafeWorkspaceExportRunChange(tx, row.run); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 6fa84a0931c..7a8d6146c9e 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; @@ -1089,6 +1090,7 @@ export async function finalizeNativeRun(input: { skipNativeReviewResponse?: boolean; skipCommittedRunProjection?: boolean; skipExternalConversationSettlement?: boolean; + skipExternalAgentSettlement?: boolean; skipNativeRecoveryNotices?: boolean; /** Workspace-only replay must not consume the provider recovery budget. */ preserveProviderAttempt?: boolean; @@ -1156,7 +1158,7 @@ export async function finalizeNativeRun(input: { }); } if (!input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: admittedRun.companyId, runId: admittedRun.id, issueId: coordinator.issueId }); - if (input.projectRunStatus) { + if (input.projectRunStatus && !input.skipExternalAgentSettlement) { const [projectedRun] = await input.db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, @@ -1635,6 +1637,7 @@ export async function finalizeNativeRun(input: { .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))) .for("update").limit(1); if (!before) return null; + if (input.projectRunStatus) await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id }); await notifyNativeRecoveryNotices(tx, { companyId: run.companyId, runId: run.id }); await notifyNativeRunProjection(tx, { companyId: run.companyId, runId: run.id }); await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id }); @@ -1673,7 +1676,7 @@ export async function finalizeNativeRun(input: { }); } if (finalizationPhase === "committed" && !input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: run.companyId, runId: run.id, issueId: coordinator.issueId }); - if (input.projectRunStatus && updatedRun) await reconcileMuseNativeTerminalRun(input.db, updatedRun); + if (input.projectRunStatus && updatedRun && !input.skipExternalAgentSettlement) await reconcileMuseNativeTerminalRun(input.db, updatedRun); return { ...coordinator, phase: finalizationPhase, diff --git a/server/src/services/native-runtime/native-run-projection.ts b/server/src/services/native-runtime/native-run-projection.ts index 746aecc663b..38fe83e8a07 100644 --- a/server/src/services/native-runtime/native-run-projection.ts +++ b/server/src/services/native-runtime/native-run-projection.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; @@ -127,6 +128,7 @@ export async function projectCommittedRun(input: { nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt }) .from(heartbeatRuns).where(needsRepair).for("update").limit(1); if (!before) return null; + await notifyExternalAgentSettlement(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyNativeRecoveryNotices(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyNativeRunProjection(tx, { companyId: input.run.companyId, runId: input.run.id }); await notifyRetainedNativeSessionCleanup(tx, { companyId: input.run.companyId, runId: input.run.id }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 69b8b5149d9..2a58aad73a3 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; import { releaseRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; import type { NativeCleanupOutcome } from "./retained-native-session-cleanup.js"; @@ -9868,6 +9869,7 @@ async function executePaperclipNativeSessionWithinScope( .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.execution.binding.runId)) .for("update"); + if (!ownershipUnverified) await notifyExternalAgentSettlement(tx, input.execution.binding); if (!ownershipUnverified) await notifyInstructionCollection(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId }); if (!ownershipUnverified) await notifyCancelledWorkspacePreservation(tx, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId }); if (!ownershipUnverified) await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId }); diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index cedd7130e8c..78078a0a7ac 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { lockCurrentNativeWorkAssessmentEvidence } from "./work-assessments.js"; import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finalization-recovery-state.js"; import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js"; @@ -1098,6 +1099,7 @@ async function materializeDecisionEffect(input: { ), ); } + await notifyExternalAgentSettlement(input.tx, { companyId: input.companyId, runId: input.runId }); await notifyInstructionCollection(input.tx, { companyId: input.companyId, runId: input.runId }); await notifyCancelledWorkspacePreservation(input.tx, { companyId: input.companyId, runId: input.runId }); await notifyUnsafeWorkspaceExportRecovery(input.tx, { companyId: input.companyId, issueId: input.issue.id }); diff --git a/server/src/services/public-mcp/oauth.ts b/server/src/services/public-mcp/oauth.ts index 4f5cdad55db..49e4e38b35a 100644 --- a/server/src/services/public-mcp/oauth.ts +++ b/server/src/services/public-mcp/oauth.ts @@ -320,7 +320,7 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { if(!candidate)return db.transaction(tx=>operation(tx as unknown as Db)); return db.transaction(tx=>withExternalAdmissionGuard(tx,candidate.companyId,candidate.agentId,async()=>{ await tx.select({id:museAgentBindings.id}).from(museAgentBindings).where(and(eq(museAgentBindings.id,candidate.id),eq(museAgentBindings.grantId,grant.id))).for("update"); - await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, "*"); return operation(tx as unknown as Db); })); } @@ -508,7 +508,7 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { await tx.insert(mcpOauthRequests).values({ id, clientId: client.id, redirectUri: p.redirect_uri, resource: config.resource, scopes, requestedCompanyId: binding.companyId, state: p.state ?? null, challenge: p.code_challenge, grantId: grant.id, codeHash: hashMcpSecret(code), decidedAt: now, expiresAt: new Date(now.getTime() + minute) }); - await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, "*"); await tx.update(museAgentBindings).set({ grantId: grant.id, status: "connected", pairedAt: now, workerActivityAt: now, ticketHash: null, ticketExpiresAt: null, clientVersion: null, revision: binding.revision + 1, updatedAt: now }).where(eq(museAgentBindings.id, binding.id)); await logActivity(tx as unknown as Db, { companyId: binding.companyId, actorType: "user", actorId: binding.operatorId, @@ -660,7 +660,7 @@ export function createPublicMcpOAuth(db: Db, config: PublicMcpConfig, options: { const [grant] = await tx.select().from(mcpOauthGrants).where(eq(mcpOauthGrants.id, token.grantId)).for("update"); if (!grant || grant.clientId !== clientId || grant.resource !== config.resource || grant.revokedAt) return null; if (token.usedAt) { - if (museAgentConnection) await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance); + if (museAgentConnection) await notifyDeliveryWork(tx, DELIVERY_QUEUES.museMaintenance, "*"); await tx.update(mcpOauthGrants).set({ revokedAt: new Date() }).where(eq(mcpOauthGrants.id, grant.id)); revokedGrant = grant; await tx.insert(activityLog).values({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 24c32a005fc..d9b422ea913 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -1,3 +1,4 @@ +import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; @@ -6262,6 +6263,7 @@ export function recoveryService( const [candidate] = await tx.select().from(heartbeatRuns).where(eligible).for("update"); if (!candidate) return null; await notifyInstructionCollection(tx, { companyId: candidate.companyId, runId: candidate.id }); + await notifyExternalAgentSettlement(tx, { companyId: candidate.companyId, runId: candidate.id }); await notifyCancelledWorkspacePreservation(tx, { companyId: candidate.companyId, runId: candidate.id }); const boundIssueId = boundRunIssueId(candidate); if (boundIssueId) {