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, packet 875c515e9. Filbert approved round 3, comment 28333, candidate manifest 0a62225e (11 paths). Sage gate on 9774d70db plus 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:
2026-10-13 19:12:54 -05:00
co-authored by Claude Opus 5.5
parent 848ceb68a4
commit badfa52b9d
11 changed files with 299 additions and 11 deletions
+2 -1
View File
@@ -11,7 +11,8 @@ import { appendFileSync, writeFileSync } from "node:fs";
import { callTool } from "../../src/tools.mjs";
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`);
process.stderr.write(`fake adapter: ${process.env.MOSAIC_MODEL}\n`);
const done = () => writeFileSync(process.env.MOSAIC_TURN_MARKER, "done\n");
+104 -4
View File
@@ -11,9 +11,9 @@ import { open } from "node:fs/promises";
import { createServer } from "node:net";
import { join } from "node:path";
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 { EXIT, founderCheck, turnRequest } from "../src/runner.mjs";
import { EXIT, founderCheck, stateBlock, turnRequest } from "../src/runner.mjs";
import { scratch } from "./helpers.mjs";
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");
});
// 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 } = {}) {
const dir = scratch(t);
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 call = (c, verb, args = {}) => broker.request(c, { verb, args });
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
@@ -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_run, "coder-run");
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));
const err = join(ctx.runDir, "turns", "0001.stderr");
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")));
});
// 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) => {
const ctx = await setup(t);
// session.json becomes a FIFO: main opens it right after its handlers go