mosaicstack/stack
25
0 Fork 1
Code Issues 404 Pull Requests 13 Packages Projects Releases 15 Wiki Activity
Files
org-cb01
stack/agents/darkwing/work/queue-121/r1/build.patch
T
jason.woltjeandClaude Opus 5.5 ef8c3fe713 agents: Darkwing row 121 round 1 packet (candidate manifest, gate, 62 mutants) 
Round 1 is a code and test review and won't land. land-work fails at
BASE on the read.mjs 3b pin, which row 49 removes.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-15 12:35:28 -05:00

2308 lines
124 KiB
Diff
Raw Permalink Blame History

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/<business>/<thread>/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/<business>/<instance>/`. The clone check runs at each start,
+so it can fetch while another thread's session is running.
+
### Git socket
The launcher also serves `<dataRoot>/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", "[email protected]", "-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/<business>/<thread>/` (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/<business>/<instance>/ the instance's workspace (0700), kept
// across launches
+// threads/<business>/<thread>/ 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): <business> and <thread> 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);
Reference in new issue View Git Blame Copy Permalink