diff --git a/docs/TOOLS.md b/docs/TOOLS.md index 6e02835e4..46a3084cc 100644 --- a/docs/TOOLS.md +++ b/docs/TOOLS.md @@ -150,7 +150,8 @@ Since #1631 the gate, its steps and the root-file rule live in `packages/seat/src/landing.mjs`, and the decision checks in `packages/bus/src/landing.mjs`. The bus host's `land` op uses both to land an agent's task on its working repository's `main`, never on `refactor` -(`packages/cli/README.md`, "Git socket"). A decision raised for one target +(`packages/cli/README.md`, "Git socket"). Since #1651 the host starts the +holder's landing run itself ("Landing runs"). A decision raised for one target can't land the other. land-work still accepts only a decision the human resolved through the CLI. @@ -276,7 +277,11 @@ unit's `MOSAIC_CLAUDE_BIN`, from `--claude PATH` or else `command -v claude` PID namespace with a generated bundle, and `--pm` launches the business's `launch.by` instance at start. A host that starts ends the launches a crashed host left as `host-lost` and refuses with exit 3 on a -`sessions.json` it can't read. `talk` sends a REQUEST and prints only the +`sessions.json` it can't read. Every 2 s it also starts each launch-block +instance that holds an approved, unused merge decision, once per decision, +with a landing turn that calls `land_work`; a holder run that started +before the approval is stopped between turns first (`land-relaunch` in the +launch log; row 156, cli README "Landing runs"). `talk` sends a REQUEST and prints only the reply to it, marking only that read (polls every 2 s, `--wait` default 900; `0` only sends); other messages stay unread for `mosaic inbox`, and `--follow` and `follow` print and read everything. `inbox --read ...` diff --git a/packages/bus/src/broker.mjs b/packages/bus/src/broker.mjs index 48394fa67..5b8385acd 100644 --- a/packages/bus/src/broker.mjs +++ b/packages/bus/src/broker.mjs @@ -175,6 +175,9 @@ export class Broker { #sessions = new Map(); #runs = new Map(); #ended = new Set(); + // Row 156 round 2 (#1651, Filbert's F1 in 29033): role runs that have asked message.receive. The + // runner asks it first after its landing turns, so a run not in here may be in one. + #received = new Set(); // Row 137 (#1631): landings checkLand let start, by ticket, until recordLand or landRefused ends them. #landings = new Map(); #secretCheck; @@ -979,19 +982,62 @@ export class Broker { this.#business(identifier(record.business)); identifier(record.role); if (record.thread !== null) identifier(record.thread); - for (const d of this.#store.all( - 'SELECT * FROM decisions WHERE business=? AND action=? AND raised_by_role=? ORDER BY seq', + return !this.#held(record.business, record.role, record.thread).next().done; + } + // Row 156 (#1651): every merge decision a non-thread instance holds as landHeld counts it, in + // decision order, with its task ref and resolution time. The launcher's landing pass starts the + // holder with these in the session file's landing. Trusted launcher API, not a socket verb. + landHolders(business) { + this.#business(identifier(business)); + return [...this.#held(business, null, null)].map(({ d, context, r }) => ({ + instance: d.raised_by_role, + decision: d.id, + task_ref: context.target, + resolved_at: r.at, + })); + } + // Row 156 (#1651): the landing pass's stop for a role run started before its decision was + // resolved, so it has no land_work for it. As threadIdle, it rechecks and releases in one + // synchronous call. Messages delivered to the run and not read are a turn in progress and keep + // it; messages still waiting stay undelivered and go to the next holder. by='land-relaunch'. + // A run that hasn't yet asked message.receive may be in a landing turn, which reads no message, + // so it is kept too (round 2, F1). A rebound run counts from its next receive. + landRelease(record) { + keys(record, ['business', 'run', 'decision'], ['business', 'run', 'decision']); + this.#business(identifier(record.business)); + const key = record.business + ':' + identifier(record.run); + const session = this.#runs.get(key) ?? fail('unknown-run'); + if (this.#ended.has(key)) fail('run-ended'); + if (session.thread) fail('thread-not-allowed'); + const h = this.#holder(record.business, session.role); + if (h?.op !== 'claim' || h.holder_run !== record.run) fail('not-holder'); + if (!this.landHolders(record.business).some((x) => x.instance === session.role && x.decision === record.decision)) + fail('decision-not-held'); + if (!this.#received.has(key)) return { released: false, landing: true }; + const { unread } = this.#store.get( + "SELECT count(*) AS unread FROM messages m JOIN deliveries d ON d.message=m.id AND d.op='delivered' WHERE m.business=? AND m.to_role=? AND d.holder_run=? AND NOT EXISTS(SELECT 1 FROM deliveries r WHERE r.message=m.id AND r.op='read')", record.business, - MERGE, - record.role, - )) { - if (d.class !== 'cross-role' || this.#raisedThread(d) !== record.thread) continue; + session.role, + record.run, + ); + if (unread) return { released: false, unread }; + this.#store.transaction(() => this.#claimEnd(session, h, 'release', null, null, 'land-relaunch')); + return { released: true }; + } + // The merge decisions landHeld would count, each with its raised context and closing event. A + // null role is every non-thread instance's (row 156): raised outside a thread by a role that + // isn't a thread instance. + *#held(business, role, thread) { + for (const d of role === null + ? this.#store.all('SELECT * FROM decisions WHERE business=? AND action=? ORDER BY seq', business, MERGE) + : this.#store.all('SELECT * FROM decisions WHERE business=? AND action=? AND raised_by_role=? ORDER BY seq', business, MERGE, role)) { + if (d.class !== 'cross-role' || this.#raisedThread(d) !== thread) continue; + if (role === null && this.#threaded(business, d.raised_by_role)) continue; const context = this.#raised(d.business, d.id); const r = this.#closed(d.id); if (context && taskRef(context.target) && r?.op === 'resolved' && r.choice === context.approvalChoice && !this.#spent(d.business, d.id)) - return true; + yield { d, context, r }; } - return false; } // Row 93: what the git socket's fetch for a reviewer may send, from the review.requested event the // reviewer cites. The caller must hold review.verdict and be the request's reviewer. Trusted host only. @@ -1203,17 +1249,25 @@ export class Broker { } } // Only the runner's bound session and delivered message identify a turn; no host path crosses the wire. + // Row 156 (#1651, sage-0808): a role run's landing turn is a merge decision id instead. Its begin + // needs the decision held for the run's role; its end needs that begin, since the landing may + // have spent the decision by then. async #requestUsage(s, verb, args) { try { keys(args, ['turn'], ['turn']); identifier(args.turn); + const phase = verb === 'usage.begin' ? 'begin' : 'end'; this.#store.transaction(() => { this.#agent(s); - if (!this.#store.get("SELECT 1 FROM messages m JOIN deliveries d ON d.message=m.id WHERE m.id=? AND m.business=? AND m.to_role=? AND d.op='delivered' AND d.holder_run=?", args.turn, s.business, s.role, s.run)) - fail('not-recipient'); + if (this.#store.get("SELECT 1 FROM messages m JOIN deliveries d ON d.message=m.id WHERE m.id=? AND m.business=? AND m.to_role=? AND d.op='delivered' AND d.holder_run=?", args.turn, s.business, s.role, s.run)) + return; + if (!s.thread && (phase === 'begin' + ? this.landHolders(s.business).some((h) => h.instance === s.role && h.decision === args.turn) + : this.#store.get("SELECT 1 FROM usage_records WHERE business=? AND run=? AND turn=? AND phase='begin'", s.business, s.run, args.turn))) + return; + fail('not-recipient'); }); if (!this.#usage) fail('usage-not-configured'); - const phase = verb === 'usage.begin' ? 'begin' : 'end'; try { return await this.#usage.collect(s, phase, args.turn); } finally { @@ -1787,7 +1841,10 @@ export class Broker { case 'message.receive': { keys(a, []); if (s.human) this.#human(s); - else if (!s.tester) this.#agent(s); + else if (!s.tester) { + this.#agent(s); + this.#received.add(s.business + ':' + s.run); + } const role = s.human ? 'human' : s.role; // A thread run receives its own thread's messages only (REQ-THREAD-1). const rows = this.#store.all( diff --git a/packages/bus/src/process.mjs b/packages/bus/src/process.mjs index ceaa38e40..ec19afead 100644 --- a/packages/bus/src/process.mjs +++ b/packages/bus/src/process.mjs @@ -18,6 +18,8 @@ const LAUNCH_OPS = new Set([ 'threadsState', 'threadIdle', 'mayLaunch', + 'landHolders', + 'landRelease', ]); function launchOp(m) { if (m.op === 'identity') return runtime.identity(m.cap); @@ -38,7 +40,14 @@ function launchOp(m) { if (m.op === 'threadIdle') return runtime.threadIdle(m.record); // Row 125 (#1619, REQ-LIMIT-4): the limit checks' answer before a launch (async). if (m.op === 'mayLaunch') return runtime.mayLaunch(m); - return runtime.credentialStatus(m.business, m.role); + // Row 156 (#1651): the launcher's landing pass: who holds a merge decision, and the stop of a + // role run started before its decision was resolved. + if (m.op === 'landHolders') return runtime.landHolders(m.business); + if (m.op === 'landRelease') return runtime.landRelease(m.record); + if (m.op === 'credentialStatus') return runtime.credentialStatus(m.business, m.role); + // Every LAUNCH_OPS name has its line above. A missing one used to answer as credentialStatus, + // whose [] for an unknown role read as "no holders" for landHolders (row 156 mutant P02). + throw new BusError('launch-op-refused'); } // The host's request id comes back on the reply, so a late reply can't // answer a later request (host.mjs). A message without one gets none. diff --git a/packages/bus/src/runtime.mjs b/packages/bus/src/runtime.mjs index ae4a3070e..3971bde8e 100644 --- a/packages/bus/src/runtime.mjs +++ b/packages/bus/src/runtime.mjs @@ -194,6 +194,9 @@ export async function startBroker({ recordLand: (ticket, land) => broker.recordLand(ticket, land), landRefused: (ticket, args) => broker.landRefused(ticket, args), landHeld: (record) => broker.landHeld(record), + // Row 156 (#1651): the launcher's landing pass. + landHolders: (business) => broker.landHolders(business), + landRelease: (r) => broker.landRelease(r), refuse: (cap, code, subject) => broker.refuse(cap, code, subject), credentialStatus, launches: bound, diff --git a/packages/bus/tests/land.test.mjs b/packages/bus/tests/land.test.mjs index 24618c57b..33d3cc222 100644 --- a/packages/bus/tests/land.test.mjs +++ b/packages/bus/tests/land.test.mjs @@ -12,10 +12,13 @@ import assert from 'node:assert/strict'; import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { fork } from 'node:child_process'; +import { once } from 'node:events'; import { Store } from '../src/store.mjs'; import { Broker } from '../src/broker.mjs'; import { readLanding } from '../src/landing.mjs'; import { fixture } from './spec-fixtures.mjs'; +import { createUsage, turnRecords } from '../src/usage.mjs'; const TASK = 'vikunja:1/2'; const MERGE = 'git.merge.protected'; @@ -41,11 +44,12 @@ const options = [ ]; const merge = { action: MERGE, target: TASK, question: 'Squash-merge work/vikunja-1-2 into main?', options, recommendation: 'yes', blocking: false }; -function setup(t, { arbiters, roles: r } = {}) { +// `extra(store, root, businesses)` adds Broker options, as the usage collector and limit checks. +function setup(t, { arbiters, roles: r, extra = () => ({}) } = {}) { const root = mkdtempSync(join(tmpdir(), 'bus-land-')); const store = new Store(root); const businesses = { demo: demo(arbiters, r) }; - const b = new Broker({ store, businesses }); + const b = new Broker({ store, businesses, ...extra(store, root, businesses) }); t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); @@ -84,7 +88,7 @@ function setup(t, { arbiters, roles: r } = {}) { const held = (role = 'merge-gate', thread = null) => b.landHeld({ business: 'demo', role, thread }); const check = (args) => b.checkLand(caps['merge-gate'], args); const routed = (d) => ({ ...store.get('SELECT class, route_to FROM decisions WHERE id=?', d.id) }); - return { b, store, businesses, caps, call, push, approve, raise, resolve, ready, land, events, merges, results, held, check, routed }; + return { b, store, root, businesses, caps, call, push, approve, raise, resolve, ready, land, events, merges, results, held, check, routed }; } test('REQ-DEC-2: a merge-gate raise its arbiter resolved lands through the host op: checkLand answers the facts and a ticket, records nothing, and recordLand spends the decision once', (t) => { @@ -353,6 +357,190 @@ test('landHeld is true only for an approved, unspent cross-role merge decision t assert.throws(() => w.b.landHeld({ business: 'nope', role: 'merge-gate', thread: null })); }); +// Row 156 (#1651): landHolders is landHeld's list for the launcher's landing pass. At each step +// of the landHeld test above, the two agree. +test('landHolders lists a merge decision exactly while landHeld is true for its non-thread instance, in decision order', (t) => { + const w = setup(t); + const holders = () => w.b.landHolders('demo'); + const agree = (msg) => assert.equal(holders().some((h) => h.instance === 'merge-gate'), w.held(), msg); + assert.deepEqual(holders(), []); + w.push(A); + w.approve(A); + const d = w.raise(); + agree('open'); + w.resolve(d, 'no'); + agree('declined'); + const notTask = w.raise('merge-gate', { target: 'main' }); + w.resolve(notTask); + agree('a target that is not a task ref'); + const gated = w.raise('coder'); + w.resolve(gated, 'yes', 'human'); + assert.deepEqual(holders(), [], 'gated'); + const first = w.raise(); + w.resolve(first); + const second = w.raise(); + w.resolve(second); + agree('approved'); + const at = (x) => w.store.get("SELECT at FROM decision_events WHERE decision=? AND op='resolved'", x.id).at; + const entry = (x) => ({ instance: 'merge-gate', decision: x.id, task_ref: TASK, resolved_at: at(x) }); + assert.deepEqual(holders(), [entry(first), entry(second)]); + // The second supersedes the first, so only the second lands. The first is never spent, and + // landHeld still counts it, so it stays listed (BUILD.md follow-up 5). + assert.throws(() => w.check({ task_ref: TASK, decision: first.id }), /superseded/); + w.b.recordLand(w.check({ task_ref: TASK, decision: second.id }).ticket, w.land); + assert.deepEqual(holders(), [entry(first)], 'spent'); + agree('spent'); + assert.throws(() => w.b.landHolders('nope'), /unknown-business/); + assert.throws(() => w.b.landHolders('no such'), /invalid-request/); +}); + +test("landHolders leaves out a thread instance's decision, even one raised before its role became one", (t) => { + const w = setup(t); + w.ready(); + assert.equal(w.b.landHolders('demo').length, 1); + // A restart with merge-gate now a thread instance: landHeld still answers for the role and + // thread asked, but the landing pass starts no non-thread run for it. + const roles2 = { ...roles, 'merge-gate': { ...roles['merge-gate'], definition: 'assistant' } }; + const b2 = new Broker({ store: w.store, businesses: { demo: demo(undefined, roles2) } }); + assert.equal(b2.landHeld({ business: 'demo', role: 'merge-gate', thread: null }), true); + assert.deepEqual(b2.landHolders('demo'), []); +}); + +// Row 156: the landing pass stops a role run started before its decision was resolved. Unread +// delivered messages keep it; waiting ones stay undelivered for the next holder. Round 2 (F1 in +// 29033): so does a run that hasn't asked message.receive yet, since it may be in a landing turn. +test('landRelease releases the holding run by land-relaunch once it has asked message.receive and nothing delivered to it is unread, and waiting messages go to the next holder', (t) => { + const w = setup(t); + const { d } = w.ready(); + const release = (extra = {}) => w.b.landRelease({ business: 'demo', run: 'merge-gate-run', decision: d.id, ...extra }); + assert.deepEqual(release(), { released: false, landing: true }, 'no receive yet'); + assert.deepEqual(release(), { released: false, landing: true }, 'asking again changes nothing'); + // The resolve in ready() sent merge-gate its DECISION message, so this receive delivers two. + const sent = w.call('human', 'message.send', { to: 'merge-gate', body: 'one' }); + const got = w.call('merge-gate', 'message.receive'); + assert.equal(got.length, 2); + assert.equal(got.filter((m) => m.id === sent.id).length, 1); + assert.deepEqual(release(), { released: false, unread: 2 }); + assert.equal(w.store.get("SELECT op FROM role_claims WHERE role='merge-gate' ORDER BY seq DESC LIMIT 1").op, 'claim', 'kept'); + w.call('merge-gate', 'message.read', { id: sent.id }); + assert.deepEqual(release(), { released: false, unread: 1 }); + for (const m of got) if (m.id !== sent.id) w.call('merge-gate', 'message.read', { id: m.id }); + const waiting = w.call('human', 'message.send', { to: 'merge-gate', body: 'two' }); + assert.throws(() => release({ decision: 'nope' }), /decision-not-held/); + assert.throws(() => release({ extra: 1 }), /invalid-request/); + assert.throws(() => w.b.landRelease({ business: 'demo', run: 'no-run', decision: d.id }), /unknown-run/); + assert.throws(() => w.b.landRelease({ business: 'demo', run: 'coder-run', decision: d.id }), /decision-not-held/); + // A run of the same instance that doesn't hold the claim releases nothing. + w.caps.other = w.b.bindLaunch({ business: 'demo', role: 'merge-gate', run: 'merge-gate-0', harness: 'pi' }); + assert.throws(() => w.b.landRelease({ business: 'demo', run: 'merge-gate-0', decision: d.id }), /not-holder/); + assert.deepEqual(release(), { released: true }); + assert.deepEqual( + { ...w.store.get("SELECT op, holder_run, \"by\", decision, reason FROM role_claims WHERE role='merge-gate' ORDER BY seq DESC LIMIT 1") }, + { op: 'release', holder_run: 'merge-gate-run', by: 'land-relaunch', decision: null, reason: null }, + ); + assert.throws(() => release(), /not-holder/, 'released once'); + assert.throws(() => w.call('merge-gate', 'message.receive'), /not-holder/); + w.b.endLaunch({ business: 'demo', run: 'merge-gate-run', reason: 'exited' }); + assert.throws(() => release(), /run-ended/); + assert.equal(w.store.get("SELECT count(*) AS n FROM deliveries WHERE message=? AND op='delivered'", waiting.id).n, 0, 'still waiting'); + // The decision is still held, and the next run gets the waiting message. + assert.equal(w.held(), true); + w.caps.next = w.b.bindLaunch({ business: 'demo', role: 'merge-gate', run: 'merge-gate-2', harness: 'pi' }); + w.call('next', 'role.claim'); + assert.deepEqual(w.call('next', 'message.receive').map((m) => m.id), [waiting.id]); + assert.equal(w.store.get("SELECT holder_run FROM deliveries WHERE message=? AND op='delivered'", waiting.id).holder_run, 'merge-gate-2'); +}); + +// Row 156 round 2 (Sage's F3 ruling on 29033): the run the landing pass starts after the +// resolution lands in its landing turn, before it has asked message.receive. Its land_work reaches +// checkLand through the git socket with the run's capability; here the call is made as that run. +test('a landing run started after the resolution passes checkLand for its held decision before any message.receive, and lands it', (t) => { + const w = setup(t); + const { d, v } = w.ready(); + w.b.endLaunch({ business: 'demo', run: 'merge-gate-run', reason: 'exited' }); + assert.deepEqual(w.b.landHolders('demo').map(({ instance, decision, task_ref }) => ({ instance, decision, task_ref })), [{ instance: 'merge-gate', decision: d.id, task_ref: TASK }]); + const cap = w.b.bindLaunch({ business: 'demo', role: 'merge-gate', run: 'merge-gate-land', harness: 'pi' }); + w.b.request(cap, { verb: 'role.claim', args: {} }); + assert.deepEqual(w.b.landRelease({ business: 'demo', run: 'merge-gate-land', decision: d.id }), { released: false, landing: true }); + const { ticket, resolved_at, ...facts } = w.b.checkLand(cap, { task_ref: TASK, decision: d.id }); + assert.match(ticket, /^[0-9a-f-]{36}$/); + assert.deepEqual([facts.decision, facts.commit, facts.verdict, facts.choice], [d.id, A, v.verdict, 'yes']); + assert.deepEqual(w.b.recordLand(ticket, w.land), { task_ref: TASK, commit: B }); + const [e] = w.merges(); + assert.deepEqual([e.actor_run, e.body.decision, e.body.commit], ['merge-gate-land', d.id, B]); + // Spent: no longer listed, and nothing was delivered to the landing run. The claim routed the + // waiting messages to it as their holder; none was read. + assert.deepEqual(w.b.landHolders('demo'), []); + assert.equal(w.store.get("SELECT count(*) AS n FROM deliveries WHERE holder_run='merge-gate-land' AND op='delivered'").n, 0); +}); + +// Row 156 (sage-0933): the bus process answers both launch ops from its own launchOp line. A +// LAUNCH_OPS name without one refuses launch-op-refused; before row 156 it answered as +// credentialStatus, whose [] for landHolders read as "no holders" (mutant P02). +test('the bus process answers landHolders and landRelease from their own lines, not as credentialStatus', async (t) => { + const root = mkdtempSync(join(tmpdir(), 'bus-land-ops-')); + t.after(() => rmSync(root, { recursive: true, force: true })); + const child = fork(new URL('../src/process.mjs', import.meta.url), [], { stdio: ['ignore', 'ignore', 'ignore', 'ipc'] }); + t.after(() => { + if (child.exitCode === null) child.kill('SIGKILL'); + }); + const ask = async (m) => { + const reply = once(child, 'message'); + child.send(m); + return (await reply)[0]; + }; + const businesses = { demo: { id: 'demo', human: 'jason', arbiters: { technical: 'pm', delivery: 'pm' }, roles } }; + assert.equal((await ask({ op: 'boot', config: { dataRoot: root, businesses, launches: [], readers: [] } })).ok, true); + // Without its launchOp line each would get launch-op-refused, not the broker's own answer. + assert.deepEqual(await ask({ op: 'landHolders', business: 'demo', id: 1 }), { ok: true, result: [], id: 1 }, 'landHolders, not launch-op-refused'); + assert.deepEqual( + await ask({ op: 'landRelease', record: { business: 'demo', run: 'no-run', decision: 'd1' }, id: 2 }), + { ok: false, error: 'unknown-run', id: 2 }, + 'landRelease, not launch-op-refused', + ); + assert.deepEqual(await ask({ op: 'credentialStatus', business: 'demo', role: 'pm', id: 3 }), { ok: true, result: [], id: 3 }); + const exit = once(child, 'exit'); + child.send({ op: 'close' }); + assert.equal((await exit)[0], 0); +}); + +// Row 156 (sage-0808): a role run's landing turn brackets its usage with the decision id. The +// begin needs the decision held for the run's instance; the end needs that begin. +test('usage.begin with a decision id needs it held for the run\'s instance, usage.end needs the begin, and the limit checks see the landing turn', async (t) => { + const asked = []; + const w = setup(t, { + extra: (store, root, businesses) => { + const usage = createUsage({ store, dataRoot: root, businesses, log: () => {} }); + return { + usage: { collect: (...a) => usage.then((u) => u.collect(...a)), view: () => ({}) }, + limits: { mustStop: async (business, subject) => { asked.push({ subject, turns: turnRecords(store, business) }); return { stop: false }; } }, + }; + }, + }); + const usage = (who, phase, turn) => w.call(who, `usage.${phase}`, { turn }); + const open = w.raise('merge-gate'); + await assert.rejects(usage('merge-gate', 'begin', open.id), { code: 'not-recipient' }, 'open'); + w.push(A); + w.approve(A); + w.resolve(open); + await assert.rejects(usage('pm', 'begin', open.id), { code: 'not-recipient' }, "another instance's"); + await assert.rejects(usage('merge-gate', 'end', open.id), { code: 'not-recipient' }, 'no begin'); + assert.deepEqual(await usage('merge-gate', 'begin', open.id), { recorded: true }); + // The landing spends the decision inside the turn; the end still records. + w.b.recordLand(w.check({ task_ref: TASK, decision: open.id }).ticket, w.land); + await assert.rejects(usage('pm', 'end', open.id), { code: 'not-recipient' }, "another run's begin"); + assert.deepEqual(await usage('merge-gate', 'end', open.id), { recorded: true }); + assert.deepEqual( + w.store.all('SELECT run, instance, turn, phase FROM usage_records ORDER BY seq').map((r) => ({ ...r })), + ['begin', 'end'].map((phase) => ({ run: 'merge-gate-run', instance: 'merge-gate', turn: open.id, phase })), + ); + assert.equal(asked.length, 1); + assert.deepEqual(asked[0].subject, { instance: 'merge-gate', provider: null, model: null }); + assert.deepEqual(asked[0].turns.map((r) => [r.run, r.turn, r.instance]), [['merge-gate-run', open.id, 'merge-gate']]); + // Spent, a second landing turn's begin refuses. + await assert.rejects(usage('merge-gate', 'begin', open.id), { code: 'not-recipient' }, 'spent'); +}); + test('no run may write the action.allowed for git.merge.protected itself', (t) => { const w = setup(t); assert.throws( diff --git a/packages/cli/README.md b/packages/cli/README.md index 38a2ac7a4..20f3fe739 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -459,6 +459,50 @@ Thread runs of one instance share its workspace, `workspaces///`. The clone check runs at each start, so it can fetch while another thread's session is running. +### Landing runs (row 156, #1651) + +Every launch of a non-thread instance asks the broker's `landHolders` +(launch op) which approved, unspent merge decisions each non-thread +instance holds, with each decision's task ref. The instance's own entries +go into its `session.json` as `landing: [{decision, task_ref}]`, and the +same answer decides `land_work` and the git socket, so a landing entry +never comes without the tool. The key is left out when there is none. The +runner takes one turn per entry before its receive loop +(`packages/harness/README.md`). A thread launch still asks `landHeld` and +gets no `landing`. + +After the thread pass, every tick asks `landHolders` again and acts on +each holder that `launch.instances` lists: + +- A holder with no running session starts, through the checks in the + table above, as the human, once per decision. A run that ends without + landing isn't started again for that decision. A refused start is + logged once per code per decision as `decision (): + landing start refused: `, with a `refused` record that carries + `decision`, and the next tick tries again. A landing turn whose state + read fails is skipped (the runner logs it). The run still counts as + started for that decision, so this host doesn't start it again, and + the role-file steps below apply. +- A running holder whose `landing` doesn't name the decision started + before the resolution, so it has no `land_work` for it. The launcher + asks the broker's `landRelease` to end it. The broker releases the + role only between turns, with nothing delivered and unread, by + `land-relaunch`; a message still waiting stays undelivered for the next + run. A landing turn reads no message, so the broker counts a run as + past its landing turns only once it has asked `message.receive`, which + the runner first does after them. While a landing turn or a message + turn runs, or a delivered message is unread, nothing is released and a + later tick asks again. A refusal is logged once per code + per run as `run : landRelease refused: `. After a release the + launch log gets a `land-relaunch` record (`instance`, `run`, + `decision`) and the runner gets SIGTERM, as `mosaic stop` sends it. The + next tick starts the instance with the decision in its `landing`. + +The launcher keeps none of this on disk. A host that restarts ends the +earlier host's runs (`host-lost`), so its first tick starts each holder. +If the host doesn't start a landing run, row 149's steps in the role +files still apply: the chief of staff ends the raising run and launches. + ### Git socket The launcher also serves `/bus-host/git.sock` (0600, @@ -468,8 +512,9 @@ commit it was asked to review (`push_working` and `fetch_review`, `packages/harness/README.md`). A role granted git.merge.protected cross-role whose merge decision was approved, by its arbiter or the human, lands its task on the repository's `main` (`land_work`, row 137, #1631). The sandbox binds the socket only for a role that holds -git.push.working or review.verdict, or one the broker's `landHeld` says -holds a merge decision it could land. A session's `session.json` names it +git.push.working or review.verdict, or one that holds a merge decision it +could land: `landHolders` lists it (a non-thread launch) or `landHeld` says +so (a thread launch). A session's `session.json` names it as `gitSocket` only then. Only a git bundle crosses. The host never runs git in a session's clone diff --git a/packages/cli/src/launcher.mjs b/packages/cli/src/launcher.mjs index deaae7840..f9271003a 100644 --- a/packages/cli/src/launcher.mjs +++ b/packages/cli/src/launcher.mjs @@ -65,6 +65,28 @@ // 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). +// +// Landing runs (row 156, #1651). Every launch of a non-thread instance +// reads the broker's landHolders (launch op): the approved, unspent merge +// decisions each non-thread instance raised. The instance's entries go into +// session.json as `landing: [{decision, task_ref}]`, and the same read +// decides land_work, so a landing turn always comes with the tool. The +// runner takes one turn per entry before its receive loop. After the +// thread pass, every tick also: +// +// - starts a holder that launch.instances lists and that isn't running, +// through check() as any launch, once per decision. A run that ended +// without landing isn't started again, and a refusal is logged once per +// code and isn't a start; +// - stops a running holder whose landing doesn't name the decision (it +// started before the resolution, so it has no land_work for it). The +// broker decides (launch op landRelease): it releases the role only +// between turns, with nothing delivered and unread, and leaves waiting +// messages undelivered for the next run. The SIGTERM goes only after it +// releases, and the next tick starts the instance. +// +// Nothing here is persisted. After a restart recover() ends the earlier +// host's runs, so the first tick starts each holder. import { spawnSync } from "node:child_process"; import { createServer } from "node:net"; @@ -174,6 +196,13 @@ export function createLauncher({ const stopping = new Set(); const refusedThreads = new Map(); const startedFor = new Map(); + // Row 156: run -> the decisions its landing names; decision -> started; decision -> last + // refusal code; run -> last landRelease refusal code; and landHolders' last refusal. + const landingOf = new Map(); + const landStarted = new Set(); + const refusedLand = new Map(); + const refusedRelease = new Map(); + let refusedHolders = null; const sockets = new Set(); const gitSocket = createGitSocket({ host, business: b, dataRoot, env, log, ...git }); @@ -273,7 +302,12 @@ export function createLauncher({ const v = versions ?? { pi: piVersion(), claude: model.harness === "claude-code" ? claudeVersion(env) : null }; // Row 137 (#1631): a role and thread holding an approved, unspent merge decision get land_work. // The broker checks the decision again on every land; this only decides what the bundle shows. - const land = await host.op({ op: "landHeld", record: { business: b.id, role: instance, thread } }); + // Row 156 (#1651): a non-thread instance's land and landing come from one landHolders read, so a + // decision resolved between two reads can't give a landing turn without land_work. + const holders = thread ? null : await host.op({ op: "landHolders", business: b.id }); + if (holders !== null && !Array.isArray(holders)) throw new Refusal("launch-failed", "landHolders: malformed answer"); + const holding = holders?.filter((h) => h.instance === instance) ?? null; + const land = thread ? await host.op({ op: "landHeld", record: { business: b.id, role: instance, thread } }) : holding.length > 0; const bundle = buildBundle({ dir: join(runDir, "bundle"), resolved, @@ -310,6 +344,7 @@ export function createLauncher({ bundle: files, toolSocket, turnMarker, + ...(holding?.length ? { landing: holding.map(({ decision, task_ref }) => ({ decision, task_ref })) } : {}), }); const paths = { @@ -341,6 +376,9 @@ export function createLauncher({ pid: child.pid, startTime: outerStart, runnerPid: runner.pid, runnerStartTime: runner.startTime, runDir, startedAt: new Date().toISOString(), }; registry.add(entry); + // Row 156: one start per decision, whichever launch wrote it into landing. + landingOf.set(run, new Set(holding?.map((h) => h.decision) ?? [])); + for (const h of holding ?? []) landStarted.add(h.decision); child.stdin.end(`${JSON.stringify({ cap: launch.cap })}\n`); } catch (e) { child.kill("SIGKILL"); @@ -362,75 +400,138 @@ export function createLauncher({ } finally { children.delete(run); stopping.delete(run); + landingOf.delete(run); + refusedRelease.delete(run); } }); 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. + // Row 121: one tick of thread starts and idle stops, then (row 156) the + // landing pass. 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 (!stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) { - let idle; - try { - idle = await host.op({ op: "threadIdle", record: { business: b.id, thread: t.thread, run: s.run } }); - } catch (e) { - log(`run ${s.run}: threadIdle refused: ${e.code ?? e.message}`); - continue; - } - // A turn in progress, or a message since the snapshot: the run keeps going. - if (!idle.released) continue; - 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; + await threadPass(); + await landPass(); + } finally { + ticking = false; + } + } + + async function threadPass() { + 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 (!stopping.has(s.run) && now - last >= b.launch.threadIdleSeconds * 1000) { + let idle; try { - await start({ instance: t.instance, launchedBy: b.human, thread: t.thread, ...(await check(t.instance, { thread: t.thread })) }); - startedFor.set(t.thread, signature); - refusedThreads.delete(t.thread); + idle = await host.op({ op: "threadIdle", record: { business: b.id, thread: t.thread, run: s.run } }); } 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}:{}) }); + log(`run ${s.run}: threadIdle refused: ${e.code ?? e.message}`); + continue; } - }); + // A turn in progress, or a message since the snapshot: the run keeps going. + if (!idle.released) continue; + 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; } - } finally { - ticking = false; + 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, ...(await 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}:{}) }); + } + }); + } + } + + // Row 156 (#1651): start each listed merge decision holder once, and stop a + // running holder that started before its decision's resolution. + async function landPass() { + if (!b.launch) return; + let holders; + try { + holders = await host.op({ op: "landHolders", business: b.id }); + if (!Array.isArray(holders)) throw new Refusal("malformed-answer"); + refusedHolders = null; + } catch (e) { + const code = e.code ?? e.message; + if (refusedHolders !== code) log(`landHolders refused: ${code}`); + refusedHolders = code; + return; + } + for (const h of holders) { + if (closed) return; + if (!b.launch.instances.includes(h.instance)) continue; + const s = registry.running(b.id, h.instance, null); + if (s) { + if (stopping.has(s.run) || landingOf.get(s.run)?.has(h.decision)) continue; + let r; + try { + r = await host.op({ op: "landRelease", record: { business: b.id, run: s.run, decision: h.decision } }); + } catch (e) { + const code = e.code ?? e.message; + if (refusedRelease.get(s.run) !== code) log(`run ${s.run}: landRelease refused: ${code}`); + refusedRelease.set(s.run, code); + continue; + } + // A landing turn or a message turn in progress: a later tick. + if (!r?.released) continue; + stopping.add(s.run); + record({ event: "land-relaunch", instance: h.instance, run: s.run, decision: h.decision }); + try { + if (startTimeOf(s.runnerPid) === s.runnerStartTime) process.kill(s.runnerPid, "SIGTERM"); + } catch {} + continue; + } + if (landStarted.has(h.decision)) continue; + await serialize(async () => { + if (closed || registry.running(b.id, h.instance, null)) return; + try { + await start({ instance: h.instance, launchedBy: b.human, ...(await check(h.instance)) }); + refusedLand.delete(h.decision); + } catch (e) { + const code = e instanceof Refusal ? e.code : "launch-failed"; + if (refusedLand.get(h.decision) === code) return; + refusedLand.set(h.decision, code); + log(`decision ${h.decision} (${h.instance}): landing start refused: ${code}`); + record({ event: "refused", by: b.human, instance: h.instance, decision: h.decision, code, why: e.message, ...(e.limit ? { limit: e.limit } : {}) }); + } + }); } } diff --git a/packages/cli/tests/launcher.test.mjs b/packages/cli/tests/launcher.test.mjs index 46a7e84a9..e7cd64b25 100644 --- a/packages/cli/tests/launcher.test.mjs +++ b/packages/cli/tests/launcher.test.mjs @@ -384,33 +384,221 @@ test("each launch makes sure the workspace has a clone of the project: cloned, f }); // Row 137 (#1631): the chief holds neither git.push.working nor -// review.verdict, so only landHeld can give it land_work and the git socket. -// landHeld is the broker's (packages/bus/tests/land.test.mjs); here the host -// answers it, since a real answer needs a merge decision the human resolved -// through the CLI. -test("a role the broker says holds a merge decision gets land_work, its prompt line and the git socket; without it, none", { skip, timeout: 30000 }, async (t) => { +// review.verdict, so only a held merge decision can give it land_work and the +// git socket. Row 156 (#1651): for a non-thread launch that read is +// landHolders, which also gives session.json its landing. Both are the +// broker's (packages/bus/tests/land.test.mjs); here the host answers, since a +// real answer needs a merge decision the human resolved through the CLI. +test("a role the broker says holds a merge decision gets land_work, its prompt line, its landing and the git socket; without it, none", { skip, timeout: 30000 }, async (t) => { const asked = []; - const answering = (held) => ({ wrap: (host) => Object.create(host, { op: { value: async (m) => (m.op === "landHeld" ? (asked.push(m.record), held) : host.op(m)) } }) }); + const held = [{ instance: "chief", decision: "d1", task_ref: "vikunja:1/2" }, { instance: "coder", decision: "d2", task_ref: "vikunja:1/3" }]; + const answering = (holders) => ({ wrap: (host) => Object.create(host, { op: { value: async (m) => (m.op === "landHolders" ? (asked.push(m), holders) : host.op(m)) } }) }); const landing = "- Landing: once your merge decision is approved, land_work it with the task."; const read = (f, run, file) => readFileSync(join(f.dataRoot, "launches", "acme", run, file), "utf8"); - const one = await setup(t, undefined, undefined, answering(true)); + const one = await setup(t, undefined, undefined, answering(held)); const pm1 = await one.launcher.launchPm(); - assert.deepEqual(asked, [{ business: "acme", role: "chief", thread: null }]); + assert.deepEqual(asked, [{ op: "landHolders", business: "acme" }]); const tool = JSON.parse(read(one.f, pm1.run, "bundle/tools.json")).find((x) => x.name === "land_work"); assert.equal(tool.verb, "working.land"); assert.equal(tool.action, "git.merge.protected"); assert.deepEqual(tool.parameters.required, ["task_ref", "decision"]); assert.ok(read(one.f, pm1.run, "bundle/prompt.md").includes(landing)); assert.equal(JSON.parse(read(one.f, pm1.run, "session.json")).gitSocket, gitSocketPath(one.f.dataRoot)); + // Only the chief's own entry, from the same read as land_work. + assert.deepEqual(JSON.parse(read(one.f, pm1.run, "session.json")).landing, [{ decision: "d1", task_ref: "vikunja:1/2" }]); - // A separate fixture, so its business and data root are new. - const two = await setup(t, undefined, undefined, answering(false)); + // A separate fixture, so its business and data root are new. Another + // instance's decision gives the chief nothing. + const two = await setup(t, undefined, undefined, answering(held.slice(1))); const pm2 = await two.launcher.launchPm(); assert.equal(asked.length, 2); assert.equal(JSON.parse(read(two.f, pm2.run, "bundle/tools.json")).some((x) => x.name === "land_work"), false); assert.equal(read(two.f, pm2.run, "bundle/prompt.md").includes(landing), false); assert.equal(JSON.parse(read(two.f, pm2.run, "session.json")).gitSocket, null); + assert.equal("landing" in JSON.parse(read(two.f, pm2.run, "session.json")), false); +}); + +// Row 156 (#1651): the landing pass. The host answers landHolders and +// landRelease from `state`, as above; the broker's side of both is in +// packages/bus/tests/land.test.mjs. With no state.release, landRelease goes +// to the real host. The tick timer is an hour, so only the test's own tick() +// calls run a pass. The coder is listed in launch.instances and the chief +// isn't. +const LANDING = "- Landing: once your merge decision is approved, land_work it with the task."; +function landingHost(state) { + const op = (host) => async (m) => { + if (m.op === "landHolders") return state.holders; + if (m.op === "landRelease") return (state.released.push(m.record), state.release ?? host.op(m)); + return host.op(m); + }; + return { threadTickMs: 3600000, wrap: (host) => Object.create(host, { op: { value: op(host) } }) }; +} +const runFile = (f, run, file) => readFileSync(join(f.dataRoot, "launches", "acme", run, file), "utf8"); +const runnerText = (f, run) => (existsSync(join(f.dataRoot, "launches", "acme", run, "runner.log")) ? runFile(f, run, "runner.log") : ""); +const landingOf = (f, run) => JSON.parse(runFile(f, run, "session.json")).landing; +const hasLand = (f, run) => JSON.parse(runFile(f, run, "bundle/tools.json")).some((x) => x.name === "land_work"); +const coderRuns = (f) => log(f).filter((e) => e.event === "launch" && e.instance === "coder").map((e) => e.run); +const D1 = { instance: "coder", decision: "d1", task_ref: "vikunja:1/2" }; + +test("REQ-LAUNCH-1 the tick starts a listed merge decision holder once, with land_work, the Landing line and its landing turn; an unlisted holder starts nothing", { skip, timeout: 30000 }, async (t) => { + const state = { holders: [{ instance: "chief", decision: "d0", task_ref: "vikunja:1/1" }, D1], release: { released: false, unread: 0 }, released: [] }; + const { f, launcher } = await setup(t, undefined, undefined, landingHost(state)); + await launcher.tick(); + assert.deepEqual(log(f).filter((e) => e.event === "launch").map((e) => [e.instance, e.launchedBy]), [["coder", "jason"]]); + const [run] = coderRuns(f); + assert.equal(hasLand(f, run), true); + assert.ok(runFile(f, run, "bundle/prompt.md").includes(LANDING)); + assert.deepEqual(landingOf(f, run), [{ decision: "d1", task_ref: "vikunja:1/2" }]); + await until(() => runnerText(f, run).includes("runner: landing turn 1 for decision d1 (vikunja:1/2): ok")); + + // Running with the decision in its landing: no second start, no release. + await launcher.tick(); + await launcher.tick(); + assert.deepEqual(coderRuns(f), [run]); + assert.deepEqual(state.released, []); + assert.equal(log(f).some((e) => e.event === "refused"), false); +}); + +test("a holder running since before the resolution is stopped once the broker releases it, land-relaunch, and the next tick starts it: landing turn first, then the message that waited", { skip, timeout: 45000 }, async (t) => { + const state = { holders: [], released: [] }; + const { f, host, launcher, client, logs } = await setup(t, undefined, undefined, landingHost(state)); + const chief = await host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) }); + await new Client({ path: host.path, cap: chief.cap }).call("role.claim"); + const launched = await launcher.fromSession(chief.cap, "coder"); + assert.equal(launched.ok, true, JSON.stringify(launched)); + const old = launched.result.run; + assert.equal(landingOf(f, old), undefined); + assert.equal(hasLand(f, old), false); + await until(() => runnerText(f, old).includes("claimed by run")); + + // Resolved now. The real broker holds no such decision and refuses through + // the broker process; the launcher logs it once, and nothing stops. + state.holders = [D1]; + await launcher.tick(); + await launcher.tick(); + assert.deepEqual(state.released, [{ business: "acme", run: old, decision: "d1" }, { business: "acme", run: old, decision: "d1" }]); + assert.deepEqual(logs.filter((l) => l.includes("landRelease refused")), [`run ${old}: landRelease refused: decision-not-held`]); + // The broker keeps the claim (a message delivered and unread): nothing stops. + state.release = { released: false, unread: 1 }; + await launcher.tick(); + assert.equal(state.released.length, 3); + assert.equal(log(f).some((e) => e.event === "land-relaunch"), false); + assert.deepEqual(coderRuns(f), [old]); + assert.equal(readSessions(f.dataRoot).find((s) => s.run === old)?.live, true); + // Row 156 round 2 (F1 in 29033): the broker keeps the claim while the run is in its landing + // turns (it hasn't asked message.receive yet). Nothing stops either; runner.test holds a real + // landing turn open against the broker for this answer. + state.release = { released: false, landing: true }; + await launcher.tick(); + assert.equal(state.released.length, 4); + assert.equal(log(f).some((e) => e.event === "land-relaunch"), false); + assert.deepEqual(coderRuns(f), [old]); + assert.equal(readSessions(f.dataRoot).find((s) => s.run === old)?.live, true); + + // Released: one land-relaunch record, a SIGTERM, and no start while it ends. + state.release = { released: true }; + await launcher.tick(); + assert.equal(state.released.length, 5); + assert.deepEqual(log(f).filter((e) => e.event === "land-relaunch").map(({ event, instance, run, decision }) => ({ event, instance, run, decision })), [{ event: "land-relaunch", instance: "coder", run: old, decision: "d1" }]); + const end = await until(() => log(f).find((e) => e.event === "end" && e.run === old)); + assert.deepEqual([end.reason, end.exitCode], ["stopped", 0]); + assert.deepEqual(coderRuns(f), [old]); + + // A message for the coder waits with no run. The next tick starts the + // holder with land_work and its landing; the landing turn is turn 1 and the + // message is answered by the new run after it. + const sent = await client.call("message.send", { to: "coder", class: "REQUEST", body: "hello" }); + await launcher.tick(); + const [, next] = coderRuns(f); + assert.ok(next); + assert.equal(hasLand(f, next), true); + assert.deepEqual(landingOf(f, next), [{ decision: "d1", task_ref: "vikunja:1/2" }]); + const reply = await until(async () => (await client.call("message.receive")).find((m) => m.in_reply_to === sent.id)); + assert.equal(reply.from_run, next); + assert.match(runnerText(f, next), /runner: landing turn 1 for decision d1 \(vikunja:1\/2\): ok/); + + // Started after the resolution: left alone. + await launcher.tick(); + assert.equal(state.released.length, 5); + assert.deepEqual(coderRuns(f), [old, next]); +}); + +test("a landing run that ended isn't started again for its decision, a spent decision starts nothing, and a new decision starts the holder", { skip, timeout: 45000 }, async (t) => { + const state = { holders: [D1], release: { released: false, unread: 0 }, released: [] }; + const { f, launcher } = await setup(t, undefined, undefined, landingHost(state)); + await launcher.tick(); + const [first] = coderRuns(f); + await until(() => runnerText(f, first).includes("landing turn 1 for decision d1")); + const s = readSessions(f.dataRoot).find((x) => x.run === first); + process.kill(s.runnerPid, "SIGTERM"); + await until(() => log(f).find((e) => e.event === "end" && e.run === first)); + + // Still listed, as if the run ended without landing: no second start. + await launcher.tick(); + assert.deepEqual(coderRuns(f), [first]); + // Spent: not listed. + state.holders = []; + await launcher.tick(); + assert.deepEqual(coderRuns(f), [first]); + // A new decision for the same instance starts it once more. + state.holders = [{ ...D1, decision: "d2" }]; + await launcher.tick(); + const runs = coderRuns(f); + assert.equal(runs.length, 2); + assert.deepEqual(landingOf(f, runs[1]), [{ decision: "d2", task_ref: "vikunja:1/2" }]); + assert.deepEqual(state.released, []); +}); + +test("after a restart, the new host's first tick starts a holder whose decision is still held", { skip, timeout: 45000 }, async (t) => { + const given = prepare(t); + const state = { holders: [D1], release: { released: false, unread: 0 }, released: [] }; + const one = await setup(t, undefined, given, landingHost(state)); + await one.launcher.tick(); + const [first] = coderRuns(given.f); + await until(() => runnerText(given.f, first).includes("landing turn 1 for decision d1")); + await one.client.call("role.release"); // the test's own cto claim, so the next host's can claim again + assert.equal(await one.close(), 0); + assert.deepEqual(readSessions(given.f.dataRoot), []); + + const two = await setup(t, undefined, given, landingHost(state)); + assert.deepEqual(coderRuns(given.f), [first]); + await two.launcher.tick(); + const runs = coderRuns(given.f); + assert.equal(runs.length, 2); + assert.equal(hasLand(given.f, runs[1]), true); + assert.deepEqual(landingOf(given.f, runs[1]), [{ decision: "d1", task_ref: "vikunja:1/2" }]); + await until(() => runnerText(given.f, runs[1]).includes("landing turn 1 for decision d1 (vikunja:1/2): ok")); + await two.client.call("role.release"); + assert.equal(await two.close(), 0); +}); + +// The reviewer and the coder share sonnet's one place, so a running coder +// makes the reviewer's landing start refuse capacity-full. +test("a refused landing start is logged once per code with its decision, isn't a start, and the next tick with room starts the holder", { skip, timeout: 45000 }, async (t) => { + const state = { holders: [], release: { released: false, unread: 0 }, released: [] }; + const { f, host, launcher, logs } = await setup(t, undefined, undefined, landingHost(state)); + const chief = await host.bindLaunch({ business: "acme", role: "chief", run: "chief-test", harness: "pi", pid: process.pid, startTime: startTimeOf(process.pid) }); + await new Client({ path: host.path, cap: chief.cap }).call("role.claim"); + const coder = await launcher.fromSession(chief.cap, "coder"); + assert.equal(coder.ok, true, JSON.stringify(coder)); + await until(() => runnerText(f, coder.result.run).includes("claimed by run")); + state.holders = [{ instance: "reviewer", decision: "d3", task_ref: "vikunja:1/4" }]; + await launcher.tick(); + await launcher.tick(); + const refused = log(f).filter((e) => e.event === "refused"); + assert.deepEqual(refused.map(({ by, instance, decision, code }) => ({ by, instance, decision, code })), [{ by: "jason", instance: "reviewer", decision: "d3", code: "capacity-full" }]); + assert.deepEqual(logs.filter((l) => l.includes("landing start refused")), ["decision d3 (reviewer): landing start refused: capacity-full"]); + assert.equal(log(f).some((e) => e.event === "launch" && e.instance === "reviewer"), false); + + const s = readSessions(f.dataRoot).find((x) => x.run === coder.result.run); + process.kill(s.runnerPid, "SIGTERM"); + await until(() => !readSessions(f.dataRoot).some((x) => x.run === coder.result.run) && log(f).some((e) => e.event === "end" && e.run === coder.result.run)); + await launcher.tick(); + const [run] = log(f).filter((e) => e.event === "launch" && e.instance === "reviewer").map((e) => e.run); + assert.ok(run); + assert.deepEqual(landingOf(f, run), [{ decision: "d3", task_ref: "vikunja:1/4" }]); + await until(() => runnerText(f, run).includes("runner: landing turn 1 for decision d3 (vikunja:1/4): ok")); }); test("a project clone that can't be made refuses checkout-failed, from --chief and from a session, with no run and no clone left", { skip, timeout: 30000 }, async (t) => { diff --git a/packages/cli/tests/thread-launch.test.mjs b/packages/cli/tests/thread-launch.test.mjs index e018cea44..62f0bc1fb 100644 --- a/packages/cli/tests/thread-launch.test.mjs +++ b/packages/cli/tests/thread-launch.test.mjs @@ -86,6 +86,8 @@ async function start(t, { f, adapter }, { tickMs = 100, ops: replace } = {}) { threadIdle: (m) => runtime.threadIdle(m.record), mayLaunch: (m) => runtime.mayLaunch(m), landHeld: (m) => runtime.landHeld(m.record), + landHolders: (m) => runtime.landHolders(m.business), + landRelease: (m) => runtime.landRelease(m.record), ...replace?.(runtime), }; const refusing = (what, fn) => async (m) => { @@ -301,6 +303,23 @@ test("REQ-ROLE-2 REQ-LAUNCH-1 a thread held by a run the launcher didn't start g assert.deepEqual(launchLog(f), []); }); +// Row 156 round 2 (Filbert's F2 in 29033): a thread launch asks landHeld, never landHolders, so a +// landHolders entry naming the thread instance gives its session no landing. Without the +// `thread ? null :` guard in start, the entry would reach session.json. +test("REQ-LAUNCH-1 a thread run's session file has no landing, even when landHolders names its instance", { skip, timeout: 30000 }, async (t) => { + const prepared = prepare(t); + const { f } = prepared; + const listed = [{ instance: "assistant", decision: "d9", task_ref: "vikunja:1/9", resolved_at: "2026-10-18T00:00:00.000Z" }]; + const host = await start(t, prepared, { ops: () => ({ landHolders: () => listed }) }); + const project = host.call("project.create", { name: "stack work", repo: "stack" }).id; + const A = host.call("thread.create", { project, name: "alpha" }).id; + const sent = host.call("thread.send", { thread: A, body: "for A" }); + await until(() => host.call("thread.messages", { thread: A }).find((m) => m.in_reply_to === sent.id && m.class === "RESULT")); + const [launch] = launchLog(f).filter((e) => e.event === "launch" && e.thread === A); + const session = JSON.parse(readFileSync(join(f.dataRoot, "launches", "acme", launch.run, "session.json"), "utf8")); + assert.equal(Object.hasOwn(session, "landing"), false, JSON.stringify(session)); +}); + // 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. diff --git a/packages/harness/src/runner.mjs b/packages/harness/src/runner.mjs index 820d03c89..cad6a0e9b 100644 --- a/packages/harness/src/runner.mjs +++ b/packages/harness/src/runner.mjs @@ -16,6 +16,9 @@ // own process group under a wall clock (S0 line 4). The turn's stdout // goes back to the sender as a RESULT, except for an incoming RESULT, // which never gets an automatic reply. Then message.read. +// Before the first message, one turn per session.json `landing` entry +// (row 156): a fixed request to land the approved merge decision, with +// no reply and no read. A malformed entry exits 2 before the claim. // 5. SIGTERM ends the turn, releases the role and exits 0; before the // claim it exits 0 without claiming. // @@ -78,6 +81,27 @@ export function turnRequest(m) { return `${head.join(", ")}:\n\n${tester}${m.body}`; } +// Row 156 (#1651): the request of a landing turn. Every value comes from the session file's landing +// entry, which the runner checked; nothing comes from a message body. +export function landingRequest({ decision, task_ref }) { + return `Landing from the host, decision ${decision}, task ${task_ref}:\n\nYour merge decision ${decision} for ${task_ref} is approved and unused. Land it now with land_work, task_ref ${task_ref} and decision ${decision}. The reply only says the landing started; the gate's outcome comes later as a RESULT from the bus. If land_work refuses decision-consumed, the landing already happened.`; +} + +// A landing entry is exactly { decision, task_ref }: a broker id and a task ref. +const LANDING_DECISION = /^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$/; +const LANDING_TASK_REF = /^vikunja:[1-9][0-9]*\/[1-9][0-9]*$/; +function landingOk(landing) { + if (landing === undefined) return true; + if (!Array.isArray(landing)) return false; + return landing.every((e) => { + if (!e || typeof e !== "object" || Array.isArray(e)) return false; + const keys = Object.keys(e).sort(); + return keys.length === 2 && keys[0] === "decision" && keys[1] === "task_ref" && + typeof e.decision === "string" && e.decision.length <= 160 && LANDING_DECISION.test(e.decision) && + typeof e.task_ref === "string" && LANDING_TASK_REF.test(e.task_ref); + }); +} + // 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) { @@ -296,6 +320,10 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l log(`runner: ${e.message}`); return EXIT.usage; } + if (!landingOk(session.landing)) { + log("runner: malformed landing entry"); + return EXIT.usage; + } if (stopping) { log("runner: stopped before claim"); return EXIT.stopped; @@ -343,7 +371,34 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l r(); }; }); - loop: while (!stopping) { + // Row 156 (#1651): the landing turns, before any message. Each reads the state as a message turn + // does and is bracketed by usage with the decision id as its turn; it sends nothing and marks + // nothing read. The turn counter is shared with message turns. + let gone = false; + for (const e of session.landing ?? []) { + if (stopping) break; + let block; + try { + block = stateBlock(await client.call("state"), new Date().toISOString()); + } catch (err) { + if (SESSION_GONE.has(err.code)) { + log(`runner: session stopped working: ${err.code}`); + code = EXIT.session; + gone = true; + break; + } + log(`runner: landing turn ${++n} for decision ${e.decision} (${e.task_ref}) failed: could not read the current state: ${err.code ?? err.message}`); + continue; + } + ++n; + try { await client.call("usage.begin", { turn: e.decision }); } + catch (err) { log(`runner: usage baseline unavailable: ${err.code ?? "failed"}`); } + const turn = await runTurn({ session, request: `${block}\n${landingRequest(e)}`, n, env: turnEnv, onChild: (c) => (child = c) }); + try { await client.call("usage.end", { turn: e.decision }); } + catch (err) { log(`runner: usage outcome unavailable: ${err.code ?? "failed"}`); } + log(`runner: landing turn ${n} for decision ${e.decision} (${e.task_ref}): ${turn.ok ? "ok" : `failed: ${turn.reason}`}`); + } + loop: while (!stopping && !gone) { let messages; try { messages = await client.call("message.receive"); diff --git a/packages/harness/tests/runner.test.mjs b/packages/harness/tests/runner.test.mjs index 964b9e581..25767b33e 100644 --- a/packages/harness/tests/runner.test.mjs +++ b/packages/harness/tests/runner.test.mjs @@ -13,7 +13,7 @@ import { join } from "node:path"; import { Store } from "../../bus/src/store.mjs"; import { Broker, BusError } from "../../bus/src/broker.mjs"; import { serve } from "../../bus/src/server.mjs"; -import { EXIT, founderCheck, stateBlock, turnRequest } from "../src/runner.mjs"; +import { EXIT, founderCheck, landingRequest, stateBlock, turnRequest } from "../src/runner.mjs"; import { typedTools } from "../src/tools.mjs"; import { resolvedFor, scratch } from "./helpers.mjs"; @@ -91,10 +91,12 @@ test("stateBlock names the launch state with its time and event, and the open ta assert.match(capped, /Open tasks: 9, the first 1 of cap 1\./); }); -async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null, limits = null } = {}) { +// `roles` replaces demo's role entries by name (row 156 round 2 gives the coder git.merge.protected). +async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null, limits = null, roles = null } = {}) { const dir = scratch(t); const store = new Store(dir); - const demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo; + let demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo; + if (roles) demo = { ...demo, roles: { ...demo.roles, ...roles } }; const broker = new Broker({ store, businesses: { demo }, organization, usage, limits }); const brokerSocket = join(dir, "bus.sock"); const server = await serve({ broker, path: brokerSocket }); @@ -637,3 +639,158 @@ test("REQ-LIMIT-4 a runner at its limit stops at the end of its turn: RESULT and const revokedAt = ctx.store.get("SELECT at FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").at; assert.ok(ctx.store.get("SELECT at FROM messages WHERE id=?", reply.id).at <= revokedAt); }); + +// Row 156 (#1651): the session file's landing. The adapter records each turn's request, so a test +// sees what the turn was given and in which order. +test("landingRequest is the fixed template, with the decision and task ref and nothing else", () => { + assert.equal( + landingRequest({ decision: "d1", task_ref: "vikunja:3/7" }), + "Landing from the host, decision d1, task vikunja:3/7:\n\nYour merge decision d1 for vikunja:3/7 is approved and unused. Land it now with land_work, task_ref vikunja:3/7 and decision d1. The reply only says the landing started; the gate's outcome comes later as a RESULT from the bus. If land_work refuses decision-consumed, the landing already happened.", + ); +}); + +function recording(ctx) { + const file = join(ctx.dir, "requests"); + writeFileSync(join(ctx.dir, "adapter.sh"), `printf '%s\\0' "$MOSAIC_REQUEST" >> '${file}'\nexec '${process.execPath}' '${FAKE}'\n`); + return () => (existsSync(file) ? readFileSync(file, "utf8").split("\0").slice(0, -1) : []); +} + +test("each landing entry is one turn before any message, with the state block and the fixed request; it sends nothing and reads nothing", async (t) => { + const landing = [{ decision: "d1", task_ref: "vikunja:3/7" }, { decision: "d2", task_ref: "vikunja:3/8" }]; + const ctx = await setup(t, { session: { landing } }); + const requests = recording(ctx); + ctx.inbox = []; + // The message waits from before the start; both landing turns still come first. + const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" }); + const r = startRunner(ctx.runDir, ctx.cap); + t.after(() => r.child.kill("SIGKILL")); + const [reply] = await until(replies(ctx)); + assert.equal(reply.in_reply_to, sent.id); + const got = requests(); + assert.equal(got.length, 3); + for (const [i, e] of landing.entries()) + assert.match(got[i], new RegExp(`^Current state, read at \\S+:\nLaunches: allowed \\(never switched\\)\\.\nOpen tasks: 0\\.\n\n${landingRequest(e).replace(/[.*+?^${}()|[\]\\/]/g, "\\$&")}$`)); + assert.match(got[2], new RegExp(`Message ${sent.id} from pm, sent \\S+, class REQUEST:\n\nhello$`)); + r.child.kill("SIGTERM"); + const out = await r.exited; + assert.equal(out.code, EXIT.stopped, out.stderr); + assert.match(out.stderr, /runner: landing turn 1 for decision d1 \(vikunja:3\/7\): ok\n[\s\S]*runner: landing turn 2 for decision d2 \(vikunja:3\/8\): ok\n/); + // Each landing turn asks usage.begin and usage.end with its decision id. These decisions aren't + // held in this broker, so both refuse not-recipient (land.test.mjs has a held one); a message + // turn with no usage configured refuses usage-not-configured instead. + for (const [i, e] of landing.entries()) + assert.match(out.stderr, new RegExp(`runner: usage baseline unavailable: not-recipient\\n(?:(?!runner: landing turn)[\\s\\S])*?runner: usage outcome unavailable: not-recipient\\n(?:(?!runner: landing turn)[\\s\\S])*?runner: landing turn ${i + 1} for decision ${e.decision} `)); + // The landing turns sent nothing and marked nothing read: one RESULT, one read, both for the message. + assert.deepEqual(ctx.store.all("SELECT in_reply_to FROM messages WHERE from_run='coder-run'").map((m) => m.in_reply_to), [sent.id]); + assert.deepEqual(ctx.store.all("SELECT message FROM deliveries WHERE op='read'").map((d) => d.message), [sent.id]); + assert.deepEqual(readdirSync(join(ctx.runDir, "turns")).sort(), ["0001.stderr", "0002.stderr", "0003.stderr"]); +}); + +test("an empty landing takes no landing turn: the message is turn 1", async (t) => { + const ctx = await setup(t, { session: { landing: [] } }); + const requests = recording(ctx); + ctx.inbox = []; + const r = startRunner(ctx.runDir, ctx.cap); + t.after(() => r.child.kill("SIGKILL")); + await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'")); + const sent = ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" }); + await until(replies(ctx)); + r.child.kill("SIGTERM"); + const out = await r.exited; + assert.equal(out.code, EXIT.stopped, out.stderr); + assert.doesNotMatch(out.stderr, /landing/); + const got = requests(); + assert.equal(got.length, 1); + assert.match(got[0], new RegExp(`Message ${sent.id} from pm`)); +}); + +// Row 156 round 2 (Filbert's F1 in 29033): while the runner is in a landing turn, the launcher's +// landing pass must not stop it for a second decision resolved meanwhile. The pass stops a run only +// when landRelease answers released (packages/cli/tests/launcher.test.mjs has the launcher's side), +// so this holds a real landing turn open on a real broker and asks landRelease as the pass does. +test("a run in a landing turn isn't released for a second held decision until its landing turns end and it has read its messages", async (t) => { + const MERGE = "git.merge.protected"; + const ctx = await setup(t, { roles: { coder: { authority: { withinRole: ["message.send"], crossRole: [MERGE] } } } }); + const merge = (target) => ({ + action: MERGE, target, question: `Merge ${target}?`, + options: [{ key: "yes", text: "Land it" }, { key: "no", text: "Not yet" }], recommendation: "yes", blocking: false, + }); + // d1 is raised and resolved before the run starts, by an earlier run of the instance. + const earlier = ctx.broker.bindLaunch({ business: "demo", role: "coder", run: "coder-0", harness: "pi" }); + ctx.call(earlier, "role.claim"); + const d1 = ctx.call(earlier, "decision.raise", merge("vikunja:1/2")); + assert.equal(d1.route_to, "pm"); + ctx.call(ctx.pm, "decision.resolve", { id: d1.id, choice: "yes" }); + ctx.broker.endLaunch({ business: "demo", run: "coder-0", reason: "exited" }); + const file = join(ctx.runDir, "session.json"); + writeFileSync(file, JSON.stringify({ ...JSON.parse(readFileSync(file, "utf8")), landing: [{ decision: d1.id, task_ref: "vikunja:1/2" }] })); + // The adapter's first turn, the landing turn, waits for the go file. + const started = join(ctx.dir, "started"); + const go = join(ctx.dir, "go"); + writeFileSync(join(ctx.dir, "adapter.sh"), `touch '${started}'\nwhile [ ! -e '${go}' ]; do sleep 0.05; done\nexec '${process.execPath}' '${FAKE}'\n`); + const r = startRunner(ctx.runDir, ctx.cap); + t.after(() => r.child.kill("SIGKILL")); + await until(() => existsSync(started)); + // d2 is resolved during the landing turn, so the landing pass lists it for the running holder. + const d2 = ctx.call(ctx.cap, "decision.raise", merge("vikunja:1/3")); + ctx.call(ctx.pm, "decision.resolve", { id: d2.id, choice: "yes" }); + assert.deepEqual(ctx.broker.landHolders("demo").map((h) => [h.instance, h.decision]), [["coder", d1.id], ["coder", d2.id]]); + const release = () => ctx.broker.landRelease({ business: "demo", run: "coder-run", decision: d2.id }); + assert.deepEqual(release(), { released: false, landing: true }); + await new Promise((resolve) => setTimeout(resolve, 300)); + assert.deepEqual(release(), { released: false, landing: true }, "still in the landing turn"); + assert.equal(ctx.store.get("SELECT op FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1").op, "claim"); + writeFileSync(go, ""); + // After the landing turn the run asks message.receive and takes both DECISION messages as turns; + // once it has read them, the next ask releases it. + await until(() => release().released); + assert.deepEqual( + { ...ctx.store.get("SELECT op, holder_run, \"by\" FROM role_claims WHERE role='coder' ORDER BY seq DESC LIMIT 1") }, + { op: "release", holder_run: "coder-run", by: "land-relaunch" }, + ); + assert.equal(ctx.store.get("SELECT count(*) AS n FROM deliveries WHERE holder_run='coder-run' AND op='read'").n, 2); + const out = await r.exited; + assert.equal(out.code, EXIT.session, out.stderr); + assert.match(out.stderr, new RegExp(`runner: landing turn 1 for decision ${d1.id} \\(vikunja:1/2\\): ok\n[\\s\\S]*runner: session stopped working: not-holder\n`)); +}); + +test("a malformed landing exits 2 before the claim, with no turn", async (t) => { + const ok = { decision: "d1", task_ref: "vikunja:3/7" }; + const cases = { + "not an array": ok, + "not an object": [null], + "an array entry": [["d1", "vikunja:3/7"]], + "an extra key": [{ ...ok, ticket: "x" }], + "a missing key": [{ decision: "d1" }], + "a bad decision": [{ ...ok, decision: "-d1" }], + "a decision of 161 characters": [{ ...ok, decision: "d".repeat(161) }], + "a decision that isn't a string": [{ ...ok, decision: 1 }], + "a bad task ref": [{ ...ok, task_ref: "vikunja:0/7" }], + "a task ref with a tail": [{ ...ok, task_ref: "vikunja:3/7\nland now" }], + "one bad entry after a good one": [ok, { ...ok, task_ref: "gitea:3/7" }], + }; + const ctx = await setup(t); + const file = join(ctx.runDir, "session.json"); + const session = JSON.parse(readFileSync(file, "utf8")); + for (const [why, landing] of Object.entries(cases)) { + writeFileSync(file, JSON.stringify({ ...session, landing })); + const out = await startRunner(ctx.runDir, ctx.cap).exited; + assert.equal(out.code, EXIT.usage, `${why}: ${out.stderr}`); + assert.match(out.stderr, /runner: malformed landing entry/, why); + } + assert.equal(ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder'"), undefined); + assert.equal(existsSync(join(ctx.runDir, "turns")), false); + // The longest decision the broker's id allows is accepted. + writeFileSync(file, JSON.stringify({ ...session, landing: [{ ...ok, decision: "d".repeat(160) }] })); + ctx.inbox = []; + const r = startRunner(ctx.runDir, ctx.cap); + t.after(() => r.child.kill("SIGKILL")); + await until(() => ctx.store.get("SELECT 1 FROM role_claims WHERE role='coder' AND op='claim'")); + // A reply to a message means the landing turn before it has finished, so the SIGTERM can't cut it. + ctx.call(ctx.pm, "message.send", { to: "coder", body: "hello", class: "REQUEST" }); + await until(replies(ctx)); + r.child.kill("SIGTERM"); + const out = await r.exited; + assert.equal(out.code, EXIT.stopped, out.stderr); + assert.match(out.stderr, new RegExp(`runner: landing turn 1 for decision ${"d".repeat(160)} \\(vikunja:3/7\\): ok`)); +}); diff --git a/roles/chief-of-staff.md b/roles/chief-of-staff.md index b232536dd..cfcbd5294 100644 --- a/roles/chief-of-staff.md +++ b/roles/chief-of-staff.md @@ -52,7 +52,9 @@ product's code, you don't review it and you don't create tasks yourself. cto's message named the instance and the decision, `launch` for that instance was refused `instance-running`, and the run belongs to that instance. Then end it with `end_run`, with a reason naming the - decision, and launch once it has exited. End no other run. + decision, and launch once it has exited. End no other run. The host + now starts that run itself once the decision is approved; these steps + are the fallback. - Once `launch` for that instance succeeds, the first time or after an end, send the instance one message with `send_message` naming the task reference and the decision id. A run takes a turn only on a message it diff --git a/roles/cto.md b/roles/cto.md index 442242408..4911db8b5 100644 --- a/roles/cto.md +++ b/roles/cto.md @@ -15,7 +15,8 @@ assigned engineering work yourself. option, send the chief of staff one message with `send_message`, naming the instance that raised it (`raised_by_role`), the decision id and the task reference (the decision's `authorization.target`). That - instance can land only in a run launched after your approval. + instance can land only in a run launched after your approval. The host + now starts that run itself; your message is the fallback. - Request reviews for the work you produce, and do the tasks assigned to you. - When the bus tells you the human saved `spec/north-star.md`, read the diff --git a/roles/merge-gate.md b/roles/merge-gate.md index 5a96628e8..f8e714c33 100644 --- a/roles/merge-gate.md +++ b/roles/merge-gate.md @@ -26,6 +26,9 @@ write tools. tools have no `land_work`, don't try to land and don't raise again. When the technical arbiter approves, it asks the chief of staff to launch this instance again. + The host now starts that run itself once the decision is approved, and + its first turn, from the host, names the task reference and the + decision id; the chief of staff's launch and message are the fallback. - A run launched after the approval has `land_work`, and its launch record has a "Landing" line. Its turn comes on the chief of staff's message naming the task reference and the decision id: call