Compare commits

...
Author SHA1 Message Date
jason.woltje f5256e6779 morgana: row 295 r2 narrow alice grant (Dewey F1, sage-3333): aliceAccess(ref) names only the test's project, set after project.create; asserts and bindLaunch lines byte-identical; BUILD.md preservation record updated 2026-10-23 04:27:51 -05:00
jason.woltje f652dfa5b8 morgana: row 295 r2 old-test setup (mos-1436 option a): message-filter's :117 and governance-pin tests get an alice collaborator access document, spec-fixtures fixture() takes an optional access (null default); asserts and bindLaunch lines byte-identical; BUILD.md gate record, assertion preservation, N1 revoke window 2026-10-23 04:23:33 -05:00
jason.woltje 9f29e1b2fd morgana: row 295 r2 first-run reds placed (mos-1272): claim rows spread off the store's null prototype in access-revoke and thread-revoke, demo switches the pi harness on for startBroker, Alice binds at her first call in thread-revoke 2026-10-23 04:16:39 -05:00
jason.woltje b16cef1d0b morgana: row 295 r2 rebase on 80793d824 (sage-3256): accessFor comment is stamp-cached (dewey-0871 N3); BUILD.md rebase note 2026-10-23 02:13:21 -05:00
jason.woltje bc8cd60d02 morgana: row 295 r2: wrap the launcher's revoked comment (dewey-0874 nit) 2026-10-23 02:11:49 -05:00
jason.woltje 2d981107eb morgana: row 295 r2 (dewey-0871): threadsState ends an expired grant's held claim from the cached read by access-expiry (B1, D7, M16, M11 claim row); access.reload scope as D8 (N1); bus README access.reload, revoked and revocation (N2); tests unrun 2026-10-23 02:11:48 -05:00
jason.woltje 60357196b6 morgana: row 295 FED R4 revocation (#1801): 30 s access timer, access.reload, one-transaction claim end on a loss, threadsState revoked, launcher SIGTERM and no start, CLI reload after grant/revoke; tests unrun (no CLEAR yet) 2026-10-23 02:11:48 -05:00
jason.woltje 943501a366 morgana: row 295 plan (FED R4 revocation, #1801): sweep on timer, access.reload and user requests; threadsState revoked; launcher skip and SIGTERM; bindLaunch and role.claim refuse thread-revoked 2026-10-23 02:11:48 -05:00
12 changed files with 985 additions and 29 deletions

No files matched your search

+207
View File
@@ -0,0 +1,207 @@
# Row 295: FED R4 revocation (#1801)
Brief: docs/plans/2026-10-23_mvp-readiness-rows.md, "Brief: FED R4, revocation".
Design: agents/gizmoduck/work/federation/DESIGN.md at b93f098f0 (r4.11), sections 3.7 and 7.
Base: origin/refactor 253346e3c (rows 206, 271 and 275 are in it). Branch morgana/row295.
Round 2 base: 80793d824 (row 294 landed as d3059c2c3), branch morgana/row295-r2 (sage-3256).
Reviewer Dewey, gate owner Sage.
Row 193's viewer level landed with row 275 (3410df7c3), so it is on this base. The downgrade test and
its mutant build here, not on a 193 candidate (Q177 not needed).
## Plan
### Broker (packages/bus/src/broker.mjs)
1. `#revoked(business, t)`: true when `t.owner` is `user:<id>` and no grant in one read of the access
file covers the thread at collaborator level (`covers(g, {business, project: t.project, instance:
t.instance}, now, {level: 'collaborator'})`). A read that throws, a null document or no `access`
function covers nothing, as `#covered` does (fail closed). A thread the business human or the tester
owns is never revoked.
2. `#revokeLost(business, by)`: one read of the file, then one transaction that ends, with
`#claimEnd(..., 'revoke', null, 'access-revoked', by)`, the held claim of every session of every
user-owned thread `#revoked` names. Returns the ended runs. The claim is ended, so the run's next
bus call refuses `not-holder` (F10) and a later `role.claim` for it refuses `run-revoked`.
3. Triggers (design 3.7):
- a trusted method `accessSweep()`, not a socket verb, that runs `#revokeLost` for each business with
`by: 'access-timer'`. The runtime calls it on a 30 s `setInterval` (unref'd, cleared in close()),
only when `access` is configured.
- `access.reload`, human-only (`#human(s)`, so not the tester and not a reader), runs `#revokeLost`
for `s.business` with `by: s.human` and returns `{ ended: [runs] }`.
- every user request (`s.user`), any user's: `#revokeLost(by: 'access-change')` before the verb,
once per stamp change (sage-3139 Q1, D2). threadsState sweeps the same way (D3).
4. `threadsState` rows gain `revoked: true|false`, from one access read per call. `#threadRows`
selects `t.owner, t.project` for it. `threadIdle` ignores the field.
5. Launch check: `bindLaunch` refuses `thread-revoked` for a new run (`!old`) of a revoked thread, the
way `!old && !harnessOn` refuses `invalid-harness`, and `role.claim` refuses `thread-revoked` too.
A rebind (`old`, recover() at launcher.mjs:1174) is not refused, so recover() can still end the
earlier host's run. bindLaunch comes after spawn and a refusal there SIGKILLs the child
(launcher.mjs:608, :620), which covers a start racing the revocation; the claim check covers a run
bound just before it.
### Runtime (packages/bus/src/runtime.mjs)
6. With `access` set: `setInterval(() => broker.accessSweep(), access.intervalMs ?? 30_000)`, unref'd,
cleared in close() and on a failed start. A sweep that throws logs one line to stderr and the next
tick tries again. `access.intervalMs` follows `librarySync.intervalMs` (C2-03a): in-process only.
### Launcher (packages/cli/src/launcher.mjs)
7. threadPass: a `revoked` row never starts. A revoked row with a running session, not already
stopping, gets `stopping.add`, a `thread-revoked` record event and the idle stop's start-time-checked
SIGTERM at once, with no `threadIdle` call (the claim is already ended).
8. A bindLaunch refusal `thread-revoked` is a start refusal, logged once per code as today.
### CLI (packages/cli/src/access.mjs)
9. grant and revoke call `access.reload` after the write (project.list stays the human proof, since
grant needs its rows for project names; D6 for `user remove`). A refused reload after a good write:
stdout still reports the write, stderr says `access.reload refused: <code>; the bus's timer
applies the change within 30 s`, exit 0, because the file is written and a retry would hit
revision-conflict (sage-3139 Q2).
10. ACCESS_HELP keeps the provider-removal order; a test pins the sentence.
## Tests (each with a mutant)
Timers: `t.mock.timers` drives `setInterval` and `Date`; no test sleeps 30 s. The bus file is
packages/bus/tests/access-revoke.test.mjs (B), the launcher file packages/cli/tests/thread-revoke.test.mjs
(L), and the CLI tests are appended to packages/cli/tests/access.test.mjs (C).
| # | Test | File | Mutant |
|---|---|---|---|
| M1 | timer sweep after a hand edit ends only alice's run (by access-timer, reason access-revoked); its next call refuses not-holder; bob's and the human's runs stay | B | `#revokeLost` ends every user thread (drop the `#revoked` filter) |
| M2 | an expiry with no write and no request ends the run at the next sweep; bob's request on the same stamp doesn't | B | `accessSweep` not forced (`#revokeLost('access-timer')`) |
| M3 | a downgrade to viewer is a loss | B | level `'viewer'` in `#revoked` |
| M4 | bob's request after a hand edit ends alice's run, and nothing ends before it (sage-3139 Q1) | B | the per-request `#revokeLost` call removed |
| M5 | one read per stamp change, none while unchanged | B | `#accessDoc` ignores the stamp |
| M6 | access.reload ends at once for the human; tester, reader, agent refused human-required/read-only, user access-refused | B | drop `#human(s)` in access.reload |
| M7 | threadsState marks `revoked: true` on the lost thread only, its pending message kept | B | threadsState never adds `revoked` |
| M8 | bindLaunch and role.claim refuse thread-revoked; another broker's rebind is allowed | B | drop the bindLaunch check |
| M9 | the runtime timer ends the run at 30 000 ms and not at 29 999; intervalMs is checked; close stops it | B | the timer callback never calls `accessSweep` |
| M10 | pending message + launcher restart starts no run, logs no launch and no refused start | L | threadPass ignores `revoked` |
| M11 | a running revoked thread's run gets SIGTERM, ends `stopped` exit 0, with no thread-idle; its last claim row is revoke/access-revoked/access-change, not a release (dewey-0871 B1) | L | the revoked branch drops the kill |
| M12 | recover() rebinds an earlier host's run on a revoked thread and ends it host-lost (sage-3139 D1) | L | the bindLaunch check without `!old` |
| M13 | grant and revoke call access.reload after the write; list and a refused write don't | C | drop the `reload` call |
| M14 | a refused reload: exit 0, stdout as before, stderr names the code and the 30 s timer (sage-3139 Q2) | C | `reload` rethrows |
| M15 | help keeps the provider-removal order with the reload sentence after it | C | drop the help sentence |
| M16 | an expiry with no write and no sweep: threadsState ends the claim (revoke, access-revoked, by access-expiry) before it returns the row revoked with holder null; one revoke row only (dewey-0871 B1) | B | drop threadsState's `#revokeLost('access-expiry', true, doc)` call |
M11 relies on the runner's poll interval (set to 600 s in L), so only the signal ends the run: without
the kill, a run that polls would exit 22 on not-holder and hide the mutant.
## Sage's rulings (sage-3139)
- Q1 resolved, yes: a request from any user counts. The cost rule: a request only stats the file;
`#revokeLost` runs only on a stat change and reuses the read R3's check makes. Tests M4 and M5.
- Q2 resolved, exit 0: stderr names the refusal code and the 30 s timer, stdout reports the write as
today. Help text next to the provider-removal order. Tests M14 and M15.
- D1 accepted with a test (M12).
- Base 253346e3c; rebase before the gate if broker.mjs or launcher.mjs change on refactor.
## Decisions
- D1: bindLaunch's revoked check skips a rebind. recover() binds the earlier host's run and then ends
it; a refusal there would leave the run unended in the broker (launcher.mjs:1173-1179).
- D2, the stamp: the runtime passes `accessStamp`, an lstat key (dev, ino, size, mtimeNs, ctimeNs,
mode, uid; a failed stat is `unreadable:<code>`). `#accessDoc` reuses the last read while the
stamp is unchanged, and the sweep runs once per stamp (`#accessSwept`). The timer and
access.reload force a read, so an expiry with no write is caught by the timer. With no stamp
function (the bus tests' old fixtures), every read reads, as R3 did, so user-principal.test's
reads+2 pin holds: a request reads once and hands the document to both the sweep and the grants.
- D3: threadsState also sweeps on a stamp change, so the launcher's tick sees the claim already
ended when it gets `revoked: true`. `revoked` is added only when true, so end-launch.test's
deepEqual of a threadsState row still holds.
- D4: `#threadLost` (bindLaunch, role.claim) only reads. role.claim runs inside the verb's
transaction and store.transaction refuses nesting, so the claim end waits for the next sweep.
- D5: a session in a clear is skipped in threadPass before the revoked branch (clearPass owns it).
Its successor's start goes through bindLaunch, which refuses thread-revoked, so no run starts.
A drained run of a revoked thread already had its claim ended by the sweep.
- D6: `user remove` doesn't call access.reload: it refuses while the user holds a grant, so it
never removes coverage. The brief names grant and revoke.
- D7 (Dewey's B1, dewey-0871): an expiry changes no stamp, so the stamp check skips the sweep while
`#revoked` with the current time already marks the row. threadsState therefore ends the lost
claims itself when a lost row still has a holder, forced, from the cached document (no read, so
sage-3139's cost rule holds), `by: 'access-expiry'`, and re-reads the rows. The launcher (2 s
tick) then never sees `revoked: true` with a held claim, and the run's end is never a release.
The request path is unchanged.
- D8 (Dewey's N1): access.reload ends lost claims in every business, since the access file is
shared, and writes the calling business's human as `by` on each. It returns only the caller's
business's ended runs, so one business's human learns nothing of another's threads.
## Review round 1 (dewey-0871, CHANGES on 527002f97)
- B1 fixed: D7, test M16, M11 assertion. N1: D8. N2: packages/bus/README.md gets the access.reload
row, `revoked` in threadsState, a revocation paragraph (timer, accessSweep, access.intervalMs,
the four triggers) and the stamp-cached wording.
- N3, at the rebase on 294 (c866a8a69): 294's accessFor passes `access` to the live boot, so the
timer and request sweep go live with both rows; fix 294's "reads the file on every user request"
comment to stamp-cached, and check its ACCESS_HELP edit against M15's ordering assertion.
- N4: the touched whole files (user-principal, viewer, thread-shares) run whole in the bus suite;
a first-run red goes per mos-1272.
## Rebase onto 80793d824 (sage-3256)
- Row 294 landed (d3059c2c3). Of 295's manifest only packages/cli/src/access.mjs drifted; the rebase
applied with no conflict and keeps 294's lines (the federation.peers help and comment,
accessContextFor's peers). The reload sentence still sits right after "so do that second.", so
M15's order assertion holds; 294 changed only lines after it.
- N3: packages/cli/src/config.mjs accessFor's comment now says the broker reads on a user request
that finds the stat changed and on the 30 s timer. config.mjs joins the manifest (comment only).
- 294 put `access` on the live boot, so the timer and the request sweep run in its new tests:
packages/cli/tests/access-boot.test.mjs and packages/webui/tests/granted-user.test.mjs run whole at
the gate with the touched files. stack-release.test's copy list carries access.mjs, which adds
no import here.
## Gate on 07802dc57 (sage-3318)
- First run, merge b42c591f8 (tree 0626929e5): access-revoke.test went 4 pass, 6 fail. These were
first-run reds in the row's new tests, placed per mos-1272 in 9f29e1b2f (sage-3326, Dewey
dewey-0905). Five claim-row compares failed because store rows have a null prototype, so they're
now spread; thread-revoke:207 had the same compare and got the same fix. The demo business switches
on the pi harness, which startBroker's harnessOn needs. In thread-revoke, Alice binds at her first
call, because a second host's bind after the hand edit refuses access-revoked.
- Second run, merge 38591cf96 (tree f719c0dc5): 12 whole files green. suite-bus stopped on 2
old-test reds in packages/bus/tests/message-filter.test.mjs. ":117 trail on a decision ..." and
":165 governance pin ..." each threw thread-revoked from bindLaunch's new check. Those fixtures
write a `user:alice` thread with no access function, and with no document no grant covers it.
Mos ruled option (a) in mos-1436: old-test setup only.
### Assertion-preservation record (mos-1436 conditions 7-9)
- message-filter.test.mjs gains `aliceAccess(ref)`. It returns the access function, whose single
grant is user alice, business demo, projects `[ref.project]`, instances `['assistant']` (the
thread's instance), collaborator level. That's the narrowest grant (Dewey's F1, sage-3333;
f652dfa5b had projects `['*']`). The broker has no accessStamp here, so #accessDoc calls the
function again on each read.
- `setup(t, { access } = {})` passes `access` to `new Broker`. Only the trail test (:117 at 9f29e1b2f, :126 now) passes it; every other
`setup(t)` gives `access: undefined`, which the constructor's default turns into null, as before.
setup() already returned `project`.
- The trail test and the governance pin (:176) each add `const ref = {};` before setup or fixture, and set
`ref.project = project;` once the project exists. The trail test also destructures `project` from setup().
The governance pin passes `access: aliceAccess(ref)` to `fixture()`.
- Nothing reads access before `ref.project` is set. The broker reads it on a user's request
(request()), in threadsState, in #covered, and in bindLaunch/role.claim on a thread.
setup() binds only the human, a reader and the tester and makes the project. fixture() binds
coder and reviewer runs with no thread. Neither binds or claims as alice, and neither calls
threadsState.
- No other line of either test changes. The bindLaunch lines (:119, :169 at 9f29e1b2f; :130, :182 now)
and every assertion are byte-identical. The removed lines are only `function setup(t) {`, the
two setup/fixture calls and the two helper lines below.
- spec-fixtures.mjs (11 importers) gains one optional parameter with a null default, which goes to
`new Broker` unchanged. No other caller changes.
-export async function fixture(t, { adapter = true, roles = {}, arbiters = { technical: 'reviewer', delivery: 'reviewer' } } = {}) {
+export async function fixture(t, { adapter = true, roles = {}, arbiters = { technical: 'reviewer', delivery: 'reviewer' }, access = null } = {}) {
- const broker = new Broker({ store, businesses: { demo: business }, project: adapter ? {
+ const broker = new Broker({ store, businesses: { demo: business }, access, project: adapter ? {
- Condition 4: M8 is the mutant. It deletes the broker.mjs:1371 thread-revoked check in bindLaunch,
and its kill test is access-revoke's bind/claim refusal test. The red-first run on 38591cf96
(both tests red), the green run and M8 run on Sage's CLEAR, followed by a full gate restart on
the new merge with a new manifest.
### N items
- N1 (morgana-0167, sage-3318): a revoked run that calls the bus after its claim ends and before the
launcher's SIGTERM gets not-holder and exits 22, not stopped (the same shape as #1809). 302's
launcher-released path doesn't cover it. Sage decides after 295 lands whether this becomes a row.
+20 -3
View File
@@ -211,6 +211,7 @@ an operator inspect content and delivery evidence after an uncertain result.
| `review.verdict` | `{task_ref,request,verdict,body,decision?}`; action `review.verdict`. `verdict` is `approve` or `changes`. `request` must be a `review.requested` event on the task addressed to the caller's role (`request-not-found`), with no verdict yet (`verdict-exists`). Writes a `review.verdict` event `{task_ref,request,commit,verdict,message}` and sends the requesting role a REVIEW-RESULT in reply to the request's message. Returns `{verdict,message,commit}`. |
| `event.emit` | `{kind:'action.allowed',body:{action:'routine',target?},subject?}`; routine observations only. |
| `launch.revoke`, `launch.restore` | No args; outside-agent human CLI only. |
| `access.reload` | No args; the outside-agent human CLI only (row 295, #1801, FED R4): the tester and an agent get `human-required`, a reader `read-only`, a user `access-refused`. Reads the access file whatever its stamp and ends the lost claims of every business, since the file is shared, with `by` the human. Returns `{ended:[{thread,session,run}]}` for the caller's business only. `mosaic access grant` and `revoke` call it after the write. |
| `inbox` | No args; plain JSON projection scoped to capability's business. |
| `tasks` | Optional args, scoped to capability's business (row 89, #1583). None: every row of the tasks projection (below). `{task_ref}` alone: that one row, or `task-not-found`. Any of `{done,assignee,label,requirement,limit,after}`: `{tasks,total,next}`, the short form (below). An unknown arg or bad value refuses `invalid-request`. |
| `agents` | No args; each held role's latest claim row in the capability's business. Human and reader sessions also get `state`, `working`, `inputNeeded` and `stranded` (Agent state, below); a role session gets the claim rows only. |
@@ -305,8 +306,9 @@ Only a human principal matches a share: the business human, a reader session as
a user. An agent session matches none, `*` included, even when it is bound to the thread.
A reader session reads as the business human. A sharee can't call `session.list` on the
thread; it stays owner-only until row 193's share wiring. The check runs on every request against the
access document the broker reads then, so a removed, expired or unreadable grant or an
unshare ends reads on the next request. Only `thread.messages` admits a sharee; `thread.open`,
access document as of that request: the broker stats the file each time and reads it again only
when the stat changed (row 295), so a removed, expired or unreadable grant or an unshare ends
reads on the next request. Only `thread.messages` admits a sharee; `thread.open`,
`thread.send` and the share verbs refuse it `thread-not-owned`. `trail` and `replies` show a
user's thread messages to the owner and readers while the check holds. The tester can't call
the share verbs (`tester-refused`), and a viewer is refused them `access-refused`; a viewer
@@ -316,7 +318,22 @@ reads a thread shared with it. Tests: `tests/thread-shares.test.mjs`.
Per thread it returns `{thread,instance,pending,unread,lastMessage,lastOpen,lastClaim,holder}`:
messages to the instance not yet delivered, messages delivered to the current holder and
not yet read, the latest message, open and claim times, and the holding run or `null`. The
launcher reads it to start and stop thread sessions (`packages/cli/README.md`).
launcher reads it to start and stop thread sessions (`packages/cli/README.md`). A row adds
`revoked: true` when its thread's user owner has lost the grant (below); the launcher never
starts such a thread and sends a running one SIGTERM.
Revocation (row 295, #1801, FED R4, federation design 3.7). A user thread is lost when no grant
in the access file covers it at collaborator level, so a downgrade to viewer is a loss. One
transaction ends the held claim of every session of every lost thread, op `revoke`, reason
`access-revoked`; the run's next bus call refuses `not-holder`, and a new run of the thread
refuses `thread-revoked` at `bindLaunch` and at `role.claim`. A rebind of an earlier host's run is
allowed, so the launcher's recovery can end it. Four triggers end the claims, each with its `by`:
the runtime's timer, every 30 s (`access.intervalMs` sets it in process, as `librarySync.intervalMs`
does), which reads the file whatever its stamp (`access-timer`); a user's request or a
`threadsState` call after the file's stat changed (`access-change`); a `threadsState` row that is
lost with its claim still held, as when a grant expired with no write, from the same read
(`access-expiry`); and `access.reload` (the human). The timer calls `accessSweep()`, a trusted
broker method, not a verb. Tests: `tests/access-revoke.test.mjs`.
`threadIdle({business,thread,run})` is the launcher's idle stop, also trusted and not a
verb. It rechecks the thread and releases the run's claim in one call, and returns
+105 -13
View File
@@ -396,6 +396,12 @@ export class Broker {
// FED R3: () => the checked access document (R1's loadAccess), read once per user request. Null,
// a throw or no document leaves every user with no grant (design 3.5, fail closed).
#access;
// FED R4 (#1801, design 3.7): () => a key that changes when the access file does (dev, inode,
// size, mtime, ctime), or null. With it a request only stats the file; without it every read reads.
#accessStamp;
// The last read, { stamp, doc }, and the stamp whose losses #revokeLost has already ended.
#accessSeen = null;
#accessSwept;
#compactions;
// r6 X3: interrupted host grants whose remote undo waits for reconcileHosts, each with its claim.
#hostUndo = [];
@@ -422,13 +428,14 @@ export class Broker {
#clock = 0;
#processing;
// credentialStatus is the runtime's metadata-only Credentials.status(); the broker never gets use().
constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, library = null, usage = null, limits = null, harnessOn = null, credentialVerbs = null, hostGrant = null, hostKeys = null, processing = null, testerOn = null, roleFile = null, giteaGrant = null, giteaKeys = null, access = null, compactions = null }) {
constructor({ store, businesses, secretCheck = () => {}, credentialStatus = () => [], organization = null, project = null, library = null, usage = null, limits = null, harnessOn = null, credentialVerbs = null, hostGrant = null, hostKeys = null, processing = null, testerOn = null, roleFile = null, giteaGrant = null, giteaKeys = null, access = null, accessStamp = null, compactions = null }) {
if (!object(businesses) || !Object.keys(businesses).length) fail('invalid-business');
// Row 194 b15 (sage-2501): the function-argument checks come before anything reads the store.
if (harnessOn !== null && typeof harnessOn !== 'function') throw new TypeError('harnessOn must be a function');
if (testerOn !== null && typeof testerOn !== 'function') throw new TypeError('testerOn must be a function');
if (roleFile !== null && typeof roleFile !== 'function') throw new TypeError('roleFile must be a function');
if (access !== null && typeof access !== 'function') throw new TypeError('access must be a function');
if (accessStamp !== null && typeof accessStamp !== 'function') throw new TypeError('accessStamp must be a function');
if (compactions !== null && typeof compactions !== 'function') throw new TypeError('compactions must be a function');
this.#businesses = structuredClone(businesses);
this.#store = store;
@@ -449,6 +456,7 @@ export class Broker {
this.#giteaGrant = giteaGrant;
this.#giteaKeys = giteaKeys;
this.#access = access;
this.#accessStamp = accessStamp;
this.#compactions = compactions;
this.#project = project;
this.#library = library;
@@ -903,13 +911,7 @@ export class Broker {
// FED R3 (design 3.5): the user's live grants in the session's business, from one read of the
// access file. Any failure to read leaves none. Row 275: a grant at any level counts here; each
// verb's own check asks the level it needs (#granted).
#userGrants(s) {
let doc = null;
try {
doc = this.#access?.() ?? null;
} catch {
doc = null;
}
#userGrants(s, doc = this.#accessDoc()) {
const now = new Date();
return (doc?.grants ?? []).filter((g) => g.user === s.user && g.business === s.business &&
covers(g, { business: s.business, project: g.projects[0], instance: g.instances[0] }, now, { level: 'viewer' }));
@@ -965,15 +967,72 @@ export class Broker {
// Does a live grant of any level cover thread `t` for `user`? One read of the access file, as
// #userGrants does; any failure to read covers nothing (design 5.6, H3).
#covered(user, business, t) {
const doc = this.#accessDoc();
const now = new Date();
return (doc?.grants ?? []).some((g) => g.user === user &&
covers(g, { business, project: t.project, instance: t.instance }, now, { level: 'viewer' }));
}
// FED R4 (#1801, design 3.7): the access document, from one read per change of the file. A stamp
// equal to the last read's reuses that read; `force` (the timer, access.reload) reads regardless.
// A read that throws, or no access function, is no document: every user loses every grant.
#accessDoc(force = false) {
let stamp = null;
try {
stamp = this.#accessStamp?.() ?? null;
} catch {
stamp = null;
}
if (!force && stamp !== null && this.#accessSeen?.stamp === stamp) return this.#accessSeen.doc;
let doc = null;
try {
doc = this.#access?.() ?? null;
} catch {
doc = null;
}
this.#accessSeen = { stamp, doc };
return doc;
}
// Is thread `t` owned by a user whom no grant in `doc` covers at collaborator level any more? A
// downgrade to viewer is a loss (design 3.7, morgana-0005). The human's and the tester's never are.
#revoked(t, doc, now) {
if (!t.owner.startsWith('user:')) return false;
const user = t.owner.slice(5);
return !(doc?.grants ?? []).some((g) => g.user === user &&
covers(g, { business: t.business, project: t.project, instance: t.instance }, now, { level: 'collaborator' }));
}
// Design 3.7: on a loss, one transaction ends the held claim of every session of every user thread
// the access file no longer covers, in every business. The run's next bus call refuses not-holder
// (F10), and a later claim refuses thread-revoked. Runs only when the read is new: a forced read,
// a stamp it hasn't swept, or no stamp at all. `doc` is a read the caller just made, so a request
// reads once. Returns the ended claims.
#revokeLost(by, force = false, doc = this.#accessDoc(force)) {
const stamp = this.#accessSeen.stamp;
if (!force && stamp !== null && stamp === this.#accessSwept) return [];
const now = new Date();
return (doc?.grants ?? []).some((g) => g.user === user &&
covers(g, { business, project: t.project, instance: t.instance }, now, { level: 'viewer' }));
const lost = this.#store.all("SELECT * FROM threads WHERE substr(owner,1,5)='user:' ORDER BY seq").filter((t) => this.#revoked(t, doc, now));
const ended = [];
if (lost.length) this.#store.transaction(() => {
for (const t of lost)
for (const h of this.#threadHolders(t.business, t.instance, t.owner))
if (h.thread === t.id) {
this.#claimEnd({ business: t.business }, h, 'revoke', null, 'access-revoked', by);
ended.push(h);
}
});
this.#accessSwept = stamp;
return ended;
}
// Is thread `id` of `business` revoked, from the file as of now? Only reads: role.claim runs it
// inside the verb's transaction, so the claim end is left to the next sweep.
#threadLost(business, id) {
const t = this.#store.get('SELECT * FROM threads WHERE id=? AND business=?', id, business);
// Only a user's thread reads the file, so a human's or tester's run never does.
return Boolean(t?.owner.startsWith('user:')) && this.#revoked(t, this.#accessDoc(), new Date());
}
// Trusted host API, not a socket verb: the runtime's 30 s timer (design 3.7 trigger 2). It reads
// the file whatever its stamp, so an expiry with no write and no request ends its runs too.
accessSweep() {
return this.#revokeLost('access-timer', true).map((h) => ({ business: h.business, thread: h.thread, session: h.session, run: h.holder_run }));
}
// The threads another principal owns that the session may read, oldest first.
#readableThreads(s, project) {
@@ -1033,9 +1092,23 @@ export class Broker {
// (a turn in progress), and the latest message, open and claim times. A message delivered
// to an earlier run and never read (it crashed mid-turn) isn't the holder's turn, so it
// doesn't keep the holder from idling out.
// FED R4 (design 3.7): a row adds revoked: true for a thread the access file no longer covers, as
// of this call, so a launcher restart never starts one. A changed file also ends its lost runs here.
// A grant that expired with no change to the file (Dewey's B1 in dewey-0871) leaves a revoked row
// whose claim the stamp check didn't end: that row ends the lost claims from the same read, by
// access-expiry, before the launcher sees it, so the run's end is never a release.
threadsState(business) {
this.#business(identifier(business));
return this.#threadRows('t.business=?', business);
this.#revokeLost('access-change');
const doc = this.#accessSeen?.doc ?? null;
const now = new Date();
const lost = (r, owner, project) => this.#revoked({ business, owner, project, instance: r.instance }, doc, now);
let rows = this.#threadRows('t.business=?', business);
if (rows.some(({ owner, project, ...r }) => r.holder !== null && lost(r, owner, project))) {
this.#revokeLost('access-expiry', true, doc);
rows = this.#threadRows('t.business=?', business);
}
return rows.map(({ owner, project, ...r }) => (lost(r, owner, project) ? { ...r, revoked: true } : r));
}
// Row 121 round 2 (#1615, Dewey's F3 in 28591): the launcher's idle stop. The tick decides
// from a threadsState snapshot, and a message the run took after it would be delivered to a
@@ -1203,7 +1276,7 @@ export class Broker {
#threadRows(where, ...params) {
return this.#store
.all(
`SELECT t.id, t.instance, se.id AS session, se.op AS session_op,
`SELECT t.id, t.instance, t.owner, t.project, se.id AS session, se.op AS session_op,
(SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=se.id AND m.to_role=t.instance
AND NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op IN ('delivered','read'))
AND NOT EXISTS(SELECT 1 FROM message_cancels x WHERE x.message=m.id)) AS pending,
@@ -1221,6 +1294,8 @@ export class Broker {
.map((r) => ({
thread: r.id,
instance: r.instance,
owner: r.owner,
project: r.project,
pending: r.pending,
unread: r.unread,
lastMessage: r.last_message,
@@ -1285,6 +1360,9 @@ export class Broker {
// harness was switched off, so it can be ended.
if (!old) this.#processingWork();
if (!old && !this.#harnessOn(record.business, record.harness)) fail('invalid-harness');
// FED R4 (design 3.7, D1): a new run of a revoked thread refuses, which covers a start racing the
// revocation. recover()'s rebind of an earlier run isn't refused, so it can still end that run.
if (!old && record.thread !== undefined && this.#threadLost(record.business, record.thread)) fail('thread-revoked');
if (old) {
if (old.actor_role !== record.role || old.body !== JSON.stringify(body)) fail('launch-record-mismatch');
// An ended run stays ended across broker restarts.
@@ -4450,8 +4528,12 @@ export class Broker {
// request reads the access file again, so a removed or expired grant refuses the next one.
let grants = null;
if (s.user) {
// FED R4 (sage-3139): any user's request that finds the file changed ends the lost runs. One
// read serves the sweep and the grants, and none at all while the file's stamp is unchanged.
const doc = this.#accessDoc();
this.#revokeLost('access-change', false, doc);
if (!USER_VERBS.includes(request.verb)) fail('access-refused');
grants = this.#userGrants(s);
grants = this.#userGrants(s, doc);
if (!grants.length) fail('access-revoked');
}
const args = request.args ?? {};
@@ -4463,6 +4545,14 @@ export class Broker {
keys(args, []);
return this.#processingState();
}
// FED R4 (#1801, design 3.7 trigger 3): the human's re-read of the access file, which R5's grant
// and revoke call after the write. Human-only, so not a tester, a reader, an agent or a user.
if (request.verb === 'access.reload') {
this.#human(s);
keys(args, []);
const ended = this.#revokeLost(s.human, true).filter((h) => h.business === s.business);
return { ended: ended.map((h) => ({ thread: h.thread, session: h.session, run: h.holder_run })) };
}
this.#processingGate(s, request.verb);
// Row 213 (C2-08, Case 2 design 5.1 M6): a creative role has no free-text decision channel. It
// runs before the spec-cited branch below, so a cited raise refuses too.
@@ -5154,6 +5244,8 @@ export class Broker {
if (s.thread) {
if (this.#store.get("SELECT 1 FROM thread_claims WHERE session=? AND holder_run=? AND op='revoke' LIMIT 1", s.session, s.run))
fail('run-revoked');
// FED R4 (design 3.7): no run takes a thread the access file no longer covers.
if (this.#threadLost(s.business, s.thread)) fail('thread-revoked');
// Row 271 (264d, B2): no run takes a session being cleared, or a retired one.
this.#sessionFence(s.session);
if (this.#sessionHolder(s.session)?.op === 'claim') fail('thread-already-held');
+32 -4
View File
@@ -1,4 +1,4 @@
import { readdirSync } from 'node:fs';
import { lstatSync, readdirSync } from 'node:fs';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { Store } from './store.mjs';
@@ -37,6 +37,8 @@ import { createContainerStopper } from '../../processing-control/src/containers.
// Row 194 (D3): the verbs the instance writer commits, whose answers carry restart.
const INSTANCE_VERBS = new Set(['instances.set', 'veto.set', 'instance.hire', 'instance.fire']);
// FED R4 (#1801, design 3.7): the revocation timer's period.
export const ACCESS_SWEEP_MS = 30_000;
// Trusted host API. S1 supplies resolved definitions, S6 supplies launch records and ends them.
// Human limits writes use the shared organization writer behind the verified transport.
// `tasks` is S3's adapter factory: ({broker, credentials, businesses}) => {handle, timeout, close}.
@@ -101,16 +103,18 @@ export async function startBroker({
// a librarySync without one is refused.
librarySync = null,
// FED R3 (#414): the access file and its loader context, { file, context } (R1's loadAccess).
// Without it no user binds. R6 and R8 pass it; the live boot config doesn't yet.
// Without it no user binds. R6 and R8 pass it; the live boot config doesn't yet. FED R4 (#1801):
// intervalMs, the revocation timer's period (design 3.7, 30 s), in-process only like librarySync's.
access = null,
}) {
if (access !== null && (typeof access?.file !== 'string' || access.context === null || typeof access.context !== 'object'))
if (access !== null && (typeof access?.file !== 'string' || access.context === null || typeof access.context !== 'object'
|| (Object.hasOwn(access, 'intervalMs') && !(Number.isSafeInteger(access.intervalMs) && access.intervalMs > 0))))
throw new BusError('invalid-host-request');
if (librarySync !== null && (typeof librarySync?.command !== 'function'
|| (Object.hasOwn(librarySync, 'intervalMs') && !(Number.isSafeInteger(librarySync.intervalMs) && librarySync.intervalMs > 0))
|| (Object.hasOwn(librarySync, 'configDir') && typeof librarySync.configDir !== 'string')))
throw new BusError('invalid-host-request');
let store, credentials, server, adapter, processing, syncTimer = null, syncPass = null;
let store, credentials, server, adapter, processing, syncTimer = null, syncPass = null, accessTimer = null;
try {
const references = {};
for (const [business, b] of Object.entries(businesses)) {
@@ -262,6 +266,16 @@ export async function startBroker({
broker = new Broker({
// FED R3: the access file is read per user request; a refusal leaves the user no grant.
access: access ? () => loadAccess(access.file, access.context) : null,
// FED R4 (#1801, design 3.7 trigger 1): a request stats the file and reads it only when this
// changes. A failed stat is a stamp of its own, so a removed file reads (and refuses) once.
accessStamp: access ? () => {
try {
const st = lstatSync(access.file, { bigint: true });
return `${st.dev}:${st.ino}:${st.size}:${st.mtimeNs}:${st.ctimeNs}:${st.mode}:${st.uid}`;
} catch (e) {
return `unreadable:${e.code ?? 'error'}`;
}
} : null,
// Row 270 (264c): session.list's compaction count, read from the session store's ledger.
compactions: ({ business, thread, session, legacy }) => {
const dir = storeDir(dataRoot, business, thread, session, { legacy });
@@ -558,6 +572,18 @@ export async function startBroker({
syncTimer.unref();
}
const stopSync = async () => { clearInterval(syncTimer); syncTimer = null; await syncPass; };
// FED R4 (#1801, design 3.7 trigger 2): every 30 s the broker re-reads the access file, so a hand
// edit or an expiry with no request ends the lost runs within one period.
if (access) {
accessTimer = setInterval(() => {
try {
broker.accessSweep();
} catch (e) {
process.stderr.write(`access: ${JSON.stringify({ event: 'access-sweep-failed', code: e.code ?? 'error' })}\n`);
}
}, access.intervalMs ?? ACCESS_SWEEP_MS);
accessTimer.unref();
}
return {
broker,
credentials,
@@ -605,6 +631,7 @@ export async function startBroker({
launches: bound,
readers: readCaps,
async close() {
clearInterval(accessTimer);
await stopSync();
await server.close();
await adapter?.close?.();
@@ -614,6 +641,7 @@ export async function startBroker({
},
};
} catch (e) {
clearInterval(accessTimer);
await server?.close();
await adapter?.close?.();
await processing?.close();
+272
View File
@@ -0,0 +1,272 @@
// FED R4 (#1801, federation design 3.7 and the R4 row of section 7): revocation. A loss of a user's
// grant ends the claim of every run on that user's threads, from one of three triggers: the timer's
// accessSweep, the human's access.reload, and the next request of any user once the file changed.
// The access document and its stamp come in as functions, as the runtime passes loadAccess and an
// lstat key; `edit` changes the document and the stamp together, as a hand edit does. BUILD.md
// (agents/morgana/work/queue-295) names the mutant each test kills.
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { Store } from '../src/store.mjs';
import { Broker } from '../src/broker.mjs';
import { ACCESS_SWEEP_MS, startBroker } from '../src/runtime.mjs';
const shipped = (name) =>
JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8'));
const grant = (over = {}) => ({
id: `g-${over.user ?? 'alice'}`, user: 'alice', business: 'demo', projects: ['*'], instances: ['assistant'], level: 'collaborator', ...over,
});
const BUSINESSES = {
demo: {
id: 'demo', human: 'jason', arbiters: { technical: 'chief', delivery: 'chief' }, vars: { 'bus.tester': true }, plugins: { harness: ['pi'] },
roles: {
chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } },
assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } },
},
definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging } },
},
};
// Alice and Bob each hold a collaborator grant and own a thread; the human owns thread H. With
// `runs`, each thread has a claimed run: a-run on A, b-run on B, h-run on H.
function setup(t, { runs = true, grants = [grant(), grant({ user: 'bob' })] } = {}) {
const root = mkdtempSync(join(tmpdir(), 'bus-access-revoke-'));
const store = new Store(root);
t.after(() => {
store.close();
rmSync(root, { recursive: true, force: true });
});
// `reads` counts loader calls; `rev` is the stamp, which a hand edit changes.
const access = { doc: { accessVersion: 1, revision: 1, users: [], grants }, reads: 0, rev: 1 };
const broker = () => new Broker({
store,
businesses: BUSINESSES,
access: () => {
access.reads++;
return access.doc;
},
accessStamp: () => String(access.rev),
});
const b = broker();
const caps = {
human: b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }),
tester: b.bindTester({ business: 'demo' }),
reader: b.bindReader({ business: 'demo' }),
};
const call = (who, verb, args = {}) => b.request(caps[who], { verb, args });
const code = (who, verb, args = {}) => {
try {
call(who, verb, args);
} catch (e) {
return e.code;
}
return null;
};
const edit = (fn) => {
access.doc = { ...access.doc, revision: access.doc.revision + 1, grants: fn(access.doc.grants) };
access.rev++;
};
const record = (name, thread) => ({ business: 'demo', role: 'assistant', run: `${name}-run`, harness: 'pi', thread });
const launch = (name, thread) => {
caps[name] = b.bindLaunch(record(name, thread));
return call(name, 'role.claim');
};
// The latest claim row of a run: its op, reason and by, spread off the store's null prototype.
const claim = (run) => ({ ...store.get('SELECT op, reason, by FROM thread_claims WHERE holder_run=? ORDER BY seq DESC LIMIT 1', run) });
const P = call('human', 'project.create', { name: 'shared' }).id;
caps.alice = b.bindUser({ business: 'demo', user: 'alice' });
caps.bob = b.bindUser({ business: 'demo', user: 'bob' });
const A = call('alice', 'thread.create', { project: P, name: 'alice work' }).id;
const B = call('bob', 'thread.create', { project: P, name: 'bob work' }).id;
const H = call('human', 'thread.create', { project: P, name: 'owner work' }).id;
if (runs) {
launch('a', A);
launch('b', B);
launch('h', H);
}
return { b, store, access, caps, call, code, edit, record, launch, claim, broker, P, A, B, H };
}
const withoutAlice = (gs) => gs.filter((g) => g.user !== 'alice');
test('R4: the timer sweep after a hand edit ends only the lost user\'s run; its next call refuses not-holder', (t) => {
const { b, code, edit, claim, A } = setup(t);
edit(withoutAlice);
const ended = b.accessSweep();
assert.equal(ended.length, 1);
assert.deepEqual({ ...ended[0], session: undefined }, { business: 'demo', thread: A, session: undefined, run: 'a-run' });
assert.equal(typeof ended[0].session, 'string');
assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-timer' });
assert.equal(code('a', 'message.receive'), 'not-holder', 'the ended run');
assert.equal(code('b', 'message.receive'), null, 'another user\'s run stays');
assert.equal(code('h', 'message.receive'), null, 'the human\'s run stays');
assert.deepEqual([claim('b-run').op, claim('h-run').op], ['claim', 'claim']);
// Ended once: a second sweep finds nothing to end.
assert.deepEqual(b.accessSweep(), []);
});
test('R4: an expiry with no write and no request ends the run at the next timer sweep', (t) => {
t.mock.timers.enable({ apis: ['Date'], now: Date.parse('2026-10-23T00:00:00Z') });
const { b, code, claim } = setup(t, { grants: [grant({ expires: '2026-10-23T00:01:00Z' }), grant({ user: 'bob' })] });
assert.deepEqual(b.accessSweep(), [], 'live');
t.mock.timers.tick(60_001);
// Bob's request sees the same stamp, so it doesn't sweep: the timer is what catches an expiry.
assert.equal(code('bob', 'thread.list'), null);
assert.equal(claim('a-run').op, 'claim');
assert.deepEqual(b.accessSweep().map((h) => h.run), ['a-run']);
assert.equal(code('a', 'message.receive'), 'not-holder');
assert.equal(code('b', 'message.receive'), null);
});
test('R4: a downgrade to viewer is a loss', (t) => {
const { b, code, edit, claim } = setup(t);
edit((gs) => gs.map((g) => (g.user === 'alice' ? { ...g, level: 'viewer' } : g)));
assert.deepEqual(b.accessSweep().map((h) => h.run), ['a-run']);
assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-timer' });
assert.equal(code('b', 'message.receive'), null);
});
// sage-3139 Q1: no timer runs here, so only the request trigger can end alice's run.
test('R4: another user\'s request after a hand edit ends the lost user\'s run, and its own run stays', (t) => {
const { code, edit, claim } = setup(t);
edit(withoutAlice);
assert.equal(claim('a-run').op, 'claim', 'nothing ends before a request');
assert.equal(code('a', 'message.receive'), null, 'an agent request is not a trigger');
assert.equal(code('bob', 'thread.list'), null);
assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-change' });
assert.equal(code('a', 'message.receive'), 'not-holder');
assert.equal(code('b', 'message.receive'), null);
assert.equal(code('h', 'message.receive'), null);
});
// sage-3139 Q1, the cost rule: a request only stats the file, and reads it once per change.
test('R4: a request reads the access file only when its stamp changed, once per change', (t) => {
const { access, code, edit } = setup(t);
const reads = access.reads;
assert.equal(code('bob', 'thread.list'), null);
assert.equal(code('alice', 'project.list'), null);
assert.equal(access.reads, reads, 'unchanged: no read');
edit((gs) => gs);
assert.equal(code('bob', 'thread.list'), null);
assert.equal(code('alice', 'thread.list'), null);
assert.equal(access.reads, reads + 1, 'changed: one read for both requests');
});
test('R4: access.reload ends the lost runs at once for the human; the tester, a reader, an agent and a user are refused', (t) => {
const { b, code, call, edit, claim, A } = setup(t);
assert.equal(code('tester', 'access.reload'), 'human-required', 'tester');
assert.equal(code('reader', 'access.reload'), 'read-only', 'reader');
assert.equal(code('b', 'access.reload'), 'human-required', 'agent');
assert.equal(code('bob', 'access.reload'), 'access-refused', 'user');
assert.deepEqual(call('human', 'access.reload'), { ended: [] }, 'nothing lost');
assert.equal(code('human', 'access.reload', { force: true }), 'invalid-request');
edit(withoutAlice);
const { ended } = call('human', 'access.reload');
assert.deepEqual(ended.map((h) => [h.thread, h.run]), [[A, 'a-run']]);
assert.equal(typeof ended[0].session, 'string');
assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'jason' });
assert.equal(code('a', 'message.receive'), 'not-holder');
assert.equal(code('b', 'message.receive'), null);
assert.deepEqual(b.accessSweep(), []);
});
test('R4: threadsState marks revoked: true on the lost user\'s thread only, a pending message included', (t) => {
const { b, call, edit, A, B, H } = setup(t, { runs: false });
call('alice', 'thread.send', { thread: A, body: 'for A' });
call('bob', 'thread.send', { thread: B, body: 'for B' });
const row = (id) => b.threadsState('demo').find((x) => x.thread === id);
assert.equal(row(A).revoked, undefined, 'covered');
edit(withoutAlice);
assert.equal(row(A).revoked, true);
assert.equal(row(A).pending, 1, 'its message is still pending');
assert.equal(Object.hasOwn(row(B), 'revoked'), false);
assert.equal(Object.hasOwn(row(H), 'revoked'), false);
assert.equal(Object.hasOwn(row(A), 'owner'), false, 'the row keeps its old fields');
});
// Dewey's B1 (dewey-0871): the launcher ticks faster than the timer, so a grant that expires with no
// write must not reach it as revoked: true with the claim still held.
test('R4: threadsState after an expiry with no write and no sweep ends the claim before it marks the row revoked', (t) => {
t.mock.timers.enable({ apis: ['Date'], now: Date.parse('2026-10-23T00:00:00Z') });
const { b, code, claim, store, A, B } = setup(t, { grants: [grant({ expires: '2026-10-23T00:01:00Z' }), grant({ user: 'bob' })] });
const row = (id) => b.threadsState('demo').find((x) => x.thread === id);
assert.equal(row(A).revoked, undefined, 'live');
t.mock.timers.tick(60_001);
const a = row(A);
assert.deepEqual([a.revoked, a.holder], [true, null]);
assert.deepEqual(claim('a-run'), { op: 'revoke', reason: 'access-revoked', by: 'access-expiry' });
assert.equal(code('a', 'message.receive'), 'not-holder');
assert.deepEqual([row(B).holder, claim('b-run').op], ['b-run', 'claim']);
// Ended once: the next call writes no second revoke.
row(A);
assert.equal(store.get("SELECT count(*) AS n FROM thread_claims WHERE holder_run='a-run' AND op='revoke'").n, 1);
});
test('R4: a new run of a revoked thread refuses thread-revoked at bind and at claim; a rebind is allowed', (t) => {
const { b, call, edit, record, broker, A, B } = setup(t, { runs: false });
const late = b.bindLaunch(record('late', A));
edit(withoutAlice);
assert.throws(() => b.request(late, { verb: 'role.claim' }), { code: 'thread-revoked' }, 'bound before, claimed after');
assert.throws(() => b.bindLaunch(record('new', A)), { code: 'thread-revoked' }, 'bound after');
assert.ok(b.bindLaunch(record('other', B)), 'bob\'s thread binds');
call('bob', 'thread.list');
// D1: a later broker over the same store rebinds the run an earlier one bound (recover()), so the
// host can end it.
const again = broker();
assert.ok(again.bindLaunch(record('late', A)));
assert.equal(again.endLaunch({ business: 'demo', run: 'late-run', reason: 'recovered' }).released, false);
});
// The runtime's timer (design 3.7 trigger 2) on a real 0600 access file read by loadAccess. The mocked
// setInterval stands in for 30 s; the run is watched through its own calls, since threadsState and a
// user request would sweep by themselves.
test('R4 runtime: a hand edit ends the run at the first timer period, intervalMs is checked, and close stops the timer', async (t) => {
t.mock.timers.enable({ apis: ['setInterval'] });
const root = mkdtempSync(join(tmpdir(), 'bus-access-timer-'));
t.after(() => rmSync(root, { recursive: true, force: true }));
const dataRoot = join(root, 'data');
mkdirSync(dataRoot, { mode: 0o700 });
const file = join(root, 'access.json');
const write = (grants) => writeFileSync(file, JSON.stringify({
accessVersion: 1, revision: 1, users: [{ id: 'alice', home: 'local', subjects: [{ provider: 'authentik', sub: 'alice-sub' }] }], grants,
}), { mode: 0o600 });
write([grant()]);
const context = { providers: ['authentik'], peers: [], businesses: { demo: { human: 'jason', instances: ['chief', 'assistant'] } }, ssoSubjects: [] };
for (const intervalMs of [0, -1, 1.5, '30000'])
await assert.rejects(startBroker({ dataRoot, businesses: BUSINESSES, access: { file, context, intervalMs } }), { code: 'invalid-host-request' }, String(intervalMs));
const runtime = await startBroker({ dataRoot, businesses: BUSINESSES, access: { file, context } });
let closed = false;
t.after(() => closed || runtime.close());
const b = runtime.broker;
const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
const alice = b.bindUser({ business: 'demo', user: 'alice' });
const P = b.request(human, { verb: 'project.create', args: { name: 'shared' } }).id;
const A = b.request(alice, { verb: 'thread.create', args: { project: P, name: 'alice work' } }).id;
const run = b.bindLaunch({ business: 'demo', role: 'assistant', run: 'a-run', harness: 'pi', thread: A });
b.request(run, { verb: 'role.claim' });
const receive = () => {
try {
b.request(run, { verb: 'message.receive' });
} catch (e) {
return e.code;
}
return null;
};
write([]);
t.mock.timers.tick(ACCESS_SWEEP_MS - 1);
assert.equal(receive(), null, 'before the first period');
t.mock.timers.tick(1);
assert.equal(receive(), 'not-holder', 'at the first period');
// After close the timer is gone: the sweep is never called again.
let sweeps = 0;
b.accessSweep = () => void sweeps++;
t.mock.timers.tick(ACCESS_SWEEP_MS);
assert.equal(sweeps, 1, 'one more period, one sweep');
closed = true;
await runtime.close();
t.mock.timers.tick(ACCESS_SWEEP_MS * 3);
assert.equal(sweeps, 1, 'closed');
});
+16 -3
View File
@@ -15,6 +15,14 @@ import { fixture } from './spec-fixtures.mjs';
const AT = '2026-10-21T00:00:00.000Z';
const within = (grants) => ({ authority: { withinRole: grants, crossRole: [] } });
const options = [{ key: 'yes', text: 'Yes' }, { key: 'no', text: 'No' }];
// FED R4 (row 295, mos-1436): a new run on a user's thread binds only while a grant covers it, so the
// two tests that launch on alice's thread pass this access function; no other test's broker has one.
// The grant names only the test's project, which the test sets in `ref` once it exists; the broker
// has no stamp here, so it calls the function again on each read.
const aliceAccess = (ref) => () => ({
accessVersion: 1, revision: 1, users: [],
grants: [{ id: 'g-alice', user: 'alice', business: 'demo', projects: [ref.project], instances: ['assistant'], level: 'collaborator' }],
});
// Row 264a (#1769): a thread made by hand gets the default session the 3k-3l migration gives a
// legacy thread, with the thread's id.
const DEFAULT_SESSION = [
@@ -55,7 +63,7 @@ const fromAlice = (store, thread, body, extra = {}) =>
const toAlice = (store, thread, body, extra = {}) =>
put(store, { thread, from_role: 'assistant', from_run: 'assistant-run', to_role: 'user', body, ...extra });
function setup(t) {
function setup(t, { access } = {}) {
const root = mkdtempSync(join(tmpdir(), 'bus-message-filter-'));
const store = new Store(root);
t.after(() => {
@@ -64,6 +72,7 @@ function setup(t) {
});
const b = new Broker({
store,
access,
businesses: {
demo: {
id: 'demo', human: 'jason', arbiters: { technical: 'chief', delivery: 'chief' }, vars: { 'bus.tester': true },
@@ -115,7 +124,9 @@ test('replies on a hidden message id answers exactly as on an unknown id, for th
});
test('trail on a decision keeps the decision and drops the user thread messages, their deliveries and their events', (t) => {
const { b, store, call, U } = setup(t);
const ref = {};
const { b, store, call, project, U } = setup(t, { access: aliceAccess(ref) });
ref.project = project;
const run = b.bindLaunch({ business: 'demo', role: 'assistant', run: 'assistant-run', harness: 'pi', thread: U });
b.request(run, { verb: 'role.claim' });
const d = b.request(run, {
@@ -163,8 +174,10 @@ test('messages in the business human and tester threads are not hidden (row 177
// Row 193's user sentence, one assertion per item: "The business owner sees your thread's agent's
// decisions, inbox entries, claims, tasks and project files, but not your messages."
test('governance pin: the owner sees decisions, inbox entries, claims, tasks and project files, but not the user messages', async (t) => {
const f = await fixture(t, { roles: { assistant: { definition: 'assistant', ...within(['message.send', 'record.write']) } } });
const ref = {};
const f = await fixture(t, { roles: { assistant: { definition: 'assistant', ...within(['message.send', 'record.write']) } }, access: aliceAccess(ref) });
const project = (await f.call(f.human, 'project.create', { name: 'alice work' })).id;
ref.project = project;
const U = threadOf(f.store, project, 'user:alice', 'alice');
const run = f.broker.bindLaunch({ business: 'demo', role: 'assistant', run: 'assistant-run', harness: 'pi', thread: U });
await f.call(run, 'role.claim');
+2 -2
View File
@@ -7,7 +7,7 @@ import { Broker } from '../src/broker.mjs';
import { initProject, readSpec, writeRecord, listProjectFiles, readProjectFile, writeHumanFile, ProjectError } from '../../project/src/index.mjs';
export const git = (repo, ...args) => execFileSync('git', ['-c', 'core.hooksPath=/dev/null', '-C', repo, ...args], { encoding: 'utf8' }).trim();
const options = [{ key: 'yes', text: 'Yes' }, { key: 'no', text: 'No' }];
export async function fixture(t, { adapter = true, roles = {}, arbiters = { technical: 'reviewer', delivery: 'reviewer' } } = {}) {
export async function fixture(t, { adapter = true, roles = {}, arbiters = { technical: 'reviewer', delivery: 'reviewer' }, access = null } = {}) {
const root = mkdtempSync(join(tmpdir(), 'bus-spec-'));
const store = new Store(root);
t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
@@ -19,7 +19,7 @@ export async function fixture(t, { adapter = true, roles = {}, arbiters = { tech
...roles,
} };
const { repo, commit } = await initProject({ business, project: 'stack', dataRoot: root });
const broker = new Broker({ store, businesses: { demo: business }, project: adapter ? {
const broker = new Broker({ store, businesses: { demo: business }, access, project: adapter ? {
readSpec: context => readSpec({ ...context, dataRoot: root }),
writeRecord: context => writeRecord({ ...context, dataRoot: root }),
listFiles: context => listProjectFiles({ ...context, dataRoot: root }),
+16 -1
View File
@@ -36,6 +36,9 @@ list is the real limit: a collaborator's message drives a granted instance with
the owner's role.
To cut a user off, revoke them here first. It acts at once. Removing a provider
from console.sso takes effect only after a Console restart, so do that second.
grant and revoke ask the bus to re-read the file after the write. If the bus
refuses, the write still stands (exit 0), stderr names the refusal, and the bus's
timer applies the change within 30 s.
user remove refuses while the user has a grant at any business: revoke those first.
A user's home is local or a peer id from federation.peers in the system config;
any other home refuses. A grant that names a sign-in subject needs console.sso
@@ -179,7 +182,7 @@ export function withLock(file, fn, fs = realFs) {
// The one human-proof call: project.list, which the bus gates with #human
// (broker.mjs:2319, and #human at :377). It also returns the rows that resolve
// project names. Row R4 swaps this call for access.reload.
// project names. R4 (#1801) keeps it, since grant needs those rows, and calls access.reload after the write.
async function proveHuman(transport, business) {
return transport(business)("project.list", {});
}
@@ -283,6 +286,18 @@ async function accessVerb(args, io, deps) {
grants: result.touched,
});
io.stdout.write(`access ${sub}: ${result.touched.join(", ")} at ${business}, revision ${result.checked.revision}\n`);
await reload(deps.transport, business, io);
}
// FED R4 (#1801, design 3.7 trigger 3): after a grant or revoke, the bus re-reads the file and ends
// the runs a loss leaves uncovered, at once. The write already stands, so a refusal here is exit 0
// (sage-3139): a nonzero exit invites a retry that would hit revision-conflict.
async function reload(transport, business, io) {
try {
await transport(business)("access.reload", {});
} catch (error) {
io.stderr.write(`access.reload refused: ${error.code ?? error.message}; the bus's timer applies the change within 30 s\n`);
}
}
// `user add ID` and `user remove ID` (row 193c): the same lock, revision check, file checks,
+4 -3
View File
@@ -85,9 +85,10 @@ function trackerFor(system, b, warn) {
}
// FED R6 (#1800, sage-3142): the runtime's `access`, { file, context }. The broker reads the file
// with R1's loadAccess on every user request, so a bad file leaves every user with no grant and
// the owner's bus still boots (sage-3153). The context is the cli's accessContextFor, peers included. No file,
// or a context that can't be built, means no `access`: no user binds, as before R6.
// with R1's loadAccess on a user request that finds the file's stat changed, and on its 30 s timer
// (FED R4, #1801), so a bad file leaves every user with no grant and the owner's bus still boots
// (sage-3153). The context is the cli's accessContextFor, peers included. No file, or a context
// that can't be built, means no `access`: no user binds, as before R6.
function accessFor(system, env, warn) {
const file = accessFilePath(configDir(env));
try {
+17
View File
@@ -70,6 +70,9 @@
// after the snapshot gets a turn from the same run or waits for the
// next one, instead of going to the stopping run. An idle thread holds
// no session and counts against no limit.
// - FED R4 (#1801): never starts a thread the broker marks revoked, and
// SIGTERMs its running session at once (the broker already ended the
// claim). bindLaunch refuses a new run of one with thread-revoked.
//
// A thread run's sessionDir is the thread's store (session.mjs
// threadStore()), so the next run for the thread resumes the harness
@@ -698,6 +701,20 @@ export function createLauncher({
const key = `${t.thread}/${t.session}`;
if (clearing.has(t.session)) continue;
const s = registry.running(b.id, t.instance, t.thread, t.session);
// FED R4 (#1801, design 3.7): the broker marks a thread whose user owner lost the grant, and has
// ended its claim before it returns the row, whether the file changed or a grant expired. A
// revoked thread never starts, and a running one gets the idle stop's start-time-checked
// SIGTERM at once, with no threadIdle: there is no claim left to release.
if (t.revoked) {
if (s && !stopping.has(s.run)) {
stopping.add(s.run);
record({ event: "thread-revoked", instance: t.instance, thread: t.thread, session: t.session, run: s.run });
try {
if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM");
} catch {}
}
continue;
}
if (s) {
const last = Math.max(ms(s.startedAt), ms(t.lastMessage), ms(t.lastOpen));
if (!stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) {
+59
View File
@@ -449,3 +449,62 @@ test("T9 a bus that refuses the human proof blocks both user verbs", async (t) =
assert.equal(readFileSync(s.file, "utf8"), before);
assert.equal(s.receipts(), "");
});
// FED R4 (#1801, design 3.7 trigger 3): grant and revoke ask the bus to re-read the file after the
// write. This stand-in CLI logs every call, one line each, and with FAKE_MODE "refuse-reload"
// refuses access.reload alone, as a bus would refuse a call it can't take. BUILD.md
// (agents/morgana/work/queue-295) names the mutant each test kills.
const RELOAD_BODY = `
import { appendFileSync, readFileSync } from "node:fs";
const input = JSON.parse(readFileSync(0, "utf8"));
appendFileSync(process.env.FAKE_LOG, JSON.stringify(input) + "\\n");
if (input.verb === "access.reload") {
if (process.env.FAKE_MODE === "refuse-reload") { process.stderr.write("human-required\\n"); process.exit(3); }
process.stdout.write(JSON.stringify({ ended: [] }));
} else process.stdout.write(JSON.stringify(${JSON.stringify(ROWS)}));
`;
function reloading(t, mode = "ok") {
const s = setup(t);
mkdirSync(join(s.root, "reload"));
const cli = fakeCli(join(s.root, "reload"), RELOAD_BODY);
const env = humanEnv({ FAKE_LOG: s.log, FAKE_MODE: mode });
s.deps.transport = (business) => humanTransport({ socket: join(s.root, "bus.sock"), business, env, cli, timeoutMs: 5000 });
const calls = () => readFileSync(s.log, "utf8").trim().split("\n").map((l) => JSON.parse(l));
return { ...s, calls };
}
test("R4 grant and revoke call access.reload after the write, and nothing else does", async (t) => {
const s = reloading(t);
await run([...GRANT, "--revision", "1"], s.deps);
assert.deepEqual(s.calls().map((c) => [c.verb, c.business, c.args]), [["project.list", "acme", {}], ["access.reload", "acme", {}]]);
await run(["revoke", "--business", "acme", "--user", "alice", "--revision", "2"], s.deps);
await run(["list", "--business", "acme"], s.deps);
assert.deepEqual(s.calls().map((c) => c.verb), ["project.list", "access.reload", "project.list", "access.reload", "project.list"]);
// A refused write never asks.
await refusal(run([...GRANT, "--revision", "1"], s.deps), /revision-conflict/);
assert.equal(s.calls().at(-1).verb, "project.list");
assert.equal(s.calls().filter((c) => c.verb === "access.reload").length, 2);
});
// sage-3139 Q2: the write stands, so a refused reload is exit 0. stdout reports the write as before,
// and stderr names the refusal and the timer.
test("R4 a refused access.reload after a good write exits 0, stdout reports the write, stderr names the code and the 30 s timer", async (t) => {
const s = reloading(t, "refuse-reload");
const i = io(humanEnv());
assert.equal(await accessCommand([...GRANT, "--revision", "1"], i, s.deps), undefined, "no error: exit 0");
assert.match(i.out.text, /^access grant: g-[0-9a-f]{8} at acme, revision 2\n$/);
assert.equal(i.err.text, "access.reload refused: human-required; the bus's timer applies the change within 30 s\n");
assert.equal(s.read().revision, 2, "the write stands");
assert.equal(s.receipts().trim().split("\n").length, 1);
});
test("R4 the help keeps the provider-removal order and says what a refused reload means", async (t) => {
const s = setup(t);
const err = await refusal(run(["help"], s.deps), /revoke them here first/);
assert.equal(err.exitCode, 4);
const text = err.message.replace(/\n/g, " ");
assert.match(text, /To cut a user off, revoke them here first\. It acts at once\. Removing a provider from console\.sso takes effect only after a Console restart, so do that second\./);
assert.match(text, /grant and revoke ask the bus to re-read the file after the write\. If the bus refuses, the write still stands \(exit 0\), stderr names the refusal, and the bus's timer applies the change within 30 s\./);
assert.ok(text.indexOf("so do that second") < text.indexOf("re-read the file"), "next to the provider-removal order");
});
+235
View File
@@ -0,0 +1,235 @@
// FED R4 (#1801, federation design 3.7): the launcher half of revocation. A thread whose user owner
// lost the grant never starts, even with a message waiting across a restart; a running one gets
// SIGTERM at once; and recover() still rebinds and ends an earlier host's run on such a thread
// (sage-3139, D1). Real runners in the bwrap sandbox, with thread-adapter.mjs as the turn.
// The bus half is packages/bus/tests/access-revoke.test.mjs; BUILD.md
// (agents/morgana/work/queue-295) names the mutant each test kills.
//
// The host is thread-launch.test.mjs's, with an access file passed to startBroker, which the live
// boot config doesn't pass yet.
import { test } from "node:test";
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { existsSync, readFileSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { fileURLToPath } from "node:url";
import { launchDoc } from "../../business/tests/helpers.mjs";
import { BusError } from "../../bus/src/broker.mjs";
import { startBroker } from "../../bus/src/runtime.mjs";
import { Store } from "../../bus/src/store.mjs";
import { gitEnv } from "../../seat/src/checkout.mjs";
import { sandboxProblem } from "../../seat/src/sandbox.mjs";
import { launchLogFile, readSessions, Registry, SESSION_CREDENTIALS } from "../../seat/src/session.mjs";
import { bootConfig, loadSystem } from "../src/config.mjs";
import { createLauncher } from "../src/launcher.mjs";
import { fakeHarnesses, fixture, tmp } from "./helpers.mjs";
const ADAPTER = fileURLToPath(new URL("./fixtures/thread-adapter.mjs", import.meta.url));
const skip = sandboxProblem() ?? false;
// A mutant's kill is an assertion failure, not a timeout (sage-2707): the wait returns what it last
// saw and the assertions after it decide.
async function settle(fn, ms = 20000) {
const end = Date.now() + ms;
let v;
while (Date.now() < end) {
v = await fn();
if (v) return v;
await new Promise((r) => setTimeout(r, 50));
}
return v;
}
function gitIn(cwd, ...args) {
const r = spawnSync("git", ["-c", "user.name=test", "-c", "[email protected]", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" });
assert.equal(r.status, 0, r.stderr);
}
// The assistant is the thread instance. Alice holds a collaborator grant on it in the access file;
// `grant(false)` is the hand edit that removes it.
function prepare(t) {
const root = tmp(t, "mosaic-revoke-");
const f = fixture(root, "acme", (doc) => {
doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5", thinking: "high" };
doc.roles.assistant = { definition: "assistant", vars: { harness: "pi", model: "claude-sonnet-5-5", thinking: "high" } };
doc.launch = launchDoc("chief", ["coder", "assistant"], { opus: 1, sonnet: 2 });
doc.launch.threadIdleSeconds = 3600;
return doc;
});
for (const k of SESSION_CREDENTIALS.flat()) delete f.env[k];
const project = f.doc.projects.stack.root;
gitIn(project, "init", "-q", "-b", "main");
writeFileSync(join(project, "file.txt"), "one\n");
gitIn(project, "add", "file.txt");
gitIn(project, "commit", "-q", "-m", "one");
const adapter = join(root, "adapter.sh");
writeFileSync(adapter, `exec '${process.execPath}' '${ADAPTER}'\n`, { mode: 0o644 });
const file = join(root, "access.json");
let revision = 0;
const grant = (on) =>
writeFileSync(file, JSON.stringify({
accessVersion: 1,
revision: ++revision,
users: [{ id: "alice", home: "local", subjects: [{ provider: "authentik", sub: "alice-sub" }] }],
grants: on ? [{ id: "g-alice", user: "alice", business: "acme", projects: ["*"], instances: ["assistant"], level: "collaborator" }] : [],
}), { mode: 0o600 });
grant(true);
const context = { providers: ["authentik"], peers: [], businesses: { acme: { human: f.doc.human, instances: Object.keys(f.doc.roles) } }, ssoSubjects: [] };
return { root, f, adapter, access: { file, context }, grant };
}
// A host and its launcher, ticking only when the test calls tick(). pollInterval is long, so a run
// with nothing to do sleeps rather than ask the bus, and only a signal ends it.
async function start(t, { f, adapter, access }) {
const system = loadSystem({ env: f.env });
const { trackers, ...boot } = bootConfig({ system, businessId: "acme", env: f.env });
assert.equal(trackers, undefined);
const runtime = await startBroker({ ...boot, access });
const ops = {
identity: (m) => runtime.identity(m.cap),
authorizeLaunch: (m) => runtime.authorizeLaunch(m.cap, m.instance),
refuse: (m) => runtime.refuse(m.cap, m.code, m.subject),
endLaunch: (m) => runtime.endLaunch(m.record),
credentialStatus: (m) => runtime.credentialStatus(m.business, m.role),
threadsState: (m) => runtime.threadsState(m.business),
threadIdle: (m) => runtime.threadIdle(m.record),
mayLaunch: (m) => runtime.mayLaunch(m),
landHeld: (m) => runtime.landHeld(m.record),
landHolders: (m) => runtime.landHolders(m.business),
landRelease: (m) => runtime.landRelease(m.record),
sessionClears: (m) => runtime.sessionClears(m.business),
};
const refusing = (what, fn) => async (m) => {
try {
return fn(m);
} catch (e) {
const code = e instanceof BusError ? e.code : `${what}-refused`;
throw Object.assign(new Error(`${what} refused: ${code}`), { code });
}
};
const hooks = [];
let closing = null;
const host = {
business: "acme",
path: runtime.path,
dataRoot: f.dataRoot,
op: refusing("launch-op", (m) => {
if (!Object.hasOwn(ops, m.op)) throw new BusError("invalid-host-request");
return ops[m.op](m);
}),
bindLaunch: refusing("launch bind", (r) => runtime.bindLaunch(r)),
beforeClose: (fn) => void hooks.push(fn),
};
const logs = [];
const launcher = createLauncher({
host,
system,
env: f.env,
makeLoader: await fakeHarnesses(t, { pi: adapter, "claude-code": adapter }),
sessionDefaults: { pollInterval: 600000 },
harnessVersion: "0.85.1",
readOnly: [ADAPTER],
threadTickMs: 3600000,
log: (l) => logs.push(l),
});
const close = () =>
(closing ??= (async () => {
for (const fn of hooks) await fn(0);
await runtime.close();
})());
t.after(close);
await launcher.listen();
const human = runtime.broker.bindHuman({ business: "acme", human: f.doc.human, via: "cli", outsideAgent: true });
// Alice binds at her first call: after the hand edit her bind refuses access-revoked.
let alice = null;
const call = (verb, args = {}) => runtime.broker.request(human, { verb, args });
const as = (verb, args = {}) => runtime.broker.request((alice ??= runtime.broker.bindUser({ business: "acme", user: "alice" })), { verb, args });
return { runtime, launcher, call, as, logs, close };
}
const launchLog = (f) =>
existsSync(launchLogFile(f.dataRoot, "acme"))
? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l))
: [];
const events = (f, event) => launchLog(f).filter((e) => e.event === event);
// Alice's thread A in the stack project.
function thread(host) {
const project = host.call("project.create", { name: "stack work", repo: "stack" }).id;
return host.as("thread.create", { project, name: "alice work" }).id;
}
test("REQ-LAUNCH-1 R4: a pending message on a revoked thread starts no run after a launcher restart", { skip, timeout: 60000 }, async (t) => {
const prepared = prepare(t);
const { f } = prepared;
const one = await start(t, prepared);
const A = thread(one);
one.as("thread.send", { thread: A, body: "for A" });
await one.close();
// The hand edit, while no host runs.
prepared.grant(false);
const two = await start(t, prepared);
for (let i = 0; i < 3; i++) await two.launcher.tick();
const row = two.runtime.threadsState("acme").find((x) => x.thread === A);
assert.deepEqual([row.revoked, row.pending, row.holder], [true, 1, null]);
assert.deepEqual(launchLog(f), [], "no launch and no refused start");
assert.deepEqual(readSessions(f.dataRoot), []);
});
test("REQ-LAUNCH-1 R4: a running thread's run gets SIGTERM at the first tick after the hand edit, with no idle check", { skip, timeout: 60000 }, async (t) => {
const prepared = prepare(t);
const { f } = prepared;
const host = await start(t, prepared);
const A = thread(host);
const sent = host.as("thread.send", { thread: A, body: "hello" });
await host.launcher.tick();
const reply = await settle(() => host.as("thread.messages", { thread: A }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"));
assert.ok(reply, JSON.stringify(launchLog(f)));
const [launch] = events(f, "launch");
assert.equal(launch.thread, A);
prepared.grant(false);
await host.launcher.tick();
assert.deepEqual(events(f, "thread-revoked").map((e) => [e.thread, e.run]), [[A, launch.run]]);
// The runner sleeps for its poll interval, so only the signal ends it, and it exits 0, stopped.
const end = await settle(() => events(f, "end").find((e) => e.run === launch.run));
assert.deepEqual([end?.reason, end?.exitCode, end?.signal], ["stopped", 0, null], JSON.stringify(launchLog(f)));
assert.deepEqual(events(f, "thread-idle"), []);
assert.deepEqual(readSessions(f.dataRoot), []);
const row = host.runtime.threadsState("acme").find((x) => x.thread === A);
assert.deepEqual([row.revoked, row.holder], [true, null]);
// More ticks start nothing.
for (let i = 0; i < 3; i++) await host.launcher.tick();
assert.equal(events(f, "launch").length, 1);
// Dewey's B1: the broker ended the claim before the launcher's kill, so the run's end didn't release it.
await host.close();
const store = new Store(f.dataRoot);
t.after(() => store.close());
assert.deepEqual(
{ ...store.get("SELECT op, reason, by FROM thread_claims WHERE holder_run=? ORDER BY seq DESC LIMIT 1", launch.run) },
{ op: "revoke", reason: "access-revoked", by: "access-change" },
);
});
// sage-3139, D1: the bind refusal skips a rebind, so recover() ends the earlier host's run.
test("REQ-LAUNCH-1 R4 D1: recover() rebinds an earlier host's run on a revoked thread and ends it, which ends its claim", { skip, timeout: 60000 }, async (t) => {
const prepared = prepare(t);
const { f } = prepared;
const one = await start(t, prepared);
const A = thread(one);
const D = one.runtime.threadsState("acme").find((x) => x.thread === A).session;
// A stand-in for the earlier host's run, with pids that aren't running.
const record = { business: "acme", role: "assistant", run: "run-left", harness: "pi", pid: 2 ** 22 + 7, startTime: "1", thread: A, session: D };
const left = one.runtime.bindLaunch(record);
one.runtime.broker.request(left.cap, { verb: "role.claim" });
await one.close();
prepared.grant(false);
new Registry(f.dataRoot).add({ business: "acme", instance: "assistant", thread: A, session: D, run: "run-left", harness: "pi", pid: record.pid, startTime: "1", runnerPid: 2 ** 22 + 9, runnerStartTime: "1" });
const two = await start(t, prepared);
assert.ok(two.logs.includes("run run-left (assistant) from an earlier host ended: host-lost"), two.logs.join("\n"));
assert.deepEqual(events(f, "end").map((e) => [e.run, e.reason]), [["run-left", "host-lost"]]);
const row = two.runtime.threadsState("acme").find((x) => x.thread === A);
assert.deepEqual([row.revoked, row.holder], [true, null]);
for (let i = 0; i < 3; i++) await two.launcher.tick();
assert.deepEqual(events(f, "launch"), []);
});