From aa8fc490f5bf0c70e3eb384132e28bf801e030ac Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Thu, 22 Oct 2026 14:55:29 -0500 Subject: [PATCH] row 271 (264d, #1776): clear context build round 1, not run (slot held). Broker: session.clear, session.cancel and session.operation, the fence on send/open/claim, cancel at arrival for routed messages, the drained receive and the end drain, trusted ops sessionClears and sessionClearExited/Uncertain/Retire/Start. Launcher: clearPass before the threads loop (stop, proof, retire, successor start through check()/start()), waiting on a live row 269 survivor. Old tests per mos-1315, rebased on row 206 (sage-2968): label 3m to 3n, future step 3n to 3o, 3m-3n.sql in the engine copies and file lists, ready lines, later3l and the rows() exclusions gain both tables as 206 did for thread_shares (mos-1292, approved in mos-1336), schema.test counts 25 tables and 96 triggers, COUNTED and T45 LISTED gain both tables after thread_shares (mos-1318, mos-1319), TESTER_VERBS pin (mos-1325), fixtures/schema-3m.sql --- agents/gizmoduck/work/queue-271/BUILD.md | 93 ++++ packages/bus/src/broker.mjs | 373 ++++++++++++- packages/bus/src/process.mjs | 11 + packages/bus/src/runtime.mjs | 6 + packages/bus/tests/account-migration.test.mjs | 12 +- packages/bus/tests/chief-of-staff.test.mjs | 4 +- packages/bus/tests/comment-migration.test.mjs | 12 +- packages/bus/tests/fixtures/schema-3m.sql | 495 ++++++++++++++++++ packages/bus/tests/gitea-migration.test.mjs | 14 +- packages/bus/tests/grant-migration.test.mjs | 14 +- packages/bus/tests/name-migration.test.mjs | 2 +- packages/bus/tests/schema.test.mjs | 5 +- packages/bus/tests/sessions.test.mjs | 6 +- packages/bus/tests/spec-migration.test.mjs | 2 +- packages/bus/tests/tester-writes.test.mjs | 2 + packages/bus/tests/thread-migration.test.mjs | 6 +- packages/bus/tests/thread-shares.test.mjs | 11 +- packages/bus/tests/usage-migration.test.mjs | 2 +- packages/cli/src/launcher.mjs | 112 ++++ packages/cli/tests/stack-release.test.mjs | 2 +- scripts/stack-release.sh | 2 +- 21 files changed, 1136 insertions(+), 50 deletions(-) create mode 100644 packages/bus/tests/fixtures/schema-3m.sql diff --git a/agents/gizmoduck/work/queue-271/BUILD.md b/agents/gizmoduck/work/queue-271/BUILD.md index e85ef1b2a..ffe28e93e 100644 --- a/agents/gizmoduck/work/queue-271/BUILD.md +++ b/agents/gizmoduck/work/queue-271/BUILD.md @@ -393,6 +393,16 @@ on 2d0d407ba with `git grep` for `'3l'`, `3k-3l` and `'3m'` in `packages/*/tests - Any other count pin found at build goes here before it's edited. - After the move, grep `packages/*/tests` for the old label. Only history fixtures and migration chains may still name it. The result is recorded here (mos-1315 item 3). +- **TESTER_VERBS pin (mos-1325 item 1).** tester-writes.test.mjs:28-29 gains one comment line and + the three names after 264a's `session.list`, `session.create`. The deepEqual stays exact. +- **Engines that load the whole current store** need the new step's file. That's the + migration-file-list allowance. account-, gitea- and grant-migration.test.mjs each gain a + `copyFileSync` of `3l-3m.sql` after their `3k-3l.sql` copy, for their `engine-failing`, which + keeps every step. The engines that truncate the step list (engine-3h, -3j, -3i, -3c, -3d, -3k) + need nothing. +- Applied on gizmoduck/row271 (edit script gizmo-271-edits/mos1315.py). The grep for `'3l'` in + `packages/*/tests/*` afterwards has no hit. `3k-3l` stays in the migration copy lists and in + sessions.test.mjs:435's "no 3k→3l step" assertion, both history. - Comment-only mentions (`3k-3l migration` in end-launch, message-filter, thread-claims, threads, session-launch, thread-cli, thread-launch, chat, session-compactions) are history and stay. @@ -415,3 +425,86 @@ on 2d0d407ba with `git grep` for `'3l'`, `3k-3l` and `'3m'` in `packages/*/tests - `uncertain` after 30 s with no proof, and no second signal - the successor start is held, started or refused, through check()/start() - the pid-namespace build check + +## Build round 1 (code on gizmoduck/row271, not run: the slot was held) + +Where the code departs from the design above. Each one is for Duckworth at the code round. + +- **No `state` key in `threadsState`.** end-launch.test.mjs:193 pins a whole threadsState row, and + a new key would be an old-test change. Instead: + - `threadRows` drops retired sessions (broker.mjs `#threadRows`), and a cancelled message isn't + `pending`. + - A new trusted op, `sessionClears {business}` (broker.mjs:988), lists each open or starting + clear: thread, instance, session, op, stage, run, generation, successor, refusal. The + launcher reads it once per tick in `clearPass` (launcher.mjs:742), before its threads loop. + Any session a listed clear owns (the old session while stopping, the successor while + starting) is skipped by the normal start and idle code, so threadPass never starts or idles + it. +- **No `thread_opens` row at retire.** The successor's start comes from `sessionClears` (stages + `retired`, `start-held`, `start-refused`), not from an open. A cancelled op isn't listed, so + nothing starts its successor. A later new send gives the successor a `pending` message, and + threadPass starts it like any idle session (sage-2830's reading). +- **The run's end is a drain too.** `endLaunch` appends `drained` (run, generation) when the + ending run's session has a `requested` op. A run that exits before its next receive needs no + signal. +- **Exit proof.** `sessionClearExited {business, thread, op, run}` refuses `run-not-ended` unless + the run has ended. With run null, the launcher first takes and releases the session's workspace + lock. A busy lock (a spawning run) throws and the clear waits a tick. + - Side effect: `sessionWorkspace` creates the session's workspace directory, so a session that + never ran gets an empty one at its clear. +- **Restart.** `recover()` SIGKILLs an earlier host's runs and ends them `host-lost`. A run that + outlived the kill is a row 269 survivor and is ended too. `clearStop` waits while a survivor + with the clear's run is live (`liveRun`), and sends it nothing. When the old run is gone, the + ended run is the proof. + - `clearStops` is in memory, so a restarted launcher has no stop times. That's safe because + recover() has already killed or listed every run from the earlier host, so none is in the new + registry to signal. An `uncertain` op it finds is never signalled. +- **Halt at retire waits.** `sessionClearRetire` throws `processing-halted` while processing is + halted, and the op stays `exited` until it lifts. Only an owner or grant refusal appends + `failed`. +- **Held codes.** A successor start refused `capacity-full` or `limit-reached` reports `held`. + Every other refusal reports `refused`, and any non-Refusal throw reports `launch-failed`. Each + code is logged and reported once per op, not once per tick. +- **Names stay bound.** `#nameTaken` counts any op's `successor_name`, a cancelled or failed one + included. So a session.create of `-1` after a cancelled clear refuses `session-exists`, and + the next clear takes `-2`. That's stricter than needed. Freeing a name would let a repeat + of the cancelled op's envelope name a session that isn't its successor. +- **Cancel after `failed`** refuses `session-clear-complete`, like cancel after `start-started`. +- **message.send** answers `cancelled: "session-clearing"` or `"session-retired"` only when its + route lands on such a session (B1). N1 grep of whole-answer pins is still to do. + +### Tests added to T4 (sage-2830) + +- After a post-retire cancel, a thread-only send to the thread (the successor is the default) + launches the successor through D8 as new input. The clear's latest row stays `cancelled`, and + no `start-*` row is appended. +- After a restart, a live survivor of the cleared run holds the clear at `drained` with no + signal. Its exit lets the clear retire. + +### Tester grant (mos-1325) + +The brief's section 3 says tester writes aren't granted by default and use fixtures only. 271 +grants the tester `session.clear`, `session.cancel` and `session.operation` because the Q173 live +check has to clear a tester thread through the tester socket. Without the grant, D7 can't be +proved live by an agent tester. +- The three verbs act only on the tester's own threads. Each resolves its thread with `#thread` + (broker.mjs:871-877), which refuses `thread-not-owned` unless the thread's owner is the caller's + `#owner`, and a tester's owner is `tester` (:879-880). So a tester can't reach a human's or a + user's thread. +- Test: a tester's `session.clear`, `session.cancel` and `session.operation` on a thread a human + owns each refuse `thread-not-owned` and write nothing (no `session_ops` or `message_cancels` + row). +- The R3 user gate refuses all three for a user (user-principal.test.mjs:133 spreads + TESTER_VERBS). Those refusals are this row's own checks (mos-1325 item 3). + +### N1 and N6 at build + +- **N1.** message.send adds `cancelled` only when `#route` returns a code (broker.mjs:5505, + `if (cancelled) return { id: mid, request, cancelled }`). Every other send answers exactly what + it did before. `#route` returns a code only for a session with an open or retired clear, and + before 271 no test can make one. So no whole-answer pin on message.send can change, and nothing + joins the mos-1315 list. +- **N6.** The launcher's only spawn is `spawnSession` at launcher.mjs:574, inside `start()` + (:422), which runs every session, thread runs included. spawnSession (seat/src/session.mjs:300-302) + builds the runner argv through `sandboxArgv`, whose argv starts with `NAMESPACES` (sandbox.mjs:168), + and `NAMESPACES` holds `--unshare-pid` and `--as-pid-1` (:68). The bwrap test stays in T4. diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index c33425486..047480661 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -127,7 +127,9 @@ export const TESTER_VERBS = Object.freeze(['message.send', 'message.receive', 'm // Row 194 (#1692, D4.1): launch and terminate, and the read of the requests they make. 'instance.launch', 'instance.end', 'instance.request', // Row 264a (#1769): the sessions of the tester's own threads. - 'session.list', 'session.create']); + 'session.list', 'session.create', + // Row 271 (264d, #1776): clear context in the tester's own threads. + 'session.clear', 'session.cancel', 'session.operation']); // FED R3 (#414, federation design 3.5): the collaborator level is this exact list. Every other verb // refuses a user access-refused before its own checks: owner verbs, board reads, message.*, // decision.*, project files, spec.read, assistant.stop, launch.*, name.set and usage.*. @@ -613,6 +615,108 @@ export class Broker { if (!this.#store.get('SELECT 1 FROM sessions WHERE id=? AND thread=?', identifier(session), t.id)) fail('session-not-found'); return session; } + // Row 271 (264d, #1776, D7): a clear of a session is one op, its rows in session_ops. The op is + // open while its latest row is requested, drained, uncertain or exited (session_ops_one_open). + #openOp(session) { + return this.#store.get( + `SELECT o.* FROM session_ops o WHERE o.session=? AND o.seq=(SELECT max(seq) FROM session_ops WHERE thread=o.thread AND op=o.op) + AND o.stage IN ('requested','drained','uncertain','exited')`, + session, + ); + } + // active, clearing, uncertain or retired, with the op that made it so. + #sessionState(session) { + const retired = this.#store.get("SELECT op, successor FROM session_ops WHERE session=? AND stage='retired'", session); + if (retired) return { state: 'retired', op: retired.op, successor: retired.successor }; + const open = this.#openOp(session); + if (!open) return { state: 'active', op: null, successor: null }; + return { state: open.stage === 'uncertain' ? 'uncertain' : 'clearing', op: open.op, successor: null, open }; + } + // The fence: no send, open or claim reaches a session being cleared, or a retired one. + #sessionFence(session) { + const { state } = this.#sessionState(session); + if (state === 'retired') fail('session-retired'); + if (state !== 'active') fail('session-clearing'); + } + // A session's revision: 0 at creation, plus one for each row of its clears. + #revision(session) { + return this.#store.get('SELECT count(*) AS n FROM session_ops WHERE session=?', session).n; + } + // A name a session has, or one a clear has bound for its successor. + #nameTaken(thread, sessionName) { + return Boolean( + this.#store.get('SELECT 1 FROM sessions WHERE thread=? AND name=?', thread, sessionName) ?? + this.#store.get('SELECT 1 FROM session_ops WHERE thread=? AND successor_name=?', thread, sessionName), + ); + } + // The successor's id, bound at accept so a repeat finds it: a lowercase RFC 9562 version 8 UUID + // from a sha256 of the op, which every session selector takes (cli/src/thread.mjs:41, gizmoduck-0274). + #successorId(business, thread, session, op) { + const h = createHash('sha256').update(JSON.stringify([business, thread, session, op])).digest('hex').split(''); + h[12] = '8'; + h[16] = '89ab'[parseInt(h[16], 16) & 3]; + const x = h.join(''); + return `${x.slice(0, 8)}-${x.slice(8, 12)}-${x.slice(12, 16)}-${x.slice(16, 20)}-${x.slice(20, 32)}`; + } + // A clear verb's payload with its caller, for the repeat and conflict checks. + #opDigest(s, ...payload) { + return createHash('sha256').update(JSON.stringify([this.#owner(s), ...payload])).digest('hex'); + } + // One session_ops row. `op` is the op's requested row, or the fields it is made from. + #appendOp(op, stage, { attempt = 0, workAttempt = null, run = null, generation = null, digest = null, refusal = null } = {}) { + this.#store.run( + 'INSERT INTO session_ops(at,business,thread,session,op,stage,attempt,work_attempt,successor,successor_name,run,generation,digest,refusal) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)', + this.#now(), + op.business, + op.thread, + op.session, + op.op, + stage, + attempt, + workAttempt, + op.successor, + op.successor_name, + run, + generation, + digest, + refusal, + ); + } + #opRow(thread, op, stage) { + return this.#store.get('SELECT * FROM session_ops WHERE thread=? AND op=? AND stage=? ORDER BY seq LIMIT 1', thread, op, stage); + } + // The answer to every clear verb, from the op's rows (BUILD.md's contract for 264e). + #clearEnvelope(thread, op) { + const rows = this.#store.all('SELECT * FROM session_ops WHERE thread=? AND op=? ORDER BY seq', thread, op); + if (!rows.length) fail('operation-not-found'); + const has = (stage) => rows.some((r) => r.stage === stage); + const last = rows.at(-1); + const start = rows.filter((r) => r.stage.startsWith('start-')).at(-1) ?? null; + const state = has('cancelled') ? 'cancelled' : has('failed') ? 'failed' : has('retired') ? 'retired' : last.stage === 'uncertain' ? 'uncertain' : 'clearing'; + const refusal = state === 'failed' ? rows.find((r) => r.stage === 'failed').refusal : start?.stage === 'start-refused' ? start.refusal : null; + return { + op, + thread, + session: rows[0].session, + successor: rows[0].successor, + state, + start: start ? start.stage.slice('start-'.length) : null, + refusal, + ...(state === 'cancelled' ? { retired: has('retired') ? true : null } : {}), + at: last.at, + }; + } + // D7 step 1: the session's undelivered inputs are cancelled, each with a receipt, never replayed. + #cancelInputs(op, instance) { + const rows = this.#store.all( + `SELECT m.id FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=? AND m.to_role=? + 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 c WHERE c.message=m.id) ORDER BY m.seq`, + op.session, + instance, + ); + for (const m of rows) this.#store.run('INSERT INTO message_cancels(at,message,session,op) VALUES(?,?,?,?)', this.#now(), m.id, op.session, op.op); + } // Row 155 (#1649): the held claims of every session of a thread instance, oldest thread first, or // only the threads `owner` made. An end of the instance ends each of them. #threadHolders(business, instance, owner = null) { @@ -951,15 +1055,157 @@ export class Broker { this.#store.transaction(() => this.#claimEnd(session, this.#own(session), 'release', null)); return { released: true }; } + // Row 271 (264d, #1776, D7 steps 3-5): the business's clears still in the launcher's hands, as of + // each op's latest row. An open op waits for its drained run's stop and exit proof, then its + // retirement. A retired op whose successor hasn't started, and wasn't cancelled, waits for its + // start through D8. `run` and `generation` are the drained row's. Trusted launcher API, not a + // socket verb. + sessionClears(business) { + this.#business(identifier(business)); + return this.#store + .all( + `SELECT o.*, t.instance FROM session_ops o JOIN threads t ON t.id=o.thread + WHERE o.business=? AND o.seq=(SELECT max(seq) FROM session_ops WHERE thread=o.thread AND op=o.op) + AND o.stage IN ('requested','drained','uncertain','exited','retired','start-held','start-refused') + ORDER BY o.seq`, + business, + ) + .map((o) => { + const drained = this.#opRow(o.thread, o.op, 'drained'); + return { + thread: o.thread, + instance: o.instance, + session: o.session, + op: o.op, + stage: o.stage, + run: drained?.run ?? null, + generation: drained?.generation ?? null, + successor: o.successor, + refusal: o.refusal, + }; + }); + } + // The requested row of a clear the launcher names, in the business it names. + #launcherOp(record, extra = []) { + keys(record, ['business', 'thread', 'op', ...extra], ['business', 'thread', 'op', ...extra]); + this.#business(identifier(record.business)); + const first = this.#opRow(identifier(record.thread), identifier(record.op), 'requested'); + if (!first || first.business !== record.business) fail('operation-not-found'); + return first; + } + // D7 step 3: the exit proof, from the launcher. The drained run's own close handler wrote its + // session.ended (endLaunch), which is checked here; a signal, a revoked claim or a dead socket + // is not proof. With no run, the proof is the launcher's: it took the session's workspace lock, + // and its registry holds no run of the session. Trusted launcher API. + sessionClearExited(record) { + const first = this.#launcherOp(record, ['run']); + if (record.run !== null) identifier(record.run); + return this.#store.transaction(() => { + const open = this.#openOp(first.session); + if (open?.op !== first.op) fail('operation-not-open'); + if (open.stage === 'exited') return this.#clearEnvelope(first.thread, first.op); + const drained = this.#opRow(first.thread, first.op, 'drained') ?? fail('not-drained'); + if (record.run !== drained.run) fail('not-drained-run'); + if (drained.run !== null && !this.#store.get("SELECT 1 FROM events WHERE business=? AND kind='session.ended' AND subject=?", first.business, drained.run)) + fail('run-not-ended'); + this.#appendOp(first, 'exited', { run: drained.run, generation: drained.generation }); + return this.#clearEnvelope(first.thread, first.op); + }); + } + // D7 step 3: 30 s after the stop with no proof, once. Nothing else is signalled or forced. + sessionClearUncertain(record) { + const first = this.#launcherOp(record); + return this.#store.transaction(() => { + const open = this.#openOp(first.session); + if (open?.op !== first.op) fail('operation-not-open'); + if (open.stage === 'drained') this.#appendOp(first, 'uncertain'); + else if (open.stage !== 'uncertain') fail('not-drained'); + return this.#clearEnvelope(first.thread, first.op); + }); + } + // D7 step 4: retirement and the successor, in one transaction, after the exit proof. Owner, grant + // and halt are rechecked here (N4). A halt refuses and writes nothing, so the launcher asks again + // once processing resumes; a lost grant ends the clear `failed`. Only the proven run may still hold + // the claim, and its claim ends here only if endRun hasn't ended it (N5). + sessionClearRetire(record) { + const first = this.#launcherOp(record); + return this.#store.transaction(() => { + const open = this.#openOp(first.session); + if (open?.op !== first.op || open.stage !== 'exited') fail('not-exited'); + this.#processingWork(); + const t = this.#store.get('SELECT * FROM threads WHERE id=?', first.thread); + const refusal = this.#ownerRefusal(t); + if (refusal) { + this.#appendOp(first, 'failed', { refusal }); + return this.#clearEnvelope(first.thread, first.op); + } + const drained = this.#opRow(first.thread, first.op, 'drained'); + const h = this.#sessionHolder(first.session); + if (h?.op === 'claim') { + if (h.holder_run !== drained.run) fail('thread-already-held'); + this.#claimEnd({ business: t.business }, h, 'release', null, null, 'session-clear'); + } + const wasDefault = this.#defaultSession(t.id) === first.session; + this.#appendOp(first, 'retired'); + const at = this.#now(); + this.#store.run( + 'INSERT INTO sessions(id,at,business,thread,instance,owner,name,op,predecessor) VALUES(?,?,?,?,?,?,?,?,?)', + first.successor, + at, + t.business, + t.id, + t.instance, + t.owner, + first.successor_name, + first.op, + first.session, + ); + // D7 step 5: the default moves only off the session that was the default. + if (wasDefault) this.#store.run('INSERT INTO session_defaults(at,thread,session,by) VALUES(?,?,?,?)', at, t.id, first.successor, t.owner); + return this.#clearEnvelope(first.thread, first.op); + }); + } + // D7 step 5, D8: the successor's start, as the launcher's check()/start() decided it. Each report + // takes the next counter (C1). A held start spends no work attempt; a refused one spends one. + sessionClearStart(record) { + keys(record, ['business', 'thread', 'op', 'start', 'refusal'], ['business', 'thread', 'op', 'start']); + const { start, refusal = null } = record; + const first = this.#launcherOp({ business: record.business, thread: record.thread, op: record.op }); + if (!['held', 'started', 'refused'].includes(start)) fail('invalid-request'); + if (start === 'started' ? refusal !== null : !id(refusal)) fail('invalid-request'); + return this.#store.transaction(() => { + if (!this.#opRow(first.thread, first.op, 'retired')) fail('not-retired'); + if (this.#opRow(first.thread, first.op, 'cancelled')) fail('session-clear-cancelled'); + if (this.#opRow(first.thread, first.op, 'start-started')) fail('session-clear-complete'); + const { reports, refused } = this.#store.get( + "SELECT count(*) AS reports, count(*) FILTER (WHERE stage='start-refused') AS refused FROM session_ops WHERE thread=? AND op=? AND stage GLOB 'start-*'", + first.thread, + first.op, + ); + this.#appendOp(first, 'start-' + start, { attempt: reports + 1, workAttempt: refused + 1, refusal }); + return this.#clearEnvelope(first.thread, first.op); + }); + } + // N4: why the thread's owner may no longer clear it, or null. A user's grants are read again; the + // business human and the tester own their threads for as long as the business names them. + #ownerRefusal(t) { + if (t.owner.startsWith('user:')) { + const s = { business: t.business, user: t.owner.slice('user:'.length) }; + return this.#granted({ ...s, grants: this.#userGrants(s) }, t.project, t.instance) ? null : 'access-refused'; + } + return t.owner === 'tester' || t.owner === this.#business(t.business).human ? null : 'thread-not-owned'; + } // Row 269 (264b, #1774): one row per session of each thread, oldest thread first, then oldest // session. `legacy` is a session the 3k-3l migration made (op legacy-3l), whose store and - // in-sandbox workspace path are the thread's old ones. + // in-sandbox workspace path are the thread's old ones. Row 271 (264d): a retired session has + // none, so nothing starts or idles it, and a cancelled message isn't pending. #threadRows(where, ...params) { return this.#store .all( `SELECT t.id, t.instance, 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'))) AS pending, + 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, (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 c.op='claim' AND EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='delivered' AND d.holder_run=c.holder_run) AND NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='read')) AS unread, @@ -968,7 +1214,7 @@ export class Broker { c.at AS claim_at, c.op AS claim_op, c.holder_run AS holder_run FROM threads t JOIN sessions se ON se.thread=t.id LEFT JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE session=se.id) - WHERE ${where} ORDER BY t.seq, se.seq`, + WHERE ${where} AND NOT EXISTS(SELECT 1 FROM session_ops o WHERE o.session=se.id AND o.stage='retired') ORDER BY t.seq, se.seq`, ...params, ) .map((r) => ({ @@ -1066,6 +1312,10 @@ export class Broker { const h = this.#own(session); const released = h?.op === 'claim' && h.holder_run === session.run; if (released) this.#claimEnd(session, h, 'release', null); + // Row 271 (264d, D7 step 2): the holder of a session being cleared that ends before its next + // receive leaves no turn running either. Its exit is the drain, recorded with its generation. + const open = released && session.thread ? this.#openOp(session.session) : null; + if (open?.stage === 'requested') this.#appendOp(open, 'drained', { run: session.run, generation: h.generation }); // Row 125 (#1619, REQ-LIMIT-4): a run a limit stopped carries the claim row's reason. const limit = this.#store.get( `SELECT reason FROM ${session.thread ? 'thread_claims' : 'role_claims'} WHERE business=? AND holder_run=? AND by='limit-check' ORDER BY seq DESC LIMIT 1`, @@ -4903,6 +5153,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'); + // 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'); this.#store.run( "INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,address,by) VALUES(?,?,?,?,1+(SELECT count(*) FROM thread_claims WHERE session=? AND op='claim'),?,?,?,?,?)", @@ -5321,10 +5573,12 @@ export class Broker { a.corrects ?? null, a.body, ); - this.#route(s.business, mid, to, thread); + const cancelled = this.#route(s.business, mid, to, thread); const request = s.human ? this.#event(s, 'human.input', { kind: 'instruction', message: mid }, mid) : null; + // Row 271 (264d, B1): kept as evidence and cancelled at arrival, so it waits for nobody. + if (cancelled) return { id: mid, request, cancelled }; const waiting = this.#waiting(s, mid, to, thread); return waiting ? { id: mid, request, waiting } : { id: mid, request }; } @@ -5335,10 +5589,21 @@ export class Broker { this.#agent(s); this.#received.add(s.business + ':' + s.run); } + // Row 271 (264d, D7 step 2): a thread run's receive comes at a completed turn with no child + // (the runner asks it only then). At a clear of its session it records the drain once and + // delivers nothing. The run keeps its claim, so no turn starts before the launcher's stop. + if (s.thread) { + const open = this.#openOp(s.session); + if (open) { + if (!this.#opRow(open.thread, open.op, 'drained')) this.#appendOp(open, 'drained', { run: s.run, generation: this.#runGeneration(s) }); + return []; + } + } const role = s.human ? 'human' : s.role; // A thread run receives its own session's messages only (REQ-THREAD-1, row 264a). const rows = this.#store.all( "SELECT * FROM messages m WHERE business=? AND to_role=? 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 c WHERE c.message=m.id)' + (s.thread ? ' AND EXISTS(SELECT 1 FROM message_threads mt WHERE mt.message=m.id AND mt.session=?)' : '') + ' ORDER BY seq LIMIT 100', s.business, @@ -5864,6 +6129,8 @@ export class Broker { keys(a, ['thread', 'session'], ['thread']); const t = this.#grantedThread(s, 'thread.open', this.#thread(s, a.thread)); const session = this.#threadSession(t, a.session); + // Row 271 (264d, D7 step 1): a session being cleared, or a retired one, isn't opened. + this.#sessionFence(session); this.#store.run('INSERT INTO thread_opens(at,thread,session,by) VALUES(?,?,?,?)', this.#now(), t.id, session, this.#owner(s)); return { opened: t.id }; } @@ -5874,6 +6141,8 @@ export class Broker { keys(a, ['thread', 'session', 'body', 'class'], ['thread', 'body']); const t = this.#grantedThread(s, 'thread.send', this.#thread(s, a.thread)); const session = this.#threadSession(t, a.session); + // Row 271 (264d, D7 step 1): nor sent to, by name or as the default. + this.#sessionFence(session); // Row 120's lists hold here as on message.send: the instance must hear from the human. this.#checkMessaging(s.business, 'human', t.instance); string(a.body, 32768); @@ -5959,9 +6228,20 @@ export class Broker { keys(a, ['thread'], ['thread']); const t = this.#thread(s, a.thread); const fallback = this.#defaultSession(t.id); + // Row 271 (264d): each row's clear state, revision and, once retired, successor. return this.#store .all('SELECT id,at,name,predecessor,op FROM sessions WHERE thread=? ORDER BY seq', t.id) - .map(({ op, ...r }) => ({ ...r, default: r.id === fallback, compactions: this.#compactionCount(t, r.id, op === 'legacy-3l') })); + .map(({ op, ...r }) => { + const { state, successor } = this.#sessionState(r.id); + return { + ...r, + default: r.id === fallback, + compactions: this.#compactionCount(t, r.id, op === 'legacy-3l'), + state, + revision: this.#revision(r.id), + successor, + }; + }); } // D8: a new session is metadata only. Its run starts when the launcher starts it (264b). // The op key makes a retry find the first row: the same op and name answer it, another name @@ -5972,14 +6252,80 @@ export class Broker { if (!name(a.name)) fail('invalid-request'); identifier(a.op); const t = this.#thread(s, a.thread); + // Row 271 (264d, B3): a clear's op is the clear's, and its successor takes it. + if (this.#store.get('SELECT 1 FROM session_ops WHERE thread=? AND op=?', t.id, a.op)) fail('session-op-conflict'); const old = this.#store.get('SELECT id,name FROM sessions WHERE thread=? AND op=?', t.id, a.op); if (old) { if (old.name !== a.name) fail('session-op-conflict'); return { id: old.id, thread: t.id, name: old.name }; } - if (this.#store.get('SELECT 1 FROM sessions WHERE thread=? AND name=?', t.id, a.name)) fail('session-exists'); + if (this.#nameTaken(t.id, a.name)) fail('session-exists'); return { id: this.#newSession(s, t, a.name, a.op, false), thread: t.id, name: a.name }; } + // Row 271 (264d, #1776, D7): clear context. This accept binds the successor, fences the session + // and cancels its undelivered inputs in one transaction. The drain (message.receive or + // endLaunch), the stop and proof (the launcher), retirement and the start follow. The answer is + // the op's envelope, and a repeat of the same payload answers it again. + case 'session.clear': { + if (!s.tester) this.#human(s); + keys(a, ['thread', 'session', 'revision', 'confirm', 'op'], ['thread', 'session', 'revision', 'confirm', 'op']); + identifier(a.op); + if (!Number.isSafeInteger(a.revision) || a.revision < 0) fail('invalid-request'); + const t = this.#thread(s, a.thread); + const session = this.#threadSession(t, a.session); + if (a.confirm !== true) fail('confirm-required'); + const digest = this.#opDigest(s, 'session.clear', t.id, session, a.revision, a.op); + const first = this.#opRow(t.id, a.op, 'requested'); + if (first) { + if (first.digest !== digest) fail('session-op-conflict'); + return this.#clearEnvelope(t.id, a.op); + } + // B3: an op names one thing in its thread, a session or a clear. + if (this.#store.get('SELECT 1 FROM sessions WHERE thread=? AND op=?', t.id, a.op)) fail('session-op-conflict'); + this.#processingWork(); + this.#sessionFence(session); + if (this.#revision(session) !== a.revision) fail('session-revision-stale'); + // Q4 (sage-2826): `-N`, the next N free in the thread. + const base = this.#store.get('SELECT name FROM sessions WHERE id=?', session).name; + let n = 1; + while (this.#nameTaken(t.id, `${base}-${n}`)) n++; + if (!name(`${base}-${n}`)) fail('session-name-too-long'); + const op = { business: s.business, thread: t.id, session, op: a.op, successor: this.#successorId(s.business, t.id, session, a.op), successor_name: `${base}-${n}` }; + this.#appendOp(op, 'requested', { digest }); + this.#cancelInputs(op, t.instance); + // With no holder there is no turn to wait for, and B2 refuses a claim from here on. + if (this.#sessionHolder(session)?.op !== 'claim') this.#appendOp(op, 'drained'); + return this.#clearEnvelope(t.id, a.op); + } + // Row 271 (sage-2828, D7:263-264): a human's cancel of a clear. Before retirement it closes the + // op, which releases the fence: the claim and the cancelled messages stay as they are, and no + // successor is made. After retirement it ends the startup only, and both sessions stay. After + // the successor started, or a failed clear, there is nothing to cancel. + case 'session.cancel': { + if (!s.tester) this.#human(s); + keys(a, ['thread', 'op', 'confirm'], ['thread', 'op', 'confirm']); + identifier(a.op); + const t = this.#thread(s, a.thread); + if (a.confirm !== true) fail('confirm-required'); + const first = this.#opRow(t.id, a.op, 'requested') ?? fail('operation-not-found'); + const digest = this.#opDigest(s, 'session.cancel', t.id, a.op); + const done = this.#opRow(t.id, a.op, 'cancelled'); + if (done) { + if (done.digest !== digest) fail('session-op-conflict'); + return this.#clearEnvelope(t.id, a.op); + } + if (this.#opRow(t.id, a.op, 'start-started') || this.#opRow(t.id, a.op, 'failed')) fail('session-clear-complete'); + this.#appendOp(first, 'cancelled', { digest }); + return this.#clearEnvelope(t.id, a.op); + } + // Row 271: a clear's envelope, for a caller that waited and reports pending. Reading writes + // nothing, and a retry asks here rather than sending the clear again. + case 'session.operation': { + if (!s.tester) this.#human(s); + keys(a, ['thread', 'op'], ['thread', 'op']); + identifier(a.op); + return this.#clearEnvelope(this.#thread(s, a.thread).id, a.op); + } case 'replies': { if (!s.human && !s.reader) fail('human-required'); keys(a, ['id'], ['id']); @@ -6108,12 +6454,21 @@ export class Broker { } // Row 121: a message names its thread (REQ-MSG-1) and goes to the holder that may have it. // Row 264a (#1769): it names one session of the thread, and the session's holder may have it. + // Row 271 (264d, B1): a message to the instance in a session being cleared, or a retired one, is + // kept and cancelled at arrival with a receipt. It is never delivered there or to the successor, + // and the answer is why; otherwise null. #route(business, mid, to, session) { if (session) { - const { thread } = this.#store.get('SELECT thread FROM sessions WHERE id=?', session) ?? fail('session-not-found'); + const { thread, instance } = this.#store.get('SELECT thread, instance FROM sessions WHERE id=?', session) ?? fail('session-not-found'); this.#store.run('INSERT INTO message_threads(message,thread,session) VALUES(?,?,?)', mid, thread, session); + const { state, op } = this.#sessionState(session); + if (to === instance && state !== 'active') { + this.#store.run('INSERT INTO message_cancels(at,message,session,op) VALUES(?,?,?,?)', this.#now(), mid, session, op); + return state === 'retired' ? 'session-retired' : 'session-clearing'; + } } this.#delivery(mid, 'routed', this.#recipient(business, to, session)); + return null; } // Row 264a (#1769): a session row and, for a new thread, its default. The 3k-3l migration made the // legacy defaults, whose id is the thread id; a new session's id is random, and a trigger refuses @@ -6266,7 +6621,7 @@ export class Broker { #routeSession(session) { const h = this.#sessionHolder(session); const rows = this.#store.all( - "SELECT m.id FROM messages m JOIN message_threads mt ON mt.message=m.id JOIN threads t ON t.id=mt.thread WHERE mt.session=? AND m.to_role=t.instance AND NOT EXISTS(SELECT 1 FROM deliveries WHERE message=m.id AND op IN ('delivered','read'))", + "SELECT m.id FROM messages m JOIN message_threads mt ON mt.message=m.id JOIN threads t ON t.id=mt.thread WHERE mt.session=? AND m.to_role=t.instance AND NOT EXISTS(SELECT 1 FROM deliveries WHERE message=m.id AND op IN ('delivered','read')) AND NOT EXISTS(SELECT 1 FROM message_cancels WHERE message=m.id)", session, ); for (const m of rows) this.#delivery(m.id, 'routed', h); diff --git a/packages/bus/src/process.mjs b/packages/bus/src/process.mjs index 321bf2658..4ab944af5 100644 --- a/packages/bus/src/process.mjs +++ b/packages/bus/src/process.mjs @@ -29,6 +29,11 @@ const LAUNCH_OPS = new Set([ 'lifecycleTake', 'lifecycleDelete', 'lifecycleDone', + 'sessionClears', + 'sessionClearExited', + 'sessionClearUncertain', + 'sessionClearRetire', + 'sessionClearStart', ]); function launchOp(m) { if (m.op === 'identity') return runtime.identity(m.cap); @@ -64,6 +69,12 @@ function launchOp(m) { if (m.op === 'lifecycleTake') return runtime.lifecycleTake(m.record); if (m.op === 'lifecycleDelete') return runtime.lifecycleDelete(m.record); if (m.op === 'lifecycleDone') return runtime.lifecycleDone(m.record); + // Row 271 (264d, #1776, D7 steps 3-5): the launcher's clear pass. + if (m.op === 'sessionClears') return runtime.sessionClears(m.business); + if (m.op === 'sessionClearExited') return runtime.sessionClearExited(m.record); + if (m.op === 'sessionClearUncertain') return runtime.sessionClearUncertain(m.record); + if (m.op === 'sessionClearRetire') return runtime.sessionClearRetire(m.record); + if (m.op === 'sessionClearStart') return runtime.sessionClearStart(m.record); if (m.op === 'credentialStatus') return runtime.credentialStatus(m.business, m.role); // Every LAUNCH_OPS name has its line above. A missing one used to answer as credentialStatus, // whose [] for an unknown role read as "no holders" for landHolders (row 156 mutant P02). diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index 6685cfe5f..7004c6acd 100644 --- a/packages/bus/src/runtime.mjs +++ b/packages/bus/src/runtime.mjs @@ -594,6 +594,12 @@ export async function startBroker({ lifecycleTake: (r) => broker.lifecycleTake(r), lifecycleDelete: (r) => broker.lifecycleDelete(r), lifecycleDone: (r) => broker.lifecycleDone(r), + // Row 271 (264d, #1776): the launcher's clear pass. + sessionClears: (business) => broker.sessionClears(business), + sessionClearExited: (r) => broker.sessionClearExited(r), + sessionClearUncertain: (r) => broker.sessionClearUncertain(r), + sessionClearRetire: (r) => broker.sessionClearRetire(r), + sessionClearStart: (r) => broker.sessionClearStart(r), refuse: (cap, code, subject) => broker.refuse(cap, code, subject), credentialStatus, launches: bound, diff --git a/packages/bus/tests/account-migration.test.mjs b/packages/bus/tests/account-migration.test.mjs index fe866b0d1..a65e02bf2 100644 --- a/packages/bus/tests/account-migration.test.mjs +++ b/packages/bus/tests/account-migration.test.mjs @@ -18,7 +18,8 @@ const here = (p) => new URL(p, import.meta.url); // Row 264a (#1769): what the 3k→3l step rebuilds or adds. sessions.test.mjs tests those objects. const SESSIONS_3L = ['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens']; // Row 206 (#1704): and the 3l→3m step's thread_shares. thread-shares.test.mjs tests it. -const later3l = (r) => [...SESSIONS_3L, 'thread_shares'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name); +// Row 271 (#1776): and the 3m→3n step's session_ops and message_cancels. session-clear.test.mjs tests them. +const later3l = (r) => [...SESSIONS_3L, 'thread_shares', 'session_ops', 'message_cancels'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name); function source3h(t) { const root = mkdtempSync(join(tmpdir(), 'account-migrate-')); t.after(() => rmSync(root, { recursive: true, force: true })); @@ -86,6 +87,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { copyFileSync(here('../src/migrations/3j-3k.sql'), join(moduleRoot, 'src/migrations/3j-3k.sql')); copyFileSync(here('../src/migrations/3k-3l.sql'), join(moduleRoot, 'src/migrations/3k-3l.sql')); copyFileSync(here('../src/migrations/3l-3m.sql'), join(moduleRoot, 'src/migrations/3l-3m.sql')); + copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } @@ -93,7 +95,7 @@ test('REQ-LIMIT-4 3h copy migrates to 3i: every row keeps its seq, the sequence const { root, path } = source3h(t), before = snapshot(path), was = master(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); // Every table's rows as they were, usage_records with its seq values and sqlite_sequence with 50. assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows); assert.deepEqual(before.rows.usage_records.map(r => r.seq), [1, 2, 3]); @@ -143,7 +145,7 @@ test('REQ-LIMIT-4 a 3h→3i step that fails partway leaves the store at 3h with assert.deepEqual(snapshot(path, before.columns).rows, before.rows); // Still a whole 3h store: the real step migrates it. new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(migrated3l(snapshot(path, before.columns).rows), before.rows); }); @@ -155,14 +157,14 @@ test('REQ-LIMIT-4 the 3h engine refuses a migrated store, migrating or not, and schema: ddl3h, store: (source) => source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3h).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3h';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3h';") .replace(/\}, \{\n {2}\/\/ Row 126 \(#1620\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'), }); const pinned = readFileSync(join(root, 'engine-3h/src/store.mjs'), 'utf8'); assert.ok(pinned.includes("export const SCHEMA_VERSION = '3h';") && !pinned.includes('3h-3i'), 'the 3h engine has no 3h→3i step'); assert.throws(() => new Store3h(root), /schema-mismatch/); assert.throws(() => new Store3h(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(master(path), was); assert.deepEqual(snapshot(path, after.columns).rows, after.rows); }); diff --git a/packages/bus/tests/chief-of-staff.test.mjs b/packages/bus/tests/chief-of-staff.test.mjs index c02487c67..8186994f8 100644 --- a/packages/bus/tests/chief-of-staff.test.mjs +++ b/packages/bus/tests/chief-of-staff.test.mjs @@ -614,7 +614,7 @@ async function database3c(t) { let source = readFileSync(here('../src/store.mjs'), 'utf8'); const pinned = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3c).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3c';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3c';") .replace(/\}, \{\n {2}from: '3c',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.notEqual(pinned, source); assert.ok(!pinned.includes('3c-3d'), 'the 3c engine has no 3c→3d step'); @@ -638,7 +638,7 @@ test('REQ-ROLE-2, REQ-COS-1: the 3c→3d migration lets the chief of staff end a assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3m'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}'); assert.equal(store.get("SELECT holder_run FROM role_claims WHERE role='coder'").holder_run, 'coder-run'); const b = new Broker({ store, businesses: demo(), organization: null }); diff --git a/packages/bus/tests/comment-migration.test.mjs b/packages/bus/tests/comment-migration.test.mjs index 9645d92ea..c661a2b49 100644 --- a/packages/bus/tests/comment-migration.test.mjs +++ b/packages/bus/tests/comment-migration.test.mjs @@ -33,7 +33,7 @@ test('explicit 3b copy migration preserves evidence and opens as 3g; normal star const original = readFileSync(source); assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3m'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); assert.equal(store.get("SELECT body FROM events WHERE id='preserved'").body, '{"fixture":true}'); assert.equal(store.get('SELECT count(*) n FROM task_comment_cursors').n, 0); const epoch = store.get("SELECT value FROM meta WHERE key='comment_cursor_epoch'").value; @@ -75,15 +75,15 @@ test('named migration registry chains 3b through 3g to a fixture-only future ver const sha = createHash('sha256').update(futureDDL).digest('hex'); let source = readFileSync(new URL('../src/store.mjs', import.meta.url), 'utf8'); source = source.replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${sha}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3n';") - .replace('}];\nconst timedTables', `}, { from: '3m', to: '3n', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`); + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3o';") + .replace('}];\nconst timedTables', `}, { from: '3n', to: '3o', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`); writeFileSync(join(moduleRoot, 'src/store.mjs'), source); writeFileSync(join(moduleRoot, 'schema.sql'), futureDDL); - for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql', '3k-3l.sql', '3l-3m.sql']) + for (const name of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql', '3i-3j.sql', '3j-3k.sql', '3k-3l.sql', '3l-3m.sql', '3m-3n.sql']) copyFileSync(new URL('../src/migrations/' + name, import.meta.url), join(moduleRoot, 'src/migrations', name)); const { Store: FutureStore } = await import(pathToFileURL(join(moduleRoot, 'src/store.mjs'))); const future = new FutureStore(root, { migrateSchema: true }); - assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); + assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3o'); assert.equal(future.get("SELECT name FROM sqlite_master WHERE name='future_fixture'").name, 'future_fixture'); assert.equal(future.get("SELECT count(*) n FROM task_comment_cursors").n, 0); assert.equal(future.get("SELECT id FROM events").id, 'preserved'); @@ -109,7 +109,7 @@ test('offline command migrates only the supplied fixture root and reports usage const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname; const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' }); assert.equal(migrated.status, 0, migrated.stderr); - assert.match(migrated.stdout, /bus schema 3m ready/); + assert.match(migrated.stdout, /bus schema 3n ready/); new Store(root).close(); const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' }); assert.equal(usage.status, 2); diff --git a/packages/bus/tests/fixtures/schema-3m.sql b/packages/bus/tests/fixtures/schema-3m.sql new file mode 100644 index 000000000..1f0dd8752 --- /dev/null +++ b/packages/bus/tests/fixtures/schema-3m.sql @@ -0,0 +1,495 @@ +PRAGMA journal_mode = WAL; +PRAGMA foreign_keys = ON; +CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL) STRICT; +CREATE TABLE events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, + at TEXT NOT NULL, + business TEXT NOT NULL, + kind TEXT NOT NULL CHECK (kind IN ( + 'session.launched', + 'session.ended', + 'action.allowed', + 'action.refused', + 'task.created', + 'task.assigned', + 'task.state', + 'task.closed', + 'task.changed.external', + 'task.conflict', + 'task.missing', + 'review.requested', + 'review.verdict', + 'human.input', + 'config.refused', + 'credential.expiring', + 'credential.expired', + 'credential.changed', + 'launch.revoked', + 'launch.restored', + 'digest.sent')), + actor_role TEXT, actor_run TEXT, + subject TEXT, + corrects TEXT REFERENCES events(id), + body TEXT NOT NULL CHECK (json_valid(body)) +) STRICT; +CREATE TABLE role_claims ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + business TEXT NOT NULL, role TEXT NOT NULL, + op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')), + holder_run TEXT NOT NULL, + harness TEXT NOT NULL, address TEXT, + by TEXT NOT NULL, reason TEXT, + decision TEXT +) STRICT; +CREATE TABLE decisions ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, + at TEXT NOT NULL, + business TEXT NOT NULL, project TEXT, + raised_by_role TEXT NOT NULL, raised_by_run TEXT NOT NULL, + class TEXT NOT NULL CHECK (class IN ('routine','within-role','cross-role','gated')), + action TEXT NOT NULL, + route_to TEXT NOT NULL, + question TEXT NOT NULL, + options TEXT NOT NULL CHECK (json_valid(options) AND json_array_length(options) BETWEEN 2 AND 9), + recommendation TEXT NOT NULL, + task_ref TEXT CHECK (task_ref IS NULL OR (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*')), requirement_ref TEXT, + blocking INTEGER NOT NULL CHECK (blocking IN (0,1)), + supersedes TEXT REFERENCES decisions(id) +, spec_path TEXT, spec_commit TEXT) STRICT; +CREATE TABLE decision_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + decision TEXT NOT NULL REFERENCES decisions(id), + at TEXT NOT NULL, + op TEXT NOT NULL CHECK (op IN ('seen','resolved','withdrawn','expired')), + by TEXT NOT NULL, + choice TEXT, note TEXT, via TEXT +) STRICT; +CREATE TABLE messages ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, + at TEXT NOT NULL, + business TEXT NOT NULL, + from_role TEXT NOT NULL, from_run TEXT NOT NULL, + to_role TEXT NOT NULL, + class TEXT NOT NULL, + in_reply_to TEXT REFERENCES messages(id), + decision TEXT REFERENCES decisions(id), + corrects TEXT REFERENCES messages(id), + body TEXT NOT NULL +, to_name TEXT) STRICT; +CREATE TABLE deliveries ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + message TEXT NOT NULL REFERENCES messages(id), + at TEXT NOT NULL, + op TEXT NOT NULL CHECK (op IN ('routed','delivered','failed','read')), + holder_run TEXT, transport TEXT, address TEXT, detail TEXT +) STRICT; +CREATE TABLE task_snapshots ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + business TEXT NOT NULL, + task_ref TEXT NOT NULL CHECK (task_ref GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(task_ref, 9) GLOB '*[^0-9/]*' AND NOT substr(task_ref, 9) GLOB '*/*/*'), + updated TEXT NOT NULL, + etag TEXT, + digest TEXT NOT NULL CHECK (length(digest) = 64 AND NOT digest GLOB '*[^0-9a-f]*'), + fields TEXT NOT NULL CHECK (json_valid(fields)), + source TEXT NOT NULL CHECK (source IN ('self','poll')), + via TEXT CHECK (via IN ('board','cursor','task','reconcile')), + read_at TEXT, + role TEXT, run TEXT, + CHECK ((source = 'self') = (role IS NOT NULL AND run IS NOT NULL)), + CHECK ((source = 'poll') = (via IS NOT NULL AND read_at IS NOT NULL)), + CHECK (json_type(fields, '$.bucket') IS 'integer' + OR (source IS 'poll' AND via IS 'task' AND json_type(fields, '$.gone') IS 'text')) +) STRICT; +CREATE INDEX task_snapshots_ref ON task_snapshots (business, task_ref, seq); +CREATE TRIGGER decisions_resolve_once BEFORE INSERT ON decision_events + WHEN NEW.op IN ('resolved','withdrawn','expired') AND EXISTS ( + SELECT 1 FROM decision_events WHERE decision = NEW.decision AND op IN ('resolved','withdrawn','expired')) + BEGIN SELECT RAISE(ABORT, 'decision already closed'); END; +CREATE TRIGGER decisions_resolved_choice BEFORE INSERT ON decision_events + WHEN NEW.op = 'resolved' AND (NEW.choice IS NULL OR NOT EXISTS ( + SELECT 1 FROM decisions d, json_each(d.options) o WHERE d.id = NEW.decision AND json_extract(o.value,'$.key') = NEW.choice)) + BEGIN SELECT RAISE(ABORT, 'resolution must name one of the options'); END; +CREATE TRIGGER role_one_holder BEFORE INSERT ON role_claims + WHEN NEW.op = 'claim' AND (SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) = 'claim' + BEGIN SELECT RAISE(ABORT, 'role already held'); END; +CREATE TRIGGER role_release_by_holder BEFORE INSERT ON role_claims + WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1),'') <> 'claim' + BEGIN SELECT RAISE(ABORT, 'role is not held'); END; +CREATE TRIGGER role_release_same_run BEFORE INSERT ON role_claims + WHEN NEW.op = 'release' AND (SELECT holder_run FROM role_claims WHERE business = NEW.business AND role = NEW.role ORDER BY seq DESC LIMIT 1) <> NEW.holder_run + BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END; +CREATE TRIGGER role_revoke_needs_decision BEFORE INSERT ON role_claims + WHEN NEW.op = 'revoke' AND NEW.decision IS NULL AND NEW.reason IS NULL + BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision or a reason'); END; +CREATE TRIGGER meta_no_update BEFORE UPDATE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END; +CREATE TRIGGER meta_no_delete BEFORE DELETE ON meta BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END; +CREATE TRIGGER events_no_update BEFORE UPDATE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END; +CREATE TRIGGER events_no_delete BEFORE DELETE ON events BEGIN SELECT RAISE(ABORT, 'events is append-only'); END; +CREATE TRIGGER role_claims_no_update BEFORE UPDATE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END; +CREATE TRIGGER role_claims_no_delete BEFORE DELETE ON role_claims BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END; +CREATE TRIGGER decisions_no_update BEFORE UPDATE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END; +CREATE TRIGGER decisions_no_delete BEFORE DELETE ON decisions BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END; +CREATE TRIGGER decision_events_no_update BEFORE UPDATE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END; +CREATE TRIGGER decision_events_no_delete BEFORE DELETE ON decision_events BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END; +CREATE TRIGGER messages_no_update BEFORE UPDATE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END; +CREATE TRIGGER messages_no_delete BEFORE DELETE ON messages BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END; +CREATE TRIGGER deliveries_no_update BEFORE UPDATE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END; +CREATE TRIGGER deliveries_no_delete BEFORE DELETE ON deliveries BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END; +CREATE TRIGGER meta_no_replace BEFORE INSERT ON meta WHEN EXISTS (SELECT 1 FROM meta WHERE key = NEW.key) BEGIN SELECT RAISE(ABORT, 'meta is append-only'); END; +CREATE TRIGGER events_no_replace BEFORE INSERT ON events WHEN EXISTS (SELECT 1 FROM events WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'events is append-only'); END; +CREATE TRIGGER role_claims_no_replace BEFORE INSERT ON role_claims WHEN EXISTS (SELECT 1 FROM role_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'role_claims is append-only'); END; +CREATE TRIGGER decisions_no_replace BEFORE INSERT ON decisions WHEN EXISTS (SELECT 1 FROM decisions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'decisions is append-only'); END; +CREATE TRIGGER decision_events_no_replace BEFORE INSERT ON decision_events WHEN EXISTS (SELECT 1 FROM decision_events WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'decision_events is append-only'); END; +CREATE TRIGGER messages_no_replace BEFORE INSERT ON messages WHEN EXISTS (SELECT 1 FROM messages WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'messages is append-only'); END; +CREATE TRIGGER deliveries_no_replace BEFORE INSERT ON deliveries WHEN EXISTS (SELECT 1 FROM deliveries WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'deliveries is append-only'); END; +CREATE TRIGGER task_snapshots_no_update BEFORE UPDATE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END; +CREATE TRIGGER task_snapshots_no_delete BEFORE DELETE ON task_snapshots BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END; +CREATE TRIGGER task_snapshots_no_replace BEFORE INSERT ON task_snapshots WHEN EXISTS (SELECT 1 FROM task_snapshots WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'task_snapshots is append-only'); END; +CREATE TRIGGER decisions_blocking_needs_task BEFORE INSERT ON decisions + WHEN NEW.blocking = 1 AND NEW.task_ref IS NULL + BEGIN SELECT RAISE(ABORT, 'a blocking decision cites the task it blocks'); END; +CREATE TRIGGER events_launch_by_human BEFORE INSERT ON events + WHEN NEW.kind IN ('launch.revoked','launch.restored') AND (NEW.actor_role IS NOT NULL OR NEW.actor_run IS NOT NULL) + BEGIN SELECT RAISE(ABORT, 'only the human revokes or restores launching'); END; +CREATE TRIGGER events_credential_body BEFORE INSERT ON events + WHEN NEW.kind GLOB 'credential.*' AND ( + json_extract(NEW.body, '$.service') IS NULL OR json_extract(NEW.body, '$.service') NOT IN ('gitea','vikunja') + OR json_extract(NEW.body, '$.instance') IS NULL) + BEGIN SELECT RAISE(ABORT, 'credential events name a service and a role instance'); END; +CREATE TRIGGER events_task_missing_body BEFORE INSERT ON events + WHEN NEW.kind = 'task.missing' AND ( + json_extract(NEW.body, '$.reason') IS NULL OR json_extract(NEW.body, '$.reason') NOT IN ('moved','not-found','no-access') + OR (json_extract(NEW.body, '$.reason') = 'moved' AND json_type(NEW.body, '$.project') IS NOT 'integer')) + BEGIN SELECT RAISE(ABORT, 'task.missing carries a reason, and the new project when moved'); END; +CREATE TRIGGER events_task_subject BEFORE INSERT ON events + WHEN NEW.kind GLOB 'task.*' AND (NEW.subject IS NULL OR NOT (NEW.subject GLOB 'vikunja:[1-9]*/[1-9]*' AND NOT substr(NEW.subject, 9) GLOB '*[^0-9/]*' AND NOT substr(NEW.subject, 9) GLOB '*/*/*')) + BEGIN SELECT RAISE(ABORT, 'a task event names its task in subject'); END; +CREATE TRIGGER events_task_created_body BEFORE INSERT ON events + WHEN NEW.kind = 'task.created' AND ( + json_type(NEW.body, '$.request') IS NOT 'text' + OR NOT EXISTS (SELECT 1 FROM events WHERE id = json_extract(NEW.body, '$.request') + AND kind = 'human.input' AND business = NEW.business) + OR json_type(NEW.body, '$.requirement') IS NOT 'text' + OR NOT json_extract(NEW.body, '$.requirement') GLOB 'REQ-[A-Z]*-[1-9]*' + OR json_extract(NEW.body, '$.requirement') GLOB 'REQ-*[^A-Z0-9-]*') + BEGIN SELECT RAISE(ABORT, 'task.created cites the human.input that asked for it and a requirement id'); END; +CREATE VIEW launch_state AS + SELECT business, CASE kind WHEN 'launch.revoked' THEN 'revoked' ELSE 'allowed' END AS state, at + FROM events e WHERE kind IN ('launch.revoked','launch.restored') + AND seq = (SELECT max(seq) FROM events WHERE business = e.business AND kind IN ('launch.revoked','launch.restored')); +CREATE VIEW task_external_changes AS + SELECT p.business, p.task_ref, p.seq, p.via, p.updated, p.digest, ls.digest AS self_digest + FROM task_snapshots p + LEFT JOIN task_snapshots ls ON ls.seq = (SELECT max(seq) FROM task_snapshots + WHERE business = p.business AND task_ref = p.task_ref AND source = 'self') + WHERE p.source = 'poll' + AND p.seq = (SELECT max(seq) FROM task_snapshots WHERE business = p.business AND task_ref = p.task_ref) + AND (ls.seq IS NULL OR (p.updated >= ls.updated AND p.read_at > ls.at AND p.digest <> ls.digest)); +CREATE VIEW task_current AS + SELECT s.seq, s.at, s.business, s.task_ref, s.updated, s.etag, s.digest, s.fields, s.source, s.via, s.read_at, s.role, s.run + FROM task_snapshots s + WHERE s.seq = (SELECT max(c.seq) FROM task_snapshots c + WHERE c.business = s.business AND c.task_ref = s.task_ref + AND (c.source = 'self' + OR c.read_at > coalesce((SELECT ls.at FROM task_snapshots ls + WHERE ls.business = c.business AND ls.task_ref = c.task_ref AND ls.source = 'self' + ORDER BY ls.seq DESC LIMIT 1), ''))); +CREATE VIEW tasks_open AS + SELECT s.business, s.task_ref, json_extract(s.fields, '$.bucket') AS bucket, s.seq + FROM task_current s + WHERE json_type(s.fields, '$.gone') IS NULL + AND json_extract(s.fields, '$.done') = 0; +CREATE VIEW urgent_inbox AS + SELECT d.id, d.business, d.task_ref, d.question, d.at + FROM decisions d + WHERE d.class = 'gated' AND d.blocking = 1 + AND NOT EXISTS (SELECT 1 FROM decision_events x WHERE x.decision = d.id AND x.op IN ('resolved','withdrawn','expired')); +CREATE TABLE task_comment_cursors ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + business TEXT NOT NULL, + task_id INTEGER NOT NULL CHECK (task_id > 0), + last_comment INTEGER NOT NULL CHECK (last_comment > 0) +) STRICT; +CREATE INDEX task_comment_cursors_latest ON task_comment_cursors (business, task_id, seq); +CREATE TRIGGER task_comment_cursors_no_update BEFORE UPDATE ON task_comment_cursors + BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END; +CREATE TRIGGER task_comment_cursors_no_delete BEFORE DELETE ON task_comment_cursors + BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END; +CREATE TRIGGER task_comment_cursors_no_replace BEFORE INSERT ON task_comment_cursors + WHEN EXISTS (SELECT 1 FROM task_comment_cursors WHERE seq = NEW.seq) + BEGIN SELECT RAISE(ABORT, 'task_comment_cursors is append-only'); END; +CREATE TRIGGER task_comment_cursors_forward BEFORE INSERT ON task_comment_cursors + WHEN NEW.last_comment <= COALESCE((SELECT last_comment FROM task_comment_cursors + WHERE business = NEW.business AND task_id = NEW.task_id ORDER BY seq DESC LIMIT 1), 0) + BEGIN SELECT RAISE(ABORT, 'comment cursor must advance'); END; +CREATE TABLE projects ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, + at TEXT NOT NULL, + business TEXT NOT NULL, + name TEXT NOT NULL, + repo TEXT, + created_by TEXT NOT NULL, + UNIQUE (business, name) +) STRICT; +CREATE TABLE threads ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, + at TEXT NOT NULL, + business TEXT NOT NULL, + project TEXT NOT NULL REFERENCES projects(id), + instance TEXT NOT NULL, + owner TEXT NOT NULL, + name TEXT NOT NULL, + UNIQUE (project, name) +) STRICT; +CREATE TABLE sessions ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + id TEXT NOT NULL UNIQUE, + at TEXT NOT NULL, + business TEXT NOT NULL, + thread TEXT NOT NULL REFERENCES threads(id), + instance TEXT NOT NULL, + owner TEXT NOT NULL, + name TEXT NOT NULL, + op TEXT NOT NULL, + predecessor TEXT REFERENCES sessions(id), + UNIQUE (thread, name), + UNIQUE (thread, op) +) STRICT; +CREATE TABLE session_defaults ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + thread TEXT NOT NULL REFERENCES threads(id), + session TEXT NOT NULL REFERENCES sessions(id), + by TEXT NOT NULL +) STRICT; +CREATE TABLE thread_claims ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + business TEXT NOT NULL, + thread TEXT NOT NULL REFERENCES threads(id), + session TEXT NOT NULL REFERENCES sessions(id), + generation INTEGER NOT NULL CHECK (generation >= 1), + op TEXT NOT NULL CHECK (op IN ('claim','release','revoke')), + holder_run TEXT NOT NULL, + harness TEXT NOT NULL, address TEXT, + by TEXT NOT NULL, reason TEXT, + decision TEXT +) STRICT; +CREATE TABLE message_threads ( + message TEXT PRIMARY KEY REFERENCES messages(id), + thread TEXT NOT NULL REFERENCES threads(id), + session TEXT NOT NULL REFERENCES sessions(id) +) STRICT; +CREATE TABLE thread_opens ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + thread TEXT NOT NULL REFERENCES threads(id), + session TEXT NOT NULL REFERENCES sessions(id), + by TEXT NOT NULL +) STRICT; +CREATE INDEX thread_claims_thread ON thread_claims (thread, seq); +CREATE INDEX thread_claims_session ON thread_claims (session, seq); +CREATE INDEX message_threads_thread ON message_threads (thread); +CREATE INDEX message_threads_session ON message_threads (session); +CREATE INDEX session_defaults_thread ON session_defaults (thread, seq); +CREATE TRIGGER threads_project_business BEFORE INSERT ON threads + WHEN (SELECT business FROM projects WHERE id = NEW.project) IS NOT NEW.business + BEGIN SELECT RAISE(ABORT, 'a thread belongs to its project''s business'); END; +CREATE TRIGGER sessions_thread_match BEFORE INSERT ON sessions + WHEN NOT EXISTS (SELECT 1 FROM threads t WHERE t.id = NEW.thread AND t.business = NEW.business AND t.instance = NEW.instance AND t.owner = NEW.owner) + BEGIN SELECT RAISE(ABORT, 'a session has its thread''s business, instance and owner'); END; +CREATE TRIGGER sessions_predecessor BEFORE INSERT ON sessions + WHEN NEW.predecessor IS NOT NULL AND (SELECT thread FROM sessions WHERE id = NEW.predecessor) IS NOT NEW.thread + BEGIN SELECT RAISE(ABORT, 'a session''s predecessor is in its thread'); END; +CREATE TRIGGER sessions_id_not_thread BEFORE INSERT ON sessions + WHEN NEW.id IS NOT NEW.thread AND EXISTS (SELECT 1 FROM threads WHERE id = NEW.id) + BEGIN SELECT RAISE(ABORT, 'a session id names another thread'); END; +CREATE TRIGGER threads_id_not_session BEFORE INSERT ON threads + WHEN EXISTS (SELECT 1 FROM sessions WHERE id = NEW.id) + BEGIN SELECT RAISE(ABORT, 'a thread id names a session'); END; +CREATE TRIGGER session_defaults_match BEFORE INSERT ON session_defaults + WHEN (SELECT thread FROM sessions WHERE id = NEW.session) IS NOT NEW.thread + BEGIN SELECT RAISE(ABORT, 'a default session is in its thread'); END; +CREATE TRIGGER thread_one_holder BEFORE INSERT ON thread_claims + WHEN NEW.op = 'claim' AND (SELECT op FROM thread_claims WHERE session = NEW.session ORDER BY seq DESC LIMIT 1) = 'claim' + BEGIN SELECT RAISE(ABORT, 'session already held'); END; +CREATE TRIGGER thread_one_per_run BEFORE INSERT ON thread_claims + WHEN NEW.op = 'claim' AND EXISTS (SELECT 1 FROM thread_claims c WHERE c.holder_run = NEW.holder_run AND c.business = NEW.business + AND c.seq = (SELECT max(seq) FROM thread_claims WHERE session = c.session) AND c.op = 'claim') + BEGIN SELECT RAISE(ABORT, 'run already holds a session'); END; +CREATE TRIGGER thread_release_by_holder BEFORE INSERT ON thread_claims + WHEN NEW.op IN ('release','revoke') AND COALESCE((SELECT op FROM thread_claims WHERE session = NEW.session ORDER BY seq DESC LIMIT 1),'') <> 'claim' + BEGIN SELECT RAISE(ABORT, 'session is not held'); END; +CREATE TRIGGER thread_release_same_run BEFORE INSERT ON thread_claims + WHEN NEW.op = 'release' AND (SELECT holder_run FROM thread_claims WHERE session = NEW.session ORDER BY seq DESC LIMIT 1) <> NEW.holder_run + BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); END; +CREATE TRIGGER thread_claim_generation BEFORE INSERT ON thread_claims + WHEN NEW.generation IS NOT (CASE WHEN NEW.op = 'claim' + THEN 1 + (SELECT count(*) FROM thread_claims WHERE session = NEW.session AND op = 'claim') + ELSE (SELECT generation FROM thread_claims WHERE session = NEW.session ORDER BY seq DESC LIMIT 1) END) + BEGIN SELECT RAISE(ABORT, 'a claim row names its session''s generation'); END; +CREATE TRIGGER thread_revoke_needs_decision BEFORE INSERT ON thread_claims + WHEN NEW.op = 'revoke' AND NEW.decision IS NULL AND NEW.reason IS NULL + BEGIN SELECT RAISE(ABORT, 'revoke needs a resolved decision or a reason'); END; +CREATE TRIGGER thread_claims_business BEFORE INSERT ON thread_claims + WHEN (SELECT business FROM threads WHERE id = NEW.thread) IS NOT NEW.business + BEGIN SELECT RAISE(ABORT, 'a thread claim names the thread''s business'); END; +CREATE TRIGGER thread_claims_session BEFORE INSERT ON thread_claims + WHEN (SELECT thread FROM sessions WHERE id = NEW.session) IS NOT NEW.thread + BEGIN SELECT RAISE(ABORT, 'a claim''s session is in its thread'); END; +CREATE TRIGGER message_threads_match BEFORE INSERT ON message_threads + WHEN NOT EXISTS (SELECT 1 FROM messages m JOIN threads t ON t.id = NEW.thread + WHERE m.id = NEW.message AND m.business = t.business AND t.instance IN (m.to_role, m.from_role)) + BEGIN SELECT RAISE(ABORT, 'a threaded message is to or from the thread''s instance'); END; +CREATE TRIGGER message_threads_session BEFORE INSERT ON message_threads + WHEN (SELECT thread FROM sessions WHERE id = NEW.session) IS NOT NEW.thread + BEGIN SELECT RAISE(ABORT, 'a threaded message''s session is in its thread'); END; +CREATE TRIGGER thread_opens_session BEFORE INSERT ON thread_opens + WHEN (SELECT thread FROM sessions WHERE id = NEW.session) IS NOT NEW.thread + BEGIN SELECT RAISE(ABORT, 'an open''s session is in its thread'); END; +CREATE TRIGGER delivery_same_session BEFORE INSERT ON deliveries + WHEN NEW.op IN ('routed','delivered') AND (NEW.op = 'delivered' OR NEW.holder_run IS NOT NULL) + AND EXISTS (SELECT 1 FROM message_threads mt JOIN messages m ON m.id = mt.message JOIN threads t ON t.id = mt.thread + WHERE mt.message = NEW.message AND m.to_role = t.instance + AND NOT EXISTS (SELECT 1 FROM thread_claims c WHERE c.seq = (SELECT max(seq) FROM thread_claims WHERE session = mt.session) + AND c.op = 'claim' AND c.holder_run IS NEW.holder_run)) + BEGIN SELECT RAISE(ABORT, 'a threaded message goes only to its session''s holder'); END; +CREATE TRIGGER projects_no_update BEFORE UPDATE ON projects BEGIN SELECT RAISE(ABORT, 'projects is append-only'); END; +CREATE TRIGGER projects_no_delete BEFORE DELETE ON projects BEGIN SELECT RAISE(ABORT, 'projects is append-only'); END; +CREATE TRIGGER projects_no_replace BEFORE INSERT ON projects WHEN EXISTS (SELECT 1 FROM projects WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'projects is append-only'); END; +CREATE TRIGGER threads_no_update BEFORE UPDATE ON threads BEGIN SELECT RAISE(ABORT, 'threads is append-only'); END; +CREATE TRIGGER threads_no_delete BEFORE DELETE ON threads BEGIN SELECT RAISE(ABORT, 'threads is append-only'); END; +CREATE TRIGGER threads_no_replace BEFORE INSERT ON threads WHEN EXISTS (SELECT 1 FROM threads WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'threads is append-only'); END; +CREATE TRIGGER sessions_no_update BEFORE UPDATE ON sessions BEGIN SELECT RAISE(ABORT, 'sessions is append-only'); END; +CREATE TRIGGER sessions_no_delete BEFORE DELETE ON sessions BEGIN SELECT RAISE(ABORT, 'sessions is append-only'); END; +CREATE TRIGGER sessions_no_replace BEFORE INSERT ON sessions WHEN EXISTS (SELECT 1 FROM sessions WHERE seq = NEW.seq OR id = NEW.id) BEGIN SELECT RAISE(ABORT, 'sessions is append-only'); END; +CREATE TRIGGER session_defaults_no_update BEFORE UPDATE ON session_defaults BEGIN SELECT RAISE(ABORT, 'session_defaults is append-only'); END; +CREATE TRIGGER session_defaults_no_delete BEFORE DELETE ON session_defaults BEGIN SELECT RAISE(ABORT, 'session_defaults is append-only'); END; +CREATE TRIGGER session_defaults_no_replace BEFORE INSERT ON session_defaults WHEN EXISTS (SELECT 1 FROM session_defaults WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'session_defaults is append-only'); END; +CREATE TRIGGER thread_claims_no_update BEFORE UPDATE ON thread_claims BEGIN SELECT RAISE(ABORT, 'thread_claims is append-only'); END; +CREATE TRIGGER thread_claims_no_delete BEFORE DELETE ON thread_claims BEGIN SELECT RAISE(ABORT, 'thread_claims is append-only'); END; +CREATE TRIGGER thread_claims_no_replace BEFORE INSERT ON thread_claims WHEN EXISTS (SELECT 1 FROM thread_claims WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'thread_claims is append-only'); END; +CREATE TRIGGER message_threads_no_update BEFORE UPDATE ON message_threads BEGIN SELECT RAISE(ABORT, 'message_threads is append-only'); END; +CREATE TRIGGER message_threads_no_delete BEFORE DELETE ON message_threads BEGIN SELECT RAISE(ABORT, 'message_threads is append-only'); END; +CREATE TRIGGER message_threads_no_replace BEFORE INSERT ON message_threads WHEN EXISTS (SELECT 1 FROM message_threads WHERE message = NEW.message) BEGIN SELECT RAISE(ABORT, 'message_threads is append-only'); END; +CREATE TRIGGER thread_opens_no_update BEFORE UPDATE ON thread_opens BEGIN SELECT RAISE(ABORT, 'thread_opens is append-only'); END; +CREATE TRIGGER thread_opens_no_delete BEFORE DELETE ON thread_opens BEGIN SELECT RAISE(ABORT, 'thread_opens is append-only'); END; +CREATE TRIGGER thread_opens_no_replace BEFORE INSERT ON thread_opens WHEN EXISTS (SELECT 1 FROM thread_opens WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'thread_opens is append-only'); END; +CREATE TABLE instance_names ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + business TEXT NOT NULL, + instance TEXT NOT NULL, + name TEXT NOT NULL, + by TEXT NOT NULL +) STRICT; +CREATE TRIGGER instance_names_unique BEFORE INSERT ON instance_names + WHEN EXISTS (SELECT 1 FROM instance_names n WHERE n.business = NEW.business AND n.name = NEW.name + AND n.instance <> NEW.instance + AND n.seq = (SELECT max(seq) FROM instance_names WHERE business = n.business AND instance = n.instance)) + BEGIN SELECT RAISE(ABORT, 'an instance name is unique in its business'); END; +CREATE TRIGGER instance_names_no_update BEFORE UPDATE ON instance_names BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END; +CREATE TRIGGER instance_names_no_delete BEFORE DELETE ON instance_names BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END; +CREATE TRIGGER instance_names_no_replace BEFORE INSERT ON instance_names WHEN EXISTS (SELECT 1 FROM instance_names WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'instance_names is append-only'); END; +-- Row 124 (#1618): immutable source snapshots and turn outcomes, never transcript text. +CREATE TABLE usage_records ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + business TEXT NOT NULL, + run TEXT NOT NULL, + instance TEXT NOT NULL, + provider TEXT, + model TEXT, + account TEXT, + turn TEXT NOT NULL, + source TEXT NOT NULL, + phase TEXT NOT NULL CHECK (phase IN ('begin','end')), + status TEXT NOT NULL CHECK (status IN ('known','unknown')), + code TEXT, + cursor TEXT CHECK (cursor IS NULL OR json_valid(cursor)), + input INTEGER, + output INTEGER, + cache_write INTEGER, + cache_read INTEGER, + UNIQUE (business,run,turn,source,phase), + CHECK ((status='unknown' AND input IS NULL AND output IS NULL AND cache_write IS NULL AND cache_read IS NULL) + OR (status='known' AND input >= 0 AND output >= 0 AND cache_write >= 0 AND cache_read >= 0 + AND input IS NOT NULL AND output IS NOT NULL AND cache_write IS NOT NULL AND cache_read IS NOT NULL)) +) STRICT; +CREATE TRIGGER usage_records_no_update BEFORE UPDATE ON usage_records BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END; +CREATE TRIGGER usage_records_no_delete BEFORE DELETE ON usage_records BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END; +CREATE TRIGGER usage_records_no_replace BEFORE INSERT ON usage_records WHEN EXISTS (SELECT 1 FROM usage_records WHERE seq=NEW.seq OR (business=NEW.business AND run=NEW.run AND turn=NEW.turn AND source=NEW.source AND phase=NEW.phase)) BEGIN SELECT RAISE(ABORT, 'usage_records is append-only'); END; +CREATE TABLE hosts ( + business TEXT NOT NULL, + host TEXT NOT NULL, + address TEXT NOT NULL, + port INTEGER NOT NULL CHECK (port BETWEEN 1 AND 65535), + user TEXT NOT NULL, + sudo INTEGER NOT NULL CHECK (sudo IN (0,1)), + host_keys TEXT NOT NULL CHECK (json_valid(host_keys) AND json_type(host_keys) = 'array' AND json_array_length(host_keys) > 0), + installer_fingerprint TEXT, + state TEXT NOT NULL CHECK (state IN ('pending-install','ready')), + decision TEXT NOT NULL REFERENCES decisions(id), + checked_at TEXT, + PRIMARY KEY (business, host), + CHECK (state = 'pending-install' OR (installer_fingerprint IS NOT NULL AND checked_at IS NOT NULL)) +) STRICT; +CREATE TABLE host_grants ( + business TEXT NOT NULL, + agent TEXT NOT NULL, + host TEXT NOT NULL, + commands TEXT NOT NULL CHECK (json_valid(commands) AND json_type(commands) = 'array' AND json_array_length(commands) > 0), + rotate_by TEXT NOT NULL, + tag TEXT NOT NULL CHECK (length(tag) = 39 AND tag GLOB 'mosaic-*' AND NOT substr(tag, 8) GLOB '*[^0-9a-f]*'), + decision TEXT NOT NULL REFERENCES decisions(id), + PRIMARY KEY (business, agent, host), + FOREIGN KEY (business, host) REFERENCES hosts(business, host) +) STRICT; +CREATE TRIGGER events_grant_actor BEFORE INSERT ON events + WHEN NEW.kind = 'action.allowed' + AND (json_extract(NEW.body, '$.action') GLOB 'grant.*' OR json_extract(NEW.body, '$.operation') GLOB 'grant.*') + AND NOT (NEW.actor_role IS NULL AND NEW.actor_run IS 'grant') + BEGIN SELECT RAISE(ABORT, 'only the grant writer records grant events'); END; +CREATE TABLE gitea_users ( + business TEXT NOT NULL, + agent TEXT NOT NULL, + user TEXT NOT NULL, + id INTEGER NOT NULL CHECK (id > 0), + nonce TEXT NOT NULL CHECK (length(nonce) = 16 AND NOT nonce GLOB '*[^0-9a-f]*'), + decision TEXT NOT NULL REFERENCES decisions(id), + PRIMARY KEY (business, agent) +) STRICT; +CREATE TABLE gitea_grants ( + business TEXT NOT NULL, + agent TEXT NOT NULL, + user TEXT NOT NULL, + token_id INTEGER NOT NULL CHECK (token_id > 0), + token_name TEXT NOT NULL, + projects TEXT NOT NULL CHECK (json_valid(projects) AND json_type(projects) = 'array' AND json_array_length(projects) > 0), + rotate_by TEXT NOT NULL, + decision TEXT NOT NULL REFERENCES decisions(id), + PRIMARY KEY (business, agent), + FOREIGN KEY (business, agent) REFERENCES gitea_users(business, agent) +) STRICT; +-- Row 206 (193b2, #1704, row 193 design 5.2): a thread owner's read-only shares. The latest row per +-- (thread, target) is the share's state. Append-only. +CREATE TABLE thread_shares ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + business TEXT NOT NULL, + thread TEXT NOT NULL REFERENCES threads(id), + op TEXT NOT NULL CHECK (op IN ('share','unshare')), + target TEXT NOT NULL, + by TEXT NOT NULL +) STRICT; +CREATE INDEX thread_shares_thread ON thread_shares (thread, target, seq); +CREATE TRIGGER thread_shares_no_update BEFORE UPDATE ON thread_shares BEGIN SELECT RAISE(ABORT, 'thread_shares is append-only'); END; +CREATE TRIGGER thread_shares_no_delete BEFORE DELETE ON thread_shares BEGIN SELECT RAISE(ABORT, 'thread_shares is append-only'); END; +CREATE TRIGGER thread_shares_no_replace BEFORE INSERT ON thread_shares WHEN EXISTS (SELECT 1 FROM thread_shares WHERE seq = NEW.seq) BEGIN SELECT RAISE(ABORT, 'thread_shares is append-only'); END; diff --git a/packages/bus/tests/gitea-migration.test.mjs b/packages/bus/tests/gitea-migration.test.mjs index d5e98c5db..f8aa649d1 100644 --- a/packages/bus/tests/gitea-migration.test.mjs +++ b/packages/bus/tests/gitea-migration.test.mjs @@ -35,11 +35,12 @@ function source3j(t) { // Row 264a (#1769): what the 3k→3l step rebuilds or adds. sessions.test.mjs tests those objects. const SESSIONS_3L = ['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens']; // Row 206 (#1704): and the 3l→3m step's thread_shares. thread-shares.test.mjs tests it. -const later3l = (r) => [...SESSIONS_3L, 'thread_shares'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name); +// Row 271 (#1776): and the 3m→3n step's session_ops and message_cancels. session-clear.test.mjs tests them. +const later3l = (r) => [...SESSIONS_3L, 'thread_shares', 'session_ops', 'message_cancels'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name); function rows(path) { const db = new DatabaseSync(path, { readOnly: true }); try { - return Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT IN ('gitea_users','gitea_grants','sessions','session_defaults','thread_shares')").all() + return Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT IN ('gitea_users','gitea_grants','sessions','session_defaults','thread_shares','session_ops','message_cancels')").all() .map(({ name }) => [name, db.prepare(`SELECT * FROM ${name} ORDER BY rowid`).all() .filter(r => name !== 'meta' || !['schema_version', 'schema_digest'].includes(r.key)).map(r => ({ ...r }))])); } finally { db.close(); } @@ -76,6 +77,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { writeFileSync(join(moduleRoot, 'src/migrations/3j-3k.sql'), step(readFileSync(here('../src/migrations/3j-3k.sql'), 'utf8'))); copyFileSync(here('../src/migrations/3k-3l.sql'), join(moduleRoot, 'src/migrations/3k-3l.sql')); copyFileSync(here('../src/migrations/3l-3m.sql'), join(moduleRoot, 'src/migrations/3l-3m.sql')); + copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } function fresh(t) { @@ -97,7 +99,7 @@ test('3j copy migrates to 3k: every row kept, and sqlite_master gains only the t const { root, path } = source3j(t), before = rows(path), was = master(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(migrated3l(rows(path)), before); const now = master(path); assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r))); @@ -123,7 +125,7 @@ test('a 3j→3k step that fails partway leaves the store at 3j with every row an assert.deepEqual(master(path), was); assert.deepEqual(rows(path), before); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); }); test('the 3j engine refuses a migrated store, migrating or not, and leaves it as it was', async t => { @@ -134,14 +136,14 @@ test('the 3j engine refuses a migrated store, migrating or not, and leaves it as schema: ddl3j, store: (source) => source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3j).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3j';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3j';") .replace(/\}, \{\n {2}\/\/ Row 236 \(#1742[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'), }); const pinned = readFileSync(join(root, 'engine-3j/src/store.mjs'), 'utf8'); assert.ok(pinned.includes("export const SCHEMA_VERSION = '3j';") && !pinned.includes('3j-3k'), 'the 3j engine has no 3j→3k step'); assert.throws(() => new Store3j(root), /schema-mismatch/); assert.throws(() => new Store3j(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(master(path), was); assert.deepEqual(rows(path), after); }); diff --git a/packages/bus/tests/grant-migration.test.mjs b/packages/bus/tests/grant-migration.test.mjs index cb2a31b9f..baa9ba326 100644 --- a/packages/bus/tests/grant-migration.test.mjs +++ b/packages/bus/tests/grant-migration.test.mjs @@ -21,7 +21,8 @@ const LATER = ['gitea_users', 'gitea_grants', 'sqlite_autoindex_gitea_users_1', // Row 264a (#1769): what the 3k→3l step rebuilds or adds. sessions.test.mjs tests those objects. const SESSIONS_3L = ['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens']; // Row 206 (#1704): and the 3l→3m step's thread_shares. thread-shares.test.mjs tests it. -const later3l = (r) => [...SESSIONS_3L, 'thread_shares'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name); +// Row 271 (#1776): and the 3m→3n step's session_ops and message_cancels. session-clear.test.mjs tests them. +const later3l = (r) => [...SESSIONS_3L, 'thread_shares', 'session_ops', 'message_cancels'].includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session', 'threads_id_not_session'].includes(r.name); function source3i(t) { const root = mkdtempSync(join(tmpdir(), 'grant-migrate-')); t.after(() => rmSync(root, { recursive: true, force: true })); @@ -41,7 +42,7 @@ function source3i(t) { function rows(path) { const db = new DatabaseSync(path, { readOnly: true }); try { - return Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT IN ('hosts','host_grants','gitea_users','gitea_grants','sessions','session_defaults','thread_shares')").all() + return Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT IN ('hosts','host_grants','gitea_users','gitea_grants','sessions','session_defaults','thread_shares','session_ops','message_cancels')").all() .map(({ name }) => [name, db.prepare(`SELECT * FROM ${name} ORDER BY rowid`).all() .filter(r => name !== 'meta' || !['schema_version', 'schema_digest'].includes(r.key)).map(r => ({ ...r }))])); } finally { db.close(); } @@ -79,6 +80,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { copyFileSync(here('../src/migrations/3j-3k.sql'), join(moduleRoot, 'src/migrations/3j-3k.sql')); copyFileSync(here('../src/migrations/3k-3l.sql'), join(moduleRoot, 'src/migrations/3k-3l.sql')); copyFileSync(here('../src/migrations/3l-3m.sql'), join(moduleRoot, 'src/migrations/3l-3m.sql')); + copyFileSync(here('../src/migrations/3m-3n.sql'), join(moduleRoot, 'src/migrations/3m-3n.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } function fresh(t) { @@ -102,7 +104,7 @@ test('3i copy migrates to 3j: every row kept, and sqlite_master gains only the t const { root, path } = source3i(t), before = rows(path), was = master(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(migrated3l(rows(path)), before); const now = master(path).filter(r => !LATER.includes(r.name)); assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3l(r)), was.filter(r => !later3l(r))); @@ -127,7 +129,7 @@ test('a 3i→3j step that fails partway leaves the store at 3i with every row an assert.deepEqual(master(path), was); assert.deepEqual(rows(path), before); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); }); test('the 3i engine refuses a migrated store, migrating or not, and leaves it as it was', async t => { @@ -138,14 +140,14 @@ test('the 3i engine refuses a migrated store, migrating or not, and leaves it as schema: ddl3i, store: (source) => source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3i).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3i';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3i';") .replace(/\}, \{\n {2}\/\/ Row 234 \(#1740\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'), }); const pinned = readFileSync(join(root, 'engine-3i/src/store.mjs'), 'utf8'); assert.ok(pinned.includes("export const SCHEMA_VERSION = '3i';") && !pinned.includes('3i-3j'), 'the 3i engine has no 3i→3j step'); assert.throws(() => new Store3i(root), /schema-mismatch/); assert.throws(() => new Store3i(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(master(path), was); assert.deepEqual(rows(path), after); }); diff --git a/packages/bus/tests/name-migration.test.mjs b/packages/bus/tests/name-migration.test.mjs index 68392e39f..e6b80e7cf 100644 --- a/packages/bus/tests/name-migration.test.mjs +++ b/packages/bus/tests/name-migration.test.mjs @@ -65,7 +65,7 @@ test('3f copy migrates to 3g preserving every old column/row, with an empty inst assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); try { - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3m'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null }); assert.equal(store.get('SELECT count(*) AS n FROM instance_names').n, 0); assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: 'docs/spec.md', spec_commit: 'c'.repeat(40) }); diff --git a/packages/bus/tests/schema.test.mjs b/packages/bus/tests/schema.test.mjs index 9c68ae5f6..0f8b2878e 100644 --- a/packages/bus/tests/schema.test.mjs +++ b/packages/bus/tests/schema.test.mjs @@ -523,7 +523,8 @@ test('v3b prototype refusals, views and append-only mutations', (t) => { // Row 236 (#1742) adds gitea_users and gitea_grants; gitea-migration.test.mjs tests them. // Row 264a (#1769) adds sessions, session_defaults and 15 triggers; sessions.test.mjs tests them. // Row 206 (#1704) adds thread_shares and its 3 append-only triggers; thread-shares.test.mjs tests them. - assert.equal(count('table') - 1, 23); - assert.equal(count('trigger'), 89); + // Row 271 (#1776) adds session_ops, message_cancels and 7 triggers; session-clear.test.mjs tests them. + assert.equal(count('table') - 1, 25); + assert.equal(count('trigger'), 96); assert.equal(count('view'), 5); }); diff --git a/packages/bus/tests/sessions.test.mjs b/packages/bus/tests/sessions.test.mjs index cfd7369fa..7b15ab2ef 100644 --- a/packages/bus/tests/sessions.test.mjs +++ b/packages/bus/tests/sessions.test.mjs @@ -374,7 +374,7 @@ test('T7: a 3k store migrates to 3l, each thread\'s default session has the thre const { root, path } = source3k(t); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); // The migrated schema is the fresh one, object for object. const freshRoot = mkdtempSync(join(tmpdir(), 'sessions-fresh-')); t.after(() => rmSync(freshRoot, { recursive: true, force: true })); @@ -430,7 +430,7 @@ async function engine3k(root) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const edited = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3k).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3k';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3k';") .replace(/\}, \{\n {2}\/\/ Row 264a \(#1769\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(edited.includes("export const SCHEMA_VERSION = '3k';") && edited.includes('3j-3k') && !edited.includes('3k-3l'), 'the 3k engine has no 3k→3l step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), edited); @@ -447,7 +447,7 @@ test('T7 D9: the 3k engine refuses a 3l store, migrating or not, and leaves it a const Store3k = await engine3k(root); assert.throws(() => new Store3k(root), /schema-mismatch/); assert.throws(() => new Store3k(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(master(path), was); assert.deepEqual(rows(path), before); // The same engine opens a fresh 3k store, so the refusal is the version's and not a broken copy. diff --git a/packages/bus/tests/spec-migration.test.mjs b/packages/bus/tests/spec-migration.test.mjs index 691aa9234..8c3b738b2 100644 --- a/packages/bus/tests/spec-migration.test.mjs +++ b/packages/bus/tests/spec-migration.test.mjs @@ -63,7 +63,7 @@ test('3e copy migrates through 3f to 3g preserving every old column/row and addi assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); try { - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3m'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); assert.deepEqual({ ...store.get('SELECT spec_path,spec_commit FROM decisions') }, { spec_path: null, spec_commit: null }); assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok'); assert.deepEqual(store.all('PRAGMA foreign_key_check'), []); diff --git a/packages/bus/tests/tester-writes.test.mjs b/packages/bus/tests/tester-writes.test.mjs index 3382aac23..5beae2699 100644 --- a/packages/bus/tests/tester-writes.test.mjs +++ b/packages/bus/tests/tester-writes.test.mjs @@ -25,6 +25,8 @@ test('row 175: TESTER_VERBS and the organization branch tester set are pinned', 'instance.launch', 'instance.end', 'instance.request', // Row 264a (#1769): the sessions of the tester's own threads. 'session.list', 'session.create', + // Row 271 (264d, #1776): clear context in the tester's own threads (mos-1325). + 'session.clear', 'session.cancel', 'session.operation', ]); assert.deepEqual([...TESTER_ORGANIZATION_VERBS], ['limits', 'plugins', 'instances', 'plugins.set', 'instances.set', 'routes.set', 'routes']); assert.ok(Object.isFrozen(TESTER_VERBS) && Object.isFrozen(TESTER_ORGANIZATION_VERBS)); diff --git a/packages/bus/tests/thread-migration.test.mjs b/packages/bus/tests/thread-migration.test.mjs index 53ab2d099..5c9e65c4f 100644 --- a/packages/bus/tests/thread-migration.test.mjs +++ b/packages/bus/tests/thread-migration.test.mjs @@ -49,7 +49,7 @@ async function database3d(t) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const pinned = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3d).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3d';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3d';") .replace(/\}, \{\n {2}from: '3d',[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(!pinned.includes('3d-3e'), 'the 3d engine has no 3d→3e step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), pinned); @@ -101,7 +101,7 @@ test('REQ-THREAD-1 REQ-ORG-2: the 3d→3e migration adds the thread tables and l assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); t.after(() => store.close()); - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3m'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); for (const table of ['projects', 'threads', 'thread_claims', 'message_threads', 'thread_opens']) assert.equal(store.get(`SELECT count(*) n FROM ${table}`).n, 0, table); // Parked: the undelivered ones are counted per business and role, and none is changed or given a thread. @@ -161,5 +161,5 @@ test('REQ-THREAD-1: the offline command prints the kept-message counts before th const command = new URL('../src/migrate-schema.mjs', import.meta.url).pathname; const migrated = spawnSync(process.execPath, [command, root], { encoding: 'utf8' }); assert.equal(migrated.status, 0, migrated.stderr); - assert.equal(migrated.stdout, [...REPORT, 'bus schema 3m ready', ''].join('\n')); + assert.equal(migrated.stdout, [...REPORT, 'bus schema 3n ready', ''].join('\n')); }); diff --git a/packages/bus/tests/thread-shares.test.mjs b/packages/bus/tests/thread-shares.test.mjs index 53125e9ca..0c586b703 100644 --- a/packages/bus/tests/thread-shares.test.mjs +++ b/packages/bus/tests/thread-shares.test.mjs @@ -403,10 +403,15 @@ test('a 3l store migrates to 3m: thread_shares is new and empty, and every other const before = tables(path); assert.throws(() => new Store(root), /schema-mismatch/); new Store(root, { migrateSchema: true }).close(); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); const after = tables(path); assert.deepEqual(after.thread_shares, []); delete after.thread_shares; + // Row 271 (#1776): the 3m→3n step's session_ops and message_cancels, new and empty too. + for (const name of ['session_ops', 'message_cancels']) { + assert.deepEqual(after[name], [], name); + delete after[name]; + } assert.deepEqual(after, before); // The migrated schema is the fresh one, object for object. const freshRoot = mkdtempSync(join(tmpdir(), 'thread-shares-fresh-')); @@ -434,7 +439,7 @@ async function engine3l(root) { const source = readFileSync(here('../src/store.mjs'), 'utf8'); const edited = source .replace(/export const SCHEMA_SHA256 = '[a-f0-9]+';/, `export const SCHEMA_SHA256 = '${createHash('sha256').update(ddl3l).digest('hex')}';`) - .replace("export const SCHEMA_VERSION = '3m';", "export const SCHEMA_VERSION = '3l';") + .replace("export const SCHEMA_VERSION = '3n';", "export const SCHEMA_VERSION = '3l';") .replace(/\}, \{\n {2}\/\/ Row 206 \(193b2[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); assert.ok(edited.includes("export const SCHEMA_VERSION = '3l';") && edited.includes('3k-3l') && !edited.includes('3l-3m'), 'the 3l engine has no 3l→3m step'); writeFileSync(join(moduleRoot, 'src/store.mjs'), edited); @@ -451,7 +456,7 @@ test('the 3l engine refuses a 3m store, migrating or not, and leaves it as it wa const Store3l = await engine3l(root); assert.throws(() => new Store3l(root), /schema-mismatch/); assert.throws(() => new Store3l(root, { migrateSchema: true }), /schema-mismatch/); - assert.equal(version(path), '3m'); + assert.equal(version(path), '3n'); assert.deepEqual(master(path), was); assert.deepEqual(tables(path), before); // The same engine opens a fresh 3l store, so the refusal is the version's and not a broken copy. diff --git a/packages/bus/tests/usage-migration.test.mjs b/packages/bus/tests/usage-migration.test.mjs index 6358a2792..904ce584c 100644 --- a/packages/bus/tests/usage-migration.test.mjs +++ b/packages/bus/tests/usage-migration.test.mjs @@ -68,7 +68,7 @@ test('REQ-LIMIT-4 3g copy migrates through 3h, 3i, 3j and 3k to 3l preserving ev assert.throws(() => new Store(root), /schema-mismatch/); const store = new Store(root, { migrateSchema: true }); try { - assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3m'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3n'); assert.deepEqual({ ...store.get('SELECT to_name FROM messages') }, { to_name: null }); assert.equal(store.get('SELECT count(*) AS n FROM usage_records').n, 0); assert.equal(store.get('SELECT name FROM instance_names').name, 'Builder'); diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index 3fefb63bc..83d278deb 100644 --- a/packages/cli/src/launcher.mjs +++ b/packages/cli/src/launcher.mjs @@ -263,6 +263,9 @@ export function createLauncher({ const stopping = new Set(); const refusedThreads = new Map(); const startedFor = new Map(); + // Row 271 (264d): each clear's stop time (by op), and each clear refusal logged once (by op and code). + const clearStops = new Map(); + const clearLogged = new Map(); // Row 194 N6: run -> when its process was spawned, so an end signals only the runs spawned by its verb. const spawnedAt = new Map(); // Row 156: run -> the decisions its landing names; decision -> started; decision -> last @@ -676,9 +679,12 @@ export function createLauncher({ const ms = (at) => (at ? Date.parse(at) : 0); // Row 269: a thread with a legacy session keeps its old store for that session only. const legacyThreads = new Set(threads.filter((t) => t.legacy).map((t) => t.thread)); + // Row 271 (264d): a session in a clear is the clear pass's, and nothing below starts or idles it. + const clearing = await clearPass(now, legacyThreads); for (const t of threads) { if (closed) return; const key = `${t.thread}/${t.session}`; + if (clearing.has(t.session)) continue; const s = registry.running(b.id, t.instance, t.thread, t.session); if (s) { const last = Math.max(ms(s.startedAt), ms(t.lastMessage), ms(t.lastOpen)); @@ -726,6 +732,112 @@ export function createLauncher({ } } + // Row 271 (264d, #1776, D7 steps 3-5): one step of each clear the broker lists. The drain comes + // from the run itself (its receive, or its end), so a requested clear waits here. A drained run + // gets the idle stop's start-time-checked SIGTERM once, and 30 s without its exit proof records + // uncertain once; nothing is signalled again or forced. The proof is the run's own close handler, + // whose endRun the broker checks, or for no run, the session's workspace lock. Then retirement, + // and the successor's start through the same check()/start() as any session. Returns the + // sessions those steps own this tick. + async function clearPass(now, legacyThreads) { + const owned = new Set(); + let clears; + try { + clears = await host.op({ op: "sessionClears", business: b.id }); + } catch (e) { + clearLog("sessionClears", e.code ?? e.message, `sessionClears refused: ${e.code ?? e.message}`); + return owned; + } + for (const c of clears) { + if (closed) break; + const start = ["retired", "start-held", "start-refused"].includes(c.stage); + owned.add(start ? c.successor : c.session); + try { + await (start ? clearStart(c, legacyThreads) : clearStop(c, now)); + } catch (e) { + clearLog(c.op, e.code ?? e.message, `clear ${c.op} (session ${c.session}): ${e.code ?? e.message}`); + } + } + return owned; + } + + function clearLog(op, code, line) { + if (clearLogged.get(op) === code) return; + clearLogged.set(op, code); + log(line); + } + + async function clearStop(c, now) { + const where = { business: b.id, thread: c.thread, op: c.op }; + let stage = c.stage; + if (stage === "drained" || stage === "uncertain") { + const s = registry.running(b.id, c.instance, c.thread, c.session); + if (s && s.run === c.run) { + if (!clearStops.has(c.op)) { + // An uncertain clear was signalled before this launcher started, and isn't signalled again. + if (stage === "uncertain") return void clearStops.set(c.op, now); + // A cancel since the snapshot leaves the run alone; one after this read is a residual. + const fresh = (await host.op({ op: "sessionClears", business: b.id })).find((x) => x.thread === c.thread && x.op === c.op); + if (fresh?.stage !== "drained") return; + clearStops.set(c.op, now); + stopping.add(s.run); + record({ event: "session-clear-stop", instance: c.instance, thread: c.thread, session: c.session, run: s.run, op: c.op }); + try { + if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM"); + } catch {} + } else if (stage === "drained" && now - clearStops.get(c.op) >= stopTimeoutMs) { + await host.op({ op: "sessionClearUncertain", record: where }); + record({ event: "session-clear-uncertain", instance: c.instance, thread: c.thread, session: c.session, run: s.run, op: c.op }); + } + return; + } + // Some other run of the session (one that never claimed) exits on its refused claim first. + if (s) return; + // An earlier host's run that outlived recover()'s kill was ended host-lost, but it still runs: + // its exit, not that end, is the proof (row 269 Q2). It isn't signalled. + if (survivors.some((x) => x.business === b.id && x.run === c.run && liveRun(x))) return; + if (c.run === null) { + // No run: the proof is the session's workspace lock, which a spawned run holds until its close. + const live = registry.sessions.filter((x) => x.business === b.id && x.workspace).map((x) => x.workspace); + const workspace = sessionWorkspace(dataRoot, b.id, c.instance, c.session, { live }); + lockSessionWorkspace(workspace, { hostPid: process.pid, hostStartTime: startTimeOf(process.pid), business: b.id, instance: c.instance, thread: c.thread, session: c.session }).release(); + } + await host.op({ op: "sessionClearExited", record: { ...where, run: c.run } }); + stage = "exited"; + } + if (stage !== "exited") return; + const out = await host.op({ op: "sessionClearRetire", record: where }); + clearStops.delete(c.op); + record({ event: out.state === "retired" ? "session-clear-retired" : "session-clear-failed", instance: c.instance, thread: c.thread, session: c.session, successor: c.successor, op: c.op, ...(out.refusal ? { code: out.refusal } : {}) }); + } + + // D8: capacity and limits hold a start, which spends no work attempt. Any other refusal is one. + const CLEAR_HELD = new Set(["capacity-full", "limit-reached"]); + + async function clearStart(c, legacyThreads) { + const where = { business: b.id, thread: c.thread, op: c.op }; + await serialize(async () => { + if (closed) return; + if (!registry.running(b.id, c.instance, c.thread, c.successor)) { + const session = { thread: c.thread, session: c.successor }; + try { + await start({ instance: c.instance, launchedBy: b.human, ...session, legacy: false, legacyThread: legacyThreads.has(c.thread), ...(await check(c.instance, session)) }); + } catch (e) { + const code = e instanceof Refusal ? e.code : "launch-failed"; + // Reported and logged once per code, not once per tick. + if (clearLogged.get(c.op) === code) return; + clearLogged.set(c.op, code); + log(`clear ${c.op}: successor ${c.successor} (${c.instance}): start ${CLEAR_HELD.has(code) ? "held" : "refused"}: ${code}`); + record({ event: "refused", by: b.human, instance: c.instance, thread: c.thread, session: c.successor, code, why: e.message, ...(e.limit ? { limit: e.limit } : {}) }); + await host.op({ op: "sessionClearStart", record: { ...where, start: CLEAR_HELD.has(code) ? "held" : "refused", refusal: code } }); + return; + } + } + clearLogged.delete(c.op); + await host.op({ op: "sessionClearStart", record: { ...where, start: "started" } }); + }); + } + // Row 156 (#1651): start each listed merge decision holder once, and stop a // running holder that started before its decision's resolution. async function landPass() { diff --git a/packages/cli/tests/stack-release.test.mjs b/packages/cli/tests/stack-release.test.mjs index beb8da252..34bc48579 100644 --- a/packages/cli/tests/stack-release.test.mjs +++ b/packages/cli/tests/stack-release.test.mjs @@ -858,7 +858,7 @@ const COUNTED=/^COUNTED="([^"]*)"$/m.exec(srText)[1].split(' '); const STORE_JS=/^STORE_JS='([\s\S]*?)'$/m.exec(srText)[1]; // sage-1163's list 2, written out so that a table dropped from the script's COUNTED fails T45 as // well as T46. -const LISTED=['role_claims','thread_claims','decision_events','messages','deliveries','task_snapshots','projects','threads','sessions','session_defaults','thread_opens','thread_shares','instance_names','usage_records']; +const LISTED=['role_claims','thread_claims','decision_events','messages','deliveries','task_snapshots','projects','threads','sessions','session_defaults','thread_opens','thread_shares','instance_names','usage_records','session_ops','message_cancels']; const LATE='2999-01-01T00:00:00.000Z'; const addRow=(file,table,at=LATE)=>{const db=new DatabaseSync(file);db.exec(`CREATE TABLE IF NOT EXISTS ${table} (seq INTEGER PRIMARY KEY, at TEXT NOT NULL)`);db.prepare(`INSERT INTO ${table} (at) VALUES (?)`).run(at);db.close()}; diff --git a/scripts/stack-release.sh b/scripts/stack-release.sh index 92bd6f7d0..7a8c347b2 100755 --- a/scripts/stack-release.sh +++ b/scripts/stack-release.sh @@ -417,7 +417,7 @@ plugin_checks() { # Row 181: the tables a bus writer fills without an event in the same transaction, in schema # order. A restore drops their rows after STAMP as it drops events, so the guard counts them too. # BUILD.md for row 181 names each writer; the T46 guard classifies every table in the schema. -COUNTED="role_claims thread_claims decision_events messages deliveries task_snapshots projects threads sessions session_defaults thread_opens thread_shares instance_names usage_records" +COUNTED="role_claims thread_claims decision_events messages deliveries task_snapshots projects threads sessions session_defaults thread_opens thread_shares instance_names usage_records session_ops message_cancels" STORE_JS=' const {DatabaseSync}=require("node:sqlite");const {pathToFileURL}=require("node:url");