fix(bus,harness): stop INFO echo replies and duplicate continuation notices (#1847, #1848)

Expose the parent message class on agent delivery so the runner can stop automatic replies to answers while preserving requests and ordinary relays. Scope continuation retry versions to the target, suppress notices during processing halts and unclaimed launches, and clarify the instance-running refusal.

The bus suite excludes project-files.test.mjs because its 1 MiB socket test timed out on both the candidate and the untouched 1186cf7b1 base.

packages/bus/src/broker.mjs

@@ -652,7 +652,8 @@ export class Broker {
         },
         continuationRouted: (message, run) => !!this.#store.get("SELECT 1 FROM deliveries WHERE message=? AND holder_run=? AND op IN ('routed','delivered')", message, run),
         continuationRetry: (state, message, target, context) => {
-          if (context.chief && target !== context.chief) this.#notice(state.business, message, context.chief, target, NOTICE_ROLE, this.#business(state.business).launch.instances.includes(target), this.#store.get('SELECT body FROM messages WHERE id=?', message)?.body);
+          this.#processingWork();
+          if (!this.#continuationStarting(state.business, target) && context.chief && target !== context.chief) this.#notice(state.business, message, context.chief, target, NOTICE_ROLE, this.#business(state.business).launch.instances.includes(target), this.#store.get('SELECT body FROM messages WHERE id=?', message)?.body);
         },
         continuationMessage: (state, target, body, context) => this.#continuationMessage(state, target, body, context),
         continuationQuote: (state, resolved = null) => this.#continuationQuote(state, resolved),
@@ -712,12 +713,16 @@ export class Broker {
     const b = this.#business(state.business), holder = this.#holder(state.business, target);
     const chief = b.launch?.by ?? null, h = chief ? this.#holder(state.business, chief) : null;
     const run = holder?.op === 'claim' ? holder.holder_run : null;
-    const revision = this.#store.get("SELECT max(seq) n FROM events WHERE business=? AND kind IN ('session.launched','session.ended','launch.revoked','launch.restored')", state.business).n;
-    const claims = this.#store.get('SELECT max(seq) n FROM role_claims WHERE business=?', state.business).n;
+    const revision = this.#store.get("SELECT max(seq) n FROM events WHERE business=? AND (kind IN ('launch.revoked','launch.restored') OR (actor_role=? AND kind IN ('session.launched','session.ended')))", state.business, target).n;
+    const claims = this.#store.get('SELECT max(seq) n FROM role_claims WHERE business=? AND role=?', state.business, target).n;
     const revoked = this.#store.get('SELECT state FROM launch_state WHERE business=?', state.business)?.state === 'revoked';
     return { run, chief, chief_run: h?.op === 'claim' ? h.holder_run : null, revoked,
       prerequisite: digestPrd({ business: b, revision, claims, revoked, credentials: this.#credentialStatus() }) };
   }
+  #continuationStarting(business, target) {
+    // A launch without an end is still starting when the target has no holder.
+    return !!this.#store.get("SELECT 1 FROM events l WHERE l.business=? AND l.actor_role=? AND l.kind='session.launched' AND NOT EXISTS (SELECT 1 FROM events e WHERE e.business=l.business AND e.kind='session.ended' AND e.subject=l.subject) LIMIT 1", business, target);
+  }
   #continuationMessage(state, target, body, context) {
     this.#processingWork();
     const b = this.#business(state.business);
@@ -727,7 +732,7 @@ export class Broker {
     this.#store.run('INSERT INTO messages(id,at,business,from_role,from_run,to_role,class,body) VALUES(?,?,?,?,?,?,?,?)',
       mid, this.#now(), state.business, NOTICE_ROLE, NOTICE_RUN, target, 'RESULT', body);
     this.#route(state.business, mid, target, null);
-    if (!context.run && context.chief && target !== context.chief)
+    if (!context.run && !this.#continuationStarting(state.business, target) && context.chief && target !== context.chief)
       this.#notice(state.business, mid, context.chief, target, NOTICE_ROLE, b.launch.instances.includes(target), body);
     return mid;
   }
@@ -7074,7 +7079,8 @@ export class Broker {
             s.human || s.tester ? null : this.#own(s),
             s.human || s.tester ? 'cli' : 'broker',
           );
-        return rows.map((m) => ({ ...m, request: this.#request(s.business, m) }));
+        return rows.map((m) => ({ ...m, request: this.#request(s.business, m),
+          ...(!s.human && !s.tester && m.in_reply_to ? { in_reply_to_class: this.#store.get('SELECT class FROM messages WHERE business=? AND id=?', s.business, m.in_reply_to)?.class ?? null } : {}) }));
       }
       case 'message.read': {
         keys(a, ['id'], ['id']);
@@ -8157,8 +8163,8 @@ export class Broker {
     if (told) this.#notice(s.business, mid, by, to, sender, launchable);
     return { role: to, launcher: by, told, launcherRunning: by === null ? null : this.#recipient(s.business, by, null) !== null, launchable };
   }
-  // One notice per waiting message: only message.send calls this, once for the message it stores, so
-  // a re-route at a claim or a broker restart never sends another. Its class is RESULT: the runner
+  // Ordinary messages get one notice; continuations may retry on a target prerequisite change.
+  // Its class is RESULT: the runner
   // takes a turn on it and sends no automatic reply, which NOTICE_ROLE could not receive.
   #notice(business, mid, by, to, sender, launchable, continuation = null) {
     const nid = randomUUID();

packages/bus/tests/launch-notice.test.mjs

@@ -6,6 +6,7 @@ import { join } from 'node:path';
 import { fileURLToPath } from 'node:url';
 import { Store } from '../src/store.mjs';
 import { Broker } from '../src/broker.mjs';
+import { createContinuations } from '../src/request-continuation.mjs';
 import { ID_PATTERN } from '../../business/src/vocabulary.mjs';
 
 // Row 146 (#1640): a message.send to a role nobody holds tells the business's launch.by
@@ -35,6 +36,72 @@ const demo = (extra = {}) => ({
   },
 });
 
+test('continuation notices ignore unrelated lifecycle events, halts, and unclaimed launches', async t => {
+  const root = mkdtempSync(join(tmpdir(), 'continuation-notice-'));
+  const store = new Store(root);
+  t.after(() => { store.close(); rmSync(root, { recursive: true, force: true }); });
+  const extraRoles = { ...roles, reviewer: { definition: 'reviewer', authority: { withinRole: ['message.send'], crossRole: [] } }, helper: { definition: 'coder', authority: { withinRole: ['message.send'], crossRole: [] } } };
+  const businesses = demo({ roles: extraRoles, definitions: { ...definitions, reviewer: {} }, launch: { by: 'chief', instances: ['pm', 'coder', 'reviewer'], models: {}, limits: {} } });
+  const states = ['pm', 'coder', 'reviewer'].map((role, i) => ({ business: 'demo', request: `request-${i}`, revision: 1, phase: 'selection', pm: role, source_message: `source-${i}`, tasks: [] }));
+  const rows = [];
+  const journal = {
+    effects: (business, request) => rows.filter(r => r.business === business && r.request === request),
+    transaction: fn => fn(),
+    prepareEffect: e => { rows.push(structuredClone(e)); return e; },
+    finishEffect: e => { rows.push(structuredClone(e)); return e; },
+  };
+  let halted = false, continuations;
+  const processing = { read: () => ({ version: 1, generation: '00000000-0000-4000-8000-000000000000', halted }) };
+  const b = new Broker({ store, businesses, processing, requestLifecycle: services => {
+    continuations = createContinuations({ journal, services });
+    const noop = () => {};
+    return { open: noop, get: noop, projects: noop, propose: noop, question: noop, answer: noop, attach: noop, cancel: noop, modeConsole: noop, draftConsole: noop, resume: noop, recover: noop };
+  } });
+  const resume = () => { for (const state of states) { try { continuations.resume(state); } catch (e) { if (e.code !== 'processing-halted') throw e; } } };
+  const notices = id => store.get("SELECT count(*) n FROM messages WHERE from_role='_bus' AND to_role='chief' AND in_reply_to=?", id).n;
+  const launch = (role, run) => b.bindLaunch({ business: 'demo', role, run, harness: 'pi' });
+  launch('helper', 'helper-old');
+  resume();
+  const waiting = store.all("SELECT id,to_role FROM messages WHERE from_role='_bus' AND body LIKE 'Request continuation %' ORDER BY seq");
+  assert.equal(waiting.length, 3);
+  for (const m of waiting) assert.equal(notices(m.id), 1);
+
+  halted = true;
+  b.endLaunch({ business: 'demo', run: 'helper-old', reason: 'exit' });
+  resume();
+  for (const m of waiting) assert.equal(notices(m.id), 1, 'halt sends no notices');
+  halted = false;
+  launch('helper', 'helper-new');
+  resume();
+  for (const m of waiting) assert.equal(notices(m.id), 1, 'unrelated launch sends no notices');
+
+  const pm = launch('pm', 'pm-new');
+  resume();
+  for (const m of waiting) assert.equal(notices(m.id), 1, 'unclaimed target launch sends no notices');
+  b.request(pm, { verb: 'role.claim', args: {} });
+  resume();
+  const pmMessage = waiting.find(m => m.to_role === 'pm');
+  assert.equal(store.get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed' ORDER BY seq DESC LIMIT 1", pmMessage.id).holder_run, 'pm-new');
+  launch('coder', 'coder-new');
+  resume();
+  for (const m of waiting) assert.equal(notices(m.id), 1, 'staggered launches do not duplicate notices');
+  states.push({ business: 'demo', request: 'request-starting', revision: 1, phase: 'selection', pm: 'coder', source_message: 'source-starting', tasks: [] });
+  resume();
+  const starting = store.get("SELECT id FROM messages WHERE to_role='coder' AND body LIKE '%request-starting%'");
+  assert.equal(notices(starting.id), 0, 'a new continuation does not announce an unclaimed launch');
+  halted = true;
+  b.endLaunch({ business: 'demo', run: 'coder-new', reason: 'exit' });
+  resume();
+  assert.equal(notices(starting.id), 0, 'target end while halted sends no notice');
+  halted = false;
+  resume();
+  assert.equal(notices(starting.id), 1, 'target end is noticed after processing resumes');
+  const replacement = launch('coder', 'coder-replacement');
+  b.request(replacement, { verb: 'role.claim', args: {} });
+  resume();
+  assert.equal(store.get("SELECT holder_run FROM deliveries WHERE message=? AND op='routed' ORDER BY seq DESC LIMIT 1", starting.id).holder_run, 'coder-replacement');
+});
+
 function setup(t, businesses = demo()) {
   const root = mkdtempSync(join(tmpdir(), 'bus-notice-'));
   let store = new Store(root);

packages/harness/src/runner.mjs

@@ -14,8 +14,9 @@
 //      harness never sees the capability.
 //   4. One harness turn per received message, through the adapter, in its
 //      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.
+//      goes back to the sender as a RESULT, except for an incoming RESULT or
+//      an agent's non-REQUEST reply to a RESULT. Human, tester and user messages
+//      and REQUESTs still get replies. 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.
@@ -665,7 +666,7 @@ export async function main(runDir, { stdin = process.stdin, env = process.env, l
       if (stopping) break loop;
       if (!turn.ok) log(`runner: turn ${n} for message ${m.id} failed: ${turn.reason}`);
       try {
-        if (m.class !== "RESULT") {
+        if (m.class !== "RESULT" && (m.class === "REQUEST" || ["human", "tester", "user"].includes(m.from_role) || m.in_reply_to_class !== "RESULT")) {
           if (!library) {
             const body = turn.ok ? turn.text || "(the session gave no answer)" : `The turn failed: ${turn.reason}`;
             await client.call("message.send", { to: m.from_role, body: clip(body) + unslop.note(), class: "RESULT", in_reply_to: m.id });

packages/harness/src/tools.mjs

@@ -98,7 +98,7 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project
   const tools = [];
   const add = (name, verb, action, description, parameters) => tools.push({ name, verb, action, description, parameters });
   if (held.has("message.send")) {
-    add("send_message", "message.send", "message.send", "Send a message to another role instance, or to \"human\". Your final answer to a message is sent back to its sender for you; use this for anything else.", obj({
+    add("send_message", "message.send", "message.send", "Send a message to another role instance, or to \"human\". Your final answer to a message is sent back to its sender for you; use this for anything else. A reply to an answer gets no automatic answer; use class REQUEST to ask a follow-up.", obj({
       to: str("Role instance, its name, or \"human\""),
       body: str("Message text"),
       class: str("Message class", { enum: MESSAGE_CLASSES }),
@@ -429,7 +429,7 @@ export function callTool(path, tool, args = {}, { timeout = 60_000 } = {}) {
       if (!input.includes("\n")) return;
       try {
         const r = JSON.parse(input);
-        r.ok ? end(null, r.result) : end(new Error(tool === "intake_question" && r.error === "outcome-unknown" ? "outcome-unknown" : tool === "launch" && r.error === "instance-running" ? "refused: instance-running: already running; wait for the existing run" : `refused: ${r.error}${typeof r.reason === "string" && tool === "advance_request" && r.error === "request-evidence-required" ? `: ${r.reason.slice(0, 300)}` : tool === "write_record" && r.error === "project-record-invalid" && typeof r.reason === "string" ? `: ${r.reason.slice(0, 300)}` : ""}`));
+        r.ok ? end(null, r.result) : end(new Error(tool === "intake_question" && r.error === "outcome-unknown" ? "outcome-unknown" : tool === "launch" && r.error === "instance-running" ? "refused: instance-running: target instance is already running and will receive its waiting messages when it claims; you must not launch it again or relay the notice" : `refused: ${r.error}${typeof r.reason === "string" && tool === "advance_request" && r.error === "request-evidence-required" ? `: ${r.reason.slice(0, 300)}` : tool === "write_record" && r.error === "project-record-invalid" && typeof r.reason === "string" ? `: ${r.reason.slice(0, 300)}` : ""}`));
       } catch {
         end(unknown("tool reply isn't JSON"));
       }

packages/harness/tests/launch-backoff.test.mjs

@@ -21,14 +21,14 @@ test('a running instance is a wait: repeated launch calls are delayed, independe
   const path=join(dir,'tools.sock');
   const server=await runner.toolServer({path,tools:[{name:'launch',verb:'launch'}],client:{},local:{launch}});
   t.after(()=>new Promise(resolve=>server.close(resolve)));
-  await assert.rejects(callTool(path,'launch',{instance:'pm'}),/instance-running.*already running.*wait/i);
+  await assert.rejects(callTool(path,'launch',{instance:'pm'}),/instance-running.*already running.*waiting messages.*claims.*must not launch.*relay/i);
   const second=callTool(path,'launch',{instance:'pm'});
   for(let n=0;n<100 && !delays.length;n++) await new Promise(resolve=>setTimeout(resolve,2));
   assert.deepEqual(calls,['pm']);
   assert.equal(delays.length,1); assert.ok(delays[0].ms >= 30000);
   assert.deepEqual(await callTool(path,'launch',{instance:'coder'}),{run:'other-run'});
   now += delays[0].ms; delays[0].resolve();
-  await assert.rejects(second,/instance-running.*already running.*wait/i);
+  await assert.rejects(second,/instance-running.*already running.*waiting messages.*claims.*must not launch.*relay/i);
   assert.deepEqual(calls,['pm','coder','pm']);
 });
 

packages/harness/tests/runner.test.mjs

@@ -102,10 +102,11 @@ test("stateBlock names the launch state with its time and event, and the open ta
 });
 
 // `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 } = {}) {
+async function setup(t, { session = {}, policy = {}, tools, as = "coder", organization = null, launchBy, usage = null, limits = null, roles = null, tester = false } = {}) {
   const dir = scratch(t);
   const store = new Store(dir);
   let demo = launchBy ? { ...businesses.demo, launch: { ...businesses.demo.launch, by: launchBy } } : businesses.demo;
+  if (tester) demo = { ...demo, vars: { ...demo.vars, "bus.tester": true } };
   if (roles) demo = { ...demo, roles: { ...demo.roles, ...roles } };
   const broker = new Broker({ store, businesses: { demo }, organization, usage, limits });
   const brokerSocket = join(dir, "bus.sock");
@@ -228,6 +229,26 @@ test("a message becomes a turn, the answer goes back as a RESULT, SIGTERM releas
   assert.ok(!existsSync(join(ctx.runDir, "tools.sock")));
 });
 
+test("agent replies to RESULT end unless they are REQUEST; ordinary INFO and human and tester follow-ups still get answers", async t => {
+  const ctx = await setup(t, { tester: true });
+  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 seed = ctx.call(ctx.pm, "message.send", { to: "coder", body: "first question", class: "REQUEST" });
+  const anchor = await until(() => ctx.store.get("SELECT id FROM messages WHERE from_run='coder-run' AND in_reply_to=? AND class='RESULT'", seed.id));
+  const info = ctx.call(ctx.pm, "message.send", { to: "coder", body: "noted", class: "INFO", in_reply_to: anchor.id });
+  const ordinary = ctx.call(ctx.pm, "message.send", { to: "coder", body: "ordinary relay", class: "INFO" });
+  const linked = ctx.call(ctx.pm, "message.send", { to: "coder", body: "linked relay", class: "INFO", in_reply_to: seed.id });
+  const request = ctx.call(ctx.pm, "message.send", { to: "coder", body: "follow-up", class: "REQUEST", in_reply_to: anchor.id });
+  const human = ctx.call(ctx.human, "message.send", { to: "coder", body: "human follow-up", in_reply_to: anchor.id });
+  const tester = ctx.broker.bindTester({ business: "demo" });
+  const testMessage = ctx.call(tester, "message.send", { to: "coder", body: "tester follow-up", in_reply_to: anchor.id });
+  await until(() => ctx.store.get("SELECT 1 FROM deliveries WHERE message=? AND op='read'", testMessage.id));
+  const answers = ctx.store.all("SELECT in_reply_to,class FROM messages WHERE from_run='coder-run' ORDER BY seq");
+  assert.deepEqual(answers.map(m => [m.in_reply_to, m.class]), [[seed.id, "RESULT"], [ordinary.id, "RESULT"], [linked.id, "RESULT"], [request.id, "RESULT"], [human.id, "RESULT"], [testMessage.id, "RESULT"]]);
+  assert.equal(ctx.store.get("SELECT count(*) n FROM messages WHERE in_reply_to=?", info.id).n, 0);
+});
+
 test("managed empty answer formatting does not mark model output", async t => {
   const ctx = await setup(t, {session: {proxyFirstTurn: true, turnEnd: "exit"}});
   writeFileSync(join(ctx.runDir, "proxy-first-turn-enabled"), "enabled\n");