bus, cli, harness: the host starts the landing run once a merge decision is approved (row 156, #1651)

Until now a landing ran only when the chief of staff or merge-gate
followed row 149's manual steps. The host now does it.

Broker. landHolders(business) lists the approved merge decisions each
instance holds, sharing the #held generator with landHeld.
landRelease(record) releases a run's claim with by='land-relaunch'.
Delivered and unread messages keep the run, and a run that hasn't had
its first message.receive answers {released:false, landing:true}
(#received, round 2), so a landing turn in progress is never cut. A
run that never reaches its first receive, a hung landing turn or a
broker restart, keeps the release closed, and the decision falls back
to row 149's path.
usage.begin accepts a decision id that landHolders lists for the run.
The bus process answers both ops from their own launchOp lines and
refuses any op it doesn't name.

Launcher. start() reads landHolders once for a non-thread instance and
takes land_work and the session's landing list from that one answer.
landPass() runs after threadPass() on each tick.

Runner. One turn per landing entry before the message loop, with a
usage bracket on the decision id.

Docs. A "Landing runs" section in the cli README and docs/TOOLS.md.
roles/merge-gate.md, cto.md and chief-of-staff.md say the host starts
the run, and row 149's steps are the fallback.

Owner Dewey, brief docs/plans/2026-10-18_landing-run-launch.md, design
DESIGN.md in the packet (sage-0808). Round 2 answers Filbert's F1-F3
(29033, rulings sage-0942). Reviewer on candidate manifest
72a5c1b6...b700: Filbert approve 29039. Seat run 4: 290/290 singles,
36/36 mutants killed. Sage gate on a618528b2 (candidate 524bfcc80): GATE_PASSED 25/25. Package tests 2258 plus queue 156, script selftests 422, 23:30:07Z to 23:38:04Z. Run 1 stopped at pkg-webui on board-held.test.mjs:163, the P19 366 timeout, 220/1. Run 2 resumed at pkg-webui on the same tree: webui 221/0 and every scripts/test-*.sh green.

Nothing live.

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>...`

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(

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.

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,

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(

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

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,78 +400,141 @@ 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;
-          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}:{}) });
-          }
-        });
-      }
+      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 {
+            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;
+        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 } : {}) });
+        }
+      });
+    }
+  }
+
   // endLaunch's answer ({ released, limit? }), or null when it refused.
   async function endRun(run, reason, exitCode) {
     try {

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) => {

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.

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");

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`));
+});

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

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

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