feat(harness): let the PM ask the owner an intake question (#1839)

BUILD-LOG.md

@@ -6117,3 +6117,4 @@ Local commit only; no push or live instance/credential-file access.
 2026-11-04T17:57:52Z | Codex implementation sub-agent, t3/lifecycle-task-6b-r2 | #797 / Gitea #1839 Task 6b rebase before/after | Before: local 347791b55 on the superseded Task 4 cherry-pick e41ec8159; origin/refactor is b5a58b48b4c254f166135c24e22bd63ce4b5e0b9 with Task 4 c098fa022, #772 793f3b157, Task 3 and Task 5b. After: replay only 347791b55 on the explicitly requested new branch t3/lifecycle-task-6b-r2; e41ec8159 is dropped. Shipped Task 3 migration 3q-3r and all seventeen existing migration files are byte-identical to the base. New 3r-3s rebuild preserves all fifteen shipped event kinds (including question.asked, question.answered, attach.selected and every Task 4 PRD/mode kind) and adds advanced, for sixteen; schema.sql matches, SCHEMA_VERSION=3s, SCHEMA_SHA256=fde7a46505a964d0e3802aaee96a1e2d9fb8c7163fbafc2f86aca914ecc50f03. The migration test carries rows of every prior kind, preserves a sqlite_sequence high-water mark, index and every request/effect guard, then appends advanced at the next sequence. Current-version, offline-chain and synthetic future-step test pins updated through 3s (future fixture 3t). Conflicts resolved by keeping both behavior sets: broker retains Console owner/verb guards, request question/answer/attach/cancel handlers, Task 5b builtin binding and scoped requirements/git admission, plus 6b advance and continuation services; lifecycle exports the union of Task 3 and 6b operations; runtime retains Task 5b builtin binding and #772 pending-product readiness while adding 6b recovery; harness keeps Task 5b project/snapshot/op schemas and catalog prose while adding 6b tools, with duplicate schema properties removed; six role duty documents merged cleanly and retained; append-only logs keep both sides. Integration corrections: unanswered owner questions suspend new continuation dispatch, answer evidence resumes it, and planning continuations carry the stable create op required by Task 5b. A real Task 5b adapter test creates/assigns the task with that op, advances it and keeps cross-project git refused. First full bus run 1554/1555 found a Task 3 global-zero-effects assertion; it now asserts attach preserves the exact preexisting continuation receipts. Another bus run and isolated #656 control reproduced the existing nice 16 versus 19 race: the installed Node scheduling rule changes startup priority. The owned FIFO child now sets priority after fixture construction immediately before the measured operation; all priority/nonblocking assertions remain intact, no daemon or live instance changed. FIFO 21/21 and final full bus 1555/1555 pass. First WebUI run 537/540, two failures and one skip: missing local esbuild 0.28.2 and one browser refresh timeout; npm ci --ignore-scripts installed only the locked WebUI dependencies, the browser file passes 2/2 in isolation, and whole WebUI passes 540/540 with MOSAIC_REQUIRE_EDITOR_BUILD=1 and test-concurrency=4. An accidental root npm test invocation reported no script and ran no tests; corrected to the bus package. Final GREEN, zero skips/cancellations: new 45/45, bus 1555/1555, harness 228/228, business 307/307, launch 8/8, whole WebUI 540/540. Every test invocation used its own mktemp -d /mnt/storage/scratch/tmp/XXXXXX TMPDIR. Logs /mnt/storage/scratch/tmp/task6b-r2-new-final.log, task6b-r2-bus-final.log, task6b-r2-harness.log, task6b-r2-business.log, task6b-r2-launch.log, task6b-r2-webui-final.log; complete per-run counts in task6b-r2-runs.json and schema evidence in task6b-r2-schema-proof.json in the same directory. No unresolved conflict, role JSON authority change, shipped migration edit, credential/config mount, push, merge, deployment or live-instance action. Queue command again refused this linked worktree; canonical coordination is unchanged. Local rebase only, with no attribution trailers.
 
 2026-11-04 | Codex implementation sub-agent | #803 Console Model proxy in Settings | Before: the owner's native Model proxy page was linked above the Console tree at /proxy. After: Settings links to /settings/model-proxy, whose page links back to Settings; /proxy redirects there only after the existing sign-in, granted-user and named-host fences. The proxy script, fixed API routes, human bus verbs, read and write authority, origin, Fetch Metadata, transport and bounded body checks are unchanged. No secret values are displayed. Tests first: expected red 3 failures (new Settings link, redirect and tester fence), 18 passes, zero skips; final targeted proxy HTTP 13/13, proxy browser 5/5, Settings browser 3/3, whole webui 558/558, zero failures, cancellations or skips. Each test command used a fresh TMPDIR under /mnt/storage/scratch/tmp/XXXXXX; logs /mnt/storage/scratch/tmp/console803-logs/{red,proxy-page,proxy-browser,views,webui}.log. Root dependencies were symlinked and webui dependencies linked per entry without installation. Queue next in this worktree refused the linked checkout; canonical queue next t3 returned nothing, with no queue movement. Local commit only; no push, live configuration or secrets access.
+2026-11-04T18:34:22Z | Codex implementation sub-agent, t3/lifecycle-intake-question | #797 / Gitea #1839 PM intake question before/after | Before: base 647626bdd exposes request.question on the bus but no PM harness tool. After: PM with message.send gets intake_question in both harnesses, forwarding only request, revision, op, text and optional bounded choices; the existing broker enforces exact shape, distinct choice keys, authority and replay. A stable caller-supplied op and exact retry use the broker journal, while bus refusals remain errors and lost/unknown outcomes are reported as unknown with no automatic retry. PM duty prose describes status reconciliation and that answers do not approve a PRD. Red evidence: request-tools red-real.log 19/25, with missing tool and unknown-outcome failures; earlier fixture corrections are retained in the logs. Final gates: harness 234/234, bus 1556/1556, business 307/307, all zero failures/skips/cancellations; tests include both harness manifests, unauthorized roles, invalid input, refusal propagation, unknown outcome, stable-op replay and owner Console answer through the real bus. Logs: /mnt/storage/scratch/tmp/W6Jjos/; each invocation used a fresh TMPDIR from /mnt/storage/scratch/tmp/XXXXXX and pipefail. No bus, role JSON, contract or migration change; local commit only, no push. Canonical queue next t3 returned nothing; coordination stays with the controller.

docs/SESSIONS.md

@@ -2014,3 +2014,4 @@ are never rewritten or removed; corrections are new entries.
 2026-11-04 | Codex implementation sub-agent, t3/lifecycle-task-6b-r2 | #797 / Gitea #1839 Task 6b rebase onto b5a58b48b | Dropped e41ec8159; merged Task 3 controls/questions, Task 5b scope/op admission and 6b continuations. New migration 3r-3s retains all fifteen shipped kinds plus advanced; all shipped migrations unchanged. Added question/op integration checks and fixed owned FIFO priority setup timing without weakening assertions. Final new 45/45, bus 1555/1555, harness 228/228, business 307/307, launch 8/8, whole WebUI 540/540, zero skips; per-run counts/log paths /mnt/storage/scratch/tmp/task6b-r2-runs.json. Both append-only logs retained; local rebase only, no attribution trailers or push/live changes; queue remains with canonical coordinator after linked-worktree refusal.
 2026-11-04 | Codex implementation sub-agent, t3/lifecycle-task-6b-r3 | #797 / Gitea #1839 Task 6b rebase onto 68bef84fa | Kept both append-only log sides, the Task 4 snapshot-counter behavior and the full 6b continuation changes; 3r-3s migration remains byte-identical to 65588327d. Gates, zero skips: bus 1556/1556, harness 228/228, business 307/307, webui 554/554; logs /mnt/storage/scratch/tmp/nj4n7n/gate-*.log. Local commit only, no push.
 2026-11-04 | Codex implementation sub-agent, t3/console-803-proxy-settings | #803 Model proxy in Settings | Protected /settings/model-proxy and checked /proxy redirect; unchanged proxy API and authority. Red first; proxy HTTP 13/13, proxy browser 5/5, Settings browser 3/3, whole webui 558/558, zero skips; logs /mnt/storage/scratch/tmp/console803-logs/. Linked-worktree queue refusal; canonical queue next t3 returned nothing. Local commit only, no push or live/secret access.
+2026-11-04 | Codex implementation sub-agent, t3/lifecycle-intake-question | #797 / Gitea #1839 PM intake question tool | PM-only typed request.question, strict bounded broker arguments, stable-op replay, refusal and unknown-outcome propagation, owner-answer integration and PM duty prose; red-first; harness 234/234, bus 1556/1556, business 307/307, zero skips; logs /mnt/storage/scratch/tmp/W6Jjos/. Local commit only; no push, bus, role JSON, contract or migration changes. Canonical queue next t3 returned nothing.

packages/harness/src/tools.mjs

@@ -127,6 +127,16 @@ export function typedTools({ resolved, arbiter = false, tracker = false, project
   }
   if (resolved.definition === "pm" && held.has("message.send")) {
     add("intake_open", "request.open", "message.send", "Open a reachable user request for pre-task intake, preserving its original intent. Opening an already open request returns its state.", obj({ request: REQUEST }, ["request"]));
+    add("intake_question", "request.question", "message.send", "Ask the owner one question during discovery, PRD review or planning. Use the current request revision and a stable op id; reuse the exact arguments and op only for a retry of the same question. An unknown outcome is not a refusal: check intake_status before retrying. Answers do not approve a PRD.", obj({
+      request: str("Original request id", { minLength: 1, maxLength: 160, pattern: "^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$" }),
+      revision: { type: "integer", minimum: 1, maximum: Number.MAX_SAFE_INTEGER, description: "Current request revision" },
+      op: str("Stable operation id for this question; reuse only with identical arguments", { minLength: 1, maxLength: 160, pattern: "^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$" }),
+      text: str("Question text", { minLength: 1, maxLength: 2000, pattern: "^[^\\u0000]+$" }),
+      options: { type: "array", maxItems: 8, description: "Optional choices with distinct keys", items: obj({
+        key: str("Choice key", { minLength: 1, maxLength: 160, pattern: "^[a-zA-Z0-9][a-zA-Z0-9_.:/-]*$" }),
+        text: str("Choice text", { minLength: 1, maxLength: 200, pattern: "^[^\\u0000]+$" }),
+      }, ["key", "text"]) },
+    }, ["request", "revision", "op", "text"]));
     add("intake_propose", "request.propose", "message.send", "Propose a project for an open request at its current revision. The broker prepares the exact resource plans and raises the user's new-project mode decision, or records a setup wait. This does not approve or bootstrap a project.", obj({
       request: REQUEST,
       revision: { type: "integer", minimum: 1, description: "Current request revision from intake_status or intake_open" },
@@ -404,19 +414,20 @@ export function callTool(path, tool, args = {}, { timeout = 60_000 } = {}) {
       socket.destroy();
       error ? reject(error) : resolve(value);
     };
-    socket.setTimeout(timeout, () => end(new Error("tool call timed out")));
+    const unknown = (message) => new Error(tool === "intake_question" ? `outcome-unknown: ${message}` : message);
+    socket.setTimeout(timeout, () => end(unknown("tool call timed out")));
     socket.on("connect", () => socket.write(JSON.stringify({ tool, args: { ...ARG_DEFAULTS[tool], ...args } }) + "\n"));
-    socket.on("error", (e) => end(new Error(`tool socket: ${e.code ?? e.message}`)));
-    socket.on("end", () => end(new Error("tool socket closed without a reply")));
+    socket.on("error", (e) => end(unknown(`tool socket: ${e.code ?? e.message}`)));
+    socket.on("end", () => end(unknown("tool socket closed without a reply")));
     socket.on("data", (b) => {
       input += b.toString("utf8");
-      if (input.length > 4 * 1024 * 1024) return end(new Error("tool reply too large"));
+      if (input.length > 4 * 1024 * 1024) return end(unknown("tool reply too large"));
       if (!input.includes("\n")) return;
       try {
         const r = JSON.parse(input);
-        r.ok ? end(null, r.result) : end(new Error(`refused: ${r.error}`));
+        r.ok ? end(null, r.result) : end(new Error(tool === "intake_question" && r.error === "outcome-unknown" ? "outcome-unknown" : `refused: ${r.error}`));
       } catch {
-        end(new Error("tool reply isn't JSON"));
+        end(unknown("tool reply isn't JSON"));
       }
     });
   });

packages/harness/tests/bundle.test.mjs

@@ -60,7 +60,7 @@ test("a pi bundle: prompt, policy, tools and manifest, 0600 in a 0700 directory"
   const prompt = readFileSync(b.files.prompt, "utf8");
   assert.match(prompt, /^# Contract\n\nDo the work\.\n\n## This session\n/);
   assert.match(prompt, /Role instance: pm \(definition pm\)\. Run: run-1\./);
-  assert.match(prompt, /Bus tools: send_message, request_staff, intake_status, intake_projects, intake_open, intake_propose, advance_request, submit_prd, task_create, /);
+  assert.match(prompt, /Bus tools: send_message, request_staff, intake_status, intake_projects, intake_open, intake_question, intake_propose, advance_request, submit_prd, task_create, /);
   assert.doesNotMatch(prompt, /Bus tools: [^.]*\blaunch\b/);
   // Row 90 (#1584): PM fault 10 (#228 c91).
   assert.match(prompt, /^- Only the human stops a run, with `mosaic stop`\. If you think a run should stop, raise a role\.revoke decision with blocking false/m);

packages/harness/tests/request-tools.test.mjs

@@ -5,6 +5,7 @@ import { spawn } from 'node:child_process';
 import { mkdirSync, readFileSync } from 'node:fs';
 import { join } from 'node:path';
 import { createInterface } from 'node:readline';
+import { createServer } from 'node:net';
 import { buildBundle, SOURCES } from '../src/bundle.mjs';
 import { callTool, typedTools } from '../src/tools.mjs';
 import * as runner from '../src/runner.mjs';
@@ -13,9 +14,10 @@ import { Broker } from '../../bus/src/broker.mjs';
 import { Store } from '../../bus/src/store.mjs';
 import { createRequestStore } from '../../bus/src/request-store.mjs';
 import { createRequestLifecycle } from '../../bus/src/request-lifecycle.mjs';
+import { requestFixture } from '../../bus/tests/request-fixture.mjs';
 
 const reads = ['intake_status', 'intake_projects'];
-const writes = ['intake_open', 'intake_propose'];
+const writes = ['intake_open', 'intake_propose', 'intake_question'];
 const intake = t => t.name.startsWith('intake_');
 const shipped = name => JSON.parse(readFileSync(join(REPO, 'roles', `${name}.json`), 'utf8'));
 const roleDefinition = role => role === 'chief' ? 'chief-of-staff' : role;
@@ -43,7 +45,6 @@ for (const harness of ['pi', 'claude-code']) {
       const tools = JSON.parse(readFileSync(b.files.tools, 'utf8'));
       assert.deepEqual(tools.filter(intake).map(t => t.name).sort(), expected.toSorted());
       assert.deepEqual(b.policy.typed.filter(n => n.startsWith('intake_')).sort(), expected.toSorted());
-      assert.ok(!tools.some(t => t.name === 'intake_question'), 'no backing request.question verb');
       for (const tool of tools.filter(intake)) {
         assert.equal(tool.parameters.additionalProperties, false);
         assert.equal(tool.parameters.properties.request.type, 'string');
@@ -195,7 +196,6 @@ test('broker keeps wrong-role intake refusals, with eligible CTO reads', async t
   await assert.rejects(x.tool('intake_open', { request: x.source.id }), /request-origin-refused/);
   await assert.rejects(x.tool('intake_status', { ...args, extra: true }), /invalid-request/);
   await assert.rejects(x.tool('intake_propose', { ...args, revision: 0, op: 'bad-revision', source_project: 'factory', project: 'timer', name: 'Timer', tracker_project: 2 }), /invalid-request/);
-  await assert.rejects(x.tool('intake_question', args), /unknown-tool/);
 });
 
 test('pre-task staffing still requires PM message.send authority', async t => {
@@ -232,3 +232,74 @@ test('product task and catalog tools carry project and exact snapshot fields', t
     assert.equal(schema.properties.requirement_snapshot.type, 'string');
   }
 });
+
+for (const harness of ['pi', 'claude-code']) test(`${harness}: PM asks once and the owner answers through the bus`, async t => {
+  const x = await requestFixture(t, 'guided'), state = x.state;
+  assert.equal(state.phase, 'discovery');
+  const pm = bundle(t, 'pm', harness), socket = join(x.f.dataRoot, 'question-tools.sock');
+  const server = await runner.toolServer({ path: socket, tools: pm.tools, local: {}, client: { call: (verb, args) => x.call('pm', verb, args) } });
+  t.after(() => new Promise(resolve => server.close(resolve)));
+  x.pm = pm; x.socket = socket; x.tool = (name, args) => callTool(socket, name, args);
+  const invoke = harness === 'claude-code' ? mcp(t, x.pm.files.tools, x.socket) : x.tool;
+  const args = { request: state.request, revision: state.revision, op: 'question-1', text: 'Which platform?', options: [{ key: 'web', text: 'Web' }, { key: 'cli', text: 'CLI' }] };
+  const asked = await invoke('intake_question', args);
+  assert.equal(asked.question.text, 'Which platform?');
+  assert.deepEqual(asked.question.options, args.options);
+  assert.equal(asked.question.answer, null);
+  assert.deepEqual(await invoke('intake_question', args), asked);
+  for (const role of ['cto', 'coder', 'reviewer', 'chief'])
+    await assert.rejects(async () => x.call(role, 'request.question', { ...args, op: 'foreign' }), /request-access-refused/);
+  assert.equal(x.store.get("SELECT count(*) n FROM request_events WHERE kind='question.asked'").n, 1);
+  await assert.rejects(x.tool('intake_question', { ...args, text: 'Changed?' }), /request-operation-conflict/);
+  const answered = await x.call('console', 'request.answer.console', { request: state.request, revision: asked.revision, question: asked.question.id, op: 'answer-1', text: 'Web' });
+  assert.ok(answered.question.answer);
+  assert.equal(x.store.get('SELECT kind FROM events WHERE id=?', answered.question.answer).kind, 'human.input');
+  assert.equal((await x.tool('intake_status', { request: state.request })).question.answer, answered.question.answer);
+});
+
+test('intake_question refuses malformed arguments and broker phase refusal', async t => {
+  const x = await setup(t), state = await x.tool('intake_open', { request: x.source.request });
+  const valid = { request: state.request, revision: state.revision, op: 'q1', text: 'Question?' };
+  for (const invalid of [
+    { ...valid, extra: true }, { ...valid, revision: 0 }, { ...valid, op: '' }, { ...valid, text: '' },
+    { ...valid, text: 'a'.repeat(2001) }, { ...valid, text: 'bad\0text' },
+    { ...valid, options: [{ key: 'a', text: 'A', extra: true }] },
+    { ...valid, options: [{ key: 'a', text: 'A' }, { key: 'a', text: 'B' }] },
+    { ...valid, options: Array.from({ length: 9 }, (_, i) => ({ key: `k${i}`, text: 'A' })) },
+    { ...valid, options: [{ key: 'a', text: 'a'.repeat(201) }] },
+  ]) await assert.rejects(x.tool('intake_question', invalid), /invalid-request/);
+  await assert.rejects(x.tool('intake_question', valid), /request-phase-refused/);
+  assert.equal(x.store.get("SELECT count(*) n FROM request_events WHERE kind='question.asked'").n, 0);
+});
+
+test('intake_question reports an unknown socket outcome without retrying', async t => {
+  const root = scratch(t), path = join(root, 'lost.sock');
+  let received = 0;
+  const server = createServer(socket => socket.on('data', () => { received++; socket.end(); }));
+  await new Promise(resolve => server.listen(path, resolve));
+  t.after(() => new Promise(resolve => server.close(resolve)));
+  await assert.rejects(callTool(path, 'intake_question', { request: 'r', revision: 1, op: 'q', text: 'Q?' }), /outcome.unknown/i);
+  assert.equal(received, 1);
+});
+
+test('intake_question exposes exactly the bounded broker fields', t => {
+  const schema = bundle(t, 'pm', 'pi').tools.find(t => t.name === 'intake_question').parameters;
+  assert.deepEqual(Object.keys(schema.properties).sort(), ['op', 'options', 'request', 'revision', 'text']);
+  assert.deepEqual(schema.required.sort(), ['op', 'request', 'revision', 'text']);
+  assert.equal(schema.additionalProperties, false);
+  assert.equal(schema.properties.text.maxLength, 2000);
+  assert.equal(schema.properties.options.maxItems, 8);
+  assert.equal(schema.properties.options.items.additionalProperties, false);
+  assert.equal(schema.properties.options.items.properties.text.maxLength, 200);
+});
+
+test('intake_question reports a bus unknown outcome without calling twice', async t => {
+  const root = scratch(t), path = join(root, 'unknown.sock'), pm = bundle(t, 'pm', 'pi');
+  let attempts = 0;
+  const server = await runner.toolServer({ path, tools: pm.tools, local: {}, client: { call: () => {
+    attempts++; throw Object.assign(new Error('lost response'), { code: 'outcome-unknown' });
+  } } });
+  t.after(() => new Promise(resolve => server.close(resolve)));
+  await assert.rejects(callTool(path, 'intake_question', { request: 'r', revision: 1, op: 'q', text: 'Q?' }), e => /outcome-unknown/.test(e.message) && !/refused/.test(e.message));
+  assert.equal(attempts, 1);
+});

packages/harness/tests/tools.test.mjs

@@ -15,7 +15,7 @@ test("the PM gets its task verbs and the reads, and no launch", (t) => {
   const tools = typedTools({ resolved: pm, arbiter: true, tracker: true });
   assert.deepEqual(names(tools), [
     "send_message", "request_staff",
-    "intake_status", "intake_projects", "intake_open", "intake_propose", "advance_request", "submit_prd",
+    "intake_status", "intake_projects", "intake_open", "intake_question", "intake_propose", "advance_request", "submit_prd",
     "task_create", "task_assign", "task_reassign", "task_schedule", "task_priority_change", "task_scope_change",
     "task_update_assigned", "task_close",
     "read_spec", "write_record", "raise_decision", "resolve_decision", "list_agents", "inbox", "trail", "list_tasks", "get_task", "list_requirements",

roles/pm.md

@@ -32,6 +32,11 @@ order. You don't write the product's code and you don't review it.
   and autonomous, guided or existing choices, or records a precise setup wait.
   The user chooses and confirms; a proposal grants no bootstrap authority.
   Check status after the answer or a setup refusal before continuing.
+- During discovery, PRD review or planning, use `intake_question` to ask the
+  owner a bounded question, optionally with choices. Cite the current revision
+  and a stable operation id. Check `intake_status` if the outcome is unknown;
+  reuse the same id and exact arguments for an identical retry. The answer is
+  user input, not PRD approval.
 - Break each approved requirement into tasks on the tracker. Every task
   you create cites the requirement id it serves, such as `REQ-TASK-1`,
   and the human request that asked for it. A task without both is