fix(server): settle Muse terminal assignments from committed changes (#17793)

<!-- entire-trail-link-start -->
https://entire.com/gh/paperclipai/-/trails/12/changes/paperclip/12
<!-- entire-trail-link-end -->

## 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.

<details><summary>Earlier revision verification and failure
history</summary>

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.


</details>

## Risks

- In-process signals do not capture independent SQL or other replicas.
Supported external events still need ingress.
- Pending assignments and operation receipts retain their original
binding identity. A new agent configuration cannot take their ownership.
- Genuine qualification, OAuth, and recovery deadlines remain
obligations. They must not be discarded merely to make the worker quiet.
- The parent coarse-hint guard prevents recurring Muse discovery; real
company/binding authority changes retain their signals.
- No full local repository suite, full chat suite, live Muse or Copilot
qualification, deployed sleep, database suspension, external alarm
coverage, or cost savings is claimed. Existing Cloud blockers remain.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning, code execution, real
PostgreSQL tests, and GitHub tools. The exact runtime model identifier
and context-window size are not exposed in this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
authored and GitHub committed 2026-10-29 06:18:40 -05:00
1 parent 4d0847c4ca
commit 829f9e7eaf
33 files changed
+983 -98

No files matched your search

@@ -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.
@@ -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
@@ -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.
+31
View File
@@ -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<string>`${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();
@@ -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<typeof createDb>;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_MUSE_SETTLEMENT_TEST_DATABASE_URL");
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let stop: ReturnType<typeof startMuseMaintenance> | undefined;
let admitted = true, recover = true;
const errors: unknown[] = [], detachers: Array<() => Promise<void>> = [];
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<boolean> = 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<ReturnType<typeof fixture>>,afterSettle?: (command:ExternalProviderOperation)=>Promise<void>) {
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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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<void>(resolve => { entered = resolve; });
const gate = new Promise<void>(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();
});
});
@@ -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");
@@ -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();
@@ -1090,7 +1090,7 @@ describe("startServer feedback export wiring", () => {
intervalCallback?.();
await new Promise<void>((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 {
+4 -1
View File
@@ -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) {
+1 -1
View File
@@ -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();
@@ -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 });
@@ -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";
@@ -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<Db, "selectDistinct">,
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);
}
+17 -10
View File
@@ -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};
@@ -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<S>(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<S>(db:Db,provider:Provider,identity:I
const [old]=await tx.execute<Assignment>(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<Assignment>(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<S>(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;
@@ -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<CancelSuppressedRetryResult> {
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<CancelStaleQueuedRunOutcome> {
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 });
@@ -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
@@ -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);
+7
View File
@@ -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 });
+2 -1
View File
@@ -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,
@@ -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);
+2
View File
@@ -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);
+47 -3
View File
@@ -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 () => {
+100 -70
View File
@@ -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<ReturnType<typeof createDeliveryWorkCoordinator>, "register">,
canSettle: () => Promise<boolean> = async () => true,
) {
const broker = museRunnerBroker(db);
const receiver = museReceiver(db);
let startupMarked = false;
let requestedSweep = 0;
let completedSweep = -1;
const requested = new Set<string>(["*"]);
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<string>();
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 });
@@ -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);
@@ -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(
@@ -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);
@@ -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,
@@ -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 });
@@ -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 });
@@ -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 });
+3 -3
View File
@@ -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({
+2
View File
@@ -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) {