harness: a new PM run reads current launch state and open tasks before each turn, and old messages show their age, row 109 (#1603)
The broker gains a read-only state verb (launch state with the event that set it, open tasks capped). The runner fetches it before each turn and puts a "Current state" block at the top of the request; a failed read fails the turn. turnRequest() adds the message time. roles/pm.md gets one rule: the block is the truth for launch state and task status. Seat session, seat sandbox and cli launcher tests re-anchored to the new request shape. Built by Researcher, packet875c515e9. Filbert approved round 3, comment 28333, candidate manifest 0a62225e (11 paths). Sage gate on9774d70dbplus row109.patch: every package node suite and every scripts/test-*.sh green on the first run, skipped 0 in bus, harness and seat. Vikunja #259. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -157,6 +157,7 @@ an operator inspect content and delivery evidence after an uncertain result.
|
|||||||
| `decisions` | `{list}`: `roles` or `recent`; human and reader sessions only. Open decisions routed to a role, or the last 50 closed. |
|
| `decisions` | `{list}`: `roles` or `recent`; human and reader sessions only. Open decisions routed to a role, or the last 50 closed. |
|
||||||
| `credentials` | No args; human and reader sessions only. `[{service,account,role,dateKind,date,state}]` for the capability's business, never a token, file path or variable name. |
|
| `credentials` | No args; human and reader sessions only. `[{service,account,role,dateKind,date,state}]` for the capability's business, never a token, file path or variable name. |
|
||||||
| `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`. |
|
| `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. |
|
| `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. |
|
||||||
|
|
||||||
### Agent state (row 59, #1548)
|
### Agent state (row 59, #1548)
|
||||||
|
|||||||
@@ -654,7 +654,7 @@ export class Broker {
|
|||||||
if (!object(args)) fail('invalid-request');
|
if (!object(args)) fail('invalid-request');
|
||||||
this.#secretCheck(args);
|
this.#secretCheck(args);
|
||||||
return this.#store.transaction(() => {
|
return this.#store.transaction(() => {
|
||||||
if (s.reader && !['inbox', 'agents', 'tasks', 'trail', 'decisions', 'credentials', 'replies', 'messages'].includes(request.verb))
|
if (s.reader && !['inbox', 'agents', 'tasks', 'trail', 'decisions', 'credentials', 'replies', 'messages', 'state'].includes(request.verb))
|
||||||
fail('read-only');
|
fail('read-only');
|
||||||
const result = this.#dispatch(s, request.verb, args);
|
const result = this.#dispatch(s, request.verb, args);
|
||||||
this.#secretCheck(result);
|
this.#secretCheck(result);
|
||||||
@@ -1086,6 +1086,23 @@ export class Broker {
|
|||||||
this.#agent(s);
|
this.#agent(s);
|
||||||
return this.#requirements;
|
return this.#requirements;
|
||||||
}
|
}
|
||||||
|
// Row 109 (#1603): what a run needs to know before it plans. The business's launch state with
|
||||||
|
// the time and id of the event that set it (`allowed`, no time and no id when none has), and
|
||||||
|
// the open tasks as the tracker snapshot shows them: refs and titles, at most `cap` of them,
|
||||||
|
// `total` counting every open task. A read for role, reader and human sessions; nothing is stored.
|
||||||
|
case 'state': {
|
||||||
|
keys(a, []);
|
||||||
|
if (!s.human && !s.reader) this.#agent(s);
|
||||||
|
const set = this.#store.get(
|
||||||
|
"SELECT id,kind,at FROM events WHERE business=? AND kind IN ('launch.revoked','launch.restored') ORDER BY seq DESC LIMIT 1",
|
||||||
|
s.business,
|
||||||
|
);
|
||||||
|
const open = this.#tasks(s.business, { done: false });
|
||||||
|
return {
|
||||||
|
launch: { state: set?.kind === 'launch.revoked' ? 'revoked' : 'allowed', at: set?.at ?? null, event: set?.id ?? null },
|
||||||
|
tasks: { open: open.tasks.map(({ task_ref, title }) => ({ task_ref, title })), total: open.total, cap: TASK_PAGE },
|
||||||
|
};
|
||||||
|
}
|
||||||
case 'inbox':
|
case 'inbox':
|
||||||
case 'agents':
|
case 'agents':
|
||||||
case 'tasks':
|
case 'tasks':
|
||||||
|
|||||||
@@ -0,0 +1,113 @@
|
|||||||
|
import test from 'node:test';
|
||||||
|
import assert from 'node:assert/strict';
|
||||||
|
import { mkdtempSync, rmSync } from 'node:fs';
|
||||||
|
import { tmpdir } from 'node:os';
|
||||||
|
import { join } from 'node:path';
|
||||||
|
import { Store } from '../src/store.mjs';
|
||||||
|
import { Broker, TASK_PAGE } from '../src/broker.mjs';
|
||||||
|
// Row 109 (#1603): `state` is what a run reads before it plans: the launch state with the time and
|
||||||
|
// id of the event that set it, and the open tasks from the tracker snapshot, capped.
|
||||||
|
const authority = { withinRole: ['message.send', 'role.launch'], crossRole: [] };
|
||||||
|
const roles = { pm: { authority }, coder: { authority } };
|
||||||
|
const businesses = {
|
||||||
|
demo: { id: 'demo', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, launch: { by: 'pm', instances: ['coder'], max: { opus: 2, sonnet: 2 } }, roles },
|
||||||
|
};
|
||||||
|
const at = '2026-10-12T12:00:00.000Z';
|
||||||
|
const digest = 'd'.repeat(64);
|
||||||
|
const ref = (i) => `vikunja:3/${i}`;
|
||||||
|
const fields = (i, done = false) => ({ project_id: 3, title: `Task ${i}`, description: `<p>${i}</p>`, done, bucket: 11 });
|
||||||
|
|
||||||
|
function setup(t) {
|
||||||
|
const root = mkdtempSync(join(tmpdir(), 'bus-state-'));
|
||||||
|
const store = new Store(root);
|
||||||
|
const b = new Broker({ store, businesses });
|
||||||
|
t.after(() => {
|
||||||
|
store.close();
|
||||||
|
rmSync(root, { recursive: true, force: true });
|
||||||
|
});
|
||||||
|
const human = b.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
|
||||||
|
const pm = b.bindLaunch({ business: 'demo', role: 'pm', run: 'demo-pm', harness: 'claude-code' });
|
||||||
|
b.request(pm, { verb: 'role.claim' });
|
||||||
|
const snap = (i, over = {}, read_at = at) => ({ task_ref: ref(i), updated: read_at, digest, fields: { ...fields(i), ...over }, via: 'board', read_at });
|
||||||
|
const record = (...snapshots) => b.recordTask({ business: 'demo', snapshots });
|
||||||
|
const state = (cap = pm, args = {}) => b.request(cap, { verb: 'state', args });
|
||||||
|
return { b, store, human, pm, snap, record, state, reader: b.bindReader({ business: 'demo' }) };
|
||||||
|
}
|
||||||
|
|
||||||
|
test('a business that never switched launches reads allowed with no time and no event', (t) => {
|
||||||
|
const { state } = setup(t);
|
||||||
|
assert.deepEqual(state().launch, { state: 'allowed', at: null, event: null });
|
||||||
|
});
|
||||||
|
|
||||||
|
test('launches revoked and then restored read allowed, with the restore time and event id', (t) => {
|
||||||
|
const { b, store, human, pm, state } = setup(t);
|
||||||
|
const revoked = b.request(human, { verb: 'launch.revoke' }).id;
|
||||||
|
assert.deepEqual(
|
||||||
|
{ state: state().launch.state, event: state().launch.event },
|
||||||
|
{ state: 'revoked', event: revoked },
|
||||||
|
);
|
||||||
|
const restored = b.request(human, { verb: 'launch.restore' }).id;
|
||||||
|
const row = store.get('SELECT at FROM events WHERE id=?', restored);
|
||||||
|
assert.deepEqual(state(pm).launch, { state: 'allowed', at: row.at, event: restored });
|
||||||
|
assert.notEqual(restored, revoked);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('a task closed on the tracker leaves the open list, and total counts the open ones', (t) => {
|
||||||
|
const { snap, record, state } = setup(t);
|
||||||
|
record(snap(1), snap(2), snap(3));
|
||||||
|
assert.deepEqual(state().tasks, {
|
||||||
|
open: [1, 2, 3].map((i) => ({ task_ref: ref(i), title: `Task ${i}` })),
|
||||||
|
total: 3,
|
||||||
|
cap: TASK_PAGE,
|
||||||
|
});
|
||||||
|
record(snap(2, { done: true }, '2026-10-12T13:00:00.000Z'));
|
||||||
|
const { open, total } = state().tasks;
|
||||||
|
assert.deepEqual(open.map((x) => x.task_ref), [ref(1), ref(3)]);
|
||||||
|
assert.equal(total, 2);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('a tombstone is not open', (t) => {
|
||||||
|
const { snap, record, state } = setup(t);
|
||||||
|
record(snap(1), { ...snap(2), fields: { gone: 'no-access' }, via: 'task' });
|
||||||
|
assert.deepEqual(state().tasks.open.map((x) => x.task_ref), [ref(1)]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('the open list stops at the cap and says so; total still counts every open task', (t) => {
|
||||||
|
const { snap, record, state } = setup(t);
|
||||||
|
record(...Array.from({ length: TASK_PAGE + 5 }, (_, k) => snap(k + 1)));
|
||||||
|
const { open, total, cap } = state().tasks;
|
||||||
|
assert.equal(open.length, TASK_PAGE);
|
||||||
|
assert.equal(total, TASK_PAGE + 5);
|
||||||
|
assert.equal(cap, TASK_PAGE);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('a role and a reader session read state; arguments, and an unbound cap, refuse', (t) => {
|
||||||
|
const { reader, pm, state } = setup(t);
|
||||||
|
assert.deepEqual(state(reader), state(pm));
|
||||||
|
assert.throws(() => state(pm, { business: 'other' }), { code: 'invalid-request' });
|
||||||
|
assert.throws(() => state('f'.repeat(64)), /unauthenticated/);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('a human session reads the same state, with the same cap', (t) => {
|
||||||
|
const { snap, record, human, pm, state } = setup(t);
|
||||||
|
record(...Array.from({ length: TASK_PAGE + 2 }, (_, k) => snap(k + 1)));
|
||||||
|
const read = state(human);
|
||||||
|
assert.deepEqual(read, state(pm));
|
||||||
|
assert.equal(read.tasks.cap, TASK_PAGE);
|
||||||
|
assert.equal(read.tasks.open.length, TASK_PAGE);
|
||||||
|
assert.equal(read.tasks.total, TASK_PAGE + 2);
|
||||||
|
});
|
||||||
|
|
||||||
|
test('reading state writes nothing: every table keeps its row count', (t) => {
|
||||||
|
const { b, store, human, pm, reader, snap, record, state } = setup(t);
|
||||||
|
b.request(human, { verb: 'launch.revoke' });
|
||||||
|
record(snap(1), snap(2));
|
||||||
|
const counts = () => Object.fromEntries(
|
||||||
|
store.all("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
|
||||||
|
.map(({ name }) => [name, store.get(`SELECT COUNT(*) AS n FROM "${name}"`).n]),
|
||||||
|
);
|
||||||
|
const before = counts();
|
||||||
|
assert.ok(before.events > 0 && before.task_snapshots > 0);
|
||||||
|
for (const cap of [pm, reader, human]) state(cap);
|
||||||
|
assert.deepEqual(counts(), before);
|
||||||
|
});
|
||||||
@@ -234,10 +234,14 @@ test("the PM launches a coder through its launch tool; the coder answers; refusa
|
|||||||
}
|
}
|
||||||
|
|
||||||
// The coder claimed its role and answers its own messages.
|
// The coder claimed its role and answers its own messages.
|
||||||
|
const before = Date.now();
|
||||||
const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
|
const hello = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" });
|
||||||
const echo = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id));
|
const echo = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === hello.id));
|
||||||
assert.equal(echo.from_role, "coder");
|
assert.equal(echo.from_role, "coder");
|
||||||
assert.match(echo.body, /^echo: Message .* from cto, class REQUEST:\n\nhello/);
|
// Row 109 (#1603): the turn opens with the state block, then the header with the time it was sent.
|
||||||
|
const turn = echo.body.match(new RegExp(`^echo: Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\nMessage ${hello.id} from cto, sent (\\S+), class REQUEST:\n\nhello$`));
|
||||||
|
assert.ok(turn, echo.body);
|
||||||
|
assert.ok(Date.parse(turn[1]) >= before - 1000 && Date.parse(turn[1]) <= Date.now(), "the sent time is when it was sent");
|
||||||
|
|
||||||
// Host refusals, asked with the cto capability (steps 2 to 8 come before
|
// Host refusals, asked with the cto capability (steps 2 to 8 come before
|
||||||
// the broker's role.launch check, which cto would fail).
|
// the broker's role.launch check, which cto would fail).
|
||||||
|
|||||||
@@ -182,6 +182,15 @@ it never touches the disk and isn't in the environment or argv.
|
|||||||
Its stdout goes back to the sender as a RESULT; a failed turn replies
|
Its stdout goes back to the sender as a RESULT; a failed turn replies
|
||||||
`The turn failed: <reason>`; an incoming RESULT never gets an automatic
|
`The turn failed: <reason>`; an incoming RESULT never gets an automatic
|
||||||
reply. Then `message.read`.
|
reply. Then `message.read`.
|
||||||
|
|
||||||
|
Row 109 (#1603): before each turn the runner calls the broker's `state`
|
||||||
|
verb and puts a block at the top of the request, headed `Current state,
|
||||||
|
read at <time>`: the launch state with the time and id of the event that
|
||||||
|
set it, and the open tasks (refs and titles, at most the stated cap, with
|
||||||
|
the total). The block is the truth over any older message. The message
|
||||||
|
header after it names when the message was sent. A failed read fails the
|
||||||
|
turn without running the adapter: the sender gets `The turn failed: could
|
||||||
|
not read the current state: <code>`, and the message is read.
|
||||||
5. SIGTERM or SIGINT kills the turn's process group, releases the role and
|
5. SIGTERM or SIGINT kills the turn's process group, releases the role and
|
||||||
exits 0. The handlers go in before anything else, because pid 1 of a
|
exits 0. The handlers go in before anything else, because pid 1 of a
|
||||||
PID namespace never sees a signal it has no handler for; a stop before
|
PID namespace never sees a signal it has no handler for; a stop before
|
||||||
|
|||||||
@@ -62,7 +62,10 @@ export function founderCheck(policy, env) {
|
|||||||
|
|
||||||
// The request text one message becomes.
|
// The request text one message becomes.
|
||||||
export function turnRequest(m) {
|
export function turnRequest(m) {
|
||||||
const head = [`Message ${m.id} from ${m.from_role}`, `class ${m.class}`];
|
const head = [`Message ${m.id} from ${m.from_role}`];
|
||||||
|
// Row 109 (#1603): when the sender sent it, so a message that waited reads as old.
|
||||||
|
if (m.at) head.push(`sent ${m.at}`);
|
||||||
|
head.push(`class ${m.class}`);
|
||||||
if (m.in_reply_to) head.push(`in reply to ${m.in_reply_to}`);
|
if (m.in_reply_to) head.push(`in reply to ${m.in_reply_to}`);
|
||||||
if (m.decision) head.push(`citing decision ${m.decision}`);
|
if (m.decision) head.push(`citing decision ${m.decision}`);
|
||||||
// Row 90 (#1584): the human input id task_create cites as its request.
|
// Row 90 (#1584): the human input id task_create cites as its request.
|
||||||
@@ -70,6 +73,22 @@ export function turnRequest(m) {
|
|||||||
return `${head.join(", ")}:\n\n${m.body}`;
|
return `${head.join(", ")}:\n\n${m.body}`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Row 109 (#1603): the block at the top of a turn's request. It is what the broker said at `readAt`,
|
||||||
|
// the truth for launch state and task status over any older message.
|
||||||
|
export function stateBlock(state, readAt) {
|
||||||
|
const { launch, tasks } = state;
|
||||||
|
const since = launch.at ? ` since ${launch.at} (event ${launch.event})` : " (never switched)";
|
||||||
|
const shown = tasks.open.length < tasks.total ? `, the first ${tasks.open.length} of cap ${tasks.cap}` : "";
|
||||||
|
const open = tasks.open.map((t) => `- ${t.task_ref} ${t.title ?? ""}`.trimEnd());
|
||||||
|
return [
|
||||||
|
`Current state, read at ${readAt}:`,
|
||||||
|
`Launches: ${launch.state}${since}.`,
|
||||||
|
`Open tasks: ${tasks.total}${shown}.`,
|
||||||
|
...open,
|
||||||
|
"",
|
||||||
|
].join("\n");
|
||||||
|
}
|
||||||
|
|
||||||
function clip(text) {
|
function clip(text) {
|
||||||
return text.length > BODY_MAX ? text.slice(0, BODY_MAX) + "\n[answer cut at 32000 characters]" : text;
|
return text.length > BODY_MAX ? text.slice(0, BODY_MAX) + "\n[answer cut at 32000 characters]" : text;
|
||||||
}
|
}
|
||||||
@@ -339,7 +358,23 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
|
|||||||
}
|
}
|
||||||
for (const m of messages) {
|
for (const m of messages) {
|
||||||
if (stopping) break loop;
|
if (stopping) break loop;
|
||||||
const turn = await runTurn({ session, request: turnRequest(m), n: ++n, env: turnEnv, onChild: (c) => (child = c) });
|
// Row 109 (#1603): read the state now, not when the message was sent. A failed read fails the turn.
|
||||||
|
let block = null;
|
||||||
|
let unread = null;
|
||||||
|
try {
|
||||||
|
block = stateBlock(await client.call("state"), new Date().toISOString());
|
||||||
|
} catch (e) {
|
||||||
|
if (SESSION_GONE.has(e.code)) {
|
||||||
|
log(`runner: session stopped working: ${e.code}`);
|
||||||
|
code = EXIT.session;
|
||||||
|
break loop;
|
||||||
|
}
|
||||||
|
unread = `could not read the current state: ${e.code ?? e.message}`;
|
||||||
|
}
|
||||||
|
++n;
|
||||||
|
const turn = unread
|
||||||
|
? { ok: false, reason: unread }
|
||||||
|
: await runTurn({ session, request: `${block}\n${turnRequest(m)}`, n, env: turnEnv, onChild: (c) => (child = c) });
|
||||||
if (stopping) break loop;
|
if (stopping) break loop;
|
||||||
if (!turn.ok) log(`runner: turn ${n} for message ${m.id} failed: ${turn.reason}`);
|
if (!turn.ok) log(`runner: turn ${n} for message ${m.id} failed: ${turn.reason}`);
|
||||||
try {
|
try {
|
||||||
|
|||||||
+2
-1
@@ -11,7 +11,8 @@ import { appendFileSync, writeFileSync } from "node:fs";
|
|||||||
import { callTool } from "../../src/tools.mjs";
|
import { callTool } from "../../src/tools.mjs";
|
||||||
|
|
||||||
const request = process.env.MOSAIC_REQUEST ?? "";
|
const request = process.env.MOSAIC_REQUEST ?? "";
|
||||||
const body = request.split("\n\n").slice(1).join("\n\n");
|
// The message body follows the "Message ...:" header line; a state block (row 109) may sit above it.
|
||||||
|
const body = request.match(/^Message [^\n]*:\n\n([\s\S]*)$/m)?.[1] ?? request.split("\n\n").slice(1).join("\n\n");
|
||||||
if (process.env.FAKE_PIDS) appendFileSync(process.env.FAKE_PIDS, `${process.pid}\n`);
|
if (process.env.FAKE_PIDS) appendFileSync(process.env.FAKE_PIDS, `${process.pid}\n`);
|
||||||
process.stderr.write(`fake adapter: ${process.env.MOSAIC_MODEL}\n`);
|
process.stderr.write(`fake adapter: ${process.env.MOSAIC_MODEL}\n`);
|
||||||
const done = () => writeFileSync(process.env.MOSAIC_TURN_MARKER, "done\n");
|
const done = () => writeFileSync(process.env.MOSAIC_TURN_MARKER, "done\n");
|
||||||
|
|||||||
@@ -11,9 +11,9 @@ import { open } from "node:fs/promises";
|
|||||||
import { createServer } from "node:net";
|
import { createServer } from "node:net";
|
||||||
import { join } from "node:path";
|
import { join } from "node:path";
|
||||||
import { Store } from "../../bus/src/store.mjs";
|
import { Store } from "../../bus/src/store.mjs";
|
||||||
import { Broker } from "../../bus/src/broker.mjs";
|
import { Broker, BusError } from "../../bus/src/broker.mjs";
|
||||||
import { serve } from "../../bus/src/server.mjs";
|
import { serve } from "../../bus/src/server.mjs";
|
||||||
import { EXIT, founderCheck, turnRequest } from "../src/runner.mjs";
|
import { EXIT, founderCheck, stateBlock, turnRequest } from "../src/runner.mjs";
|
||||||
import { scratch } from "./helpers.mjs";
|
import { scratch } from "./helpers.mjs";
|
||||||
|
|
||||||
const RUNNER = new URL("../src/runner.mjs", import.meta.url).pathname;
|
const RUNNER = new URL("../src/runner.mjs", import.meta.url).pathname;
|
||||||
@@ -56,6 +56,25 @@ test("turnRequest names the sender, class, reply and decision", () => {
|
|||||||
assert.equal(turnRequest({ id: "m4", from_role: "cto", class: "RESULT", request: null, body: "b" }), "Message m4 from cto, class RESULT:\n\nb");
|
assert.equal(turnRequest({ id: "m4", from_role: "cto", class: "RESULT", request: null, body: "b" }), "Message m4 from cto, class RESULT:\n\nb");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Row 109 (#1603): the header carries when the message was sent, not when it is read.
|
||||||
|
test("turnRequest shows the message's own sent time, a day old as well as new", () => {
|
||||||
|
const sent = "2026-10-12T09:30:00.000Z";
|
||||||
|
assert.equal(
|
||||||
|
turnRequest({ id: "m5", from_role: "jason", class: "REQUEST", at: sent, body: "resume" }),
|
||||||
|
`Message m5 from jason, sent ${sent}, class REQUEST:\n\nresume`,
|
||||||
|
);
|
||||||
|
assert.equal(turnRequest({ id: "m6", from_role: "cto", class: "RESULT", at: sent, in_reply_to: "m5", body: "b" }), `Message m6 from cto, sent ${sent}, class RESULT, in reply to m5:\n\nb`);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("stateBlock names the launch state with its time and event, and the open tasks with the cap", () => {
|
||||||
|
const read = "2026-10-13T10:00:00.000Z";
|
||||||
|
const restored = stateBlock({ launch: { state: "allowed", at: "2026-10-12T20:00:00.000Z", event: "ev-9" }, tasks: { open: [{ task_ref: "vikunja:3/7", title: "Seven" }], total: 1, cap: 50 } }, read);
|
||||||
|
assert.equal(restored, "Current state, read at 2026-10-13T10:00:00.000Z:\nLaunches: allowed since 2026-10-12T20:00:00.000Z (event ev-9).\nOpen tasks: 1.\n- vikunja:3/7 Seven\n");
|
||||||
|
const capped = stateBlock({ launch: { state: "allowed", at: null, event: null }, tasks: { open: [{ task_ref: "vikunja:3/1", title: "One" }], total: 9, cap: 1 } }, read);
|
||||||
|
assert.match(capped, /Launches: allowed \(never switched\)\./);
|
||||||
|
assert.match(capped, /Open tasks: 9, the first 1 of cap 1\./);
|
||||||
|
});
|
||||||
|
|
||||||
async function setup(t, { session = {}, policy = {}, tools } = {}) {
|
async function setup(t, { session = {}, policy = {}, tools } = {}) {
|
||||||
const dir = scratch(t);
|
const dir = scratch(t);
|
||||||
const store = new Store(dir);
|
const store = new Store(dir);
|
||||||
@@ -113,7 +132,8 @@ async function setup(t, { session = {}, policy = {}, tools } = {}) {
|
|||||||
const pm = broker.bindLaunch({ business: "demo", role: "pm", run: "pm-run", harness: "pi" });
|
const pm = broker.bindLaunch({ business: "demo", role: "pm", run: "pm-run", harness: "pi" });
|
||||||
const call = (c, verb, args = {}) => broker.request(c, { verb, args });
|
const call = (c, verb, args = {}) => broker.request(c, { verb, args });
|
||||||
call(pm, "role.claim");
|
call(pm, "role.claim");
|
||||||
return { dir, runDir, store, broker, server, cap, pm, call, launches, pids: join(dir, "pids") };
|
const human = broker.bindHuman({ business: "demo", human: "jason", via: "cli", outsideAgent: true });
|
||||||
|
return { dir, runDir, store, broker, server, cap, pm, human, call, launches, pids: join(dir, "pids") };
|
||||||
}
|
}
|
||||||
|
|
||||||
// A runner that hasn't exited after a minute is killed, so a test that
|
// A runner that hasn't exited after a minute is killed, so a test that
|
||||||
@@ -162,7 +182,8 @@ test("a message becomes a turn, the answer goes back as a RESULT, SIGTERM releas
|
|||||||
assert.equal(reply.from_role, "coder");
|
assert.equal(reply.from_role, "coder");
|
||||||
assert.equal(reply.from_run, "coder-run");
|
assert.equal(reply.from_run, "coder-run");
|
||||||
assert.equal(reply.in_reply_to, sent.id);
|
assert.equal(reply.in_reply_to, sent.id);
|
||||||
assert.equal(reply.body, `echo: Message ${sent.id} from pm, class REQUEST:\n\nhello`);
|
const sentAt = ctx.store.get("SELECT at FROM messages WHERE id=?", sent.id).at;
|
||||||
|
assert.match(reply.body, new RegExp(`^echo: Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\nMessage ${sent.id} from pm, sent ${sentAt}, class REQUEST:\n\nhello$`));
|
||||||
await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
|
await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
|
||||||
const err = join(ctx.runDir, "turns", "0001.stderr");
|
const err = join(ctx.runDir, "turns", "0001.stderr");
|
||||||
assert.equal(statSync(err).mode & 0o777, 0o600);
|
assert.equal(statSync(err).mode & 0o777, 0o600);
|
||||||
@@ -176,6 +197,85 @@ test("a message becomes a turn, the answer goes back as a RESULT, SIGTERM releas
|
|||||||
assert.ok(!existsSync(join(ctx.runDir, "tools.sock")));
|
assert.ok(!existsSync(join(ctx.runDir, "tools.sock")));
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// Row 109 (#1603): each turn reads the state when it starts. The fake adapter echoes the request,
|
||||||
|
// so a reply shows the block the turn was given.
|
||||||
|
const ask = async (ctx, body = "what now") => {
|
||||||
|
const before = ctx.inbox.length;
|
||||||
|
ctx.call(ctx.pm, "message.send", { to: "coder", body, class: "REQUEST" });
|
||||||
|
await until(() => replies(ctx)() && ctx.inbox.length > before);
|
||||||
|
return ctx.inbox.at(-1);
|
||||||
|
};
|
||||||
|
const snapshot = (i, over = {}, read = "2026-10-12T12:00:00.000Z") => ({
|
||||||
|
task_ref: `vikunja:3/${i}`,
|
||||||
|
updated: read,
|
||||||
|
digest: "d".repeat(64),
|
||||||
|
fields: { project_id: 3, title: `Task ${i}`, description: "", done: false, bucket: 11, ...over },
|
||||||
|
via: "board",
|
||||||
|
read_at: read,
|
||||||
|
});
|
||||||
|
|
||||||
|
test("launches revoked and then restored: the turn's block reads allowed and gives the restore time", async (t) => {
|
||||||
|
const ctx = await setup(t);
|
||||||
|
ctx.inbox = [];
|
||||||
|
const r = startRunner(ctx.runDir, ctx.cap);
|
||||||
|
t.after(() => r.child.kill("SIGKILL"));
|
||||||
|
ctx.call(ctx.human, "launch.revoke");
|
||||||
|
assert.match((await ask(ctx)).body, /\nLaunches: revoked since \S+ \(event \S+\)\.\n/);
|
||||||
|
ctx.call(ctx.human, "launch.restore");
|
||||||
|
const restored = ctx.store.get("SELECT id,at FROM events WHERE kind='launch.restored'");
|
||||||
|
assert.ok((await ask(ctx)).body.includes(`\nLaunches: allowed since ${restored.at} (event ${restored.id}).\n`));
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a task closed on the tracker is missing from the next turn's open list", async (t) => {
|
||||||
|
const ctx = await setup(t);
|
||||||
|
ctx.inbox = [];
|
||||||
|
ctx.broker.recordTask({ business: "demo", snapshots: [snapshot(1), snapshot(2)] });
|
||||||
|
const r = startRunner(ctx.runDir, ctx.cap);
|
||||||
|
t.after(() => r.child.kill("SIGKILL"));
|
||||||
|
const first = (await ask(ctx)).body;
|
||||||
|
assert.match(first, /\nOpen tasks: 2\.\n- vikunja:3\/1 Task 1\n- vikunja:3\/2 Task 2\n/);
|
||||||
|
ctx.broker.recordTask({ business: "demo", snapshots: [snapshot(2, { done: true }, "2026-10-12T13:00:00.000Z")] });
|
||||||
|
const second = (await ask(ctx)).body;
|
||||||
|
assert.match(second, /\nOpen tasks: 1\.\n- vikunja:3\/1 Task 1\n\nMessage /);
|
||||||
|
assert.ok(!second.includes("vikunja:3/2"));
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a message read a day after it was sent shows the time it was sent", async (t) => {
|
||||||
|
const ctx = await setup(t);
|
||||||
|
ctx.inbox = [];
|
||||||
|
const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "resume the launch", class: "REQUEST" });
|
||||||
|
const sentAt = ctx.store.get("SELECT at FROM messages WHERE id=?", sent.id).at;
|
||||||
|
// The broker's clock jumps a day before the runner reads: the header keeps the original time.
|
||||||
|
const now = Date.now;
|
||||||
|
Date.now = () => now() + 86_400_000;
|
||||||
|
t.after(() => (Date.now = now));
|
||||||
|
const r = startRunner(ctx.runDir, ctx.cap);
|
||||||
|
t.after(() => r.child.kill("SIGKILL"));
|
||||||
|
const [reply] = await until(replies(ctx));
|
||||||
|
assert.ok(reply.body.includes(`Message ${sent.id} from pm, sent ${sentAt}, class REQUEST:`));
|
||||||
|
assert.ok(Date.parse(reply.at) - Date.parse(sentAt) >= 86_000_000);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a failed state read fails the turn: no adapter runs, the sender hears why, the message is read", async (t) => {
|
||||||
|
const ctx = await setup(t);
|
||||||
|
ctx.inbox = [];
|
||||||
|
const request = ctx.broker.request.bind(ctx.broker);
|
||||||
|
ctx.broker.request = (cap, req) => {
|
||||||
|
if (req.verb === "state") throw new BusError("state-unavailable");
|
||||||
|
return request(cap, req);
|
||||||
|
};
|
||||||
|
const r = startRunner(ctx.runDir, ctx.cap);
|
||||||
|
t.after(() => r.child.kill("SIGKILL"));
|
||||||
|
const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "plan the week", class: "REQUEST" });
|
||||||
|
const [reply] = await until(replies(ctx));
|
||||||
|
assert.equal(reply.class, "RESULT");
|
||||||
|
assert.equal(reply.in_reply_to, sent.id);
|
||||||
|
assert.equal(reply.body, "The turn failed: could not read the current state: state-unavailable");
|
||||||
|
await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", sent.id));
|
||||||
|
assert.ok(!existsSync(join(ctx.runDir, "turns", "0001.stderr")));
|
||||||
|
assert.match(r.stderr(), /turn 1 for message \S+ failed: could not read the current state: state-unavailable/);
|
||||||
|
});
|
||||||
|
|
||||||
test("a SIGTERM before the claim stops the runner with exit 0 and no claim", async (t) => {
|
test("a SIGTERM before the claim stops the runner with exit 0 and no claim", async (t) => {
|
||||||
const ctx = await setup(t);
|
const ctx = await setup(t);
|
||||||
// session.json becomes a FIFO: main opens it right after its handlers go
|
// session.json becomes a FIFO: main opens it right after its handlers go
|
||||||
|
|||||||
@@ -214,7 +214,8 @@ async function session(t, { expose = () => [] } = {}) {
|
|||||||
writeFileSync(join(bundle, "tools.json"), "[]");
|
writeFileSync(join(bundle, "tools.json"), "[]");
|
||||||
const adapter = join(root, "adapter.sh");
|
const adapter = join(root, "adapter.sh");
|
||||||
writeFileSync(adapter, [
|
writeFileSync(adapter, [
|
||||||
`body=$(printf '%s\\n' "$MOSAIC_REQUEST" | sed '1,/^$/d')`,
|
// Row 109 (#1603): a state block opens the request, so the body is what follows the "Message ...:" header and its blank line.
|
||||||
|
`body=$(printf '%s\\n' "$MOSAIC_REQUEST" | sed '1,/^Message .*:$/d' | sed '1d')`,
|
||||||
`case $body in TOOL\\ *) exec '${process.execPath}' '${FAKE}' ;; esac`,
|
`case $body in TOOL\\ *) exec '${process.execPath}' '${FAKE}' ;; esac`,
|
||||||
`out=$(cd "$MOSAIC_WORKSPACE" && /bin/sh -c "$body" 2>&1); rc=$?`,
|
`out=$(cd "$MOSAIC_WORKSPACE" && /bin/sh -c "$body" 2>&1); rc=$?`,
|
||||||
`printf 'rc=%s\\n%s' "$rc" "$out"`,
|
`printf 'rc=%s\\n%s' "$rc" "$out"`,
|
||||||
|
|||||||
@@ -224,7 +224,9 @@ test("a session runs in its sandbox as pid 1 of its namespace, claims, answers,
|
|||||||
if (!reply) await new Promise((r) => setTimeout(r, 50));
|
if (!reply) await new Promise((r) => setTimeout(r, 50));
|
||||||
}
|
}
|
||||||
assert.ok(reply, "the session answered");
|
assert.ok(reply, "the session answered");
|
||||||
assert.match(reply.body, /^echo: Message .* from pm, class REQUEST:\n\nhello/);
|
// Row 109 (#1603): the turn opens with the state block, then the header with the time it was sent.
|
||||||
|
const sentAt = store.get("SELECT at FROM messages WHERE id=?", sent.id).at;
|
||||||
|
assert.match(reply.body, new RegExp(`^echo: Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\nMessage ${sent.id} from pm, sent ${sentAt}, class REQUEST:\n\nhello$`));
|
||||||
|
|
||||||
const registry = new Registry(dataRoot);
|
const registry = new Registry(dataRoot);
|
||||||
registry.add({ business: "acme", instance: "coder", run, pid: child.pid, startTime: startTimeOf(child.pid), runnerPid: runner.pid, runnerStartTime: runner.startTime });
|
registry.add({ business: "acme", instance: "coder", run, pid: child.pid, startTime: startTimeOf(child.pid), runnerPid: runner.pid, runnerStartTime: runner.startTime });
|
||||||
|
|||||||
@@ -53,5 +53,10 @@ cross-role. Raise it as a decision for the technical arbiter.
|
|||||||
|
|
||||||
- You act on the tracker and on Gitea only through the typed tools the
|
- You act on the tracker and on Gitea only through the typed tools the
|
||||||
broker gives you. You never hold a token.
|
broker gives you. You never hold a token.
|
||||||
|
- Each turn opens with a "Current state" block, read when the turn
|
||||||
|
started. It is the truth for launch state and task status. A message
|
||||||
|
or a note of yours that disagrees with it is out of date. Before you
|
||||||
|
plan to resume or launch work on a task, check the task with
|
||||||
|
`get_task`.
|
||||||
- Send messages through the bus, addressed to a role, not a person.
|
- Send messages through the bus, addressed to a role, not a person.
|
||||||
- Report a blocker as soon as you find it, with what you tried.
|
- Report a blocker as soon as you find it, with what you tried.
|
||||||
|
|||||||
Reference in New Issue
Block a user