feat(acp-bridge): route sessions across legacy and managed engines (#12698)

* feat(acp-bridge): route sessions across legacy and managed engines

* fix(acp-bridge): preserve idle reclamation across engine restores

* test(sdk-java): align tool response assertion timeouts

* fix(sdk-java): preserve subsecond writer leases

* fix(acp-bridge): isolate live replay and runtime reservations by engine

* test(live): make stale discovery owner liveness deterministic

* fix(acp-bridge): close routing and idle lifecycle gaps

* fix(acp-bridge): quarantine null session creation responses
This commit is contained in:
Shaojin Wen authored and GitHub committed 2026-09-26 05:35:07 +00:00
1 parent 89b057befd
commit 342ba1bcdd
15 files changed
+2724 -176

No files matched your search

+183
View File
@@ -0,0 +1,183 @@
# ACP Bridge execution engines
[English](./acp-bridge-execution-engines.md) | [简体中文](./acp-bridge-execution-engines.zh-CN.md)
## Status
Implementation plan for the Stage B Bridge slice of #12380, based on upstream
`790bd83c2b`. This slice supplies an opt-in Bridge construction API. Ordinary
serve factories and the Hosted model/tool loop are separate integration work.
## Problem and current behavior
The Bridge owns one reusable ACP channel and one startup promise. Sessions
already keep their own channel and connection, but callbacks look up sessions
through a shared ID map without consistently checking the sending channel.
Adding a second engine must preserve shared admission, reservations, replay,
and physical cleanup rather than duplicating the session control plane.
## Scope
Support Legacy and Managed channels inside one Bridge, fixed session ownership,
engine receipts, channel-scoped callbacks, and complete lifecycle accounting.
Keep existing callers of `channelFactory` compatible.
This change does not implement a Managed Harness, write owner records, select
compatible deployment configurations, wire the ordinary serve factories,
change the public REST API, or implement Stage G takeover. Managed branching
remains unavailable. Workspace control and preheat remain on the Legacy channel.
## Proposed design
### Construction and ownership contract
`executionEngines` contains `legacy` and `managed` factories and a server-owned
`select` callback. It is mutually exclusive with `channelFactory`. Selection
receives a snapshot of the validated spawn/load/resume request, including the
canonical workspace and trusted standalone purpose. It returns only `legacy`
or `managed`. Client metadata cannot override selection.
Selection runs after shared capacity and ID reservation, with the operation
already visible to shutdown. Cold load/resume requires the caller's selector
to read verified durable ownership; ambiguous or unreadable history must reject.
Hot attach uses the existing entry without invoking the selector again. A branch
checks its source engine on cold restore, hot attach and coalesced restore.
The selector retains its original receiver, including prototype methods.
Paired channels require the actual ACP new/load/resume response to contain
`_meta['qwen.session.executionEngine']` matching the selected engine. This key
matches the pinned design/reference implementation. The host must persist or
verify ownership before returning this receipt and before initialization side
effects. A receipt is a trusted host assertion, not a filesystem proof performed
by the Bridge. The generic single-factory API does not require this receipt.
Owner persistence stays outside this package. #12693 currently introduces its
reader and writer foundations; the issue discussion must still settle the
minimal dependency and the relationship between `session_execution_engine`
and `managed_session_header_v1`. This slice does not create another format or
require all of Stage G. Production enablement requires that host integration.
### Channels and admission
Keep one `byId`, default attach entry, ID reservation map, admission budget,
runtime epoch source and set of physically owned channels. Add engine slots for
reusable channels and startup promises. Coalesce startup within an engine;
allow the two engines to start independently. Track dying generations until
physical exit. Channel quarantine blocks fresh work only for that engine.
Idle timers belong to the actual channel. Unbound spawns and restores protect
channels until their work transfers to a channel counter. Binding or settling
these operations rechecks deferred idle timers, including timers consumed while
selection had not yet bound an owner. An unrelated idle channel must not wait
for the selected engine's restore RPC or cleanup to finish. Re-evaluation rearms
only deferred timers without extending another channel's existing idle deadline.
Bare Legacy preheat at idle timeout zero remains timer-free until that channel
is used; unrelated Managed work must not activate its idle cleanup. A settlement
uses a snapshot of its channels and cannot reclaim a replacement published later.
A late exit from an old generation must not cancel another channel's timer.
Runtime-operation reservations protect only their engine's channels; workspace
activity and stop checks still count reservations across both engines.
Shutdown awaits every engine startup, selection, session operation and owned
channel; force shutdown reaches all
tracked children. No failure path switches to the other factory.
### Registration and cleanup
Before dispatch, reject an unaddressable caller-supplied ID or an ID already
owned by a live session. Before registration, check the actual engine receipt
and the returned session ID. Invalid, conflicting or missing receipts reject
registration. Close a safely addressable unregistered session on its original connection. If its ID
cannot be safely addressed, quarantine the original channel, let other sessions
drain, and retain admission until physical exit. Never close another session
merely because a malformed response returned its ID.
Existing sessions on a quarantined channel can continue prompting. Fresh work
for that engine remains blocked until they drain and the channel exits; there
is no bounded drain deadline or automatic recycle in this slice. Before
production enablement, #12380 must define an operational recovery policy that
accounts for those live sessions and physical admission ownership.
Restore failures after a successful ACP response use the same original-channel
cleanup discipline. Public timeout is not evidence of physical completion.
A late success without a session ID remains a success requiring cleanup or quarantine; it is distinct
from an explicit RPC failure.
Keep reservations until the original operation settles and cleanup completes.
The cleanup fence covers receipt rejection as well as timeout; its retry hint
is a backoff policy, not an estimate of when cleanup will complete.
### Live routing and workspace operations
Prompt, cancellation, approvals, model changes and session close use the entry's
bound connection. Inbound session lookups, restore replay, background admission,
and generation events must also match the sending channel/connection. Live
transcript reads, turn-index reads and flushes use that owner as well; only
persisted reads without a live entry use the Legacy workspace fallback.
Workspace MCP, configuration/status control and preheat use Legacy. Aggregate
liveness and activity inspect both engines; idle reclamation locates the actual
candidate by ID. Preheat keepalive extends only Legacy's idle deadline. Managed
sessions and in-flight work prevent their own channel from being reclaimed;
Managed has no independent preheat keepalive in this slice. Child resource
sampling and user-language delivery also remain Legacy-only. Before production
enablement, #12380 must define per-engine resource aggregation and language
propagation together with host wiring.
Production paired-host wiring is also blocked on these workspace contracts:
- Separate aggregate runtime liveness from Legacy workspace-control readiness.
Coordinator preheat decisions and workspace-service readiness must check the
capability they need; Managed alone cannot make Legacy preheat successful.
- Define capability generations independently of the shared epoch allocator.
Starting Managed must not invalidate unchanged Legacy skills/MCP preparation,
and losing Legacy must invalidate its capabilities even if Managed survives.
Workspace-stop confirmation must retain a deliberate generation policy.
- Deliver session-affecting workspace changes to every relevant live engine,
including permission rules, approval/workflow settings, providers and skills.
Define acknowledgements and partial-failure handling; a newly applied deny
rule must not silently leave existing Managed sessions on old permissions.
These are acceptance gates for #12380 host integration, not capabilities supplied
by the current Legacy-only workspace-control implementation.
The existing workspace-stop receipt addresses one physical channel, so stopping
multiple live channels is explicitly blocked until that receipt is extended.
A single live channel remains stoppable. Managed branch/side-task requests reject
before mutating history.
## Files and consumers
| Area | Files / consumers |
| ------------------- | ------------------------------------------------------------------------------------------ |
| Public construction | `bridgeOptions.ts`, package `index.ts`; daemon, Channels, SDK/embedded Bridge constructors |
| Channel ownership | `channel-lifecycle.ts`, `channel-startup.ts`, `channel-harness.ts` |
| Session routing | `session-control-plane.ts`, `BridgeClient` callbacks |
| Verification | Collocated Bridge/lifecycle tests and isolated process test script |
Existing daemon, Channels and embedded constructors remain on the single-factory
path. There are no new daemon routes. Workspace control is workspace scoped;
all session operations belong to the live session owner. A missing or failed
Managed owner must never resolve through a Legacy or primary-runtime fallback.
## Validation and acceptance criteria
1. Both engines coexist, with startup coalescing and shared session/ID limits.
2. Changing the selector default cannot change an attached or durably owned
session. Selection and Managed failures cause zero other-engine dispatches.
3. Matching receipts permit registration; missing/conflicting receipts and
malformed/colliding IDs reject without losing physical resource accounting.
4. Prompts, cancellation, permissions, replay and close stay on their owner;
another channel cannot inject events or answer for that session.
5. Idle cleanup, quarantine, delayed startup, late responses and shutdown cover
both engines and preserve replacement generations.
6. Existing single-factory tests remain green. Run build, typecheck, bundle,
focused unit tests and isolated process verification. Test doubles prove the
Bridge contract, not durable host persistence or Hosted inference.
## Risks and open questions
The highest risks are releasing admission before cleanup and treating one
engine's current channel as the entire workspace. Tests must observe actual
factory/connection calls and pending teardown, not only final session counts.
The owner persistence dependency and production host receipt implementation
remain the integration questions posted in #12380; the Bridge seam is usable
for contract tests while those are resolved.
@@ -0,0 +1,140 @@
# ACP Bridge 执行引擎
[English](./acp-bridge-execution-engines.md) | [简体中文](./acp-bridge-execution-engines.zh-CN.md)
## 状态
#12380 的 Stage B Bridge 切片实施方案,基于上游 `790bd83c2b`。本切片提供
显式启用的 Bridge 构造 API。普通 serve factory 和 Hosted 模型/工具闭环另行接线。
## 问题与当前行为
Bridge 目前持有一个可复用 ACP channel 和一个启动 promise。Session 已保存自己的
channel 与 connection,但回调通过共享 ID 表查找 Session 时,尚未始终核对发送通道。
加入第二个引擎必须保留共享准入、ID 占用、回放与物理清理,不能复制会话控制面。
## 范围
在同一 Bridge 中支持 Legacy/Managed 通道、固定会话归属、引擎回执、按通道隔离的
回调和完整生命周期记账。保持现有 `channelFactory` 调用方兼容。
本变更不实现 Managed Harness、不写 owner 记录、不选择兼容部署配置、不接线普通
serve factory、不改变公共 REST API,也不实现 Stage G 接管。Managed 分支暂不可用。
工作区控制与预热仍使用 Legacy 通道。
## 方案
### 构造与归属契约
`executionEngines` 包含 `legacy`、`managed` factory 和服务端 `select` 回调,
与 `channelFactory` 互斥。选择器收到已验证的 spawn/load/resume 请求快照,包含
canonical workspace 和已确认的 standalone 用途,只能返回 `legacy` 或 `managed`。
客户端 metadata 不能覆盖选择。
先进行共享容量和 ID 占用,将操作登记到 shutdown 可见的在途集合,再执行选择。
冷 load/resume 要求调用方选择器读取已验证的持久 owner;历史含糊或不可读必须拒绝。
热 attach 使用已有 entry,不重新调用选择器。分支在冷恢复、热 attach 和合并恢复
入口都核对源引擎。选择器保留原始 receiver,包括原型方法。
配对通道要求实际 ACP new/load/resume 响应携带
`_meta['qwen.session.executionEngine']`,且匹配所选引擎。该 key 与冻结设计和参考
实现一致。Host 必须先持久化或验证归属,再返回回执,并且该过程要早于初始化副作用。
回执是受信 host 的声明,不代表 Bridge 自己验证过文件持久化。普通单 factory API
不要求此回执。
Owner 持久化留在包外。#12693 正在引入 reader/writer 基础;最小依赖以及
`session_execution_engine` 与 `managed_session_header_v1` 的关系仍需在 issue
中对齐。本切片不新建格式,也不要求整个 Stage G 完成。生产启用需要完成 host 接线。
### 通道与准入
保留唯一的 `byId`、默认 attach entry、ID 占用表、准入预算、runtime epoch 来源和
物理通道集合。按引擎划分可复用通道与启动 promise,同引擎合并启动,不同引擎可独立
启动。正在退出的代数持续记账到物理退出。通道隔离只阻止该引擎的新会话。
空闲定时器归实际通道所有。未绑定的创建和恢复操作保护通道,直到工作转交给通道计数器。
这些操作绑定或结算时重新检查被推迟的空闲定时器,包括选择器尚未绑定归属期间已触发
并消耗定时器的通道。无关的空闲通道不应等待所选引擎的恢复 RPC 或清理结束。重新检查只重挂被推迟的定时器,不延长另一通道已有的空闲期限。
空闲超时为零的 Legacy 裸预热保持无定时器,直到该通道被使用;无关 Managed 工作
不能启动其空闲回收。结算遍历通道快照,不能回收结算开始后发布的替代通道。
旧代迟到的退出不能取消另一通道的定时器。
运行时操作占用仅保护所属引擎的通道;工作区活动与停止检查仍汇总两个引擎的占用。
Shutdown 等待所有引擎启动、选择、会话操作和物理通道,强制退出覆盖全部已登记子进程。失败路径
绝不切换到另一 factory。
### 注册与清理
派发前拒绝无法寻址或已被活跃会话占用的调用方指定 ID。注册前核验实际引擎回执和
返回的 Session ID。缺失、非法或冲突回执拒绝注册。可安全寻址的未注册 Session 在原 connection 上关闭;ID 无法安全寻址时隔离原通道,等待
其他会话排空,并保持准入到物理退出。不能因异常响应返回了其他 Session 的 ID 就关闭它。
隔离通道上的已有会话仍可继续 Prompt。该引擎的新工作会持续被阻止,直到这些会话
排空且通道退出;本切片没有有限排空期限或自动回收机制。生产启用前,#12380 必须
定义运维恢复策略,并考虑这些活跃会话及物理准入占用。
ACP 恢复成功后发生的校验失败,仍按原通道清理。对外超时不代表物理操作完成。
迟到成功即使缺少 Session ID,也仍需清理或隔离,不能等同明确的 RPC 失败。
原操作结算且清理完成前保留占用。清理围栏同时覆盖回执拒绝与超时;重试提示是退避
策略,不是清理完成时间的估计。
### 活跃路由与工作区操作
Prompt、取消、审批、模型变更和关闭使用 entry 绑定的 connection。入站 Session
查找、恢复回放、后台准入与 generation 事件也必须匹配发送的 channel/connection。
活跃会话的 transcript、turn-index 读取和 flush 同样走 owner;仅不存在活跃 entry
的持久历史读取使用 Legacy 工作区回退。
工作区 MCP、配置/状态控制与预热使用 Legacy。整体存活与活动状态检查两个引擎;
空闲回收按实际 candidate ID 找到通道。预热 keepalive 只延长 Legacy 的空闲期限;
Managed 的会话和在途工作仍会阻止其通道被回收,本切片不提供独立的 Managed 预热
keepalive。子进程资源采样与用户语言下发也仍只覆盖 Legacy。生产启用前,#12380
需结合 host 接线确定按引擎聚合资源与传播语言的行为。
生产配对 host 接线还必须先满足以下工作区契约:
- 区分整体 runtime 存活与 Legacy 工作区控制就绪。Coordinator 的预热判断和
workspace-service 的就绪结果必须检查所需能力;仅 Managed 存活不能使 Legacy
预热被判为成功。
- 在共享 epoch 分配器之外定义能力代际。Managed 启动不能使未改变的 Legacy
skills/MCP 准备失效;Legacy 退出必须使其能力失效,即使 Managed 仍存活。
工作区 stop 确认也必须保留明确的代际策略。
- 将影响会话的工作区变更传给所有相关的活跃引擎,包括权限规则、审批/workflow
设置、providers 和 skills。定义确认和部分失败处理;新生效的 deny 规则不能
悄悄让已有 Managed 会话继续使用旧权限。
这些是 #12380 host 集成的验收门槛,当前仅走 Legacy 的工作区控制实现尚不提供这些能力。
现有工作区 stop 回执只表示一个物理通道,因此多个通道存活时明确拒绝 stop,直到
回执扩展;只有一个通道时仍可停止。
Managed branch/side-task 在修改历史前拒绝。
## 文件与消费者
| 区域 | 文件 / 消费者 |
| -------- | --------------------------------------------------------------------------- |
| 公共构造 | `bridgeOptions.ts`、包 `index.ts`;daemon、Channels、SDK/嵌入 Bridge 构造方 |
| 通道归属 | `channel-lifecycle.ts`、`channel-startup.ts`、`channel-harness.ts` |
| 会话路由 | `session-control-plane.ts`、`BridgeClient` 回调 |
| 验证 | 同目录 Bridge/lifecycle 测试与隔离进程脚本 |
现有 daemon、Channels 与嵌入构造方保持单 factory 路径,不增加 daemon route。
工作区控制归工作区,所有会话操作归活跃 Session owner。Managed owner 缺失或失败
时绝不回退到 Legacy 或 primary runtime。
## 验证与验收标准
1. 两种引擎共存,按引擎合并启动,共享 Session/ID 限额。
2. 修改默认选择不能改变已 attach 或持久归属的 Session。选择器和 Managed 失败时,
另一引擎的 dispatch 次数为零。
3. 匹配回执允许注册;缺失/冲突回执及非法/冲突 ID 拒绝,物理资源继续正确记账。
4. Prompt、取消、权限、回放与关闭保持归属,另一通道不能注入事件或代答。
5. 空闲清理、隔离、延迟启动、迟到响应和 shutdown 覆盖两个引擎,并保留替代代数。
6. 现有单 factory 测试通过;运行 build、typecheck、bundle、定向单测和隔离进程验证。
测试替身仅证明 Bridge 契约,不证明 host 持久化或 Hosted 推理已经完成。
## 风险与待定项
主要风险是清理完成前释放准入,或把一个引擎的 current channel 当作整个工作区。
测试必须观察真实 factory/connection 调用和未完成清理,而不只检查最终 Session 数量。
Owner 持久化依赖与生产 host 回执实现仍是 #12380 中待对齐的接线问题;在此期间可通过
Bridge 注入接口验证契约。
+4 -4
View File
@@ -162,9 +162,9 @@ export class SessionArchivingError extends Error {
*
* `restore_in_progress` is the ordinary case: a restore is running and the
* caller can retry shortly. `awaiting_abandoned_cleanup` means the public
* caller already received a timeout, but the non-cancellable ACP registration
* request (and its cleanup) has not settled yet — retrying at the ordinary
* cadence just re-hits the fence, so clients must back off much further.
* caller already received a timeout or registration failure, but registration
* or cleanup has not settled yet. Its conservative backoff is not an estimate
* of when cleanup will complete.
*/
export type RestoreInProgressReason =
| 'restore_in_progress'
@@ -205,7 +205,7 @@ export class RestoreInProgressError extends Error {
: `session/${activeAction}`;
super(
reason === 'awaiting_abandoned_cleanup'
? `Session "${sessionId}" timed out during ${activeTarget} and its abandoned registration has not settled yet; retry ${retryTarget} once cleanup completes`
? `Session "${sessionId}" has an abandoned registration from ${activeTarget} whose cleanup has not settled yet; retry ${retryTarget} once cleanup completes`
: activeAction === 'spawn'
? `Session "${sessionId}" is already being registered by ${activeTarget}; retry ${retryTarget} after it completes`
: `Session "${sessionId}" is already being restored via ${activeTarget}; retry ${retryTarget} after it completes`,
+28
View File
@@ -23,6 +23,26 @@ import type { ServePreflightCell, ServeWorkspaceEnvStatus } from './status.js';
import type { BridgeFileSystem } from './bridgeFileSystem.js';
import type { JournalGrowthSessionLimit } from './replayWindowLimits.js';
import type { PromptLedgerRecord } from './prompt-ledger.js';
import type {
BridgeSpawnRequest,
BridgeRestoreSessionRequest,
} from './bridgeTypes.js';
export type BridgeExecutionEngine = 'legacy' | 'managed';
export const SESSION_EXECUTION_ENGINE_META_KEY = 'qwen.session.executionEngine';
export type BridgeExecutionSelection = {
readonly daemonOwnedStandalone: boolean;
} & (
| {
readonly operation: 'spawn';
readonly request: Readonly<BridgeSpawnRequest>;
}
| {
readonly operation: 'load' | 'resume';
readonly request: Readonly<BridgeRestoreSessionRequest>;
}
);
/**
* Sink for serve-level diagnostic lines (set by the cli daemon logger).
@@ -261,6 +281,14 @@ export interface BridgeOptions {
sessionScope?: 'single' | 'thread';
/** Channel factory; defaults to spawning `qwen --acp` as a child process. */
channelFactory?: ChannelFactory;
/** Server-owned selection; restore must use verified durable ownership. */
executionEngines?: {
legacy: ChannelFactory;
managed: ChannelFactory;
select(
context: BridgeExecutionSelection,
): BridgeExecutionEngine | Promise<BridgeExecutionEngine>;
};
/** Workspace-scoped epoch source shared across Bridge replacement. */
runtimeEpochSource?: BridgeRuntimeEpochSource;
/** Daemon-global admission for the process-wide MCP OAuth callback port. */
@@ -101,6 +101,50 @@ describe('channel harness ownership', () => {
}
});
it('does not visit a new preheated channel while an older settlement awaits exit', async () => {
let releaseExit!: () => void;
let startedKill!: () => void;
const exit = new Promise<void>((resolve) => {
releaseExit = resolve;
});
const killing = new Promise<void>((resolve) => {
startedKill = resolve;
});
const old = makeChannel({ extMethodImpl: () => ({ available: true }) });
const fresh = makeChannel();
const kill = old.channel.kill.bind(old.channel);
vi.spyOn(old.channel, 'kill').mockImplementation(async () => {
startedKill();
await exit;
await kill();
});
const factory = vi
.fn<() => Promise<AcpChannel>>()
.mockResolvedValueOnce(old.channel)
.mockResolvedValue(fresh.channel);
const bridge = makeBridge({
channelFactory: factory,
channelIdleTimeoutMs: 0,
});
let operation: Promise<boolean> | undefined;
try {
await bridge.preheat();
operation = bridge.isWorkspaceMemoryRememberAvailable!();
await killing;
await bridge.preheat();
expect(fresh.killed).toBe(false);
releaseExit();
expect(await operation).toBe(true);
expect(fresh.killed).toBe(false);
expect(bridge.isChannelLive()).toBe(true);
expect(factory).toHaveBeenCalledTimes(2);
} finally {
releaseExit();
await operation;
await bridge.shutdown();
}
});
it('keeps the replacement idle timer armed when an old channel exits late', async () => {
vi.useFakeTimers();
const old = makeChannel();
+109 -55
View File
@@ -28,6 +28,7 @@ import { BridgeTimeoutError, SERVE_CONTROL_EXT_METHODS } from './status.js';
import { terminateChannel } from './channel-transport.js';
import { WorkspaceDrainingError } from './bridgeErrors.js';
import { writeStderrLine } from './internal/stderrLine.js';
import type { BridgeExecutionEngine } from './bridgeOptions.js';
export interface ChannelWorkExclusions {
ignoreCurrentSessionSpawn?: boolean;
@@ -68,14 +69,21 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
channelShouldReapWhenIdle,
sessionCount,
} = options;
const channelLifecycle = createChannelLifecycle();
const defaultEngine = options.executionEngines ? 'legacy' : undefined;
const channelLifecycle = createChannelLifecycle(defaultEngine);
let keepAliveUntil = 0;
let runtimeOperationReservations = 0;
const runtimeOperationReservations = new Map<
BridgeExecutionEngine | undefined,
number
>();
const pendingKeepAliveDeadlines = new Map<symbol, number>();
let idleTimer: ReturnType<typeof setTimeout> | undefined;
const idleTimers = new Map<HarnessChannel, ReturnType<typeof setTimeout>>();
const pendingIdleTimers = new Set<HarnessChannel>();
function liveHarnessChannel(): HarnessChannel | undefined {
const channel = channelLifecycle.current;
function liveHarnessChannel(
engine: BridgeExecutionEngine | undefined = defaultEngine,
): HarnessChannel | undefined {
const channel = channelLifecycle.currentFor(engine);
return channel && !channel.isDying ? channel : undefined;
}
@@ -88,14 +96,17 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
return (
ci.workspaceControlInFlight === 0 &&
hasNoWorkspaceWork(ci) &&
runtimeOperationReservations === 0
(runtimeOperationReservations.get(ci.executionEngine) ?? 0) === 0
);
}
function cancelIdleTimer(): void {
if (idleTimer !== undefined) {
clearTimeout(idleTimer);
idleTimer = undefined;
function cancelIdleTimer(channel?: HarnessChannel): void {
if (channel) pendingIdleTimers.delete(channel);
else pendingIdleTimers.clear();
for (const [owner, timer] of idleTimers) {
if (channel !== undefined && owner !== channel) continue;
clearTimeout(timer);
idleTimers.delete(owner);
}
}
@@ -104,6 +115,7 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
context?: string,
): Promise<void> {
ci.isDying = true;
cancelIdleTimer(ci);
ci.channelLiveness?.stop();
await terminateChannel(
ci.channel,
@@ -148,8 +160,9 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
: 0;
}
function resolvedChannelIdleTimeoutMs(): number {
function resolvedChannelIdleTimeoutMs(engine = defaultEngine): number {
const configured = configuredChannelIdleTimeoutMs();
if (engine !== defaultEngine) return configured;
const now = Date.now();
let pendingKeepAliveMs = 0;
for (const deadline of pendingKeepAliveDeadlines.values()) {
@@ -161,35 +174,43 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
async function startIdleTimer(
ci: HarnessChannel,
context?: string,
exclusions?: ChannelWorkExclusions,
): Promise<void> {
if (
options.isRuntimeStopping() ||
ci.isDying ||
liveHarnessChannel() !== ci
liveHarnessChannel(ci.executionEngine) !== ci
)
return;
const timeoutMs = resolvedChannelIdleTimeoutMs();
if (!hasNoChannelWork(ci, exclusions)) {
pendingIdleTimers.add(ci);
return;
}
cancelIdleTimer(ci);
const timeoutMs = resolvedChannelIdleTimeoutMs(ci.executionEngine);
if (timeoutMs <= 0) {
await killChannelWithLog(ci, context);
return;
}
cancelIdleTimer();
idleTimer = setTimeout(() => {
idleTimer = undefined;
const idleTimer = setTimeout(() => {
idleTimers.delete(ci);
if (hasNoChannelWork(ci)) {
writeStderrLine(
`qwen serve: idle timeout (${timeoutMs}ms) expired, killing channel`,
);
void killChannelWithLog(ci, 'idle timeout');
} else {
pendingIdleTimers.add(ci);
}
}, timeoutMs);
idleTimers.set(ci, idleTimer);
idleTimer.unref();
}
function retireChannel(info: HarnessChannel, context: string) {
info.isDying = true;
cancelIdleTimer();
keepAliveUntil = 0;
cancelIdleTimer(info);
if (info.executionEngine === defaultEngine) keepAliveUntil = 0;
info.channelLiveness?.stop();
return terminateChannel(info.channel, initTimeoutMs, context);
}
@@ -218,7 +239,7 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
): Promise<T> {
if (options.isRuntimeStopping())
throw new WorkspaceDrainingError(options.boundWorkspace ?? '');
if (liveHarnessChannel() === ci) cancelIdleTimer();
if (liveHarnessChannel(ci.executionEngine) === ci) cancelIdleTimer(ci);
if (recordUse) ci.lastUsedAt = Date.now();
ci.workspaceControlInFlight++;
try {
@@ -233,45 +254,67 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
ci.workspaceControlInFlight - 1,
);
await reapPendingEmptyChannel(ci);
if (!ci.isDying && liveHarnessChannel() === ci && hasNoChannelWork(ci)) {
if (!ci.isDying && liveHarnessChannel(ci.executionEngine) === ci) {
await startIdleTimer(ci, 'workspace control');
}
}
}
async function settleReleasedRuntimeWork(
context: string,
armIdleTimer = true,
): Promise<void> {
for (const ci of Array.from(channelLifecycle.values())) {
async function settleReleasedRuntimeWork(context: string): Promise<void> {
const channels = Array.from(channelLifecycle.values());
for (const ci of channels) {
await reapPendingEmptyChannel(ci);
}
if (!armIdleTimer) return;
const ci = liveHarnessChannel();
if (ci && hasNoChannelWork(ci)) {
await startIdleTimer(ci, context);
for (const ci of channels) {
if (
!ci.isDying &&
pendingIdleTimers.has(ci) &&
!idleTimers.has(ci) &&
hasNoChannelWork(ci)
) {
await startIdleTimer(ci, context);
}
}
}
function reserveRuntimeOperation(
engine: BridgeExecutionEngine | undefined = defaultEngine,
): void {
runtimeOperationReservations.set(
engine,
(runtimeOperationReservations.get(engine) ?? 0) + 1,
);
}
function decrementRuntimeOperationReservation(
engine: BridgeExecutionEngine | undefined = defaultEngine,
): void {
const remaining = (runtimeOperationReservations.get(engine) ?? 0) - 1;
if (remaining > 0) runtimeOperationReservations.set(engine, remaining);
else runtimeOperationReservations.delete(engine);
}
async function releaseRuntimeOperationReservation(
context: string,
engine: BridgeExecutionEngine | undefined = defaultEngine,
): Promise<void> {
runtimeOperationReservations = Math.max(
0,
runtimeOperationReservations - 1,
);
decrementRuntimeOperationReservation(engine);
const channel = liveHarnessChannel(engine);
if (channel) pendingIdleTimers.add(channel);
await settleReleasedRuntimeWork(context);
}
/**
* Get-or-create the daemon's single `qwen --acp` channel. N sessions
* Get-or-create the selected engine's ACP channel. N sessions
* multiplex onto it via `connection.newSession()`. Concurrent callers
* coalesce through `inFlightChannelSpawn` so we never spawn two
* children. Wires up the one-and-only `channel.exited` cleanup on
* first creation so the late-arriving event tears down ALL
* multiplexed sessions.
*/
async function ensureChannel(): Promise<HarnessChannel> {
async function ensureChannel(
engine: BridgeExecutionEngine | undefined = defaultEngine,
): Promise<HarnessChannel> {
if (options.isRuntimeStopping())
throw new WorkspaceDrainingError(options.boundWorkspace ?? '');
if (isShuttingDown()) {
@@ -281,16 +324,20 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
// mid-SIGTERM-or-already-dead and `connection.newSession()` on it
// would either hang or land the caller with a sessionId that
// immediately 404s on every follow-up.
cancelIdleTimer();
if (channelLifecycle.current && !channelLifecycle.current.isDying)
return channelLifecycle.current;
if (channelLifecycle.starting) return await channelLifecycle.starting;
const current = channelLifecycle.currentFor(engine);
if (current) cancelIdleTimer(current);
if (current && !current.isDying) return current;
const starting = channelLifecycle.startingFor(engine);
if (starting) return await starting;
const promise = channelLifecycle.startSpawn(channelStartup.start);
const promise = channelLifecycle.startSpawn(
() => channelStartup.start(engine),
engine,
);
try {
return await promise;
} finally {
channelLifecycle.finishSpawn();
channelLifecycle.finishSpawn(engine);
}
}
@@ -298,7 +345,7 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
if (isShuttingDown()) {
throw new Error('AcpSessionBridge is shutting down');
}
runtimeOperationReservations++;
reserveRuntimeOperation();
const rawKeepAliveMs = options?.keepAliveMs;
const keepAliveMs =
rawKeepAliveMs !== undefined &&
@@ -314,12 +361,14 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
Date.now() + keepAliveMs,
);
}
let channel: HarnessChannel | undefined;
try {
await telemetry.withSpan(
'channel.preheat',
{ 'qwen-code.daemon.bridge.operation': 'channel.preheat' },
async () => {
const info = await ensureChannel();
channel = info;
info.lastUsedAt = Date.now();
if (keepAliveMs !== undefined) {
keepAliveUntil = Math.max(keepAliveUntil, Date.now() + keepAliveMs);
@@ -330,14 +379,15 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
if (pendingKeepAliveToken) {
pendingKeepAliveDeadlines.delete(pendingKeepAliveToken);
}
runtimeOperationReservations = Math.max(
0,
runtimeOperationReservations - 1,
);
await settleReleasedRuntimeWork(
'channel preheat',
resolvedChannelIdleTimeoutMs() > 0,
);
decrementRuntimeOperationReservation();
if (
channel &&
channelLifecycle.has(channel) &&
resolvedChannelIdleTimeoutMs() > 0
) {
pendingIdleTimers.add(channel);
}
await settleReleasedRuntimeWork('channel preheat');
}
};
@@ -348,7 +398,7 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
handleChannelExit(info, exitInfo) {
info.channelLiveness?.stop();
options.handleChannelTransportUnavailable(info);
if (channelLifecycle.current === info) cancelIdleTimer();
cancelIdleTimer(info);
options.beforeChannelExit(info);
channelLifecycle.remove(info);
options.handleChannelExit(info, exitInfo);
@@ -362,11 +412,17 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
get starting() {
return channelLifecycle.starting;
},
currentFor: channelLifecycle.currentFor,
startingFor: channelLifecycle.startingFor,
startups: channelLifecycle.startups,
get epoch() {
return channelStartup.epoch;
},
get runtimeOperationReservations() {
return runtimeOperationReservations;
return Array.from(runtimeOperationReservations.values()).reduce(
(total, count) => total + count,
0,
);
},
get pendingKeepAliveCount() {
return pendingKeepAliveDeadlines.size;
@@ -434,9 +490,7 @@ export function createChannelHarness(options: ChannelHarnessOptions) {
hasNoChannelWork,
reapPendingEmptyChannel,
withWorkspaceControl,
reserveRuntimeOperation() {
runtimeOperationReservations++;
},
reserveRuntimeOperation,
releaseRuntimeOperationReservation,
settleReleasedRuntimeWork,
preheat,
+39 -15
View File
@@ -8,9 +8,11 @@ import type { ClientSideConnection } from '@agentclientprotocol/sdk';
import type { ActiveWorkHoldCategory, ChildHeapReport } from './bridgeTypes.js';
import type { AcpChannel } from './channel.js';
import type { ChannelLivenessMonitor } from './channel-liveness.js';
import type { BridgeExecutionEngine } from './bridgeOptions.js';
export interface HarnessChannel {
readonly id: string;
readonly executionEngine?: BridgeExecutionEngine;
lastUsedAt: number;
readonly channel: AcpChannel;
readonly connection: ClientSideConnection;
@@ -112,17 +114,27 @@ export interface ChannelLifecycle {
readonly current: HarnessChannel | undefined;
readonly starting: Promise<HarnessChannel> | undefined;
readonly size: number;
currentFor(engine?: BridgeExecutionEngine): HarnessChannel | undefined;
startingFor(
engine?: BridgeExecutionEngine,
): Promise<HarnessChannel> | undefined;
startups(): IterableIterator<Promise<HarnessChannel>>;
has(info: HarnessChannel): boolean;
values(): IterableIterator<HarnessChannel>;
track(info: HarnessChannel): void;
publish(info: HarnessChannel): void;
remove(info: HarnessChannel): void;
startSpawn(spawn: () => Promise<HarnessChannel>): Promise<HarnessChannel>;
finishSpawn(): void;
startSpawn(
spawn: () => Promise<HarnessChannel>,
engine?: BridgeExecutionEngine,
): Promise<HarnessChannel>;
finishSpawn(engine?: BridgeExecutionEngine): void;
}
export function createChannelLifecycle(): ChannelLifecycle {
// `channelInfo` is the SINGLE attach-available channel. Cleared
export function createChannelLifecycle(
defaultEngine?: BridgeExecutionEngine,
): ChannelLifecycle {
// Each engine slot has one attach-available channel. Cleared
// ONLY by the `channel.exited` handler in channel-harness.ts when the OS
// reaps the underlying child process. Teardown initiators
// (`killSession` last-session-leaving — via `startIdleTimer` ->
@@ -138,7 +150,7 @@ export function createChannelLifecycle(): ChannelLifecycle {
// handshake completes). Race-aware code paths (`ensureChannel`,
// `killAllSync`) gate on `isDying` rather than presence; see
// `HarnessChannel.isDying` for the per-set-site rationale.
let channelInfo: HarnessChannel | undefined;
const channels = new Map<BridgeExecutionEngine | undefined, HarnessChannel>();
// BkUyD: superset of `channelInfo` covering channels
// that are dying but not yet OS-reaped. `killSession` /
// `doSpawn`-newSession-failure / `shutdown` mark a channel as
@@ -156,16 +168,22 @@ export function createChannelLifecycle(): ChannelLifecycle {
const aliveChannels = new Set<HarnessChannel>();
// Coalesces a concurrent second `ensureChannel()` call onto the
// first one's spawn so we never create two children for the same
// daemon. Cleared in the `finally` of the creator.
let inFlightChannelSpawn: Promise<HarnessChannel> | undefined;
// engine. Cleared in the `finally` of the creator.
const starting = new Map<
BridgeExecutionEngine | undefined,
Promise<HarnessChannel>
>();
return {
get current() {
return channelInfo;
return channels.get(defaultEngine);
},
get starting() {
return inFlightChannelSpawn;
return starting.get(defaultEngine);
},
currentFor: (engine = defaultEngine) => channels.get(engine),
startingFor: (engine = defaultEngine) => starting.get(engine),
startups: () => starting.values(),
get size() {
return aliveChannels.size;
},
@@ -179,17 +197,23 @@ export function createChannelLifecycle(): ChannelLifecycle {
aliveChannels.add(info);
},
publish(info) {
channelInfo = info;
channels.set(info.executionEngine, info);
},
remove(info) {
aliveChannels.delete(info);
if (channelInfo === info) channelInfo = undefined;
if (channels.get(info.executionEngine) === info)
channels.delete(info.executionEngine);
},
startSpawn(spawn) {
return (inFlightChannelSpawn ??= spawn());
startSpawn(spawn, engine = defaultEngine) {
let promise = starting.get(engine);
if (!promise) {
promise = spawn();
starting.set(engine, promise);
}
return promise;
},
finishSpawn() {
inFlightChannelSpawn = undefined;
finishSpawn(engine = defaultEngine) {
starting.delete(engine);
},
};
}
+28 -10
View File
@@ -16,6 +16,8 @@ import type { HarnessChannel, ChannelLifecycle } from './channel-lifecycle.js';
import type {
BridgeRuntimeEpochSource,
BridgeTelemetry,
BridgeExecutionEngine,
BridgeOptions,
} from './bridgeOptions.js';
import type { NdJsonQueueLimitError } from './ndJsonStream.js';
import {
@@ -42,6 +44,10 @@ import { writeStderrLine } from './internal/stderrLine.js';
export interface ChannelStartupOptions {
channelFactory: ChannelFactory;
executionEngines?: Pick<
NonNullable<BridgeOptions['executionEngines']>,
'legacy' | 'managed'
>;
boundWorkspace: string;
childEnvOverrides: Readonly<Record<string, string | undefined>>;
initTimeoutMs: number;
@@ -52,7 +58,11 @@ export interface ChannelStartupOptions {
delegateReadTextFileToClient: boolean;
isExternalToolGuardRequired(): boolean;
isShuttingDown(): boolean;
constructHarnessChannel(channel: AcpChannel, id: string): HarnessChannel;
constructHarnessChannel(
channel: AcpChannel,
id: string,
engine?: BridgeExecutionEngine,
): HarnessChannel;
handleChannelTransportUnavailable(info: HarnessChannel): void;
handleChannelExit(
info: HarnessChannel,
@@ -101,6 +111,7 @@ function safeTransportFailureDetail(error: unknown): string | undefined {
export function createChannelStartup({
channelFactory,
executionEngines,
boundWorkspace,
childEnvOverrides,
initTimeoutMs,
@@ -119,7 +130,11 @@ export function createChannelStartup({
}: ChannelStartupOptions) {
let runtimeEpoch = initialRuntimeEpoch;
async function start(): Promise<HarnessChannel> {
async function start(
engine?: BridgeExecutionEngine,
): Promise<HarnessChannel> {
const factory =
engine === undefined ? channelFactory : executionEngines![engine];
const privateParentCapability = randomBytes(32).toString('base64url');
const acpChannelId = randomUUID();
const startupStartedAt = Date.now();
@@ -133,7 +148,7 @@ export function createChannelStartup({
'qwen-code.daemon.acp_channel.id': acpChannelId,
},
async () =>
await channelFactory(
await factory(
boundWorkspace,
{
...childEnvOverrides,
@@ -173,7 +188,7 @@ export function createChannelStartup({
}
let info: HarnessChannel;
try {
info = constructHarnessChannel(channel, acpChannelId);
info = constructHarnessChannel(channel, acpChannelId, engine);
} catch (error) {
try {
channel.killSync();
@@ -219,18 +234,21 @@ export function createChannelStartup({
channelLifecycle.track(info);
// Belt-and-suspenders leak detection. The set is intentionally
// multi-entry to cover the `killSession`-then-`spawnOrAttach`
// overlap window (size 2 is legitimate: one dying + one fresh
// overlap window (size 2 per engine is legitimate: one dying + one fresh
// attach-target). Anything higher implies a `channel.exited`
// handler never fired for some prior channel — a real leak we'd
// otherwise notice only as gradually-growing RSS over hours.
// The warning surfaces it the moment it happens. Threshold is
// 2 because that's the design ceiling; bumping it requires
// 2 per engine because that's the design ceiling; bumping it requires
// updating both this guard and the comments around
// `aliveChannels` declaration.
if (channelLifecycle.size > 2) {
const engineChannelCount = [...channelLifecycle.values()].filter(
(candidate) => candidate.executionEngine === engine,
).length;
if (engineChannelCount > 2) {
writeStderrLine(
`qwen serve: WARNING aliveChannels.size=${channelLifecycle.size} ` +
`(expected 1, max 2 during killSession-then-spawnOrAttach ` +
`qwen serve: WARNING engine=${engine ?? 'single'} channelCount=${engineChannelCount} ` +
`(expected 1, max 2 per engine during killSession-then-spawnOrAttach ` +
`overlap) — possible channel leak; check that prior channels' ` +
`channel.exited fired and the handler ran cleanup.`,
);
@@ -420,7 +438,7 @@ export function createChannelStartup({
}),
onFailure: failChannelLiveness,
isActive: () =>
channelLifecycle.current === info &&
channelLifecycle.currentFor(engine) === info &&
channelLifecycle.has(info) &&
!info.isDying &&
!isShuttingDown(),
File diff suppressed because it is too large. Load diff
File diff suppressed because it is too large. Load diff
@@ -218,6 +218,16 @@ describe('qwen serve Live Host discovery', () => {
await fs.readFile(runtimeDiscoveryPath),
{ mode: 0o600 },
);
const kill = process.kill.bind(process);
const killSpy = vi
.spyOn(process, 'kill')
.mockImplementation((pid, signal) => {
if (pid === 999_999 && signal === 0) {
throw Object.assign(new Error('No such process'), { code: 'ESRCH' });
}
return kill(pid, signal);
});
let handle: Awaited<ReturnType<typeof runQwenServe>> | undefined;
try {
handle = await runQwenServe(
@@ -262,6 +272,7 @@ describe('qwen serve Live Host discovery', () => {
).toBe(true);
} finally {
await handle?.close();
killSpy.mockRestore();
if (previousQwenHome === undefined) delete process.env['QWEN_HOME'];
else process.env['QWEN_HOME'] = previousQwenHome;
if (previousRuntimeDir === undefined)
+8
View File
@@ -1020,6 +1020,14 @@ describe('LiveDaemon', () => {
});
it('reclaims a stale dead-owner discovery record and republishes', async () => {
const kill = process.kill.bind(process);
vi.spyOn(process, 'kill').mockImplementation((pid, signal) => {
if (pid === 999_999 && signal === 0) {
throw Object.assign(new Error('No such process'), { code: 'ESRCH' });
}
return kill(pid, signal);
});
const config = await testConfig();
await plantDiscoveryRecord(config.discoveryDir, {
url: 'http://127.0.0.1:3210',
@@ -236,7 +236,7 @@ class ManagedAgentMySqlIT {
}
@ParameterizedTest
@ValueSource(strings = {"UTC", "+08:00"})
@ValueSource(strings = {"LOCAL", "UTC", "+08:00", "Asia/Shanghai"})
@Order(2)
void shortWriterLeaseKeepsSubsecondDatabasePrecision(String connectionTimeZone)
throws InterruptedException {
@@ -297,6 +297,7 @@ class ManagedAgentMySqlIT {
new AcquireWriterRequest("mysql-lease-workspace",
"mysql-lease-writer", 60_000L)));
assertThat(reacquired.writerGeneration()).isEqualTo(1);
assertThat(reacquired.replayed()).isTrue();
assertLeaseDeadline(jdbc, tenant, session, reacquired);
WriterGrant renewed = inTransaction(transactions,
@@ -317,6 +318,19 @@ class ManagedAgentMySqlIT {
"mysql-replacement-writer", 60_000L)));
assertThat(takenOver.writerGeneration()).isEqualTo(2);
assertLeaseDeadline(jdbc, tenant, session, takenOver);
inTransaction(transactions,
() -> store.sealWriter(tenant, session,
"dddddddddddddddddddddddddddddddd",
new SealWriterRequest("mysql-lease-workspace",
"mysql-replacement-writer", 2)));
WriterGrant afterSeal = inTransaction(transactions,
() -> store.acquireWriter(tenant, session,
"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
new AcquireWriterRequest("mysql-lease-workspace",
"mysql-post-seal-writer", 1_000L)));
assertThat(afterSeal.writerGeneration()).isEqualTo(3);
assertLeaseDeadline(jdbc, tenant, session, afterSeal);
}
private static void assertLeaseDeadline(JdbcTemplate jdbc, String tenant,
@@ -921,13 +921,13 @@ class HttpRuntimeTransportTest {
() -> {
switch (operation) {
case "execute" -> transport.execute(lease, session,
reference).toCompletableFuture().get(2,
reference).toCompletableFuture().get(5,
TimeUnit.SECONDS);
case "status" -> transport.status(lease, session,
reference, 0).toCompletableFuture().get(2,
reference, 0).toCompletableFuture().get(5,
TimeUnit.SECONDS);
case "cancel" -> transport.cancel(lease, session,
reference).toCompletableFuture().get(2,
reference).toCompletableFuture().get(5,
TimeUnit.SECONDS);
default -> throw new AssertionError(
"unknown operation");
+3
View File
@@ -57,6 +57,8 @@ const EN: Messages = {
'capacityChoice.resume': 'Resume conversation',
'capacityChoice.blocked.stopping': 'Stopping',
'capacityChoice.blocked.not_live': 'No live ACP',
'capacityChoice.blocked.multiple_engine_channels':
'Multiple execution engines are running',
'capacityChoice.blocked.release_unavailable':
'Owned process release cannot be observed',
'capacityChoice.blocked.session_start_pending':
@@ -4174,6 +4176,7 @@ const ZH: Messages = {
'capacityChoice.resume': '恢复会话',
'capacityChoice.blocked.stopping': '正在停止',
'capacityChoice.blocked.not_live': '没有运行中的 ACP',
'capacityChoice.blocked.multiple_engine_channels': '多个执行引擎正在运行',
'capacityChoice.blocked.release_unavailable': '无法确认自有进程释放',
'capacityChoice.blocked.session_start_pending': '正在启动或恢复会话',
'capacityChoice.blocked.workspace_control_pending':