From 829f9e7eafe57588d214697efca336b35aecf45d Mon Sep 17 00:00:00 2001
From: Dotta <34892728+cryppadotta@users.noreply.github.com>
Date: Thu, 29 Oct 2026 06:18:40 -0500
Subject: [PATCH] fix(server): settle Muse terminal assignments from committed
changes (#17793)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
https://entire.com/gh/paperclipai/-/trails/12/changes/paperclip/12
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - Its external-agent module owns Muse assignments, operation receipts,
and worker holds.
> - Running assignments and retained terminal history must not cause an
empty maintenance scan every five seconds.
> - Terminal run changes and operation results already have durable
database transactions.
> - This pull request wakes the existing maintenance worker from those
commits and reconstructs work at startup.
> - The worker retries only concrete settlement debt. It retains
qualification deadlines and authority checks.
## Linked Issues or Issue Description
Refs: #17772. This is slice 57. Parent #17772 has merged.
Related: #17000, #17525, #17752, #17758. The publication targets master
and preserves the merged parent.
**What existing behavior does this improve?**
Muse terminal assignment settlement and the existing Muse maintenance
worker.
**Current behavior**
A running assignment can keep maintenance scanning every five seconds.
Terminal reconciliation rewrites unchanged history. Native finalization
also calls that reconciliation during periodic replay. Discovery
excludes a revoked original binding even when its assignment still needs
settlement.
**Proposed behavior**
Committed run changes and operation receipts wake the existing worker
with the original company, run or binding identity. Startup reconstructs
actionable work from existing rows. Maintenance processes bounded
batches and disarms after settlement. Default live and manual callbacks
remain. The periodic callback delegates this one duty.
**Reason and benefit**
Configured but idle instances stop querying for this obligation. Late
receipts, restarts, drain and historical bindings still have recovery
paths. No table or independent timer is added.
**Breaking changes**
No schema, endpoint or provider protocol changes. Explicit manual
reconciliation retains its unbounded default. This does not alter Cloud
sleep admission or claim that all instance work can sleep.
## What Changed
- Add an owner predicate shared by discovery and locked settlement.
Fenced private-worker uncertainty remains held without an automatic
retry loop.
- Preserve original binding generation and company/agent/run identity,
including revoked bindings. Fence only the terminal assignment during
reconciliation.
- Extend existing maintenance with scoped hints, bounded batches,
concrete-debt retry, startup reconstruction and quiet drain behavior.
- Register notifications inside existing terminal, projection,
cancellation, operation-result and send-failure transactions. Keep
qualification and OAuth signals explicitly global when coalesced with
scoped changes.
- Serialize only four domain IDs. Real native-finalizer tests caught and
prevent a circular-object serialization failure at this boundary.
- Use existing restore/worktree suppression and settings reactivation.
Delegate both periodic finalizer branches while retaining resource
callbacks and default live/manual settlement.
- Add the producer audit and regression evidence in
`doc/plans/2026-10-26-muse-terminal-settlement-sleep.md`.
## Verification
Accepted production source: `584015018ed4a9d57d2b0f192b75847503b54e53`.
Publication head: `305014b030d6c5eb891c70bc57036f4192d4fdad`. Normal
adoption of the actual parent squash preserves the complete accepted
child tree. Only audit text, a migration-count clarification, and the
proved Photon test extension follow.
- **3,715 distinct cases in 52 complete server files pass**, plus **20
selected chat cases** (3,735 selected total). The main 49-file batch
passes 3,008; process recovery passes 410; review response passes 273;
the full Photon file passes 24. The other 1,794 chat cases were not run
locally.
- **Quiet producer proof:** moving only the Muse fanout outside its
parent guard produces six candidate SELECTs and fails the regression.
Keeping the guard yields zero and still settles a real terminal run. The
same test hash spans the controlled red/green pair.
- **Photon review evidence:** repeated active callbacks produce zero
settlement hints/candidate SQL. Three concurrent attention recoveries
produce exactly one of each. The existing locked attention guard was
correct; production needed no fix. The parent security check accepted
the resolved finding before merge. This PR keeps the additional
settlement queue in the existing regression.
- **Static checks pass:** database build; server typecheck/build; module
boundaries; migration order; current Copilot identity. Packages equal
the accepted parent byte for byte. No new live provider qualification is
claimed.
- **Fixtures and failures:** 34 fresh PostgreSQL 17 fixtures have 412
applied migration receipts through tag 0413_tired_kronos, plus fresh
per-call helper fixtures. All 52 original test hashes restore exactly.
The additional Photon test hash matches its passing proof. Earlier
remote chat passive-idle failure remains recorded; both passive/missing
cases pass in the new selection. Parent handoff timeout/fresh-file-pass
accounting remains in its audit. No assertion or timeout was weakened.
- [Integration and quiet producer
audit](https://github.com/paperclipai/paperclip/blob/305014b030d6c5eb891c70bc57036f4192d4fdad/doc/plans/2026-10-26-muse-terminal-settlement-sleep.md#current-master-integration-and-quiet-producers-2026-10-29).
Fresh CI, security, and exact-head Greptile 5/5 must pass before merge.
Earlier revision verification and failure
history
Receiver failure fairness: a controlled 1,100 ms batch reproduced a
30-second delay for healthy due receivers. The worker now supplies its
existing per-item deadline through `nextRunAtOnError`. All 20 receiver
lifecycle tests pass on the owner and combined tip
`31c12f1d68ff4a51e06fc2be2abf2c2aee9f9d3a`; server production
typechecking passes on the owner. The failed receiver retains its own
backoff, and settled receivers return to zero recurring SQL. The
five-second assertion budget is unchanged. See [the owning
audit](https://github.com/paperclipai/paperclip/blob/f7477933469fdc960f555425dce66eff25a94e54/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md#healthy-receiver-progress-after-a-slow-failed-batch).
Current-head CI and review remain required.
Approved delegated Test-tab calls now evaluate caller and target
policies in the existing claim transaction. Real HTTP regressions first
showed a missing caller charge and shared-limit self-rejection. All 78
connection-management cases and 92 gateway cases now pass on the owning
branch, along with server production typechecking. All 78
connection-management cases also pass at combined tip
`d25f830db89c1623bff3547bfc68015c1c66e0cf`. The three new cases cover
shared charges, separate charges, rollback on denial, duplicate
approval, and rejection of a later authorized call. Intermediate
counter/identity fixture failures and a reused-database Slack collision
are recorded in [the owning
audit](https://github.com/paperclipai/paperclip/blob/bbd31dea5ef3a4875950a87b4636af8b6bde858e/doc/idle-approved-tool-recovery.md#delegated-test-tab-admission-after-policy-changes).
The earlier combined checks below remain evidence for their stated
revisions. Current-head CI and review remain required.
Inherited voice-cleanup verification: the controlled slow-provider
regression and all 13 lifecycle tests passed on its owner and combined
tip `ab22d8ed0c09028315607414b83b375e5cc0a60a`. That repair changed only
the fixture clock advancement and its audit note.
Current head: `f7477933469fdc960f555425dce66eff25a94e54`. This remains a
single-duty PR stacked on #17772.
- Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5`
with normal merges. Every branch passed generated-schema checks and its
target-branch migration-order gate. Existing SQL is unchanged; the 25
stack migrations follow master’s new migration.
- Duty-specific producer coverage, lifecycle tests, and earlier failure
records:
[2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/ff196791db6dada2dc04ea9db0d58c5c3beb4a41/doc/plans/2026-10-24-recurring-server-sleep-stack.md),
[2026-10-26-muse-terminal-settlement-sleep.md](https://github.com/paperclipai/paperclip/blob/ff196791db6dada2dc04ea9db0d58c5c3beb4a41/doc/plans/2026-10-26-muse-terminal-settlement-sleep.md).
- Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355
startup/cancellation/recovery/dispatch/preservation tests**, **75
connection-management tests**, and production server typechecking pass.
These combined checks do not mean every intermediate PR independently
ran the full repository suite.
- Real master-to-stack database upgrade passes: **379 unchanged prior
migration receipts plus 25 new migrations**, with the gateway
responsibility fixture preserved.
- Owning-branch checks: 84 runner installation/settlement cases plus
runner typechecking and generated TypeScript/Rust profile agreement; 253
GitHub review workflow cases. Stop-attribution and held-worker readiness
regressions failed before their fixes.
- Failure accounting: the initial combined run passed 349 cases and
failed six idle assertions in a reused preservation database containing
old cancelled runs. Repeating the combined command with a fresh
preservation database passed. Earlier invalid project selection, merge
import/type errors, and the interrupted full local suite are not counted
as passing checks.
- **Latest-head CI and review remain required.** Prior CI findings
include GitHub idle observation, route teardown, attention cleanup,
touch-picker E2E, workspace health, and voice-cleanup timing; local
passes do not erase those failures. No full local repository-suite,
deployed sleep, fleet cost, or live Copilot qualification claim is made.
(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) {