diff --git a/agents/gizmoduck/work/queue-264/BUILD.md b/agents/gizmoduck/work/queue-264/BUILD.md index dab45f44f..bb968704b 100644 --- a/agents/gizmoduck/work/queue-264/BUILD.md +++ b/agents/gizmoduck/work/queue-264/BUILD.md @@ -43,7 +43,7 @@ One seat can't build D1-D9 in six days and keep every slice reviewable. I propos under row 264, each with its own manifest and Duckworth round: 1. 264a bus. Session tables, claim key `(organization, thread, instance, session)` with a - generation, the D3 default session, the D9 bus mapping, `session.list|get|create|open|send` and an + generation, the D3 default session, the D9 bus mapping, `session.list` and `session.create`, and an optional session selector on the thread verbs. Tests: T1 bus half, T2 claims, T7 legacy clients and the rollback refusal. 2. 264b seat and launcher. Per-session native store and workspace, the single-writer fence, the @@ -126,8 +126,10 @@ Line numbers are at da9555ca2 unless they say otherwise. - The thread verbs take an optional `session`, and leaving it out uses the default (D3). `thread.create` also makes the default session. `thread.messages` without a session reads the default session. - - `session.list {thread}`, `session.get {session}`, `session.create {thread, name, op}` (D8: - metadata only), `session.open {session}` and `session.send {session, body, class}`. + - `session.list {thread}` and `session.create {thread, name, op}` (D8: metadata only). The build + dropped `session.get`, `session.open` and `session.send`. `thread.open`, `thread.send` and + `thread.messages` take an optional `session`, which does the same job with one owner check and + one code path. - All of them are human or tester only, through `#human`/`#thread`, the same owner check as the thread verbs (Q7: single human, no shares). They go into `TESTER_VERBS`. - Other readers. `#grantNotice`, `#raisedThread` and the decision-close path (:736, :1500, :2941) @@ -148,6 +150,41 @@ Line numbers are at da9555ca2 unless they say otherwise. packages/cli/tests/stack-release.test.mjs:860-934. Also `timedTables` (store.mjs:65-79) and the broker clock-seed list (:325-336). +### Duckworth's design checks (duckworth-0106) + +Duckworth had no objection to the in-place rebuild and asked for seven checks. Where each one is: + +1. The cross-table trigger. 3j-3k.sql drops `delivery_same_thread` before any rename and creates + `delivery_same_session` after the new tables. +2. Index names. The step drops `thread_claims_thread` and `message_threads_thread` before the + renames, so the new tables can reuse the names. +3. Per-table triggers. The step drops each guard and append-only trigger and creates it again with + text that matches schema.sql. The digest check at store.mjs:231 enforces that, and the T7 + `sqlite_master` comparison shows it. +4. Generation counts claim rows only. T7's fixture is Duckworth's: claim, release, claim, revoke, + giving 1, 1, 2, 2, with t2's claim between them. +5. Sequences. Rows keep their seq values, `sqlite_sequence` carries over, and T7 checks that the + next insert is 41 after 40. +6. Id collisions. A legacy session id is its thread's id. `sessions_id_not_thread` refuses a session + id that names another thread, and `threads_id_not_session` refuses a thread id that names a + session. New ids come from randomUUID. +7. `thread_one_per_run`. The inner `max(seq)` is per session. + +### Dry run + +Before writing any node test, I ran 3j-3k.sql in an in-memory Python sqlite3 on a 3j fixture +(gizmo-264-out/dry-3k.py, no repo code). The claims were kept with generations 1, 1, 1, 2, 2 in seq +order for the mixed t1/t2 fixture. Seq 40 carried over and the next insert was 41. The migrated +`sqlite_master` matched a fresh schema.sql, 119 rows each. Every guard refused its bad write. + +### Hand-made threads in tests + +A thread that a test writes by hand has no default session, because only `thread.create` and the +migration write one. The broker refuses such a thread with `session-not-found`, and the launcher view +leaves it out. So every test that inserts a `threads` row now also writes the legacy-shape default +session (`DEFAULT_SESSION` in threads, end-launch, thread-cli, thread-launch and chat tests). That +mirrors the migration's rows rather than inventing a third shape. + ## Runs None yet. Every run waits for Sage's CLEAR. diff --git a/packages/bus/README.md b/packages/bus/README.md index bfcc1c5d7..4732aa45f 100644 --- a/packages/bus/README.md +++ b/packages/bus/README.md @@ -236,9 +236,11 @@ a receipt with empty `changes`. A refusal is `action.refused` with its code alon | `project.list` | No args; human sessions only. The business's projects, oldest first. | | `thread.create` | `{project,name}`; human sessions only. The thread belongs to the caller and to the business's one thread instance (Projects and threads, below). Returns `{id,instance}`. Refuses `project-not-found`, `thread-exists`, `thread-instance-missing` or `thread-instance-ambiguous`. | | `thread.list` | `{project?}`; human sessions only. The caller's own threads, oldest first. | -| `thread.open` | `{thread}`; human sessions only. Records a `thread_opens` row, which asks the launcher to start the thread's session. Returns `{opened}`. | -| `thread.send` | `{thread,body,class?}`; human sessions only. A message to the thread's instance, in that thread, with its `human.input` event as `request`. Returns `{id,request}`. | -| `thread.messages` | `{thread,after?,limit?}`; human sessions only. The thread's messages oldest first, after the message `after` names (`message-not-found` when it isn't in the thread), at most `limit` (default 50, 1 to 200). Writes nothing. | +| `thread.open` | `{thread,session?}`; human sessions only. Records a `thread_opens` row, which asks the launcher to start the thread's session. Returns `{opened}`. | +| `thread.send` | `{thread,session?,body,class?}`; human sessions only. A message to the thread's instance, in that thread, with its `human.input` event as `request`. Returns `{id,request}`. | +| `thread.messages` | `{thread,session?,after?,limit?}`; human sessions only. The session's messages oldest first, after the message `after` names (`message-not-found` when it isn't in the session), at most `limit` (default 50, 1 to 200). Writes nothing. | +| `session.list` | `{thread}`; human and tester sessions (row 264a, #1769). The thread's sessions oldest first, `[{id,at,name,predecessor,default}]`. Writes nothing. | +| `session.create` | `{thread,name,op}`; human and tester sessions. A new session of the thread, metadata only. A retry with the same `op` and name returns the first row; another name under that `op` refuses `session-op-conflict`, and a taken name `session-exists`. Returns `{id,thread,name}`. | ### Projects and threads (row 121, #1615) @@ -306,6 +308,33 @@ the threads they made. A revoked run can't claim its thread again (`run-revoked` thread run can still be stopped another way: the launcher stops it once the thread is idle, the human stops it with `mosaic stop `, and the run's end releases its claim. `thread_revoke_needs_decision` guards the table against a direct write. +### Sessions (row 264a, #1769) + +Schema `3k` puts sessions below a thread. `sessions` and `session_defaults` are new and +append-only. `thread_claims`, `message_threads` and `thread_opens` each gain a `session` +column, and `thread_claims` gains `generation`, the claim's number in its session counting +from 1. A release or revoke carries its claim's number. + +Every thread has a default session, the latest `session_defaults` row (D3). `thread.create` +writes one with a random id. The 3j→3k step gives each existing thread a default session whose +id is the thread's own, named `default` with op `legacy-3k`, and files every old claim, message +and open under it. A thread verb, or a `bindLaunch` record, that names no `session` works the +default, so a client from before 3k keeps working. A `session` from another thread refuses +`session-not-found`, and a non-thread role that names one refuses `thread-not-allowed`. + +The thread rules above hold per session. One run holds a session at a time, and a run holds at +most one session in its business, while sibling sessions of a thread may each have a holder. +A thread run receives and replies in its own session, and the chief of staff's reply returns +to the session that sent the work. `threadsState` and `threadIdle` keep one row per thread, +computed on its default session; 264b's launcher starts the others. + +The triggers check that a session has its thread's business, instance and owner, that its +predecessor and any default, claim, message or open name a session of the same thread, and that +a claim's generation is the one its session expects. `delivery_same_session` replaces +`delivery_same_thread`. `sessions_id_not_thread` and `threads_id_not_session` keep a session id +from naming another thread and a thread id from naming a session. A 3j engine refuses a 3k +store with `schema-mismatch`. + ### Agent state (row 59, #1548) For a human or reader session, `agents` derives each held role's state from @@ -711,10 +740,10 @@ scope probes and rotation/expiry notifications; S2 never probes real services. ## Schema and verification -`schema.sql` is v3j, SHA-256 -`0a7b633ce7eec9d0bb37c3255f4244d97bd0c8eac38db91a11b4ef6cc982ec8c`. -It has 18 tables, 72 triggers and five views. Source digest and a trusted full -sqlite_master digest are checked at open, alongside schema version `3j`, WAL, +`schema.sql` is v3k, SHA-256 +`860374bde3f063ecfd72618913e05924587b7c39b87b763eee4c6aeb59922f46`. +It has 20 tables, 87 triggers and five views. Source digest and a trusted full +sqlite_master digest are checked at open, alongside schema version `3k`, WAL, quick_check and foreign-key integrity. Missing metadata refuses; it never reinitializes an existing file. Full synchronous WAL transactions either commit or roll back. SQL UPDATE, DELETE and INSERT OR REPLACE refuse on evidence tables. diff --git a/packages/bus/schema.sql b/packages/bus/schema.sql index fe664c8e1..b63e79405 100644 --- a/packages/bus/schema.sql +++ b/packages/bus/schema.sql @@ -248,11 +248,34 @@ CREATE TABLE threads ( 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, @@ -261,55 +284,95 @@ CREATE TABLE thread_claims ( ) STRICT; CREATE TABLE message_threads ( message TEXT PRIMARY KEY REFERENCES messages(id), - thread TEXT NOT NULL REFERENCES threads(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 thread = NEW.thread ORDER BY seq DESC LIMIT 1) = 'claim' - BEGIN SELECT RAISE(ABORT, 'thread already held'); END; + 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 thread = c.thread) AND c.op = 'claim') - BEGIN SELECT RAISE(ABORT, 'run already holds a thread'); END; + 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 thread = NEW.thread ORDER BY seq DESC LIMIT 1),'') <> 'claim' - BEGIN SELECT RAISE(ABORT, 'thread is not held'); END; + 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 thread = NEW.thread ORDER BY seq DESC LIMIT 1) <> NEW.holder_run + 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 delivery_same_thread BEFORE INSERT ON deliveries +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 thread = t.id) + 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 thread''s holder'); END; + 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; diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index f01ca0918..14e2a90ba 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -116,7 +116,9 @@ export const TASK_VERBS = Object.freeze(ACTIONS.filter((a) => a.startsWith('task // Row 177 (#1675, Q173): the tester's own threads, owner `tester`, and a stop of their runs. export const TESTER_VERBS = Object.freeze(['message.send', 'message.receive', 'message.read', 'inbox', 'agents', 'tasks', 'trail', 'decisions', 'messages', 'replies', 'usage', 'names', 'limits', 'plugins', 'instances', 'project.list', 'thread.list', 'name.set', 'project.create', 'credentials', - 'thread.create', 'thread.open', 'thread.send', 'thread.messages', 'assistant.stop']); + 'thread.create', 'thread.open', 'thread.send', 'thread.messages', 'assistant.stop', + // Row 264a (#1769): the sessions of the tester's own threads. + 'session.list', 'session.create']); // Row 175: the organization verbs a tester may call. Its writes run every check the human's do // and log actor `tester`; row 176 adds routes reads; limits.set, organization.set and organization.revision refuse it. export const TESTER_ORGANIZATION_VERBS = Object.freeze(['limits', 'plugins', 'instances', 'plugins.set', 'instances.set', 'routes.set', 'routes']); @@ -329,6 +331,8 @@ export class Broker { 'task_snapshots', 'projects', 'threads', + 'sessions', + 'session_defaults', 'thread_claims', 'thread_opens', ]) { @@ -389,37 +393,49 @@ export class Broker { #threaded(business, role) { return this.#business(business).roles[role]?.definition === THREAD_DEFINITION; } - #threadHolder(thread) { - return this.#store.get('SELECT * FROM thread_claims WHERE thread=? ORDER BY seq DESC LIMIT 1', thread); + // Row 264a (#1769): a claim holds one session of a thread, and each session has its own holder. + #sessionHolder(session) { + return this.#store.get('SELECT * FROM thread_claims WHERE session=? ORDER BY seq DESC LIMIT 1', session); } - // Row 155 (#1649): the held claims of every thread of a thread instance, oldest thread first, or + // D3: the thread's default session, the latest session_defaults row. thread.create and the 3j-3k + // migration write one for every thread. + #defaultSession(thread) { + return this.#store.get('SELECT session FROM session_defaults WHERE thread=? ORDER BY seq DESC LIMIT 1', thread)?.session ?? fail('session-not-found'); + } + // The session a thread verb names, or the thread's default when it names none. + #threadSession(t, session) { + if (session === undefined) return this.#defaultSession(t.id); + if (!this.#store.get('SELECT 1 FROM sessions WHERE id=? AND thread=?', identifier(session), t.id)) fail('session-not-found'); + return session; + } + // 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) { return this.#store.all( - "SELECT c.* FROM threads t JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE thread=t.id) WHERE t.business=? AND t.instance=? AND (? IS NULL OR t.owner=?) AND c.op='claim' ORDER BY t.seq", + "SELECT c.* FROM threads t JOIN sessions se ON se.thread=t.id JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE session=se.id) WHERE t.business=? AND t.instance=? AND (? IS NULL OR t.owner=?) AND c.op='claim' ORDER BY t.seq, se.seq", business, instance, owner, owner, ); } - // The claim a run acts under: its thread's for a thread run, its role's otherwise. + // The claim a run acts under: its session's for a thread run, its role's otherwise. #own(s) { - return s.thread ? this.#threadHolder(s.thread) : this.#holder(s.business, s.role); + return s.thread ? this.#sessionHolder(s.session) : this.#holder(s.business, s.role); } - // Who a message to `to` goes to now: the thread's holder when the message is in a thread of - // `to`, the role's otherwise. Null while nobody holds it, and the message waits. - #recipient(business, to, thread) { - const h = thread && this.#threaded(business, to) ? this.#threadHolder(thread) : this.#holder(business, to); + // Who a message to `to` goes to now: the session's holder when the message is in a session of a + // thread of `to`, the role's otherwise. Null while nobody holds it, and the message waits. + #recipient(business, to, session) { + const h = session && this.#threaded(business, to) ? this.#sessionHolder(session) : this.#holder(business, to); return h?.op === 'claim' ? h : null; } - // The thread a new message belongs to (REQ-MSG-1). A thread run's message is in its thread. - // A message to a thread instance from anyone else is in the thread of the nearest message up - // its in_reply_to chain that has one, so the chief of staff's reply returns to the thread + // The session a new message belongs to (REQ-MSG-1). A thread run's message is in its session. + // A message to a thread instance from anyone else is in the session of the nearest message up + // its in_reply_to chain that has one, so the chief of staff's reply returns to the session // that sent the work. Messages are append-only and in_reply_to names an earlier message, so // the chain has an end; `seen` stops the walk there and needs no depth constant. - #threadFor(s, to, inReplyTo) { - if (s.thread) return s.thread; + #sessionFor(s, to, inReplyTo) { + if (s.thread) return s.session; if (!this.#threaded(s.business, to)) return null; // The user names the thread with thread.send, which checks the owner (REQ-USER-2). if (s.human) fail('thread-required-use-thread-send'); @@ -427,18 +443,20 @@ export class Broker { for (let at = inReplyTo ?? null; at !== null && !seen.has(at); ) { seen.add(at); const m = this.#store.get( - 'SELECT m.in_reply_to, m.from_role, m.body, t.thread FROM messages m LEFT JOIN message_threads t ON t.message=m.id WHERE m.id=? AND m.business=?', + 'SELECT m.in_reply_to, m.from_role, m.body, t.thread, t.session FROM messages m LEFT JOIN message_threads t ON t.message=m.id WHERE m.id=? AND m.business=?', at, s.business, ); if (m?.thread) { if (this.#store.get('SELECT instance FROM threads WHERE id=?', m.thread).instance !== to) fail('thread-instance-mismatch'); - return this.#testerOwns(s, m.thread); + this.#testerOwns(s, m.thread); + return m.session; } - // Row 143: a chain that reaches a spec notice belongs to the user's thread in that notice's project. + // Row 143: a chain that reaches a spec notice belongs to the user's thread in that notice's + // project, in its default session. if (m?.from_role === NOTICE_ROLE && m.body.startsWith(SPEC_NOTICE)) { const spec = this.#specThread(s, to, m.body); - if (spec) return this.#testerOwns(s, spec); + if (spec) return this.#defaultSession(this.#testerOwns(s, spec)); } at = m?.in_reply_to ?? null; } @@ -500,26 +518,30 @@ export class Broker { const key = record.business + ':' + identifier(record.run); const session = this.#runs.get(key) ?? fail('unknown-run'); if (this.#ended.has(key)) fail('run-ended'); - // A run holds at most one thread, so this also refuses another thread and a non-thread run. + // A run holds at most one session, so this also refuses another thread and a non-thread run. + // Row 264a: the row is the thread's default session's, so a run of another session is refused too. const [t] = this.#threadRows('t.business=? AND t.id=?', record.business, identifier(record.thread)); if (t?.holder !== record.run) fail('not-holder'); if (t.pending || t.unread) return { released: false, pending: t.pending, unread: t.unread }; - this.#store.transaction(() => this.#claimEnd(session, this.#threadHolder(t.thread), 'release', null)); + this.#store.transaction(() => this.#claimEnd(session, this.#own(session), 'release', null)); return { released: true }; } + // Row 264a (#1769): each row is the thread's default session's. Until 264b the launcher starts + // default sessions only, so another session's messages wait. #threadRows(where, ...params) { return this.#store .all( `SELECT t.id, t.instance, - (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.thread=t.id AND m.to_role=t.instance + (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=ds.session 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, - (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.thread=t.id AND m.to_role=t.instance + (SELECT count(*) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=ds.session 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, - (SELECT max(m.at) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.thread=t.id) AS last_message, - (SELECT max(at) FROM thread_opens WHERE thread=t.id) AS last_open, + (SELECT max(m.at) FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=ds.session) AS last_message, + (SELECT max(at) FROM thread_opens WHERE session=ds.session) AS last_open, c.at AS claim_at, c.op AS claim_op, c.holder_run AS holder_run - FROM threads t LEFT JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE thread=t.id) + FROM threads t JOIN session_defaults ds ON ds.seq=(SELECT max(seq) FROM session_defaults WHERE thread=t.id) + LEFT JOIN thread_claims c ON c.seq=(SELECT max(seq) FROM thread_claims WHERE session=ds.session) WHERE ${where} ORDER BY t.seq`, ...params, ) @@ -538,7 +560,7 @@ export class Broker { bindLaunch(record) { keys( record, - ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime', 'thread', 'provider', 'model', 'account', 'transcript'], + ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime', 'thread', 'session', 'provider', 'model', 'account', 'transcript'], ['business', 'role', 'run', 'harness'], ); const b = this.#business(identifier(record.business)); @@ -561,10 +583,13 @@ export class Broker { if (record.thread === undefined) fail('thread-required'); if (!this.#store.get('SELECT 1 FROM threads WHERE id=? AND business=? AND instance=?', identifier(record.thread), record.business, record.role)) fail('thread-not-found'); - } else if (record.thread !== undefined) fail('thread-not-allowed'); + } else if (record.thread !== undefined || record.session !== undefined) fail('thread-not-allowed'); + // Row 264a (#1769): a thread run works one session of its thread, the thread's default when the + // record names none (D3). + const bound = record.thread === undefined ? null : this.#threadSession({ id: record.thread }, record.session); const key = record.business + ':' + record.run; if (this.#runs.has(key)) fail('duplicate-run'); - const session = { ...record, address: record.address ?? null, thread: record.thread ?? null, human: false }; + const session = { ...record, address: record.address ?? null, thread: record.thread ?? null, session: bound, human: false }; this.#secretCheck(record); const old = this.#store.get( "SELECT actor_role,body FROM events WHERE business=? AND kind='session.launched' AND subject=? ORDER BY seq LIMIT 1", @@ -578,6 +603,8 @@ export class Broker { pid: record.pid ?? null, startTime: record.startTime ?? null, ...(record.thread === undefined ? {} : { thread: record.thread }), + // A record without a session keeps the body it had before row 264a. + ...(record.session === undefined ? {} : { session: record.session }), }; // A run an earlier broker bound is bound again unchanged (the host's recover()), even after its // harness was switched off, so it can be ended. @@ -733,14 +760,14 @@ export class Broker { // close is. #grantNotice(business, decision, body) { const d = this.#store.get('SELECT raised_by_role,raised_by_run FROM decisions WHERE id=?', decision); - const thread = this.#threaded(business, d.raised_by_role) + const session = this.#threaded(business, d.raised_by_role) ? (this.#store.get( - "SELECT thread FROM thread_claims WHERE business=? AND holder_run=? AND op='claim' ORDER BY seq DESC LIMIT 1", + "SELECT session FROM thread_claims WHERE business=? AND holder_run=? AND op='claim' ORDER BY seq DESC LIMIT 1", business, d.raised_by_run, - )?.thread ?? null) + )?.session ?? null) : null; - for (const [to, t] of [['human', null], [d.raised_by_role, thread]]) { + for (const [to, t] of [['human', null], [d.raised_by_role, session]]) { const mid = randomUUID(); this.#store.run( 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,in_reply_to,decision,corrects,body) VALUES(?,?,?,?,?,?,?,?,?,?,?)', @@ -1958,7 +1985,7 @@ export class Broker { null, body, ); - this.#route(s.business, mid, s.role, s.thread); + this.#route(s.business, mid, s.role, s.session); } // Trusted launcher API: whether the role, in the thread (null for a role run), holds a merge // decision a landing could use: cross-role, raised there with a task ref as its target, resolved @@ -2647,22 +2674,25 @@ export class Broker { keys(a, []); if (s.human) fail('agent-required'); // Row 121: a thread run claims its thread, so idle threads hold nothing (REQ-THREAD-2). + // Row 264a (#1769): the claim is the run's session's, with the session's next generation. if (s.thread) { - if (this.#store.get("SELECT 1 FROM thread_claims WHERE thread=? AND holder_run=? AND op='revoke' LIMIT 1", s.thread, s.run)) + 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'); - if (this.#threadHolder(s.thread)?.op === 'claim') fail('thread-already-held'); + if (this.#sessionHolder(s.session)?.op === 'claim') fail('thread-already-held'); this.#store.run( - 'INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,address,by) VALUES(?,?,?,?,?,?,?,?)', + "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'),?,?,?,?,?)", this.#now(), s.business, s.thread, + s.session, + s.session, 'claim', s.run, s.harness, s.address, by, ); - this.#routeThread(s.thread); + this.#routeSession(s.session); return { role: s.role, run: s.run, thread: s.thread }; } if ( @@ -2711,7 +2741,7 @@ export class Broker { identifier(a.decision); this.#consumeAuthority(s, 'role.revoke', { decision: a.decision, target: a.role }); } - // Row 155 (#1649): a thread instance holds a claim per thread, so the end revokes each one. + // Row 155 (#1649): a thread instance holds a claim per session, so the end revokes each one. if (this.#threaded(s.business, a.role)) { const held = this.#threadHolders(s.business, a.role); if (held.length === 0) fail('not-held'); @@ -2934,14 +2964,14 @@ export class Broker { op === 'resolved' ? `resolved by ${by}: ${a.choice} (${JSON.parse(d.options).find((o) => o.key === a.choice).text})` : `withdrawn by ${by}`; - // A close to a thread instance goes to the thread the raising run held. A decision + // A close to a thread instance goes to the session the raising run held. A decision // raised before row 121 has no thread claim and stays unthreaded. - const thread = this.#threaded(s.business, d.raised_by_role) + const session = this.#threaded(s.business, d.raised_by_role) ? (this.#store.get( - "SELECT thread FROM thread_claims WHERE business=? AND holder_run=? AND op='claim' ORDER BY seq DESC LIMIT 1", + "SELECT session FROM thread_claims WHERE business=? AND holder_run=? AND op='claim' ORDER BY seq DESC LIMIT 1", s.business, d.raised_by_run, - )?.thread ?? null) + )?.session ?? null) : null; const mid = randomUUID(); this.#store.run( @@ -2958,7 +2988,7 @@ export class Broker { null, `Decision ${d.id} on ${d.action} was ${outcome}.${a.note === undefined ? '' : `\nNote: ${a.note}`}`, ); - this.#route(s.business, mid, d.raised_by_role, thread); + this.#route(s.business, mid, d.raised_by_role, session); } return grant ? { id: d.id, op, grant: 'started' } : { id: d.id, op }; } @@ -3025,7 +3055,7 @@ export class Broker { } if (a.class === 'BLOCK') this.#checkBlock(s, to, a.in_reply_to); if (a.decision) this.#decision(s, a.decision); - const thread = this.#threadFor(s, to, a.in_reply_to); + const thread = this.#sessionFor(s, to, a.in_reply_to); // Row 170 (#1667): a tester's message to a role nobody holds is refused before it is // stored, so a PM test never waits for the role's next run. A run launched but not yet // claimed counts as no holder, as in #waiting. No lock: a holder that releases once the @@ -3062,14 +3092,14 @@ export class Broker { this.#received.add(s.business + ':' + s.run); } const role = s.human ? 'human' : s.role; - // A thread run receives its own thread's messages only (REQ-THREAD-1). + // 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'))" + - (s.thread ? ' AND EXISTS(SELECT 1 FROM message_threads mt WHERE mt.message=m.id AND mt.thread=?)' : '') + + (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, role, - ...(s.thread ? [s.thread] : []), + ...(s.thread ? [s.session] : []), ); for (const m of rows) this.#delivery( @@ -3481,6 +3511,8 @@ export class Broker { this.#owner(s), a.name, ); + // Row 264a (#1769, D3): every thread has a default session from the start. + this.#newSession(s, { id: tid, business: s.business, instance: instances[0], owner: this.#owner(s) }, 'default', 'default', true); return { id: tid, instance: instances[0] }; } // Row 177 (#1675): each lists its own threads. The human's list leaves out the tester's, @@ -3498,20 +3530,23 @@ export class Broker { ...(a.project === undefined ? [] : [a.project]), ); } - // Opening a thread asks the launcher to start its session (REQ-LAUNCH-1). + // Opening a thread asks the launcher to start its session (REQ-LAUNCH-1). Row 264a: a thread + // verb names one session of the thread, or works on its default (D3). case 'thread.open': { if (!s.tester) this.#human(s); - keys(a, ['thread'], ['thread']); + keys(a, ['thread', 'session'], ['thread']); const t = this.#thread(s, a.thread); - this.#store.run('INSERT INTO thread_opens(at,thread,by) VALUES(?,?,?)', this.#now(), t.id, this.#owner(s)); + const session = this.#threadSession(t, a.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 }; } // Row 177 (#1675): a tester's message is the tester's, from tester-cli, and writes no // human.input, as on message.send. In its own thread the lists read it as the human. case 'thread.send': { if (!s.tester) this.#human(s); - keys(a, ['thread', 'body', 'class'], ['thread', 'body']); + keys(a, ['thread', 'session', 'body', 'class'], ['thread', 'body']); const t = this.#thread(s, a.thread); + const session = this.#threadSession(t, a.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); @@ -3531,31 +3566,60 @@ export class Broker { null, a.body, ); - this.#route(s.business, mid, t.instance, t.id); + this.#route(s.business, mid, t.instance, session); return { id: mid, request: s.tester ? null : this.#event(s, 'human.input', { kind: 'instruction', message: mid }, mid) }; } // The thread's history, oldest first. Reading writes nothing. case 'thread.messages': { if (!s.tester) this.#human(s); - keys(a, ['thread', 'after', 'limit'], ['thread']); + keys(a, ['thread', 'session', 'after', 'limit'], ['thread']); const t = this.#thread(s, a.thread); + const session = this.#threadSession(t, a.session); const limit = a.limit ?? THREAD_PAGE; if (!Number.isInteger(limit) || limit < 1 || limit > THREAD_PAGE_MAX) fail('invalid-request'); let after = 0; if (a.after !== undefined) { after = this.#store.get( - 'SELECT m.seq FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE m.id=? AND mt.thread=?', + 'SELECT m.seq FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE m.id=? AND mt.session=?', identifier(a.after), - t.id, + session, )?.seq; if (after === undefined) fail('message-not-found'); } return this.#store.all( - `SELECT m.* FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.thread=? AND m.seq>? ORDER BY m.seq LIMIT ${limit}`, - t.id, + `SELECT m.* FROM messages m JOIN message_threads mt ON mt.message=m.id WHERE mt.session=? AND m.seq>? ORDER BY m.seq LIMIT ${limit}`, + session, after, ); } + // Row 264a (#1769, T1 item 7): a thread's sessions, oldest first, and which one is the default. + // Each user reaches only their own threads' sessions (REQ-USER-2); there are no shares (Q7). + case 'session.list': { + if (!s.tester) this.#human(s); + keys(a, ['thread'], ['thread']); + const t = this.#thread(s, a.thread); + const fallback = this.#defaultSession(t.id); + return this.#store + .all('SELECT id,at,name,predecessor FROM sessions WHERE thread=? ORDER BY seq', t.id) + .map((r) => ({ ...r, default: r.id === fallback })); + } + // 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 + // under the same op refuses. + case 'session.create': { + if (!s.tester) this.#human(s); + keys(a, ['thread', 'name', 'op'], ['thread', 'name', 'op']); + if (!name(a.name)) fail('invalid-request'); + identifier(a.op); + const t = this.#thread(s, a.thread); + 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'); + return { id: this.#newSession(s, t, a.name, a.op, false), thread: t.id, name: a.name }; + } case 'replies': { if (!s.human && !s.reader) fail('human-required'); keys(a, ['id'], ['id']); @@ -3611,10 +3675,12 @@ export class Broker { #claimEnd(s, h, op, decision, reason = null, by = s.run ?? s.human) { if (h.thread) return this.#store.run( - 'INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,address,by,decision,reason) VALUES(?,?,?,?,?,?,?,?,?,?)', + 'INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,address,by,decision,reason) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)', this.#now(), s.business, h.thread, + h.session, + h.generation, op, h.holder_run, h.harness, @@ -3657,7 +3723,7 @@ export class Broker { } // The message a review verb sends, routed like message.send's. #reviewMessage(s, mid, { to, cls, in_reply_to = null, decision = null, body }) { - const thread = this.#threadFor(s, to, in_reply_to); + const thread = this.#sessionFor(s, to, in_reply_to); this.#store.run( 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,in_reply_to,decision,corrects,body) VALUES(?,?,?,?,?,?,?,?,?,?,?)', mid, @@ -3675,9 +3741,35 @@ export class Broker { this.#route(s.business, mid, to, thread); } // Row 121: a message names its thread (REQ-MSG-1) and goes to the holder that may have it. - #route(business, mid, to, thread) { - if (thread) this.#store.run('INSERT INTO message_threads(message,thread) VALUES(?,?)', mid, thread); - this.#delivery(mid, 'routed', this.#recipient(business, to, thread)); + // Row 264a (#1769): it names one session of the thread, and the session's holder may have it. + #route(business, mid, to, session) { + if (session) { + const { thread } = this.#store.get('SELECT thread FROM sessions WHERE id=?', session); + this.#store.run('INSERT INTO message_threads(message,thread,session) VALUES(?,?,?)', mid, thread, session); + } + this.#delivery(mid, 'routed', this.#recipient(business, to, session)); + } + // Row 264a (#1769): a session row and, for a new thread, its default. The 3j-3k migration made the + // legacy defaults, whose id is the thread id; a new session's id is random, and a trigger refuses + // one that names another thread. + #newSession(s, t, sessionName, op, makeDefault) { + const sid = randomUUID(); + const at = this.#now(); + this.#store.run( + 'INSERT INTO sessions(id,at,business,thread,instance,owner,name,op,predecessor) VALUES(?,?,?,?,?,?,?,?,?)', + sid, + at, + t.business, + t.id, + t.instance, + t.owner, + sessionName, + op, + null, + ); + if (makeDefault) + this.#store.run('INSERT INTO session_defaults(at,thread,session,by) VALUES(?,?,?,?)', at, t.id, sid, this.#owner(s)); + return sid; } // Row 146 (#1640): a message.send that nobody holds the target for waits for a claim. Null when // the message was routed to a holder, or is the human's, a tester's, or in a thread to a thread @@ -3804,12 +3896,12 @@ export class Broker { ); for (const m of rows) this.#delivery(m.id, 'routed', h); } - // A new thread holder gets the thread's waiting messages to the instance. - #routeThread(thread) { - const h = this.#threadHolder(thread); + // A new session holder gets the session's waiting messages to the instance. + #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.thread=? AND m.to_role=t.instance AND NOT EXISTS(SELECT 1 FROM deliveries WHERE message=m.id AND op IN ('delivered','read'))", - thread, + "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'))", + session, ); for (const m of rows) this.#delivery(m.id, 'routed', h); } diff --git a/packages/bus/src/migrations/3j-3k.sql b/packages/bus/src/migrations/3j-3k.sql new file mode 100644 index 000000000..24fec14b7 --- /dev/null +++ b/packages/bus/src/migrations/3j-3k.sql @@ -0,0 +1,168 @@ +-- Row 264a (#1769, T1 item 7): sessions below a thread. Each existing thread gets one legacy default +-- session whose id is the thread id, and its claims, messages and opens name that session (D9's +-- mapping). thread_claims, message_threads and thread_opens are rebuilt with a session column, and +-- thread_claims gains generation, the claim's number in its session. Rows are copied with their seq +-- values and old columns as they were, and the sequences carry over, as 3h-3i did. The guards and the +-- delivery trigger that read the old tables are dropped first, so the renames rewrite nothing. +DROP TRIGGER thread_one_holder; +DROP TRIGGER thread_one_per_run; +DROP TRIGGER thread_release_by_holder; +DROP TRIGGER thread_release_same_run; +DROP TRIGGER thread_revoke_needs_decision; +DROP TRIGGER thread_claims_business; +DROP TRIGGER message_threads_match; +DROP TRIGGER delivery_same_thread; +DROP TRIGGER thread_claims_no_update; +DROP TRIGGER thread_claims_no_delete; +DROP TRIGGER thread_claims_no_replace; +DROP TRIGGER message_threads_no_update; +DROP TRIGGER message_threads_no_delete; +DROP TRIGGER message_threads_no_replace; +DROP TRIGGER thread_opens_no_update; +DROP TRIGGER thread_opens_no_delete; +DROP TRIGGER thread_opens_no_replace; +DROP INDEX thread_claims_thread; +DROP INDEX message_threads_thread; +ALTER TABLE thread_claims RENAME TO thread_claims_3j; +ALTER TABLE message_threads RENAME TO message_threads_3j; +ALTER TABLE thread_opens RENAME TO thread_opens_3j; +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; +INSERT INTO sessions(at,business,thread,id,instance,owner,name,op,predecessor) + SELECT at,business,id,id,instance,owner,'default','legacy-3k',NULL FROM threads ORDER BY seq; +INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3k' FROM threads ORDER BY seq; +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; +INSERT INTO thread_claims(seq,at,business,thread,session,generation,op,holder_run,harness,address,by,reason,decision) + SELECT c.seq,c.at,c.business,c.thread,c.thread, + (SELECT count(*) FROM thread_claims_3j p WHERE p.thread = c.thread AND p.op = 'claim' AND p.seq <= c.seq), + c.op,c.holder_run,c.harness,c.address,c.by,c.reason,c.decision FROM thread_claims_3j c ORDER BY c.seq; +INSERT INTO message_threads(message,thread,session) SELECT message,thread,thread FROM message_threads_3j ORDER BY rowid; +INSERT INTO thread_opens(seq,at,thread,session,by) SELECT seq,at,thread,thread,by FROM thread_opens_3j ORDER BY seq; +DELETE FROM sqlite_sequence WHERE name IN ('thread_claims','thread_opens'); +INSERT INTO sqlite_sequence(name,seq) SELECT 'thread_claims', seq FROM sqlite_sequence WHERE name='thread_claims_3j'; +INSERT INTO sqlite_sequence(name,seq) SELECT 'thread_opens', seq FROM sqlite_sequence WHERE name='thread_opens_3j'; +DROP TABLE thread_claims_3j; +DROP TABLE message_threads_3j; +DROP TABLE thread_opens_3j; +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 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 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; diff --git a/packages/bus/src/store.mjs b/packages/bus/src/store.mjs index 4e7280eec..b8189cf78 100644 --- a/packages/bus/src/store.mjs +++ b/packages/bus/src/store.mjs @@ -13,8 +13,8 @@ import { join, isAbsolute } from 'node:path'; import { createHash } from 'node:crypto'; const ddl = readFileSync(new URL('../schema.sql', import.meta.url), 'utf8'); -export const SCHEMA_SHA256 = '0a7b633ce7eec9d0bb37c3255f4244d97bd0c8eac38db91a11b4ef6cc982ec8c'; -export const SCHEMA_VERSION = '3j'; +export const SCHEMA_SHA256 = '860374bde3f063ecfd72618913e05924587b7c39b87b763eee4c6aeb59922f46'; +export const SCHEMA_VERSION = '3k'; // Ordered, named offline steps. New schema changes append a step and keep prior DDL pinned. const legacyDDL = readFileSync(new URL('./migrations/3b.sql', import.meta.url), 'utf8'); const LEGACY_SHA256 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65'; @@ -61,6 +61,12 @@ const migrations = [{ from: '3i', to: '3j', sql: readFileSync(new URL('./migrations/3i-3j.sql', import.meta.url), 'utf8'), +}, { + // Row 264a (#1769): sessions below a thread. Each thread gets a default session whose id is the + // thread id, and thread_claims, message_threads and thread_opens are rebuilt to name it. + from: '3j', + to: '3k', + sql: readFileSync(new URL('./migrations/3j-3k.sql', import.meta.url), 'utf8'), }]; const timedTables = [ 'events', @@ -72,6 +78,8 @@ const timedTables = [ 'task_snapshots', 'projects', 'threads', + 'sessions', + 'session_defaults', 'thread_claims', 'thread_opens', 'instance_names', diff --git a/packages/bus/tests/account-migration.test.mjs b/packages/bus/tests/account-migration.test.mjs index 4cd14db44..3b9fa3f59 100644 --- a/packages/bus/tests/account-migration.test.mjs +++ b/packages/bus/tests/account-migration.test.mjs @@ -15,6 +15,9 @@ const AT = '2026-10-18T00:00:00.000Z'; const SEQUENCE = 50; const ddl3h = readFileSync(new URL('./fixtures/schema-3h.sql', import.meta.url), 'utf8'); const here = (p) => new URL(p, import.meta.url); +// Row 264a (#1769): what the 3j→3k step rebuilds or adds. sessions.test.mjs tests those objects. +const SESSIONS_3K = ['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens']; +const later3k = (r) => SESSIONS_3K.includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session'].includes(r.name); function source3h(t) { const root = mkdtempSync(join(tmpdir(), 'account-migrate-')); t.after(() => rmSync(root, { recursive: true, force: true })); @@ -70,6 +73,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { copyFileSync(here('../src/migrations/' + file), join(moduleRoot, 'src/migrations', file)); writeFileSync(join(moduleRoot, 'src/migrations/3h-3i.sql'), step(readFileSync(here('../src/migrations/3h-3i.sql'), 'utf8'))); copyFileSync(here('../src/migrations/3i-3j.sql'), join(moduleRoot, 'src/migrations/3i-3j.sql')); + copyFileSync(here('../src/migrations/3j-3k.sql'), join(moduleRoot, 'src/migrations/3j-3k.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } @@ -77,7 +81,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), '3j'); + assert.equal(version(path), '3k'); // Every table's rows as they were, usage_records with its seq values and sqlite_sequence with 50. assert.deepEqual(snapshot(path, before.columns).rows, before.rows); assert.deepEqual(before.rows.usage_records.map(r => r.seq), [1, 2, 3]); @@ -87,7 +91,7 @@ test('REQ-LIMIT-4 3h copy migrates to 3i: every row keeps its seq, the sequence assert.equal(was.filter(r => r.sql?.includes(old)).map(r => r.name).join(), 'usage_records'); // Row 234 (#1740) adds 3i→3j's own objects after this step. Its test is grant-migration.test.mjs. const later = ['hosts', 'host_grants', 'events_grant_actor', 'sqlite_autoindex_hosts_1', 'sqlite_autoindex_host_grants_1']; - assert.deepEqual(master(path).filter(r => !later.includes(r.name)), was.map(r => r.name === 'usage_records' ? { ...r, sql: r.sql.replace(old, ' account TEXT,\n') } : r)); + assert.deepEqual(master(path).filter(r => !later.includes(r.name) && !later3k(r)), was.filter(r => !later3k(r)).map(r => r.name === 'usage_records' ? { ...r, sql: r.sql.replace(old, ' account TEXT,\n') } : r)); assert.equal(was.filter(r => r.type === 'trigger' && r.tbl_name === 'usage_records').length, 3); const store = new Store(root); try { @@ -125,7 +129,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), '3j'); + assert.equal(version(path), '3k'); assert.deepEqual(snapshot(path, before.columns).rows, before.rows); }); @@ -137,14 +141,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 = '3j';", "export const SCHEMA_VERSION = '3h';") + .replace("export const SCHEMA_VERSION = '3k';", "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), '3j'); + assert.equal(version(path), '3k'); 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 299ffb985..3f148f080 100644 --- a/packages/bus/tests/chief-of-staff.test.mjs +++ b/packages/bus/tests/chief-of-staff.test.mjs @@ -606,7 +606,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 = '3j';", "export const SCHEMA_VERSION = '3c';") + .replace("export const SCHEMA_VERSION = '3k';", "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'); @@ -630,7 +630,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, '3j'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3k'); 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 161fdd63a..50b0f90d5 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, '3j'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3k'); 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 = '3j';", "export const SCHEMA_VERSION = '3k';") - .replace('}];\nconst timedTables', `}, { from: '3j', to: '3k', sql: ${JSON.stringify(futureSQL)} }];\nconst timedTables`); + .replace("export const SCHEMA_VERSION = '3k';", "export const SCHEMA_VERSION = '3l';") + .replace('}];\nconst timedTables', `}, { from: '3k', to: '3l', 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']) + 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']) 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, '3k'); + assert.equal(future.get("SELECT value FROM meta WHERE key='schema_version'").value, '3l'); 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 3j ready/); + assert.match(migrated.stdout, /bus schema 3k ready/); new Store(root).close(); const usage = spawnSync(process.execPath, [command], { encoding: 'utf8' }); assert.equal(usage.status, 2); diff --git a/packages/bus/tests/end-launch.test.mjs b/packages/bus/tests/end-launch.test.mjs index 6540277e1..49c11d492 100644 --- a/packages/bus/tests/end-launch.test.mjs +++ b/packages/bus/tests/end-launch.test.mjs @@ -8,6 +8,12 @@ import { join } from 'node:path'; import { Store } from '../src/store.mjs'; import { Broker } from '../src/broker.mjs'; import { Client } from '../src/client.mjs'; +// Row 264a (#1769): a thread made by hand gets the default session the 3j-3k migration gives a +// legacy thread, with the thread's id. +const DEFAULT_SESSION = [ + 'INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3k' FROM threads WHERE id=?', + 'INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3k' FROM threads WHERE id=?', +]; const businesses = { demo: { @@ -149,7 +155,8 @@ test('broker process: launch ops authorize role.launch, record refusals and end const side = new Store(data); side.run('INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', 'p-1', '2026-10-15T00:00:00.000Z', 'demo', 'work', null, 'jason'); side.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', 'th-1', '2026-10-15T00:00:00.000Z', 'demo', 'p-1', 'coder', 'jason', 'one'); - side.run('INSERT INTO thread_opens(at,thread,by) VALUES(?,?,?)', '2026-10-15T00:00:01.000Z', 'th-1', 'jason'); + for (const sql of DEFAULT_SESSION) side.run(sql, 'th-1'); + side.run('INSERT INTO thread_opens(at,thread,session,by) VALUES(?,?,?,?)', '2026-10-15T00:00:01.000Z', 'th-1', 'th-1', 'jason'); side.close(); const ready = await ask(child, { op: 'boot', diff --git a/packages/bus/tests/fixtures/schema-3j.sql b/packages/bus/tests/fixtures/schema-3j.sql new file mode 100644 index 000000000..fe664c8e1 --- /dev/null +++ b/packages/bus/tests/fixtures/schema-3j.sql @@ -0,0 +1,396 @@ +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 thread_claims ( + 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 ('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) +) STRICT; +CREATE TABLE thread_opens ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + at TEXT NOT NULL, + thread TEXT NOT NULL REFERENCES threads(id), + by TEXT NOT NULL +) STRICT; +CREATE INDEX thread_claims_thread ON thread_claims (thread, seq); +CREATE INDEX message_threads_thread ON message_threads (thread); +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 thread_one_holder BEFORE INSERT ON thread_claims + WHEN NEW.op = 'claim' AND (SELECT op FROM thread_claims WHERE thread = NEW.thread ORDER BY seq DESC LIMIT 1) = 'claim' + BEGIN SELECT RAISE(ABORT, 'thread 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 thread = c.thread) AND c.op = 'claim') + BEGIN SELECT RAISE(ABORT, 'run already holds a thread'); 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 thread = NEW.thread ORDER BY seq DESC LIMIT 1),'') <> 'claim' + BEGIN SELECT RAISE(ABORT, 'thread 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 thread = NEW.thread ORDER BY seq DESC LIMIT 1) <> NEW.holder_run + BEGIN SELECT RAISE(ABORT, 'only the holder releases; others revoke'); 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 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 delivery_same_thread 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 thread = t.id) + AND c.op = 'claim' AND c.holder_run IS NEW.holder_run)) + BEGIN SELECT RAISE(ABORT, 'a threaded message goes only to its thread''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 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; diff --git a/packages/bus/tests/grant-migration.test.mjs b/packages/bus/tests/grant-migration.test.mjs index 2715c1d62..d07c7032a 100644 --- a/packages/bus/tests/grant-migration.test.mjs +++ b/packages/bus/tests/grant-migration.test.mjs @@ -16,6 +16,9 @@ const TAG = 'mosaic-' + 'a'.repeat(32); const ddl3i = readFileSync(new URL('./fixtures/schema-3i.sql', import.meta.url), 'utf8'); const here = (p) => new URL(p, import.meta.url); const NEW = ['hosts', 'host_grants', 'events_grant_actor', 'sqlite_autoindex_hosts_1', 'sqlite_autoindex_host_grants_1']; +// Row 264a (#1769): what the 3j→3k step rebuilds or adds. sessions.test.mjs tests those objects. +const SESSIONS_3K = ['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens']; +const later3k = (r) => SESSIONS_3K.includes(r.tbl_name) || ['delivery_same_thread', 'delivery_same_session'].includes(r.name); function source3i(t) { const root = mkdtempSync(join(tmpdir(), 'grant-migrate-')); t.after(() => rmSync(root, { recursive: true, force: true })); @@ -35,7 +38,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')").all() + return Object.fromEntries(db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT IN ('hosts','host_grants','sessions','session_defaults')").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(); } @@ -61,6 +64,7 @@ async function engine(root, name, { store, schema, step = (s) => s }) { for (const file of ['3b.sql', '3b-3c.sql', '3c-3d.sql', '3d-3e.sql', '3e-3f.sql', '3f-3g.sql', '3g-3h.sql', '3h-3i.sql']) copyFileSync(here('../src/migrations/' + file), join(moduleRoot, 'src/migrations', file)); writeFileSync(join(moduleRoot, 'src/migrations/3i-3j.sql'), step(readFileSync(here('../src/migrations/3i-3j.sql'), 'utf8'))); + copyFileSync(here('../src/migrations/3j-3k.sql'), join(moduleRoot, 'src/migrations/3j-3k.sql')); return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; } function fresh(t) { @@ -84,10 +88,10 @@ 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), '3j'); + assert.equal(version(path), '3k'); assert.deepEqual(rows(path), before); const now = master(path); - assert.deepEqual(now.filter(r => !NEW.includes(r.name)), was); + assert.deepEqual(now.filter(r => !NEW.includes(r.name) && !later3k(r)), was.filter(r => !later3k(r))); assert.deepEqual(now.filter(r => NEW.includes(r.name)).map(r => r.name).sort(), [...NEW].sort()); const store = new Store(root); try { @@ -109,10 +113,10 @@ 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), '3j'); + assert.equal(version(path), '3k'); }); -test('the 3i engine refuses a 3j store, migrating or not, and leaves it at 3j', async t => { +test('the 3i engine refuses a migrated store, migrating or not, and leaves it as it was', async t => { const { root, path } = source3i(t); new Store(root, { migrateSchema: true }).close(); const after = rows(path), was = master(path); @@ -120,14 +124,14 @@ test('the 3i engine refuses a 3j store, migrating or not, and leaves it at 3j', 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 = '3j';", "export const SCHEMA_VERSION = '3i';") + .replace("export const SCHEMA_VERSION = '3k';", "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), '3j'); + assert.equal(version(path), '3k'); 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 e8e660c87..92b4cc4b3 100644 --- a/packages/bus/tests/name-migration.test.mjs +++ b/packages/bus/tests/name-migration.test.mjs @@ -55,7 +55,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, '3j'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3k'); 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 e295c0b09..8cf06e491 100644 --- a/packages/bus/tests/schema.test.mjs +++ b/packages/bus/tests/schema.test.mjs @@ -520,7 +520,8 @@ test('v3b prototype refusals, views and append-only mutations', (t) => { // Row 141 (#1635) adds instance_names and its 4 triggers; names.test.mjs tests them. // Row 124 adds usage_records and its three append-only triggers. // Row 234 (#1740) adds hosts, host_grants and the events_grant_actor guard; grant-migration.test.mjs tests them. - assert.equal(count('table') - 1, 18); - assert.equal(count('trigger'), 71); + // Row 264a (#1769) adds sessions, session_defaults and 15 triggers; sessions.test.mjs tests them. + assert.equal(count('table') - 1, 20); + assert.equal(count('trigger'), 86); assert.equal(count('view'), 5); }); diff --git a/packages/bus/tests/sessions.test.mjs b/packages/bus/tests/sessions.test.mjs new file mode 100644 index 000000000..3a8c4ce15 --- /dev/null +++ b/packages/bus/tests/sessions.test.mjs @@ -0,0 +1,392 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { DatabaseSync } from 'node:sqlite'; +import { chmodSync, copyFileSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createHash } from 'node:crypto'; +import { fileURLToPath, pathToFileURL } from 'node:url'; +import { Store, schemaDigest } from '../src/store.mjs'; +import { Broker } from '../src/broker.mjs'; + +// Row 264a (#1769, T1 item 7): sessions below a thread, in the bus. T1's bus half, T2 and T7 of +// docs/plans/2026-10-21_t1-sessions.md. The launcher, seat and Console halves are 264b-264e's. +const shipped = (name) => + JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), 'utf8')); +const roles = { + chief: { definition: 'chief-of-staff', authority: { withinRole: ['message.send'], crossRole: [] } }, + assistant: { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } }, + pm: { definition: 'pm', authority: { withinRole: ['message.send'], crossRole: [] } }, +}; +const businesses = { + demo: { + id: 'demo', + human: 'jason', + arbiters: { technical: 'pm', delivery: 'chief' }, + roles, + definitions: { 'chief-of-staff': {}, assistant: { messaging: shipped('assistant').messaging }, pm: {} }, + projects: { stack: { root: '/nonexistent' } }, + }, +}; +const AT = '2026-10-21T00:00:00.000Z'; +const now = () => new Date().toISOString(); +const here = (p) => new URL(p, import.meta.url); +const ddl3j = readFileSync(here('./fixtures/schema-3j.sql'), 'utf8'); + +function broker(t, store) { + const b = new Broker({ store, businesses }); + const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }); + const caps = { human }; + const call = (who, verb, args = {}) => b.request(caps[who], { verb, args }); + const launch = (name, role, extra = {}) => { + caps[name] = b.bindLaunch({ business: 'demo', role, run: `${name}-run`, harness: 'pi', ...extra }); + return caps[name]; + }; + const claim = (name) => call(name, 'role.claim'); + const sessionOf = (message) => store.get('SELECT session FROM message_threads WHERE message=?', message)?.session ?? null; + const claims = (session) => store.all('SELECT op, holder_run, generation FROM thread_claims WHERE session=? ORDER BY seq', session).map((r) => ({ ...r })); + return { b, store, caps, call, launch, claim, sessionOf, claims }; +} + +function setup(t) { + const root = mkdtempSync(join(tmpdir(), 'bus-sessions-')); + const store = new Store(root); + t.after(() => { + store.close(); + rmSync(root, { recursive: true, force: true }); + }); + const ctx = broker(t, store); + const project = ctx.call('human', 'project.create', { name: 'stack work', repo: 'stack' }).id; + const A = ctx.call('human', 'thread.create', { project, name: 'alpha' }).id; + for (const r of ['chief', 'pm']) { + ctx.launch(r, r); + ctx.claim(r); + } + const [first] = ctx.call('human', 'session.list', { thread: A }); + return { ...ctx, root, project, A, D: first.id }; +} + +test('T1 REQ-ROLE-2: two sessions of one thread keep their own history, delivery and replies', (t) => { + const { call, launch, claim, sessionOf, store, project, A, D } = setup(t); + // thread.create made one default session, with a random id that isn't the thread's. + assert.deepEqual(call('human', 'session.list', { thread: A }).map((r) => [r.name, r.predecessor, r.default]), [['default', null, true]]); + assert.notEqual(D, A); + assert.match(D, /^[a-f0-9-]{36}$/); + const S = call('human', 'session.create', { thread: A, name: 'second', op: 'op-second' }).id; + assert.deepEqual(call('human', 'session.list', { thread: A }).map((r) => [r.id, r.name, r.default]), [ + [D, 'default', true], + [S, 'second', false], + ]); + // A run with no session named works the default (D3). Each session has its own holder. + launch('a1', 'assistant', { thread: A }); + launch('a2', 'assistant', { thread: A, session: S }); + assert.deepEqual(claim('a1'), { role: 'assistant', run: 'a1-run', thread: A }); + assert.deepEqual(claim('a2'), { role: 'assistant', run: 'a2-run', thread: A }); + const toDefault = call('human', 'thread.send', { thread: A, body: 'to the default' }); + const toSecond = call('human', 'thread.send', { thread: A, session: S, body: 'to the second' }); + assert.equal(sessionOf(toDefault.id), D); + assert.equal(sessionOf(toSecond.id), S); + assert.deepEqual(call('a1', 'message.receive').map((m) => m.body), ['to the default']); + assert.deepEqual(call('a2', 'message.receive').map((m) => m.body), ['to the second']); + assert.deepEqual(call('a1', 'message.receive'), []); + // A reply stays in its run's session, and each history reads only its own. + const reply = call('a2', 'message.send', { to: 'human', body: 'second answers', in_reply_to: toSecond.id }); + assert.equal(sessionOf(reply.id), S); + assert.deepEqual(call('human', 'thread.messages', { thread: A }).map((m) => m.body), ['to the default']); + assert.deepEqual(call('human', 'thread.messages', { thread: A, session: S }).map((m) => m.body), ['to the second', 'second answers']); + // The chief's reply to the second session's work comes back to that session. + const ask = call('a2', 'message.send', { to: 'chief', body: 'need help', class: 'REQUEST' }); + const back = call('chief', 'message.send', { to: 'assistant', body: 'here', in_reply_to: ask.id }); + assert.equal(sessionOf(back.id), S); + assert.deepEqual(call('a2', 'message.receive').map((m) => m.body), ['here']); + assert.deepEqual(call('a1', 'message.receive'), []); + // An open names its session too. + call('human', 'thread.open', { thread: A, session: S }); + assert.equal(store.get('SELECT session FROM thread_opens ORDER BY seq DESC LIMIT 1').session, S); + // A non-thread role names no session, and a session of another thread is refused. + assert.throws(() => launch('p2', 'pm', { session: S }), { code: 'thread-not-allowed' }); + const B = call('human', 'thread.create', { project, name: 'beta' }).id; + assert.throws(() => launch('a3', 'assistant', { thread: B, session: S }), { code: 'session-not-found' }); + assert.throws(() => call('human', 'thread.send', { thread: B, session: S, body: 'x' }), { code: 'session-not-found' }); + assert.throws(() => call('human', 'thread.messages', { thread: B, session: 'no-such-session' }), { code: 'session-not-found' }); +}); + +test('T2 REQ-ROLE-2: one holder per session, generations count its claims, and a sibling session\'s message never reaches the other holder', (t) => { + const { b, call, launch, claim, claims, store, project, A, D } = setup(t); + const S = call('human', 'session.create', { thread: A, name: 'second', op: 'op-second' }).id; + launch('a1', 'assistant', { thread: A }); + claim('a1'); + launch('a1b', 'assistant', { thread: A }); + assert.throws(() => claim('a1b'), { code: 'thread-already-held' }); + // A sibling session is free to claim while the default is held. + launch('a2', 'assistant', { thread: A, session: S }); + claim('a2'); + // Release and claim again: the second claim is generation 2, and its release carries it. + call('a2', 'role.release'); + launch('a3', 'assistant', { thread: A, session: S }); + claim('a3'); + b.endLaunch({ business: 'demo', run: 'a3-run', reason: 'idle' }); + assert.deepEqual(claims(S), [ + { op: 'claim', holder_run: 'a2-run', generation: 1 }, + { op: 'release', holder_run: 'a2-run', generation: 1 }, + { op: 'claim', holder_run: 'a3-run', generation: 2 }, + { op: 'release', holder_run: 'a3-run', generation: 2 }, + ]); + // The human's stop revokes each held session, with that claim's generation. + launch('a4', 'assistant', { thread: A, session: S }); + claim('a4'); + assert.deepEqual(call('human', 'assistant.stop', { instance: 'assistant' }), { + stopped: [ + { thread: A, run: 'a1-run' }, + { thread: A, run: 'a4-run' }, + ], + }); + assert.deepEqual(claims(D).at(-1), { op: 'revoke', holder_run: 'a1-run', generation: 1 }); + assert.deepEqual(claims(S).at(-1), { op: 'revoke', holder_run: 'a4-run', generation: 3 }); + // The triggers hold these for a write that skips the broker. + const raw = (thread, session, generation, run) => + store.run( + "INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by) VALUES(?,'demo',?,?,?,'claim',?,'pi',?)", + now(), + thread, + session, + generation, + run, + run, + ); + assert.throws(() => raw(A, S, 5, 'x-run'), /a claim row names its session's generation/); + assert.throws(() => raw(A, S, 3, 'x-run'), /a claim row names its session's generation/); + const B = call('human', 'thread.create', { project, name: 'beta' }).id; + const [{ id: BD }] = call('human', 'session.list', { thread: B }); + assert.throws(() => raw(A, BD, 1, 'x-run'), /a claim's session is in its thread/); + raw(A, S, 4, 'x-run'); + assert.throws(() => raw(B, BD, 1, 'x-run'), /run already holds a session/); + // A threaded message goes only to its own session's holder. + launch('a5', 'assistant', { thread: A }); + claim('a5'); + const m = call('human', 'thread.send', { thread: A, session: S, body: 'for x' }); + assert.throws( + () => store.run("INSERT INTO deliveries(message,at,op,holder_run) VALUES(?,?,'delivered',?)", m.id, now(), 'a5-run'), + /a threaded message goes only to its session's holder/, + ); + store.run("INSERT INTO deliveries(message,at,op,holder_run) VALUES(?,?,'delivered',?)", m.id, now(), 'x-run'); +}); + +test('session.create is idempotent on its op, refuses a reused op or a taken name, and reaches only the owner\'s threads', (t) => { + const { call, store, project, A, D } = setup(t); + const first = call('human', 'session.create', { thread: A, name: 'second', op: 'op-1' }); + assert.deepEqual(call('human', 'session.create', { thread: A, name: 'second', op: 'op-1' }), first); + assert.throws(() => call('human', 'session.create', { thread: A, name: 'third', op: 'op-1' }), { code: 'session-op-conflict' }); + assert.throws(() => call('human', 'session.create', { thread: A, name: 'second', op: 'op-2' }), { code: 'session-exists' }); + assert.throws(() => call('human', 'session.create', { thread: A, name: 'default', op: 'op-3' }), { code: 'session-exists' }); + // thread.create's op answers a retry that names the default. + assert.deepEqual(call('human', 'session.create', { thread: A, name: 'default', op: 'default' }), { id: D, thread: A, name: 'default' }); + for (const args of [{ thread: A, name: 'x\ny', op: 'op-4' }, { thread: A, name: 'x' }, { thread: A, name: 'x', op: 'op-5', extra: 1 }]) + assert.throws(() => call('human', 'session.create', args), { code: 'invalid-request' }); + assert.equal(store.get('SELECT count(*) n FROM sessions WHERE thread=?', A).n, 2); + // Another owner's thread, made by hand with the session the broker would write. + const other = 'thread-of-someone-else'; + store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', other, now(), 'demo', project, 'assistant', 'someone-else', 'theirs'); + store.run("INSERT INTO sessions(id,at,business,thread,instance,owner,name,op) VALUES('s-theirs',?,'demo',?,'assistant','someone-else','default','default')", now(), other); + store.run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,?,'s-theirs','someone-else')", now(), other); + for (const [verb, args] of [ + ['session.list', { thread: other }], + ['session.create', { thread: other, name: 'mine', op: 'op-6' }], + ['thread.send', { thread: other, session: 's-theirs', body: 'hi' }], + ]) + assert.throws(() => call('human', verb, args), { code: 'thread-not-owned' }, verb); + assert.throws(() => call('human', 'session.list', { thread: 'nope' }), { code: 'thread-not-found' }); + // Agents don't call the session verbs. + assert.throws(() => call('pm', 'session.list', { thread: A }), { code: 'human-required' }); +}); + +test('the 3k triggers keep sessions in their thread and session ids apart from thread ids', (t) => { + // A bare database from schema.sql, so only the triggers stand between a write and the tables. + const root = mkdtempSync(join(tmpdir(), 'bus-sessions-sql-')); + const db = new DatabaseSync(join(root, 'bus.sqlite')); + t.after(() => { + db.close(); + rmSync(root, { recursive: true, force: true }); + }); + db.exec(readFileSync(here('../schema.sql'), 'utf8')); + const run = (sql, ...args) => db.prepare(sql).run(...args); + const session = (id, thread, extra = {}) => + run( + 'INSERT INTO sessions(id,at,business,thread,instance,owner,name,op,predecessor) VALUES(?,?,?,?,?,?,?,?,?)', + id, + now(), + extra.business ?? 'demo', + thread, + extra.instance ?? 'assistant', + extra.owner ?? 'jason', + extra.name ?? id, + extra.op ?? id, + extra.predecessor ?? null, + ); + run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES('p-1',?,'demo','work',NULL,'jason')", now()); + for (const id of ['th-1', 'th-2']) + run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,'demo','p-1','assistant','jason',?)", id, now(), id); + session('s-1', 'th-1'); + // The legacy shape: a session whose id is its own thread's. + session('th-1', 'th-1', { name: 'default', op: 'legacy-3k' }); + // A session id that names another thread, and a thread id that names a session, are refused. + assert.throws(() => session('th-2', 'th-1'), /a session id names another thread/); + assert.throws( + () => run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES('s-1',?,'demo','p-1','assistant','jason','x')", now()), + /a thread id names a session/, + ); + // A session has its thread's business, instance and owner, and its predecessor is in its thread. + for (const extra of [{ business: 'other' }, { instance: 'pm' }, { owner: 'someone-else' }]) + assert.throws(() => session('s-x', 'th-1', extra), /a session has its thread's business, instance and owner/, JSON.stringify(extra)); + assert.throws(() => session('s-x', 'th-2', { predecessor: 's-1' }), /a session's predecessor is in its thread/); + session('s-2', 'th-1', { predecessor: 's-1' }); + // The names and op keys are unique in a thread. + assert.throws(() => session('s-3', 'th-1', { name: 's-1' }), /UNIQUE/); + assert.throws(() => session('s-3', 'th-1', { op: 's-1' }), /UNIQUE/); + // A default is a session of its thread. + assert.throws(() => run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,'th-2','s-1','jason')", now()), /a default session is in its thread/); + run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,'th-1','s-1','jason')", now()); + // A message's and an open's session is in their thread. + run("INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES('m-1',?,'demo','human','human-cli','assistant','INFO','hi')", now()); + assert.throws(() => run("INSERT INTO message_threads(message,thread,session) VALUES('m-1','th-2','s-1')"), /a threaded message's session is in its thread/); + assert.throws(() => run("INSERT INTO thread_opens(at,thread,session,by) VALUES(?,'th-2','s-1','jason')", now()), /an open's session is in its thread/); +}); + +// T7: a 3j store from the pinned 3j schema (the parent of this change's schema.sql) and fixture +// rows. No live store is read or copied. +function source3j(t) { + const root = mkdtempSync(join(tmpdir(), 'sessions-migrate-')); + t.after(() => rmSync(root, { recursive: true, force: true })); + mkdirSync(join(root, 'bus'), { mode: 0o700 }); + const path = join(root, 'bus/bus.sqlite'), db = new DatabaseSync(path); + assert.equal(createHash('sha256').update(ddl3j).digest('hex'), '0a7b633ce7eec9d0bb37c3255f4244d97bd0c8eac38db91a11b4ef6cc982ec8c'); + db.exec(ddl3j); + const run = (sql, ...args) => db.prepare(sql).run(...args); + run('INSERT INTO meta VALUES (?,?)', 'schema_version', '3j'); + run('INSERT INTO meta VALUES (?,?)', 'schema_digest', schemaDigest(db)); + run('INSERT INTO meta VALUES (?,?)', 'comment_cursor_epoch', AT); + run('INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', 'p-1', AT, 'demo', 'stack work', 'stack', 'jason'); + for (const id of ['t1', 't2']) + run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', id, AT, 'demo', 'p-1', 'assistant', 'jason', id); + // Duckworth's fixture: claim, release, claim, revoke on t1, with t2's claim between, at seqs + // with gaps so a renumbered copy would show. + const claimRow = (seq, thread, op, holder, extra = {}) => + run( + 'INSERT INTO thread_claims(seq,at,business,thread,op,holder_run,harness,by,reason) VALUES(?,?,?,?,?,?,?,?,?)', + seq, AT, 'demo', thread, op, holder, 'pi', extra.by ?? holder, extra.reason ?? null, + ); + claimRow(10, 't1', 'claim', 'a-run'); + claimRow(11, 't2', 'claim', 'b-run'); + claimRow(12, 't1', 'release', 'a-run'); + claimRow(20, 't1', 'claim', 'c-run'); + claimRow(40, 't1', 'revoke', 'c-run', { by: 'jason', reason: 'stuck' }); + for (const [id, thread] of [['m-1', 't1'], ['m-2', 't2'], ['m-3', 't1']]) { + run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', id, AT, 'demo', 'human', 'human-cli', 'assistant', 'INFO', id); + run('INSERT INTO message_threads VALUES(?,?)', id, thread); + } + run('INSERT INTO thread_opens(seq,at,thread,by) VALUES(?,?,?,?)', 7, AT, 't1', 'jason'); + db.close(); chmodSync(path, 0o600); + return { root, path }; +} +const master = (path) => { + const db = new DatabaseSync(path, { readOnly: true }); + try { + return db.prepare('SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name').all().map((r) => ({ ...r })); + } finally { db.close(); } +}; +const rows = (path) => { + const db = new DatabaseSync(path, { readOnly: true }); + try { + return Object.fromEntries(['sessions', 'session_defaults', 'thread_claims', 'message_threads', 'thread_opens', 'meta'] + .map((name) => [name, db.prepare(`SELECT * FROM ${name} ORDER BY rowid`).all().map((r) => ({ ...r }))])); + } finally { db.close(); } +}; +const version = (path) => { + const db = new DatabaseSync(path, { readOnly: true }); + try { return db.prepare("SELECT value FROM meta WHERE key='schema_version'").get().value; } finally { db.close(); } +}; + +test('T7: a 3j store migrates to 3k, each thread\'s default session has the thread\'s id, and old rows keep their seqs', (t) => { + const { root, path } = source3j(t); + assert.throws(() => new Store(root), /schema-mismatch/); + new Store(root, { migrateSchema: true }).close(); + assert.equal(version(path), '3k'); + // 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 })); + new Store(freshRoot).close(); + assert.deepEqual(master(path), master(join(freshRoot, 'bus/bus.sqlite'))); + const store = new Store(root); + t.after(() => store.close()); + assert.deepEqual(store.all('SELECT id,at,business,thread,instance,owner,name,op,predecessor FROM sessions ORDER BY seq').map((r) => ({ ...r })), ['t1', 't2'].map((id) => ( + { id, at: AT, business: 'demo', thread: id, instance: 'assistant', owner: 'jason', name: 'default', op: 'legacy-3k', predecessor: null }))); + assert.deepEqual(store.all('SELECT at,thread,session,by FROM session_defaults ORDER BY seq').map((r) => ({ ...r })), ['t1', 't2'].map((id) => ( + { at: AT, thread: id, session: id, by: 'migration-3k' }))); + // Generation counts claim rows only: claim, release, claim, revoke on t1 is 1, 1, 2, 2. + assert.deepEqual(store.all('SELECT seq,thread,session,op,generation FROM thread_claims ORDER BY seq').map((r) => ({ ...r })), [ + { seq: 10, thread: 't1', session: 't1', op: 'claim', generation: 1 }, + { seq: 11, thread: 't2', session: 't2', op: 'claim', generation: 1 }, + { seq: 12, thread: 't1', session: 't1', op: 'release', generation: 1 }, + { seq: 20, thread: 't1', session: 't1', op: 'claim', generation: 2 }, + { seq: 40, thread: 't1', session: 't1', op: 'revoke', generation: 2 }, + ]); + assert.deepEqual(store.all('SELECT message,thread,session FROM message_threads ORDER BY rowid').map((r) => ({ ...r })), [ + { message: 'm-1', thread: 't1', session: 't1' }, + { message: 'm-2', thread: 't2', session: 't2' }, + { message: 'm-3', thread: 't1', session: 't1' }, + ]); + assert.deepEqual({ ...store.get('SELECT seq,thread,session,by FROM thread_opens') }, { seq: 7, thread: 't1', session: 't1', by: 'jason' }); + for (const [name, seq] of [['thread_claims', 40], ['thread_opens', 7]]) + assert.equal(store.get('SELECT seq FROM sqlite_sequence WHERE name=?', name).seq, seq, name); + assert.equal(store.get('PRAGMA quick_check').quick_check, 'ok'); + assert.deepEqual(store.all('PRAGMA foreign_key_check'), []); + // A legacy client names no session and gets the default: t2's holder releases at generation 1, + // and t1's next claim is its third. + const { call, launch, claim, claims, sessionOf } = broker(t, store); + for (const r of ['chief', 'pm']) { + launch(r, r); + claim(r); + } + assert.throws(() => call('human', 'session.create', { thread: 't1', name: 'default', op: 'op-new' }), { code: 'session-exists' }); + store.run("INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by) VALUES(?,'demo','t2','t2',1,'release','b-run','pi','b-run')", now()); + assert.equal(store.get("SELECT seq FROM thread_claims WHERE thread='t2' ORDER BY seq DESC LIMIT 1").seq, 41); + launch('d', 'assistant', { thread: 't1' }); + claim('d'); + assert.deepEqual(claims('t1').at(-1), { op: 'claim', holder_run: 'd-run', generation: 3 }); + const sent = call('human', 'thread.send', { thread: 't1', body: 'after the step' }); + assert.equal(sessionOf(sent.id), 't1'); + assert.deepEqual(call('d', 'message.receive').map((m) => m.body), ['m-1', 'm-3', 'after the step']); + assert.deepEqual(call('human', 'thread.messages', { thread: 't1' }).map((m) => m.body), ['m-1', 'm-3', 'after the step']); +}); + +// The 3j engine is store.mjs at this change with the 3j schema and version and no 3j→3k step. +async function engine3j(root) { + const moduleRoot = join(root, 'engine-3j'); + mkdirSync(join(moduleRoot, 'src/migrations'), { recursive: true }); + 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(ddl3j).digest('hex')}';`) + .replace("export const SCHEMA_VERSION = '3k';", "export const SCHEMA_VERSION = '3j';") + .replace(/\}, \{\n {2}\/\/ Row 264a \(#1769\)[\s\S]*?\}\];\nconst timedTables/, '}];\nconst timedTables'); + assert.ok(edited.includes("export const SCHEMA_VERSION = '3j';") && !edited.includes('3j-3k'), 'the 3j engine has no 3j→3k step'); + writeFileSync(join(moduleRoot, 'src/store.mjs'), edited); + writeFileSync(join(moduleRoot, 'schema.sql'), ddl3j); + for (const file 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']) + copyFileSync(here('../src/migrations/' + file), join(moduleRoot, 'src/migrations', file)); + return (await import(pathToFileURL(join(moduleRoot, 'src/store.mjs')))).Store; +} + +test('T7 D9: the 3j engine refuses a 3k store, migrating or not, and leaves it as it was', async (t) => { + const { root, path } = source3j(t); + new Store(root, { migrateSchema: true }).close(); + const was = master(path), before = rows(path); + const Store3j = await engine3j(root); + assert.throws(() => new Store3j(root), /schema-mismatch/); + assert.throws(() => new Store3j(root, { migrateSchema: true }), /schema-mismatch/); + assert.equal(version(path), '3k'); + assert.deepEqual(master(path), was); + assert.deepEqual(rows(path), before); + // The same engine opens a fresh 3j store, so the refusal is the version's and not a broken copy. + const { root: old } = source3j(t); + new Store3j(old).close(); +}); diff --git a/packages/bus/tests/spec-migration.test.mjs b/packages/bus/tests/spec-migration.test.mjs index 8f7fd1ac6..5e3e68979 100644 --- a/packages/bus/tests/spec-migration.test.mjs +++ b/packages/bus/tests/spec-migration.test.mjs @@ -53,7 +53,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, '3j'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3k'); 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 bd9964332..78fd7f9d0 100644 --- a/packages/bus/tests/tester-writes.test.mjs +++ b/packages/bus/tests/tester-writes.test.mjs @@ -21,6 +21,8 @@ test('row 175: TESTER_VERBS and the organization branch tester set are pinned', 'usage', 'names', 'limits', 'plugins', 'instances', 'project.list', 'thread.list', 'name.set', 'project.create', 'credentials', // Row 177 (#1675): the tester's own threads. 'thread.create', 'thread.open', 'thread.send', 'thread.messages', 'assistant.stop', + // Row 264a (#1769): the sessions of the tester's own threads. + 'session.list', 'session.create', ]); 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 e4c0e49c8..5c2f9e775 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 = '3j';", "export const SCHEMA_VERSION = '3d';") + .replace("export const SCHEMA_VERSION = '3k';", "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, '3j'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3k'); 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 3j ready', ''].join('\n')); + assert.equal(migrated.stdout, [...REPORT, 'bus schema 3k ready', ''].join('\n')); }); diff --git a/packages/bus/tests/threads.test.mjs b/packages/bus/tests/threads.test.mjs index 311a65dbc..1d2f14da6 100644 --- a/packages/bus/tests/threads.test.mjs +++ b/packages/bus/tests/threads.test.mjs @@ -7,6 +7,12 @@ import { join } from 'node:path'; import { fileURLToPath } from 'node:url'; import { Store } from '../src/store.mjs'; import { Broker, THREAD_PAGE } from '../src/broker.mjs'; +// Row 264a (#1769): a thread made by hand gets the default session the 3j-3k migration gives a +// legacy thread, with the thread's id. +const DEFAULT_SESSION = [ + 'INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3k' FROM threads WHERE id=?', + 'INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3k' FROM threads WHERE id=?', +]; // Row 121 (#1615): projects and threads in the bus. The assistant's messaging lists are the // shipped roles/assistant.json's, so the chain below runs under the lists the stack runs. @@ -81,32 +87,36 @@ test('REQ-ROLE-2 REQ-THREAD-1: one holder per thread, through the broker and the const { b, store, call, launch, claim, A, B } = twoThreads(t); launch('a2', 'assistant', { thread: A }); assert.throws(() => claim('a2'), { code: 'thread-already-held' }); - // The trigger refuses a second claim that skips the broker. + // The trigger refuses a second claim that skips the broker. Row 264a: the claim is of the + // thread's default session, with the generation the trigger expects. + const sessionOf = (thread) => store.get('SELECT session FROM session_defaults WHERE thread=?', thread).session; const second = () => store.run( - "INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by) VALUES(?,?,?,'claim',?,?,?)", + "INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by) VALUES(?,?,?,?,2,'claim',?,?,?)", now(), 'demo', A, + sessionOf(A), 'a2-run', 'pi', 'a2-run', ); - assert.throws(second, /thread already held/); - // Second case: one run holds one thread. a-run already holds A. + assert.throws(second, /session already held/); + // Second case: one run holds one session. a-run already holds A's. call('b', 'role.release'); assert.throws( () => store.run( - "INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by) VALUES(?,?,?,'claim',?,?,?)", + "INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by) VALUES(?,?,?,?,2,'claim',?,?,?)", now(), 'demo', B, + sessionOf(B), 'a-run', 'pi', 'a-run', ), - /run already holds a thread/, + /run already holds a session/, ); // A released thread can be claimed again by a new run; the claim is the thread's, not the role's. launch('b2', 'assistant', { thread: B }); @@ -128,12 +138,20 @@ test('REQ-ROLE-2 REQ-THREAD-1: the 3e triggers refuse writes that skip the broke }); db.exec(readFileSync(new URL('../schema.sql', import.meta.url), 'utf8')); const run = (sql, ...args) => db.prepare(sql).run(...args); + // Row 264a (#1769): a claim row names its session and the generation thread_claim_generation + // expects, so each refusal below comes from the trigger it names. + const generation = (op) => + op === 'claim' + ? 1 + db.prepare("SELECT count(*) n FROM thread_claims WHERE session='s-1' AND op='claim'").get().n + : (db.prepare("SELECT generation FROM thread_claims WHERE session='s-1' ORDER BY seq DESC LIMIT 1").get()?.generation ?? null); const claim = (op, run_, extra = {}) => run( - 'INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by,reason,decision) VALUES(?,?,?,?,?,?,?,?,?)', + 'INSERT INTO thread_claims(at,business,thread,session,generation,op,holder_run,harness,by,reason,decision) VALUES(?,?,?,?,?,?,?,?,?,?,?)', now(), extra.business ?? 'demo', extra.thread ?? 'th-1', + 's-1', + generation(op), op, run_, 'pi', @@ -143,9 +161,11 @@ test('REQ-ROLE-2 REQ-THREAD-1: the 3e triggers refuse writes that skip the broke ); run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES('p-1',?,'demo','work',NULL,'jason')", now()); run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES('th-1',?,'demo','p-1','assistant','jason','one')", now()); - run("INSERT INTO thread_opens(at,thread,by) VALUES(?,'th-1','jason')", now()); + run("INSERT INTO sessions(id,at,business,thread,instance,owner,name,op) VALUES('s-1',?,'demo','th-1','assistant','jason','default','default')", now()); + run("INSERT INTO session_defaults(at,thread,session,by) VALUES(?,'th-1','s-1','jason')", now()); + run("INSERT INTO thread_opens(at,thread,session,by) VALUES(?,'th-1','s-1','jason')", now()); run("INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES('m-1',?,'demo','human','human-cli','assistant','INFO','hi')", now()); - run("INSERT INTO message_threads(message,thread) VALUES('m-1','th-1')"); + run("INSERT INTO message_threads(message,thread,session) VALUES('m-1','th-1','s-1')"); // A thread is in its project's business. assert.throws( () => run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES('th-x',?,'other','p-1','assistant','jason','x')", now()), @@ -154,20 +174,22 @@ test('REQ-ROLE-2 REQ-THREAD-1: the 3e triggers refuse writes that skip the broke // A claim names the thread's business. assert.throws(() => claim('claim', 'a-run', { business: 'other' }), /a thread claim names the thread's business/); // Nothing releases or revokes a thread nobody holds. - assert.throws(() => claim('release', 'a-run'), /thread is not held/); - assert.throws(() => claim('revoke', 'a-run', { reason: 'stuck' }), /thread is not held/); + assert.throws(() => claim('release', 'a-run'), /session is not held/); + assert.throws(() => claim('revoke', 'a-run', { reason: 'stuck' }), /session is not held/); claim('claim', 'a-run'); // Only the holder releases; a revoke needs a decision or a reason. assert.throws(() => claim('release', 'b-run'), /only the holder releases; others revoke/); assert.throws(() => claim('revoke', 'a-run'), /revoke needs a resolved decision or a reason/); claim('revoke', 'a-run', { reason: 'stuck' }); - // Every 3e table is append-only: no UPDATE, no DELETE, no INSERT OR REPLACE. + // Every 3e table, and row 264a's two, is append-only: no UPDATE, no DELETE, no INSERT OR REPLACE. const rows = { projects: ["id='p-1'", 'name', "INSERT OR REPLACE INTO projects(seq,id,at,business,name,created_by) SELECT seq,id,at,business,name,created_by FROM projects WHERE id='p-1'"], threads: ["id='th-1'", 'name', "INSERT OR REPLACE INTO threads(seq,id,at,business,project,instance,owner,name) SELECT seq,id,at,business,project,instance,owner,name FROM threads WHERE id='th-1'"], - thread_claims: ['seq=1', 'reason', 'INSERT OR REPLACE INTO thread_claims(seq,at,business,thread,op,holder_run,harness,by) SELECT seq,at,business,thread,op,holder_run,harness,by FROM thread_claims WHERE seq=1'], - message_threads: ["message='m-1'", 'thread', "INSERT OR REPLACE INTO message_threads(message,thread) VALUES('m-1','th-1')"], - thread_opens: ['seq=1', 'by', 'INSERT OR REPLACE INTO thread_opens(seq,at,thread,by) SELECT seq,at,thread,by FROM thread_opens WHERE seq=1'], + sessions: ["id='s-1'", 'name', "INSERT OR REPLACE INTO sessions(seq,id,at,business,thread,instance,owner,name,op) SELECT seq,id,at,business,thread,instance,owner,name,op FROM sessions WHERE id='s-1'"], + session_defaults: ['seq=1', 'by', 'INSERT OR REPLACE INTO session_defaults(seq,at,thread,session,by) SELECT seq,at,thread,session,by FROM session_defaults WHERE seq=1'], + thread_claims: ['seq=1', 'reason', 'INSERT OR REPLACE INTO thread_claims(seq,at,business,thread,session,generation,op,holder_run,harness,by) SELECT seq,at,business,thread,session,generation,op,holder_run,harness,by FROM thread_claims WHERE seq=1'], + message_threads: ["message='m-1'", 'thread', "INSERT OR REPLACE INTO message_threads(message,thread,session) VALUES('m-1','th-1','s-1')"], + thread_opens: ['seq=1', 'by', 'INSERT OR REPLACE INTO thread_opens(seq,at,thread,session,by) SELECT seq,at,thread,session,by FROM thread_opens WHERE seq=1'], }; for (const [table, [where, col, replace]] of Object.entries(rows)) { const only = new RegExp(`${table} is append-only`); @@ -225,7 +247,7 @@ test('REQ-MSG-1: a message to or from the assistant names its thread, or is refu // The store refuses a thread on a message the instance is not party to. const pmOnly = call('pm', 'message.send', { to: 'coder', body: 'unrelated' }); assert.throws( - () => store.run('INSERT INTO message_threads(message,thread) VALUES(?,?)', pmOnly.id, A), + () => store.run('INSERT INTO message_threads(message,thread,session) VALUES(?,?,(SELECT session FROM session_defaults WHERE thread=?))', pmOnly.id, A, A), /to or from the thread's instance/, ); void b; @@ -277,8 +299,10 @@ test('REQ-MSG-1: a reply can\'t carry one thread instance\'s thread to another t const helper = { definition: 'assistant', authority: { withinRole: ['message.send'], crossRole: [] } }; const { store, call, launch, claim, count } = setup(t, demo({ roles: { ...roles, helper } })); const project = call('human', 'project.create', { name: 'two instances' }).id; - for (const [id, instance] of [['th-a', 'assistant'], ['th-h', 'helper']]) + for (const [id, instance] of [['th-a', 'assistant'], ['th-h', 'helper']]) { store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', id, now(), 'demo', project, instance, 'jason', id); + for (const sql of DEFAULT_SESSION) store.run(sql, id); + } launch('chief', 'chief'); claim('chief'); launch('a', 'assistant', { thread: 'th-a' }); @@ -437,6 +461,7 @@ test('Q169 REQ-USER-2: assistant.stop ends the human\'s own assistant thread run 'someone-else', 'theirs', ); + for (const sql of DEFAULT_SESSION) store.run(sql, other); launch('o', 'assistant', { thread: other }); claim('o'); const last = (thread) => ({ @@ -486,6 +511,7 @@ test('REQ-USER-2: the human verbs refuse another owner\'s thread', (t) => { 'someone-else', 'theirs', ); + for (const sql of DEFAULT_SESSION) store.run(sql, other); for (const [verb, args] of [ ['thread.open', { thread: other }], ['thread.send', { thread: other, body: 'hi' }], diff --git a/packages/bus/tests/usage-migration.test.mjs b/packages/bus/tests/usage-migration.test.mjs index 02c5aa05a..2628423eb 100644 --- a/packages/bus/tests/usage-migration.test.mjs +++ b/packages/bus/tests/usage-migration.test.mjs @@ -51,12 +51,12 @@ function snapshot(path, columns) { return { columns, rows }; } finally { db.close(); } } -test('REQ-LIMIT-4 3g copy migrates through 3h and 3i to 3j preserving every old column, row and sequence; new usage table is empty', t => { +test('REQ-LIMIT-4 3g copy migrates through 3h, 3i and 3j to 3k preserving every old column, row and sequence; new usage table is empty', t => { const { root, path, copy } = source3g(t), before = snapshot(path), untouched = readFileSync(copy); 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, '3j'); + assert.equal(store.get("SELECT value FROM meta WHERE key='schema_version'").value, '3k'); 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/tests/stack-release.test.mjs b/packages/cli/tests/stack-release.test.mjs index 625995e65..91ad25cd7 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','thread_opens','instance_names','usage_records']; +const LISTED=['role_claims','thread_claims','decision_events','messages','deliveries','task_snapshots','projects','threads','sessions','session_defaults','thread_opens','instance_names','usage_records']; 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/packages/cli/tests/thread-cli.test.mjs b/packages/cli/tests/thread-cli.test.mjs index fb278d792..c016ed8eb 100644 --- a/packages/cli/tests/thread-cli.test.mjs +++ b/packages/cli/tests/thread-cli.test.mjs @@ -17,6 +17,12 @@ import { run } from "../src/cli.mjs"; import { CliError } from "../src/errors.mjs"; import { busExit } from "../src/transport.mjs"; import { humanEnv, io } from "./helpers.mjs"; +// Row 264a (#1769): a thread made by hand gets the default session the 3j-3k migration gives a +// legacy thread, with the thread's id. +const DEFAULT_SESSION = [ + "INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3k' FROM threads WHERE id=?", + "INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3k' FROM threads WHERE id=?", +]; const shipped = (name) => JSON.parse(readFileSync(fileURLToPath(new URL(`../../../roles/${name}.json`, import.meta.url)), "utf8")); const roles = { @@ -185,6 +191,7 @@ test("REQ-CLI-1 a thread is its id, an id prefix of 8 characters or more, or (await s.cli(["thread", "open", ref])); const opened = (r) => r.out.trim(); @@ -322,6 +329,7 @@ test("REQ-USER-2 a thread of another user is refused for talk, open, read and fo const theirs = randomUUID(); s.store.run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)", "p-theirs", at, "demo", "theirs", "stack", "someone-else"); s.store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", theirs, at, "demo", "p-theirs", "assistant", "someone-else", "private"); + for (const sql of DEFAULT_SESSION) s.store.run(sql, theirs); // A message waiting for the user: message.receive would write a delivery row for it. s.call("chief", "message.send", { to: "human", body: "waiting", class: "INFO" }); const before = s.writes(); diff --git a/packages/cli/tests/thread-launch.test.mjs b/packages/cli/tests/thread-launch.test.mjs index 2ef1ea804..5c249f4d3 100644 --- a/packages/cli/tests/thread-launch.test.mjs +++ b/packages/cli/tests/thread-launch.test.mjs @@ -27,6 +27,12 @@ import { bootConfig, loadSystem } from "../src/config.mjs"; import { startTimeOf } from "../src/host.mjs"; import { createLauncher } from "../src/launcher.mjs"; import { fakeHarnesses, fixture, tmp } from "./helpers.mjs"; +// Row 264a (#1769): a thread made by hand gets the default session the 3j-3k migration gives a +// legacy thread, with the thread's id. +const DEFAULT_SESSION = [ + "INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3k' FROM threads WHERE id=?", + "INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3k' FROM threads WHERE id=?", +]; const ADAPTER = fileURLToPath(new URL("./fixtures/thread-adapter.mjs", import.meta.url)); const skip = sandboxProblem() ?? false; @@ -356,7 +362,8 @@ test("REQ-THREAD-2 REQ-LAUNCH-1 a thread id that isn't a safe segment, or a thre store.run("INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)", "p-hand", at(), "acme", "hand made", "stack", "jason"); for (const [id, instance] of [["../evil", "assistant"], ["coder-thread", "coder"]]) { store.run("INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)", id, at(), "acme", "p-hand", instance, "jason", id); - store.run("INSERT INTO thread_opens(at,thread,by) VALUES(?,?,?)", at(), id, "jason"); + for (const sql of DEFAULT_SESSION) store.run(sql, id); + store.run("INSERT INTO thread_opens(at,thread,session,by) VALUES(?,?,?,?)", at(), id, id, "jason"); } store.close(); diff --git a/packages/webui/tests/chat.test.mjs b/packages/webui/tests/chat.test.mjs index ab13e34c3..3860c36e9 100644 --- a/packages/webui/tests/chat.test.mjs +++ b/packages/webui/tests/chat.test.mjs @@ -5,6 +5,12 @@ import { createChat } from '../src/chat.mjs'; import { startServer } from '../src/serve.mjs'; import { BusReadError } from '../src/bus.mjs'; import { close } from './fixture.mjs'; +// Row 264a (#1769): a thread made by hand gets the default session the 3j-3k migration gives a +// legacy thread, with the thread's id. +const DEFAULT_SESSION = [ + 'INSERT INTO sessions(at,business,thread,id,instance,owner,name,op) SELECT at,business,id,id,instance,owner,'default','legacy-3k' FROM threads WHERE id=?', + 'INSERT INTO session_defaults(at,thread,session,by) SELECT at,id,id,'migration-3k' FROM threads WHERE id=?', +]; const P = '10000000-0000-4000-8000-000000000001'; const T = '20000000-0000-4000-8000-000000000002'; @@ -220,15 +226,17 @@ test('REQ-USER-2 real broker: owned catalog, create/name/filter, isolated histor const other = '40000000-0000-4000-8000-000000000004'; store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', other, new Date().toISOString(), 'demo', project, 'assistant', 'someone-else', 'PRIVATE-NAME'); + for (const sql of DEFAULT_SESSION) store.run(sql, other); store.run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)', M, new Date().toISOString(), 'demo', 'human', 'human-cli', 'assistant', 'INFO', 'PRIVATE-BODY'); - store.run('INSERT INTO message_threads(message,thread) VALUES(?,?)', M, other); + store.run('INSERT INTO message_threads(message,thread,session) VALUES(?,?,?)', M, other, other); const foreignProject = '50000000-0000-4000-8000-000000000005'; const foreignThread = '60000000-0000-4000-8000-000000000006'; store.run('INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', foreignProject, new Date().toISOString(), 'foreign', 'FOREIGN-PROJECT', null, 'jason'); store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', foreignThread, new Date().toISOString(), 'foreign', foreignProject, 'assistant', 'jason', 'FOREIGN-THREAD'); + for (const sql of DEFAULT_SESSION) store.run(sql, foreignThread); assert.equal((await get('/api/chat/threads')).data.rows.length, 2); const before = ['messages', 'thread_opens', "events WHERE kind='action.allowed'"].map(table => store.get(`SELECT count(*) n FROM ${table}`).n); for (const id of [other, T, foreignThread]) for (const suffix of ['messages', 'open']) { diff --git a/scripts/stack-release.sh b/scripts/stack-release.sh index f0c144fe5..aed858ca7 100755 --- a/scripts/stack-release.sh +++ b/scripts/stack-release.sh @@ -401,7 +401,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 thread_opens instance_names usage_records" +COUNTED="role_claims thread_claims decision_events messages deliveries task_snapshots projects threads sessions session_defaults thread_opens instance_names usage_records" STORE_JS=' const {DatabaseSync}=require("node:sqlite");const {pathToFileURL}=require("node:url");