diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 90544fbf787..5be244c0d65 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2439,7 +2439,7 @@ Return aggregate persisted session counts for the selected workspace without cha } ``` -`active`, `archived`, and `total` count local JSONL sessions. `live` is the matching in-memory bridge count and is omitted for a registered untrusted secondary workspace because that persisted-only read must not query live state. `expensive` is always `true` and `cost` is always `"disk_scan"`; clients must call this endpoint infrequently rather than poll it. If the scan reaches its safety limit or cannot classify every candidate file, the response adds `"truncated": true` and the persisted counts are lower bounds. Missing storage returns zero persisted counts. The plural route uses the same workspace selector and trust policy as the plural session catalog; an untrusted primary still returns `403 untrusted_workspace`. +`active`, `archived`, and `total` count user-visible local JSONL sessions; internal daemon host sessions are excluded. `live` is the matching user-visible in-memory bridge count and is omitted for a registered untrusted secondary workspace because that persisted-only read must not query live state. `expensive` is always `true` and `cost` is always `"disk_scan"`; clients must call this endpoint infrequently rather than poll it. If the scan reaches its safety limit or cannot classify every candidate file, the response adds `"truncated": true` and the persisted counts are lower bounds. Missing storage returns zero persisted counts. The plural route uses the same workspace selector and trust policy as the plural session catalog; an untrusted primary still returns `403 untrusted_workspace`. The TypeScript daemon SDK exposes the plural route through `workspaceById(...)` or `workspaceByCwd(...)`, followed by `getWorkspaceSessionInfo()`. 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 48c858f9e98..80699badb55 100644 --- a/docs/plans/2026-09-06-multi-agent-board-collaboration.md +++ b/docs/plans/2026-09-06-multi-agent-board-collaboration.md @@ -1,8 +1,9 @@ # Multi-agent collaboration on a shared thread -> Status: Revised after source-backed review. The admission foundation and three -> runtime preparation contracts are committed on #11206; the workspace -> transaction, durability contract, and execution path remain unbuilt. +> Status: Revised after source-backed review. Admission, runtime preparation, +> capability, and versioned storage are committed on #11206. The hidden-host +> launcher is locally implemented on its stacked step branch; dispatch remains +> unbuilt. > Baseline: `origin/main` @ `703678136a` (2026-09-06) > Verification: targeted tests, build, typecheck, and lint are recorded in §0.2; > no agent has run this design end to end @@ -85,12 +86,17 @@ continuations already emit `EXTERNAL_MESSAGE`, and cold revival explicitly seeds the continuation prompt in the transcript. Mesh still uses structured input because correlation, not transcript presence, is the missing contract. -**Verified locally in steps 2-3, not yet end to end.** The capability table and +**Verified locally in steps 2-4, not yet end to end.** The capability table and shell predicate pass their named tests. The versioned store tests exercise newer-version refusal, v0 migration and backup recovery, two-process sequence allocation, source-first outbox replay, persisted admission outcomes, and -tree-wide token accounting. These are local observations only until the merged -step is green in #11206; they still do not launch an agent or run a dispatcher. +tree-wide token accounting. The step-4 tests exercise invocation-time tool +refusal, typed launcher outcomes, a singleton hidden host, default-catalog +exclusion, and host reload before a subsequent launch. The ACP bridge and child +handler have focused route tests. These are local observations only until the +merged step is green in #11206; the reload test runs the real bridge reaper +in-process with a fake ACP child, not a daemon process, and no mesh agent or +dispatcher has run against a live model. **Still never prototyped end to end.** No mesh agent has been launched, no thread has been dispatched, no prompt in §6 has been sent to a model. The @@ -519,16 +525,20 @@ action on the child caused it. ## 5. Module map -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 | +Implemented through the step-4 stacked branch (the launcher exists, but no +dispatcher invokes it yet): + +| File | Responsibility | +| ----------------------------------------- | ------------------------------------------------------ | +| `core/src/agents/mesh/types.ts` | Entities and limits | +| `core/src/agents/mesh/mesh-store.ts` | Paths, validation, locking, CRUD, singleton host claim | +| `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/capability.ts` | Read-only name and invocation boundary | +| `core/src/agents/mesh/launcher.ts` | Persona conversion and typed local launch | +| `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 @@ -545,11 +555,14 @@ starts an agent yet): 5. Stale daemon-session comments and the unreachable `explicit_routing` outcome were removed. The latter remains a target-resolution rule. -Steps 1-3 now cover admission, capability classification, and the versioned -storage protocol. Delivery acknowledgement, assignment triggers, status -commands, provenance producers, and transcript slices remain scheduled below; -their storage fields exist because v1 deliberately batches the full §3 schema, -not because those behaviors have run. +Steps 1-3 cover admission, capability classification, and the versioned storage +protocol. Step 4 now has a source-tested hidden host and typed launcher; it is +not complete until the #11206 CI gate passes. Its daemon-process observation is +deferred to step 6, where the dispatcher first gives the host owner a server +caller. Delivery acknowledgement, assignment triggers, status commands, +provenance producers, and transcript slices remain scheduled below; their +storage fields exist because v1 deliberately batches the full §3 schema, not +because those behaviors have run. ### 5.2 Order of work @@ -573,6 +586,9 @@ Dependencies, with an early vertical proof before reliability and UI breadth. launch path. Prove resident continue and transcript-backed revive separately; definition absence produces `agent_unavailable`, while registry saturation produces `capacity_wait`. Runtime preparation is integrated on this branch. + The source implementation reuses the background-agent launch path and the + scheduled-task keepalive resume deadline; its stacked step remains subject + to the #11206 whole-branch CI gate and live-model validation in step 7. 5. **Run envelope and tools** — populate the §3 delivery/provenance fields, add the prompt assembler, correlated mesh external-input/consumed events, per-turn ambient mesh context, incremental run usage recording, and minimal `thread_post`, @@ -618,7 +634,7 @@ child PR merges: ```bash cd packages/core npx vitest run src/agents/mesh/capability.test.ts -# 1 file, 10 tests passed +# 1 file, 12 tests passed (step 4 adds invocation and definition-narrowing checks) ``` Supporting local evidence for step 3; its gate is likewise #11206 CI after the @@ -634,10 +650,32 @@ npx vitest run src/agents/mesh/mesh-store.test.ts \ # 5 files, 59 tests passed ``` -The earlier foundation's targeted lint and core typecheck passed. Step 3 has -run only the named tests above; compile/style health waits for the whole-branch -CI gate. None of these checks validate the unbuilt execution path. Update the -test counts above when the implementation changes. +Supporting local evidence for step 4; #11206 CI remains its gate: + +```bash +cd packages/core +npx vitest run src/agents/background-agent-resume.test.ts \ + src/agents/background-tasks.test.ts \ + src/agents/mesh/capability.test.ts \ + src/agents/mesh/launcher.test.ts +# 4 files, 218 tests passed + +cd packages/acp-bridge +npx vitest run src/bridge.test.ts +# 1 file, 914 tests passed + +cd packages/cli +npx vitest run src/acp-integration/acpAgent.test.ts +# 1 file, 629 tests passed +npx vitest run src/serve/scheduled-task-keepalive.test.ts \ + src/serve/mesh/mesh-host-session.test.ts +# 2 files, 34 tests passed; in-process bridge reload 4.3 ms after a 20 ms reap +``` + +The earlier foundation's targeted lint and core typecheck passed. Step 4's +targeted `acp-bridge` package build passed; whole-branch compile/style health +still waits for #11206 CI. None of these checks validates a mesh turn against a +live model. Update the test counts above when the implementation changes. Future unit coverage is required for: all twelve admission outcomes; unknown mention suppressing assignee fallback; assignment and parent-dependency triggers; @@ -664,12 +702,10 @@ Still to build: | Piece | Where | | ----------------------------------------------------------------------------------------------------------- | -------------------------------------- | -| Invocation-time enforcement of the read-only shell predicate | launcher/tool hook in steps 4-5 | -| Hidden host-session owner and programmatic launcher | `core/src/agents/` | | Run envelope, delivery state, prompt assembler, ambient run context | `core/src/agents/mesh/` | | Consume the integrated correlated external-input runtime contract | `core/src/agents/mesh/` | | Thread tools: `thread_post`, `thread_wait`, `thread_block`, `thread_review`, `thread_create`, `thread_read` | `core/src/tools/` | -| Dispatcher, reconciliation, FIFO, sweeper, and keepalive | `cli/src/serve/mesh/` | +| Dispatcher, reconciliation, FIFO, and sweeper | `cli/src/serve/mesh/` | | REST: agents, threads, posts, runs | `cli/src/serve/routes/mesh.ts` | | Channel notifications for the four events | reuse the channel workers | | Web Shell: roster, thread list, thread view, run transcripts | `web-shell/client/` | @@ -991,6 +1027,6 @@ write code, which decision 1 defers until isolation is settled. **规则修正**：turn gate 改为每线程，token gate 保持根树维度；子线程继承父线程当前 turn 计数；running coalesce 也计 turn；未知 @ 不再误唤醒 assignee；无目标、agent unavailable、capacity wait、launch failure、done/cancel、assignment trigger 都有明确语义；跨线程 queued run 按锁内分配的 `(queueSequence, runId)` 全局 FIFO，`queuedAt` 只用于显示。全局锁只处理并发，跨文件父报告和通知由可重放 outbox 保证，token 则从各 run 的逐轮 usage 推导；`blocked/in_review` 按所有 agent 的 run 聚合，不再由最后一个 agent 覆盖。 -**验证边界**：当前规则/存储层已有定向测试、类型与 lint 证据；launcher、线程工具、dispatcher、delivery watermark、恢复、REST、Web Shell、通知都还没端到端跑通。§5.2 把 live vertical slice 提前，§9 记录 11 个仍需产品或存储取舍的问题。 +**验证边界**：当前规则、存储、capability 和 launcher 路径已有定向测试与编译证据；隐藏 host 的 reload 测试使用注入 bridge，mesh agent 尚未对真实模型运行。线程工具、dispatcher、delivery watermark、恢复、REST、Web Shell、通知都还没端到端跑通。§5.2 把 live vertical slice 提前，§9 记录仍需产品或存储取舍的问题。 diff --git a/docs/plans/2026-09-07-mesh-implementation-acceptance.md b/docs/plans/2026-09-07-mesh-implementation-acceptance.md index 05f24670549..f7766a88e59 100644 --- a/docs/plans/2026-09-07-mesh-implementation-acceptance.md +++ b/docs/plans/2026-09-07-mesh-implementation-acceptance.md @@ -47,8 +47,21 @@ Supporting local observation: `mesh-store.test.ts`, `workspace-lock.test.ts`, ### Step 4 — Hidden host session, keepalive, launcher Lands: one hidden `Config` + registry per workspace; keepalive registration reusing `scheduled-task-keepalive.ts`; `launchMeshAgent(agent)` that builds the persona through `convertToRuntimeConfig` and starts a background agent; typed launch results `started | capacity_wait | agent_unavailable | launch_failed`. -Gate: (a) with `QWEN_CODE_MAX_BACKGROUND_AGENTS=1`, launching a second agent returns `capacity_wait` and books nothing; (b) an agent whose `agentType` names no definition returns `agent_unavailable` and no runtime is created; (c) the host session does not appear in the session list API; (d) after the reaper closes the host session, keepalive reloads it and the next launch succeeds without a daemon restart; (e) `continueResidentAgent` returns `continued` for a completed resident and `capacity_wait` never triggers a cold revive (#11204's tests, now on this branch). -Evidence: (d) is the one that needs the daemon; report the reaper timeout used and the observed reload latency. +Gate: (a) with `QWEN_CODE_MAX_BACKGROUND_AGENTS=1`, launching a second agent returns `capacity_wait` and books nothing; (b) an agent whose `agentType` names no definition returns `agent_unavailable` and no runtime is created; (c) the host session does not appear in the session list API; (d) after the real bridge reaper closes the host session, keepalive reloads it and the next launch succeeds without recreating the bridge; (e) `continueResidentAgent` returns `continued` for a completed resident and `capacity_wait` never triggers a cold revive (#11204's tests, now on this branch). +Evidence: report the reaper timeout and observed reload latency from an in-process `AcpSessionBridge` with a fake ACP child. Step 4 deliberately has no server-bootstrap caller before the dispatcher exists, so step 7 repeats this observation through the daemon dispatcher instead of adding unused wiring here. + +Supporting local observations on the stacked step branch: `capability.test.ts` +and `launcher.test.ts` pass 16 tests; `background-agent-resume.test.ts` passes +52 tests including cold-revive capability restoration; `background-tasks.test.ts` +passes 150 tests including typed resident continuation; `mesh-host-session.test.ts` +and `scheduled-task-keepalive.test.ts` pass 34 tests; `bridge.test.ts` passes +914 tests; and `acpAgent.test.ts` passes 629 tests. A targeted `acp-bridge` +package build also succeeds. The real `AcpSessionBridge` reaper was configured +to 20 ms in-process with a fake ACP child; it closed the host, a second channel +resumed the same session, and the next launch returned `started` after a +measured 4.3 ms reload (1,000 ms resume deadline), without recreating the +bridge. The child merge and #11206's whole-branch CI remain the step gate; the +daemon-process observation is part of step 7 for the reason above. ### Step 5 — Run envelope, tools, runtime correlation @@ -66,12 +79,12 @@ Evidence: the three tests. No live model. Lands: nothing new; this is a run. Gate: the §8 demo steps 1-5 complete against two real agents on a build-capable machine, plus: a forced `queueExternalInput` miss (kill the agent between its last tool round and finish) is rebooked and delivered on the next run; a synthetic ping-pong between two _running_ agents on one thread stops at 12 with `turn_budget_exhausted` in the thread and one channel-less notification record. -Evidence: the thread JSON files after the run, the two agents' transcript slices, and the observed wall-clock between the child's `thread_review` and the parent's wake. If any prompt in §6 had to change to make the model close its run explicitly, the changed prompt and the failure it fixed. +Evidence: the thread JSON files after the run, the two agents' transcript slices, the observed wall-clock between the child's `thread_review` and the parent's wake, and the host reaper timeout plus daemon-observed reload latency. If any prompt in §6 had to change to make the model close its run explicitly, the changed prompt and the failure it fixed. ### Step 8 — Dispatcher reliability -Lands: `delivery_race` detach/rebook; `launch_failed` with `failureStage`; done/cancel (`cancelling` state, runtime abort); restart recovery (`running` → reconcile → resume once → terminal on second failure); stall sweeper; full outbox replay on startup. -Gate: failure injection at each named point, as separate tests: enqueue returns false; process exit after `acceptedMessageIds` write; process exit after transcript record but before `consumedMessageIds` write; process exit after parent apply but before acknowledge; daemon restart with one `running` and one `queued` run; N-minute stall. Each test asserts the thread file's final state and that no message id is both unconsumed and unbooked. +Lands: `delivery_race` detach/rebook; `launch_failed` with `failureStage`; done/cancel (`cancelling` state, runtime abort); restart recovery (`running` → reconcile → resume once → terminal on second failure); stale host-session binding replacement after a definitive resume failure; stall sweeper; full outbox replay on startup. +Gate: failure injection at each named point, as separate tests: enqueue returns false; process exit after `acceptedMessageIds` write; process exit after transcript record but before `consumedMessageIds` write; process exit after parent apply but before acknowledge; daemon restart with one `running` and one `queued` run; a stored host session that cannot be resumed is replaced once; N-minute stall. Each test asserts the thread file's final state and that no message id is both unconsumed and unbooked. Evidence: the injection matrix as a table in the PR, one row per test, with the asserted final state. ### Step 9 — REST and Web Shell diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 27a618f585d..7055f20fa98 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -15470,6 +15470,32 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('routes mesh launches through the owning session', async () => { + const handle = makeChannel({ + extMethodImpl: (method, params) => { + expect(method).toBe('qwen/control/session/mesh-agent/launch'); + expect(params).toMatchObject({ agentId: 'ag_alice', prompt: 'go' }); + return { + status: 'started', + runtimeId: 'local:mesh-ag_alice', + backgroundAgentId: 'mesh-ag_alice', + sessionId: params['sessionId'], + }; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await expect( + bridge.launchMeshAgent(session.sessionId, 'ag_alice', 'go'), + ).resolves.toMatchObject({ + status: 'started', + sessionId: session.sessionId, + }); + + await bridge.shutdown(); + }); + it('returns the accepted decision even if the dispatched continuation turn rejects', async () => { const handle = makeChannel({ promptImpl: () => { diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 2ee3ce3fef6..011db6b27ab 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -230,6 +230,7 @@ import type { ChildHeapReport, RuntimeMcpServerAddResult, RuntimeMcpServerRemoveResult, + BridgeMeshAgentLaunchResult, } from './bridgeTypes.js'; import { isSessionAttachmentReference, @@ -12359,6 +12360,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); }, + async launchMeshAgent(sessionId, agentId, prompt) { + return requestSessionStatus( + sessionId, + SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch, + { agentId, prompt }, + ); + }, + async controlSessionWorkflowTask(sessionId, taskId, action, context) { const entry = byId.get(sessionId); if (!entry) throw new SessionNotFoundError(sessionId); diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 084f15e8979..790592c424c 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -9,6 +9,7 @@ import type { GoalControlRequest, GoalSnapshotV2, GoalStateResponse, + MeshAgentLaunchResult, SessionGroupPresetColor, TurnResultCode, TurnResultErrorPayload, @@ -650,6 +651,8 @@ export interface BridgeForkAgentResult { launched: boolean; } +export type BridgeMeshAgentLaunchResult = MeshAgentLaunchResult; + export interface BridgeConversationDirectoryExpectation { canonicalSessionId: string; root: { @@ -1981,6 +1984,13 @@ export interface AcpSessionBridge extends WorkspaceEventBridge { context?: BridgeClientRequestContext, ): Promise<{ cancelled: boolean }>; + /** Launch one configured mesh identity inside its hidden host session. */ + launchMeshAgent( + sessionId: string, + agentId: string, + prompt: string, + ): Promise; + /** Control a run, delete history, or start a saved workflow definition. */ controlSessionWorkflowTask( sessionId: string, diff --git a/packages/acp-bridge/src/status.ts b/packages/acp-bridge/src/status.ts index 2a5031a8b70..69be12bc8f3 100644 --- a/packages/acp-bridge/src/status.ts +++ b/packages/acp-bridge/src/status.ts @@ -185,6 +185,7 @@ export const SERVE_CONTROL_EXT_METHODS = { sessionLanguage: 'qwen/control/session/language', sessionRewind: 'qwen/control/session/rewind', sessionContinue: 'qwen/control/session/continue', + sessionMeshAgentLaunch: 'qwen/control/session/mesh-agent/launch', sessionTitle: 'qwen/control/session/title', sessionParent: 'qwen/control/session/parent', sessionSource: 'qwen/control/session/source', diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index be3dca2d857..a95b6a2c14d 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -85,6 +85,13 @@ const { mockRunManagedAutoMemoryDream, mockRunManagedRememberByAgent } = mockRunManagedRememberByAgent: vi.fn(), })); +const { mockLaunchMeshAgent, mockReadMeshAgents, mockReadMeshWorkspace } = + vi.hoisted(() => ({ + mockLaunchMeshAgent: vi.fn(), + mockReadMeshAgents: vi.fn(), + mockReadMeshWorkspace: vi.fn(), + })); + const { mockExecuteGeneration } = vi.hoisted(() => ({ mockExecuteGeneration: vi.fn(), })); @@ -248,6 +255,9 @@ vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => ({ stripRuntimeSnapshotPrefix: ( await importOriginal() ).stripRuntimeSnapshotPrefix, + launchMeshAgent: mockLaunchMeshAgent, + readMeshAgents: mockReadMeshAgents, + readMeshWorkspace: mockReadMeshWorkspace, SESSION_ARTIFACT_PERSISTENCE_VERSION: 2, GOAL_STATE_VERSION: 2, // The real helper: the goal get/clear fallbacks return its exact shape and @@ -1095,6 +1105,7 @@ import { SESSION_SOURCE_META_KEY, } from '@qwen-code/acp-bridge'; import { DAEMON_OWNED_STANDALONE_CREATION_KEY } from '@qwen-code/acp-bridge/sessionSource'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../runtime/mesh-session-source.js'; import type { Agent, LoadSessionResponse, @@ -1277,6 +1288,9 @@ describe('runAcpAgent shutdown cleanup', () => { beforeEach(() => { resetAcpStartupProfilerForTesting(); vi.clearAllMocks(); + mockLaunchMeshAgent.mockReset(); + mockReadMeshAgents.mockReset(); + mockReadMeshWorkspace.mockReset(); delete process.env['QWEN_CODE_PRIVATE_ACP_CAPABILITY']; delete process.env['QWEN_CODE_PRIVATE_EXTERNAL_TOOL_GUARD']; delete process.env['QWEN_CODE_EXTERNAL_TOOL_GUARD_TOKEN']; @@ -7123,33 +7137,64 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); - it('rejects direct mutation to the reserved standalone source', async () => { - const sessionId = 'session-A'; - const recording = { - recordSessionSource: vi.fn().mockResolvedValue(true), - }; - const innerConfig = await setupSessionMocks(sessionId); - innerConfig.getChatRecordingService = vi.fn().mockReturnValue(recording); - const { agent, agentPromise } = await bootAcpAgent(); + it.each(['standalone', MESH_HOST_SESSION_SOURCE_TYPE])( + 'rejects direct mutation to the reserved %s source', + async (sourceType) => { + const sessionId = 'session-A'; + const recording = { + recordSessionSource: vi.fn().mockResolvedValue(true), + }; + const innerConfig = await setupSessionMocks(sessionId); + innerConfig.getChatRecordingService = vi.fn().mockReturnValue(recording); + const { agent, agentPromise } = await bootAcpAgent(); + + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionSource, { + sessionId, + sourceType, + }), + ).rejects.toThrow( + sourceType === 'standalone' + ? '`standalone` is reserved for daemon-owned session creation' + : '`mesh` is reserved for daemon-owned host creation', + ); + + expect(recording.recordSessionSource).not.toHaveBeenCalled(); + expect(lastSessionMock?.enableLiveScreenContext).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }, + ); + + it('rejects forged daemon-owned standalone creation from an untrusted parent', async () => { + await setupSessionMocks('11111111-1111-4111-8111-111111111111'); + const { agent, agentPromise } = await bootInitializedAcpAgent( + makeSessionSettings(), + ); - await agent.newSession({ cwd: '/tmp', mcpServers: [] }); await expect( - agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionSource, { - sessionId, - sourceType: 'standalone', + agent.newSession({ + cwd: '/tmp', + mcpServers: [], + _meta: { + [SESSION_SOURCE_META_KEY]: { + sourceType: 'standalone', + [DAEMON_OWNED_STANDALONE_CREATION_KEY]: true, + }, + }, }), ).rejects.toThrow( '`standalone` is reserved for daemon-owned session creation', ); - - expect(recording.recordSessionSource).not.toHaveBeenCalled(); - expect(lastSessionMock?.enableLiveScreenContext).not.toHaveBeenCalled(); + expect(loadCliConfig).not.toHaveBeenCalled(); mockConnectionState.resolve(); await agentPromise; }); - it('rejects forged daemon-owned standalone creation from an untrusted parent', async () => { + it('rejects forged mesh host creation from an untrusted parent', async () => { await setupSessionMocks('11111111-1111-4111-8111-111111111111'); const { agent, agentPromise } = await bootInitializedAcpAgent( makeSessionSettings(), @@ -7161,14 +7206,11 @@ describe('QwenAgent MCP SSE/HTTP support', () => { mcpServers: [], _meta: { [SESSION_SOURCE_META_KEY]: { - sourceType: 'standalone', - [DAEMON_OWNED_STANDALONE_CREATION_KEY]: true, + sourceType: MESH_HOST_SESSION_SOURCE_TYPE, }, }, }), - ).rejects.toThrow( - '`standalone` is reserved for daemon-owned session creation', - ); + ).rejects.toThrow('`mesh` is reserved for daemon-owned host creation'); expect(loadCliConfig).not.toHaveBeenCalled(); mockConnectionState.resolve(); @@ -7219,6 +7261,102 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('launches a configured agent only from a trusted mesh host session', async () => { + const sessionId = '11111111-1111-4111-8111-111111111111'; + const meshAgent = { id: 'ag_alice', name: 'alice', createdAt: 1 }; + const innerConfig = await setupSessionMocks(sessionId); + innerConfig.getSessionSourceType = vi + .fn() + .mockReturnValue(MESH_HOST_SESSION_SOURCE_TYPE); + innerConfig.getProjectRoot = vi.fn().mockReturnValue('/tmp'); + mockReadMeshWorkspace.mockResolvedValue({ hostSessionId: sessionId }); + mockReadMeshAgents.mockResolvedValue([meshAgent]); + mockLaunchMeshAgent.mockResolvedValue({ + status: 'started', + runtimeId: 'local:mesh-ag_alice', + backgroundAgentId: 'mesh-ag_alice', + sessionId, + }); + const { agent, agentPromise } = await bootInitializedAcpAgent( + makeSessionSettings(), + 'trusted-capability', + ); + + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch, { + sessionId, + agentId: meshAgent.id, + prompt: 'go', + }), + ).resolves.toMatchObject({ status: 'started', sessionId }); + expect(mockReadMeshWorkspace).toHaveBeenCalledWith('/tmp'); + expect(mockReadMeshAgents).toHaveBeenCalledWith('/tmp'); + expect(mockLaunchMeshAgent).toHaveBeenCalledWith( + innerConfig, + meshAgent, + 'go', + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('rejects a labelled session that is not the claimed mesh host', async () => { + const sessionId = '11111111-1111-4111-8111-111111111111'; + const innerConfig = await setupSessionMocks(sessionId); + innerConfig.getSessionSourceType = vi + .fn() + .mockReturnValue(MESH_HOST_SESSION_SOURCE_TYPE); + innerConfig.getProjectRoot = vi.fn().mockReturnValue('/tmp'); + mockReadMeshWorkspace.mockResolvedValue({ hostSessionId: 'other-host' }); + const { agent, agentPromise } = await bootInitializedAcpAgent( + makeSessionSettings(), + 'trusted-capability', + ); + + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch, { + sessionId, + agentId: 'ag_alice', + prompt: 'go', + }), + ).rejects.toThrow(/claimed mesh host/); + expect(mockReadMeshAgents).not.toHaveBeenCalled(); + expect(mockLaunchMeshAgent).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('returns launch_failed when the mesh store cannot be read', async () => { + const sessionId = '11111111-1111-4111-8111-111111111111'; + const innerConfig = await setupSessionMocks(sessionId); + innerConfig.getSessionSourceType = vi + .fn() + .mockReturnValue(MESH_HOST_SESSION_SOURCE_TYPE); + innerConfig.getProjectRoot = vi.fn().mockReturnValue('/tmp'); + mockReadMeshWorkspace.mockRejectedValue(new Error('mesh store busy')); + const { agent, agentPromise } = await bootInitializedAcpAgent( + makeSessionSettings(), + 'trusted-capability', + ); + + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch, { + sessionId, + agentId: 'ag_alice', + prompt: 'go', + }), + ).resolves.toEqual({ status: 'launch_failed', error: 'mesh store busy' }); + expect(mockLaunchMeshAgent).not.toHaveBeenCalled(); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('defers standalone new-session workspace setup until managed activation', async () => { const sessionId = '11111111-1111-4111-8111-111111111111'; const innerConfig = await setupSessionMocks(sessionId); @@ -22494,6 +22632,7 @@ describe('QwenAgent unstable_listSessions cursor parsing', () => { expect(listSessions).toHaveBeenCalledWith({ cursor: undefined, size: undefined, + excludeSourceType: 'mesh', }); } } finally { @@ -22537,6 +22676,7 @@ describe('QwenAgent unstable_listSessions cursor parsing', () => { expect(listSessions).toHaveBeenCalledWith({ cursor: undefined, size: undefined, + excludeSourceType: 'mesh', }); } } finally { @@ -22577,6 +22717,7 @@ describe('QwenAgent unstable_listSessions cursor parsing', () => { expect(listSessions).toHaveBeenCalledWith({ cursor: undefined, size: expected, + excludeSourceType: 'mesh', }); } } finally { @@ -22633,6 +22774,7 @@ describe('QwenAgent unstable_listSessions cursor parsing', () => { expect(listSessions).toHaveBeenCalledWith({ cursor: 1_797_860_000_000.5, size: 2, + excludeSourceType: 'mesh', }); } finally { mockConnectionState.resolve(); @@ -23175,9 +23317,14 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { }, ); - it.each(['load', 'resume'] as const)( - '%s rejects a standalone restore without a trusted daemon parent', - async (action) => { + it.each([ + ['load', 'standalone'], + ['resume', 'standalone'], + ['load', MESH_HOST_SESSION_SOURCE_TYPE], + ['resume', MESH_HOST_SESSION_SOURCE_TYPE], + ] as const)( + '%s rejects a %s restore without a trusted daemon parent', + async (action, sourceType) => { bindRestoreMocks({ sessionExists: true }); const { agent, agentPromise } = await spawnAgent(); @@ -23188,8 +23335,10 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { mcpServers: [], _meta: { [SESSION_SOURCE_META_KEY]: { - sourceType: 'standalone', - [DAEMON_OWNED_STANDALONE_CREATION_KEY]: true, + sourceType, + ...(sourceType === 'standalone' + ? { [DAEMON_OWNED_STANDALONE_CREATION_KEY]: true } + : {}), }, }, }; @@ -23199,7 +23348,9 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { ? agent.loadSession(request) : agent.unstable_resumeSession(request), ).rejects.toThrow( - '`standalone` is reserved for daemon-owned session restore', + sourceType === 'standalone' + ? '`standalone` is reserved for daemon-owned session restore' + : '`mesh` is reserved for daemon-owned host restore', ); } finally { mockConnectionState.resolve(); diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 410c68a12fc..31add00aac1 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -147,6 +147,11 @@ import { listWorkflowSnapshots, type TurnResultRecordPayload, sessionIdContext, + launchMeshAgent, + readMeshAgents, + readMeshWorkspace, + type MeshAgent, + type MeshAgentLaunchResult, } from '@qwen-code/qwen-code-core'; import { randomUUID, timingSafeEqual } from 'node:crypto'; import { performance } from 'node:perf_hooks'; @@ -401,6 +406,7 @@ import { parseSessionSource, SESSION_SOURCE_META_KEY, } from '@qwen-code/acp-bridge/sessionSource'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../runtime/mesh-session-source.js'; import { ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM, ACTIVE_WORK_HEARTBEAT_META_KEY, @@ -4991,6 +4997,15 @@ class QwenAgent implements Agent { ); initializationDeadline?.signal.throwIfAborted(); const sessionSource = getSessionSource(params); + if ( + sessionSource?.sourceType === MESH_HOST_SESSION_SOURCE_TYPE && + !this.isTrustedManagedParent() + ) { + throw RequestError.invalidParams( + undefined, + '`mesh` is reserved for daemon-owned host creation', + ); + } const provisionalStandalone = isReservedStandaloneSessionSourceType( sessionSource?.sourceType, ); @@ -5116,6 +5131,15 @@ class QwenAgent implements Agent { ): Promise { let sessionId = initialSessionId; const sessionSource = getSessionSource(params); + if ( + sessionSource?.sourceType === MESH_HOST_SESSION_SOURCE_TYPE && + !this.isTrustedManagedParent() + ) { + throw RequestError.invalidParams( + undefined, + '`mesh` is reserved for daemon-owned host restore', + ); + } const provisionalStandalone = isReservedStandaloneSessionSourceType( sessionSource?.sourceType, ); @@ -5595,6 +5619,15 @@ class QwenAgent implements Agent { ): Promise { let sessionId = initialSessionId; const sessionSource = getSessionSource(params); + if ( + sessionSource?.sourceType === MESH_HOST_SESSION_SOURCE_TYPE && + !this.isTrustedManagedParent() + ) { + throw RequestError.invalidParams( + undefined, + '`mesh` is reserved for daemon-owned host restore', + ); + } const provisionalStandalone = isReservedStandaloneSessionSourceType( sessionSource?.sourceType, ); @@ -5843,6 +5876,7 @@ class QwenAgent implements Agent { return sessionService.listSessions({ cursor: numericCursor, size, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); }); @@ -8459,12 +8493,13 @@ class QwenAgent implements Agent { } : params; if ( - method === SERVE_CONTROL_EXT_METHODS.sessionBackgroundNotification && + (method === SERVE_CONTROL_EXT_METHODS.sessionBackgroundNotification || + method === SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch) && this.privateParentState !== 'trusted' ) { throw RequestError.invalidParams( undefined, - 'Background notifications require a trusted private ACP parent', + 'This operation requires a trusted private ACP parent', ); } const sessionId = normalizedParams['sessionId']; @@ -10409,6 +10444,17 @@ class QwenAgent implements Agent { } } const session = this.sessionOrThrow(sessionId); + if ( + source.sourceType === MESH_HOST_SESSION_SOURCE_TYPE && + (!this.isTrustedManagedParent() || + session.getConfig().getSessionSourceType() !== + MESH_HOST_SESSION_SOURCE_TYPE) + ) { + throw RequestError.invalidParams( + undefined, + '`mesh` is reserved for daemon-owned host creation', + ); + } if (isCompatibleLiveSessionSource(source)) { await session.enableLiveScreenContext(); } @@ -12211,6 +12257,65 @@ class QwenAgent implements Agent { ); return result; } + case SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch: { + const sessionId = params['sessionId']; + const agentId = params['agentId']; + const prompt = params['prompt']; + if ( + typeof sessionId !== 'string' || + sessionId.length === 0 || + typeof agentId !== 'string' || + agentId.length === 0 || + typeof prompt !== 'string' || + prompt.length === 0 + ) { + throw RequestError.invalidParams( + undefined, + 'Invalid mesh agent launch request', + ); + } + const session = this.sessionOrThrow(sessionId); + const config = session.getConfig(); + if (config.getSessionSourceType() !== MESH_HOST_SESSION_SOURCE_TYPE) { + throw RequestError.invalidParams( + undefined, + 'Mesh agents require a mesh host session', + ); + } + const projectRoot = config.getProjectRoot(); + let workspace: Awaited>; + try { + workspace = await readMeshWorkspace(projectRoot); + } catch (error) { + return { + status: 'launch_failed', + error: error instanceof Error ? error.message : String(error), + } satisfies MeshAgentLaunchResult; + } + if (workspace.hostSessionId !== config.getSessionId()) { + throw RequestError.invalidParams( + undefined, + 'Mesh agents require the claimed mesh host session', + ); + } + let agents: MeshAgent[]; + try { + agents = await readMeshAgents(projectRoot); + } catch (error) { + return { + status: 'launch_failed', + error: error instanceof Error ? error.message : String(error), + } satisfies MeshAgentLaunchResult; + } + const agent = agents.find((candidate) => candidate.id === agentId); + if (!agent) { + return { + status: 'agent_unavailable', + error: `Mesh agent "${agentId}" is unavailable.`, + } satisfies MeshAgentLaunchResult; + } + return launchMeshAgent(config, agent, prompt); + } case SERVE_CONTROL_EXT_METHODS.workspaceMcpRuntimeAdd: { const request = readRuntimeMcpAddRequest(params); const result = await addRuntimeMcpServer( diff --git a/packages/cli/src/commands/sessions/list.test.ts b/packages/cli/src/commands/sessions/list.test.ts index 3f5f482ea2c..3be69a8e4c6 100644 --- a/packages/cli/src/commands/sessions/list.test.ts +++ b/packages/cli/src/commands/sessions/list.test.ts @@ -190,6 +190,7 @@ describe('sessions list command', () => { expect(mockListSessions).toHaveBeenCalledWith({ size: 10, + excludeSourceType: 'mesh', }); }); @@ -200,6 +201,7 @@ describe('sessions list command', () => { expect(mockListSessions).toHaveBeenCalledWith({ size: 20, + excludeSourceType: 'mesh', }); }); diff --git a/packages/cli/src/commands/sessions/list.ts b/packages/cli/src/commands/sessions/list.ts index 3bb2f83be42..49e2022cf02 100644 --- a/packages/cli/src/commands/sessions/list.ts +++ b/packages/cli/src/commands/sessions/list.ts @@ -11,6 +11,7 @@ import type { ListSessionsResult, } from '@qwen-code/qwen-code-core'; import stringWidth from 'string-width'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import { escapeAnsiCtrlCodes } from '../../ui/utils/textUtils.js'; import { initSessionService } from './common.js'; import { writeStdoutLine, writeStderrLine } from '../../utils/stdioHelpers.js'; @@ -173,6 +174,7 @@ export async function handleList(argv: ListArgs): Promise { try { result = await svc.listSessions({ size: argv.limit ?? 20, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); } catch (err) { writeStderrLine(`Error: failed to list sessions: ${formatError(err)}`); diff --git a/packages/cli/src/config/config.ts b/packages/cli/src/config/config.ts index d1e65a738b4..660cb5adc28 100755 --- a/packages/cli/src/config/config.ts +++ b/packages/cli/src/config/config.ts @@ -52,6 +52,7 @@ import { stripAnsiAndControl, type OutputStyleDefinition, } from '@qwen-code/qwen-code-core'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../runtime/mesh-session-source.js'; import { extensionsCommand } from '../commands/extensions.js'; import { hooksCommand } from '../commands/hooks.js'; import { resolveAcpChannelFallback } from './acp-channel-fallback.js'; @@ -1964,7 +1965,9 @@ export async function loadCliConfig( if (argv.continue || argv.resume) { const sessionService = new SessionService(cwd); if (argv.continue) { - sessionData = await sessionService.loadLastSession(); + sessionData = await sessionService.loadLastSession({ + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, + }); if (sessionData) { sessionId = sessionData.conversation.sessionId; } else if (argv.forkSession) { diff --git a/packages/cli/src/runtime/mesh-session-source.ts b/packages/cli/src/runtime/mesh-session-source.ts new file mode 100644 index 00000000000..d08d962fc86 --- /dev/null +++ b/packages/cli/src/runtime/mesh-session-source.ts @@ -0,0 +1,7 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +export const MESH_HOST_SESSION_SOURCE_TYPE = 'mesh'; diff --git a/packages/cli/src/serve/live/realtime-startup-context.ts b/packages/cli/src/serve/live/realtime-startup-context.ts index 81c57ee941e..34bfd66ad7c 100644 --- a/packages/cli/src/serve/live/realtime-startup-context.ts +++ b/packages/cli/src/serve/live/realtime-startup-context.ts @@ -14,6 +14,7 @@ import { type ChatRecord, type SessionListItem, } from '@qwen-code/qwen-code-core'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import type { WorkspaceRegistry, WorkspaceRuntime, @@ -214,6 +215,7 @@ async function loadRecentThreads( ).listSessions({ size: MAX_RECENT_THREADS, archiveState: 'active', + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); return page.items.map(recentThread); }), diff --git a/packages/cli/src/serve/mesh/mesh-host-session.test.ts b/packages/cli/src/serve/mesh/mesh-host-session.test.ts new file mode 100644 index 00000000000..4c9dbcfa54c --- /dev/null +++ b/packages/cli/src/serve/mesh/mesh-host-session.test.ts @@ -0,0 +1,395 @@ +/** + * @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 { + readMeshWorkspace, + Storage, + type MeshAgent, +} from '@qwen-code/qwen-code-core'; +import { + makeBridge, + makeChannel, +} from '@qwen-code/acp-bridge/internal/testUtils'; +import { SERVE_CONTROL_EXT_METHODS } from '@qwen-code/acp-bridge/status'; +import type { AcpSessionBridge } from '../acp-session-bridge.js'; +import { + getWorkspaceSessionInfoForResponse, + listLiveWorkspaceSessionsForResponse, + listWorkspaceSessionsForResponse, + searchWorkspaceSessionsForResponse, +} from '../server/session-list.js'; +import { startMeshHostSessionOwner } from './mesh-host-session.js'; + +const AGENT: MeshAgent = { id: 'ag_alice', name: 'alice', createdAt: 1 }; + +async function writeStoredSession( + workspace: string, + sessionId: string, + sourceType: string, + mtime: Date, +): Promise { + const chatsDir = path.join(new Storage(workspace).getProjectDir(), 'chats'); + await fs.mkdir(chatsDir, { recursive: true }); + const filePath = path.join(chatsDir, `${sessionId}.jsonl`); + const records = [ + { + uuid: `${sessionId}-user`, + parentUuid: null, + sessionId, + timestamp: mtime.toISOString(), + type: 'user', + message: { role: 'user', parts: [{ text: sessionId }] }, + cwd: workspace, + }, + { + uuid: `${sessionId}-source`, + parentUuid: `${sessionId}-user`, + sessionId, + timestamp: mtime.toISOString(), + type: 'system', + subtype: 'session_source', + systemPayload: { sourceType }, + cwd: workspace, + }, + ]; + await fs.writeFile( + filePath, + `${records.map((record) => JSON.stringify(record)).join('\n')}\n`, + ); + await fs.utimes(filePath, mtime, mtime); +} + +describe('mesh host session owner', () => { + let scratch: string; + let workspace: string; + + beforeEach(async () => { + scratch = await fs.mkdtemp(path.join(os.tmpdir(), 'mesh-host-test-')); + workspace = path.join(scratch, 'workspace'); + await fs.mkdir(workspace); + Storage.setRuntimeBaseDir(scratch); + }); + + afterEach(async () => { + Storage.setRuntimeBaseDir(null); + await fs.rm(scratch, { recursive: true, force: true }); + }); + + it('owns one hidden session and revives it before the next launch', async () => { + let resident = false; + let spawnCount = 0; + const resumes: string[] = []; + const launches: string[] = []; + const bridge = { + recordHeartbeat: (sessionId: string) => { + if (!resident) throw new Error(`${sessionId} was reaped`); + }, + spawnOrAttach: vi.fn(async () => { + spawnCount++; + resident = true; + return { sessionId: `mesh-host-${spawnCount}` }; + }), + resumeSession: vi.fn(async ({ sessionId }: { sessionId: string }) => { + resumes.push(sessionId); + resident = true; + return { sessionId }; + }), + closeSession: vi.fn(async () => {}), + launchMeshAgent: vi.fn( + async (sessionId: string, agentId: string, prompt: string) => { + launches.push(`${sessionId}:${agentId}:${prompt}`); + return { + status: 'started' as const, + runtimeId: `local:mesh-${agentId}`, + backgroundAgentId: `mesh-${agentId}`, + sessionId, + }; + }, + ), + }; + const owner = startMeshHostSessionOwner({ + bridge, + workspaceCwd: workspace, + intervalMs: 60_000, + resumeTimeoutMs: 5_000, + }); + + await expect(owner.launch(AGENT, 'first')).resolves.toMatchObject({ + status: 'started', + sessionId: 'mesh-host-1', + }); + resident = false; + const startedAt = performance.now(); + await owner.tick(); + const reloadMs = performance.now() - startedAt; + await expect(owner.launch(AGENT, 'second')).resolves.toMatchObject({ + status: 'started', + sessionId: 'mesh-host-1', + }); + owner.stop(); + + expect(spawnCount).toBe(1); + expect(resumes).toEqual(['mesh-host-1']); + const meshWorkspace = await readMeshWorkspace(workspace); + expect(bridge.spawnOrAttach).toHaveBeenCalledWith( + expect.objectContaining({ + workspaceCwd: workspace, + sessionScope: 'thread', + sourceType: 'mesh', + sourceId: meshWorkspace.workspaceId, + }), + ); + expect(launches).toEqual([ + 'mesh-host-1:ag_alice:first', + 'mesh-host-1:ag_alice:second', + ]); + expect(reloadMs).toBeLessThan(1_000); + }); + + it('closes the losing host when two owners start concurrently', async () => { + let releaseSpawns: (() => void) | undefined; + const bothSpawned = new Promise((resolve) => { + releaseSpawns = resolve; + }); + let spawnCount = 0; + const bridge = { + recordHeartbeat: vi.fn(), + spawnOrAttach: vi.fn(async () => { + const sessionId = `mesh-host-${++spawnCount}`; + if (spawnCount === 2) releaseSpawns?.(); + await bothSpawned; + return { sessionId }; + }), + resumeSession: vi.fn(), + closeSession: vi.fn(async () => {}), + launchMeshAgent: vi.fn(), + }; + const owners = [ + startMeshHostSessionOwner({ + bridge, + workspaceCwd: workspace, + intervalMs: 60_000, + }), + startMeshHostSessionOwner({ + bridge, + workspaceCwd: workspace, + intervalMs: 60_000, + }), + ]; + try { + const sessionIds = await Promise.all( + owners.map((owner) => owner.ensureResident()), + ); + expect(new Set(sessionIds).size).toBe(1); + expect(bridge.spawnOrAttach).toHaveBeenCalledTimes(2); + expect(bridge.closeSession).toHaveBeenCalledOnce(); + expect(bridge.closeSession).not.toHaveBeenCalledWith(sessionIds[0]); + } finally { + owners.forEach((owner) => owner.stop()); + } + }); + + it('reloads after the daemon bridge reaper closes the host', async () => { + const handles: Array> = []; + const bridge = makeBridge({ + boundWorkspace: workspace, + sessionReapIntervalMs: 10, + sessionIdleTimeoutMs: 20, + channelFactory: async () => { + const handle = makeChannel({ + extMethodImpl: (method, params) => + method === SERVE_CONTROL_EXT_METHODS.sessionMeshAgentLaunch + ? { + status: 'started', + runtimeId: `local:mesh-${String(params['agentId'])}`, + backgroundAgentId: `mesh-${String(params['agentId'])}`, + sessionId: String(params['sessionId']), + } + : method === SERVE_CONTROL_EXT_METHODS.sessionClose + ? { closed: true, holds: [] } + : {}, + }); + handles.push(handle); + return handle.channel; + }, + }); + const owner = startMeshHostSessionOwner({ + bridge, + workspaceCwd: workspace, + intervalMs: 60_000, + resumeTimeoutMs: 1_000, + }); + try { + const sessionId = await owner.ensureResident(); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0), { + timeout: 2_000, + }); + const startedAt = performance.now(); + await expect(owner.launch(AGENT, 'after reaper')).resolves.toMatchObject({ + status: 'started', + sessionId, + }); + const reloadMs = performance.now() - startedAt; + + expect(handles).toHaveLength(2); + expect(handles[0]?.killed).toBe(true); + expect(bridge.listWorkspaceSessions(workspace)).toEqual([ + expect.objectContaining({ sessionId, sourceType: 'mesh' }), + ]); + expect(reloadMs).toBeLessThan(1_000); + } finally { + owner.stop(); + await bridge.shutdown(); + } + }); + + it('is excluded from the unfiltered session catalog', async () => { + const bridge = { + listWorkspaceSessions: () => [ + { + sessionId: 'default-session', + cwd: workspace, + sourceType: 'default', + }, + { sessionId: 'mesh-host', cwd: workspace, sourceType: 'mesh' }, + ], + } as unknown as AcpSessionBridge; + + const result = await listWorkspaceSessionsForResponse( + bridge, + workspace, + undefined, + { runtimeBaseDir: scratch }, + ); + expect(result.sessions.map((session) => session.sessionId)).toEqual([ + 'default-session', + ]); + const liveResult = await listLiveWorkspaceSessionsForResponse( + bridge, + workspace, + undefined, + { runtimeBaseDir: scratch }, + ); + expect(liveResult.sessions.map((session) => session.sessionId)).toEqual([ + 'default-session', + ]); + await expect( + getWorkspaceSessionInfoForResponse(bridge, workspace), + ).resolves.toMatchObject({ active: 0, total: 0, live: 1 }); + }); + + it('filters persisted hosts before paginating public catalogs', async () => { + const visibleId = '00000000-0000-4000-8000-000000000001'; + const hiddenId = '00000000-0000-4000-8000-000000000002'; + await writeStoredSession( + workspace, + visibleId, + 'default', + new Date('2026-09-06T00:00:00.000Z'), + ); + await writeStoredSession( + workspace, + hiddenId, + 'mesh', + new Date('2026-09-06T00:01:00.000Z'), + ); + const bridge = { + listWorkspaceSessions: () => [], + } as unknown as AcpSessionBridge; + + const page = await listWorkspaceSessionsForResponse( + bridge, + workspace, + { size: 1 }, + { runtimeBaseDir: scratch, mergeLive: false }, + ); + expect(page.sessions.map((session) => session.sessionId)).toEqual([ + visibleId, + ]); + expect(page.nextCursor).toBeUndefined(); + + const organized = await listWorkspaceSessionsForResponse( + bridge, + workspace, + { size: 1, view: 'organized' }, + { runtimeBaseDir: scratch, mergeLive: false }, + ); + expect(organized.sessions.map((session) => session.sessionId)).toEqual([ + visibleId, + ]); + + const explicitMesh = await listWorkspaceSessionsForResponse( + bridge, + workspace, + { sourceType: 'mesh' }, + { runtimeBaseDir: scratch, mergeLive: false }, + ); + expect(explicitMesh.sessions).toEqual([]); + + await expect( + searchWorkspaceSessionsForResponse( + workspace, + hiddenId, + {}, + { + runtimeBaseDir: scratch, + }, + ), + ).resolves.toEqual({ results: [] }); + + await expect( + getWorkspaceSessionInfoForResponse(bridge, workspace), + ).resolves.toMatchObject({ active: 1, archived: 0, total: 1, live: 0 }); + }); + + it('does not duplicate a resume that outlives its deadline', async () => { + const first = startMeshHostSessionOwner({ + bridge: { + recordHeartbeat: () => {}, + spawnOrAttach: async () => ({ sessionId: 'mesh-host' }), + resumeSession: async () => ({ sessionId: 'mesh-host' }), + closeSession: async () => {}, + launchMeshAgent: async () => ({ status: 'capacity_wait' as const }), + }, + workspaceCwd: workspace, + intervalMs: 60_000, + }); + await first.ensureResident(); + first.stop(); + + let settleResume: (() => void) | undefined; + const resumeSession = vi.fn( + () => + new Promise((resolve) => { + settleResume = resolve; + }), + ); + const owner = startMeshHostSessionOwner({ + bridge: { + recordHeartbeat: () => { + throw new Error('reaped'); + }, + spawnOrAttach: async () => ({ sessionId: 'unexpected' }), + resumeSession, + closeSession: async () => {}, + launchMeshAgent: async () => ({ status: 'capacity_wait' as const }), + }, + workspaceCwd: workspace, + intervalMs: 60_000, + resumeTimeoutMs: 5, + }); + + await expect(owner.tick()).rejects.toThrow(/timed out/); + await expect(owner.tick()).rejects.toThrow(/timed out/); + expect(resumeSession).toHaveBeenCalledOnce(); + settleResume?.(); + owner.stop(); + }); +}); diff --git a/packages/cli/src/serve/mesh/mesh-host-session.ts b/packages/cli/src/serve/mesh/mesh-host-session.ts new file mode 100644 index 00000000000..69e91a29bf8 --- /dev/null +++ b/packages/cli/src/serve/mesh/mesh-host-session.ts @@ -0,0 +1,141 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + claimMeshHostSession, + readMeshWorkspace, + type MeshAgent, + type MeshAgentLaunchResult, +} from '@qwen-code/qwen-code-core'; +import type { AcpSessionBridge } from '../acp-session-bridge.js'; +import { beginKeepaliveSessionResume } from '../scheduled-task-keepalive.js'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; + +const DEFAULT_MESH_KEEPALIVE_INTERVAL_MS = 30_000; +const DEFAULT_MESH_RESUME_TIMEOUT_MS = 70_000; + +interface MeshHostBridge { + recordHeartbeat(sessionId: string): unknown; + resumeSession( + request: Parameters[0], + ): Promise; + spawnOrAttach( + request: Parameters[0], + ): Promise<{ sessionId: string }>; + closeSession(sessionId: string): Promise; + launchMeshAgent: AcpSessionBridge['launchMeshAgent']; +} + +export interface MeshHostSessionOwner { + ensureResident(): Promise; + launch(agent: MeshAgent, prompt: string): Promise; + tick(): Promise; + stop(): void; +} + +export function startMeshHostSessionOwner(options: { + bridge: MeshHostBridge; + workspaceCwd: string; + intervalMs?: number; + resumeTimeoutMs?: number; +}): MeshHostSessionOwner { + const { bridge, workspaceCwd } = options; + const intervalMs = options.intervalMs ?? DEFAULT_MESH_KEEPALIVE_INTERVAL_MS; + const resumeTimeoutMs = + options.resumeTimeoutMs ?? DEFAULT_MESH_RESUME_TIMEOUT_MS; + let ensuring: Promise | undefined; + let reviving: + | { completion: Promise; deadline: Promise } + | undefined; + + const ensure = async (): Promise => { + const workspace = await readMeshWorkspace(workspaceCwd); + if (workspace.hostSessionId) { + try { + bridge.recordHeartbeat(workspace.hostSessionId); + } catch { + if (!reviving) { + const started = beginKeepaliveSessionResume( + bridge, + { + sessionId: workspace.hostSessionId, + workspaceCwd, + sourceType: MESH_HOST_SESSION_SOURCE_TYPE, + sourceId: workspace.workspaceId, + }, + resumeTimeoutMs, + ); + reviving = started; + void started.completion + .finally(() => { + if (reviving === started) reviving = undefined; + }) + .catch(() => {}); + } + await reviving.deadline; + } + return workspace.hostSessionId; + } + + const spawned = await bridge.spawnOrAttach({ + workspaceCwd, + sessionScope: 'thread', + sourceType: MESH_HOST_SESSION_SOURCE_TYPE, + sourceId: workspace.workspaceId, + }); + let winner: string; + try { + winner = await claimMeshHostSession(workspaceCwd, spawned.sessionId); + } catch (error) { + await bridge.closeSession(spawned.sessionId).catch(() => {}); + throw error; + } + if (winner !== spawned.sessionId) { + await bridge.closeSession(spawned.sessionId).catch(() => {}); + return ensure(); + } + return winner; + }; + + const ensureResident = (): Promise => { + ensuring ??= ensure().finally(() => { + ensuring = undefined; + }); + return ensuring; + }; + + const tick = async (): Promise => { + const workspace = await readMeshWorkspace(workspaceCwd); + if (workspace.hostSessionId) await ensureResident(); + }; + + let running = false; + const timer = setInterval(() => { + if (running) return; + running = true; + void tick() + .catch(() => {}) + .finally(() => { + running = false; + }); + }, intervalMs); + timer.unref?.(); + + let stopped = false; + return { + ensureResident, + async launch(agent, prompt) { + const sessionId = await ensureResident(); + return bridge.launchMeshAgent(sessionId, agent.id, prompt); + }, + tick, + stop() { + if (stopped) return; + stopped = true; + clearInterval(timer); + }, + }; +} diff --git a/packages/cli/src/serve/multi-workspace-sessions.test.ts b/packages/cli/src/serve/multi-workspace-sessions.test.ts index fbc4b5cbfee..2ffb67466d8 100644 --- a/packages/cli/src/serve/multi-workspace-sessions.test.ts +++ b/packages/cli/src/serve/multi-workspace-sessions.test.ts @@ -6964,6 +6964,32 @@ describe('workspace session live-state route', () => { ]); }); + it('omits the hidden mesh host without changing the catalog version', async () => { + const { app } = makeHarness({ + primarySummaries: [ + makeSummary('11111111-1111-4111-a111-111111111111', PRIMARY_CWD), + makeSummary('22222222-2222-4222-a222-222222222222', PRIMARY_CWD, { + sourceType: 'mesh', + }), + ], + }); + + const res = await request(app) + .get(liveStatePath('primary-id')) + .set('Host', host()) + .expect(200); + + expect(res.body.catalogVersion).toEqual({ + generation: expect.any(String), + revision: expect.any(Number), + }); + expect( + res.body.sessions.map( + (session: { sessionId: string }) => session.sessionId, + ), + ).toEqual(['11111111-1111-4111-a111-111111111111']); + }); + it('omits updatedAt when the bridge summary has no activity watermark', async () => { // A live entry that has not settled a running turn in this bridge (fresh // spawn, restore) legitimately carries no watermark. The key must be diff --git a/packages/cli/src/serve/routes/goals.test.ts b/packages/cli/src/serve/routes/goals.test.ts index de38198f8d9..e3299c6ed88 100644 --- a/packages/cli/src/serve/routes/goals.test.ts +++ b/packages/cli/src/serve/routes/goals.test.ts @@ -106,6 +106,24 @@ describe('GET /goals', () => { expect(res.body).toEqual({ v: 1, goals: [], droppedCount: 0 }); }); + it('does not probe or expose the hidden mesh host', async () => { + const getSessionGoal = vi.fn(async () => noGoal); + const app = makeApp({ + listWorkspaceSessions: () => [ + summary('visible'), + summary('mesh-host', { sourceType: 'mesh' }), + ], + getSessionGoal, + }); + + const res = await request(app).get('/goals'); + + expect(res.status).toBe(200); + expect(getSessionGoal).toHaveBeenCalledOnce(); + expect(getSessionGoal).toHaveBeenCalledWith('visible'); + expect(res.body).toEqual({ v: 1, goals: [], droppedCount: 0 }); + }); + it('rejects reads when the live primary workspace is untrusted', async () => { const listWorkspaceSessions = vi.fn(() => []); const app = makeApp( diff --git a/packages/cli/src/serve/routes/goals.ts b/packages/cli/src/serve/routes/goals.ts index 9b248ddd2a6..0bd29a19332 100644 --- a/packages/cli/src/serve/routes/goals.ts +++ b/packages/cli/src/serve/routes/goals.ts @@ -27,6 +27,7 @@ import type { BridgeSessionGoal, BridgeSessionSummary, } from '@qwen-code/acp-bridge'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import { writeStderrLine } from '../../utils/stdioHelpers.js'; import { sendGenerationClosedError, @@ -118,7 +119,11 @@ export function registerGoalsRoutes( const assertGenerationOpen = deps.captureGenerationAssertion?.(); try { assertGenerationOpen?.(); - const sessions = bridge.listWorkspaceSessions(boundWorkspace); + const sessions = bridge + .listWorkspaceSessions(boundWorkspace) + .filter( + (session) => session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE, + ); const settled = await allSettledWithLimit( sessions, PROBE_CONCURRENCY, diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index a5f205b0dc0..c46c61d4b00 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -138,6 +138,7 @@ import { withPromptTerminals, } from '../prompt-terminal-ledger.js'; import { createSessionOrganizationService } from '../session-organization-helpers.js'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import { omitSkillDetailsForSdkSurface, omitSkillDetailsFromReplayArrays, @@ -734,6 +735,13 @@ function parseRequestedSessionSource( body: Record, res: Response, ): { sourceType?: string; sourceId?: string } | null { + if (body['sourceType'] === MESH_HOST_SESSION_SOURCE_TYPE) { + res.status(400).json({ + error: 'The requested session source is reserved for mesh hosts.', + code: 'reserved_session_source', + }); + return null; + } if ( isReservedStandaloneSessionSource({ sourceType: @@ -1483,6 +1491,7 @@ export function registerSessionRoutes( archiveState: 'active', size: 1, signal, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); signal.throwIfAborted(); return page.items.length > 0; @@ -3013,7 +3022,12 @@ export function registerSessionRoutes( // by a "new chat" does not block a fresh branch session. const sharedCheckoutSession = runtime.bridge .listWorkspaceSessions(workspaceCwd) - .find((session) => !session.worktree && session.clientCount > 0); + .find( + (session) => + session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE && + !session.worktree && + session.clientCount > 0, + ); if (sharedCheckoutSession) { res.status(409).json({ error: @@ -7711,6 +7725,9 @@ export function registerSessionRoutes( } const sessions = bridge .listWorkspaceSessions(runtime.workspaceCwd) + .filter( + (session) => session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE, + ) .map((session) => ({ sessionId: session.sessionId, clientCount: session.clientCount, diff --git a/packages/cli/src/serve/scheduled-task-keepalive.ts b/packages/cli/src/serve/scheduled-task-keepalive.ts index cf2fe18a012..38f59a19da2 100644 --- a/packages/cli/src/serve/scheduled-task-keepalive.ts +++ b/packages/cli/src/serve/scheduled-task-keepalive.ts @@ -107,6 +107,29 @@ export interface KeepaliveBridge { ): unknown; } +export interface KeepaliveSessionResumeRequest { + sessionId: string; + workspaceCwd: string; + sourceType?: string; + sourceId?: string; +} + +export function beginKeepaliveSessionResume( + bridge: Pick, + request: KeepaliveSessionResumeRequest, + timeoutMs: number, +): { completion: Promise; deadline: Promise } { + const completion = bridge.resumeSession(request); + return { + completion, + deadline: withTimeout( + completion, + timeoutMs, + `resumeSession(${request.sessionId})`, + ), + }; +} + /** Default caller headroom above the bridge's 60-second restore deadline. */ const KEEPALIVE_REVIVE_TIMEOUT_MS = 70_000; /** Per-task spawn timeout: a hung spawnOrAttach must not stall the sweep. */ @@ -352,11 +375,15 @@ export function startScheduledTaskKeepalive( const metadata = await new SessionService( boundWorkspace, ).readCreationMetadata(sessionId); - const resume = bridge.resumeSession({ - sessionId, - workspaceCwd: boundWorkspace, - ...metadata, - }); + const { completion: resume, deadline } = beginKeepaliveSessionResume( + bridge, + { + sessionId, + workspaceCwd: boundWorkspace, + ...metadata, + }, + reviveTimeoutMs, + ); // Clear the in-flight guard on the resume's TRUE settlement (not the // timeout below) so a still-running load keeps blocking a duplicate. void resume @@ -365,11 +392,7 @@ export function startScheduledTaskKeepalive( reviving.delete(sessionId); }); try { - await withTimeout( - resume, - reviveTimeoutMs, - `resumeSession(${sessionId})`, - ); + await deadline; log.debug('keepalive: revived non-resident session', sessionId); reviveState.delete(sessionId); } catch (loadErr) { diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 6a05be9b7f9..d24fe491e95 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -12734,6 +12734,23 @@ describe('createServeApp', () => { expect(bridge.calls).toHaveLength(0); }); + it('rejects the reserved mesh host source', async () => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ sourceType: 'mesh' }); + + expect(res.status).toBe(400); + expect(res.body.code).toBe('reserved_session_source'); + expect(bridge.calls).toHaveLength(0); + }); + it('forwards a valid UUID sessionId to the bridge', async () => { const bridge = fakeBridge(); const app = createServeApp( @@ -14437,6 +14454,44 @@ describe('createServeApp', () => { } }); + it('allows branch creation when only the hidden mesh host shares the workspace', async () => { + const bridge = fakeBridge({ + listImpl: () => [ + { + sessionId: 'mesh-host', + workspaceCwd: WS_BOUND, + createdAt: '2026-01-01T00:00:00.000Z', + clientCount: 1, + hasActivePrompt: false, + sourceType: 'mesh', + }, + ], + }); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + mockWt.impl = () => ({ + isGitRepository: () => Promise.resolve(true), + getCurrentBranch: () => Promise.resolve('main'), + }); + mockBranchOps.getHeadCommit = () => Promise.resolve('abc123'); + + try { + const res = await request(app) + .post('/session') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ branch: { name: 'feat/x' } }); + + expect(res.status).toBe(200); + expect(bridge.calls).toHaveLength(1); + } finally { + mockWt.impl = undefined; + mockBranchOps.getHeadCommit = undefined; + } + }); + it('allows branch creation when the sharing session has no attached client', async () => { // A detached session (e.g. left behind by a "new chat") is not actively // running, so it must not block a fresh branch session. @@ -18808,6 +18863,7 @@ describe('createServeApp', () => { archiveState: 'active', size: 1, signal: preflightSignal, + excludeSourceType: 'mesh', }); catalogRequest.abort(); await vi.waitFor(() => expect(preflightSignal?.aborted).toBe(true)); @@ -20351,6 +20407,7 @@ describe('createServeApp', () => { cursor: 1000123.456, size: 20, archiveState: 'active', + excludeSourceType: 'mesh', }); } finally { listSessionsSpy.mockRestore(); diff --git a/packages/cli/src/serve/server/session-list.ts b/packages/cli/src/serve/server/session-list.ts index df2a044d2f0..46f99865452 100644 --- a/packages/cli/src/serve/server/session-list.ts +++ b/packages/cli/src/serve/server/session-list.ts @@ -32,6 +32,7 @@ import { import { laterActivityTimestamp } from './activity-timestamp.js'; import { classifyTopLevelConversationSource } from '../../runtime/live-session-source.js'; import { parseCallerSuppliedSessionId } from '../../config/session-id.js'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; const DEFAULT_SESSION_PAGE_SIZE = 20; const MAX_SESSION_PAGE_SIZE = 100; @@ -688,6 +689,7 @@ async function loadAllPersistedSummaries( size: 10_000, archiveState, signal, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); signal.throwIfAborted(); const remaining = MAX_ORGANIZED_SESSIONS - sessions.length; @@ -1051,6 +1053,7 @@ async function listOrganizedWorkspaceSessionsForResponse( } const filtered = [...bySessionId.values()].filter((session) => { + if (session.sourceType === MESH_HOST_SESSION_SOURCE_TYPE) return false; if (!matchesSessionMetadataSource(session, options)) return false; if (group === 'all') return true; if (group === 'pinned') return session.isPinned === true; @@ -1243,6 +1246,7 @@ async function listWorkspaceSessionsByMetadataForResponse( const matches = [...bySessionId.values()] .filter( (session) => + session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE && (filter.parentSessionId === undefined || session.parentSessionId === filter.parentSessionId) && matchesSessionMetadataSource(session, filter), @@ -1401,6 +1405,7 @@ async function listWorkspaceSessionsForResponseInRuntime( cursor: numericCursor, size: pageSize, archiveState, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, ...(readOptions.signal ? { signal: readOptions.signal } : {}), }); readOptions.signal?.throwIfAborted(); @@ -1431,7 +1436,9 @@ async function listWorkspaceSessionsForResponseInRuntime( return { sessions, nextCursor }; } - const liveSessions = bridge.listWorkspaceSessions(workspaceCwd); + const liveSessions = bridge + .listWorkspaceSessions(workspaceCwd) + .filter((session) => session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE); for (const live of liveSessions) { const existing = bySessionId.get(live.sessionId); if (existing) { @@ -1498,6 +1505,7 @@ export async function listLiveWorkspaceSessionsForResponse( : undefined; const sessions = bridge .listWorkspaceSessions(workspaceCwd) + .filter((session) => session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE) .sort((a, b) => compareLiveSessionCursorKeys( getLiveSessionCursorKey(a), @@ -1578,7 +1586,7 @@ export async function searchWorkspaceSessionsForResponse( for (const hit of hits) { readOptions.signal?.throwIfAborted(); const item = await sessionService.getSessionListItem(hit.sessionId); - if (item) + if (item?.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE) bySessionId.set( hit.sessionId, applyOrganization( @@ -1621,14 +1629,22 @@ export async function getWorkspaceSessionInfoForResponse( workspaceCwd: string, options: { includeLive?: boolean } = {}, ): Promise { - const counts = await new SessionService(workspaceCwd).getSessionInfoCounts(); + const counts = await new SessionService(workspaceCwd).getSessionInfoCounts({ + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, + }); return { active: counts.active, archived: counts.archived, total: counts.total, ...(options.includeLive === false ? {} - : { live: bridge.listWorkspaceSessions(workspaceCwd).length }), + : { + live: bridge + .listWorkspaceSessions(workspaceCwd) + .filter( + (session) => session.sourceType !== MESH_HOST_SESSION_SOURCE_TYPE, + ).length, + }), expensive: true, cost: 'disk_scan', ...(counts.truncated ? { truncated: true } : {}), diff --git a/packages/cli/src/ui/components/StandaloneSessionPicker.tsx b/packages/cli/src/ui/components/StandaloneSessionPicker.tsx index abc9f390588..5303fc2a298 100644 --- a/packages/cli/src/ui/components/StandaloneSessionPicker.tsx +++ b/packages/cli/src/ui/components/StandaloneSessionPicker.tsx @@ -8,6 +8,7 @@ import { useState } from 'react'; import { render, Box, useApp } from 'ink'; import type { Config } from '@qwen-code/qwen-code-core/config/config.js'; import { SessionService } from '@qwen-code/qwen-code-core/services/sessionService.js'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import type { SessionListItem } from '@qwen-code/qwen-code-core/services/sessionService.js'; import { getGitBranch } from '@qwen-code/qwen-code-core/utils/gitUtils.js'; import { KeypressProvider } from '../contexts/KeypressContext.js'; @@ -108,7 +109,9 @@ export async function showResumeSessionPicker( initialSessions?: SessionListItem[], ): Promise { const sessionService = new SessionService(cwd); - const hasSession = await sessionService.loadLastSession(); + const hasSession = await sessionService.loadLastSession({ + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, + }); if (!hasSession) { writeStdoutLine('No sessions found. Start a new session with `qwen`.'); return undefined; diff --git a/packages/cli/src/ui/hooks/session-completion.test.ts b/packages/cli/src/ui/hooks/session-completion.test.ts index 6f357f74a86..d3f5f31610e 100644 --- a/packages/cli/src/ui/hooks/session-completion.test.ts +++ b/packages/cli/src/ui/hooks/session-completion.test.ts @@ -48,6 +48,10 @@ describe('getSessionSuggestions', () => { hasMore: false, }); const out = await getSessionSuggestions('/proj', ''); + expect(mockListSessions).toHaveBeenCalledWith({ + size: 20, + excludeSourceType: 'mesh', + }); expect(out).toHaveLength(2); expect(out[0]).toMatchObject({ label: 'Fix auth bug', diff --git a/packages/cli/src/ui/hooks/session-completion.ts b/packages/cli/src/ui/hooks/session-completion.ts index bcdd40d1c10..e6d91a586cf 100644 --- a/packages/cli/src/ui/hooks/session-completion.ts +++ b/packages/cli/src/ui/hooks/session-completion.ts @@ -6,6 +6,7 @@ import { SessionService } from '@qwen-code/qwen-code-core'; import type { SessionListItem } from '@qwen-code/qwen-code-core'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import type { Suggestion } from '../components/SuggestionsDisplay.js'; import { buildSessionRef, @@ -29,7 +30,7 @@ interface CacheEntry { expiresAt: number; } -// Cache the UNFILTERED listing keyed by cwd; pattern filtering is cheap and +// Cache the visible listing keyed by cwd; pattern filtering is cheap and // always applied fresh below. const listingCache = new Map(); @@ -49,6 +50,7 @@ async function listSessionsCached( try { const res = await new SessionService(cwd).listSessions({ size: MAX_SESSION_SUGGESTIONS, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); listingCache.set(cwd, { items: res.items, diff --git a/packages/cli/src/ui/hooks/useSessionPicker.ts b/packages/cli/src/ui/hooks/useSessionPicker.ts index 0966ca08e92..2f0e7765c3c 100644 --- a/packages/cli/src/ui/hooks/useSessionPicker.ts +++ b/packages/cli/src/ui/hooks/useSessionPicker.ts @@ -14,6 +14,7 @@ */ import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import type { ListSessionsResult, SessionListItem, @@ -264,6 +265,7 @@ export function useSessionPicker({ try { const result: ListSessionsResult = await sessionService.listSessions({ size: SESSION_PAGE_SIZE, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); setSessionState({ sessions: result.items, @@ -288,6 +290,7 @@ export function useSessionPicker({ const result: ListSessionsResult = await sessionService.listSessions({ size: SESSION_PAGE_SIZE, cursor: sessionState.nextCursor, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, }); setSessionState((prev) => ({ sessions: [...prev.sessions, ...result.items], diff --git a/packages/cli/src/ui/opentui/dialogs-misc.tsx b/packages/cli/src/ui/opentui/dialogs-misc.tsx index 7e4068dcc57..62288340829 100644 --- a/packages/cli/src/ui/opentui/dialogs-misc.tsx +++ b/packages/cli/src/ui/opentui/dialogs-misc.tsx @@ -22,6 +22,7 @@ import { type ReactNode, } from 'react'; import { useRenderer, useKeyboard } from '@opentui/react'; +import { MESH_HOST_SESSION_SOURCE_TYPE } from '../../runtime/mesh-session-source.js'; import type { Config } from '@qwen-code/qwen-code-core/config/config.js'; import type { SessionListItem } from '@qwen-code/qwen-code-core/services/sessionService.js'; import type { EditorType } from '@qwen-code/qwen-code-core/utils/editor.js'; @@ -468,7 +469,10 @@ export function OpenTuiResumeDialog({ return; } svc - .listSessions({ size: 10 }) + .listSessions({ + size: 10, + excludeSourceType: MESH_HOST_SESSION_SOURCE_TYPE, + }) .then((res) => { if (!alive) return; setRows(res.items ?? []); diff --git a/packages/core/src/agents/index.ts b/packages/core/src/agents/index.ts index 09eb614a5bf..0d4723b37d5 100644 --- a/packages/core/src/agents/index.ts +++ b/packages/core/src/agents/index.ts @@ -29,7 +29,12 @@ export { sanitizeFilenameComponent, } from './agent-transcript.js'; export type { AgentTrace, AgentTraceNode } from './agent-transcript.js'; +export { + claimMeshHostSession, + readMeshAgents, + readMeshWorkspace, +} from './mesh/mesh-store.js'; export { launchMeshAgent } from './mesh/launcher.js'; export type { MeshAgentLaunchResult } from './mesh/launcher.js'; -export type { MeshAgent } from './mesh/types.js'; +export type { MeshAgent, MeshWorkspaceState } from './mesh/types.js'; export * from './tasks/types.js'; diff --git a/packages/core/src/agents/mesh/mesh-store.ts b/packages/core/src/agents/mesh/mesh-store.ts index 3908b2ba34c..1d582d12fab 100644 --- a/packages/core/src/agents/mesh/mesh-store.ts +++ b/packages/core/src/agents/mesh/mesh-store.ts @@ -1021,6 +1021,25 @@ export async function readMeshWorkspace( }); } +export async function claimMeshHostSession( + projectRoot: string, + candidateSessionId: string, +): Promise { + if (!isNonEmptyString(candidateSessionId)) { + throw new Error('Mesh host session id must be a non-empty string.'); + } + return withWorkspaceLock(projectRoot, async () => { + const workspace = await ensureMigratedUnlocked(projectRoot); + if (workspace.hostSessionId) return workspace.hostSessionId; + await atomicWriteJSON( + getWorkspaceFilePath(projectRoot), + { ...workspace, hostSessionId: candidateSessionId }, + { noFollow: true }, + ); + return candidateSessionId; + }); +} + export async function readMeshAgents( projectRoot: string, ): Promise { diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index 611af2fe0d8..9b2b1154605 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -7694,6 +7694,60 @@ describe('SessionService', () => { systemPayload: { text: 'x'.repeat(350) }, })); + it('excludes a source before applying the page size', async () => { + const visibleId = '00000000-0000-4000-8000-000000000001'; + const hiddenId = '00000000-0000-4000-8000-000000000002'; + const visibleFile = writeSession(visibleId, [ + userLine(visibleId, 'visible'), + ]); + const hiddenFile = writeSession(hiddenId, [ + userLine(hiddenId, 'hidden'), + { + ...sessionSourceLine(hiddenId), + systemPayload: { sourceType: 'mesh' }, + }, + ]); + fs.utimesSync(visibleFile, new Date(1), new Date(1)); + fs.utimesSync(hiddenFile, new Date(2), new Date(2)); + + await expect( + service.listSessions({ size: 1, excludeSourceType: 'mesh' }), + ).resolves.toMatchObject({ + items: [{ sessionId: visibleId }], + hasMore: false, + nextCursor: undefined, + }); + }); + + it('excludes the source from active and archived counts', async () => { + const visibleId = '00000000-0000-4000-8000-000000000001'; + const hiddenId = '00000000-0000-4000-8000-000000000002'; + writeSession(visibleId, [userLine(visibleId, 'visible')]); + const archiveDir = realPath.join(getChatsDir(), 'archive'); + fs.mkdirSync(archiveDir, { recursive: true }); + fs.writeFileSync( + realPath.join(archiveDir, `${hiddenId}.jsonl`), + `${[ + userLine(hiddenId, 'hidden'), + { + ...sessionSourceLine(hiddenId), + systemPayload: { sourceType: 'mesh' }, + }, + ] + .map((line) => JSON.stringify(line)) + .join('\n')}\n`, + ); + + await expect( + service.getSessionInfoCounts({ excludeSourceType: 'mesh' }), + ).resolves.toEqual({ + active: 1, + archived: 0, + total: 1, + truncated: false, + }); + }); + // Short complete fixtures answer from parsed records before the scan runs. // The long and truncated fixtures below drive the real tail-window scan // and pin the production marker (`"subtype":"goal_state"`) and field name. diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index 23115daa3c3..30ad4b8efae 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -274,6 +274,8 @@ export interface ListSessionsOptions { archiveState?: SessionArchiveState; /** Aborts an in-progress catalog scan. */ signal?: AbortSignal; + /** Omits records carrying this immutable creator attribution. */ + excludeSourceType?: string; } /** @@ -2520,7 +2522,13 @@ export class SessionService { async listSessions( options: ListSessionsOptions = {}, ): Promise { - const { cursor, size = 20, archiveState = 'active', signal } = options; + const { + cursor, + size = 20, + archiveState = 'active', + signal, + excludeSourceType, + } = options; const chatsDir = this.getChatsDirForState(archiveState); const isArchived = archiveState === 'archived'; signal?.throwIfAborted(); @@ -2625,6 +2633,17 @@ export class SessionService { continue; } + const source = this.extractCreationMetadataFromFile( + filePath, + records, + tailBuffer, + ); + if ( + excludeSourceType !== undefined && + source.sourceType === excludeSourceType + ) { + continue; + } const prompt = this.extractFirstPromptFromRecords(records); signal?.throwIfAborted(); const titleInfo = this.readSessionTitleInfoFromFile(filePath, tailBuffer); @@ -2637,11 +2656,6 @@ export class SessionService { readResult.complete, tailBuffer, ); - const source = this.extractCreationMetadataFromFile( - filePath, - records, - tailBuffer, - ); items.push({ sessionId: firstRecord.sessionId, cwd: firstRecord.cwd, @@ -2742,16 +2756,19 @@ export class SessionService { * * Same disk-walk shape as {@link findSessionTitlesByPrefix} / * {@link findSessionsByTitle}: `readdir` the chats dir, cap at the - * file-processing safety limit, then read only the first JSONL record for - * project membership. Title/prompt/message hydration is skipped entirely. + * file-processing safety limit, then read the first JSONL record for project + * membership. A requested source exclusion adds one bounded tail read; + * title/prompt/message hydration is still skipped. * * Still an O(n) disk walk — callers (and HTTP clients of * `GET .../session-info`) must not poll this in a tight loop. */ - async getSessionInfoCounts(): Promise { + async getSessionInfoCounts( + options: { excludeSourceType?: string } = {}, + ): Promise { const [active, archived] = await Promise.all([ - this.countSessionsInState('active'), - this.countSessionsInState('archived'), + this.countSessionsInState('active', options.excludeSourceType), + this.countSessionsInState('archived', options.excludeSourceType), ]); return { active: active.count, @@ -2763,6 +2780,7 @@ export class SessionService { private async countSessionsInState( archiveState: SessionArchiveState, + excludeSourceType?: string, ): Promise<{ count: number; truncated: boolean }> { const chatsDir = this.getChatsDirForState(archiveState); let fileNames: string[]; @@ -2778,6 +2796,10 @@ export class SessionService { let count = 0; let filesProcessed = 0; let truncated = false; + const tailBuffer = + excludeSourceType === undefined + ? undefined + : Buffer.alloc(LITE_READ_BUF_SIZE); for (const name of fileNames) { if (!SESSION_FILE_PATTERN.test(name)) continue; @@ -2806,6 +2828,13 @@ export class SessionService { ) { continue; } + if ( + excludeSourceType !== undefined && + this.extractCreationMetadataFromFile(filePath, records, tailBuffer) + .sourceType === excludeSourceType + ) { + continue; + } count++; } catch { truncated = true; @@ -4345,8 +4374,10 @@ export class SessionService { * * @returns Session data for resumption, or undefined if no sessions exist */ - async loadLastSession(): Promise { - const result = await this.listSessions({ size: 1 }); + async loadLastSession( + options: Pick = {}, + ): Promise { + const result = await this.listSessions({ size: 1, ...options }); if (result.items.length === 0) { return; }