feat(mesh): record run closes and derive thread state from them

docs/plans/2026-09-06-multi-agent-board-collaboration.md

@@ -522,14 +522,15 @@ action on the child caused it.
 Pre-existing on this branch (five production files plus two tests; nothing
 starts an agent yet):
 
-| File                                      | Responsibility                                     |
-| ----------------------------------------- | -------------------------------------------------- |
-| `core/src/agents/mesh/types.ts`           | Entities and limits                                |
-| `core/src/agents/mesh/mesh-store.ts`      | Paths, validation, locking, CRUD                   |
-| `core/src/agents/mesh/mentions.ts`        | `@name` → agent ids                                |
-| `core/src/agents/mesh/dispatch-policy.ts` | `decideDispatch` — pure                            |
-| `core/src/agents/mesh/thread-actions.ts`  | `postMessage` — append and book under one lock     |
-| `core/src/agents/mesh/thread-status.ts`   | Aggregate status over every run's close obligation |
+| File                                      | Responsibility                                        |
+| ----------------------------------------- | ----------------------------------------------------- |
+| `core/src/agents/mesh/types.ts`           | Entities and limits                                   |
+| `core/src/agents/mesh/mesh-store.ts`      | Paths, validation, locking, CRUD                      |
+| `core/src/agents/mesh/mentions.ts`        | `@name` → agent ids                                   |
+| `core/src/agents/mesh/dispatch-policy.ts` | `decideDispatch` — pure                               |
+| `core/src/agents/mesh/thread-actions.ts`  | `postMessage` — append and book under one lock        |
+| `core/src/agents/mesh/thread-status.ts`   | Aggregate status over every run's close obligation    |
+| `core/src/agents/mesh/run-lifecycle.ts`   | Run close, terminal state, status application, outbox |
 
 ### 5.1 Local review correction — committed and verified
 

docs/plans/2026-09-07-mesh-implementation-acceptance.md

@@ -58,6 +58,8 @@ Evidence: the assembled prompt text for cases first-entry / delta / gap / retry,
 
 **Aggregate status landed.** `thread-status.ts` derives the status from every run's close obligation rather than letting the last run to finish stamp it, and the three round-2 findings are each pinned by a test: a same-thread wait is discharged by a later close (I1), any later successful booking discharges an earlier failure or unclosed return (I2), and a quiescent thread whose last admission booked nothing becomes `blocked` (I6). Also covered: a live run outranks another agent's review, a blocker outranks a review, a wait is `in_progress` only while a child can wake it, `done` is sticky against a late post, and a failed run reports as a failure even when it recorded a close kind. Observed locally: `thread-status.test.ts` → 1 file, 13 tests passed; targeted ESLint clean. The producers that write `closeKind` are the thread tools in 5b, so nothing calls this resolver yet.
 
+**Run close and status application landed.** `run-lifecycle.ts` splits closing into two writes: the tool records `closeKind` and moves the run to `finishing`, ending the agent's turn, and the runtime callback records the terminal state — the only place the aggregate status is recomputed. A `waiting` close is refused when nothing could wake it, and a live _descendant_ counts while a mere sibling under the same root does not. Any close discharges peers' waits on the same thread. A clean exit with no closing tool is recorded as `unclosed`, never as implicit success. `finishRun` now delegates to this one path, and `postMessage` discharges outstanding obligations when it books work and then applies the aggregate status, so the I2 and I6 fixes have producers rather than only a resolver. Observed locally: `run-lifecycle.test.ts`, `thread-actions.test.ts`, `thread-status.test.ts`, `mesh-store.test.ts` → 4 files, 57 tests passed; targeted ESLint clean. The six thread tools that call `closeRun` are still to come, so gates (a), (b), (c) and (e) remain unexecuted.
+
 ### Step 6 — Minimal in-process dispatcher, no recovery
 
 Lands: pick the lowest `queueSequence` queued run per agent; branch on registry state `completed+resident → continue`, `completed → revive`, `paused → resume`, `unbound → launch`; `startRun` / `finishRun`; consume the parent-report outbox; leave `capacity_wait` queued.

packages/core/src/agents/mesh/run-lifecycle.test.ts

@@ -0,0 +1,344 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import * as fs from 'node:fs/promises';
+import * as os from 'node:os';
+import * as path from 'node:path';
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+
+import { Storage } from '../../config/storage.js';
+import {
+  createThread,
+  readThread,
+  updateMeshAgents,
+  writeThread,
+} from './mesh-store.js';
+import {
+  closeRun,
+  finishRunInTransaction,
+  hasLiveDescendant,
+  MeshCloseRejectedError,
+} from './run-lifecycle.js';
+import { withMeshStoreTransaction } from './mesh-store.js';
+import { postMessage } from './thread-actions.js';
+import {
+  HUMAN_AUTHOR_ID,
+  type MeshAgent,
+  type Thread,
+  type ThreadRun,
+} from './types.js';
+
+const PROJECT_ROOT = '/mesh-lifecycle-test';
+const ALICE: MeshAgent = { id: 'ag_alice', name: 'alice', createdAt: 1 };
+const BOB: MeshAgent = { id: 'ag_bob', name: 'bob', createdAt: 1 };
+
+function run(overrides: Partial<ThreadRun> = {}): ThreadRun {
+  return {
+    id: 'rn_alice',
+    agentId: ALICE.id,
+    status: 'running',
+    triggerMessageIds: [],
+    acceptedMessageIds: [],
+    consumedMessageIds: [],
+    usageByRound: [],
+    // Well clear of the workspace counter: these fixtures are hand-written and
+    // must not collide with a sequence the store allocates during the test.
+    queueSequence: 100,
+    queuedAt: 1_000,
+    attempts: 1,
+    ...overrides,
+  };
+}
+
+async function seed(overrides: Partial<Thread> = {}): Promise<Thread> {
+  const created = await createThread(PROJECT_ROOT, { title: 'Investigate' });
+  const thread: Thread = {
+    ...created,
+    status: 'in_progress',
+    runs: [run()],
+    ...overrides,
+  };
+  await writeThread(PROJECT_ROOT, thread);
+  return thread;
+}
+
+function finish(
+  threadId: string,
+  runId: string,
+  outcome: Parameters<typeof finishRunInTransaction>[1]['outcome'],
+) {
+  return withMeshStoreTransaction(PROJECT_ROOT, (transaction) =>
+    finishRunInTransaction(transaction, { threadId, runId, outcome }),
+  );
+}
+
+describe('mesh run lifecycle', () => {
+  let runtimeDir: string;
+
+  beforeEach(async () => {
+    runtimeDir = await fs.mkdtemp(path.join(os.tmpdir(), 'mesh-lifecycle-'));
+    Storage.setRuntimeBaseDir(runtimeDir);
+    await updateMeshAgents(PROJECT_ROOT, () => [ALICE, BOB]);
+  });
+
+  afterEach(async () => {
+    Storage.setRuntimeBaseDir(null);
+    await fs.rm(runtimeDir, { recursive: true, force: true });
+  });
+
+  it('posts the question, records the close, and ends the turn without finishing the run', async () => {
+    const thread = await seed();
+
+    const result = await closeRun(PROJECT_ROOT, {
+      threadId: thread.id,
+      runId: 'rn_alice',
+      agentId: ALICE.id,
+      request: { kind: 'blocked', question: 'which retry path?' },
+    });
+
+    expect(result.message?.text).toBe('which retry path?');
+    expect(result.message?.authorKind).toBe('agent');
+    expect(result.message?.sourceRunId).toBe('rn_alice');
+    expect(result.message?.authorNameSnapshot).toBe('alice');
+    // The runtime is still executing, so the run may not be marked terminal.
+    expect(result.thread.runs[0]?.status).toBe('finishing');
+    expect(result.thread.runs[0]?.closeKind).toBe('blocked');
+    expect(result.thread.runs[0]?.finalMessageId).toBe(result.message?.id);
+    expect(result.thread.status).toBe('in_progress');
+    expect(result.thread.outbox).toHaveLength(1);
+    expect(result.thread.outbox[0]?.payload['event']).toBe('blocker_raised');
+  });
+
+  it('refuses a wait that nothing could ever wake', async () => {
+    const thread = await seed();
+
+    await expect(
+      closeRun(PROJECT_ROOT, {
+        threadId: thread.id,
+        runId: 'rn_alice',
+        agentId: ALICE.id,
+        request: { kind: 'waiting' },
+      }),
+    ).rejects.toThrow(MeshCloseRejectedError);
+  });
+
+  it('allows a wait once a sub-thread is open, and not for a mere sibling', async () => {
+    const parent = await seed();
+    const child = await createThread(PROJECT_ROOT, {
+      title: 'read the code',
+      parentThreadId: parent.id,
+    });
+
+    const waited = await closeRun(PROJECT_ROOT, {
+      threadId: parent.id,
+      runId: 'rn_alice',
+      agentId: ALICE.id,
+      request: { kind: 'waiting' },
+    });
+    expect(waited.thread.runs[0]?.closeKind).toBe('waiting');
+
+    // A sibling under the same root is not this thread's dependency.
+    const sibling = await createThread(PROJECT_ROOT, {
+      title: 'unrelated',
+      parentThreadId: parent.id,
+    });
+    const threads = [
+      { ...parent },
+      { ...child, status: 'done' as const },
+      { ...sibling, status: 'done' as const },
+    ];
+    expect(hasLiveDescendant(threads, parent.id)).toBe(false);
+    expect(hasLiveDescendant([{ ...parent }, { ...child }], parent.id)).toBe(
+      true,
+    );
+  });
+
+  it('refuses a close for a run the caller does not own', async () => {
+    const thread = await seed();
+
+    await expect(
+      closeRun(PROJECT_ROOT, {
+        threadId: thread.id,
+        runId: 'rn_alice',
+        agentId: BOB.id,
+        request: { kind: 'review', summary: 'done' },
+      }),
+    ).rejects.toThrow(/not a running run of agent "ag_bob"/);
+  });
+
+  it('discharges a peer wait so a review is not reported as blocked', async () => {
+    const thread = await seed({
+      runs: [
+        run({ id: 'rn_wait', status: 'completed', closeKind: 'waiting' }),
+        run({
+          id: 'rn_bob',
+          agentId: BOB.id,
+          status: 'running',
+          queueSequence: 101,
+        }),
+      ],
+    });
+
+    const closed = await closeRun(PROJECT_ROOT, {
+      threadId: thread.id,
+      runId: 'rn_bob',
+      agentId: BOB.id,
+      request: { kind: 'review', summary: 'the flake is the retry path' },
+    });
+    expect(
+      closed.thread.runs.find((entry) => entry.id === 'rn_wait')
+        ?.closeAcknowledgedAtSequence,
+    ).toBe(1);
+
+    const finished = await finish(thread.id, 'rn_bob', { status: 'completed' });
+    expect(finished.status).toBe('in_review');
+  });
+
+  it('records a clean exit with no closing tool as unclosed and blocks', async () => {
+    const thread = await seed();
+
+    const finished = await finish(thread.id, 'rn_alice', {
+      status: 'completed',
+    });
+
+    expect(finished.runs[0]?.closeKind).toBe('unclosed');
+    expect(finished.status).toBe('blocked');
+    expect(
+      finished.outbox.some(
+        (event) => event.payload['event'] === 'thread_blocked',
+      ),
+    ).toBe(true);
+  });
+
+  it('reports a child in review to its parent exactly once', async () => {
+    const parent = await createThread(PROJECT_ROOT, { title: 'parent' });
+    const created = await createThread(PROJECT_ROOT, {
+      title: 'child',
+      parentThreadId: parent.id,
+    });
+    await writeThread(PROJECT_ROOT, {
+      ...created,
+      status: 'in_progress',
+      runs: [run()],
+    });
+
+    await closeRun(PROJECT_ROOT, {
+      threadId: created.id,
+      runId: 'rn_alice',
+      agentId: ALICE.id,
+      request: { kind: 'review', summary: 'root cause found' },
+    });
+    const finished = await finish(created.id, 'rn_alice', {
+      status: 'completed',
+    });
+
+    expect(finished.status).toBe('in_review');
+    const reports = finished.outbox.filter(
+      (event) => event.kind === 'parent_report',
+    );
+    expect(reports).toHaveLength(1);
+    expect(reports[0]?.payload['parentThreadId']).toBe(parent.id);
+
+    // Re-running the terminal write must not enqueue a second report.
+    const again = await finish(created.id, 'rn_alice', { status: 'completed' });
+    expect(again.outbox.filter((e) => e.kind === 'parent_report')).toHaveLength(
+      1,
+    );
+  });
+
+  it('carries a typed failure stage onto the run and blocks the thread', async () => {
+    const thread = await seed();
+
+    const finished = await finish(thread.id, 'rn_alice', {
+      status: 'failed',
+      error: 'definition missing',
+      failureStage: 'launch',
+    });
+
+    expect(finished.runs[0]?.failureStage).toBe('launch');
+    expect(finished.status).toBe('blocked');
+  });
+
+  it('refuses any close on a thread a person already marked done', async () => {
+    const thread = await seed({ status: 'done' });
+
+    await expect(
+      closeRun(PROJECT_ROOT, {
+        threadId: thread.id,
+        runId: 'rn_alice',
+        agentId: ALICE.id,
+        request: { kind: 'review', summary: 'late' },
+      }),
+    ).rejects.toThrow(/is done/);
+  });
+
+  it('clears an obsolete failure when a later post books real work', async () => {
+    const thread = await seed({ assigneeAgentId: ALICE.id });
+    const failed = await finish(thread.id, 'rn_alice', {
+      status: 'failed',
+      error: 'launch failed',
+    });
+    expect(failed.status).toBe('blocked');
+
+    const posted = await postMessage(PROJECT_ROOT, thread.id, {
+      from: HUMAN_AUTHOR_ID,
+      text: 'try again please',
+    });
+
+    expect(posted.dispatched).toHaveLength(1);
+    expect(posted.thread.status).toBe('in_progress');
+    expect(
+      posted.thread.runs.find((entry) => entry.id === 'rn_alice')
+        ?.closeAcknowledgedAtSequence,
+    ).toBe(1);
+  });
+
+  it('blocks a quiescent thread whose post books nothing at all', async () => {
+    const created = await createThread(PROJECT_ROOT, { title: 'unassigned' });
+
+    const posted = await postMessage(PROJECT_ROOT, created.id, {
+      from: HUMAN_AUTHOR_ID,
+      text: 'anyone?',
+    });
+
+    expect(posted.dispatched).toHaveLength(0);
+    expect(posted.thread.status).toBe('blocked');
+    expect(
+      posted.thread.outbox.some(
+        (event) => event.payload['event'] === 'thread_blocked',
+      ),
+    ).toBe(true);
+  });
+
+  it('leaves a thread in_progress while another run is still live', async () => {
+    const thread = await seed({
+      runs: [
+        run(),
+        run({
+          id: 'rn_bob',
+          agentId: BOB.id,
+          status: 'queued',
+          queueSequence: 101,
+        }),
+      ],
+    });
+
+    await closeRun(PROJECT_ROOT, {
+      threadId: thread.id,
+      runId: 'rn_alice',
+      agentId: ALICE.id,
+      request: { kind: 'review', summary: 'my part is done' },
+    });
+    const finished = await finish(thread.id, 'rn_alice', {
+      status: 'completed',
+    });
+
+    expect(finished.status).toBe('in_progress');
+    expect(await readThread(PROJECT_ROOT, thread.id)).toMatchObject({
+      status: 'in_progress',
+    });
+  });
+});