diff --git a/agents/morgana/work/queue-295/BUILD.md b/agents/morgana/work/queue-295/BUILD.md index 6e1ee9bfb..edeb8d626 100644 --- a/agents/morgana/work/queue-295/BUILD.md +++ b/agents/morgana/work/queue-295/BUILD.md @@ -27,8 +27,8 @@ its mutant build here, not on a 193 candidate (Q177 not needed). 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`): `#revokeLost(s.business, 'user:')` before the verb, so a - user's own request after a hand edit also ends the lost 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 @@ -53,35 +53,66 @@ its mutant build here, not on a 193 candidate (Q177 not needed). ### CLI (packages/cli/src/access.mjs) -9. grant, revoke and `user remove` call `access.reload` after the write (project.list stays the human - proof, since grant needs its rows for project names). A refused reload after a good write: stdout - still reports the write, stderr says `reload refused: ; the broker's 30 s timer applies it`, - exit 0, because the file is written and a retry would hit revision-conflict. +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: ; 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: `mock.timers.enable({ apis: ['setInterval', 'Date'] })` and `tick(30_000)`; no test sleeps. +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 | -|---|---|---| -| hand edit removing the grant ends the run within one timer period; its next call refuses not-holder | bus access-revoke.test.mjs (runtime) | timer never calls accessSweep | -| expiry with no request ends the run (Date ticked past `expires`) | bus | `#revoked` ignores `now` (covers called with a fixed early time) | -| pending message + launcher restart starts no run | cli launcher thread test | threadPass ignores `revoked` | -| running revoked thread gets SIGTERM with no threadIdle | cli launcher | revoked branch drops the kill | -| downgrade to viewer counts as a loss | bus | level `'viewer'` in `#revoked` | -| another user's runs stay, and the human's threads stay | bus | `#revokeLost` ends every user thread in the business | -| access.reload ends the run at once; tester, reader and agent refused | bus | drop `#human` in access.reload | -| bindLaunch and role.claim refuse thread-revoked | bus | drop the bindLaunch check | -| cli grant/revoke call access.reload after the write | cli access.test | drop the call | +| # | 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 | 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 | -## Open points for Sage +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. -- Q1: the per-user-request sweep (item 3, third bullet). Design 3.7 lists the next user request as a - trigger; R3 already refuses the user. This adds the claim end on that path. Keep or drop? -- Q2: the CLI's exit 0 on a refused reload after a good write (item 9). +## 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:`). `#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. diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index b05922479..018e37fc2 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -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 = []; @@ -426,13 +432,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; @@ -453,6 +460,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; @@ -907,13 +915,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' })); @@ -969,15 +971,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) { @@ -1038,9 +1097,15 @@ 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. 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(); + return this.#threadRows('t.business=?', business).map(({ owner, project, ...r }) => + (this.#revoked({ business, owner, project, instance: r.instance }, doc, now) ? { ...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 @@ -1209,7 +1274,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, @@ -1227,6 +1292,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, @@ -1291,6 +1358,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. @@ -4458,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 ?? {}; @@ -4471,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. @@ -5165,6 +5247,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'); diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index 7004c6acd..c0309e86c 100644 --- a/packages/bus/src/runtime.mjs +++ b/packages/bus/src/runtime.mjs @@ -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(); diff --git a/packages/bus/tests/access-revoke.test.mjs b/packages/bus/tests/access-revoke.test.mjs new file mode 100644 index 000000000..4bddf4198 --- /dev/null +++ b/packages/bus/tests/access-revoke.test.mjs @@ -0,0 +1,254 @@ +// 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 }, + 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. + 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'); +}); + +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'); +}); diff --git a/packages/cli/src/access.mjs b/packages/cli/src/access.mjs index 9f1134347..0e21fa620 100644 --- a/packages/cli/src/access.mjs +++ b/packages/cli/src/access.mjs @@ -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, diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index bfe55b61e..8ec164267 100644 --- a/packages/cli/src/launcher.mjs +++ b/packages/cli/src/launcher.mjs @@ -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,19 @@ 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. 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) { diff --git a/packages/cli/tests/access.test.mjs b/packages/cli/tests/access.test.mjs index d801f44bc..18b09ac15 100644 --- a/packages/cli/tests/access.test.mjs +++ b/packages/cli/tests/access.test.mjs @@ -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"); +}); diff --git a/packages/cli/tests/thread-revoke.test.mjs b/packages/cli/tests/thread-revoke.test.mjs new file mode 100644 index 000000000..c5f306d55 --- /dev/null +++ b/packages/cli/tests/thread-revoke.test.mjs @@ -0,0 +1,225 @@ +// 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 { 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", "user.email=test@example.invalid", "-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 }); + const alice = runtime.broker.bindUser({ business: "acme", user: "alice" }); + const call = (verb, args = {}) => runtime.broker.request(human, { verb, args }); + const as = (verb, args = {}) => runtime.broker.request(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); +}); + +// 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"), []); +});