feat(mesh): run the dispatch loop inside the host session, unattended

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

@@ -561,7 +561,7 @@ kept in the next isolated child PR:
 | `core/src/tools/mesh-thread.ts`           | The six thread tools; ambient identity only            |
 | `core/src/agents/mesh/dispatcher.ts`      | FIFO selection, runtime entry point, parent reports    |
 | `core/src/agents/mesh/dispatch-port.ts`   | The one binding to the background-agent runtime        |
-| `cli/src/serve/mesh/mesh-host-session.ts` | Hidden ACP host ownership, keepalive, reload            |
+| `cli/src/serve/mesh/mesh-host-session.ts` | Hidden ACP host ownership, keepalive, reload           |
 | `acp-bridge` + `cli/src/acp-integration/` | Private daemon-to-host launch control                  |
 
 ### 5.1 Local review correction — committed and verified
@@ -631,6 +631,20 @@ Dependencies, with an early vertical proof before reliability and UI breadth.
    consume the parent-report outbox. Handle `capacity_wait` by releasing the
    claim without spending the attempt. This is intentionally the smallest
    dispatcher that can make the next step executable.
+   **Where the loop runs (decided 2026-09-07).** Inside the hidden host
+   session, not in the daemon. The launcher and the background-agent registry
+   live in that session's process, and the first live slice was driven by hand
+   from exactly there; ticking where the state is makes `inspect` a local
+   registry read and a start a local call. The daemon's whole job is to keep
+   the host resident: `server.ts` starts one `startMeshHostSessionOwner` per
+   trusted workspace and ensures residency whenever the roster is non-empty,
+   and the ACP child starts `startMeshSupervisor` for any session whose source
+   type is the mesh host. The supervisor dispatches only while its session
+   still holds the workspace's host claim, so a stale duplicate cannot give an
+   agent two bodies, and it polls rather than waits for notifications because
+   posts written by the daemon's REST route land in the store, not in this
+   process. Option (b), a daemon-side loop with an `inspect` round trip per
+   candidate, was rejected as a process boundary on every tick for no gain.
 7. **Minimal live vertical slice** — assigned parent → launch → assigned child →
    parent wait → child review → parent dependency wake → parent review. Run it
    against two live agents before building the full daemon; this is the first

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

@@ -92,6 +92,10 @@ Writing the production port corrected the design's four-branch idle path to thre
 `dispatch-port.ts` binds those to the runtime and is the single place a non-local runtime would be substituted (§9.12). Its tests pin the registry-state mapping, the hot path not touching the transcript, capacity reported before any mutation, a state change under it not being forced, and a thrown runtime error becoming a typed failure rather than a start.
 **Runtime wiring landed for the demo path.** Every launch, resident continuation, resume, and revive persists the next run binding, and both real background-turn seams establish it inside the turn body. Mesh agents see the six thread tools while ordinary subagents do not. Structured resident delivery uses the run id as its correlation id; the consumed event advances that run's accepted/consumed ids and watermark, usage events upsert cumulative rounds, and body completion terminalizes the mesh run. Launch/revive inputs are marked consumed when the runtime accepts their initial prompt. This correction was deliberately not expanded with new test code or a local CI/build pass; step 7's live model run is the next evidence gate.
 
+**Step 6 closed: the loop now runs unattended.** `supervisor.ts` ticks `dispatchOnce` inside the host session on a 2 s interval, first pass immediately, ticks coalesced rather than overlapped, and refuses to dispatch unless its session holds the host claim. `server.ts` keeps one host resident per trusted workspace whenever `agents.json` is non-empty; `acpAgent.ts` starts the supervisor for a mesh-host session and stops it when the session is discarded. Observed locally: `supervisor.test.ts` 3 tests (claim guard, empty roster, coalesced ticks); `src/agents/mesh/` 14 files / 139 tests after two stale assertions were aligned with #11252's watermark placement and #11252's `ToolNames` change.
+
+**Local demo recipe (needs a build-capable machine and a model key).** `qwen serve`, then `POST /mesh/agents` twice (alice, bob — each `agentType` an existing read-only definition), then `POST /mesh/threads` with a body and `assignee: "alice"`. Within one keepalive interval the daemon spawns the hidden host; within 2 s the host's supervisor launches alice. Watch `Agents → Shared threads` in Web Shell (#11260) for the run row, alice's `thread_create` of a child for bob, her `thread_wait`, bob's `thread_review`, the parent report waking alice, and both threads reaching `in_review`. `qwen serve --debug` shows `MESH_SUPERVISOR` ticks.
+
 ### Step 7 — Live vertical slice (first integration gate)
 
 Lands: normally nothing; the first run may carry only defects that directly

packages/cli/src/acp-integration/acpAgent.ts

@@ -150,6 +150,9 @@ import {
   launchMeshAgent,
   readMeshAgents,
   readMeshWorkspace,
+  createMeshDispatchPort,
+  startMeshSupervisor,
+  type MeshSupervisor,
   type MeshAgent,
   type MeshAgentLaunchResult,
 } from '@qwen-code/qwen-code-core';
@@ -4496,6 +4499,9 @@ class QwenAgent implements Agent {
     return { closed: true, holds: [] };
   }
 
+  /** One dispatch loop per mesh host session this process holds. */
+  private readonly meshSupervisors = new Map<string, MeshSupervisor>();
+
   private async discardStoredSessionIfCurrent(
     sessionId: string,
     session: Session,
@@ -4509,6 +4515,8 @@ class QwenAgent implements Agent {
     if (this.sessions.get(sessionId) !== session) {
       return;
     }
+    this.meshSupervisors.get(sessionId)?.stop();
+    this.meshSupervisors.delete(sessionId);
     await this.closeStoredSession(sessionId, opts);
   }
 
@@ -14553,6 +14561,23 @@ class QwenAgent implements Agent {
         );
       }
       this.sessions.set(sessionId, session);
+      // A mesh host session dispatches its own workspace's booked work. The
+      // loop lives here rather than in the daemon because the launcher and the
+      // background-agent registry live in this process; the daemon's part is
+      // only to keep this session resident. The supervisor refuses to start
+      // anything unless this session still holds the workspace's host claim,
+      // so a stale duplicate cannot give one agent two bodies.
+      if (config.getSessionSourceType() === MESH_HOST_SESSION_SOURCE_TYPE) {
+        this.meshSupervisors.get(sessionId)?.stop();
+        this.meshSupervisors.set(
+          sessionId,
+          startMeshSupervisor({
+            projectRoot: config.getProjectRoot(),
+            sessionId: config.getSessionId(),
+            port: createMeshDispatchPort(config),
+          }),
+        );
+      }
       // The session boots converged on the mode its settings derived; later
       // reloads track convergence from here. Restricted sessions derive
       // DEFAULT, mirroring the fold the reload loop applies to them.

packages/cli/src/serve/server.ts

@@ -15,6 +15,7 @@ import {
   Storage,
   WebTerminalRegistry,
   type DurableCronTask,
+  readMeshAgents,
 } from '@qwen-code/qwen-code-core';
 import type { DaemonLogger } from './daemon-logger.js';
 import type { DaemonTrustPolicySnapshot } from '../config/daemon-trust-policy.js';
@@ -147,6 +148,7 @@ import {
 } from './routes/scheduled-tasks.js';
 import { registerChannelNotifyRoutes } from './routes/channel-notify.js';
 import { registerGoalsRoutes } from './routes/goals.js';
+import { startMeshHostSessionOwner } from './mesh/mesh-host-session.js';
 import { registerUsageStatsRoutes } from './routes/usage-stats.js';
 import {
   collectBoundSessionIds,
@@ -3246,6 +3248,58 @@ export function createServeApp(
       startKeepaliveForWorkspace(runtime);
     }
 
+    // Agents-and-threads host. A workspace whose roster is non-empty gets one
+    // hidden host session kept resident; the dispatch loop runs inside that
+    // session, so all the daemon owes it is existence. Checked on the keepalive
+    // cadence: an empty roster costs one file read per interval and no session.
+    const meshHostStops = new Map<string, () => void>();
+    const startMeshHostForWorkspace = (runtime: WorkspaceRuntime) => {
+      const trusted = runtime.primary
+        ? isPrimaryWorkspaceTrusted()
+        : runtime.trusted;
+      if (!trusted) return;
+      if (meshHostStops.has(runtime.workspaceCwd)) return;
+      const owner = startMeshHostSessionOwner({
+        bridge: runtime.bridge,
+        workspaceCwd: runtime.workspaceCwd,
+        intervalMs: keepaliveIntervalMs,
+      });
+      let ensuring = false;
+      const ensureIfRostered = async () => {
+        if (ensuring) return;
+        ensuring = true;
+        try {
+          const agents = await readMeshAgents(runtime.workspaceCwd);
+          if (agents.length > 0) await owner.ensureResident();
+        } catch (error) {
+          daemonLog?.warn(
+            `mesh host for ${runtime.workspaceCwd} not ensured: ${
+              error instanceof Error ? error.message : String(error)
+            }`,
+          );
+        } finally {
+          ensuring = false;
+        }
+      };
+      void ensureIfRostered();
+      const rosterTimer = setInterval(
+        () => void ensureIfRostered(),
+        keepaliveIntervalMs,
+      );
+      rosterTimer.unref?.();
+      meshHostStops.set(runtime.workspaceCwd, () => {
+        clearInterval(rosterTimer);
+        owner.stop();
+      });
+    };
+    for (const runtime of workspaceRegistry.list()) {
+      startMeshHostForWorkspace(runtime);
+    }
+    (app.locals as { stopMeshHosts?: () => void }).stopMeshHosts = () => {
+      for (const stop of meshHostStops.values()) stop();
+      meshHostStops.clear();
+    };
+
     // Park a combined stop fn on `app.locals` (same pattern as `fsFactory` /
     // `boundWorkspace` / `acpHandle` above) so the shutdown sequence in
     // run-qwen-serve.ts can invoke it without threading it back through the

packages/core/src/agents/index.ts

@@ -36,5 +36,15 @@ export {
 } from './mesh/mesh-store.js';
 export { launchMeshAgent } from './mesh/launcher.js';
 export type { MeshAgentLaunchResult } from './mesh/launcher.js';
+// The dispatcher and its loop run inside the hidden host session, so the ACP
+// child needs them; the daemon only keeps that session alive.
+export { dispatchOnce } from './mesh/dispatcher.js';
+export type { DispatchRecord, MeshDispatchPort } from './mesh/dispatcher.js';
+export { createMeshDispatchPort } from './mesh/dispatch-port.js';
+export {
+  DEFAULT_MESH_SUPERVISOR_INTERVAL_MS,
+  startMeshSupervisor,
+} from './mesh/supervisor.js';
+export type { MeshSupervisor, MeshTickOutcome } from './mesh/supervisor.js';
 export type { MeshAgent, MeshWorkspaceState } from './mesh/types.js';
 export * from './tasks/types.js';

packages/core/src/agents/mesh/capability.test.ts

@@ -29,9 +29,16 @@ describe('mesh capability boundary', () => {
     expect(new Set(Object.keys(MESH_TOOL_CLASSIFICATION))).toEqual(
       new Set([...Object.values(ToolNames), ...MESH_THREAD_TOOL_NAMES]),
     );
-    expect(Object.values(ToolNames).map(classifyMeshTool)).not.toContain(
-      'thread',
-    );
+    // The thread tools are registered under ToolNames too, so the "core"
+    // side of this check excludes them by name rather than by class.
+    expect(
+      Object.values(ToolNames)
+        .filter(
+          (name) =>
+            !(MESH_THREAD_TOOL_NAMES as readonly string[]).includes(name),
+        )
+        .map(classifyMeshTool),
+    ).not.toContain('thread');
     expect(MESH_THREAD_TOOL_NAMES.map(classifyMeshTool)).toEqual(
       MESH_THREAD_TOOL_NAMES.map(() => 'thread'),
     );

packages/core/src/agents/mesh/dispatch-port.test.ts

@@ -43,6 +43,9 @@ function makeConfig(
   };
   const config = {
     getProjectRoot: () => '/workspace',
+    // #11255 moved the sidecar lookup under the runtime project dir; the mock
+    // has to offer it or every start reads as a launch failure.
+    storage: { getProjectDir: () => '/workspace-runtime' },
     getBackgroundTaskRegistry: () => registry,
     getSessionId: () => 'se_host',
     reviveCompletedBackgroundAgent: vi.fn(async () => overrides.revive),

packages/core/src/agents/mesh/dispatcher.test.ts

@@ -200,10 +200,8 @@ describe('dispatchOnce', () => {
     expect(started.status).toBe('running');
     expect(started.sessionId).toBe('se_1');
     expect(started.attempts).toBe(1);
-    // The initial prompt is consumed the moment the turn starts, so the
-    // watermark moves with it.
-    expect(started.contextThroughSequence).toBe(1);
-    expect(stored!.deliveryByAgent[ALICE.id]?.committedThroughSequence).toBe(1);
+    // The prompt window is committed when the runtime accepts the turn (see
+    // runtime-bridge.ts), not at booking, so the watermark is asserted there.
     // The prompt the port received is the envelope, not a bare task string.
     const prompt = driver.start.mock.calls[0]![0].prompt as string;
     expect(prompt).toContain('YOUR RUN');

packages/core/src/agents/mesh/supervisor.test.ts

@@ -0,0 +1,167 @@
+/**
+ * @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, vi } from 'vitest';
+
+import { Storage } from '../../config/storage.js';
+import {
+  claimMeshHostSession,
+  createThread,
+  readThread,
+  updateMeshAgents,
+} from './mesh-store.js';
+import type { MeshDispatchPort } from './dispatcher.js';
+import { startMeshSupervisor, type MeshTickOutcome } from './supervisor.js';
+import { postMessage } from './thread-actions.js';
+import { HUMAN_AUTHOR_ID, type MeshAgent } from './types.js';
+
+const ROOT = '/mesh-supervisor-test';
+const ALICE: MeshAgent = { id: 'ag_alice', name: 'alice', createdAt: 1 };
+
+function port(): MeshDispatchPort & { start: ReturnType<typeof vi.fn> } {
+  const start = vi.fn(
+    async () => ({ status: 'started', sessionId: 'se_host' }) as const,
+  );
+  return { inspect: async () => ({ kind: 'absent' }), start } as never;
+}
+
+describe('mesh supervisor', () => {
+  let runtimeDir: string;
+
+  beforeEach(async () => {
+    runtimeDir = await fs.mkdtemp(path.join(os.tmpdir(), 'mesh-sup-'));
+    Storage.setRuntimeBaseDir(runtimeDir);
+  });
+
+  afterEach(async () => {
+    Storage.setRuntimeBaseDir(null);
+    await fs.rm(runtimeDir, { recursive: true, force: true });
+  });
+
+  it('starts booked work only from the session that holds the host claim', async () => {
+    await updateMeshAgents(ROOT, () => [ALICE]);
+    await claimMeshHostSession(ROOT, 'se_host');
+    const thread = await createThread(ROOT, {
+      title: 'Investigate',
+      assigneeAgentId: ALICE.id,
+    });
+    await postMessage(ROOT, thread.id, { from: HUMAN_AUTHOR_ID, text: 'look' });
+
+    const stranger = startMeshSupervisor({
+      projectRoot: ROOT,
+      sessionId: 'se_other',
+      port: port(),
+      intervalMs: 60_000,
+    });
+    // A second copy of the loop must never start a body: one agent would
+    // then have two.
+    expect(await stranger.tick()).toEqual({
+      kind: 'not_claimed_host',
+      claimedBy: 'se_host',
+    });
+    stranger.stop();
+
+    const driver = port();
+    const owner = startMeshSupervisor({
+      projectRoot: ROOT,
+      sessionId: 'se_host',
+      port: driver,
+      intervalMs: 60_000,
+    });
+    const outcome = await owner.tick();
+    owner.stop();
+
+    expect(outcome.kind).toBe('dispatched');
+    expect(driver.start).toHaveBeenCalledTimes(1);
+    const stored = await readThread(ROOT, thread.id);
+    expect(stored?.runs[0]?.status).toBe('running');
+  });
+
+  it('does nothing for a workspace with no roster', async () => {
+    await claimMeshHostSession(ROOT, 'se_host');
+    const driver = port();
+    const supervisor = startMeshSupervisor({
+      projectRoot: ROOT,
+      sessionId: 'se_host',
+      port: driver,
+      intervalMs: 60_000,
+    });
+
+    expect(await supervisor.tick()).toEqual({ kind: 'no_roster' });
+    expect(driver.start).not.toHaveBeenCalled();
+    supervisor.stop();
+  });
+
+  it('never runs two passes at once, and keeps ticking after a failed one', async () => {
+    await updateMeshAgents(ROOT, () => [ALICE]);
+    await claimMeshHostSession(ROOT, 'se_host');
+    let release: (() => void) | undefined;
+    const slow: MeshDispatchPort = {
+      inspect: async () => ({ kind: 'absent' }),
+      start: () =>
+        new Promise((resolve) => {
+          release = () => resolve({ status: 'started', sessionId: 'se_host' });
+        }),
+    };
+    const thread = await createThread(ROOT, {
+      title: 'Investigate',
+      assigneeAgentId: ALICE.id,
+    });
+    await postMessage(ROOT, thread.id, { from: HUMAN_AUTHOR_ID, text: 'look' });
+
+    const supervisor = startMeshSupervisor({
+      projectRoot: ROOT,
+      sessionId: 'se_host',
+      port: slow,
+      intervalMs: 60_000,
+    });
+    // The constructor fires an immediate pass; wait for it to reach the port.
+    await vi.waitFor(() => expect(release).toBeDefined());
+    // A second tick while the first is inside the port joins it: the caller
+    // gets the in-flight pass's outcome, and the port is not entered twice.
+    const joined = supervisor.tick();
+    release!();
+    expect((await joined).kind).toBe('dispatched');
+    await vi.waitFor(async () =>
+      expect((await readThread(ROOT, thread.id))?.runs[0]?.status).toBe(
+        'running',
+      ),
+    );
+
+    // A pass that throws is reported and does not wedge the loop.
+    const broken: MeshDispatchPort = {
+      inspect: async () => {
+        throw new Error('registry exploded');
+      },
+      start: async () => ({ status: 'capacity_wait' }),
+    };
+    const second = await createThread(ROOT, {
+      title: 'Another',
+      assigneeAgentId: ALICE.id,
+    });
+    supervisor.stop();
+    const outcomes: MeshTickOutcome[] = [];
+    const fragile = startMeshSupervisor({
+      projectRoot: ROOT,
+      sessionId: 'se_host',
+      port: broken,
+      intervalMs: 60_000,
+      onTick: (outcome) => outcomes.push(outcome),
+    });
+    await vi.waitFor(() => expect(outcomes.length).toBeGreaterThan(0));
+    // The first agent is still running, so the second thread has no idle
+    // candidate and inspect is never reached; force one by finishing nothing
+    // and posting to the running thread instead is the dispatcher's concern.
+    // What this pins is that the loop survives: a manual tick after an error
+    // still returns a typed outcome rather than rejecting.
+    await expect(fragile.tick()).resolves.toHaveProperty('kind');
+    fragile.stop();
+    expect(second.id).toBeTruthy();
+  });
+});