From ac2f371c44be62e81e8efd8314b82b980033f923 Mon Sep 17 00:00:00 2001 From: Shaojin Wen Date: Thu, 9 Jul 2026 20:02:39 +0800 Subject: [PATCH] feat(scheduled-tasks): add isolated run mode via create_sub_session tool (#6535) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(scheduled-tasks): add isolated run mode via create_sub_session tool Introduce a new `create_sub_session` tool (daemon-only) that spawns a fresh top-level sub-session with its own clean context and transcript. Wire it into the cron scheduler as an `isolated` run mode so each scheduled fire dispatches its prompt into a fresh sub-session instead of accumulating in one shared transcript. - Add `create_sub_session` tool with `first-turn` and `sent` completion modes - Add `SubSessionLauncher` in cli/serve with concurrency cap, timeout, and truncation - Extend ACP bridge `extMethod` dispatch for child→daemon sub-session requests - Add `runMode` field (`shared`|`isolated`) to DurableCronTask, CronJob, and API types - Add run-mode radio picker to ScheduledTasksDialog UI - Fix AuthMessage hardcoded placeholder to use i18n key * fix(scheduled-tasks): dispatch isolated fires daemon-side, not via the model An `isolated` fire was relayed through the model: the fired prompt was wrapped with an instruction to call `create_sub_session`. That tool's default permission is `'ask'`, so under `ApprovalMode.DEFAULT` an unattended fire reached `client.requestPermission`, found no SSE subscriber, and was cancelled by the daemon's 5-minute permission timeout. The task never ran, and the cancel was booked as a successful run — the headline use case of a scheduled task was broken. Route isolated fires straight to the daemon instead: the cron `onFire` handler in `Session` calls the sub-session spawner directly, with no model relay and no tool-permission gate. The prompt was already approved when the task was created; laundering it back through the model only re-opened that gate. `create_sub_session` keeps `'ask'` for model-initiated calls, and the attended "Run now" button keeps its relay (a user is present to answer the prompt). Also fix orphan-session cleanup in the launcher. `closeSession` was guarded only by `.catch()`, which covers an async rejection but not a synchronous throw; because the call sits inside the launcher's own `catch (err)` block, a sync throw escaped and replaced the real launch error. Guard both shapes. Tests: - Cover isolated routing: dispatch, in-session fallback with no spawner, missed one-shot, dispatch failure (dropped, never run inline), and shared mode. - Cover the orphan close, including a `closeSession` that throws. - Replace the sent-mode concurrency test, which only asserted the slot was eventually released (moving the release to the drain's *start* kept it green) with one that asserts the slot is HELD while the drain runs, plus one that asserts it is released at `turn_complete`. * fix(scheduled-tasks): honor the caller's AbortSignal and harden the spawn boundary Four findings from review, all in the model-initiated `create_sub_session` path (the scheduled `isolated` dispatch reaches none of them). `execute()` took no parameters, so it silently dropped the parent turn's `AbortSignal`. `Session.ts` awaits `invocation.execute(signal)` without racing the abort itself, so cancelling a turn with a `first-turn` sub-session in flight pinned the caller's tool loop until the daemon's 5-minute ceiling. Accept the signal and return as soon as it fires. The sub-session is deliberately NOT cancelled and deliberately KEEPS its concurrency slot: `sendPrompt` has no abort seam, so the sub-session runs on. Releasing its slot on cancel — as the review suggested — would let the caller over-admit against sub-sessions that are still consuming a bridge session and model quota. `handleCreateSubSession` trusted the child-supplied `callerSessionId` verbatim, and that id keys the launcher's per-caller concurrency bucket: a fabricated id starts a fresh bucket at zero (cap evasion) and a victim's id burns their slots (DoS). Validate it with the connection's existing `ownsSession` seam. Every daemon session wires a spawner, sub-sessions included, and each gets its own cap-sized bucket — so one prompt could fan out 5ⁿ sub-sessions until `maxSessions` ran dry. Gate nesting at one level: the launcher remembers the sessions it spawned and refuses to spawn from them. With `callerSessionId` now authenticated, the gate cannot be sidestepped. Cap the prompt at 100,000 chars (matching the scheduled-task REST route) and the display name at 200, both at the bridge trust boundary and, for the prompt, in the tool's own validation so the model gets an actionable error. Not changed: `create_sub_session` stays in `PermissionManager.CORE_TOOLS`. Membership there SUBJECTS a tool to the `coreTools` allowlist; it does not exempt it. Removing it — as the review suggested — is what would let the tool bypass a user's allowlist, the way `agent` and `send_message` do today. * fix(core): do not spawn a sub-session for an already-cancelled turn `raceCancellation(spawner({…}), signal)` evaluated the spawner as an argument, so the spawn started before the abort was ever checked. A turn cancelled before `execute()` ran still created a sub-session on the daemon — and it kept a concurrency slot — while the tool reported itself cancelled. Take a thunk instead, so the pre-abort check happens before any daemon work is started. Track whether the spawn actually began, and say so: "cancelled before it started, no sub-session was created" is a different fact from "a sub-session may already have been created and is not cancelled". Regression test asserts the spawner is never called for a pre-aborted signal; it fails against the eager-argument form. * fix(serve): require callerSessionId and stop misreporting an early stream close Two findings from review. `awaitFirstTurn`'s `'incomplete'` stopReason was unreachable. The cleanup `finally` calls `ac.abort()` unconditionally to tear the subscription down, so by the time the stopReason ternary read `ac.signal.aborted` it was always true. An event stream that closed before the turn finished (bridge teardown, WS drop) was reported as a 5-minute wall-clock `'timeout'` — indistinguishable from a real one. Track the timer firing in its own flag. `callerSessionId` was validated only when present. Omitting it handed the launcher `undefined`, which minted an `anon:` bucket — a fresh concurrency bucket per call, so no cap — and skipped the depth-1 nesting gate (`info.callerSessionId !== undefined && …`). Authenticating the id closed forgery but not omission. It is now required at the bridge boundary, and required in `CreateSubSessionInfo`, so the launcher's anonymous fallback and the gate's presence check are both gone. Every real caller has a session id — the tool only ever runs inside a session's turn. * fix(serve): surface dropped fires and drain timeouts; bound sub-sessions per workspace Three findings from review. A dropped `isolated` scheduled fire left no trace. `debugLogger.warn` writes nothing unless a debug log session is active, and the scheduler persists the fire as a run before dispatch — so a nightly task could fail forever while its history claimed it ran. It now also writes to stderr, which the daemon forwards from the child. A sent-mode drain that hit its 30-minute ceiling was equally silent: the catch saw `drainAc.signal.aborted` and skipped logging, the `finally` freed the concurrency slot, and the sub-session — which the abort does not cancel — kept burning a bridge session and model quota. The timer now records its own firing (the controller cannot: `finally` aborts it on every exit path) and the timeout is written to stderr. The drain ceiling is injectable for tests, mirroring `firstTurnTimeoutMs`. The per-caller concurrency cap trusts `callerSessionId`, and the bridge can only authenticate that id as "a session on this channel". Every session of a workspace shares one child process, so nothing at the transport can prove which of them issued the call — and a per-session secret would be readable by the whole process anyway. Rather than pretend otherwise, add a workspace-wide ceiling on concurrent sub-sessions that holds no matter which bucket a launch is charged to. --- packages/acp-bridge/src/bridge.ts | 4 + packages/acp-bridge/src/bridgeClient.test.ts | 186 ++++++ packages/acp-bridge/src/bridgeClient.ts | 117 +++- packages/acp-bridge/src/bridgeOptions.ts | 57 ++ packages/acp-bridge/src/status.ts | 9 + .../acp-integration/session/Session.test.ts | 173 ++++++ .../src/acp-integration/session/Session.ts | 106 +++- .../session/Session.worktree.test.ts | 2 + packages/cli/src/i18n/locales/en.js | 1 + packages/cli/src/i18n/locales/zh-TW.js | 1 + packages/cli/src/i18n/locales/zh.js | 1 + .../cli/src/serve/create-sub-session.test.ts | 568 ++++++++++++++++++ packages/cli/src/serve/create-sub-session.ts | 533 ++++++++++++++++ .../src/serve/routes/scheduled-tasks.test.ts | 59 ++ .../cli/src/serve/routes/scheduled-tasks.ts | 33 + packages/cli/src/serve/run-qwen-serve.test.ts | 2 + packages/cli/src/serve/run-qwen-serve.ts | 54 ++ packages/core/src/config/config.ts | 61 ++ .../src/permissions/permission-manager.ts | 1 + .../core/src/services/cronScheduler.test.ts | 29 + packages/core/src/services/cronScheduler.ts | 10 + .../core/src/services/cronTasksFile.test.ts | 25 + packages/core/src/services/cronTasksFile.ts | 17 + .../core/src/tools/create-sub-session.test.ts | 172 ++++++ packages/core/src/tools/create-sub-session.ts | 318 ++++++++++ packages/core/src/tools/tool-names.ts | 2 + packages/web-shell/client/App.tsx | 13 + .../dialogs/ScheduledTasksDialog.module.css | 26 + .../dialogs/ScheduledTasksDialog.test.tsx | 75 ++- .../dialogs/ScheduledTasksDialog.tsx | 84 ++- .../client/components/messages/Markdown.tsx | 24 + .../client/components/messages/ToolGroup.tsx | 47 +- packages/web-shell/client/i18n.tsx | 12 + packages/webui/src/daemon/workspace/types.ts | 11 + 34 files changed, 2814 insertions(+), 19 deletions(-) create mode 100644 packages/cli/src/serve/create-sub-session.test.ts create mode 100644 packages/cli/src/serve/create-sub-session.ts create mode 100644 packages/core/src/tools/create-sub-session.test.ts create mode 100644 packages/core/src/tools/create-sub-session.ts diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 38f910face..92ff2af52c 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -1726,6 +1726,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { outputTokens, durationMs, ), + // `create_sub_session` tool: forward the request/response hook so a child + // tool can ask the daemon to spawn a sub-session and (for 'first-turn') + // return its result. Omitted → the method reports daemon-only. + opts.onCreateSubSession, ); const connection = new ClientSideConnection(() => client, channel.stream); diff --git a/packages/acp-bridge/src/bridgeClient.test.ts b/packages/acp-bridge/src/bridgeClient.test.ts index ebca09023a..18d1d5a6aa 100644 --- a/packages/acp-bridge/src/bridgeClient.test.ts +++ b/packages/acp-bridge/src/bridgeClient.test.ts @@ -48,6 +48,10 @@ import { } from '@qwen-code/qwen-code-core'; import type { JSONRPCMessage } from '@modelcontextprotocol/sdk/types.js'; import { BridgeClient } from './bridgeClient.js'; +import { + MAX_SUB_SESSION_NAME_CHARS, + MAX_SUB_SESSION_PROMPT_CHARS, +} from './bridgeOptions.js'; import type { BridgeFileSystem } from './bridgeFileSystem.js'; import type { MidTurnQueueEntry } from './bridgeTypes.js'; import type { ClientMcpMessageSender } from './bridgeOptions.js'; @@ -717,6 +721,188 @@ describe('BridgeClient — token usage accounting', () => { }); }); +describe('BridgeClient — create-sub-session extMethod dispatch', () => { + const noFlow = () => { + throw new Error('test: should not run'); + }; + + function makeClientWithCreateSubSession( + onCreateSubSession: + | ((info: { + prompt: string; + completion: 'sent' | 'first-turn'; + model?: string; + name?: string; + callerSessionId?: string; + }) => Promise<{ + sessionId: string; + result?: string; + stopReason?: string; + }>) + | undefined, + ownsSession?: (sessionId: string) => boolean, + ) { + return new BridgeClient( + (() => undefined) as never, // resolveEntry + noFlow as never, + { request: noFlow } as never, + 0, + Infinity, + undefined, // fileSystem + undefined, // onModelPromoted + undefined, // onModePromoted + undefined, // clientMcpSender + ownsSession as never, // ownsSession (undefined → defaults to () => true) + undefined, // onTokenUsage + onCreateSubSession, + ); + } + + const METHOD = 'qwen/control/create-sub-session'; + + it('forwards a valid request to the host handler and returns its result', async () => { + const onCreate = vi.fn(async () => ({ + sessionId: 'sub-9', + result: 'done', + stopReason: 'end_turn', + })); + const client = makeClientWithCreateSubSession(onCreate); + + const res = await client.extMethod(METHOD, { + prompt: 'summarize', + completion: 'first-turn', + model: 'm1', + name: 'digest', + callerSessionId: 'caller-1', + }); + + expect(onCreate).toHaveBeenCalledWith({ + prompt: 'summarize', + completion: 'first-turn', + model: 'm1', + name: 'digest', + callerSessionId: 'caller-1', + }); + expect(res).toEqual({ + sessionId: 'sub-9', + result: 'done', + stopReason: 'end_turn', + }); + }); + + it('omits result/stopReason when the handler does not return them (sent mode)', async () => { + const client = makeClientWithCreateSubSession(async () => ({ + sessionId: 'sub-10', + })); + const res = await client.extMethod(METHOD, { + prompt: 'go', + completion: 'sent', + callerSessionId: 'caller-1', + }); + expect(res).toEqual({ sessionId: 'sub-10' }); + }); + + it('rejects methodNotFound when no host handler is wired (non-daemon)', async () => { + const client = makeClientWithCreateSubSession(undefined); + await expect( + client.extMethod(METHOD, { + prompt: 'x', + completion: 'sent', + callerSessionId: 'caller-1', + }), + ).rejects.toThrow(); + }); + + it('rejects invalid params (missing prompt, bad completion)', async () => { + const onCreate = vi.fn(); + const client = makeClientWithCreateSubSession( + onCreate as unknown as Parameters< + typeof makeClientWithCreateSubSession + >[0], + ); + await expect( + client.extMethod(METHOD, { completion: 'sent' }), + ).rejects.toThrow(); + await expect( + client.extMethod(METHOD, { prompt: 'x', completion: 'weird' }), + ).rejects.toThrow(); + expect(onCreate).not.toHaveBeenCalled(); + }); + + it('rejects a callerSessionId this connection does not own', async () => { + // The key names the launcher's per-caller concurrency bucket. A child that + // can name any session evades its own cap (a fabricated id starts a fresh + // bucket at zero) and can burn a victim session's slots. + const onCreate = vi.fn(async () => ({ sessionId: 'sub-11' })); + const client = makeClientWithCreateSubSession( + onCreate, + (id) => id === 'mine', + ); + + await expect( + client.extMethod(METHOD, { + prompt: 'x', + completion: 'sent', + callerSessionId: 'victim', + }), + ).rejects.toThrow(/callerSessionId/i); + expect(onCreate).not.toHaveBeenCalled(); + + // An owned id passes through untouched. + await client.extMethod(METHOD, { + prompt: 'x', + completion: 'sent', + callerSessionId: 'mine', + }); + expect(onCreate).toHaveBeenCalledWith({ + prompt: 'x', + completion: 'sent', + callerSessionId: 'mine', + }); + + // Omitting it is NOT legal: an absent id would give the launcher an + // anonymous per-call bucket (no cap) and skip its depth-1 nesting gate. + onCreate.mockClear(); + await expect( + client.extMethod(METHOD, { prompt: 'x', completion: 'sent' }), + ).rejects.toThrow(/callerSessionId/i); + expect(onCreate).not.toHaveBeenCalled(); + }); + + it('rejects an over-long prompt and an over-long name', async () => { + const onCreate = vi.fn(async () => ({ sessionId: 'sub-12' })); + const client = makeClientWithCreateSubSession(onCreate); + + await expect( + client.extMethod(METHOD, { + prompt: 'x'.repeat(MAX_SUB_SESSION_PROMPT_CHARS + 1), + completion: 'sent', + callerSessionId: 'caller-1', + }), + ).rejects.toThrow(new RegExp(`${MAX_SUB_SESSION_PROMPT_CHARS}`)); + + await expect( + client.extMethod(METHOD, { + prompt: 'x', + completion: 'sent', + name: 'n'.repeat(MAX_SUB_SESSION_NAME_CHARS + 1), + callerSessionId: 'caller-1', + }), + ).rejects.toThrow(new RegExp(`${MAX_SUB_SESSION_NAME_CHARS}`)); + + expect(onCreate).not.toHaveBeenCalled(); + + // Both boundaries are accepted. + await client.extMethod(METHOD, { + prompt: 'x'.repeat(MAX_SUB_SESSION_PROMPT_CHARS), + completion: 'sent', + name: 'n'.repeat(MAX_SUB_SESSION_NAME_CHARS), + callerSessionId: 'caller-1', + }); + expect(onCreate).toHaveBeenCalledTimes(1); + }); +}); + describe('BridgeClient — artifact ingress', () => { const noPermissionFlow = () => { throw new Error('test: permission flow should not run'); diff --git a/packages/acp-bridge/src/bridgeClient.ts b/packages/acp-bridge/src/bridgeClient.ts index 7dbc1e732d..c55752e80f 100644 --- a/packages/acp-bridge/src/bridgeClient.ts +++ b/packages/acp-bridge/src/bridgeClient.ts @@ -27,7 +27,14 @@ import { MID_TURN_MESSAGE_INJECTED_EVENT } from './daemonEventTypes.js'; import { MID_TURN_QUEUE_DRAIN_METHOD } from './bridgeTypes.js'; import type { MidTurnQueueEntry } from './bridgeTypes.js'; import { SERVE_CONTROL_EXT_METHODS } from './status.js'; -import type { ClientMcpMessageSender } from './bridgeOptions.js'; +import type { + ClientMcpMessageSender, + CreateSubSessionHandler, +} from './bridgeOptions.js'; +import { + MAX_SUB_SESSION_NAME_CHARS, + MAX_SUB_SESSION_PROMPT_CHARS, +} from './bridgeOptions.js'; import type { BridgeFileSystem } from './bridgeFileSystem.js'; import { CANCEL_VOTE_SENTINEL } from './permissionMediator.js'; // Narrowed from the concrete `MultiClientPermissionMediator` to the @@ -502,6 +509,15 @@ export class BridgeClient implements Client { outputTokens: number, durationMs?: number, ) => void, + /** + * Daemon-host seam for the `create_sub_session` tool. Invoked from the + * `extMethod` dispatch (a child→daemon REQUEST, so it returns a Promise the + * child awaits) with the prompt, completion mode, and optional model/name; + * the host spawns a sub-session and, for `'first-turn'`, returns its result. + * Omitted by tests / Mode A / non-daemon — the method then reports + * `methodNotFound` and the tool surfaces itself as daemon-only. + */ + private readonly onCreateSubSession?: CreateSubSessionHandler, ) {} async requestPermission( @@ -818,9 +834,12 @@ export class BridgeClient implements Client { /** * Handle child->bridge ACP `extMethod` requests (calls that expect a - * response, unlike `extNotification`). The only method served today is - * `craft/drainMidTurnQueue`: the ACP child calls it between tool batches to - * pull any messages the browser queued mid-turn. We splice the per-session + * response, unlike `extNotification`). Served methods: + * `qwen/control/client_mcp/message` (reverse tool channel), + * `qwen/control/create-sub-session` (the `create_sub_session` tool → daemon + * spawns a sub-session and, for `'first-turn'`, returns its first-turn + * result), and `craft/drainMidTurnQueue`: the ACP child calls the last one + * between tool batches to pull any messages the browser queued mid-turn. We splice the per-session * queue, return them to the child as the response, and — when non-empty — * publish a `mid_turn_message_injected` SSE frame so the browser can move * those messages out of its pending queue (a dedupe signal, not a transcript @@ -842,6 +861,9 @@ export class BridgeClient implements Client { if (method === SERVE_CONTROL_EXT_METHODS.clientMcpMessage) { return this.handleClientMcpMessage(params); } + if (method === SERVE_CONTROL_EXT_METHODS.createSubSession) { + return this.handleCreateSubSession(params); + } if (method !== MID_TURN_QUEUE_DRAIN_METHOD) { throw RequestError.methodNotFound(method); } @@ -942,6 +964,89 @@ export class BridgeClient implements Client { return { payload: response as Record }; } + /** + * Handle the `create_sub_session` tool's request: validate, then forward to + * the daemon-host `onCreateSubSession` callback (which spawns a fresh + * top-level sub-session and, for `'first-turn'`, waits for its first turn and + * returns the result). No host wired → `methodNotFound`, which the tool + * surfaces as "daemon-only". + */ + private async handleCreateSubSession( + params: Record, + ): Promise> { + if (!this.onCreateSubSession) { + throw RequestError.methodNotFound( + SERVE_CONTROL_EXT_METHODS.createSubSession, + ); + } + const prompt = params['prompt']; + if (typeof prompt !== 'string' || prompt.length === 0) { + throw RequestError.invalidParams( + undefined, + '`prompt` must be a non-empty string', + ); + } + // The child is a separate process; this is a trust boundary. Without a cap + // it can hand the daemon a multi-MB string to deserialize, copy for the + // display name, and dispatch into a new session. Same ceiling the + // scheduled-task REST route applies to the prompts it accepts. + if (prompt.length > MAX_SUB_SESSION_PROMPT_CHARS) { + throw RequestError.invalidParams( + undefined, + `\`prompt\` exceeds the ${MAX_SUB_SESSION_PROMPT_CHARS}-character limit`, + ); + } + const completion = params['completion']; + if (completion !== 'sent' && completion !== 'first-turn') { + throw RequestError.invalidParams( + undefined, + "`completion` must be 'sent' or 'first-turn'", + ); + } + const name = params['name']; + if (typeof name === 'string' && name.length > MAX_SUB_SESSION_NAME_CHARS) { + throw RequestError.invalidParams( + undefined, + `\`name\` exceeds the ${MAX_SUB_SESSION_NAME_CHARS}-character limit`, + ); + } + // `callerSessionId` keys the launcher's per-caller concurrency bucket AND + // its depth-1 nesting gate. A child that names a session it does not own + // could evade its own cap (a fabricated id starts every bucket at zero) or + // exhaust a victim session's; a child that OMITS the field would get an + // anonymous per-call bucket and skip the nesting gate entirely. Neither is + // acceptable, so it is required and authenticated. Every real caller has + // one — the tool runs inside a session's turn. + const callerSessionId = params['callerSessionId']; + if ( + typeof callerSessionId !== 'string' || + callerSessionId.length === 0 || + !this.ownsSession(callerSessionId) + ) { + throw RequestError.invalidParams( + undefined, + '`callerSessionId` is required and must name a session owned by this connection', + ); + } + const model = params['model']; + const result = await this.onCreateSubSession({ + prompt, + completion, + ...(typeof model === 'string' && model.length > 0 && model.length <= 128 + ? { model } + : {}), + ...(typeof name === 'string' && name.length > 0 ? { name } : {}), + callerSessionId, + }); + return { + sessionId: result.sessionId, + ...(result.result !== undefined ? { result: result.result } : {}), + ...(result.stopReason !== undefined + ? { stopReason: result.stopReason } + : {}), + }; + } + /** * Handle child->bridge ACP `extNotification` calls. Recognized methods are * `qwen/notify/session/model-update`, @@ -951,8 +1056,8 @@ export class BridgeClient implements Client { * `qwen/notify/session/artifact-event` (hook artifacts), * `qwen/notify/session/terminal-sequence`, and * `qwen/notify/session/mcp-budget-event` — each translated into a - * session-scoped SSE frame. Unknown methods are dropped silently - * for forward-compat. + * session-scoped SSE frame. Unknown methods are dropped silently for + * forward-compat. */ async extNotification( method: string, diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index 97ea71f1ad..939b0da72f 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -440,6 +440,16 @@ export interface BridgeOptions { * receives an SDK MCP runtime server, so the method is never called. */ clientMcpSender?: ClientMcpMessageSender; + /** + * Daemon-host seam for the `create_sub_session` tool. When a tool running + * inside a child's agent turn asks (over `extMethod`) to spawn a fresh + * top-level sub-session and run a prompt in it, the bridge's `extMethod` + * dispatch forwards it here. It RETURNS A PROMISE so the `'first-turn'` completion + * mode can wait for the sub-session's first turn and return its result to the + * caller. Omitted by tests / Mode A / non-daemon embeds — the tool then + * reports itself unavailable (daemon-only). + */ + onCreateSubSession?: CreateSubSessionHandler; } /** @@ -453,3 +463,50 @@ export interface BridgeOptions { export type ClientMcpMessageSender = ( serverName: string, ) => ((payload: unknown) => Promise) | undefined; + +/** Ceiling on a sub-session prompt arriving over `extMethod`. The child is a + * separate process, so this is a trust boundary — mirrors the scheduled-task + * REST route's `MAX_PROMPT_LENGTH` and the core tool's own client-side check. */ +export const MAX_SUB_SESSION_PROMPT_CHARS = 100_000; + +/** Ceiling on the sub-session display name. It is a label — the launcher + * truncates it to 60 chars for display anyway. */ +export const MAX_SUB_SESSION_NAME_CHARS = 200; + +/** + * Payload the bridge forwards to {@link BridgeOptions.onCreateSubSession} when + * the `create_sub_session` tool requests a sub-session. + */ +export interface CreateSubSessionInfo { + /** Prompt to run in the freshly-spawned sub-session. */ + prompt: string; + /** When the request resolves: `'sent'` = as soon as the prompt is dispatched; + * `'first-turn'` = after the sub-session's first turn completes (result + * returned). Extensible — more criteria may be added later. */ + completion: 'sent' | 'first-turn'; + /** Optional model service id for the sub-session (falls back to default). */ + model?: string; + /** Optional display name for the sub-session in the session list. */ + name?: string; + /** + * The calling session's id. REQUIRED, and authenticated against the + * connection's owned sessions before it reaches the host — it keys the + * per-caller concurrency bucket and the depth-1 nesting gate, so a caller + * that could omit it would face neither. + */ + callerSessionId: string; +} + +/** Result the daemon host returns for a create-sub-session request. `result` + * (the sub-session's first-turn output) is present only for `'first-turn'`. */ +export interface CreateSubSessionResult { + sessionId: string; + result?: string; + stopReason?: string; +} + +/** Daemon-host callback that spawns a sub-session and (for `'first-turn'`) waits + * for its first turn. Returns a Promise so the tool can block on the result. */ +export type CreateSubSessionHandler = ( + info: CreateSubSessionInfo, +) => Promise; diff --git a/packages/acp-bridge/src/status.ts b/packages/acp-bridge/src/status.ts index 9cce58cd14..bf2480f743 100644 --- a/packages/acp-bridge/src/status.ts +++ b/packages/acp-bridge/src/status.ts @@ -165,6 +165,15 @@ export const SERVE_CONTROL_EXT_METHODS = { */ clientMcpMessage: 'qwen/control/client_mcp/message', sessionCd: 'qwen/control/session/cd', + /** + * Also called by the CHILD UP into the parent (like `clientMcpMessage`): the + * `create_sub_session` tool, running inside a child's agent turn, asks the + * daemon to spawn a fresh top-level sub-session and run a prompt in it. Params: + * `{ prompt, completion:'sent'|'first-turn', model?, name?, callerSessionId? }`; + * result: `{ sessionId, result?, stopReason? }` (result present only for the + * `first-turn` mode, which waits for the sub-session's first turn to finish). + */ + createSubSession: 'qwen/control/create-sub-session', } as const; export type ServeStatus = diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index 8464387679..6a3d427cc2 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -469,6 +469,8 @@ describe('Session', () => { getMonitorRegistry: vi.fn().mockReturnValue(mockMonitorRegistry), getFileHistoryService: vi.fn().mockReturnValue(mockFileHistoryService), getDisabledSkillNames: vi.fn().mockReturnValue(new Set()), + setSubSessionSpawner: vi.fn(), + getSubSessionSpawner: vi.fn(), getExtensions: vi.fn().mockReturnValue([]), } as unknown as Config; @@ -8559,6 +8561,177 @@ describe('Session', () => { ); }); + describe('isolated scheduled tasks', () => { + /** Mock scheduler that delivers exactly one job through `start`. */ + function schedulerFiring(job: { + prompt: string; + cronExpr?: string; + missed?: boolean; + runMode?: 'shared' | 'isolated'; + }) { + return { + size: 1, + hasPendingWork: true, + start: vi.fn((callback: (j: typeof job) => void) => callback(job)), + stop: vi.fn(), + getExitSummary: vi.fn().mockReturnValue(undefined), + }; + } + + it('dispatches an isolated fire into a sub-session, never into the bound session', async () => { + // The whole point of `isolated`: the fire must NOT be relayed through + // the model in this session. Relaying it would put the run behind + // create_sub_session's 'ask' permission, which an unattended fire has + // nobody to answer. + const spawner = vi.fn().mockResolvedValue({ sessionId: 'sub-abc' }); + const scheduler = schedulerFiring({ + prompt: 'nightly report', + runMode: 'isolated', + }); + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + mockConfig.getSubSessionSpawner = vi.fn().mockReturnValue(spawner); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + + await session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'hello' }], + }); + + await vi.waitFor(() => expect(spawner).toHaveBeenCalledTimes(1)); + // Raw prompt, fire-and-forget — no wrapper instructing the model. + expect(spawner).toHaveBeenCalledWith({ + prompt: 'nightly report', + completion: 'sent', + }); + // Only the user's own turn reached the model; the cron fire did not. + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(1); + }); + + it('runs an isolated fire in-session when no sub-session spawner is wired', async () => { + // Outside `qwen serve` there is no bridge to spawn into. Losing the + // task is worse than losing isolation, so the fire runs inline. + const scheduler = schedulerFiring({ + prompt: 'nightly report', + runMode: 'isolated', + }); + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + mockConfig.getSubSessionSpawner = vi.fn().mockReturnValue(undefined); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + + await session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'hello' }], + }); + + await vi.waitFor(() => + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(2), + ); + const cronCall = ( + mockChat.sendMessageStream as ReturnType + ).mock.calls[1]; + const text = (cronCall![1].message as Array<{ text?: string }>) + .map((p) => p.text ?? '') + .join(''); + expect(text).toContain('nightly report'); + }); + + it('keeps a missed isolated one-shot on the in-session confirm-first path', async () => { + const spawner = vi.fn().mockResolvedValue({ sessionId: 'sub-abc' }); + const scheduler = schedulerFiring({ + prompt: 'nightly report', + runMode: 'isolated', + missed: true, + }); + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + mockConfig.getSubSessionSpawner = vi.fn().mockReturnValue(spawner); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + + await session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'hello' }], + }); + + await vi.waitFor(() => + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(2), + ); + expect(spawner).not.toHaveBeenCalled(); + }); + + it('drops the fire (never falls back in-session) when the dispatch fails', async () => { + // A failed dispatch may already have sent the prompt, and running an + // isolated task inside the bound session would defeat the isolation the + // user asked for. Log and drop. + // + // The scheduler has already persisted this fire as a run, and + // `debugLogger.warn` writes nothing unless a debug log session is + // active — so the drop MUST also reach stderr, or a nightly task can + // fail forever while its history claims it ran. + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockReturnValue(true); + const spawner = vi.fn().mockRejectedValue(new Error('cap reached')); + const scheduler = schedulerFiring({ + prompt: 'nightly report', + runMode: 'isolated', + }); + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + mockConfig.getSubSessionSpawner = vi.fn().mockReturnValue(spawner); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + + await session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'hello' }], + }); + + await vi.waitFor(() => + expect(debugLoggerWarnSpy).toHaveBeenCalledWith( + expect.stringContaining('Isolated scheduled task dispatch failed'), + ), + ); + const stderr = stderrWrite.mock.calls.map((c) => String(c[0])).join(''); + expect(stderr).toContain('isolated scheduled task dispatch failed'); + expect(stderr).toContain('cap reached'); + stderrWrite.mockRestore(); + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(1); + }); + + it('leaves a shared fire on the in-session path', async () => { + const spawner = vi.fn().mockResolvedValue({ sessionId: 'sub-abc' }); + const scheduler = schedulerFiring({ + prompt: 'nightly report', + runMode: 'shared', + }); + mockConfig.isCronEnabled = vi.fn().mockReturnValue(true); + mockConfig.getCronScheduler = vi.fn().mockReturnValue(scheduler); + mockConfig.getSubSessionSpawner = vi.fn().mockReturnValue(spawner); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + + await session.prompt({ + sessionId: 'test-session-id', + prompt: [{ type: 'text', text: 'hello' }], + }); + + await vi.waitFor(() => + expect(mockChat.sendMessageStream).toHaveBeenCalledTimes(2), + ); + expect(spawner).not.toHaveBeenCalled(); + }); + }); + describe('hooks', () => { describe('PermissionDenied hook', () => { it('fires PermissionDenied hooks for AUTO classifier blocks', async () => { diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index b619198627..9a62cb1d5b 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -34,6 +34,7 @@ import type { LoopTickResult, ToolArtifact, VisionBridgeResult, + SubSessionSpawner, } from '@qwen-code/qwen-code-core'; import { AuthType, @@ -131,6 +132,7 @@ import { NOT_CURRENTLY_GENERATING_CANCEL_MESSAGE } from '@qwen-code/acp-bridge/b // Single source of truth shared with the daemon-side answerer (BridgeClient), // so a rename can't desync caller and answerer into a silent -32601 latch. import { MID_TURN_QUEUE_DRAIN_METHOD } from '@qwen-code/acp-bridge/bridgeTypes'; +import { SERVE_CONTROL_EXT_METHODS } from '@qwen-code/acp-bridge/status'; import { getCommandSubcommandNames } from '../../services/commandMetadata.js'; import { getEffectiveSupportedModes } from '../../services/commandUtils.js'; import { @@ -979,6 +981,47 @@ export class Session implements SessionContext { this.#installGoalTerminalObserver(); this.#registerBackgroundNotificationCallbacks(); + this.#registerSubSessionSpawner(); + } + + /** + * Wire the sub-session spawner to the daemon over the ACP `extMethod` request + * channel. Two callers: the `create_sub_session` tool (model-initiated) and + * `#dispatchIsolatedCronFire` (scheduler-initiated). ONLY the ACP/daemon + * session wires it, so the tool is inert (reports daemon-only) in interactive + * TUI / headless, where no bridge exists. + * + * A tool-initiated request runs while the caller's turn is suspended in the + * tool await — safe because the ACP channel supports concurrent bidirectional + * in-flight requests and prompts serialize per-session, not per-child. + */ + #registerSubSessionSpawner(): void { + this.config.setSubSessionSpawner(async (req) => { + const resp = await this.client.extMethod( + SERVE_CONTROL_EXT_METHODS.createSubSession, + { + prompt: req.prompt, + completion: req.completion, + ...(req.model ? { model: req.model } : {}), + ...(req.name ? { name: req.name } : {}), + callerSessionId: this.sessionId, + }, + ); + if (typeof resp['sessionId'] !== 'string' || !resp['sessionId']) { + throw new Error( + 'create_sub_session: bridge returned non-string sessionId', + ); + } + return { + sessionId: resp['sessionId'], + ...(typeof resp['result'] === 'string' + ? { result: resp['result'] } + : {}), + ...(typeof resp['stopReason'] === 'string' + ? { stopReason: resp['stopReason'] } + : {}), + }; + }); } getId(): string { @@ -1053,6 +1096,7 @@ export class Session implements SessionContext { this.config.getMonitorRegistry().setNotificationCallback(undefined); this.config.getBackgroundShellRegistry().setNotificationCallback(undefined); this.config.getChatRecordingService()?.setTitleRecordedCallback(undefined); + this.config.setSubSessionSpawner(undefined); clearGoalTerminalObserver(this.sessionId); } @@ -2837,9 +2881,32 @@ export class Session implements SessionContext { if (!scheduler.hasPendingWork) return; scheduler.start( - (job: { prompt: string; cronExpr?: string; missed?: boolean }) => { + (job: { + prompt: string; + cronExpr?: string; + missed?: boolean; + runMode?: 'shared' | 'isolated'; + }) => { if (this.cronDisabledByTokenLimit) return; if (job.missed && detectAutonomousSentinel(job.prompt)) return; + // An `isolated` task gets a FRESH sub-session per fire instead of + // accumulating in this bound session. Dispatch it straight to the + // daemon rather than asking the model to call `create_sub_session`: + // that tool's permission is 'ask', and under the default approval mode + // an unattended fire has nobody to answer the prompt — the call would + // hang until the daemon's permission timeout cancels it. A `missed` + // one-shot keeps the in-session confirm-first path. + if (!job.missed && job.runMode === 'isolated') { + const spawner = this.config.getSubSessionSpawner(); + if (spawner) { + void this.#dispatchIsolatedCronFire(spawner, job.prompt); + return; + } + // Defensive: an isolated task can only be created through the daemon's + // REST route, which binds it to a session that always has a spawner + // (and a bound task fires nowhere else). Should that ever change, run + // the fire in-session — losing the task is worse than losing isolation. + } this.cronQueue.push({ prompt: job.prompt, source: job.cronExpr === '@wakeup' ? 'loop' : 'cron', @@ -2849,6 +2916,43 @@ export class Session implements SessionContext { ); } + /** + * Runs one `isolated` scheduled fire in a fresh sub-session. Fire-and-forget: + * `'sent'` resolves once the prompt is dispatched, and the daemon-side + * launcher holds a concurrency slot until that sub-session's turn drains. + * + * A dispatch failure (concurrency cap reached, bridge gone, daemon shutting + * down) is logged and the fire is dropped. It deliberately does NOT fall back + * to an in-session run: the prompt may already have been dispatched, and + * silently running an isolated task inside the bound session would defeat the + * isolation the user asked for. + */ + async #dispatchIsolatedCronFire( + spawner: SubSessionSpawner, + prompt: string, + ): Promise { + try { + const { sessionId } = await spawner({ prompt, completion: 'sent' }); + debugLogger.info( + `Isolated scheduled task dispatched into sub-session ${sessionId} [session ${this.sessionId}]`, + ); + } catch (err) { + const detail = err instanceof Error ? err.message : String(err); + // Must reach stderr. `debugLogger.warn` writes nothing unless a debug log + // session is active, and the scheduler has already persisted this fire as + // a run — so a dropped dispatch would otherwise look like a successful + // one, with no trace anywhere. The daemon forwards child stderr. + writeStderrLine( + `qwen serve: isolated scheduled task dispatch failed — the fire was ` + + `dropped [session ${this.sessionId}]: ${detail}`, + ); + debugLogger.warn( + `Isolated scheduled task dispatch failed — the fire was dropped ` + + `[session ${this.sessionId}]: ${detail}`, + ); + } + } + /** * Processes queued cron prompts one at a time. Uses `cronProcessing` * as a mutex to prevent concurrent access to the chat. diff --git a/packages/cli/src/acp-integration/session/Session.worktree.test.ts b/packages/cli/src/acp-integration/session/Session.worktree.test.ts index a01c1e75b6..19f93ea70a 100644 --- a/packages/cli/src/acp-integration/session/Session.worktree.test.ts +++ b/packages/cli/src/acp-integration/session/Session.worktree.test.ts @@ -147,6 +147,8 @@ describe('Session.pendingWorktreeNotice', () => { getBackgroundShellRegistry: vi.fn().mockReturnValue({ setNotificationCallback: vi.fn(), }), + setSubSessionSpawner: vi.fn(), + getSubSessionSpawner: vi.fn(), } as unknown as Config; mockClient = { diff --git a/packages/cli/src/i18n/locales/en.js b/packages/cli/src/i18n/locales/en.js index 82b538c19c..120c9870c8 100644 --- a/packages/cli/src/i18n/locales/en.js +++ b/packages/cli/src/i18n/locales/en.js @@ -199,6 +199,7 @@ export default { 'toolDisplayName.CronList': 'toolDisplayName.CronList', 'toolDisplayName.CronDelete': 'toolDisplayName.CronDelete', 'toolDisplayName.LoopWakeup': 'toolDisplayName.LoopWakeup', + 'toolDisplayName.CreateSubSession': 'toolDisplayName.CreateSubSession', 'toolDisplayName.TaskCreate': 'toolDisplayName.TaskCreate', 'toolDisplayName.TaskUpdate': 'toolDisplayName.TaskUpdate', 'toolDisplayName.TaskList': 'toolDisplayName.TaskList', diff --git a/packages/cli/src/i18n/locales/zh-TW.js b/packages/cli/src/i18n/locales/zh-TW.js index 8d1bb7b212..52b499fad5 100644 --- a/packages/cli/src/i18n/locales/zh-TW.js +++ b/packages/cli/src/i18n/locales/zh-TW.js @@ -190,6 +190,7 @@ export default { 'toolDisplayName.CronList': '定時任務清單', 'toolDisplayName.CronDelete': '刪除定時任務', 'toolDisplayName.LoopWakeup': '循環喚醒', + 'toolDisplayName.CreateSubSession': '建立子會話', 'toolDisplayName.TaskCreate': '建立任務', 'toolDisplayName.TaskUpdate': '更新任務', 'toolDisplayName.TaskList': '任務列表', diff --git a/packages/cli/src/i18n/locales/zh.js b/packages/cli/src/i18n/locales/zh.js index 5c9c2b9ad3..87289f7685 100644 --- a/packages/cli/src/i18n/locales/zh.js +++ b/packages/cli/src/i18n/locales/zh.js @@ -191,6 +191,7 @@ export default { 'toolDisplayName.CronList': '定时任务列表', 'toolDisplayName.CronDelete': '删除定时任务', 'toolDisplayName.LoopWakeup': '循环唤醒', + 'toolDisplayName.CreateSubSession': '创建子会话', 'toolDisplayName.TaskCreate': '创建任务', 'toolDisplayName.TaskUpdate': '更新任务', 'toolDisplayName.TaskList': '任务列表', diff --git a/packages/cli/src/serve/create-sub-session.test.ts b/packages/cli/src/serve/create-sub-session.test.ts new file mode 100644 index 0000000000..899ac10b53 --- /dev/null +++ b/packages/cli/src/serve/create-sub-session.test.ts @@ -0,0 +1,568 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { AcpSessionBridge } from '@qwen-code/acp-bridge/bridgeTypes'; + +/** Captures the launcher's operator-facing stderr output. */ +const { stderrLines } = vi.hoisted(() => ({ stderrLines: [] as string[] })); +vi.mock('../utils/stdioHelpers.js', () => ({ + writeStderrLine: (line: string) => stderrLines.push(line), +})); + +const { + createSubSessionLauncher, + MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER, + MAX_CONCURRENT_SUB_SESSIONS_TOTAL, +} = await import('./create-sub-session.js'); + +type FakeEvent = { type: string; data: unknown }; + +const chunk = (text: string): FakeEvent => ({ + type: 'session_update', + data: { update: { sessionUpdate: 'agent_message_chunk', content: { text } } }, +}); +const turnComplete = ( + promptId: string, + stopReason = 'end_turn', +): FakeEvent => ({ + type: 'turn_complete', + data: { sessionId: '', stopReason, promptId }, +}); +const turnError = (promptId: string, message: string): FakeEvent => ({ + type: 'turn_error', + data: { sessionId: '', message, promptId }, +}); + +/** A fake bridge whose `subscribeEvents` yields a scripted stream (built from + * the captured promptId) and can optionally block until the abort signal fires + * — used to exercise the timeout and concurrency-cap paths. */ +function makeFakeBridge(opts?: { + events?: (promptId: string) => FakeEvent[]; + blockAfterEvents?: boolean; + sendPromptRejects?: string; + /** How the orphan-cleanup `closeSession` fails, if at all. A real bridge can + * throw synchronously (e.g. an unknown session id hits an assertion before + * the first await), which must not clobber the launch error. */ + closeSessionFails?: 'sync' | 'async'; +}) { + const spawns: Array<{ + workspaceCwd: string; + sessionScope?: string; + modelServiceId?: string; + }> = []; + const prompts: Array<{ sessionId: string; promptId?: string; text: string }> = + []; + const names: Array<{ sessionId: string; displayName?: string }> = []; + const closes: string[] = []; + let subscribeCalls = 0; + let capturedPromptId = ''; + let n = 0; + + const bridge = { + spawnOrAttach: async (req: { + workspaceCwd: string; + sessionScope?: 'single' | 'thread'; + modelServiceId?: string; + }) => { + spawns.push(req); + return { sessionId: `sub-${++n}` }; + }, + updateSessionMetadata: ( + sessionId: string, + metadata: { displayName?: string }, + ) => { + names.push({ sessionId, displayName: metadata.displayName }); + return metadata; + }, + getSessionLastEventId: () => 0, + sendPrompt: ( + sessionId: string, + req: { prompt: Array<{ type: string; text?: string }> }, + _signal: unknown, + ctx?: { promptId?: string }, + ) => { + capturedPromptId = ctx?.promptId ?? ''; + prompts.push({ + sessionId, + promptId: capturedPromptId, + text: req.prompt.map((p) => p.text ?? '').join(''), + }); + if (opts?.sendPromptRejects) { + return Promise.reject(new Error(opts.sendPromptRejects)); + } + // Never resolves — the first-turn result comes from the event stream. + return new Promise(() => {}); + }, + closeSession: (sessionId: string) => { + closes.push(sessionId); + if (opts?.closeSessionFails === 'sync') { + throw new Error('closeSession exploded'); + } + if (opts?.closeSessionFails === 'async') { + return Promise.reject(new Error('closeSession rejected')); + } + return Promise.resolve(); + }, + async *subscribeEvents(_sessionId: string, o?: { signal?: AbortSignal }) { + subscribeCalls++; + const evs = opts?.events ? opts.events(capturedPromptId) : []; + for (const e of evs) { + if (o?.signal?.aborted) return; + yield e; + } + if (opts?.blockAfterEvents) { + await new Promise((resolve) => { + if (o?.signal) { + o.signal.addEventListener('abort', () => resolve(), { once: true }); + } + }); + } + }, + }; + return { + bridge: bridge as unknown as AcpSessionBridge, + spawns, + prompts, + names, + closes, + subscribeCalls: () => subscribeCalls, + }; +} + +describe('sub-session launcher', () => { + const WS = '/tmp/ws'; + + beforeEach(() => { + stderrLines.length = 0; + }); + + it('sent: spawns a thread-scoped session, dispatches, returns the id (background subscribe holds slot)', async () => { + const fake = makeFakeBridge(); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + + const res = await launcher.launch({ + prompt: 'do the thing', + completion: 'sent', + name: 'my task', + callerSessionId: 'caller-1', + }); + + expect(res).toEqual({ sessionId: 'sub-1' }); + expect(fake.spawns).toEqual([{ workspaceCwd: WS, sessionScope: 'thread' }]); + expect(fake.prompts[0]!.text).toBe('do the thing'); + expect(fake.names[0]!.displayName).toContain('my task'); + // 'sent' returns immediately but starts a background subscription to hold + // the concurrency slot until the sub-session's turn finishes (so the cap + // stays meaningful). The subscription is fire-and-forget — the launch + // result is already returned before any events are consumed. + expect(fake.subscribeCalls()).toBe(1); + }); + + it('first-turn: accumulates chunk text until turn_complete and returns it', async () => { + const fake = makeFakeBridge({ + events: (pid) => [chunk('Hello '), chunk('world'), turnComplete(pid)], + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + + const res = await launcher.launch({ + prompt: 'greet', + completion: 'first-turn', + model: 'model-x', + callerSessionId: 'caller-1', + }); + + expect(res).toEqual({ + sessionId: 'sub-1', + result: 'Hello world', + stopReason: 'end_turn', + }); + // model flows through as modelServiceId on the spawn. + expect(fake.spawns[0]).toEqual({ + workspaceCwd: WS, + sessionScope: 'thread', + modelServiceId: 'model-x', + }); + expect(fake.subscribeCalls()).toBe(1); + }); + + it('first-turn: reports turn_error with the partial text and error stopReason', async () => { + const fake = makeFakeBridge({ + events: (pid) => [chunk('partial'), turnError(pid, 'model exploded')], + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + const res = await launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }); + expect(res.sessionId).toBe('sub-1'); + expect(res.stopReason).toBe('error'); + expect(res.result).toContain('partial'); + expect(res.result).toContain('model exploded'); + }); + + it('first-turn: truncates an over-long result', async () => { + const fake = makeFakeBridge({ + events: (pid) => [chunk('x'.repeat(40_000)), turnComplete(pid)], + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + const res = await launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }); + expect(res.result!.length).toBeLessThan(40_000); + expect(res.result).toContain('truncated'); + }); + + it('first-turn: times out (returns partial text + timeout stopReason)', async () => { + const fake = makeFakeBridge({ + events: () => [chunk('slow...')], + blockAfterEvents: true, + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + firstTurnTimeoutMs: 60, + }); + const res = await launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }); + expect(res.stopReason).toBe('timeout'); + expect(res.result).toContain('slow...'); + }); + + it('caps concurrent first-turn runs per caller, rejecting the overflow without spawning', async () => { + const fake = makeFakeBridge({ blockAfterEvents: true }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + firstTurnTimeoutMs: 80, // held runs settle via timeout so the test ends + }); + + const promises = []; + for (let i = 0; i < MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER + 1; i++) { + promises.push( + launcher.launch({ + prompt: `p${i}`, + completion: 'first-turn', + callerSessionId: 'same-caller', + }), + ); + } + const settled = await Promise.allSettled(promises); + const rejected = settled.filter((s) => s.status === 'rejected'); + expect(rejected).toHaveLength(1); + expect((rejected[0] as PromiseRejectedResult).reason.message).toMatch( + /cap/i, + ); + // The overflow was rejected BEFORE spawning — exactly cap sessions spawned. + expect(fake.spawns).toHaveLength(MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER); + }); + + it('rejects when the bridge is unavailable', async () => { + const launcher = createSubSessionLauncher({ + getBridge: () => undefined, + boundWorkspace: WS, + }); + await expect( + launcher.launch({ + prompt: 'x', + completion: 'sent', + callerSessionId: 'c', + }), + ).rejects.toThrow(); + }); + + it('rejects new launches after stop()', async () => { + const fake = makeFakeBridge(); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + launcher.stop(); + await expect( + launcher.launch({ + prompt: 'x', + completion: 'sent', + callerSessionId: 'c', + }), + ).rejects.toThrow(/shutting down/i); + expect(fake.spawns).toHaveLength(0); + }); + + it('first-turn: sendPrompt rejection fails fast (not after timeout)', async () => { + // blockAfterEvents keeps the subscription alive so the turnError race + // is the only way to settle — proving the rejection short-circuits the + // 5-min timeout instead of silently timing out. + const fake = makeFakeBridge({ + sendPromptRejects: 'API 429 rate limit', + events: () => [], + blockAfterEvents: true, + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + firstTurnTimeoutMs: 60_000, // would wait 1 min without the race + }); + await expect( + launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }), + ).rejects.toThrow(/dispatch failed.*API 429/i); + // The session was already spawned when the dispatch failed — close it so it + // doesn't linger in the bridge's pool while launch() reports failure. + expect(fake.closes).toEqual(['sub-1']); + }); + + it('first-turn: a throwing closeSession does not mask the launch error', async () => { + // Orphan cleanup runs inside the launcher's catch block. A synchronous + // throw there would escape and replace the real failure ('API 429') with + // the cleanup failure — the caller would be told the wrong thing. + for (const closeSessionFails of ['sync', 'async'] as const) { + const fake = makeFakeBridge({ + sendPromptRejects: 'API 429 rate limit', + events: () => [], + blockAfterEvents: true, + closeSessionFails, + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + firstTurnTimeoutMs: 60_000, + }); + await expect( + launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }), + ).rejects.toThrow(/dispatch failed.*API 429/i); + expect(fake.closes).toEqual(['sub-1']); + } + }); + + it('sent mode: holds the concurrency slot while the drain is still running', async () => { + // No turn_complete and a stream that blocks: every drain stays in flight, + // so every slot stays held. Releasing at drain *start* instead of drain + // *end* would silently admit the overflow launch below — that is exactly + // the "cap is a no-op for sent mode" bug this guards. + const fake = makeFakeBridge({ events: () => [], blockAfterEvents: true }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + for (let i = 0; i < MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER; i++) { + await launcher.launch({ + prompt: `p${i}`, + completion: 'sent', + callerSessionId: 'same-caller', + }); + } + await expect( + launcher.launch({ + prompt: 'overflow', + completion: 'sent', + callerSessionId: 'same-caller', + }), + ).rejects.toThrow(/cap/i); + // Rejected before spawning — exactly cap sessions exist. + expect(fake.spawns).toHaveLength(MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER); + launcher.stop(); // unblock the drains so the test leaves nothing pending + }); + + it('sent mode: releases the slot once the drain sees turn_complete', async () => { + // blockAfterEvents keeps the stream open past the scripted events, so the + // ONLY way a drain can end is by matching its own turn_complete promptId. + const fake = makeFakeBridge({ + events: (pid) => [turnComplete(pid)], + blockAfterEvents: true, + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + for (let i = 0; i < MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER; i++) { + await launcher.launch({ + prompt: `p${i}`, + completion: 'sent', + callerSessionId: 'same-caller', + }); + } + // Drains are fire-and-forget; let them observe turn_complete and release. + await vi.waitFor(() => + expect(fake.subscribeCalls()).toBe( + MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER, + ), + ); + await new Promise((r) => setTimeout(r, 10)); + // All slots freed — a launch beyond the cap now succeeds. + const fresh = await launcher.launch({ + prompt: 'after-drain', + completion: 'sent', + callerSessionId: 'same-caller', + }); + expect(fresh.sessionId).toBeTruthy(); + launcher.stop(); + }); + + it('first-turn: reports "incomplete" when the stream ends before the turn does', async () => { + // Bridge teardown / WS drop: the subscription ends with no turn_complete and + // no deadline passed. Reading `ac.signal.aborted` here would always say + // "timeout" — the cleanup `finally` aborts that controller unconditionally. + const fake = makeFakeBridge({ + events: () => [chunk('partial')], + blockAfterEvents: false, // stream ends on its own + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + firstTurnTimeoutMs: 60_000, // nowhere near firing + }); + const res = await launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }); + expect(res.stopReason).toBe('incomplete'); + expect(res.result).toContain('partial'); + }); + + it('sent mode: a drain timeout reaches stderr before the slot is released', async () => { + // 30 min of model compute and a bridge session went nowhere. `log.debug` is + // a no-op unless a debug log session is active, so without this the hang + // leaves no trace anywhere. + const fake = makeFakeBridge({ events: () => [], blockAfterEvents: true }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + sentModeDrainTimeoutMs: 20, + }); + await launcher.launch({ + prompt: 'x', + completion: 'sent', + callerSessionId: 'c', + }); + await vi.waitFor(() => + expect(stderrLines.some((l) => /drain timed out/i.test(l))).toBe(true), + ); + const line = stderrLines.find((l) => /drain timed out/i.test(l))!; + expect(line).toContain('sub-1'); + expect(line).toMatch(/may still be running/i); + // The slot is freed, so a fresh launch from the same caller succeeds. + await launcher.launch({ + prompt: 'y', + completion: 'sent', + callerSessionId: 'c', + }); + launcher.stop(); + }); + + it('caps concurrent sub-sessions workspace-wide, even across rotated caller ids', async () => { + // The per-caller cap trusts `callerSessionId`, which the bridge can only + // authenticate as "a session on this channel" — all of a workspace's + // sessions share one child process. A caller rotating ids never trips the + // per-caller bucket; this backstop does not depend on the id being honest. + const fake = makeFakeBridge({ events: () => [], blockAfterEvents: true }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + for (let i = 0; i < MAX_CONCURRENT_SUB_SESSIONS_TOTAL; i++) { + await launcher.launch({ + prompt: `p${i}`, + completion: 'sent', + callerSessionId: `rotated-${i}`, // a fresh bucket every time + }); + } + await expect( + launcher.launch({ + prompt: 'overflow', + completion: 'sent', + callerSessionId: 'rotated-fresh', + }), + ).rejects.toThrow(/workspace/i); + expect(fake.spawns).toHaveLength(MAX_CONCURRENT_SUB_SESSIONS_TOTAL); + launcher.stop(); + }); + + it('refuses to spawn from a session it already spawned (depth-1 gate)', async () => { + // Every daemon session wires a spawner, sub-sessions included, and each + // gets its own cap-sized bucket. Without this gate one prompt fans out 5ⁿ. + const fake = makeFakeBridge({ events: (pid) => [turnComplete(pid)] }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + }); + + const first = await launcher.launch({ + prompt: 'top level', + completion: 'sent', + callerSessionId: 'anchor', + }); + expect(first.sessionId).toBe('sub-1'); + + // 'sub-1' is now a known sub-session — it may not spawn further ones. + await expect( + launcher.launch({ + prompt: 'nested', + completion: 'sent', + callerSessionId: first.sessionId, + }), + ).rejects.toThrow(/nesting/i); + // Rejected before spawning: still exactly one session. + expect(fake.spawns).toHaveLength(1); + + // A sibling top-level caller is unaffected. + const sibling = await launcher.launch({ + prompt: 'other top level', + completion: 'sent', + callerSessionId: 'anchor-2', + }); + expect(sibling.sessionId).toBe('sub-2'); + launcher.stop(); + }); + + it('stop() mid-first-turn returns stopReason "shutdown"', async () => { + const fake = makeFakeBridge({ + events: () => [chunk('partial')], + blockAfterEvents: true, // holds until signal aborts + }); + const launcher = createSubSessionLauncher({ + getBridge: () => fake.bridge, + boundWorkspace: WS, + firstTurnTimeoutMs: 60_000, + }); + const promise = launcher.launch({ + prompt: 'x', + completion: 'first-turn', + callerSessionId: 'c', + }); + // Let the launch start and subscribe, then stop. + await new Promise((r) => setTimeout(r, 10)); + launcher.stop(); + const res = await promise; + expect(res.stopReason).toBe('shutdown'); + expect(res.result).toContain('partial'); + }); +}); diff --git a/packages/cli/src/serve/create-sub-session.ts b/packages/cli/src/serve/create-sub-session.ts new file mode 100644 index 0000000000..f00d83f2c7 --- /dev/null +++ b/packages/cli/src/serve/create-sub-session.ts @@ -0,0 +1,533 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Daemon-host handler for sub-session spawn requests. + * + * A child sends a `create-sub-session` `extMethod` request UP to the daemon (see + * `BridgeOptions.onCreateSubSession`) — either from the `create_sub_session` tool + * inside an agent turn, or from the ACP session's `isolated` scheduled-task + * dispatch. This handler spawns a FRESH top-level sub-session, runs the prompt in + * it (`spawnOrAttach` thread scope → `sendPrompt`), and RETURNS a result. + * + * Completion modes: + * - `'sent'` — dispatch the prompt and return `{ sessionId }` immediately; + * the sub-session keeps running and is idle-reaped later. + * A background event-stream subscription holds the concurrency + * slot until the turn finishes (or `stop()` aborts it), so the + * per-caller cap stays meaningful for fire-and-forget runs. + * - `'first-turn'`— subscribe to the sub-session's event stream, accumulate its + * `agent_message_chunk` text until `turn_complete`/`turn_error` + * (correlated on `promptId`), and return it. `sendPrompt`'s + * promise only carries `stopReason` (no text), so the result + * must come from the stream. `stop()` aborts the subscription + * via a composed `AbortSignal`; the sub-session's turn itself + * is NOT cancelled (sendPrompt has no abort seam) and will + * complete or be idle-reaped independently. + * + * The sub-session is fire-and-forget w.r.t. lifecycle: it is NOT kept resident, + * so once idle the bridge's reaper closes it; its transcript persists. + */ + +import { randomUUID } from 'node:crypto'; +import { + createDebugLogger, + stripTerminalControlSequences, +} from '@qwen-code/qwen-code-core'; +import type { AcpSessionBridge } from '@qwen-code/acp-bridge/bridgeTypes'; +import type { + CreateSubSessionInfo, + CreateSubSessionResult, +} from '@qwen-code/acp-bridge/bridgeOptions'; +import { writeStderrLine } from '../utils/stdioHelpers.js'; + +const log = createDebugLogger('SUB_SESSION'); + +/** Per-caller ceiling on concurrent in-flight sub-sessions. A `first-turn` + * request holds a slot until its turn finishes; parallel tool calls from one + * caller must not spawn unbounded sub-sessions. Over the cap the request is + * rejected (surfaced as the tool's error), never silently dropped. */ +export const MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER = 5; + +/** + * Ceiling on concurrent in-flight sub-sessions across ALL callers of this + * workspace's launcher. + * + * The per-caller cap is keyed on `callerSessionId`, and the daemon can only + * authenticate that id as "a session on this channel" — every session of a + * workspace shares ONE child process, so nothing at the transport can prove + * *which* of them issued the call. A child running attacker code could rotate + * ids to open a fresh bucket per launch, or charge them to a sibling. This + * bound does not depend on the id being honest: it holds whichever bucket the + * launch is charged to. + */ +export const MAX_CONCURRENT_SUB_SESSIONS_TOTAL = 20; + +/** Wall-clock ceiling for `first-turn`: a hung sub-session turn must not block + * the caller forever. On timeout we return whatever text accumulated so far. */ +const FIRST_TURN_TIMEOUT_MS = 5 * 60_000; + +/** Wall-clock ceiling for the sent-mode background drain. Generous enough for + * long-running sub-sessions but prevents a hung turn from permanently consuming + * a concurrency slot (the idle reaper may not fire if the sub-session is still + * "actively" running from the daemon's perspective). */ +const SENT_MODE_DRAIN_TIMEOUT_MS = 30 * 60_000; + +/** Cap on returned first-turn text so a runaway sub-session can't flood the + * caller's context. Excess is dropped with a truncation marker. */ +const MAX_RESULT_CHARS = 32_000; + +/** Cap on the session display name (a label, not the full prompt). */ +const MAX_NAME_LENGTH = 60; + +/** How many spawned sub-session ids the depth-1 gate remembers. Far above any + * plausible live sub-session count (`maxSessions` defaults to 20), so eviction + * only ever discards long-reaped sessions. */ +const MAX_TRACKED_SPAWNED_SESSIONS = 1024; + +export interface SubSessionLauncher { + /** The `onCreateSubSession` callback wired into the bridge. Returns a Promise + * the child's tool awaits. */ + launch(info: CreateSubSessionInfo): Promise; + /** Stop accepting new sub-sessions (daemon shutdown). Idempotent. */ + stop(): void; +} + +export interface CreateSubSessionLauncherOptions { + getBridge: () => AcpSessionBridge | undefined; + boundWorkspace: string; + /** Per-request `first-turn` wall-clock timeout; defaults to + * {@link FIRST_TURN_TIMEOUT_MS}. Exposed for tests. */ + firstTurnTimeoutMs?: number; + /** Sent-mode background-drain ceiling; defaults to + * {@link SENT_MODE_DRAIN_TIMEOUT_MS}. Exposed for tests. */ + sentModeDrainTimeoutMs?: number; +} + +/** A readable, control-char-free session name (the bridge's title guard rejects + * control chars, silently dropping an unsanitized rename). Prefixed with a + * thread glyph so sub-sessions are recognizable in the list. */ +// Unicode Bidi_Control marks — ALM (U+061C), LRM/RLM (U+200E/200F), the +// embedding/override set (U+202A..U+202E), and the isolates (U+2066..U+2069): a +// Trojan-Source-style reordering defense for the session list, mirroring the +// scheduled-task session namer. Built from a string (not a literal regex) so no +// invisible control chars appear in the source. +const BIDI_CONTROL_MARKS = new RegExp( + '[\\u061C\\u200E\\u200F\\u202A-\\u202E\\u2066-\\u2069]', + 'g', +); + +function subSessionName(label: string): string { + const cleaned = stripTerminalControlSequences(label) + .replace(BIDI_CONTROL_MARKS, '') + .trim() + .replace(/\s+/g, ' '); + let short = cleaned; + if (cleaned.length > MAX_NAME_LENGTH) { + let cut = MAX_NAME_LENGTH - 1; + const boundary = cleaned.charCodeAt(cut - 1); + if (boundary >= 0xd800 && boundary <= 0xdbff) cut -= 1; + short = `${cleaned.slice(0, cut)}…`; + } + return `🧵 ${short}`; +} + +/** Accumulate the sub-session's first-turn text from its event stream, stopping + * at `turn_complete`/`turn_error` for `promptId` (or a wall-clock timeout, or + * an external shutdown signal from `stop()`). */ +async function awaitFirstTurn( + bridge: AcpSessionBridge, + sessionId: string, + promptId: string, + lastEventId: number, + timeoutMs: number, + stopSignal?: AbortSignal, +): Promise<{ result: string; stopReason: string }> { + const ac = new AbortController(); + // `ac.signal.aborted` cannot report whether the deadline passed: the `finally` + // below aborts unconditionally to tear the subscription down, so by the time + // the stopReason is computed the signal is always aborted. Record the timer + // firing separately, or a stream that closes early (bridge teardown, WS drop) + // is misreported as a 5-minute wall-clock timeout. + let timedOut = false; + const timer = setTimeout(() => { + timedOut = true; + ac.abort(); + }, timeoutMs); + if (typeof timer.unref === 'function') timer.unref(); + // Compose: the subscription ends on timeout OR daemon shutdown, whichever + // fires first. Without this, stop() cannot interrupt a first-turn await and + // shutdown hangs for up to timeoutMs (5 min default). + const composed = stopSignal + ? AbortSignal.any([ac.signal, stopSignal]) + : ac.signal; + + let acc = ''; + let truncated = false; + let stopReason: string | undefined; + + const appendChunk = (text: string): void => { + if (truncated) return; + if (acc.length + text.length > MAX_RESULT_CHARS) { + // Surrogate-pair-safe: if the cut lands on a high surrogate, back up + // one code unit so we don't emit a lone leading surrogate. + let cut = Math.max(0, MAX_RESULT_CHARS - acc.length); + if (cut > 0) { + const code = text.charCodeAt(cut - 1); + if (code >= 0xd800 && code <= 0xdbff) cut -= 1; + } + acc += text.slice(0, cut); + truncated = true; + } else { + acc += text; + } + }; + + try { + for await (const e of bridge.subscribeEvents(sessionId, { + lastEventId, + signal: composed, + })) { + if (e.type === 'session_update') { + const d = e.data as { + update?: { sessionUpdate?: string; content?: { text?: string } }; + }; + if ( + d?.update?.sessionUpdate === 'agent_message_chunk' && + typeof d.update.content?.text === 'string' + ) { + appendChunk(d.update.content.text); + } + } else if (e.type === 'turn_complete') { + const d = e.data as { promptId?: string; stopReason?: string }; + if (d?.promptId === promptId) { + stopReason = d.stopReason ?? 'end_turn'; + break; + } + } else if (e.type === 'turn_error') { + const d = e.data as { promptId?: string; message?: string }; + if (d?.promptId === promptId) { + stopReason = 'error'; + if (d.message && !truncated) { + const suffix = `${acc ? '\n' : ''}[turn error] ${d.message}`; + if (acc.length + suffix.length <= MAX_RESULT_CHARS) { + acc += suffix; + } else { + truncated = true; + } + } + break; + } + } + } + } finally { + clearTimeout(timer); + ac.abort(); // tear down the subscription on any exit + } + + if (stopReason === undefined) { + // Distinguish shutdown (stop() called) from timeout from bus-closure so the + // caller can tell the difference between "daemon is going away", "the + // sub-session turn didn't finish in time", and "the event stream ended + // before the turn did". + stopReason = stopSignal?.aborted + ? 'shutdown' + : timedOut + ? 'timeout' + : 'incomplete'; + } + if (truncated) acc += '\n[…output truncated]'; + return { result: acc, stopReason }; +} + +export function createSubSessionLauncher( + opts: CreateSubSessionLauncherOptions, +): SubSessionLauncher { + const { getBridge, boundWorkspace } = opts; + const firstTurnTimeoutMs = opts.firstTurnTimeoutMs ?? FIRST_TURN_TIMEOUT_MS; + const sentModeDrainTimeoutMs = + opts.sentModeDrainTimeoutMs ?? SENT_MODE_DRAIN_TIMEOUT_MS; + const inflight = new Map(); + // Ids of the sub-sessions this launcher spawned — the depth-1 gate reads it. + // Insertion-ordered and evicted FIFO past the cap so a long-lived daemon + // can't accumulate ids forever; an evicted id belongs to a sub-session old + // enough to have been idle-reaped long ago. + const spawnedSessionIds = new Set(); + // Shared AbortController — stop() aborts it, tearing down every active + // subscription (first-turn awaits AND sent-mode background drains). This + // prevents shutdown from waiting up to 5 min per in-flight session. + const stopAc = new AbortController(); + + // Sum of `inflight`, tracked separately so the workspace-wide cap holds even + // when a caller opens a fresh bucket per launch. + let inflightTotal = 0; + + const release = (key: string): void => { + const n = (inflight.get(key) ?? 1) - 1; + if (n <= 0) inflight.delete(key); + else inflight.set(key, n); + inflightTotal = Math.max(0, inflightTotal - 1); + }; + + const rememberSpawned = (sessionId: string): void => { + spawnedSessionIds.add(sessionId); + while (spawnedSessionIds.size > MAX_TRACKED_SPAWNED_SESSIONS) { + const oldest = spawnedSessionIds.values().next().value; + if (oldest === undefined) break; + spawnedSessionIds.delete(oldest); + } + }; + + const launch = async ( + info: CreateSubSessionInfo, + ): Promise => { + if (stopAc.signal.aborted) { + throw new Error( + 'The daemon is shutting down; cannot create a sub-session.', + ); + } + const bridge = getBridge(); + if (!bridge) { + throw new Error('Session bridge is not available.'); + } + + // Depth-1 gate. Every daemon session wires a spawner, sub-sessions included, + // and each gets its own 5-slot bucket — so without this a sub-session can + // spawn 5 more, each of which spawns 5 more (5ⁿ), exhausting `maxSessions` + // from one prompt. `callerSessionId` is required and authenticated at the + // bridge (`ownsSession`), so it can neither be forged nor omitted to + // sidestep this gate or the per-caller cap below. + if (spawnedSessionIds.has(info.callerSessionId)) { + throw new Error( + 'A sub-session cannot create further sub-sessions (nesting is capped ' + + 'at one level).', + ); + } + + // Per-caller concurrency key. Always a real, bridge-authenticated session + // id: an anonymous fallback (a per-launch UUID) would give every call its + // own bucket, which is the same as having no cap at all. + const key = info.callerSessionId; + const current = inflight.get(key) ?? 0; + if (current >= MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER) { + throw new Error( + `Too many concurrent sub-sessions for this session ` + + `(cap ${MAX_CONCURRENT_SUB_SESSIONS_PER_CALLER}); wait for one to finish.`, + ); + } + // Forge-proof backstop: the per-caller cap above trusts `callerSessionId`, + // this one does not. See MAX_CONCURRENT_SUB_SESSIONS_TOTAL. + if (inflightTotal >= MAX_CONCURRENT_SUB_SESSIONS_TOTAL) { + throw new Error( + `Too many concurrent sub-sessions in this workspace ` + + `(cap ${MAX_CONCURRENT_SUB_SESSIONS_TOTAL}); wait for one to finish.`, + ); + } + inflight.set(key, current + 1); + inflightTotal += 1; + // Per-acquire idempotent release: prevents double-free when an error + // propagates through both the inner finally (first-turn path) and the + // outer catch. Without this, each failure loosens the cap by one slot; + // repeated failures drive the counter below the real in-flight count + // and over-admit concurrent sub-sessions past the documented cap. + let released = false; + const releaseOnce = (): void => { + if (released) return; + released = true; + release(key); + }; + // Set after a successful spawnOrAttach; if a later step fails the launch + // we close this session so it isn't orphaned (the slot was consumed and + // the prompt may have been dispatched, but launch() reports failure). + let spawnedSessionId: string | undefined; + + try { + const sub = await bridge.spawnOrAttach({ + workspaceCwd: boundWorkspace, + sessionScope: 'thread', // force a fresh top-level session, never attach + ...(info.model ? { modelServiceId: info.model } : {}), + }); + spawnedSessionId = sub.sessionId; + const sessionId = sub.sessionId; + rememberSpawned(sessionId); + + try { + bridge.updateSessionMetadata(sessionId, { + displayName: subSessionName(info.name ?? info.prompt), + }); + } catch (err) { + log.debug('sub-session: updateSessionMetadata failed', sessionId, err); + } + + // Capture the event cursor BEFORE dispatching so subscriptions can replay + // every chunk of the turn (no early-chunk loss). Called unconditionally + // — even sent mode needs it for the background drain that holds the + // concurrency slot; hardcoding 0 would work on a fresh bus but is + // load-bearing and subtle, so always ask the bridge. + const lastEventId = bridge.getSessionLastEventId(sessionId); + + const promptId = randomUUID(); + const turn = bridge.sendPrompt( + sessionId, + { + sessionId, + prompt: [{ type: 'text', text: info.prompt }], + } as Parameters[1], + undefined, + { promptId }, + ); + // The result comes from the event stream (turn_error surfaces failures); + // swallow the promise so it can't raise an unhandled rejection, but log + // the error so dispatch failures are not invisible. + void turn.catch((err) => { + log.debug('sub-session: sendPrompt rejected', sessionId, String(err)); + }); + + if (info.completion === 'sent') { + // Hold the concurrency slot until the sub-session's turn finishes + // (or the daemon shuts down via stop(), or a wall-clock ceiling is + // reached). Without this the cap is a no-op for sent mode — the + // fire-and-forget path returns immediately and the slot releases + // before the sub-session has done any work, letting a looping + // isolated task exhaust the daemon's session pool. + const drainAc = new AbortController(); + // Recorded in the timer, not read off `drainAc.signal.aborted`: the + // `finally` below aborts that controller on every exit path, so the + // signal cannot tell a 30-minute hang from a clean drain. + let drainTimedOut = false; + const drainTimer = setTimeout(() => { + drainTimedOut = true; + drainAc.abort(); + }, sentModeDrainTimeoutMs); + if (typeof drainTimer.unref === 'function') drainTimer.unref(); + const drainSignal = AbortSignal.any([stopAc.signal, drainAc.signal]); + void (async () => { + try { + // Race the turn promise against the drain: if sendPrompt rejects + // (API 429, network timeout), the turn will never emit + // turn_complete/turn_error, so abort the drain immediately. + const turnSettled = turn.then( + () => 'ok' as const, + () => 'rejected' as const, + ); + const drainDone = (async () => { + for await (const e of bridge.subscribeEvents(sessionId, { + lastEventId, + signal: drainSignal, + })) { + if (e.type === 'turn_complete' || e.type === 'turn_error') { + const d = e.data as { promptId?: string }; + if (d?.promptId === promptId) break; + } + } + return 'drained' as const; + })(); + await Promise.race([turnSettled, drainDone]); + } catch (err) { + // AbortError from stop()/timeout or bus closure is expected. + // Other errors (bus corruption, internal bridge failures) are + // real and should surface — don't silently swallow them. + if ( + !(err instanceof Error && err.name === 'AbortError') && + !(stopAc.signal.aborted || drainAc.signal.aborted) + ) { + log.debug( + 'sub-session: sent-mode drain error', + sessionId, + String(err), + ); + } + } finally { + clearTimeout(drainTimer); + if (drainTimedOut) { + // The slot is about to be freed while the sub-session is very + // likely still running (`sendPrompt` has no abort seam), so it + // keeps burning a bridge session and model quota with nobody + // watching. `log.debug` is a no-op unless a debug log session is + // active — this has to reach stderr or it leaves no trace at all. + writeStderrLine( + `qwen serve: sub-session ${sessionId} drain timed out after ` + + `${Math.round(sentModeDrainTimeoutMs / 60_000)}min; releasing its ` + + `concurrency slot (the sub-session may still be running)`, + ); + } + drainAc.abort(); + // Use releaseOnce (not raw release) — if spawn succeeded but the + // outer catch also fires release, using raw release would double- + // free the slot. + releaseOnce(); + } + })(); + return { sessionId }; + } + + // first-turn: hold the slot synchronously until the turn completes. + // stopAc.signal is composed inside awaitFirstTurn so stop() aborts + // the subscription (stopReason: 'shutdown'). Also race against the + // sendPrompt promise — if it rejects (API 429, network timeout, auth + // failure), turn_complete/turn_error never fire and the caller would + // otherwise wait the full timeout. + try { + const turnError: Promise = turn.then( + () => new Promise(() => {}), // never resolves on success + (err) => + Promise.reject( + new Error( + `sub-session dispatch failed: ${err instanceof Error ? err.message : String(err)}`, + ), + ), + ); + const firstTurn = awaitFirstTurn( + bridge, + sessionId, + promptId, + lastEventId, + firstTurnTimeoutMs, + stopAc.signal, + ); + const { result, stopReason } = await Promise.race([ + firstTurn, + turnError, + ]); + return { sessionId, result, stopReason }; + } finally { + releaseOnce(); + } + } catch (err) { + // Spawn/admission failure — surface it as the tool's error. + releaseOnce(); + // If the spawn succeeded but a later step failed (e.g. sendPrompt threw + // synchronously), close the orphaned session so it doesn't leak a slot + // in the bridge's session pool while this launch reports failure. + if (spawnedSessionId !== undefined) { + // Both guards are load-bearing. `.catch()` swallows the async + // rejection; the try/catch contains a SYNCHRONOUS throw. We are already + // inside the catch block, so an escaping throw here would replace `err` + // — the real launch failure — with the cleanup failure. + try { + void bridge.closeSession(spawnedSessionId).catch(() => {}); + } catch (closeErr) { + log.debug( + 'sub-session: closeSession threw', + spawnedSessionId, + closeErr, + ); + } + } + writeStderrLine( + `qwen serve: create_sub_session failed: ${err instanceof Error ? err.message : String(err)}`, + ); + throw err instanceof Error ? err : new Error(String(err)); + } + }; + + return { + launch, + stop: () => { + stopAc.abort(); // tears down every active subscription → releases slots + }, + }; +} diff --git a/packages/cli/src/serve/routes/scheduled-tasks.test.ts b/packages/cli/src/serve/routes/scheduled-tasks.test.ts index c5da3f7071..b0dadefc45 100644 --- a/packages/cli/src/serve/routes/scheduled-tasks.test.ts +++ b/packages/cli/src/serve/routes/scheduled-tasks.test.ts @@ -148,6 +148,35 @@ describe('scheduled-tasks routes', () => { const list = await request(h.app).get('/scheduled-tasks'); expect(list.body.tasks).toHaveLength(1); expect(list.body.tasks[0].id).toBe(res.body.id); + // runMode defaults to 'shared' when omitted (the #6389 model). + expect(res.body.runMode).toBe('shared'); + expect(list.body.tasks[0].runMode).toBe('shared'); + }); + + it('creates an isolated task, persisting runMode and still minting an anchor', async () => { + const res = await create({ + cron: '0 9 * * *', + prompt: 'p', + runMode: 'isolated', + }); + expect(res.status).toBe(201); + expect(res.body.runMode).toBe('isolated'); + // The anchor (ticker) session is still minted for an isolated task — it is + // what fires the cron; the runs happen in fresh siblings the daemon spawns. + expect(h.bridge.spawned).toHaveLength(1); + expect(res.body.sessionId).toBe(h.bridge.spawned[0]); + const list = await request(h.app).get('/scheduled-tasks'); + expect(list.body.tasks[0].runMode).toBe('isolated'); + }); + + it('rejects an invalid runMode on create', async () => { + const res = await create({ + cron: '0 9 * * *', + prompt: 'p', + runMode: 'per-run', + }); + expect(res.status).toBe(400); + expect(res.body.code).toBe('invalid_run_mode'); }); it('binds a created task to a freshly minted session', async () => { @@ -292,6 +321,36 @@ describe('scheduled-tasks routes', () => { expect(list.body.tasks[0].enabled).toBe(false); }); + it('switches runMode via PATCH, and rejects an invalid value', async () => { + const created = await create({ cron: '0 9 * * *', prompt: 'x' }); + const id = created.body.id as string; + expect(created.body.runMode).toBe('shared'); + + const patch = await request(h.app) + .patch(`/scheduled-tasks/${id}`) + .send({ runMode: 'isolated' }); + expect(patch.status).toBe(200); + expect(patch.body.runMode).toBe('isolated'); + const list = await request(h.app).get('/scheduled-tasks'); + expect(list.body.tasks[0].runMode).toBe('isolated'); + + // Reverse direction: isolated → shared clears the on-disk field (regression + // for PATCH runMode=undefined spread-overwrite subtlety). + const revert = await request(h.app) + .patch(`/scheduled-tasks/${id}`) + .send({ runMode: 'shared' }); + expect(revert.status).toBe(200); + expect(revert.body.runMode).toBe('shared'); + const listAfterRevert = await request(h.app).get('/scheduled-tasks'); + expect(listAfterRevert.body.tasks[0].runMode).toBe('shared'); + + const bad = await request(h.app) + .patch(`/scheduled-tasks/${id}`) + .send({ runMode: 'per-run' }); + expect(bad.status).toBe(400); + expect(bad.body.code).toBe('invalid_run_mode'); + }); + it('clears the name when patched to an empty string', async () => { const created = await create({ name: 'Named', diff --git a/packages/cli/src/serve/routes/scheduled-tasks.ts b/packages/cli/src/serve/routes/scheduled-tasks.ts index 9a99d781eb..fb0c0f6d74 100644 --- a/packages/cli/src/serve/routes/scheduled-tasks.ts +++ b/packages/cli/src/serve/routes/scheduled-tasks.ts @@ -125,6 +125,7 @@ interface ScheduledTaskView { lastFiredAt: number | null; nextRunAt: number | null; sessionId: string | null; + runMode: 'shared' | 'isolated'; runs: CronTaskRun[]; } @@ -160,6 +161,9 @@ function toView(task: DurableCronTask): ScheduledTaskView { typeof task.sessionId === 'string' && task.sessionId.length > 0 ? task.sessionId : null, + // Absent runMode normalizes to 'shared' so the client never special-cases + // undefined (tool-created / legacy tasks omit it). + runMode: task.runMode === 'isolated' ? 'isolated' : 'shared', // Absent runs (tool-created / never-fired) normalizes to [] so the client // never special-cases undefined. runs: Array.isArray(task.runs) ? task.runs : [], @@ -297,8 +301,20 @@ export function registerScheduledTasksRoutes( }); return; } + if ( + body['runMode'] !== undefined && + body['runMode'] !== 'shared' && + body['runMode'] !== 'isolated' + ) { + res.status(400).json({ + error: "`runMode` must be 'shared' or 'isolated'", + code: 'invalid_run_mode', + }); + return; + } const recurring = body['recurring'] !== false; const enabled = body['enabled'] !== false; + const runMode = body['runMode'] === 'isolated' ? 'isolated' : 'shared'; // Mint the task's dedicated session up front. The task is BOUND to it and // fires only inside it — its transcript becomes the task's run history, and @@ -372,6 +388,8 @@ export function registerScheduledTasksRoutes( enabled, ...(boundSessionId !== undefined ? { sessionId: boundSessionId } : {}), ...(nameResult.value !== undefined ? { name: nameResult.value } : {}), + // Omit when 'shared' so tool-created / default tasks stay byte-identical. + ...(runMode === 'isolated' ? { runMode } : {}), }; // Best-effort teardown of the just-minted session when the create can't be @@ -500,6 +518,21 @@ export function registerScheduledTasksRoutes( } patch.enabled = body['enabled']; } + if ('runMode' in body) { + if (body['runMode'] !== 'shared' && body['runMode'] !== 'isolated') { + res.status(400).json({ + error: "`runMode` must be 'shared' or 'isolated'", + code: 'invalid_run_mode', + }); + return; + } + // Switching mode only changes fire behavior; the child scheduler picks up + // the new runMode on its next file-watch reload (via durableTaskToJob). No + // anchor re-seat needed — the schedule itself is unchanged. Setting to + // 'shared' explicitly clears the field (same on-disk representation as + // legacy / default tasks — absent means 'shared'). + patch.runMode = body['runMode'] === 'isolated' ? 'isolated' : undefined; + } if (Object.keys(patch).length === 0 && !clearName) { res.status(400).json({ diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index 15a00e7efe..d060ccc693 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -4544,6 +4544,8 @@ describe('runQwenServe channel worker supervisor', () => { const listenError = new Error('listen failed') as NodeJS.ErrnoException; listenError.code = 'EADDRINUSE'; vi.spyOn(serverModule, 'createServeApp').mockReturnValue({ + // A real express app always exposes `locals`; the runtime parks the + // scheduled-task keepalive/launcher stoppers there, so the stub needs it. locals: {}, listen: vi.fn(() => { const srv = createServer(); diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 905df849ea..3fb8a3cf7a 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -41,6 +41,9 @@ import type { TelemetrySettings, } from '@qwen-code/qwen-code-core'; import { createBridgeFileSystemAdapter } from './bridge-file-system-adapter.js'; +// Dynamic-imported below (not at module scope) so the serve fast-path bundle +// closure check doesn't trace create-sub-session's transitive deps through +// the run-qwen-serve chunk. The launcher is only needed after listen(). import { PathMutexRegistry } from './fs/path-mutex-registry.js'; import { isLoopbackBind } from './loopback-binds.js'; import { RUNTIME_STARTUP_CANCELLED_MESSAGE } from './runtime-startup-errors.js'; @@ -2536,12 +2539,27 @@ export async function runQwenServe( ); } }); + // `create_sub_session` tool: spawn a fresh top-level sub-session on request + // from a child's agent turn and (for 'first-turn') return its result. + // Dynamic-imported (not at module scope) so the serve fast-path bundle + // closure check doesn't trace create-sub-session's transitive deps. + const { createSubSessionLauncher } = await import( + './create-sub-session.js' + ); + // Late-binds the bridge (constructed just below) via `() => bridgeRef`. Only + // wired on the daemon-created bridge — an injected `deps.bridge` (embed/test) + // brings its own options. + const subSessionLauncher = createSubSessionLauncher({ + getBridge: () => bridgeRef, + boundWorkspace, + }); const bridge = deps.bridge ?? runtime.createAcpSessionBridge({ // Reverse tool channel: let `BridgeClient.extMethod` reach the WS // connection that hosts a named client MCP server (#5626). clientMcpSender: clientMcpSenderRegistry.lookup, + onCreateSubSession: subSessionLauncher.launch, maxSessions: opts.maxSessions, freshSessionAdmission: totalSessionAdmission.admit, sessionLifecycle: sessionOwnerIndex.handleBridgeSessionLifecycle, @@ -2747,6 +2765,11 @@ export async function runQwenServe( }; }; + // Collects stop() callbacks from every per-workspace sub-session launcher + // (primary + secondaries). Called during shutdown so no new sub-sessions + // are admitted while bridges are being torn down. + const subSessionStoppers: Array<() => void> = []; + for (const workspaceInput of workspaceInputs.slice(1)) { let secondarySettings: | ReturnType @@ -2809,8 +2832,20 @@ export async function runQwenServe( : {}), }); const secondaryClientMcpSenderRegistry = new ClientMcpSenderRegistry(); + // Wire sub-session support for the secondary workspace too — without + // this, isolated scheduled tasks and create_sub_session calls from + // sessions bound to a secondary workspace hit methodNotFound. + // eslint-disable-next-line prefer-const -- assigned once after bridge creation; `let` required because the launcher closure captures it before the assignment. + let secondaryBridgeRef: + | ReturnType + | undefined; + const secondarySubSessionLauncher = createSubSessionLauncher({ + getBridge: () => secondaryBridgeRef, + boundWorkspace: workspaceInput.cwd, + }); const secondaryBridge = runtime.createAcpSessionBridge({ clientMcpSender: secondaryClientMcpSenderRegistry.lookup, + onCreateSubSession: secondarySubSessionLauncher.launch, maxSessions: opts.maxSessions, freshSessionAdmission: totalSessionAdmission.admit, sessionLifecycle: sessionOwnerIndex.handleBridgeSessionLifecycle, @@ -2856,8 +2891,10 @@ export async function runQwenServe( fresh.setValue(WORKSPACE_SETTING_SCOPE, 'tools.approvalMode', mode); }), }); + secondaryBridgeRef = secondaryBridge; runtimeBridges.push(secondaryBridge); internalRuntimeBridgesForCleanup.push(secondaryBridge); + subSessionStoppers.push(secondarySubSessionLauncher.stop); const secondaryWorkspaceService = runtime.createDaemonWorkspaceService({ boundWorkspace: workspaceInput.cwd, contextFilename: contextFilenameForInit ?? 'QWEN.md', @@ -3192,6 +3229,14 @@ export async function runQwenServe( }, ), }); + // Park the sub-session launcher's stop on app.locals so the close handler + // can flip it off before tearing down the bridge it spawns into (symmetric + // with stopScheduledTaskKeepalive). Defensive: a launch during drain would + // otherwise just fail its spawnOrAttach against the shutting-down bridge. + ( + app.locals as { subSessionStoppers?: Array<() => void> } + ).subSessionStoppers = subSessionStoppers; + subSessionStoppers.push(subSessionLauncher.stop); return { app, bridge }; }; @@ -3704,6 +3749,15 @@ export async function runQwenServe( ( app.locals as { stopScheduledTaskKeepalive?: () => void } ).stopScheduledTaskKeepalive?.(); + // Same rationale for the create_sub_session launchers: stop accepting + // new sub-session spawns before the bridges are torn down. Calls + // every workspace's launcher stop (primary + secondaries). + const stoppers = ( + app.locals as { subSessionStoppers?: Array<() => void> } + ).subSessionStoppers; + if (stoppers) { + for (const stop of stoppers) stop(); + } clearRuntimeStartAfterHealthTimer(); clearRuntimeStartFallbackTimer(); cancelDeferredRuntimeStartup(); diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index e26e098684..7f65edd781 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -1442,6 +1442,38 @@ function resolveCronRecurringMaxAgeDays(setting: number | undefined): number { return normalizeRecurringMaxAge(raw, DEFAULT_RECURRING_MAX_AGE_DAYS); } +/** Request from the `create_sub_session` tool to spawn a fresh top-level + * sub-session and run a prompt in it. */ +export interface SubSessionSpawnRequest { + prompt: string; + /** `'sent'` = resolve as soon as the prompt is dispatched; `'first-turn'` = + * resolve after the sub-session's first turn finishes (result returned). */ + completion: 'sent' | 'first-turn'; + /** Optional model service id for the sub-session. */ + model?: string; + /** Optional display name for the sub-session in the session list. */ + name?: string; +} + +/** Result returned to the `create_sub_session` tool. `result` (the sub-session's + * first-turn output) is present only for `completion: 'first-turn'`. */ +export interface SubSessionSpawnResult { + sessionId: string; + result?: string; + stopReason?: string; +} + +/** + * Injected capability that spawns a sub-session. Used by the `create_sub_session` + * tool and by the ACP session's `isolated` scheduled-task dispatch. Wired ONLY by + * the daemon/ACP session layer (`Session.ts` → `this.client.extMethod`); absent in + * interactive TUI / headless (no bridge), which is precisely the tool's + * daemon-only gate. + */ +export type SubSessionSpawner = ( + req: SubSessionSpawnRequest, +) => Promise; + export class Config { private sessionId: string; private sessionData?: ResumedSessionData; @@ -6398,6 +6430,18 @@ export class Config { }); } + // create_sub_session: spawn a fresh top-level sub-session and run a prompt + // in it. Only functional under `qwen serve` (needs the bridge, wired as a + // spawner by the ACP session); the tool's execute() reports a clear + // daemon-only error otherwise. Registered unconditionally so the message is + // available rather than the tool silently missing. + await registerLazy(ToolNames.CREATE_SUB_SESSION, async () => { + const { CreateSubSessionTool } = await import( + '../tools/create-sub-session.js' + ); + return new CreateSubSessionTool(this); + }); + // Register team collaboration tools (experimental). The team-specific // tools (team_create/team_delete/task_create/task_update/task_list) // are gated on this flag. @@ -6536,4 +6580,21 @@ export class Config { // Pre-init path: stash for `createToolRegistry` to consume. this.pendingMcpBudgetCallback = cb; } + + private subSessionSpawner?: SubSessionSpawner; + + /** + * Wire the sub-session spawner used by the `create_sub_session` tool. Set by + * the daemon/ACP session layer (which routes it to the bridge over + * `extMethod`); left unset in interactive TUI / headless — the tool then + * reports itself as daemon-only. `undefined` clears it on session teardown. + */ + setSubSessionSpawner(spawner: SubSessionSpawner | undefined): void { + this.subSessionSpawner = spawner; + } + + /** The injected sub-session spawner, or undefined outside daemon mode. */ + getSubSessionSpawner(): SubSessionSpawner | undefined { + return this.subSessionSpawner; + } } diff --git a/packages/core/src/permissions/permission-manager.ts b/packages/core/src/permissions/permission-manager.ts index c4f07e9b75..c71d63891e 100644 --- a/packages/core/src/permissions/permission-manager.ts +++ b/packages/core/src/permissions/permission-manager.ts @@ -566,6 +566,7 @@ export class PermissionManager { 'cron_list', 'cron_delete', 'loop_wakeup', + 'create_sub_session', 'monitor', ]); diff --git a/packages/core/src/services/cronScheduler.test.ts b/packages/core/src/services/cronScheduler.test.ts index 35052f35d7..d50d377b38 100644 --- a/packages/core/src/services/cronScheduler.test.ts +++ b/packages/core/src/services/cronScheduler.test.ts @@ -1191,6 +1191,35 @@ describe('CronScheduler', () => { }); }); + it('delivers runMode on the job so the Session layer can route isolated fires', async () => { + // Isolated tasks fire through the same onFire channel as any other durable + // task; the runMode ride-along is what lets Session.onFire dispatch the + // prompt into a fresh sub-session instead of running it in the bound + // session. The scheduler itself persists a run normally — no special + // isolated persist path. + await writeCronTasks(tmpDir, [ + { ...diskTask('iso1'), runMode: 'isolated' }, + ]); + await scheduler.enableDurable('session-1'); + const fired: CronJob[] = []; + scheduler.start((job) => fired.push(job)); + + scheduler.tick(new Date(2025, 0, 15, 10, 30, 59)); + const firstMinute = new Date(2025, 0, 15, 10, 30, 0).getTime(); + + expect(fired).toHaveLength(1); + expect(fired[0]!.runMode).toBe('isolated'); + + await vi.waitFor(async () => { + const task = (await readCronTasks(tmpDir))[0]!; + expect(task.lastFiredAt).toBe(firstMinute); + // Records a run like any durable fire (attributed to this session). + expect(task.runs).toEqual([ + { at: firstMinute, kind: 'scheduled', sessionId: 'session-1' }, + ]); + }); + }); + // Settle + tear down a second scheduler sharing this tmpDir, so its // fire-and-forget writes don't race the afterEach rm. async function settle(s: CronScheduler): Promise { diff --git a/packages/core/src/services/cronScheduler.ts b/packages/core/src/services/cronScheduler.ts index 11820d325a..dba7cafdb9 100644 --- a/packages/core/src/services/cronScheduler.ts +++ b/packages/core/src/services/cronScheduler.ts @@ -87,6 +87,14 @@ export interface CronJob { * absent, the task uses the shared model: only the lock owner fires it. */ boundSessionId?: string; + /** + * How a durable fire runs. Carried from the task so `onFire` can branch: + * `'isolated'` dispatches the fired prompt into a fresh sub-session instead + * of running it in the bound session. Absent/`'shared'` runs in-session (the + * #6389 model). The scheduler itself treats both identically — it only ferries + * the field. See {@link DurableCronTask.runMode}. + */ + runMode?: 'shared' | 'isolated'; /** One-shot that was due while no owning session ran — fired late. */ missed?: boolean; } @@ -1504,6 +1512,7 @@ function durableTaskToJob( jitterMs, durable: true, ...(task.sessionId ? { boundSessionId: task.sessionId } : {}), + ...(task.runMode ? { runMode: task.runMode } : {}), }; } @@ -1516,6 +1525,7 @@ function jobToDurableTask(job: CronJob): DurableCronTask { createdAt: job.createdAt, lastFiredAt: job.lastFiredAt ?? null, ...(job.boundSessionId ? { sessionId: job.boundSessionId } : {}), + ...(job.runMode ? { runMode: job.runMode } : {}), }; } diff --git a/packages/core/src/services/cronTasksFile.test.ts b/packages/core/src/services/cronTasksFile.test.ts index e3a268fc2a..76520c8cbd 100644 --- a/packages/core/src/services/cronTasksFile.test.ts +++ b/packages/core/src/services/cronTasksFile.test.ts @@ -170,6 +170,31 @@ describe('cronTasksFile', () => { await expect(readCronTasks(tmpDir)).rejects.toThrow(/Invalid task entry/); }); + it('round-trips the optional runMode field', async () => { + const isolated = makeTask({ id: 'iso', runMode: 'isolated' }); + const shared = makeTask({ id: 'sh', runMode: 'shared' }); + await writeCronTasks(tmpDir, [isolated, shared]); + const result = await readCronTasks(tmpDir); + expect(result).toEqual([isolated, shared]); + }); + + it('accepts legacy tasks with no runMode field', async () => { + const legacy = makeTask(); + await seedTasksFile(tmpDir, JSON.stringify([legacy])); + const result = await readCronTasks(tmpDir); + expect(result[0]!.runMode).toBeUndefined(); + }); + + it('rejects a task whose runMode is an unknown string', async () => { + // A typo must route through fix-or-delete rather than being silently + // treated as 'shared' — otherwise per-run isolation would quietly turn off. + await seedTasksFile( + tmpDir, + JSON.stringify([{ ...makeTask(), runMode: 'per-run' }]), + ); + await expect(readCronTasks(tmpDir)).rejects.toThrow(/Invalid task entry/); + }); + it('round-trips the optional runs history', async () => { const task = makeTask({ lastFiredAt: 1718000300000, diff --git a/packages/core/src/services/cronTasksFile.ts b/packages/core/src/services/cronTasksFile.ts index 7549ea38f4..0d4ae40cdb 100644 --- a/packages/core/src/services/cronTasksFile.ts +++ b/packages/core/src/services/cronTasksFile.ts @@ -87,6 +87,17 @@ export interface DurableCronTask { * (`cron_create`) and legacy tasks, which keep the shared-owner firing model. */ sessionId?: string; + /** + * How each scheduled fire runs. Absent or `'shared'` = the #6389 model: the + * task fires inside its single bound {@link sessionId} session and every run + * accumulates in that one transcript. `'isolated'` = the owning session + * dispatches each fire straight into a FRESH sub-session (its own clean + * context and transcript) and never runs the prompt inline. Absent defaults to + * `'shared'` so tool-created and legacy tasks are unchanged. The scheduler + * treats both modes identically — it only carries the field to `onFire`, which + * is where the routing happens. + */ + runMode?: 'shared' | 'isolated'; /** * Bounded, newest-last history of recent fires (capped at MAX_TASK_RUNS). * Absent on tool-created tasks and on any task that has not fired yet. @@ -394,6 +405,12 @@ function isValidTask(value: unknown): value is DurableCronTask { // would treat it as unbound, so a "bound" task would silently run unbound. (obj['sessionId'] === undefined || (typeof obj['sessionId'] === 'string' && obj['sessionId'].length > 0)) && + // Absent = 'shared'. Any string other than the two known modes routes + // through fix-or-delete rather than being silently treated as 'shared', + // so a typo can't quietly disable per-run isolation. + (obj['runMode'] === undefined || + obj['runMode'] === 'shared' || + obj['runMode'] === 'isolated') && (obj['runs'] === undefined || isValidRuns(obj['runs'])) ); } diff --git a/packages/core/src/tools/create-sub-session.test.ts b/packages/core/src/tools/create-sub-session.test.ts new file mode 100644 index 0000000000..581a6d6cdb --- /dev/null +++ b/packages/core/src/tools/create-sub-session.test.ts @@ -0,0 +1,172 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it, vi } from 'vitest'; +import { + CreateSubSessionTool, + MAX_SUB_SESSION_PROMPT_CHARS, +} from './create-sub-session.js'; +import type { Config, SubSessionSpawner } from '../config/config.js'; + +function makeConfig(spawner?: SubSessionSpawner): Config { + return { + getSubSessionSpawner: () => spawner, + } as unknown as Config; +} + +describe('CreateSubSessionTool', () => { + it('has the correct name', () => { + expect(new CreateSubSessionTool(makeConfig()).name).toBe( + 'create_sub_session', + ); + }); + + it('defaults to ask permission so delegated prompts face classifier review', async () => { + const tool = new CreateSubSessionTool(makeConfig()); + const invocation = tool.build({ prompt: 'do X' }); + expect(await invocation.getDefaultPermission()).toBe('ask'); + }); + + it('reports daemon-only when no spawner is wired (interactive / headless)', async () => { + const tool = new CreateSubSessionTool(makeConfig(undefined)); + const res = await tool + .build({ prompt: 'do X' }) + .execute(new AbortController().signal); + expect(res.error?.message).toContain('qwen serve'); + expect(res.returnDisplay).toContain('daemon-only'); + }); + + it('first-turn (default): passes trimmed params to the spawner and returns its result', async () => { + const spawner = vi.fn(async () => ({ + sessionId: 'sub-1', + result: 'the answer', + stopReason: 'end_turn', + })); + const tool = new CreateSubSessionTool(makeConfig(spawner)); + const res = await tool + .build({ prompt: ' summarize ', model: 'm1', name: 'digest' }) + .execute(new AbortController().signal); + + expect(spawner).toHaveBeenCalledWith({ + prompt: 'summarize', + completion: 'first-turn', + model: 'm1', + name: 'digest', + }); + expect(res.error).toBeUndefined(); + expect(res.llmContent).toContain('the answer'); + expect(res.llmContent).toContain('sub-1'); + }); + + it('sent: returns immediately with the session id, not a result', async () => { + const spawner = vi.fn(async () => ({ sessionId: 'sub-2' })); + const tool = new CreateSubSessionTool(makeConfig(spawner)); + const res = await tool + .build({ prompt: 'go', completion: 'sent' }) + .execute(new AbortController().signal); + + expect(spawner).toHaveBeenCalledWith({ prompt: 'go', completion: 'sent' }); + expect(res.error).toBeUndefined(); + expect(res.llmContent).toContain('sub-2'); + expect(res.llmContent).toMatch(/did not wait/i); + }); + + it('reports a completed-but-empty first turn without an error', async () => { + const spawner = vi.fn(async () => ({ + sessionId: 'sub-3', + result: '', + stopReason: 'end_turn', + })); + const tool = new CreateSubSessionTool(makeConfig(spawner)); + const res = await tool + .build({ prompt: 'x' }) + .execute(new AbortController().signal); + expect(res.error).toBeUndefined(); + expect(res.llmContent).toContain('no text output'); + }); + + it('surfaces a spawner error as a tool error', async () => { + const spawner = vi.fn(async () => { + throw new Error('spawn boom'); + }); + const tool = new CreateSubSessionTool(makeConfig(spawner)); + const res = await tool + .build({ prompt: 'x' }) + .execute(new AbortController().signal); + expect(res.error?.message).toContain('spawn boom'); + }); + + it('returns as soon as the turn is cancelled, without waiting for the spawn', async () => { + // Session.ts awaits execute() without racing the abort, so a tool that + // ignores its signal pins the caller's tool loop until the daemon's + // 5-minute first-turn ceiling. This spawner never settles. + let rejectSpawn!: (err: Error) => void; + const spawner = vi.fn( + () => + new Promise<{ sessionId: string }>((_, reject) => { + rejectSpawn = reject; + }), + ); + const ac = new AbortController(); + const tool = new CreateSubSessionTool(makeConfig(spawner)); + + const settled = tool.build({ prompt: 'x' }).execute(ac.signal); + ac.abort(); + const res = await settled; + + expect(res.returnDisplay).toBe('Cancelled'); + expect(res.llmContent).toMatch(/cancelled/i); + // The sub-session is NOT cancelled — it has no abort seam — so the tool + // must say so rather than implying the work was undone. + expect(res.llmContent).toMatch(/runs independently|not cancelled/i); + // The abandoned spawn must not raise an unhandled rejection. + rejectSpawn(new Error('late failure nobody is listening for')); + await new Promise((r) => setTimeout(r, 0)); + }); + + it('never calls the spawner when the signal is already aborted', async () => { + // The abort must be checked BEFORE the spawn starts. Passing `spawner(…)` + // as an argument evaluates it first, so a pre-cancelled turn would still + // create a sub-session on the daemon (and consume its concurrency slot) + // before reporting itself cancelled. + const spawner = vi.fn(async () => ({ sessionId: 'sub-should-not-exist' })); + const ac = new AbortController(); + ac.abort(); + + const res = await new CreateSubSessionTool(makeConfig(spawner)) + .build({ prompt: 'x' }) + .execute(ac.signal); + + expect(spawner).not.toHaveBeenCalled(); + expect(res.returnDisplay).toBe('Cancelled'); + // Nothing was created, so the message must not hedge. + expect(res.llmContent).toMatch(/no sub-session was created/i); + expect(res.llmContent).not.toMatch(/may already have been created/i); + }); + + it('does not report cancellation when the spawn wins the race', async () => { + const spawner = vi.fn(async () => ({ sessionId: 'sub-7', result: 'done' })); + const ac = new AbortController(); + const res = await new CreateSubSessionTool(makeConfig(spawner)) + .build({ prompt: 'x' }) + .execute(ac.signal); + expect(res.returnDisplay).not.toBe('Cancelled'); + expect(res.llmContent).toContain('done'); + }); + + it('rejects a prompt over the character limit before reaching the spawner', async () => { + const spawner = vi.fn(async () => ({ sessionId: 'sub-8' })); + const tool = new CreateSubSessionTool(makeConfig(spawner)); + expect(() => + tool.build({ prompt: 'x'.repeat(MAX_SUB_SESSION_PROMPT_CHARS + 1) }), + ).toThrow(new RegExp(`${MAX_SUB_SESSION_PROMPT_CHARS}`)); + expect(spawner).not.toHaveBeenCalled(); + // The boundary itself is accepted. + expect(() => + tool.build({ prompt: 'x'.repeat(MAX_SUB_SESSION_PROMPT_CHARS) }), + ).not.toThrow(); + }); +}); diff --git a/packages/core/src/tools/create-sub-session.ts b/packages/core/src/tools/create-sub-session.ts new file mode 100644 index 0000000000..8e19286dda --- /dev/null +++ b/packages/core/src/tools/create-sub-session.ts @@ -0,0 +1,318 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * `create_sub_session` tool — spawns a FRESH top-level sub-session (a sibling + * of the current session, its own transcript) and runs a prompt in it. + * + * Daemon-only: it works only when running under `qwen serve`, where the ACP + * session wires a {@link SubSessionSpawner} that routes the request to the + * daemon bridge (`spawnOrAttach` + `sendPrompt`). In interactive TUI / headless + * there is no bridge, so no spawner is wired and the tool reports itself + * unavailable. + * + * Two completion modes: + * - `'sent'` — resolve as soon as the prompt is dispatched (fire-and- + * forget); the sub-session keeps running independently. + * - `'first-turn'`— wait for the sub-session's first turn to finish and return + * its result to the caller (default). + */ + +import type { ToolInvocation, ToolResult } from './tools.js'; +import { BaseDeclarativeTool, BaseToolInvocation, Kind } from './tools.js'; +import { ToolNames, ToolDisplayNames } from './tool-names.js'; +import type { Config } from '../config/config.js'; +import type { PermissionDecision } from '../permissions/types.js'; + +export interface CreateSubSessionParams { + prompt: string; + completion?: 'sent' | 'first-turn'; + model?: string; + name?: string; +} + +const DAEMON_ONLY_MESSAGE = + 'create_sub_session is only available when running under `qwen serve` ' + + '(daemon mode). There is no session bridge in this environment, so a ' + + 'sub-session cannot be spawned.'; + +/** Ceiling on the delegated prompt. Mirrors the scheduled-task REST route's + * `MAX_PROMPT_LENGTH`: both hand a model-authored prompt to a fresh session, so + * they cap it the same way. Rejected here (a clear tool error the model can act + * on) as well as at the bridge boundary, which cannot trust this side. */ +export const MAX_SUB_SESSION_PROMPT_CHARS = 100_000; + +/** Sentinel for "the caller's turn was cancelled while the spawn was in + * flight". A symbol, so it can never collide with a spawner result. */ +const CANCELLED = Symbol('create_sub_session:cancelled'); + +/** + * Resolve as soon as EITHER the spawn settles or `signal` aborts. + * + * `Session.ts` awaits `invocation.execute(signal)` without racing the abort + * itself, so a tool that ignores its signal pins the caller's tool loop for as + * long as it runs — here, up to the daemon's 5-minute `first-turn` ceiling. + * + * Takes a thunk, not a promise: an already-aborted turn must not create daemon + * work at all. Passing `spawner(…)` directly would evaluate it as an argument, + * spawning a sub-session before the abort was ever checked. + */ +async function raceCancellation( + start: () => Promise, + signal: AbortSignal, +): Promise { + if (signal.aborted) return CANCELLED; + const spawn = start(); + // The losing branch's rejection still needs a handler, or Node reports an + // unhandled rejection once the cancel branch wins. + void spawn.catch(() => {}); + return new Promise((resolve, reject) => { + const onAbort = (): void => resolve(CANCELLED); + signal.addEventListener('abort', onAbort, { once: true }); + spawn.then( + (value) => { + signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (err) => { + signal.removeEventListener('abort', onAbort); + reject(err); + }, + ); + }); +} + +class CreateSubSessionInvocation extends BaseToolInvocation< + CreateSubSessionParams, + ToolResult +> { + constructor( + private config: Config, + params: CreateSubSessionParams, + ) { + super(params); + } + + getDescription(): string { + const mode = this.params.completion ?? 'first-turn'; + // Sanitize: strip control chars / markdown-injection from user-controlled + // label so the description can't break tool-call UI or prompt injection. + const raw = this.params.name ?? this.params.prompt; + // eslint-disable-next-line no-control-regex -- stripping C0 control chars + const cleaned = raw.replace(/[\r\n\t\x00-\x1f]/g, ' ').trim(); + const label = cleaned.length > 80 ? cleaned.slice(0, 77) + '…' : cleaned; + return `[${mode}] ${label}`; + } + + /** + * `create_sub_session` runs a model-authored prompt with full tool access in + * a fresh session — the same privileged-sink shape as `cron_create`, + * `send_message` and `task_create`, which all return `'ask'`. The L3 default + * must NOT be `'allow'`: AUTO mode short-circuits before the L5 classifier + * when `finalPermission === 'allow'`, and DEFAULT mode skips confirmation, + * so the delegated prompt would never be reviewed. `'ask'` lets AUTO route + * the call through the classifier, which resolves it without a human. + * + * This gate applies to MODEL-initiated calls only. An `isolated` scheduled + * task is dispatched daemon-side (`Session.#dispatchIsolatedCronFire`) without + * going through the tool, so an unattended fire never blocks on a permission + * request nobody is there to answer. Its prompt was approved when the task was + * created; laundering it back through the model would re-open that gate. + */ + override async getDefaultPermission(): Promise { + return 'ask'; + } + + async execute(signal: AbortSignal): Promise { + const spawner = this.config.getSubSessionSpawner(); + if (!spawner) { + return { + llmContent: DAEMON_ONLY_MESSAGE, + returnDisplay: 'Unavailable (daemon-only)', + error: { message: DAEMON_ONLY_MESSAGE }, + }; + } + + const completion = this.params.completion ?? 'first-turn'; + const prompt = this.params.prompt.trim(); + + // Set only once the spawn actually starts, which tells the two cancellation + // outcomes apart: a turn cancelled before `execute` ran leaves nothing + // behind, while one cancelled mid-flight may have created a sub-session. + let spawnStarted = false; + + try { + const res = await raceCancellation(() => { + spawnStarted = true; + return spawner({ + prompt, + completion, + ...(this.params.model ? { model: this.params.model } : {}), + ...(this.params.name ? { name: this.params.name } : {}), + }); + }, signal); + + if (res === CANCELLED) { + // Return the caller's turn to it immediately. A sub-session that DID + // start is NOT cancelled — `sendPrompt` has no abort seam — so it keeps + // running and keeps its concurrency slot until its own turn drains. + // Freeing the slot here would let the caller over-admit against a + // sub-session that is still consuming a bridge session and model quota. + const message = spawnStarted + ? 'create_sub_session was cancelled. A sub-session may already have ' + + 'been created; it runs independently and is not cancelled.' + : 'create_sub_session was cancelled before it started. No ' + + 'sub-session was created.'; + return { llmContent: message, returnDisplay: 'Cancelled' }; + } + + // Embed a clickable session link in the display output so the web shell + // can render a "jump to session" button. The `qwen-session://` scheme is + // intercepted by the markdown renderer and dispatched as a DOM event. + const sessionLink = `[🧵 ${res.sessionId.slice(0, 8)}](qwen-session://${res.sessionId})`; + + if (completion === 'sent') { + // Fire-and-forget: report the id; the caller did not wait for a result. + return { + llmContent: + `Sub-session ${sessionLink} created and the prompt was ` + + 'dispatched. It runs independently — this call did not wait for a ' + + 'result.', + returnDisplay: `${sessionLink} started`, + }; + } + + // first-turn: return the sub-session's first-turn output to the caller. + const stop = res.stopReason ? ` (stopReason: ${res.stopReason})` : ''; + const body = + res.result && res.result.length > 0 + ? res.result + : `Sub-session ${sessionLink} completed its first turn but ` + + 'produced no text output.'; + return { + llmContent: `Sub-session ${sessionLink} first-turn result${stop}:\n\n${body}`, + returnDisplay: `${sessionLink} completed${stop}`, + }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + return { + llmContent: `Error creating sub-session: ${message}`, + returnDisplay: message, + error: { message }, + }; + } + } +} + +export class CreateSubSessionTool extends BaseDeclarativeTool< + CreateSubSessionParams, + ToolResult +> { + static readonly Name = ToolNames.CREATE_SUB_SESSION; + + constructor(private config: Config) { + super( + CreateSubSessionTool.Name, + ToolDisplayNames.CREATE_SUB_SESSION, + 'Spawn a fresh, independent sub-session (its own clean context and ' + + 'transcript) and run a prompt in it. Use to fan work out into a ' + + 'separate session — e.g. a self-contained sub-task you want isolated ' + + 'from this conversation.\n\n' + + 'ONLY available when running under `qwen serve` (daemon mode); it is ' + + 'inert in a plain interactive session.\n\n' + + '## Completion modes\n' + + "- `first-turn` (default): waits for the sub-session's first turn to " + + 'finish and returns its result to you. Use when you need the answer ' + + 'back.\n' + + '- `sent`: returns immediately after dispatching the prompt, without ' + + 'waiting. Use for fire-and-forget launches whose output you do not ' + + 'need inline.\n\n' + + 'The sub-session runs the prompt with full tool access, starting from ' + + 'zero context — brief it completely in `prompt` (it cannot see this ' + + 'conversation).', + Kind.Other, + { + type: 'object', + properties: { + prompt: { + type: 'string', + description: + 'The full, self-contained prompt to run in the new sub-session. ' + + 'It starts with no context from this conversation, so include ' + + 'everything it needs.', + }, + completion: { + type: 'string', + enum: ['sent', 'first-turn'], + description: + "'first-turn' (default) waits for the sub-session's first turn " + + "and returns its result. 'sent' returns immediately after the " + + 'prompt is dispatched (fire-and-forget).', + }, + model: { + type: 'string', + description: + 'Optional model service id for the sub-session. Omit to use the ' + + 'default model.', + }, + name: { + type: 'string', + description: + 'Optional display name for the sub-session in the session list.', + }, + }, + required: ['prompt'], + additionalProperties: false, + }, + true, // isOutputMarkdown + false, // canUpdateOutput + true, // shouldDefer — spawning is infrequent + false, // alwaysLoad + 'sub-session spawn delegate fan-out isolated session', + ); + } + + protected createInvocation( + params: CreateSubSessionParams, + ): ToolInvocation { + return new CreateSubSessionInvocation(this.config, params); + } + + protected override validateToolParamValues( + params: CreateSubSessionParams, + ): string | null { + if (!params.prompt || params.prompt.trim() === '') { + return 'Parameter "prompt" must be a non-empty string.'; + } + if (params.prompt.length > MAX_SUB_SESSION_PROMPT_CHARS) { + return `Parameter "prompt" exceeds the ${MAX_SUB_SESSION_PROMPT_CHARS}-character limit.`; + } + if ( + params.completion !== undefined && + params.completion !== 'sent' && + params.completion !== 'first-turn' + ) { + return 'Parameter "completion" must be "sent" or "first-turn".'; + } + return null; + } + + /** + * Surface the delegated prompt + mode to the AUTO classifier. The sub-session + * executes this prompt with tool access, so it must face the same scrutiny as + * a direct command — without this the classifier sees `create_sub_session({})` + * and is blind to what the sub-session will be asked to do. + */ + override toAutoClassifierInput( + params: CreateSubSessionParams, + ): Record { + return { + prompt: params.prompt, + completion: params.completion ?? 'first-turn', + ...(params.model ? { model: params.model } : {}), + }; + } +} diff --git a/packages/core/src/tools/tool-names.ts b/packages/core/src/tools/tool-names.ts index ff72e901ff..fca00f8608 100644 --- a/packages/core/src/tools/tool-names.ts +++ b/packages/core/src/tools/tool-names.ts @@ -38,6 +38,7 @@ export const ToolNames = { CRON_LIST: 'cron_list', CRON_DELETE: 'cron_delete', LOOP_WAKEUP: 'loop_wakeup', + CREATE_SUB_SESSION: 'create_sub_session', TASK_STOP: 'task_stop', TASK_CREATE: 'task_create', TASK_UPDATE: 'task_update', @@ -90,6 +91,7 @@ export const ToolDisplayNames = { CRON_LIST: 'CronList', CRON_DELETE: 'CronDelete', LOOP_WAKEUP: 'LoopWakeup', + CREATE_SUB_SESSION: 'CreateSubSession', TASK_STOP: 'TaskStop', TASK_CREATE: 'TaskCreate', TASK_UPDATE: 'TaskUpdate', diff --git a/packages/web-shell/client/App.tsx b/packages/web-shell/client/App.tsx index 74652947cd..966e155251 100644 --- a/packages/web-shell/client/App.tsx +++ b/packages/web-shell/client/App.tsx @@ -2656,6 +2656,19 @@ export function App({ [loadSidebarSession, reportError], ); + // Listen for `qwen:open-session` events dispatched by the markdown renderer + // when a `qwen-session://` link is clicked. Navigate to the session. + useEffect(() => { + const handler = (e: Event) => { + const sessionId = (e as CustomEvent).detail; + if (typeof sessionId === 'string' && sessionId) { + handleOpenSessionFromOverview(sessionId); + } + }; + window.addEventListener('qwen:open-session', handler); + return () => window.removeEventListener('qwen:open-session', handler); + }, [handleOpenSessionFromOverview]); + useEffect(() => { if ( sidebarSwitchingSessionId !== null && diff --git a/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.module.css b/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.module.css index 4a11e60098..0cdc419c80 100644 --- a/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.module.css +++ b/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.module.css @@ -98,6 +98,32 @@ margin-left: 2px; } +.fieldHint { + font-size: 11px; + line-height: 1.4; + color: var(--muted-foreground); +} + +.radioGroup { + display: flex; + flex-wrap: wrap; + gap: 16px; + padding: 2px 0; +} + +.radioOption { + display: flex; + align-items: center; + gap: 6px; + font-size: 13px; + cursor: pointer; +} + +.radioOption input { + cursor: pointer; + margin: 0; +} + .input, .textarea, .select { diff --git a/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.test.tsx b/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.test.tsx index d462a2ca40..0a4ea16723 100644 --- a/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.test.tsx +++ b/packages/web-shell/client/components/dialogs/ScheduledTasksDialog.test.tsx @@ -26,7 +26,12 @@ interface MockTask { lastFiredAt: number | null; nextRunAt: number | null; sessionId: string | null; - runs: Array<{ at: number; kind?: 'scheduled' | 'catch-up' }>; + runMode?: 'shared' | 'isolated'; + runs: Array<{ + at: number; + kind?: 'scheduled' | 'catch-up'; + sessionId?: string; + }>; } const { actions } = vi.hoisted(() => ({ @@ -104,6 +109,23 @@ function findButton(label: string): HTMLButtonElement | undefined { ); } +// Run mode is a radio group; the frequency picker is the (only) setRunMode(mode)} + /> + {t(`scheduledTasks.runMode.${mode}`)} + + ))} + + + {t(`scheduledTasks.runMode.${runMode}.hint`)} + + +
)} {!isTodo && expanded && detailView && ( diff --git a/packages/web-shell/client/i18n.tsx b/packages/web-shell/client/i18n.tsx index cd11058d09..f785c5781b 100644 --- a/packages/web-shell/client/i18n.tsx +++ b/packages/web-shell/client/i18n.tsx @@ -731,6 +731,13 @@ const EN: Messages = { 'scheduledTasks.dur.h': 'h', 'scheduledTasks.dur.m': 'm', 'scheduledTasks.dur.s': 's', + 'scheduledTasks.runMode': 'Run mode', + 'scheduledTasks.runMode.shared': 'Shared session', + 'scheduledTasks.runMode.isolated': 'Isolated (fresh session per run)', + 'scheduledTasks.runMode.shared.hint': + 'Runs accumulate in one session transcript.', + 'scheduledTasks.runMode.isolated.hint': + 'Each run gets a clean session context.', 'sidebar.label': 'Workspace sidebar', 'sidebar.toggleMenu': 'Toggle menu', 'sidebar.newChat': 'New chat', @@ -2413,6 +2420,11 @@ const ZH: Messages = { 'scheduledTasks.dur.h': '小时', 'scheduledTasks.dur.m': '分', 'scheduledTasks.dur.s': '秒', + 'scheduledTasks.runMode': '运行模式', + 'scheduledTasks.runMode.shared': '共享会话', + 'scheduledTasks.runMode.isolated': '隔离（每次运行新建会话）', + 'scheduledTasks.runMode.shared.hint': '所有运行记录累积在同一个会话中。', + 'scheduledTasks.runMode.isolated.hint': '每次运行获得独立的会话上下文。', 'sidebar.label': '工作区侧边栏', 'sidebar.toggleMenu': '切换菜单', 'sidebar.newChat': '新对话', diff --git a/packages/webui/src/daemon/workspace/types.ts b/packages/webui/src/daemon/workspace/types.ts index e742ca42e6..c568b5e6b8 100644 --- a/packages/webui/src/daemon/workspace/types.ts +++ b/packages/webui/src/daemon/workspace/types.ts @@ -188,6 +188,13 @@ export interface DaemonScheduledTask { /** Id of the dedicated session this task is bound to — its transcript is the * task's run history. Null for unbound tool-created/legacy tasks. */ sessionId: string | null; + /** How each fire runs. `'shared'` (default) runs in the bound session so runs + * accumulate in one transcript; `'isolated'` dispatches each scheduled fire + * into a fresh sub-session daemon-side, so the bound session's transcript + * stays empty. Normalized (never undefined). Note: `runs[].sessionId` always + * records the anchor (bound) session — the sub-session id is not surfaced + * here. */ + runMode: 'shared' | 'isolated'; /** Bounded, newest-last history of recent fires. Empty for tasks that have * not fired (and, by nature, for one-shots — they are deleted on fire). */ runs: DaemonScheduledTaskRun[]; @@ -202,6 +209,9 @@ export interface DaemonCreateScheduledTaskRequest { recurring?: boolean; /** Defaults to true. */ enabled?: boolean; + /** Defaults to `'shared'` (the #6389 single-session model). `'isolated'` + * spawns a fresh session per fire. */ + runMode?: 'shared' | 'isolated'; } /** Partial update. `name: null` (or '') clears the name. Omitted fields are @@ -212,6 +222,7 @@ export interface DaemonUpdateScheduledTaskRequest { name?: string | null; recurring?: boolean; enabled?: boolean; + runMode?: 'shared' | 'isolated'; } export interface DaemonWorkspaceActions {