1362 lines
79 KiB
Diff
1362 lines
79 KiB
Diff
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 <id>...`
|
|
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/<business>/<instance>/`. 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 <id> (<instance>):
|
|
+ landing start refused: <code>`, 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 <run>: landRelease refused: <code>`. 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 `<dataRoot>/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
|