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)
This commit is contained in:
jason.woltje committed
1 parent
249d42dfc5
commit
ff54a15563
8 files changed
+752
-40
No files matched your search
@@ -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:<id>')` 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: <code>; 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: <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: `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:<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.
|
||||
+97
-13
@@ -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');
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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');
|
||||
});
|
||||
@@ -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,
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
@@ -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", "[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 });
|
||||
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"), []);
|
||||
});
|
||||
Reference in new issue
Block a user