diff --git a/packages/bus/README.md b/packages/bus/README.md index 4694e08b5..b16d72eb1 100644 --- a/packages/bus/README.md +++ b/packages/bus/README.md @@ -159,6 +159,57 @@ an operator inspect content and delivery evidence after an uncertain result. | `requirements` | No args; role sessions only, as `task.create` is (row 90, #1584). The approved PRD's requirement ids, `{prd,version,requirements:[{id,title}]}`, from `packages/tasks/requirements.json`, loaded when the broker starts; a bad file refuses the start. A human session refuses `agent-required`, a reader `read-only`. | | `state` | No args; role, human and reader sessions (row 109, #1603). `{launch:{state,at,event},tasks:{open:[{task_ref,title}],total,cap}}` for the capability's business. `launch` is the latest `launch.revoked` or `launch.restored` event: `allowed` with `at` and `event` null when none exists. `open` is the tracker snapshot's open tasks in `task_ref` order, at most `cap` (the `tasks` page size, 50); `total` counts every open task. The harness runner reads it before each turn. Writes nothing. | | `replies` | `{id}`: a message UUID; human and reader sessions only. The messages whose `in_reply_to` is that id, oldest first, at most 50. Writes nothing. | +| `project.create` | `{name,repo?}`; human sessions only (row 121, #1615). `repo` must name one of the business's `projects` (`unknown-repo`); a second project with the same name refuses `project-exists`. Returns `{id}`. | +| `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. | + +### Projects and threads (row 121, #1615) + +A thread is one conversation between a user and the business's thread instance, the role +whose definition is `assistant` (REQ-THREAD-1). Projects group threads (REQ-ORG-2). Schema +`3e` adds five append-only tables: `projects`, `threads`, `thread_claims`, `message_threads` +and `thread_opens`. A message's thread is a `message_threads` row written with the message, +so `messages` keeps its old columns. + +A thread instance's run is for one thread. `bindLaunch` refuses a record for that role +without `thread` (`thread-required`) or naming a thread of another instance or business +(`thread-not-found`), and a record for any other role that names one (`thread-not-allowed`). +The run's `role.claim` writes a `thread_claims` row, not a `role_claims` row. One run holds +a thread at a time (`thread-already-held`), and a run holds at most one thread. An idle +thread has no run and holds nothing, so it counts against no launch limit (REQ-THREAD-2). + +Every message to or from a thread instance names its thread (REQ-MSG-1): + +- A thread run's own message is in its thread. +- A human reaches a thread instance only through `thread.send`. A `message.send` to it + refuses `thread-required-use-thread-send`. +- Any other role's message to a thread instance needs `in_reply_to`. The broker walks the + `in_reply_to` chain to the nearest message that has a thread, so the chief of staff's + reply returns to the thread that sent the work. The walk stops where the chain ends; there + is no depth constant. No threaded ancestor refuses `thread-required-reply-to-thread-message`, + which is the whole text a run sees, so it says what to do. A threaded ancestor in another + instance's thread refuses `thread-instance-mismatch`. +- A thread run's `message.receive` returns only its own thread's messages. + +Triggers back the broker. `thread_one_holder` and `thread_one_per_run` refuse a second +holder and a second thread. `message_threads_match` refuses filing a message under a thread +of another business, or under a thread whose instance neither sent nor receives it. +`delivery_same_thread` refuses routing or delivering a threaded message to the instance to +any run but the thread's current holder. + +A user reaches only their own threads (REQ-USER-2): `thread.open`, `thread.send` and +`thread.messages` refuse `thread-not-found` for an unknown id and `thread-not-owned` for +another user's thread. + +`threadsState(business)` is a trusted launcher API, like `bindLaunch`, not a socket verb. +Per thread it returns `{thread,instance,pending,unread,lastMessage,lastOpen,lastClaim,holder}`: +messages to the instance not yet delivered, messages delivered and not yet read, the +latest message, open and claim times, and the holding run or `null`. The launcher reads it +to start and stop thread sessions (`packages/cli/README.md`). ### Agent state (row 59, #1548) diff --git a/packages/bus/schema.sql b/packages/bus/schema.sql index 6aef89004..26157ff1d 100644 --- a/packages/bus/schema.sql +++ b/packages/bus/schema.sql @@ -209,3 +209,95 @@ CREATE VIEW urgent_inbox AS 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 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; diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index f10e11b37..fded7106b 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -97,6 +97,14 @@ const VERDICTS = ['approve', 'changes']; // Row 93 (#1587): the review loop's evidence, written only by the verbs below. A push is an // action.allowed for git.push.working that names its commit, written only by recordPush. const REVIEW_KINDS = ['review.requested', 'review.verdict']; +// Row 121 (#1615): an instance of this definition talks with the user in threads (REQ-THREAD-1). +// Its runs claim a thread, not the role, and every message to or from it names its thread (REQ-MSG-1). +const THREAD_DEFINITION = 'assistant'; +// How many messages `thread.messages` returns by default and at most, oldest first. +export const THREAD_PAGE = 50; +export const THREAD_PAGE_MAX = 200; +// A project or thread name: what the user calls it, one line. +const name = (x) => text(x, 200) && !/[\r\n]/.test(x); export class Broker { #store; @@ -127,6 +135,10 @@ export class Broker { 'messages', 'deliveries', 'task_snapshots', + 'projects', + 'threads', + 'thread_claims', + 'thread_opens', ]) { const at = store.get(`SELECT max(at) at FROM ${table}`).at; if (at) { @@ -181,11 +193,61 @@ export class Broker { role, ); } + #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); + } + // The claim a run acts under: its thread's for a thread run, its role's otherwise. + #own(s) { + return s.thread ? this.#threadHolder(s.thread) : 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); + 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 + // 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; + 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'); + const seen = new Set(); + for (let at = inReplyTo ?? null; at !== null && !seen.has(at); ) { + seen.add(at); + const m = this.#store.get( + 'SELECT m.in_reply_to, t.thread 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 m.thread; + } + at = m?.in_reply_to ?? null; + } + // The code is the whole message a run sees (runner.mjs forwards codes only), so it says what to do. + fail('thread-required-reply-to-thread-message'); + } + #thread(s, id) { + const t = this.#store.get('SELECT * FROM threads WHERE id=? AND business=?', identifier(id), s.business); + if (!t) fail('thread-not-found'); + // REQ-USER-2: a user reaches only their own threads. + if (t.owner !== s.human) fail('thread-not-owned'); + return t; + } #agent(s) { if (s.tester) fail('tester-refused'); if (s.reader) fail('read-only'); if (s.human) fail('agent-required'); - const h = this.#holder(s.business, s.role); + const h = this.#own(s); if (h?.op !== 'claim' || h.holder_run !== s.run) fail('not-holder'); return h; } @@ -198,11 +260,43 @@ export class Broker { this.#sessions.set(cap, Object.freeze(s)); return cap; } + // Row 121 (#1615): what the launcher needs to start, stop and recover thread sessions. + // Trusted launcher API, not a socket verb. Per thread: messages to the instance that no + // run has been delivered yet, messages delivered and not yet read (a turn in progress), + // and the latest message, open and claim times. + threadsState(business) { + this.#business(identifier(business)); + 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 + 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 + AND EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op='delivered') + 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, + 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) + WHERE t.business=? ORDER BY t.seq`, + business, + ) + .map((r) => ({ + thread: r.id, + instance: r.instance, + pending: r.pending, + unread: r.unread, + lastMessage: r.last_message, + lastOpen: r.last_open, + lastClaim: r.claim_at, + holder: r.claim_op === 'claim' ? r.holder_run : null, + })); + } // Trusted launcher API, deliberately NOT a socket verb. S6 supplies immutable launch records. bindLaunch(record) { keys( record, - ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime'], + ['business', 'role', 'run', 'harness', 'address', 'pid', 'startTime', 'thread'], ['business', 'role', 'run', 'harness'], ); const b = this.#business(identifier(record.business)); @@ -210,20 +304,28 @@ export class Broker { identifier(record.run); if (!['pi', 'claude-code'].includes(record.harness)) fail('invalid-harness'); if (record.address !== undefined) string(record.address, 1024); + // Row 121: a thread instance's run is for one thread, and no other run names one. + if (this.#threaded(record.business, record.role)) { + 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'); const key = record.business + ':' + record.run; if (this.#runs.has(key)) fail('duplicate-run'); - const session = { ...record, address: record.address ?? null, human: false }; + const session = { ...record, address: record.address ?? null, thread: record.thread ?? null, 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", record.business, record.run, ); + // A run without a thread keeps the body it had before row 121. const body = { harness: record.harness, address: record.address ?? null, pid: record.pid ?? null, startTime: record.startTime ?? null, + ...(record.thread === undefined ? {} : { thread: record.thread }), }; if (old) { if (old.actor_role !== record.role || old.body !== JSON.stringify(body)) fail('launch-record-mismatch'); @@ -250,7 +352,7 @@ export class Broker { for (const [cap, s] of this.#sessions) if (!s.human && !s.reader && s.business === record.business && s.run === record.run) this.#sessions.delete(cap); return this.#store.transaction(() => { - const h = this.#holder(session.business, session.role); + const h = this.#own(session); const released = h?.op === 'claim' && h.holder_run === session.run; if (released) this.#claimEnd(session, h, 'release', null); this.#event(session, 'session.ended', { reason: record.reason, exitCode, released }, session.run); @@ -753,6 +855,25 @@ export class Broker { case 'role.claim': { keys(a, []); if (s.human) fail('agent-required'); + // Row 121: a thread run claims its thread, so idle threads hold nothing (REQ-THREAD-2). + 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)) + fail('run-revoked'); + if (this.#threadHolder(s.thread)?.op === 'claim') fail('thread-already-held'); + this.#store.run( + 'INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,address,by) VALUES(?,?,?,?,?,?,?,?)', + this.#now(), + s.business, + s.thread, + 'claim', + s.run, + s.harness, + s.address, + by, + ); + this.#routeThread(s.thread); + return { role: s.role, run: s.run, thread: s.thread }; + } if ( this.#store.get( "SELECT 1 FROM role_claims WHERE business=? AND role=? AND holder_run=? AND op='revoke' LIMIT 1", @@ -943,6 +1064,15 @@ 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 + // raised before row 121 has no thread claim and stays unthreaded. + const thread = 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", + s.business, + d.raised_by_run, + )?.thread ?? null) + : null; 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(?,?,?,?,?,?,?,?,?,?,?)', @@ -958,8 +1088,7 @@ export class Broker { null, `Decision ${d.id} on ${d.action} was ${outcome}.${a.note === undefined ? '' : `\nNote: ${a.note}`}`, ); - const h = this.#holder(s.business, d.raised_by_role); - this.#delivery(mid, 'routed', h?.op === 'claim' ? h : null); + this.#route(s.business, mid, d.raised_by_role, thread); } return { id: d.id, op }; } @@ -1012,6 +1141,7 @@ export class Broker { } } if (a.decision) this.#decision(s, a.decision); + const thread = this.#threadFor(s, a.to, a.in_reply_to); 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(?,?,?,?,?,?,?,?,?,?,?)', @@ -1027,8 +1157,7 @@ export class Broker { a.corrects ?? null, a.body, ); - const h = this.#holder(s.business, a.to); - this.#delivery(mid, 'routed', h?.op === 'claim' ? h : null); + this.#route(s.business, mid, a.to, thread); const request = s.human ? this.#event(s, 'human.input', { kind: 'instruction', message: mid }, mid) : null; @@ -1039,16 +1168,20 @@ export class Broker { if (s.human) this.#human(s); else if (!s.tester) this.#agent(s); const role = s.human ? 'human' : s.role; + // A thread run receives its own thread's messages only (REQ-THREAD-1). 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')) ORDER BY seq LIMIT 100", + "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=?)' : '') + + ' ORDER BY seq LIMIT 100', s.business, role, + ...(s.thread ? [s.thread] : []), ); for (const m of rows) this.#delivery( m.id, 'delivered', - s.human || s.tester ? null : this.#holder(s.business, s.role), + s.human || s.tester ? null : this.#own(s), s.human || s.tester ? 'cli' : 'broker', ); return rows.map((m) => ({ ...m, request: this.#request(s.business, m) })); @@ -1065,7 +1198,7 @@ export class Broker { ); if (!delivery || delivery.holder_run !== (s.run ?? null)) fail('not-recipient'); if (!this.#store.get("SELECT seq FROM deliveries WHERE message=? AND op='read'", m.id)) - this.#delivery(m.id, 'read', s.human || s.tester ? null : this.#holder(s.business, s.role)); + this.#delivery(m.id, 'read', s.human || s.tester ? null : this.#own(s)); return { read: true }; } // Row 93 (#1587): the coder asks a reviewer role to review the task's latest pushed commit. @@ -1300,6 +1433,123 @@ export class Broker { // reader sessions only, like decisions. It writes no delivery or read row, so a reply it // lists still reaches the next `mosaic talk`. An id naming no message in the business // refuses message-not-found, as message.send does for in_reply_to; trail would return []. + // Row 121 (#1615): projects group the user's threads (REQ-ORG-2). `repo` names a + // project from the business file; the rename to repos is a P19 follow-up. + case 'project.create': { + this.#human(s); + keys(a, ['name', 'repo'], ['name']); + if (!name(a.name)) fail('invalid-request'); + if (a.repo !== undefined && !Object.hasOwn(b.projects ?? {}, identifier(a.repo))) fail('unknown-repo'); + if (this.#store.get('SELECT 1 FROM projects WHERE business=? AND name=?', s.business, a.name)) fail('project-exists'); + const pid = randomUUID(); + this.#store.run( + 'INSERT INTO projects(id,at,business,name,repo,created_by) VALUES(?,?,?,?,?,?)', + pid, + this.#now(), + s.business, + a.name, + a.repo ?? null, + s.human, + ); + return { id: pid }; + } + case 'project.list': { + this.#human(s); + keys(a, []); + return this.#store.all('SELECT id,at,name,repo FROM projects WHERE business=? ORDER BY seq', s.business); + } + // A thread belongs to the user who made it and to the business's one thread instance. + case 'thread.create': { + this.#human(s); + keys(a, ['project', 'name'], ['project', 'name']); + if (!name(a.name)) fail('invalid-request'); + if (!this.#store.get('SELECT 1 FROM projects WHERE id=? AND business=?', identifier(a.project), s.business)) + fail('project-not-found'); + // The bus never picks between instances; the business file declares exactly one. + const instances = Object.keys(b.roles).filter((r) => this.#threaded(s.business, r)); + if (!instances.length) fail('thread-instance-missing'); + if (instances.length > 1) fail('thread-instance-ambiguous'); + if (this.#store.get('SELECT 1 FROM threads WHERE project=? AND name=?', a.project, a.name)) fail('thread-exists'); + const tid = randomUUID(); + this.#store.run( + 'INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', + tid, + this.#now(), + s.business, + a.project, + instances[0], + s.human, + a.name, + ); + return { id: tid, instance: instances[0] }; + } + case 'thread.list': { + this.#human(s); + keys(a, ['project']); + if (a.project !== undefined) identifier(a.project); + return this.#store.all( + 'SELECT id,at,project,instance,name FROM threads WHERE business=? AND owner=?' + + (a.project === undefined ? '' : ' AND project=?') + + ' ORDER BY seq', + s.business, + s.human, + ...(a.project === undefined ? [] : [a.project]), + ); + } + // Opening a thread asks the launcher to start its session (REQ-LAUNCH-1). + case 'thread.open': { + this.#human(s); + keys(a, ['thread'], ['thread']); + const t = this.#thread(s, a.thread); + this.#store.run('INSERT INTO thread_opens(at,thread,by) VALUES(?,?,?)', this.#now(), t.id, s.human); + return { opened: t.id }; + } + case 'thread.send': { + this.#human(s); + keys(a, ['thread', 'body', 'class'], ['thread', 'body']); + const t = this.#thread(s, a.thread); + string(a.body, 32768); + if (a.class !== undefined && !['REQUEST', 'INFO', 'REACTION'].includes(a.class)) fail('invalid-class'); + 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(?,?,?,?,?,?,?,?,?,?,?)', + mid, + this.#now(), + s.business, + 'human', + 'human-cli', + t.instance, + a.class ?? 'INFO', + null, + null, + null, + a.body, + ); + this.#route(s.business, mid, t.instance, t.id); + return { id: mid, request: this.#event(s, 'human.input', { kind: 'instruction', message: mid }, mid) }; + } + // The thread's history, oldest first. Reading writes nothing. + case 'thread.messages': { + this.#human(s); + keys(a, ['thread', 'after', 'limit'], ['thread']); + const t = this.#thread(s, a.thread); + 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=?', + identifier(a.after), + t.id, + )?.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, + after, + ); + } case 'replies': { if (!s.human && !s.reader) fail('human-required'); keys(a, ['id'], ['id']); @@ -1348,6 +1598,19 @@ export class Broker { return { ...c, state, working, inputNeeded, stranded }; } #claimEnd(s, h, op, decision) { + if (h.thread) + return this.#store.run( + 'INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,address,by,decision) VALUES(?,?,?,?,?,?,?,?,?)', + this.#now(), + s.business, + h.thread, + op, + h.holder_run, + h.harness, + h.address, + s.run ?? s.human, + decision, + ); this.#store.run( 'INSERT INTO role_claims(at,business,role,op,holder_run,harness,address,by,decision) VALUES(?,?,?,?,?,?,?,?,?)', this.#now(), @@ -1381,6 +1644,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); this.#store.run( 'INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,in_reply_to,decision,corrects,body) VALUES(?,?,?,?,?,?,?,?,?,?,?)', mid, @@ -1395,8 +1659,12 @@ export class Broker { null, body, ); - const h = this.#holder(s.business, to); - this.#delivery(mid, 'routed', h?.op === 'claim' ? h : null); + 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)); } #delivery(mid, op, h, transport = null) { this.#store.run( @@ -1418,6 +1686,15 @@ 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); + 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, + ); + for (const m of rows) this.#delivery(m.id, 'routed', h); + } #trail(s, subject) { const rows = new Map(); const add = (table, list) => { diff --git a/packages/bus/src/process.mjs b/packages/bus/src/process.mjs index 0cb3f3e05..a2183b7fa 100644 --- a/packages/bus/src/process.mjs +++ b/packages/bus/src/process.mjs @@ -11,6 +11,7 @@ const LAUNCH_OPS = new Set([ 'checkPush', 'recordPush', 'reviewFetch', + 'threadsState', ]); function launchOp(m) { if (m.op === 'identity') return runtime.identity(m.cap); @@ -21,6 +22,8 @@ function launchOp(m) { if (m.op === 'checkPush') return runtime.checkPush(m.cap, m.task_ref); if (m.op === 'recordPush') return runtime.recordPush(m.cap, m.push); if (m.op === 'reviewFetch') return runtime.reviewFetch(m.cap, m.args); + // Row 121 (#1615): the launcher's view of the business's threads. + if (m.op === 'threadsState') return runtime.threadsState(m.business); return runtime.credentialStatus(m.business, m.role); } // The host's request id comes back on the reply, so a late reply can't diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index c28716de4..39c0674f7 100644 --- a/packages/bus/src/runtime.mjs +++ b/packages/bus/src/runtime.mjs @@ -119,6 +119,7 @@ export async function startBroker({ path, bindLaunch, endLaunch: (r) => broker.endLaunch(r), + threadsState: (business) => broker.threadsState(business), identity: (cap) => broker.identity(cap), authorizeLaunch, checkPush, diff --git a/packages/bus/src/store.mjs b/packages/bus/src/store.mjs index 755e43041..2258e2519 100644 --- a/packages/bus/src/store.mjs +++ b/packages/bus/src/store.mjs @@ -13,7 +13,9 @@ 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 = '179ffe356d4ff19a49b5ebad39b6c6bfd7771deb8e1b7c55b5e746f039d69e65'; +export const SCHEMA_SHA256 = '7f0d07248f47ffdc4737161e7b76db98ddcb40fb1ae9b3f24851219a428ec14f'; +// Provisional until rows 49 and 123 land: then 3e is the step after their 3d. +export const SCHEMA_VERSION = '3e'; const timedTables = [ 'events', 'role_claims', @@ -22,6 +24,10 @@ const timedTables = [ 'messages', 'deliveries', 'task_snapshots', + 'projects', + 'threads', + 'thread_claims', + 'thread_opens', ]; const hex = (value) => createHash('sha256').update(value).digest('hex'); if (hex(ddl) !== SCHEMA_SHA256) throw Error('schema-source-mismatch'); @@ -104,13 +110,13 @@ export class Store { this.#db.exec('PRAGMA journal_mode=WAL'); this.transaction(() => { this.#db.exec(ddl.replace('PRAGMA journal_mode = WAL;', '')); - this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', '3b'); + this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_version', SCHEMA_VERSION); this.run('INSERT INTO meta(key,value) VALUES (?,?)', 'schema_digest', EXPECTED_DIGEST); }); } if ( schemaDigest(this.#db) !== EXPECTED_DIGEST || - this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== '3b' || + this.get("SELECT value FROM meta WHERE key='schema_version'")?.value !== SCHEMA_VERSION || this.get("SELECT value FROM meta WHERE key='schema_digest'")?.value !== EXPECTED_DIGEST || this.get('PRAGMA journal_mode').journal_mode !== 'wal' ) diff --git a/packages/bus/tests/end-launch.test.mjs b/packages/bus/tests/end-launch.test.mjs index 68720d711..acfcd86ef 100644 --- a/packages/bus/tests/end-launch.test.mjs +++ b/packages/bus/tests/end-launch.test.mjs @@ -142,6 +142,13 @@ test('broker process: launch ops authorize role.launch, record refusals and end const child = launch(t); const data = join(root, 'data'); mkdirSync(data); + // Row 121 (#1615, REQ-LAUNCH-1): a thread for the launcher's thread view below, written + // before the child takes the store's writer lock. The human verbs need an assistant instance. + 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'); + side.close(); const ready = await ask(child, { op: 'boot', config: { dataRoot: data, businesses: config, launches: [], readers: [] }, @@ -173,6 +180,10 @@ test('broker process: launch ops authorize role.launch, record refusals and end { service: 'vikunja', date: '2099-01-01', state: 'valid' }, ]); assert.deepEqual((await ask(child, { op: 'credentialStatus', business: 'demo', role: 'pm' })).result, []); + // The launcher's thread view, through the same child. + assert.deepEqual((await ask(child, { op: 'threadsState', business: 'demo' })).result, [ + { thread: 'th-1', instance: 'coder', pending: 0, unread: 0, lastMessage: null, lastOpen: '2026-10-15T00:00:01.000Z', lastClaim: null, holder: null }, + ]); const end = await ask(child, { op: 'endLaunch', record: { business: 'demo', run: 'pm-1', reason: 'stopped', exitCode: 0 } }); assert.deepEqual(end, { ok: true, result: { released: true } }); // The host's request id comes back on every launch-op and bind reply, refusals too. diff --git a/packages/bus/tests/messaging.test.mjs b/packages/bus/tests/messaging.test.mjs index a1d6f9e53..b1b762e28 100644 --- a/packages/bus/tests/messaging.test.mjs +++ b/packages/bus/tests/messaging.test.mjs @@ -64,21 +64,26 @@ function setup(t, businesses = demo()) { }); const b = new Broker({ store, businesses }); const caps = {}; + caps.human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }); + // Row 121 (#1615): the assistant runs for one thread, and the user writes to it with thread.send. + const project = b.request(caps.human, { verb: 'project.create', args: { name: 'demo' } }).id; + const thread = b.request(caps.human, { verb: 'thread.create', args: { project, name: 'main' } }).id; for (const role of Object.keys(roles)) { - caps[role] = b.bindLaunch({ business: 'demo', role, run: `${role}-run`, harness: 'pi' }); + caps[role] = b.bindLaunch({ business: 'demo', role, run: `${role}-run`, harness: 'pi', ...(role === 'assistant' ? { thread } : {}) }); b.request(caps[role], { verb: 'role.claim' }); } - caps.human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true }); caps.tester = b.bindTester({ business: 'demo' }); const call = (who, verb, args = {}) => b.request(caps[who], { verb, args }); const send = (from, to, extra = {}) => - call(from, 'message.send', { to, body: from === 'tester' ? '[mos test] hello' : 'hello', ...extra }); + from === 'human' && to === 'assistant' + ? call(from, 'thread.send', { thread, body: 'hello', ...extra }) + : call(from, 'message.send', { to, body: from === 'tester' ? '[mos test] hello' : 'hello', ...extra }); const count = (sql) => store.get(sql).n; const refusals = () => store .all("SELECT actor_role, body FROM events WHERE kind='action.refused' ORDER BY seq") .map((e) => [e.actor_role, JSON.parse(e.body).code]); - return { b, store, caps, call, send, count, refusals }; + return { b, store, caps, call, send, count, refusals, thread }; } const RECIPIENT = 'message-recipient-refused'; @@ -118,8 +123,10 @@ test('an assistant send to any other role, or to the tester, is refused and logg test('only the user and the chief of staff may send to the assistant; the tester is refused by default', (t) => { const { send, store, count, refusals } = setup(t); + // The chief writes to the assistant in reply to the assistant's own message (row 121). + const opened = send('assistant', 'chief'); for (const from of ['human', 'chief']) { - const sent = send(from, 'assistant'); + const sent = send(from, 'assistant', from === 'chief' ? { in_reply_to: opened.id } : {}); assert.equal(store.get('SELECT to_role FROM messages WHERE id=?', sent.id).to_role, 'assistant'); } const messages = count('SELECT count(*) n FROM messages'); @@ -182,7 +189,9 @@ test('a business with no role definitions at all is unlimited', (t) => { const businesses = demo(); delete businesses.demo.definitions; const { send, refusals } = setup(t, businesses); - for (const [from, to] of [['assistant', 'pm'], ['pm', 'assistant'], ['tester', 'assistant']]) send(from, to); + // A send to the assistant still answers a message in its thread (row 121). + const opened = send('assistant', 'pm'); + for (const from of ['pm', 'tester']) send(from, 'assistant', { in_reply_to: opened.id }); assert.deepEqual(refusals(), []); }); diff --git a/packages/bus/tests/schema.test.mjs b/packages/bus/tests/schema.test.mjs index 4d00d56f4..590f6fcd8 100644 --- a/packages/bus/tests/schema.test.mjs +++ b/packages/bus/tests/schema.test.mjs @@ -516,7 +516,8 @@ test('v3b prototype refusals, views and append-only mutations', (t) => { const count = (type) => db.prepare('SELECT count(*) n FROM sqlite_master WHERE type = ?').get(type).n; for (const label of refused) assert.ok(tested.has(label), label); - assert.equal(count('table') - 1, 8); - assert.equal(count('trigger'), 35); + // Row 121 (#1615) adds 5 tables and 24 triggers for projects and threads; threads.test.mjs tests them. + assert.equal(count('table') - 1, 13); + assert.equal(count('trigger'), 59); assert.equal(count('view'), 5); }); diff --git a/packages/bus/tests/threads.test.mjs b/packages/bus/tests/threads.test.mjs new file mode 100644 index 000000000..6b96d9f86 --- /dev/null +++ b/packages/bus/tests/threads.test.mjs @@ -0,0 +1,418 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { DatabaseSync } from 'node:sqlite'; +import { mkdtempSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { Store } from '../src/store.mjs'; +import { Broker, THREAD_PAGE } from '../src/broker.mjs'; + +// 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. +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: [] } }, + coder: { definition: 'coder', authority: { withinRole: ['message.send'], crossRole: [] } }, +}; +const definitions = { + 'chief-of-staff': {}, + assistant: { messaging: shipped('assistant').messaging }, + pm: {}, + coder: {}, +}; +const demo = (extra = {}) => ({ + demo: { + id: 'demo', + human: 'jason', + arbiters: { technical: 'pm', delivery: 'chief' }, + roles, + definitions, + projects: { stack: { root: '/nonexistent' } }, + ...extra, + }, +}); + +function setup(t, businesses = demo()) { + const root = mkdtempSync(join(tmpdir(), 'bus-threads-')); + const store = new Store(root); + t.after(() => { + store.close(); + rmSync(root, { recursive: true, force: true }); + }); + 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 count = (sql, ...args) => store.get(sql, ...args).n; + const threadOf = (message) => store.get('SELECT thread FROM message_threads WHERE message=?', message)?.thread ?? null; + return { b, store, caps, call, launch, claim, count, threadOf }; +} + +// Two threads in one project, each with its own assistant run holding it. +function twoThreads(t, businesses) { + const ctx = setup(t, businesses); + const project = ctx.call('human', 'project.create', { name: 'stack work', repo: 'stack' }).id; + const A = ctx.call('human', 'thread.create', { project, name: 'alpha' }).id; + const B = ctx.call('human', 'thread.create', { project, name: 'beta' }).id; + for (const r of ['chief', 'pm', 'coder']) { + ctx.launch(r, r); + ctx.claim(r); + } + ctx.launch('a', 'assistant', { thread: A }); + ctx.launch('b', 'assistant', { thread: B }); + ctx.claim('a'); + ctx.claim('b'); + return { ...ctx, project, A, B }; +} + +const now = () => new Date().toISOString(); + +test('REQ-ROLE-2 REQ-THREAD-1: one holder per thread, through the broker and the trigger', (t) => { + 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. + const second = () => + store.run( + "INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by) VALUES(?,?,?,'claim',?,?,?)", + now(), + 'demo', + A, + 'a2-run', + 'pi', + 'a2-run', + ); + assert.throws(second, /thread already held/); + // Second case: one run holds one thread. a-run already holds A. + call('b', 'role.release'); + assert.throws( + () => + store.run( + "INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by) VALUES(?,?,?,'claim',?,?,?)", + now(), + 'demo', + B, + 'a-run', + 'pi', + 'a-run', + ), + /run already holds a thread/, + ); + // 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 }); + assert.deepEqual(claim('b2'), { role: 'assistant', run: 'b2-run', thread: B }); + assert.equal(store.get("SELECT count(*) n FROM role_claims WHERE role='assistant'").n, 0, 'no role claim for the instance'); + // endLaunch releases the thread claim. + b.endLaunch({ business: 'demo', run: 'b2-run', reason: 'idle' }); + assert.equal(store.get('SELECT op FROM thread_claims WHERE thread=? ORDER BY seq DESC LIMIT 1', B).op, 'release'); +}); + +test('REQ-ROLE-2 REQ-THREAD-1: the 3e triggers refuse writes that skip the broker', (t) => { + // A bare database from schema.sql, as schema.test.mjs does, so nothing but the triggers stands + // between a write and the tables. + const root = mkdtempSync(join(tmpdir(), 'bus-threads-sql-')); + const db = new DatabaseSync(join(root, 'bus.sqlite')); + t.after(() => { + db.close(); + rmSync(root, { recursive: true, force: true }); + }); + db.exec(readFileSync(new URL('../schema.sql', import.meta.url), 'utf8')); + const run = (sql, ...args) => db.prepare(sql).run(...args); + const claim = (op, run_, extra = {}) => + run( + 'INSERT INTO thread_claims(at,business,thread,op,holder_run,harness,by,reason,decision) VALUES(?,?,?,?,?,?,?,?,?)', + now(), + extra.business ?? 'demo', + extra.thread ?? 'th-1', + op, + run_, + 'pi', + run_, + extra.reason ?? null, + null, + ); + 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 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')"); + // 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()), + /a thread belongs to its project's business/, + ); + // 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/); + 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. + 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'], + }; + for (const [table, [where, col, replace]] of Object.entries(rows)) { + const only = new RegExp(`${table} is append-only`); + assert.throws(() => db.exec(`UPDATE ${table} SET ${col}=${col} WHERE ${where}`), only, `${table} UPDATE`); + assert.throws(() => db.exec(`DELETE FROM ${table} WHERE ${where}`), only, `${table} DELETE`); + assert.throws(() => db.exec(replace), only, `${table} INSERT OR REPLACE`); + } +}); + +test('REQ-MSG-1: a message to or from the assistant names its thread, or is refused', (t) => { + const { b, store, call, launch, count, threadOf, A, B } = twoThreads(t); + const before = count('SELECT count(*) n FROM messages'); + // The chief writes to the assistant only in reply to thread work. + assert.throws(() => call('chief', 'message.send', { to: 'assistant', body: 'hi' }), { + code: 'thread-required-reply-to-thread-message', + }); + // The user writes to a thread with thread.send, which checks the owner. + assert.throws(() => call('human', 'message.send', { to: 'assistant', body: 'hi' }), { + code: 'thread-required-use-thread-send', + }); + assert.equal(count('SELECT count(*) n FROM messages'), before, 'a refused send stores nothing'); + // A run of the instance names its thread at bind, and no other run names one. + assert.throws(() => launch('x', 'assistant'), { code: 'thread-required' }); + assert.throws(() => launch('y', 'assistant', { thread: 'no-such-thread' }), { code: 'thread-not-found' }); + assert.throws(() => launch('z', 'pm', { thread: A }), { code: 'thread-not-allowed' }); + // Accepted messages: the user's, the assistant's to the user and the chief, a decision close. + const asked = call('human', 'thread.send', { thread: A, body: 'look at the build' }); + const toUser = call('a', 'message.send', { to: 'human', body: 'on it', in_reply_to: asked.id }); + const toChief = call('b', 'message.send', { to: 'chief', body: 'need a coder', class: 'REQUEST' }); + assert.equal(threadOf(asked.id), A); + assert.equal(threadOf(toUser.id), A); + assert.equal(threadOf(toChief.id), B); + // The decision close to the assistant goes to the thread the raising run held. + const d = call('b', 'decision.raise', { + action: 'spend', + target: 'demo', + question: 'spend?', + options: [ + { key: 'yes', text: 'Yes' }, + { key: 'no', text: 'No' }, + ], + recommendation: 'no', + blocking: false, + }); + call('human', 'decision.resolve', { id: d.id, choice: 'no' }); + const close = store.get("SELECT id FROM messages WHERE class='DECISION' AND decision=?", d.id); + assert.equal(threadOf(close.id), B); + assert.equal( + count( + "SELECT count(*) n FROM messages m WHERE 'assistant' IN (m.to_role, m.from_role) AND NOT EXISTS(SELECT 1 FROM message_threads t WHERE t.message=m.id)", + ), + 0, + 'every message to or from the assistant has its thread', + ); + // 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), + /to or from the thread's instance/, + ); + void b; +}); + +test('REQ-MSG-1: the chief of staff\'s reply comes back to the thread that sent the work', (t) => { + const { call, threadOf, A, B } = twoThreads(t); + // Each thread asks the chief, who hands the work to the coder and answers when the coder does. + const run = (who) => { + const ask = call(who, 'message.send', { to: 'chief', body: `work for ${who}`, class: 'REQUEST' }); + return ask; + }; + const askA = run('a'); + const askB = run('b'); + // The assistant's messages reach the chief through the broker; the trigger only binds messages to the instance. + const got = call('chief', 'message.receive').map((m) => m.id); + assert.deepEqual(got, [askA.id, askB.id]); + const assignB = call('chief', 'message.send', { to: 'coder', body: 'do B', class: 'ASSIGNMENT', in_reply_to: askB.id }); + const assignA = call('chief', 'message.send', { to: 'coder', body: 'do A', class: 'ASSIGNMENT', in_reply_to: askA.id }); + const doneA = call('coder', 'message.send', { to: 'chief', body: 'A done', class: 'RESULT', in_reply_to: assignA.id }); + const doneB = call('coder', 'message.send', { to: 'chief', body: 'B done', class: 'RESULT', in_reply_to: assignB.id }); + const replyB = call('chief', 'message.send', { to: 'assistant', body: 'B is done', class: 'RESULT', in_reply_to: doneB.id }); + const replyA = call('chief', 'message.send', { to: 'assistant', body: 'A is done', class: 'RESULT', in_reply_to: doneA.id }); + assert.equal(threadOf(replyA.id), A); + assert.equal(threadOf(replyB.id), B); + assert.deepEqual(call('a', 'message.receive').map((m) => m.id), [replyA.id]); + assert.deepEqual(call('b', 'message.receive').map((m) => m.id), [replyB.id]); + // A long chain that never loops: the walk follows it to the end, with no depth limit. + let at = call('a', 'message.send', { to: 'chief', body: 'long one', class: 'REQUEST' }).id; + for (let i = 0; i < 400; i++) + at = call(i % 2 ? 'chief' : 'coder', 'message.send', { + to: i % 2 ? 'coder' : 'chief', + body: `step ${i}`, + in_reply_to: at, + }).id; + const far = call('chief', 'message.send', { to: 'assistant', body: 'long one done', in_reply_to: at }); + assert.equal(threadOf(far.id), A); + // A coder message with no threaded ancestor breaks the chain, and the refusal says how to recover. + const loose = call('coder', 'message.send', { to: 'chief', body: 'unrelated', class: 'INFO' }); + assert.throws(() => call('chief', 'message.send', { to: 'assistant', body: 'fyi', in_reply_to: loose.id }), { + code: 'thread-required-reply-to-thread-message', + }); +}); + +test('REQ-MSG-1: a reply can\'t carry one thread instance\'s thread to another thread instance', (t) => { + // thread.create refuses a business with two thread instances, so the second instance's thread + // is made by hand. The trigger would refuse the message_threads row too; the broker refuses first, + // with a code that says why. + 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']]) + store.run('INSERT INTO threads(id,at,business,project,instance,owner,name) VALUES(?,?,?,?,?,?,?)', id, now(), 'demo', project, instance, 'jason', id); + launch('chief', 'chief'); + claim('chief'); + launch('a', 'assistant', { thread: 'th-a' }); + claim('a'); + const ask = call('a', 'message.send', { to: 'chief', body: 'work', class: 'REQUEST' }); + const before = count('SELECT count(*) n FROM messages'); + assert.throws(() => call('chief', 'message.send', { to: 'helper', body: 'done', in_reply_to: ask.id }), { + code: 'thread-instance-mismatch', + }); + assert.equal(count('SELECT count(*) n FROM messages'), before); + // The same reply to the thread's own instance is fine. + call('chief', 'message.send', { to: 'assistant', body: 'done', in_reply_to: ask.id }); +}); + +test('REQ-THREAD-1 REQ-USER-2: one thread\'s messages never reach another thread\'s holder', (t) => { + const { store, call, threadOf, A, B } = twoThreads(t); + const toA = call('human', 'thread.send', { thread: A, body: 'for A' }); + const toB = call('human', 'thread.send', { thread: B, body: 'for B' }); + assert.deepEqual(call('b', 'message.receive').map((m) => m.id), [toB.id]); + assert.deepEqual(call('a', 'message.receive').map((m) => m.id), [toA.id]); + assert.equal(threadOf(toA.id), A); + // A delivery of A's message to B's holder that skips the broker is refused by the store. + const toA2 = call('human', 'thread.send', { thread: A, body: 'again for A' }); + for (const op of ['routed', 'delivered']) + assert.throws( + () => + store.run( + 'INSERT INTO deliveries(message,at,op,holder_run,transport,address) VALUES(?,?,?,?,?,?)', + toA2.id, + now(), + op, + 'b-run', + 'pi', + null, + ), + /only to its thread's holder/, + op, + ); + // B's holder can't read A's message either. + assert.throws(() => call('b', 'message.read', { id: toA.id }), { code: 'not-recipient' }); +}); + +test('REQ-LAUNCH-1: a message to a thread with no holder waits, and the next holder gets it', (t) => { + const { b, call, launch, claim, A, B } = twoThreads(t); + b.endLaunch({ business: 'demo', run: 'a-run', reason: 'idle' }); + const waiting = call('human', 'thread.send', { thread: A, body: 'while idle' }); + const state = () => Object.fromEntries(b.threadsState('demo').map((s) => [s.thread, s])); + assert.equal(state()[A].pending, 1); + assert.equal(state()[A].holder, null); + assert.equal(state()[B].holder, 'b-run'); + assert.equal(state()[B].pending, 0); + launch('a2', 'assistant', { thread: A }); + claim('a2'); + assert.equal(state()[A].holder, 'a2-run'); + assert.deepEqual(call('a2', 'message.receive').map((m) => m.id), [waiting.id]); + assert.equal(state()[A].pending, 0); + assert.equal(state()[A].unread, 1, 'delivered and not read: a turn in progress'); + call('a2', 'message.read', { id: waiting.id }); + assert.equal(state()[A].unread, 0); + // An open is recorded for the launcher, with no message. + assert.equal(state()[B].lastOpen, null); + call('human', 'thread.open', { thread: B }); + assert.ok(state()[B].lastOpen); +}); + +test('REQ-USER-2: the human verbs refuse another owner\'s thread', (t) => { + const { store, call, project, A } = twoThreads(t); + 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', + ); + for (const [verb, args] of [ + ['thread.open', { thread: other }], + ['thread.send', { thread: other, body: 'hi' }], + ['thread.messages', { thread: other }], + ]) + assert.throws(() => call('human', verb, args), { code: 'thread-not-owned' }, verb); + assert.throws(() => call('human', 'thread.send', { thread: 'nope', body: 'hi' }), { code: 'thread-not-found' }); + assert.deepEqual( + call('human', 'thread.list').map((x) => x.id), + [A, store.get("SELECT id FROM threads WHERE name='beta'").id], + 'the list shows the user\'s own threads only', + ); + // Agents have no thread verbs. + assert.throws(() => call('pm', 'thread.list'), { code: 'human-required' }); +}); + +test('REQ-THREAD-1: thread.create refuses when the business has no thread instance or more than one', (t) => { + const none = demo({ roles: { ...roles, assistant: { ...roles.assistant, definition: 'pm' } } }); + const a = setup(t, none); + const p1 = a.call('human', 'project.create', { name: 'p' }).id; + assert.throws(() => a.call('human', 'thread.create', { project: p1, name: 'x' }), { code: 'thread-instance-missing' }); + const two = demo({ roles: { ...roles, helper: { ...roles.assistant } } }); + const c = setup(t, two); + const p2 = c.call('human', 'project.create', { name: 'p' }).id; + assert.throws(() => c.call('human', 'thread.create', { project: p2, name: 'x' }), { code: 'thread-instance-ambiguous' }); + assert.equal(a.count('SELECT count(*) n FROM threads'), 0); + assert.equal(c.count('SELECT count(*) n FROM threads'), 0); +}); + +test('REQ-ORG-2: projects and threads are named once, and the repo names a business project', (t) => { + const { call, project, count } = twoThreads(t); + assert.throws(() => call('human', 'project.create', { name: 'stack work' }), { code: 'project-exists' }); + assert.throws(() => call('human', 'project.create', { name: 'n', repo: 'missing' }), { code: 'unknown-repo' }); + assert.throws(() => call('human', 'project.create', { name: 'two\nlines' }), { code: 'invalid-request' }); + assert.throws(() => call('human', 'thread.create', { project, name: 'alpha' }), { code: 'thread-exists' }); + assert.throws(() => call('human', 'thread.create', { project: 'nope', name: 'x' }), { code: 'project-not-found' }); + assert.deepEqual( + call('human', 'project.list').map(({ name, repo }) => ({ name, repo })), + [{ name: 'stack work', repo: 'stack' }], + ); + assert.equal(count('SELECT count(*) n FROM projects'), 1); +}); + +test('REQ-THREAD-2: thread.messages returns the thread\'s history in pages and writes nothing', (t) => { + const { store, call, A, B } = twoThreads(t); + const sent = []; + for (let i = 0; i < THREAD_PAGE + 3; i++) sent.push(call('human', 'thread.send', { thread: A, body: `m${i}` }).id); + call('human', 'thread.send', { thread: B, body: 'other thread' }); + const writes = () => store.get('SELECT (SELECT count(*) FROM deliveries) + (SELECT count(*) FROM events) n').n; + const before = writes(); + const page = call('human', 'thread.messages', { thread: A }); + assert.deepEqual(page.map((m) => m.id), sent.slice(0, THREAD_PAGE)); + const rest = call('human', 'thread.messages', { thread: A, after: page.at(-1).id }); + assert.deepEqual(rest.map((m) => m.id), sent.slice(THREAD_PAGE)); + assert.deepEqual(call('human', 'thread.messages', { thread: A, limit: 2 }).map((m) => m.id), sent.slice(0, 2)); + assert.equal(writes(), before, 'reading the history writes nothing'); + for (const limit of [0, 201, 1.5, '5']) + assert.throws(() => call('human', 'thread.messages', { thread: A, limit }), { code: 'invalid-request' }, String(limit)); +}); diff --git a/packages/business/README.md b/packages/business/README.md index d2967cb33..4d7da01a5 100644 --- a/packages/business/README.md +++ b/packages/business/README.md @@ -100,6 +100,15 @@ instance whose definition holds `role.launch` within-role; otherwise the file refuses. Only that instance keeps `role.launch`. Every other instance loses it, so for them it's gated. +`launch.threadIdleSeconds` (optional, row 121) is how long a thread's +session runs with nothing pending and no new message or open before the +launcher stops it. Absent means 900 (`THREAD_IDLE_SECONDS` in +`src/business.mjs`). A set value is a positive integer with no upper bound; +anything else refuses the file. The key is accepted whether or not the file +declares a thread instance. A thread instance is one whose definition is +`assistant`. The launcher starts it once per thread, when the thread has a +pending message or was opened, and never on its own. + `plugins` (optional) switches on plug-in modules, one key per kind from `packages/plugins/src/kinds.mjs`, each a list of module names matching `^[a-z][a-z0-9-]*$` with no name twice. An unknown kind or a bad name diff --git a/packages/business/src/business.mjs b/packages/business/src/business.mjs index 49ac1fa45..e817ef36d 100644 --- a/packages/business/src/business.mjs +++ b/packages/business/src/business.mjs @@ -15,6 +15,9 @@ import { } from "./util.mjs"; const BOT_NAME = /^bot-[a-z0-9][a-z0-9._-]{0,60}$/; +// How long a thread's session stays up with no turn before the launcher stops +// it, when launch.threadIdleSeconds is absent (REQ-THREAD-2, row 121). +export const THREAD_IDLE_SECONDS = 900; const TOP_KEYS = ["businessVersion", "id", "human", "arbiters", "projects", "vars", "tracker", "roles", "launch", "plugins"]; // The config directory: next to the system config, $MOSAIC_CONFIG or @@ -115,7 +118,7 @@ function checkLaunch(launch, roles, definitions, file) { const where = `${file} launch`; requireObject(launch, where); if (Object.hasOwn(launch,"max")) refuse(`${where}.max needs migration: mosaic limits migrate --file FILE --models FILE`); - rejectUnknownKeys(launch, ["by", "instances", "models", "limits"], where); + rejectUnknownKeys(launch, ["by", "instances", "models", "limits", "threadIdleSeconds"], where); requireId(launch.by, `${where}.by`); if (!Object.hasOwn(roles, launch.by)) refuse(`${where}.by names ${launch.by}, which the business file doesn't declare`); const launcher = definitions[roles[launch.by].definition]; @@ -130,7 +133,9 @@ function checkLaunch(launch, roles, definitions, file) { const models=checkModels(launch.models,`${where}.models`); const limits=checkLimits(launch.limits===undefined?{}:launch.limits,`${where}.limits`); for(const [instance,role] of Object.entries(roles))if(role.vars.model!==undefined&&!Object.hasOwn(models,role.vars.model))refuse(`${where}: instance ${instance} selects unknown model ${role.vars.model}`); - return { by: launch.by, instances, models, limits }; + // No upper bound: limits have no ceiling in code (Q150). + const threadIdleSeconds = launch.threadIdleSeconds === undefined ? THREAD_IDLE_SECONDS : requirePositiveInt(launch.threadIdleSeconds, `${where}.threadIdleSeconds`); + return { by: launch.by, instances, models, limits, threadIdleSeconds }; } // Validate a parsed business document read from `file`. `rolesDir` is where diff --git a/packages/business/tests/business.test.mjs b/packages/business/tests/business.test.mjs index 48e181f26..152c3e0dd 100644 --- a/packages/business/tests/business.test.mjs +++ b/packages/business/tests/business.test.mjs @@ -5,6 +5,7 @@ import { chmodSync, lstatSync, mkdirSync, readdirSync, readFileSync, symlinkSync import { join } from "node:path"; import { pathToFileURL } from "node:url"; import { BusinessError, businessFilePath, configDir, loadBusiness, validateBusinessDocument } from "../src/index.mjs"; +import { THREAD_IDLE_SECONDS } from "../src/business.mjs"; import { launchDoc, businessDoc, REPO, REPO_ROLES, rolesCopy, tmp, writeJson } from "./helpers.mjs"; function refuses(fn, pattern, exitCode = 2) { @@ -39,7 +40,7 @@ test("the fixture business validates and comes back frozen", () => { assert.equal(business.roles.reviewer.holder, null); assert.equal(business.roles.coder.vars["limits.network"], "none"); assert.equal(business.definitions.pm.roleVersion, 2); - assert.deepEqual(business.launch, launchDoc()); + assert.deepEqual(business.launch, { ...launchDoc(), threadIdleSeconds: THREAD_IDLE_SECONDS }); assert.ok(Object.isFrozen(business.roles.pm.credentials.gitea)); assert.throws(() => { business.roles.pm.holder = "x"; }, TypeError); }); @@ -134,6 +135,19 @@ test("launch", () => { assert.deepEqual(check((d) => { d.launch.limits = {total:0}; }).launch.limits, {total:0}); }); +// REQ-THREAD-2 (row 121, test 10): an absent threadIdleSeconds is the one +// exported default, a set value is used, and anything but a positive integer +// refuses. No upper bound (Q150). +test("REQ-THREAD-2 REQ-LAUNCH-1 launch.threadIdleSeconds: absent is the default, a positive integer is used, anything else refuses", () => { + assert.equal(THREAD_IDLE_SECONDS, 900); + assert.equal(check().launch.threadIdleSeconds, 900); + assert.equal(check((d) => { d.launch.threadIdleSeconds = 60; }).launch.threadIdleSeconds, 60); + for (const bad of [0, -1, 1.5, "900", null]) { + refuses(() => check((d) => { d.launch.threadIdleSeconds = bad; }), /launch\.threadIdleSeconds must be a positive integer/); + } + assert.equal(check((d) => { d.launch.threadIdleSeconds = Number.MAX_SAFE_INTEGER; }).launch.threadIdleSeconds, Number.MAX_SAFE_INTEGER); +}); + test("loadBusiness: file checks", () => { const root = tmp(); const dir = join(root, "config"); diff --git a/packages/business/tests/chief-of-staff.test.mjs b/packages/business/tests/chief-of-staff.test.mjs index b993cf5c6..5a71f4917 100644 --- a/packages/business/tests/chief-of-staff.test.mjs +++ b/packages/business/tests/chief-of-staff.test.mjs @@ -9,6 +9,7 @@ import assert from "node:assert/strict"; import { readFileSync } from "node:fs"; import { join } from "node:path"; import { ACTIONS, BusinessError, classify, loadBusiness, loadRole, resolveInstance } from "../src/index.mjs"; +import { THREAD_IDLE_SECONDS } from "../src/business.mjs"; import { businessDoc, launchDoc, REPO_ROLES, systemFor, tmp, writeJson } from "./helpers.mjs"; // Decision 104's list plus role.launch (decision 139). A role file that @@ -58,7 +59,7 @@ test("the role file loads and resolves through loadBusiness", () => { assert.equal(record.definition, "chief-of-staff"); assert.equal(record.contract, join(REPO_ROLES, "chief-of-staff.md")); assert.equal(record.tracker, null); - assert.deepEqual(record.launch, launchDoc(INSTANCE)); + assert.deepEqual(record.launch, { ...launchDoc(INSTANCE), threadIdleSeconds: THREAD_IDLE_SECONDS }); }); test("the shipped file is exactly the approved list", () => { diff --git a/packages/cli/README.md b/packages/cli/README.md index 90836150c..817d991ce 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -337,6 +337,46 @@ the recorded start time and that its command line is the runner's, then sends SIGTERM until it exits (30 s). A stale entry prints "not running" and exits 0; an unknown run exits 2. +### Thread sessions (row 121, #1615) + +A thread instance is an instance whose role definition is `assistant` +(`packages/bus/README.md`, "Projects and threads"). It runs once per +thread and never on its own: a `launch` request or `--pm` for it refuses +`thread-required`, and a thread launch of any other instance refuses +`thread-not-allowed`. `instance-running` counts a thread instance per +thread, so two threads run at once when the limits have room. + +Every `threadTickMs` (2 s) the launcher asks the broker for the business's +threads (`threadsState`) and acts on each one: + +- A thread with no running session starts one when a message to its + instance waits undelivered, or when `thread.open` came after its last + claim (REQ-LAUNCH-1). The start runs the checks in the table above, as + the human, so a full limit refuses `capacity-full` and the next tick + tries again. A refusal is logged once per code per thread, not once per + tick. Each new message or open starts at most one session: a session + that ends without taking the work waits for the next message or open. +- A running thread session with nothing delivered and unread and nothing + waiting stops once the thread has had no message or open for + `launch.threadIdleSeconds` (`packages/business/README.md`). The launch + log gets a `thread-idle` line, and the runner gets SIGTERM, as + `mosaic stop` sends it. An idle thread has no session, so it counts + against no limit (REQ-THREAD-2). + +A thread run's session directory isn't in its run directory. It is +`threads///session/` under the data root, kept across +runs (`packages/seat/README.md`, "Thread stores"), so the next run for the +thread resumes the harness session (REQ-THREAD-2). The launcher checks the +store before anything else; a thread id that isn't a plain id, or an unsafe +directory, refuses `thread-store-refused` and launches nothing. Launch, +refusal and end lines for a thread run carry `thread`, and so does its +`sessions.json` entry, so a host that starts after a crash binds the run +again with its thread and ends it as `host-lost`. + +Thread runs of one instance share its workspace, +`workspaces///`. The clone check runs at each start, +so it can fetch while another thread's session is running. + ### Git socket The launcher also serves `/bus-host/git.sock` (0600, diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index 40f85ad42..8c92975e6 100644 --- a/packages/cli/src/launcher.mjs +++ b/packages/cli/src/launcher.mjs @@ -39,6 +39,25 @@ // listen() also serves the git socket (src/git-socket.mjs, row 93), which // the sandbox binds for a role that holds git.push.working or // review.verdict; session.json names it as gitSocket. +// +// Thread sessions (row 121, #1615). A thread instance (an instance whose +// role definition is the assistant) runs once per thread, never on its own: +// a launch from a session or --pm refuses thread-required. Every tick +// (threadTickMs) the host asks the broker for the business's threads +// (launch op threadsState) and: +// +// - starts a session for a thread with no running session that has an +// undelivered message to its instance, or an open newer than its last +// claim (REQ-LAUNCH-1). The start goes through check(), so it refuses +// as any launch does, and the next tick tries again; +// - stops a thread session that is between turns (nothing delivered and +// unread, nothing waiting) once the thread has had no message or open +// for launch.threadIdleSeconds, the way close() stops a run. An idle +// thread holds no session and counts against no limit. +// +// A thread run's sessionDir is the thread's store (session.mjs +// threadStore()), so the next run for the thread resumes the harness +// session (REQ-THREAD-2). import { spawnSync } from "node:child_process"; import { createServer } from "node:net"; @@ -60,6 +79,7 @@ import { runnerOf, SESSION_CREDENTIALS, sessionEnv, + threadStore, sessionExit, spawnSession, workspaceDir, @@ -75,6 +95,10 @@ import { CliError } from "./errors.mjs"; const PI_PACKAGE = "@earendil-works/pi-coding-agent"; const STOP_TIMEOUT_MS = 30000; const LINE_MAX = 4096; +// How often the host asks for the threads' state; a poll period, not a limit. +const THREAD_TICK_MS = 2000; +// The broker's thread instance definition (packages/bus/src/broker.mjs). +const THREAD_DEFINITION = "assistant"; // A refusal the host decided; `code` goes back to the caller and to the broker. class Refusal extends Error { @@ -103,8 +127,8 @@ export function claudeVersion(env = process.env) { // host: startHost()'s handle. tracker: true when the broker has task verbs for // the business. adapters, bwrap, readOnly, sessionDefaults, versions, -// stopTimeoutMs and git (createGitSocket options) exist for the tests; the -// command line never sets them. +// stopTimeoutMs, threadTickMs and git (createGitSocket options) exist for +// the tests; the command line never sets them. export function createLauncher({ host, system, @@ -116,6 +140,7 @@ export function createLauncher({ sessionDefaults = {}, versions = null, stopTimeoutMs = STOP_TIMEOUT_MS, + threadTickMs = THREAD_TICK_MS, git = {}, log = (l) => process.stderr.write(`mosaic-bus: ${l}\n`), }) { @@ -137,6 +162,11 @@ export function createLauncher({ let chain = Promise.resolve(); let closed = false; let server = null; + let ticker = null; + let ticking = false; + const stopping = new Set(); + const refusedThreads = new Map(); + const startedFor = new Map(); const sockets = new Set(); const gitSocket = createGitSocket({ host, business: b, dataRoot, env, log, ...git }); @@ -155,12 +185,16 @@ export function createLauncher({ } // Steps 2 to 8. Returns what the launch needs. Only `--pm` may name - // launch.by itself. - function check(instance, { pm = false } = {}) { + // launch.by itself. A thread instance launches only for a thread, and + // nothing else launches for one. + function check(instance, { pm = false, thread = null } = {}) { try { b=loadBusiness(host.business,{dir:configDir(env),rolesDir:rolesDir(env)}); } catch(e){throw new Refusal("organization-refused",e.message);} if (!b.launch) throw new Refusal("launch-not-configured", `business ${b.id} has no launch block`); if (!(pm ? instance === b.launch.by : b.launch.instances.includes(instance))) throw new Refusal("instance-not-listed"); - if (registry.running(b.id, instance)) throw new Refusal("instance-running"); + const threaded = b.roles[instance]?.definition === THREAD_DEFINITION; + if (threaded && !thread) throw new Refusal("thread-required", `${instance} runs only for a thread`); + if (!threaded && thread) throw new Refusal("thread-not-allowed", `${instance} isn't a thread instance`); + if (registry.running(b.id, instance, thread)) throw new Refusal("instance-running"); let resolved, model; try { resolved = resolveInstance({ system: vars, business: b, instance }); @@ -181,7 +215,17 @@ export function createLauncher({ } // Bundle, spawn, bind, register, hand over the capability. - async function start({ instance, launchedBy, resolved, model }) { + async function start({ instance, launchedBy, resolved, model, thread = null }) { + // A thread's store is checked first, so a bad id or an unsafe directory + // launches nothing. + let store = null; + if (thread) { + try { + store = threadStore(dataRoot, b.id, thread); + } catch (e) { + throw new Refusal("thread-store-refused", e.message); + } + } const credentials = await host.op({ op: "credentialStatus", business: b.id, role: instance }); // The clones come before the run, so one that can't be made leaves no // run behind. The host's git treats each clone as hostile, and its @@ -201,10 +245,10 @@ export function createLauncher({ if (c.outcome === "kept") record({ event: "checkout-kept", instance, project: c.project, reason: c.reason }); else if (c.outcome === "fetched") record({ event: "checkout-fetched", instance, project: c.project, reason: c.reason }); } - ({resolved,model}=check(instance,{pm:instance===b.launch.by})); + ({resolved,model}=check(instance,{pm:instance===b.launch.by,thread})); const { run, runDir } = newRun(dataRoot, b.id); - const sessionDir = join(runDir, "session"); - mkdirSync(sessionDir, { mode: 0o700 }); + const sessionDir = store ? store.session : join(runDir, "session"); + if (!store) mkdirSync(sessionDir, { mode: 0o700 }); const toolSocket = join(runDir, "tools.sock"); const turnMarker = join(runDir, "turn.done"); const v = versions ?? { pi: piVersion(), claude: model.harness === "claude-code" ? claudeVersion(env) : null }; @@ -228,6 +272,7 @@ export function createLauncher({ ...sessionDefaults, business: b.id, instance, + ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, @@ -252,6 +297,7 @@ export function createLauncher({ launchSocket: resolved.launch ? launchSocketPath(dataRoot) : null, gitSocket: git, claudeBin: env.MOSAIC_CLAUDE_BIN || null, + threadStore: store, readOnly, }; const child = spawnSession({ runDir, env: sessionEnv(env, run), paths, business: b.id, role: instance, bwrap }); @@ -265,10 +311,10 @@ export function createLauncher({ const outerStart = startTimeOf(child.pid); const runner = await runnerOf(child); if (!outerStart) throw new Error("the session process exited at once"); - const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, pid: child.pid, startTime: outerStart }); + const launch = await host.bindLaunch({ business: b.id, role: instance, run, harness: model.harness, pid: child.pid, startTime: outerStart, ...(thread ? { thread } : {}) }); bound = true; entry = { - business: b.id, instance, run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey, + business: b.id, instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey, pid: child.pid, startTime: outerStart, runnerPid: runner.pid, runnerStartTime: runner.startTime, runDir, startedAt: new Date().toISOString(), }; registry.add(entry); @@ -285,18 +331,76 @@ export function createLauncher({ try { await endRun(run, reason, signal ? null : code); registry.remove(b.id, run); - record({ event: "end", instance, run, reason, exitCode: code, signal }); + record({ event: "end", instance, ...(thread ? { thread } : {}), run, reason, exitCode: code, signal }); log(`run ${run} (${instance}) ended: ${reason}`); } catch (e) { log(`run ${run}: ${e.message}`); } finally { children.delete(run); + stopping.delete(run); } }); - record({ event: "launch", instance, run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey, pid: child.pid, runnerPid: entry.runnerPid, manifest }); + record({ event: "launch", instance, ...(thread ? { thread } : {}), run, launchedBy, harness: model.harness, model: model.model, provider: model.provider, modelKey: model.modelKey, pid: child.pid, runnerPid: entry.runnerPid, manifest }); return { run, instance, harness: model.harness, model: model.model }; } + // Row 121: one tick of thread starts and idle stops. A tick that is still + // running when the next is due skips it. + async function tick() { + if (ticking || closed) return; + ticking = true; + try { + let threads; + try { + threads = await host.op({ op: "threadsState", business: b.id }); + } catch (e) { + log(`threadsState refused: ${e.code ?? e.message}`); + return; + } + const now = Date.now(); + const ms = (at) => (at ? Date.parse(at) : 0); + for (const t of threads) { + if (closed) return; + const s = registry.running(b.id, t.instance, t.thread); + if (s) { + const last = Math.max(ms(s.startedAt), ms(t.lastMessage), ms(t.lastOpen)); + if (t.pending === 0 && t.unread === 0 && !stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) { + stopping.add(s.run); + record({ event: "thread-idle", instance: t.instance, thread: t.thread, run: s.run }); + try { + if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM"); + } catch {} + } + continue; + } + const opened = ms(t.lastOpen) > ms(t.lastClaim); + if (t.holder || !(t.pending > 0 || opened)) continue; + // One start per message or open: a session that ended without taking + // the work (it died before its claim, say) waits for the next one + // rather than start again every tick. A refused start isn't a start. + const signature = `${t.lastMessage}|${t.lastOpen}`; + if (startedFor.get(t.thread) === signature) continue; + await serialize(async () => { + if (closed || registry.running(b.id, t.instance, t.thread)) return; + try { + await start({ instance: t.instance, launchedBy: b.human, thread: t.thread, ...check(t.instance, { thread: t.thread }) }); + startedFor.set(t.thread, signature); + refusedThreads.delete(t.thread); + } catch (e) { + const code = e instanceof Refusal ? e.code : "launch-failed"; + // Logged once per code, not once per tick. + if (refusedThreads.get(t.thread) === code) return; + refusedThreads.set(t.thread, code); + log(`thread ${t.thread} (${t.instance}): start refused: ${code}`); + record({ event: "refused", by: b.human, instance: t.instance, thread: t.thread, code, why: e.message, ...(e.limit?{limit:e.limit}:{}) }); + } + }); + } + } finally { + ticking = false; + } + } + async function endRun(run, reason, exitCode) { try { await host.op({ op: "endLaunch", record: { business: b.id, run, reason, exitCode } }); @@ -376,7 +480,7 @@ export function createLauncher({ while (startTimeOf(s.pid) === s.startTime && Date.now() < end) await new Promise((r) => setTimeout(r, 20)); } try { - await host.bindLaunch({ business: b.id, role: s.instance, run: s.run, harness: s.harness, pid: s.pid, startTime: s.startTime }); + await host.bindLaunch({ business: b.id, role: s.instance, run: s.run, harness: s.harness, pid: s.pid, startTime: s.startTime, ...(s.thread ? { thread: s.thread } : {}) }); } catch (e) { // A host that died after ending the run but before updating sessions.json. if (e.code === "run-ended") log(`run ${s.run} (${s.instance}) from an earlier host was already ended`); @@ -384,7 +488,7 @@ export function createLauncher({ continue; } await endRun(s.run, "host-lost", null); - record({ event: "end", instance: s.instance, run: s.run, reason: "host-lost", exitCode: null, signal: null }); + record({ event: "end", instance: s.instance, ...(s.thread ? { thread: s.thread } : {}), run: s.run, reason: "host-lost", exitCode: null, signal: null }); log(`run ${s.run} (${s.instance}) from an earlier host ended: host-lost`); } leftover = []; @@ -436,12 +540,14 @@ export function createLauncher({ }); chmodSync(path, 0o600); await gitSocket.listen(); + if (b.launch) ticker = setInterval(() => tick().catch((e) => log(`thread tick: ${e.message}`)), threadTickMs); } // SIGTERM every runner, wait, then end what's left. Runs before the host // stops the broker, so endLaunch still works. async function close() { closed = true; + if (ticker) clearInterval(ticker); if (server) { // server.close waits for every connection, and a client that never // ends its side (or never sends) would hold it; so they go first. @@ -479,5 +585,5 @@ export function createLauncher({ } host.beforeClose(close); - return { business: b, registry, listen, launchPm, fromSession, close }; + return { business: b, registry, listen, launchPm, fromSession, tick, close }; } diff --git a/packages/cli/tests/fixtures/thread-adapter.mjs b/packages/cli/tests/fixtures/thread-adapter.mjs new file mode 100644 index 000000000..ae2d98efc --- /dev/null +++ b/packages/cli/tests/fixtures/thread-adapter.mjs @@ -0,0 +1,21 @@ +// A stand-in harness turn for the thread launch tests (row 121, #1615). It +// keeps a history in $MOSAIC_SESSION_DIR, which for a thread run is the +// thread's store, and a transcript per turn under $HOME/.claude/projects. +// It answers with the message body and what both held before this turn, so +// a test reads from the reply what a later run of the same thread kept. + +import { appendFileSync, existsSync, readdirSync, readFileSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; + +const request = process.env.MOSAIC_REQUEST ?? ""; +const body = request.match(/^Message [^\n]*:\n\n([\s\S]*)$/m)?.[1] ?? ""; +const historyFile = join(process.env.MOSAIC_SESSION_DIR, "history"); +const before = existsSync(historyFile) ? readFileSync(historyFile, "utf8").split("\n").filter(Boolean) : []; +const projects = join(process.env.HOME, ".claude", "projects"); +const transcripts = existsSync(projects) ? readdirSync(projects).sort() : null; +// A turn longer than the test's idle time. +if (body === "slow") Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 3000); +appendFileSync(historyFile, `${body}\n`); +if (transcripts) writeFileSync(join(projects, `turn-${before.length + 1}.jsonl`), `${JSON.stringify(body)}\n`); +process.stdout.write(`THREAD ${JSON.stringify({ body, before, transcripts })}\n`); +writeFileSync(process.env.MOSAIC_TURN_MARKER, "done\n"); diff --git a/packages/cli/tests/thread-launch.test.mjs b/packages/cli/tests/thread-launch.test.mjs new file mode 100644 index 000000000..f9413fc88 --- /dev/null +++ b/packages/cli/tests/thread-launch.test.mjs @@ -0,0 +1,326 @@ +// Thread sessions (row 121, #1615): the launcher starts a thread instance's +// session when its thread has a pending message or was opened, stops it once +// the thread is idle, and gives every run of a thread the thread's own store. +// Real runners in the bwrap sandbox; a stub adapter keeps a history in the +// store and answers with what it found there. +// +// The host is in process: startBroker on bootConfig's config, with the ops +// the broker child forwards (packages/bus/src/process.mjs). The real host +// proves a human CLI process for every human verb, which a test run under an +// agent can't, so the test's human capability comes from bindHuman directly. + +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { existsSync, mkdirSync, readdirSync, readFileSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { launchDoc } from "../../business/tests/helpers.mjs"; +import { BusError } from "../../bus/src/broker.mjs"; +import { Client } from "../../bus/src/client.mjs"; +import { startBroker } from "../../bus/src/runtime.mjs"; +import { Store } from "../../bus/src/store.mjs"; +import { gitEnv } from "../../seat/src/checkout.mjs"; +import { sandboxProblem } from "../../seat/src/sandbox.mjs"; +import { launchLogFile, readSessions, SESSION_CREDENTIALS } from "../../seat/src/session.mjs"; +import { bootConfig, loadSystem } from "../src/config.mjs"; +import { startTimeOf } from "../src/host.mjs"; +import { createLauncher } from "../src/launcher.mjs"; +import { fixture, tmp } from "./helpers.mjs"; + +const ADAPTER = fileURLToPath(new URL("./fixtures/thread-adapter.mjs", import.meta.url)); +const skip = sandboxProblem() ?? false; + +async function until(fn, ms = 20000) { + const end = Date.now() + ms; + while (Date.now() < end) { + const v = await fn(); + if (v) return v; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error("timed out waiting"); +} + +function gitIn(cwd, ...args) { + const r = spawnSync("git", ["-c", "user.name=test", "-c", "user.email=test@example.invalid", "-C", cwd, ...args], { env: gitEnv(), encoding: "utf8" }); + assert.equal(r.status, 0, r.stderr); +} + +// The assistant is the thread instance; sonnet is its model's limit. +function prepare(t, { sonnet = 2, idle = 1 } = {}) { + const root = tmp(t, "mosaic-thread-"); + const f = fixture(root, "acme", (doc) => { + doc.roles.chief.vars = { harness: "pi", model: "claude-opus-5-5" }; + doc.roles.assistant = { definition: "assistant", vars: { harness: "pi", model: "claude-sonnet-5-5" } }; + doc.launch = launchDoc("chief", ["coder", "assistant"], { opus: 1, sonnet }); + doc.launch.threadIdleSeconds = idle; + return doc; + }); + for (const k of SESSION_CREDENTIALS.flat()) delete f.env[k]; + mkdirSync(f.dataRoot, { mode: 0o700 }); + const project = f.doc.projects.stack.root; + gitIn(project, "init", "-q", "-b", "main"); + writeFileSync(join(project, "file.txt"), "one\n"); + gitIn(project, "add", "file.txt"); + gitIn(project, "commit", "-q", "-m", "one"); + const adapter = join(root, "adapter.sh"); + writeFileSync(adapter, `exec '${process.execPath}' '${ADAPTER}'\n`, { mode: 0o644 }); + return { root, f, adapter }; +} + +// A host and its launcher on the prepared root. close() runs the launcher's +// close hook, then closes the broker, as the real host does. +async function start(t, { f, adapter }, { tickMs = 100 } = {}) { + const system = loadSystem({ env: f.env }); + const { trackers, ...boot } = bootConfig({ system, businessId: "acme", env: f.env }); + assert.equal(trackers, undefined); + const runtime = await startBroker(boot); + const ops = { + identity: (m) => runtime.identity(m.cap), + authorizeLaunch: (m) => runtime.authorizeLaunch(m.cap, m.instance), + refuse: (m) => runtime.refuse(m.cap, m.code, m.subject), + endLaunch: (m) => runtime.endLaunch(m.record), + credentialStatus: (m) => runtime.credentialStatus(m.business, m.role), + threadsState: (m) => runtime.threadsState(m.business), + }; + const refusing = (what, fn) => async (m) => { + try { + return fn(m); + } catch (e) { + const code = e instanceof BusError ? e.code : `${what}-refused`; + throw Object.assign(new Error(`${what} refused: ${code}`), { code }); + } + }; + const hooks = []; + let closing = null; + const host = { + business: "acme", + path: runtime.path, + dataRoot: f.dataRoot, + op: refusing("launch-op", (m) => { + if (!Object.hasOwn(ops, m.op)) throw new BusError("invalid-host-request"); + return ops[m.op](m); + }), + bindLaunch: refusing("launch bind", (r) => runtime.bindLaunch(r)), + beforeClose: (fn) => void hooks.push(fn), + }; + const logs = []; + const launcher = createLauncher({ + host, + system, + env: f.env, + adapters: { pi: adapter, "claude-code": adapter }, + sessionDefaults: { pollInterval: 100 }, + versions: { pi: "0.85.1", claude: null }, + readOnly: [ADAPTER], + threadTickMs: tickMs, + log: (l) => logs.push(l), + }); + const close = () => + (closing ??= (async () => { + for (const fn of hooks) await fn(0); + await runtime.close(); + })()); + t.after(close); + await launcher.listen(); + const human = runtime.broker.bindHuman({ business: "acme", human: f.doc.human, via: "cli", outsideAgent: true }); + const call = (verb, args = {}) => runtime.broker.request(human, { verb, args }); + return { host, runtime, launcher, call, logs, close }; +} + +const launchLog = (f) => + existsSync(launchLogFile(f.dataRoot, "acme")) + ? readFileSync(launchLogFile(f.dataRoot, "acme"), "utf8").trim().split("\n").map((l) => JSON.parse(l)) + : []; +const threadRuns = (f, thread) => readSessions(f.dataRoot).filter((s) => s.thread === thread); + +// Send to a thread and wait for the assistant's RESULT reply in that thread. +async function ask(call, thread, body) { + const sent = call("thread.send", { thread, body }); + const reply = await until(() => call("thread.messages", { thread }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT")); + assert.equal(reply.from_role, "assistant"); + assert.equal(reply.to_role, "human"); + return { sent, reply, saw: JSON.parse(reply.body.match(/THREAD (\{.*\})/)[1]) }; +} + +// The run stops on its own once the thread is idle: no session, no holder. +async function idleStop(f, host, thread, run) { + await until(() => launchLog(f).find((e) => e.event === "end" && e.run === run)); + assert.ok(launchLog(f).find((e) => e.event === "thread-idle" && e.run === run && e.thread === thread)); + assert.deepEqual(threadRuns(f, thread), []); + assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === thread).holder, null); +} + +test("REQ-THREAD-2 REQ-LAUNCH-1 REQ-MSG-1 a thread message starts the thread's session, which stops when idle; its history survives the stop and a restart and no other thread sees it", { skip, timeout: 90000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + let host = await start(t, prepared); + const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; + const A = host.call("thread.create", { project, name: "alpha" }).id; + const B = host.call("thread.create", { project, name: "beta" }).id; + + // Nothing runs for a thread with nothing pending. + await host.launcher.tick(); + assert.deepEqual(readSessions(f.dataRoot), []); + + const first = await ask(host.call, A, "first for A"); + assert.deepEqual(first.saw, { body: "first for A", before: [], transcripts: [] }); + const [run1] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A).map((e) => e.run); + assert.ok(run1); + const store = join(f.dataRoot, "threads", "acme", A); + assert.equal(readFileSync(join(store, "session", "history"), "utf8"), "first for A\n"); + assert.deepEqual(readdirSync(join(store, "claude-projects")), ["turn-1.jsonl"]); + assert.equal(existsSync(join(f.dataRoot, "launches", "acme", run1, "session")), false, "a thread run keeps no per-run session directory"); + await idleStop(f, host, A, run1); + + // A second run of the same thread finds the first run's history. + const second = await ask(host.call, A, "second for A"); + assert.deepEqual(second.saw, { body: "second for A", before: ["first for A"], transcripts: ["turn-1.jsonl"] }); + const run2 = launchLog(f).filter((e) => e.event === "launch" && e.thread === A).at(-1).run; + assert.notEqual(run2, run1); + await idleStop(f, host, A, run2); + + // A restart: a new host and launcher on the same data root. + await host.close(); + host = await start(t, prepared); + const third = await ask(host.call, A, "third for A"); + assert.deepEqual(third.saw, { body: "third for A", before: ["first for A", "second for A"], transcripts: ["turn-1.jsonl", "turn-2.jsonl"] }); + + // Thread B has its own store, empty, and A's messages aren't in it. + const other = await ask(host.call, B, "for B"); + assert.deepEqual(other.saw, { body: "for B", before: [], transcripts: [] }); + assert.ok(!host.call("thread.messages", { thread: B }).some((m) => /for A/.test(m.body))); + assert.ok(!host.call("thread.messages", { thread: A }).some((m) => /for B/.test(m.body))); + + // Opening a thread starts its session with nothing pending; it idles out too. + for (const r of readSessions(f.dataRoot)) await idleStop(f, host, r.thread, r.run); + host.call("thread.open", { thread: B }); + const opened = await until(() => threadRuns(f, B)[0]); + await idleStop(f, host, B, opened.run); +}); + +test("REQ-THREAD-2 REQ-LAUNCH-1 a running thread counts against its model's limit and an idle-stopped one doesn't", { skip, timeout: 60000 }, async (t) => { + const prepared = prepare(t, { sonnet: 1 }); + const { f } = prepared; + const host = await start(t, prepared); + const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; + const A = host.call("thread.create", { project, name: "alpha" }).id; + const B = host.call("thread.create", { project, name: "beta" }).id; + + const toA = host.call("thread.send", { thread: A, body: "for A" }); + const toB = host.call("thread.send", { thread: B, body: "for B" }); + const refused = await until(() => launchLog(f).find((e) => e.event === "refused" && e.thread === B)); + assert.equal(refused.code, "capacity-full"); + assert.equal(refused.instance, "assistant"); + const replyB = await until(() => host.call("thread.messages", { thread: B }).find((m) => m.in_reply_to === toB.id), 30000); + assert.ok(host.call("thread.messages", { thread: A }).find((m) => m.in_reply_to === toA.id)); + + // B started only after A's run ended, and its refusal was logged once. + const entries = launchLog(f); + const endA = entries.findIndex((e) => e.event === "end" && e.run === entries.find((x) => x.event === "launch" && x.thread === A).run); + const launchB = entries.findIndex((e) => e.event === "launch" && e.thread === B); + assert.ok(endA >= 0 && launchB > endA, JSON.stringify(entries.map((e) => [e.event, e.thread, e.code]))); + assert.equal(entries.filter((e) => e.event === "refused" && e.thread === B).length, 1); + assert.match(replyB.body, /THREAD/); +}); + +test("REQ-THREAD-2 REQ-LAUNCH-1 a turn longer than the idle time isn't stopped, and the thread idles out after it", { skip, timeout: 60000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + const host = await start(t, prepared); + const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; + const A = host.call("thread.create", { project, name: "alpha" }).id; + // The stub's "slow" turn takes 3 s and the idle time is 1 s. The message is + // unread for the whole turn. + const sent = host.call("thread.send", { thread: A, body: "slow" }); + const reply = () => host.call("thread.messages", { thread: A }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT"); + await until(() => reply() || launchLog(f).find((e) => e.event === "thread-idle"), 30000); + assert.ok(reply(), JSON.stringify(launchLog(f))); + await idleStop(f, host, A, launchLog(f).find((e) => e.event === "launch" && e.thread === A).run); +}); + +// A thread held by a run the launcher has no session for, as after a restart +// before recovery. The hold is the bus's, so the launcher starts nothing. +test("REQ-ROLE-2 REQ-LAUNCH-1 a thread held by a run the launcher didn't start gets no second session", { skip, timeout: 30000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + const host = await start(t, prepared); + const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; + const A = host.call("thread.create", { project, name: "alpha" }).id; + const held = host.runtime.bindLaunch({ business: "acme", role: "assistant", run: "held-by-hand", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid), thread: A }); + host.runtime.broker.request(held.cap, { verb: "role.claim" }); + assert.equal(host.runtime.threadsState("acme").find((x) => x.thread === A).holder, "held-by-hand"); + host.call("thread.send", { thread: A, body: "for A" }); + for (let i = 0; i < 3; i++) await host.launcher.tick(); + assert.deepEqual(readSessions(f.dataRoot), []); + assert.deepEqual(launchLog(f), []); +}); + +// A run that ends before it takes the message it was started for. The next +// start waits for a new message or open, so a run that dies at once doesn't +// start again on every tick. +test("REQ-LAUNCH-1 a run that ends without taking its thread's message isn't started again for that message", { skip, timeout: 30000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + // Ticks only when the test calls tick(). + const host = await start(t, prepared, { tickMs: 3600000 }); + const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; + const A = host.call("thread.create", { project, name: "alpha" }).id; + const held = host.runtime.bindLaunch({ business: "acme", role: "assistant", run: "held-by-hand", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid), thread: A }); + host.call("thread.send", { thread: A, body: "for A" }); + await host.launcher.tick(); + // tick() returns once the run is bound and its capability is queued on its + // stdin. This claim is synchronous, so it comes before the run's own. + host.runtime.broker.request(held.cap, { verb: "role.claim" }); + const [launch] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A); + const end = await until(() => launchLog(f).find((e) => e.event === "end" && e.run === launch.run)); + assert.equal(end.exitCode, 21, "the run's claim was refused"); + host.runtime.broker.request(held.cap, { verb: "role.release" }); + const state = host.runtime.threadsState("acme").find((x) => x.thread === A); + assert.deepEqual([state.holder, state.pending], [null, 1]); + for (let i = 0; i < 3; i++) await host.launcher.tick(); + assert.equal(launchLog(f).filter((e) => e.event === "launch").length, 1); + assert.deepEqual(readSessions(f.dataRoot), []); +}); + +// A thread id that can't be a path segment, made by hand in the store before +// the host starts, and a thread of a non-thread instance. Neither launches. +test("REQ-THREAD-2 REQ-LAUNCH-1 a thread id that isn't a safe segment, or a thread of a non-thread instance, launches nothing", { skip, timeout: 60000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + const store = new Store(f.dataRoot); + const at = () => new Date().toISOString(); + 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"); + } + store.close(); + + const host = await start(t, prepared); + const refusals = await until(() => { + const r = launchLog(f).filter((e) => e.event === "refused"); + return r.length === 2 && r; + }); + assert.deepEqual(refusals.map((e) => [e.thread, e.instance, e.code]).sort(), [["../evil", "assistant", "thread-store-refused"], ["coder-thread", "coder", "thread-not-allowed"]]); + assert.match(refusals.find((e) => e.thread === "../evil").why, /thread "..\/evil" isn't an id/); + // More ticks log nothing more and start nothing. + for (let i = 0; i < 3; i++) await host.launcher.tick(); + assert.equal(launchLog(f).length, 2); + assert.deepEqual(readSessions(f.dataRoot), []); + assert.equal(existsSync(join(f.dataRoot, "threads")), false); + assert.equal(existsSync(join(f.dataRoot, "evil")), false); + assert.equal(existsSync(join(f.dataRoot, "launches", "acme")), false, "no run directory"); + assert.equal(existsSync(join(f.dataRoot, "workspaces")), false, "no workspace"); +}); + +test("REQ-THREAD-1 REQ-LAUNCH-1 the chief can't launch the assistant outside a thread", { skip, timeout: 30000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + const host = await start(t, prepared); + const chief = await host.host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) }); + await new Client({ path: host.host.path, cap: chief.cap }).call("role.claim"); + assert.deepEqual(await host.launcher.fromSession(chief.cap, "assistant"), { ok: false, error: "thread-required" }); + assert.deepEqual(readSessions(f.dataRoot), []); + assert.deepEqual(launchLog(f).map((e) => [e.event, e.instance, e.code]), [["refused", "assistant", "thread-required"]]); +}); diff --git a/packages/seat/README.md b/packages/seat/README.md index 7796d8887..1742a3853 100644 --- a/packages/seat/README.md +++ b/packages/seat/README.md @@ -247,6 +247,29 @@ The sandbox doesn't confine the network, abstract Unix sockets, or the user: a session runs as the host's UID. `packages/harness/README.md` ("Limits") says what that leaves open. +### Thread stores (row 121, #1615) + +A thread's harness session outlives each of the thread's runs, so the next +run for the thread resumes it (REQ-THREAD-2). `threadStore(dataRoot, +business, thread)` returns its paths under the data root: +`threads///` (0700) holds `session/`, which the launcher +uses as the run's session directory (`MOSAIC_SESSION_DIR`), and +`claude-projects/`, Claude Code's transcripts. + +It checks both segments before it joins them. Each must match the business +vocabulary's id pattern and hold no `..`, so neither has a `/` or a leading +`.`. Then it makes each directory from `threads/` down, and refuses one that +is a symlink, isn't a directory, isn't the user's, or has group or other +bits, as the bus store checks its own directory. A refusal is a `SeatError`; +the launcher turns it into `thread-store-refused`. + +`binds()` takes the store and adds two read-write binds: `session/` at its +own path, and `claude-projects/` at the run's `home/.claude/projects`, a +mount point `prepareRun` makes. A run gets only its own thread's store. +`Registry.running(business, instance, thread)` keys a session by its thread +too, `null` for a session that holds none, so one thread instance runs once +per thread. + ### The project clones (#1585) An instance's workspace holds one git clone per project in the business diff --git a/packages/seat/src/sandbox.mjs b/packages/seat/src/sandbox.mjs index fdf41bbef..bb992ad94 100644 --- a/packages/seat/src/sandbox.mjs +++ b/packages/seat/src/sandbox.mjs @@ -95,8 +95,9 @@ function systemBinds(node = process.execPath) { // the harness ("pi" or "claude-code"), the broker socket, the launch socket // (null for a role that may not launch), the git socket (null for a role // that neither pushes nor reviews), the Claude Code binary -// (MOSAIC_CLAUDE_BIN, or null) and node. `readOnly` adds files, for tests only. -export function binds({ runDir, workspace, adapter, harness, brokerSocket, launchSocket = null, gitSocket = null, claudeBin = null, node = process.execPath, readOnly = [] }) { +// (MOSAIC_CLAUDE_BIN, or null), the thread's session store (session.mjs +// threadStore(), or null) and node. `readOnly` adds files, for tests only. +export function binds({ runDir, workspace, adapter, harness, brokerSocket, launchSocket = null, gitSocket = null, claudeBin = null, threadStore = null, node = process.execPath, readOnly = [] }) { const out = systemBinds(node); for (const f of RUNNER_FILES) out.push({ kind: "ro", src: f, why: "a repository file the runner loads" }); out.push({ kind: "ro", src: adapter, why: "the harness adapter the runner starts each turn" }); @@ -116,6 +117,13 @@ export function binds({ runDir, workspace, adapter, harness, brokerSocket, launc { kind: "ro", src: join(runDir, "session.json"), why: "the session file stays as the host wrote it" }, { kind: "rw", src: join(runDir, "tmp"), dest: "/tmp", why: "a private /tmp, on disk in the run directory" }, ); + // Row 121: only this thread's store, and only its two directories. + if (threadStore) { + out.push( + { kind: "rw", src: threadStore.session, why: "the thread's session directory, kept across its runs (MOSAIC_SESSION_DIR)" }, + { kind: "rw", src: threadStore.projects, dest: join(runDir, "home", ".claude", "projects"), why: "the thread's Claude Code transcripts, kept across its runs" }, + ); + } return out; } @@ -132,8 +140,10 @@ function argsOf(list) { } // The run's home/ and tmp/, made before the spawn: binds need their sources. -export function prepareRun(runDir) { +// A thread run also needs the mount point for its transcripts. +export function prepareRun(runDir, { threadStore = null } = {}) { for (const d of ["home", "tmp"]) mkdirSync(join(runDir, d), { recursive: true, mode: 0o700 }); + if (threadStore) mkdirSync(join(runDir, "home", ".claude", "projects"), { recursive: true, mode: 0o700 }); } // The git identity a session commits under (#1598): the role instance's diff --git a/packages/seat/src/session.mjs b/packages/seat/src/session.mjs index 19c4fc53d..7ffb89c04 100644 --- a/packages/seat/src/session.mjs +++ b/packages/seat/src/session.mjs @@ -20,17 +20,21 @@ // every launch, refusal and end // workspaces/// the instance's workspace (0700), kept // across launches +// threads/// a thread's session store (0700), kept +// across the thread's runs (row 121): +// session/ and claude-projects/ // bus-host/sessions.json the running sessions (0600), written // only by the host; `mosaic stop` reads it import { spawn } from "node:child_process"; import { randomBytes } from "node:crypto"; -import { appendFileSync, closeSync, existsSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync } from "node:fs"; +import { appendFileSync, closeSync, existsSync, lstatSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync } from "node:fs"; import { constants } from "node:os"; import { dirname, join } from "node:path"; import { childrenOf, cmdlineOf, startTimeOf } from "./proc.mjs"; import { binds, prepareRun, REPO, sandboxArgv } from "./sandbox.mjs"; import { SeatError } from "./seat.mjs"; +import { ID_PATTERN } from "../../business/src/vocabulary.mjs"; export { REPO }; export const RUNNER = join(REPO, "packages", "harness", "src", "runner.mjs"); @@ -48,6 +52,32 @@ export const runsDir = (dataRoot, business) => join(dataRoot, "launches", busine export const launchLogFile = (dataRoot, business) => join(dataRoot, "launches", `${business}.jsonl`); export const workspaceDir = (dataRoot, business, instance) => join(dataRoot, "workspaces", business, instance); +// Row 121 (#1615): a thread's session store, which outlives each of the +// thread's runs so the next run resumes the harness session (REQ-THREAD-2). +// Both segments are checked before the join: an id, no "..", so no "/" and +// no leading ".". Then every directory from threads/ down must be ours, a +// real directory and private, as store.mjs safe() checks the bus. session/ +// is the run's sessionDir; claude-projects/ is bound at the run's +// home/.claude/projects (sandbox.mjs binds()). Throws SeatError. +const segment = (x) => typeof x === "string" && ID_PATTERN.test(x) && !x.includes(".."); +export function threadStore(dataRoot, business, thread) { + for (const [what, x] of [["business", business], ["thread", thread]]) { + if (!segment(x)) throw new SeatError(`thread store: ${what} ${JSON.stringify(x)} isn't an id`, 2); + } + const dir = join(dataRoot, "threads", business, thread); + const out = { dir, session: join(dir, "session"), projects: join(dir, "claude-projects") }; + for (const d of [join(dataRoot, "threads"), join(dataRoot, "threads", business), dir, out.session, out.projects]) { + try { + mkdirSync(d, { mode: 0o700 }); + } catch (e) { + if (e.code !== "EEXIST") throw e; + } + const st = lstatSync(d); + if (st.isSymbolicLink() || !st.isDirectory() || st.uid !== process.getuid() || st.mode & 0o077) throw new SeatError(`thread store: ${d} is unsafe`, 2); + } + return out; +} + // The session's environment is an allowlist: nothing else from the host's // environment passes, so a founder credential variable (GITEA_TOKEN, // GH_TOKEN, SSH_AUTH_SOCK, ...) never reaches a session (REQ-CRED-2). @@ -110,10 +140,10 @@ export function appendLaunchLog(dataRoot, business, entry) { // the host writes once the broker has bound the launch; stdout and stderr go // to runner.log. paths: what sandbox.mjs's binds() takes besides the run // directory (workspace, adapter, harness, brokerSocket, launchSocket, -// claudeBin; readOnly for tests). business and role (the instance) give the -// session its git identity. `bwrap` is for the tests. +// claudeBin, threadStore; readOnly for tests). business and role (the +// instance) give the session its git identity. `bwrap` is for the tests. export function spawnSession({ runDir, env, paths, business, role, node = process.execPath, bwrap }) { - prepareRun(runDir); + prepareRun(runDir, { threadStore: paths.threadStore ?? null }); const argv = sandboxArgv(binds({ ...paths, runDir, node }), runDir, [node, RUNNER, runDir], { business, role, bwrap }); const out = openSync(join(runDir, "runner.log"), "a", 0o600); try { @@ -181,8 +211,10 @@ export class Registry { count(business, provider, model = undefined) { return this.#sessions.filter((s) => s.business === business && s.provider === provider && (model === undefined || s.model === model)).length; } - running(business, instance) { - return this.#sessions.find((s) => s.business === business && s.instance === instance) ?? null; + // A thread instance runs once per thread (row 121), so `thread` is part of + // the key; null is a session that holds no thread. + running(business, instance, thread = null) { + return this.#sessions.find((s) => s.business === business && s.instance === instance && (s.thread ?? null) === thread) ?? null; } #write() { const file = sessionsFile(this.#dataRoot); diff --git a/packages/seat/tests/sandbox.test.mjs b/packages/seat/tests/sandbox.test.mjs index 77168d405..7cbb25327 100644 --- a/packages/seat/tests/sandbox.test.mjs +++ b/packages/seat/tests/sandbox.test.mjs @@ -9,7 +9,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import { spawn, spawnSync } from "node:child_process"; -import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, writeFileSync } from "node:fs"; +import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { basename, dirname, join, resolve } from "node:path"; import { Broker } from "../../bus/src/broker.mjs"; @@ -70,6 +70,29 @@ test("every bind says why; the bundle and session file stay read-only over the r assert.ok(!list.some((b) => b.src === realpathSync(join(REPO, "node_modules"))), "Claude Code gets no node_modules"); }); +// Row 121 (#1615): a thread run gets only its own thread's two directories, +// writable, and a home/.claude/projects mount point made before the sandbox starts. +test("REQ-THREAD-2 a thread store binds its session directory and Claude transcripts, and nothing more", (t) => { + const root = scratch(t); + const runDir = join(root, "run"); + const store = { dir: join(root, "threads", "acme", "th"), session: join(root, "threads", "acme", "th", "session"), projects: join(root, "threads", "acme", "th", "claude-projects") }; + const args = { runDir, workspace: join(root, "ws"), adapter: join(root, "a.sh"), harness: "claude-code", brokerSocket: join(root, "bus.sock") }; + const list = binds({ ...args, threadStore: store }); + const plain = binds(args); + assert.deepEqual(list.slice(0, plain.length), plain, "the other binds are unchanged"); + assert.deepEqual(list.slice(plain.length).map(({ kind, src, dest }) => ({ kind, src, dest })), [ + { kind: "rw", src: store.session, dest: undefined }, + { kind: "rw", src: store.projects, dest: join(runDir, "home", ".claude", "projects") }, + ]); + assert.ok(list.every((b) => b.why)); + assert.equal(list.findIndex((b) => b.src === store.dir || b.src === join(root, "threads")), -1, "not the store's parent, not other threads"); + + prepareRun(runDir); + assert.equal(existsSync(join(runDir, "home", ".claude")), false); + prepareRun(runDir, { threadStore: store }); + assert.equal(statSync(join(runDir, "home", ".claude", "projects")).mode & 0o777, 0o700); +}); + // Row 93 (#1587): the git socket is bound read-only, before the workspace, when the launcher passes one. test("the host's git socket is bound only when given, read-only and before the writable binds", (t) => { const root = scratch(t); diff --git a/packages/seat/tests/session.test.mjs b/packages/seat/tests/session.test.mjs index c9babcd6a..a31297594 100644 --- a/packages/seat/tests/session.test.mjs +++ b/packages/seat/tests/session.test.mjs @@ -5,7 +5,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; -import { existsSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { chmodSync, existsSync, lstatSync, mkdirSync, mkdtempSync, readFileSync, realpathSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { Broker } from "../../bus/src/broker.mjs"; @@ -30,6 +30,7 @@ import { spawnSession, sessionExit, stopSession, + threadStore, writeSessionFile, } from "../src/session.mjs"; import { BWRAP, sandboxProblem } from "../src/sandbox.mjs"; @@ -132,6 +133,47 @@ test("Registry mirrors to sessions.json; readSessions marks live entries; bad fi assert.throws(() => readSessions(root), (e) => e instanceof SeatError && /malformed/.test(e.message)); }); +// Row 121 (#1615): a thread instance runs once per thread, so the registry +// keys a thread run by its thread too. +test("REQ-THREAD-1 Registry.running keys a thread run by business, instance and thread", (t) => { + const r = new Registry(scratch(t)); + const base = { business: "acme", instance: "assistant", provider: "zai", model: "m", runnerPid: process.pid, runnerStartTime: startTimeOf(process.pid) }; + r.add({ ...base, run: "r1", thread: "th-a" }); + assert.equal(r.running("acme", "assistant", "th-a").run, "r1"); + assert.equal(r.running("acme", "assistant", "th-b"), null); + assert.equal(r.running("acme", "assistant"), null, "a thread run isn't the instance's non-thread run"); + r.add({ ...base, instance: "coder", run: "r2" }); + assert.equal(r.running("acme", "coder").run, "r2"); + assert.equal(r.running("acme", "coder", "th-a"), null); +}); + +// Row 121 (#1615, sage-0308 change 3): and are checked +// against the id pattern before the path join, then every directory is lstat-checked. +test("REQ-THREAD-2 REQ-LAUNCH-1 threadStore checks both segments before the join and refuses an unsafe directory", (t) => { + const root = scratch(t); + for (const [business, thread] of [["acme", ".."], ["acme", "a/b"], ["acme", ".x"], ["acme", "a..b"], ["..", "th"], ["acme", ""], ["acme", "Th"], ["acme", 7], [undefined, "th"]]) { + assert.throws(() => threadStore(root, business, thread), (e) => e instanceof SeatError && /^thread store: (business|thread) .* isn't an id$/.test(e.message), `${business}/${thread}`); + } + assert.equal(existsSync(join(root, "threads")), false, "a refused segment makes nothing"); + + const s = threadStore(root, "acme", "th-1"); + assert.deepEqual(s, { dir: join(root, "threads", "acme", "th-1"), session: join(root, "threads", "acme", "th-1", "session"), projects: join(root, "threads", "acme", "th-1", "claude-projects") }); + for (const d of [join(root, "threads"), join(root, "threads", "acme"), s.dir, s.session, s.projects]) assert.equal(mode(d), 0o700, d); + writeFileSync(join(s.session, "history"), "kept\n"); + assert.deepEqual(threadStore(root, "acme", "th-1"), s, "a second call reuses the store"); + assert.equal(readFileSync(join(s.session, "history"), "utf8"), "kept\n"); + + chmodSync(s.session, 0o755); + assert.throws(() => threadStore(root, "acme", "th-1"), (e) => e instanceof SeatError && /is unsafe$/.test(e.message)); + chmodSync(s.session, 0o700); + const elsewhere = join(root, "elsewhere"); + mkdirSync(elsewhere, { mode: 0o700 }); + symlinkSync(elsewhere, join(root, "threads", "acme", "th-2")); + assert.throws(() => threadStore(root, "acme", "th-2"), (e) => e instanceof SeatError && /th-2 is unsafe$/.test(e.message)); + assert.equal(existsSync(join(elsewhere, "session")), false, "nothing is made through the link"); + assert.ok(lstatSync(join(root, "threads", "acme", "th-2")).isSymbolicLink()); +}); + test("stopSession refuses an unknown run, reports a stale one, and won't signal a pid that isn't a runner", async (t) => { const root = scratch(t); const r = new Registry(root);