diff --git a/docs/plans/2026-09-06-multi-agent-board-collaboration.md b/docs/plans/2026-09-06-multi-agent-board-collaboration.md index 7d811b7d303..94a51bbcc25 100644 --- a/docs/plans/2026-09-06-multi-agent-board-collaboration.md +++ b/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 diff --git a/docs/plans/2026-09-07-mesh-implementation-acceptance.md b/docs/plans/2026-09-07-mesh-implementation-acceptance.md index 15ae3e79584..59252c2036b 100644 --- a/docs/plans/2026-09-07-mesh-implementation-acceptance.md +++ b/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 diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 31add00aac1..3d610236c04 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/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(); + 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. diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 85dd726a411..f2d4c169217 100644 --- a/packages/cli/src/serve/server.ts +++ b/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 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 diff --git a/packages/core/src/agents/index.ts b/packages/core/src/agents/index.ts index 0d4723b37d5..60042d7681e 100644 --- a/packages/core/src/agents/index.ts +++ b/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'; diff --git a/packages/core/src/agents/mesh/capability.test.ts b/packages/core/src/agents/mesh/capability.test.ts index d19d5db53d3..be98c0d863a 100644 --- a/packages/core/src/agents/mesh/capability.test.ts +++ b/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'), ); diff --git a/packages/core/src/agents/mesh/dispatch-port.test.ts b/packages/core/src/agents/mesh/dispatch-port.test.ts index d4d99fac57b..3e0c195b8c5 100644 --- a/packages/core/src/agents/mesh/dispatch-port.test.ts +++ b/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), diff --git a/packages/core/src/agents/mesh/dispatcher.test.ts b/packages/core/src/agents/mesh/dispatcher.test.ts index 61a08262495..8adaec8bc3a 100644 --- a/packages/core/src/agents/mesh/dispatcher.test.ts +++ b/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'); diff --git a/packages/core/src/agents/mesh/supervisor.test.ts b/packages/core/src/agents/mesh/supervisor.test.ts new file mode 100644 index 00000000000..95a74b5974f --- /dev/null +++ b/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 } { + 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(); + }); +}); diff --git a/packages/core/src/agents/mesh/supervisor.ts b/packages/core/src/agents/mesh/supervisor.ts new file mode 100644 index 00000000000..538cf7c1e60 --- /dev/null +++ b/packages/core/src/agents/mesh/supervisor.ts @@ -0,0 +1,141 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * @fileoverview The loop that keeps booked work moving. + * + * Runs **inside the hidden host session**, not in the daemon. The design + * originally placed the dispatcher daemon-side, but the launcher and the + * background-agent registry live in the host session's process, and the first + * live slice was driven by hand from exactly there. Ticking where the state is + * means `inspect` is a local registry read and a start needs no round trip; + * the daemon's only job is to keep this session resident. That is decision (a) + * in the design's §5.2 step 6, and it is why nothing here talks over ACP. + * + * Two guards make it safe to run unattended: + * + * - **Only the claimed host dispatches.** The workspace record names one host + * session. A second copy of this loop — a stale session the reaper did not + * get to, or a session that lost the claim race — must never start a body, + * or one agent would have two. + * - **Ticks never overlap.** A slow launch must not let the next interval + * start a second pass over the same queued runs. A tick requested while one + * is in flight joins it and receives its outcome rather than being refused, + * so a caller that wants "the state after the next pass" always gets one. + * + * Deliberately polling. A post written by the daemon's REST route lands in the + * store, not in this process, and the design treats in-process notifications + * as hints rather than the source of truth. A few seconds of latency is the + * price of never missing a durable trigger. + */ + +import { createDebugLogger } from '../../utils/debugLogger.js'; +import { + dispatchOnce, + type DispatchRecord, + type MeshDispatchPort, +} from './dispatcher.js'; +import { readMeshAgents, readMeshWorkspace } from './mesh-store.js'; + +const log = createDebugLogger('MESH_SUPERVISOR'); + +/** Long enough that an idle workspace costs nothing noticeable, short enough + * that a person posting sees the agent start before they wonder. */ +export const DEFAULT_MESH_SUPERVISOR_INTERVAL_MS = 2_000; + +export type MeshTickOutcome = + | { kind: 'dispatched'; records: DispatchRecord[] } + | { kind: 'not_claimed_host'; claimedBy?: string } + | { kind: 'no_roster' } + | { kind: 'error'; error: string }; + +export interface MeshSupervisor { + /** One pass, now. Returns what it decided; never throws. */ + tick(): Promise; + stop(): void; +} + +export interface StartMeshSupervisorInput { + projectRoot: string; + /** This process's session id; dispatch only while it holds the claim. */ + sessionId: string; + port: MeshDispatchPort; + intervalMs?: number; + /** Observers for tests and for the daemon's status surface. */ + onTick?: (outcome: MeshTickOutcome) => void; +} + +export function startMeshSupervisor( + input: StartMeshSupervisorInput, +): MeshSupervisor { + const intervalMs = input.intervalMs ?? DEFAULT_MESH_SUPERVISOR_INTERVAL_MS; + let inFlight: Promise | undefined; + let stopped = false; + + const pass = async (): Promise => { + const workspace = await readMeshWorkspace(input.projectRoot); + if (workspace.hostSessionId !== input.sessionId) { + return { + kind: 'not_claimed_host', + ...(workspace.hostSessionId + ? { claimedBy: workspace.hostSessionId } + : {}), + }; + } + const agents = await readMeshAgents(input.projectRoot); + if (agents.length === 0) return { kind: 'no_roster' }; + const records = await dispatchOnce(input.projectRoot, input.port); + return { kind: 'dispatched', records }; + }; + + const runPass = async (): Promise => { + let outcome: MeshTickOutcome; + try { + outcome = await pass(); + } catch (error) { + outcome = { + kind: 'error', + error: error instanceof Error ? error.message : String(error), + }; + log.warn(`dispatch pass failed: ${outcome.error}`); + } + if (outcome.kind === 'dispatched' && outcome.records.length > 0) { + log.debug( + `dispatched: ${outcome.records + .map((record) => `${record.agentId}=${record.kind}`) + .join(' ')}`, + ); + } + input.onTick?.(outcome); + return outcome; + }; + + const tick = (): Promise => { + if (inFlight) return inFlight; + inFlight = runPass().finally(() => { + inFlight = undefined; + }); + return inFlight; + }; + + const timer = setInterval(() => { + if (stopped) return; + void tick(); + }, intervalMs); + timer.unref?.(); + // The first pass runs now: a host that was just revived has work waiting + // for it, and making it wait a full interval is a visible pause. + void tick(); + + return { + tick, + stop() { + if (stopped) return; + stopped = true; + clearInterval(timer); + }, + }; +}