diff --git a/.env.example b/.env.example index 040344d210..9d83075f09 100644 --- a/.env.example +++ b/.env.example @@ -63,9 +63,9 @@ COLLECTIVE_GITHUB_CLIENT_SECRET= # Default Service opened by the Clowder "Enter Collective" launch surface. NEXT_PUBLIC_COLLECTIVE_SERVICE_URL=http://localhost:5201 -# CLI inactivity timeout in milliseconds. -# Default is 1800000 = 30 minutes; set to 0 to disable the CLI-layer timeout. -# CLI_TIMEOUT_MS=1800000 +# CLI inactivity timeout in milliseconds: a member with no output for this long is stopped. +# Default is 0 = no timeout (only a manual stop ends a silent turn). Example: 3600000 = 60 minutes. +# CLI_TIMEOUT_MS=3600000 # ── Workspace Access 工作目录 ─────────────────────────────── # Comma-separated list of directories the MCP server is allowed to access. diff --git a/docs/configuration/environment.md b/docs/configuration/environment.md index 7d5c753c8c..73a092d30d 100644 --- a/docs/configuration/environment.md +++ b/docs/configuration/environment.md @@ -60,4 +60,4 @@ The `.env` file supports key variables as a legacy fallback, but the UI approach |---------------------------|------------------|-------------| | `NEXT_PUBLIC_API_URL` | *(auto-derived)* | API URL used by the frontend. Normally computed automatically from `API_SERVER_PORT`; override only if you use a reverse proxy or custom domain | | `NEXT_PUBLIC_BRAND_NAME` | `Clowder AI` | Brand name shown in the UI (界面显示的品牌名称) | -| `CLI_TIMEOUT_MS` | `1800000` | CLI inactivity timeout in milliseconds (30 minutes). Set to `0` to disable (CLI 不活跃超时，0 为禁用) | +| `CLI_TIMEOUT_MS` | `0` | Member no-output timeout in milliseconds: a member with no output (text, tool calls, thinking) for this long is stopped as if by Stop, and its reply fails as timed out; a member whose process is still using CPU gets at most twice as long. `0` (default) disables it, so only a manual stop ends a silent turn (成员无输出超时，默认 0 即不超时) | diff --git a/docs/configuration/environment.zh-CN.md b/docs/configuration/environment.zh-CN.md index b5cffbb0d3..901dc1aa5b 100644 --- a/docs/configuration/environment.zh-CN.md +++ b/docs/configuration/environment.zh-CN.md @@ -60,4 +60,4 @@ API 密钥最好在启动后通过 UI 管理： |---------------------------|------------------|-------------| | `NEXT_PUBLIC_API_URL` | *(自动推导)* | 前端使用的 API URL。通常会根据 `API_SERVER_PORT` 自动计算；仅在使用反向代理或自定义域名时才需覆盖 | | `NEXT_PUBLIC_BRAND_NAME` | `Clowder AI` | UI 中显示的品牌名称 | -| `CLI_TIMEOUT_MS` | `1800000` | CLI 不活跃超时，单位毫秒（30 分钟）。设为 `0` 可禁用 | +| `CLI_TIMEOUT_MS` | `0` | 成员无输出超时，单位毫秒：一个成员这么久没有任何输出（文字、工具调用、思考）就按「停止」停下，回复记为超时失败；进程仍在占用 CPU 时最多顺延到 2 倍。默认 `0`，即不超时，静默的一轮只能手动停止 | diff --git a/docs/features/F117-message-delivery-lifecycle.md b/docs/features/F117-message-delivery-lifecycle.md index f200f03214..379fb19af1 100644 --- a/docs/features/F117-message-delivery-lifecycle.md +++ b/docs/features/F117-message-delivery-lifecycle.md @@ -1077,6 +1077,31 @@ Antigravity、PTY 五个 carrier；而且在默认 `CLI_TIMEOUT_MS=0` 下，这 都没有时不算正在处理。快照不完整时，没有任何证据的 running 记录也保守列出，保持 AC-E7 的契约。复审建议的「待对账」与 KD-10 / AC-E7 冲突，没有采用：无法核实时保守地显示为运行中， 由停止（AC-E7）或下一次完整快照时的 read-repair 收尾（见 J3「为什么不单列『待对账』」）。这一取舍待砚砚复审确认。 +#### J4 实现（路线 Phase 4，2026-09-27，基于 develop_base `5bd5e008e`） + +按上文 J4 落地，代码锚点对照过：外层 2× 定时器、cli-spawn 的无输出定时器、#774、Codex app-server 空闲中断、 +tmux 空闲、AGY `--print-timeout` 都还在设计写的位置；生产派发只有 QueueProcessor → `routeExecution` 一个入口， +stop hook 从这里注入即覆盖全部。和设计的出入与补充： + +- **定时器**：`MemberOutputTimeout`（`member-output-timeout.ts`）在 invoke-single-cat 启动，只有 route 传入 + `onMemberTimeout` 时才布防（生产都经 QueueProcessor 的 `stopMember`）。顺延看进程是否仍在占 CPU + （`ProcessLivenessProbe.activity()`，按两次采样的 CPU 增长，与最近是否有 stdout 无关），由 cli-spawn 按 invocation + 登记（`process-activity-registry.ts`）。没有登记进程的接入不顺延，其中包括 **SDK 接入**：它的引擎子进程由 SDK + 自己拉起，拿不到进程号。以前 SDK 接入只受外层 2× 定时器约束（3001 上是 60 分钟），现在是 `CLI_TIMEOUT_MS` + （30 分钟）无输出就停；Claude Code 的 Bash 工具本身 10 分钟封顶，更长的命令本来就该交给托管命令。 +- **触发 = Stop**：定时器触发时先把诊断交给 route（abort 之后的事件会被 route 丢弃），再由 QueueProcessor 的 + `createMemberTimeoutStop` 同步比对 `getExecutionId === executionId`（父执行 ID）后 `cancel(…, 'timeout')`。 +- **结局**：R 的终态规则收成一处 `resolveResponseTerminal`（原来 route-serial 两份、route-parallel 一份），`timeout` + → failed / `timeout`，R 带 `timeoutDiagnostics` 和失败文字。R 提交后 route 补发一条带诊断的 error 事件， + QueueProcessor 因此把该成员记为失败：`resolveFinalStatus` 与 disposition 的 `isCanceled` 都不把 timeout 墓碑 + 算作用户取消。TurnExecution failed / `timeout`，路由信号 `timeout` → `provider_timeout`；旧记录里的 + `invocation_timeout` 仍按 `provider_timeout` 读。 +- **删除范围**：cli-spawn 与 tmux 的超时不再默认读 `CLI_TIMEOUT_MS`，只给显式传 `timeoutMs` 的调用（例如 opencode + 自动审批探测）；成员派发从不传。AGY 不再传 `--print-timeout`（其默认 0，等整轮结束）。Codex app-server 不再传 + 空闲超时，客户端的 `timeoutMs` 选项已无调用方，留待后续清理。 +- **#774**：重试条件收窄为「未收到首帧」的启动超时（`isCliStartupTimeoutError`），保留 tmux 首事件看门狗在 resume + 时的重试；成员开始输出之后的静默不再被吞掉重跑。 + ### Phase K（路线 Phase 2b：Claude Agent SDK 接入，2026-09-25） diff --git a/docs/features/F254-side-effect-freshness-gate.md b/docs/features/F254-side-effect-freshness-gate.md index d590740cd8..64bc46f1bb 100644 --- a/docs/features/F254-side-effect-freshness-gate.md +++ b/docs/features/F254-side-effect-freshness-gate.md @@ -517,7 +517,7 @@ F233 projector 可选读取此流做聚合报告（通过 `FreshnessAttentionEve **(b) Per-Invocation Operational State**（hot path / 决策）： -Redis-backed per-invocation counters（TTL = invocation timeout，如 30min，自动清理）： +Redis-backed per-invocation counters（TTL 30min，自动清理；与 `CLI_TIMEOUT_MS` 无关，后者默认 0 即不超时）： ```typescript // key: `freshness:state:{invocationId}` diff --git a/packages/api/src/config/env-registry.ts b/packages/api/src/config/env-registry.ts index 0617209349..5f7a97bca1 100644 --- a/packages/api/src/config/env-registry.ts +++ b/packages/api/src/config/env-registry.ts @@ -857,7 +857,7 @@ export const ENV_VARS: EnvDefinition[] = [ { name: 'CLI_TIMEOUT_MS', defaultValue: DEFAULT_CLI_TIMEOUT_LABEL, - description: 'CLI 自动终止超时（0 = 关闭，仅人工取消）', + description: '成员无输出超时：到时按停止处理，回复记为超时失败（占用 CPU 时最多 2 倍；0 = 关闭，仅人工停止）', category: 'cli', sensitive: false, runtimeEditable: false, diff --git a/packages/api/src/domains/cats/services/agents/invocation/InvocationTracker.ts b/packages/api/src/domains/cats/services/agents/invocation/InvocationTracker.ts index edc5f091d5..d6943c8679 100644 --- a/packages/api/src/domains/cats/services/agents/invocation/InvocationTracker.ts +++ b/packages/api/src/domains/cats/services/agents/invocation/InvocationTracker.ts @@ -13,6 +13,7 @@ import type { LifecycleActiveRun } from '@cat-cafe/shared'; import { createModuleLogger } from '../../../../../infrastructure/logger.js'; import type { AgentClientActiveRunDispatcher } from '../../types.js'; +import { MEMBER_TIMEOUT_REASON } from './member-output-timeout.js'; const log = createModuleLogger('invocation-tracker'); export const DEFAULT_INVOCATION_SLOT_TTL_MS = 75 * 60_000; @@ -687,6 +688,12 @@ export class InvocationTracker { return inv.state; } + /** F117 KD-22: a member stopped by its output timeout failed; nobody cancelled it. */ + isTimedOut(threadId: string, catId: string): boolean { + const inv = this.active.get(this.slotKey(threadId, catId)); + return inv?.state === 'canceled' && inv.cancelReason === MEMBER_TIMEOUT_REASON; + } + /** Exact canceled-slot fence for read repair; slot state alone is too broad. */ getCanceledSlotIdentity(threadId: string, catId: string): { executionId: string; userId: string } | undefined { const inv = this.active.get(this.slotKey(threadId, catId)); @@ -700,7 +707,8 @@ export class InvocationTracker { * - whole-invocation abort (batch gate aborted: cancelAll / force / thread-delete / preempt) * → 'canceled_by_user' (user_cancel/cancel_all reason) or 'canceled' (other reasons) * - else if EVERY target cat is a canceled tombstone → 'canceled_by_user' (cancelled cat-by-cat) - * - else → 'succeeded' (at least one cat ran to completion) + * - else → 'succeeded': not cancelled, so each cat's own outcome decides. A cat stopped by its + * output timeout (F117 KD-22) failed — it is not a cancellation, so it never counts here. * `controller.signal.aborted` alone now means ONLY whole-invocation abort — a single-cat cancel * no longer aborts the batch gate, so callers must use this aggregate rather than raw `.aborted`. */ @@ -713,7 +721,9 @@ export class InvocationTracker { return batch.reason === 'user_cancel' || batch.reason === 'cancel_all' ? 'canceled_by_user' : 'canceled'; } if (targetCats.length === 0) return 'succeeded'; - const allCanceled = targetCats.every((c) => this.getSlotState(threadId, c) === 'canceled'); + const allCanceled = targetCats.every( + (c) => this.getSlotState(threadId, c) === 'canceled' && !this.isTimedOut(threadId, c), + ); return allCanceled ? 'canceled_by_user' : 'succeeded'; } diff --git a/packages/api/src/domains/cats/services/agents/invocation/QueueProcessor.ts b/packages/api/src/domains/cats/services/agents/invocation/QueueProcessor.ts index bec7153fef..3f54eaf737 100644 --- a/packages/api/src/domains/cats/services/agents/invocation/QueueProcessor.ts +++ b/packages/api/src/domains/cats/services/agents/invocation/QueueProcessor.ts @@ -122,6 +122,7 @@ import { } from './InvocationTracker.js'; import { projectLifecycleAppendAction } from './lifecycle-append-projection.js'; import { emitLifecycleMessageUpdated } from './lifecycle-message-update.js'; +import { createMemberTimeoutStop } from './member-output-timeout.js'; import { requireOwnerAuthProvenance } from './owner-auth-provenance.js'; import { isTerminalDispositionEvent, @@ -180,6 +181,10 @@ interface TrackerLike { ): boolean; has(threadId: string, catId?: string): boolean; cancelInvocation(threadId: string, catIds: string[], userId?: string, reason?: string): unknown; + /** F117 KD-22: stop one member the way Stop does (the member timeout passes reason `timeout`). */ + cancel?(threadId: string, catId: string, requestUserId?: string, abortReason?: string): { cancelled: boolean }; + /** F117 KD-22: the member was stopped by its output timeout — a failure, not a cancellation. */ + isTimedOut?(threadId: string, catId: string): boolean; getUserId?(threadId: string, catId: string): string | null; getExecutionId?(threadId: string, catId: string): string | undefined; /** F-parallel-cancel: expose a slot's own controller for per-cat cancel isolation. */ @@ -3425,7 +3430,10 @@ export class QueueProcessor { let lifecycleClaimRestorePromise: Promise | undefined; const terminalDispositions = new PerCatTerminalDispositionCollector({ targetCatIds: targetCats, - isCanceled: (catId) => invocationTracker.getSlotState?.(threadId, catId) === 'canceled', + // A member stopped by its output timeout failed (F117 KD-22); its route reports the failure. + isCanceled: (catId) => + invocationTracker.getSlotState?.(threadId, catId) === 'canceled' && + !invocationTracker.isTimedOut?.(threadId, catId), }); let responseText = ''; // F122B B6: completion hooks are registered before drain; keep their @@ -4155,6 +4163,22 @@ export class QueueProcessor { // abort (cancelAll / force / thread-delete), never on single-cat cancel — the sibling // keeps streaming. (See InvocationTracker.startAll returning a fresh batchController.) signalForCat: (catId: string) => invocationTracker.getController?.(threadId, catId)?.signal, + // F117 KD-22 (J4): a member's own invocation owns its output timeout; when it fires, the + // Queue stops only that member, and only while its slot still runs this execution. + ...(invocationTracker.getExecutionId && invocationTracker.cancel + ? { + stopMember: createMemberTimeoutStop({ + invocationTracker: { + getExecutionId: (tid, catId) => invocationTracker.getExecutionId?.(tid, catId), + cancel: (tid, catId, requestUserId, abortReason) => + invocationTracker.cancel?.(tid, catId, requestUserId, abortReason) ?? { cancelled: false }, + }, + threadId, + ownerUserId: userId, + log, + }), + } + : {}), getQueuedFreshnessMessagesForCat: (tid: string, uid: string, catId: string, parentInvocationId?: string) => queue.getQueuedFreshnessMessagesForCat(tid, uid, catId, { excludeEntryId: entry.id, parentInvocationId }), commitCompletedA2AWake: (input: Parameters>[0]) => diff --git a/packages/api/src/domains/cats/services/agents/invocation/invoke-helpers.ts b/packages/api/src/domains/cats/services/agents/invocation/invoke-helpers.ts index 6ee5b0be30..14be8c7010 100644 --- a/packages/api/src/domains/cats/services/agents/invocation/invoke-helpers.ts +++ b/packages/api/src/domains/cats/services/agents/invocation/invoke-helpers.ts @@ -140,6 +140,15 @@ export function isCliTimeoutError(message: string | undefined): boolean { return /CLI (?:响应超时|idle-silent 超时)/i.test(message); } +/** + * A CLI timeout before its first event (a startup watchdog such as tmux's): the member never came + * up. Silence after it started is the member output timeout's (F117 KD-22), which stops the member + * instead of surfacing here. + */ +export function isCliStartupTimeoutError(message: string | undefined): boolean { + return isCliTimeoutError(message) && /未收到首帧/.test(message ?? ''); +} + /** F215: Detect malformed tool-call error emitted by ClaudeAgentService (form A / B). * Used in invoke-single-cat to trigger seal+fresh-context+46接力 fallback chain. */ export function isMalformedToolCallError(message: string | undefined): boolean { diff --git a/packages/api/src/domains/cats/services/agents/invocation/invoke-single-cat.ts b/packages/api/src/domains/cats/services/agents/invocation/invoke-single-cat.ts index 91fbda9e0f..e8f9101d4a 100644 --- a/packages/api/src/domains/cats/services/agents/invocation/invoke-single-cat.ts +++ b/packages/api/src/domains/cats/services/agents/invocation/invoke-single-cat.ts @@ -79,6 +79,7 @@ import { resolveCliCommand } from '../../../../../utils/cli-resolve.js'; import { resolveCliTimeoutMs } from '../../../../../utils/cli-timeout.js'; import { findMonorepoRoot, isSameProject } from '../../../../../utils/monorepo-root.js'; import { resolvePersistentProjectPathDetailed } from '../../../../../utils/persistent-project-path.js'; +import { readProcessActivity } from '../../../../../utils/process-activity-registry.js'; import { pathsEqual } from '../../../../../utils/project-path.js'; import { tcpProbe } from '../../../../../utils/tcp-probe.js'; import { estimateTokens } from '../../../../../utils/token-counter.js'; @@ -602,7 +603,7 @@ import type { ResumeFailureKind } from './invoke-helpers.js'; import { classifyResumeFailure, extractTaskProgress, - isCliTimeoutError, + isCliStartupTimeoutError, isContextWindowOverflowError, isMalformedToolCallError, isMissingClaudeSessionError, @@ -612,6 +613,7 @@ import { isTransientCliExitCode1, preflightRace, } from './invoke-helpers.js'; +import { MEMBER_TIMEOUT_REASON, MemberOutputTimeout, type MemberTimeoutEvent } from './member-output-timeout.js'; import type { TaskProgressItem, TaskProgressStatus, TaskProgressStore } from './TaskProgressStore.js'; import { assertToolExecutionPolicySupported } from './tool-execution-policy.js'; @@ -1276,6 +1278,11 @@ export interface InvocationParams { readonly contentBlocks?: readonly MessageContent[]; readonly uploadDir?: string; readonly signal?: AbortSignal; + /** + * F117 KD-22 (J4): arms this member's output timeout (`CLI_TIMEOUT_MS`). Called once when it + * fires; the route keeps the diagnostics and has the Queue stop the member. Absent → no timeout. + */ + readonly onMemberTimeout?: (timeout: MemberTimeoutEvent) => void; readonly isLastCat: boolean; /** Static identity prompt — prepended to prompt on new sessions (gated by F-BLOAT logic) */ readonly systemPrompt?: string; @@ -1542,31 +1549,23 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP const triggerType = params.a2aTriggerMessageId ? 'mention' : params.parentInvocationId ? 'routing' : 'default'; catInvocationCount.add(1, { [AGENT_ID]: catId, [TRIGGER]: triggerType }); - // F089: Optional invocation-level timeout — independent of the NDJSON stream timeout. - // CLI_TIMEOUT_MS=0 means manual-cancel-only, so no automatic outer timer is armed. - const INVOCATION_TIMEOUT_MULTIPLIER = 2; - const cliTimeoutMs = resolveCliTimeoutMs(undefined); - const invocationTimeoutMs = cliTimeoutMs * INVOCATION_TIMEOUT_MULTIPLIER; - const invocationAc = new AbortController(); - let invocationTimer: ReturnType | null = null; - const resetInvocationTimeout = (): void => { - if (invocationTimer) clearTimeout(invocationTimer); - if (invocationTimeoutMs <= 0) { - invocationTimer = null; - return; - } - invocationTimer = setTimeout(() => { - log.error({ invocationId, catId, threadId, timeoutMs: invocationTimeoutMs }, 'Invocation hard timeout fired'); - invocationAc.abort(new Error('invocation_timeout')); - }, invocationTimeoutMs); - invocationTimer.unref(); - }; - resetInvocationTimeout(); - - // Merge caller signal (user cancel) with invocation timeout — neither loses semantics. - const signal: AbortSignal | undefined = callerSignal - ? AbortSignal.any([callerSignal, invocationAc.signal]) - : invocationAc.signal; + // F117 KD-22 (J4): this member's one timeout — no output for CLI_TIMEOUT_MS (0 = never). When it + // fires the member is stopped the way Stop stops it (reason `timeout`), through the caller's + // signal, so the run winds down through the same path as a Stop. + const onMemberTimeout = params.onMemberTimeout; + const memberTimeout = onMemberTimeout + ? new MemberOutputTimeout({ + timeoutMs: resolveCliTimeoutMs(undefined), + probeProcess: () => readProcessActivity(invocationId), + invocationId, + onTimeout: (diagnostics) => { + log.warn({ invocationId, catId, threadId, diagnostics }, 'Member output timeout fired; stopping the member'); + onMemberTimeout({ executionId: executionParentInvocationId, diagnostics }); + }, + }) + : undefined; + // A run without a caller signal (no Queue slot) can still be torn down only by its own end. + const signal: AbortSignal = callerSignal ?? new AbortController().signal; log.info({ invocationId, catId, threadId, userId }, 'Created invocation'); @@ -2107,9 +2106,9 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP source: participation, service, callbackEnv, - signal: signal!, + signal, })) { - resetInvocationTimeout(); + memberTimeout?.observe(message); if (message.type === 'error') hadError = true; if (message.type === 'done' && !hadError) turnExecutionCompletedSuccessfully = true; yield { ...message, turnInvocationId: invocationId, turnExecutionStartedAt: executionStartedAt }; @@ -5330,11 +5329,9 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP msg = projected; } if (currentRequestGenerationCommit) await observeCurrentRequestGeneration(msg); - // F149: provider_signal / liveness_signal must NOT reset timeout — prevents "续命" - // F198 Phase C P2-1: status (daemon detail progress) also must NOT reset timeout — - // a daemon sending frequent status updates must not evade the 30-min kill deadline. - if (msg.type !== 'provider_signal' && msg.type !== 'liveness_signal' && msg.type !== 'status') - resetInvocationTimeout(); + // F117 KD-22: only member output restarts the timeout; signals, status and diagnostics + // never do (F149, F198 Phase C), so a member cannot keep itself alive without working. + memberTimeout?.observe(msg); if (msg.contextCompaction) { // F117 K2: in-process hooks are authenticated by construction and registered by the // carrier itself; only a project hook depends on callback auth and the workspace files. @@ -5452,8 +5449,10 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP suppressedTransientCliError = msg; continue; } - // #774 self-heal: CLI timeout during session resume with no substantive output - // → likely stale/unreachable session. Suppress and retry without session. + // #774 self-heal: a resumed session whose CLI never produced its first event (startup + // watchdog) → likely stale/unreachable session. Suppress and retry without session. + // F117 KD-22: silence after the member started is not retried — its output timeout stops + // the member as an ordinary, resendable failure. // Uses attemptHasSubstantiveOutput (not attemptHasContentOutput) because // timeout_diagnostics (system_info) must NOT block the retry path. if ( @@ -5461,7 +5460,7 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP options.sessionId && !attemptHasSubstantiveOutput && msg.type === 'error' && - isCliTimeoutError(msg.error) + isCliStartupTimeoutError(msg.error) ) { suppressedTimeoutError = msg; continue; @@ -5928,8 +5927,9 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP didComplete = true; // F118 AC-C5: Normal completion reached } catch (err) { await closeActiveServiceIterator(); + // F117 KD-22: a member stopped by its output timeout failed; any other stop cancelled it. await terminateCurrentRequestGeneration( - signal?.aborted ? 'cancelled' : 'error', + signal.aborted && signal.reason !== MEMBER_TIMEOUT_REASON ? 'cancelled' : 'error', err instanceof Error ? err.message : String(err), ); // F152: Record error on invocation span + OTel log @@ -5979,26 +5979,29 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP tracing: { traceId: sc.traceId, spanId: sc.spanId, ...(parentSid ? { parentSpanId: parentSid } : {}) }, }; } finally { + memberTimeout?.close(); + // F117 KD-22: a member stopped by its output timeout failed; any other stop cancelled it. + const timedOut = callerSignal?.aborted === true && callerSignal.reason === MEMBER_TIMEOUT_REASON; await closeActiveServiceIterator(); await terminateCurrentRequestGeneration( turnExecutionCompletedSuccessfully ? 'accepted' - : callerSignal?.aborted || invocationAc.signal.aborted + : callerSignal?.aborted && !timedOut ? 'cancelled' - : hadError || turnExecutionFailureReason !== undefined + : timedOut || hadError || turnExecutionFailureReason !== undefined ? 'error' : 'unknown', - turnExecutionFailureReason ?? turnExecutionInterruptionReason, + timedOut ? MEMBER_TIMEOUT_REASON : (turnExecutionFailureReason ?? turnExecutionInterruptionReason), ); await presentationDelivery?.release('invocation_finalized_without_delivery'); if (deps.turnExecutionStore && ownsTurnExecution) { let terminal: TurnExecutionTerminalInput; if (turnExecutionCompletedSuccessfully) { terminal = { status: 'succeeded', endedAt: Date.now() }; + } else if (timedOut) { + terminal = { status: 'failed', endedAt: Date.now(), terminalReason: MEMBER_TIMEOUT_REASON }; } else if (callerSignal?.aborted) { terminal = { status: 'canceled', endedAt: Date.now(), terminalReason: 'user_cancel' }; - } else if (invocationAc.signal.aborted) { - terminal = { status: 'failed', endedAt: Date.now(), terminalReason: 'invocation_timeout' }; } else if (turnExecutionFailureReason !== undefined || hadError) { terminal = { status: 'failed', @@ -6056,9 +6059,6 @@ export async function* invokeSingleCat(deps: InvocationDeps, params: InvocationP // (abort / error / timeout). Mirrors PR #732 mention_dispatch abort-safety pattern. toolSpanTracker.endAllOrphans('aborted'); - // F089: Clear invocation hard timeout - if (invocationTimer) clearTimeout(invocationTimer); - // F118/#1329: Release runtime resume custody before conversation policy // custody. Every release is idempotent, including partial acquisition. for (const release of sessionMutexReleases.reverse()) release(); diff --git a/packages/api/src/domains/cats/services/agents/invocation/member-output-timeout.ts b/packages/api/src/domains/cats/services/agents/invocation/member-output-timeout.ts new file mode 100644 index 0000000000..d24b36e573 --- /dev/null +++ b/packages/api/src/domains/cats/services/agents/invocation/member-output-timeout.ts @@ -0,0 +1,146 @@ +/** + * F117 KD-22 (Phase J, J4): each dispatched member has exactly one timeout — no real output within + * `CLI_TIMEOUT_MS` (0 = never). Only output the model produced restarts it; heartbeats, status and + * diagnostics do not (F149). A member whose process is still using CPU gets more time, but never + * more than 2 × `CLI_TIMEOUT_MS` since its last output. When it fires, the owner stops the member + * the way Stop does, with reason {@link MEMBER_TIMEOUT_REASON}. + */ +import type { ProcessActivity } from '../../../../../utils/process-activity-registry.js'; +import type { AgentMessage, TimeoutDiagnostics } from '../../types.js'; + +/** The abort reason a member stopped by its output timeout carries (Stop carries `user_cancel`). */ +export const MEMBER_TIMEOUT_REASON = 'timeout'; + +/** The process behind a member when its timer fires; `absent` when there is no process to probe. */ +export type MemberProcessActivity = ProcessActivity | 'absent'; + +/** Raised by the member's own invocation when its timer fires, before the member is stopped. */ +export interface MemberTimeoutEvent { + /** The Queue execution that armed the timer (the invocation's parent execution id). */ + readonly executionId: string; + readonly diagnostics: TimeoutDiagnostics; +} + +/** Stop one member of a Queue execution with reason `timeout`; false when its slot moved on. */ +export type MemberTimeoutStop = (catId: string, executionId: string) => boolean; + +/** + * The Queue side of the timeout: stop the member the way Stop does, but only while its slot still + * runs the execution that armed the timer, so a timer left over from an earlier execution never + * stops the one that took the slot after it. Reading the slot and cancelling it is one synchronous + * step. The execution then winds itself down: the route commits the response, the Queue frees the + * slot, and nothing is broadcast as a user Stop. + */ +export function createMemberTimeoutStop(input: { + readonly invocationTracker: { + getExecutionId(threadId: string, catId: string): string | undefined; + cancel(threadId: string, catId: string, requestUserId?: string, abortReason?: string): { cancelled: boolean }; + }; + readonly threadId: string; + readonly ownerUserId: string; + readonly log: { info(obj: Record, msg: string): void }; +}): MemberTimeoutStop { + const { invocationTracker, threadId, ownerUserId, log } = input; + return (catId, executionId) => { + const current = invocationTracker.getExecutionId(threadId, catId); + if (current !== executionId) { + log.info( + { threadId, catId, executionId, current: current ?? null }, + '[member-timeout] the slot no longer runs the execution that timed out; leaving it alone', + ); + return false; + } + return invocationTracker.cancel(threadId, catId, ownerUserId, MEMBER_TIMEOUT_REASON).cancelled; + }; +} + +/** The failure text a timed-out member's response carries. */ +export function memberTimeoutErrorText(diagnostics: Pick): string { + const seconds = Math.max(1, Math.round(diagnostics.silenceDurationMs / 1000)); + const silence = seconds >= 60 ? `${Math.round(seconds / 60)} 分钟` : `${seconds} 秒`; + return `响应超时：${silence}没有任何输出，已停止。`; +} + +/** Output the model produced: text, tool calls and results, thinking and rich blocks. */ +export function isMemberOutput(message: Pick): boolean { + if (message.type === 'text' || message.type === 'tool_use' || message.type === 'tool_result') return true; + if (message.type !== 'system_info' || !message.content) return false; + try { + const payload = JSON.parse(message.content) as { type?: unknown } | null; + return payload?.type === 'thinking' || payload?.type === 'rich_block'; + } catch { + return false; + } +} + +export interface MemberOutputTimeoutOptions { + /** `CLI_TIMEOUT_MS`; 0 disables the timeout. */ + readonly timeoutMs: number; + /** Read when the timer fires; only `busy` defers it. */ + readonly probeProcess: () => MemberProcessActivity; + /** Called at most once, before the member is stopped, so the diagnostics outlive the abort. */ + readonly onTimeout: (diagnostics: TimeoutDiagnostics) => void; + readonly invocationId?: string; + readonly now?: () => number; +} + +export class MemberOutputTimeout { + readonly #options: MemberOutputTimeoutOptions; + readonly #now: () => number; + #lastOutputAt: number; + #firstOutputAt: number | undefined; + #lastOutputType: string | undefined; + #timer: ReturnType | undefined; + #finished = false; + + constructor(options: MemberOutputTimeoutOptions) { + this.#options = options; + this.#now = options.now ?? Date.now; + this.#lastOutputAt = this.#now(); + if (options.timeoutMs > 0) this.#arm(options.timeoutMs); + } + + /** Restart the clock if this message is member output. */ + observe(message: Pick): void { + if (this.#finished || this.#options.timeoutMs <= 0 || !isMemberOutput(message)) return; + const at = this.#now(); + this.#firstOutputAt ??= at; + this.#lastOutputAt = at; + this.#lastOutputType = message.type; + this.#arm(this.#options.timeoutMs); + } + + close(): void { + this.#finished = true; + if (this.#timer) clearTimeout(this.#timer); + this.#timer = undefined; + } + + #arm(delayMs: number): void { + if (this.#timer) clearTimeout(this.#timer); + this.#timer = setTimeout(() => this.#expire(), delayMs); + this.#timer.unref?.(); + } + + #expire(): void { + if (this.#finished) return; + const { timeoutMs } = this.#options; + const silenceMs = this.#now() - this.#lastOutputAt; + const activity = this.#options.probeProcess(); + const cap = 2 * timeoutMs; + if (activity === 'busy' && silenceMs < cap) { + this.#arm(Math.min(timeoutMs, cap - silenceMs)); + return; + } + this.close(); + this.#options.onTimeout({ + silenceDurationMs: silenceMs, + processAlive: activity !== 'dead', + ...(this.#lastOutputType ? { lastEventType: this.#lastOutputType } : {}), + ...(this.#firstOutputAt !== undefined + ? { firstEventAt: this.#firstOutputAt, lastEventAt: this.#lastOutputAt } + : {}), + ...(this.#options.invocationId ? { invocationId: this.#options.invocationId } : {}), + }); + } +} diff --git a/packages/api/src/domains/cats/services/agents/providers/CodexAgentService.ts b/packages/api/src/domains/cats/services/agents/providers/CodexAgentService.ts index 618acaa8c6..08a2305f00 100644 --- a/packages/api/src/domains/cats/services/agents/providers/CodexAgentService.ts +++ b/packages/api/src/domains/cats/services/agents/providers/CodexAgentService.ts @@ -63,7 +63,6 @@ import { spawnCli, withVerdictGhGuardEnv, } from '../../../../../utils/cli-spawn.js'; -import { parseCliTimeoutMs, resolveCliTimeoutMs } from '../../../../../utils/cli-timeout.js'; import type { SpawnFn } from '../../../../../utils/cli-types.js'; import { findMonorepoRoot } from '../../../../../utils/monorepo-root.js'; import { sanitizeCliStderr } from '../../../../../utils/sanitize-cli-stderr.js'; @@ -2185,7 +2184,8 @@ export class CodexAgentService implements AgentService { : {}), ...(options?.activeRunDispatch ? { activeRunDispatch: options.activeRunDispatch } : {}), ...(options?.signal ? { signal: options.signal } : {}), - timeoutMs: resolveCliTimeoutMs(parseCliTimeoutMs(codexEnv.CLI_TIMEOUT_MS ?? undefined)), + // F117 KD-22: no idle interrupt of its own — the member's one timeout is its + // invocation's, and it stops the turn through the signal above. interruptGraceMs: KILL_GRACE_MS, }, retryBudget: 1, diff --git a/packages/api/src/domains/cats/services/agents/providers/GeminiAgentService.ts b/packages/api/src/domains/cats/services/agents/providers/GeminiAgentService.ts index 8fe5b5c532..7178558d06 100644 --- a/packages/api/src/domains/cats/services/agents/providers/GeminiAgentService.ts +++ b/packages/api/src/domains/cats/services/agents/providers/GeminiAgentService.ts @@ -49,7 +49,6 @@ import { isLivenessWarning, spawnCli, } from '../../../../../utils/cli-spawn.js'; -import { resolveCliTimeoutMs } from '../../../../../utils/cli-timeout.js'; import type { SpawnFn } from '../../../../../utils/cli-types.js'; import { readJsonlTail } from '../../../../../utils/jsonl-tail-reader.js'; import { sanitizeCliStderr } from '../../../../../utils/sanitize-cli-stderr.js'; @@ -352,11 +351,6 @@ function readLatestGeminiContextTokens( .find((message) => matchesCurrentAssistantText(message.content, normalizedAssistantText))?.tokens?.input; } -function formatAgyPrintTimeout(timeoutMs: number): string | null { - if (timeoutMs <= 0) return null; - return `${Math.max(1, Math.ceil(timeoutMs / 1000))}s`; -} - function appendAgyMcpIdentityContract(prompt: string, catId: CatId, hasCallbackEnv: boolean): string { if (!hasCallbackEnv) return prompt; return [ @@ -942,8 +936,8 @@ export class GeminiAgentService implements AgentService { effectivePrompt = appendLocalImagePathHints(effectivePrompt, imagePaths); effectivePrompt = appendAgyMcpIdentityContract(effectivePrompt, this.catId, options?.callbackEnv !== undefined); - const timeoutMs = resolveCliTimeoutMs(undefined); - const printTimeout = formatAgyPrintTimeout(timeoutMs); + // F117 KD-22: AGY runs until its turn completes (`--print-timeout` defaults to 0); the member's + // one timeout is its invocation's, so neither AGY nor cli-spawn gets a second one. const agyLogPath = options?.agyLogPathOverride ?? join(tmpdir(), `cat-cafe-agy-${randomUUID()}.log`); const args: string[] = ['--add-dir', workingDirectory]; if (agyProfile?.autoApprove && !readOnly) { @@ -953,9 +947,6 @@ export class GeminiAgentService implements AgentService { for (const dir of imageAccessDirs) { args.push('--add-dir', dir); } - if (printTimeout) { - args.push('--print-timeout', printTimeout); - } if (agyModel) { args.push('--model', agyModel); } @@ -1176,7 +1167,6 @@ export class GeminiAgentService implements AgentService { args, outputMode: 'plainText' as const, cwd: agySpawnCwd, - timeoutMs, ...(childEnv ? { env: childEnv } : {}), ...(options?.signal ? { signal: options.signal } : {}), ...(options?.invocationId ? { invocationId: options.invocationId } : {}), diff --git a/packages/api/src/domains/cats/services/agents/routing/response-terminal.ts b/packages/api/src/domains/cats/services/agents/routing/response-terminal.ts new file mode 100644 index 0000000000..b38efa1055 --- /dev/null +++ b/packages/api/src/domains/cats/services/agents/routing/response-terminal.ts @@ -0,0 +1,46 @@ +import { MEMBER_TIMEOUT_REASON } from '../invocation/member-output-timeout.js'; + +export type ResponseTerminalStatus = 'completed' | 'failed' | 'canceled' | 'interrupted'; + +function nonEmpty(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + +/** + * How a member's response ends, given how its run ended — one rule for every route (F117): + * - output the Queue refused to commit → interrupted, `output_commit_rejected`; + * - stopped by its output timeout → failed, `timeout` (KD-22: a timeout is an ordinary failure); + * - stopped by the user (Stop, stop all) → canceled; stopped for any other reason → interrupted; + * - a provider failure → failed; otherwise → completed. + * A stopped or failed response names why: its error code, else the stop reason. + */ +export function resolveResponseTerminal(input: { + readonly aborted: boolean; + readonly abortReason: unknown; + readonly failed: boolean; + readonly errorCode?: unknown; + readonly outputCommitRejected?: boolean; +}): { readonly status: ResponseTerminalStatus; readonly reason?: string } { + if (input.outputCommitRejected) return { status: 'interrupted', reason: 'output_commit_rejected' }; + const abortReason = nonEmpty(input.abortReason); + if (input.aborted && abortReason === MEMBER_TIMEOUT_REASON) { + return { status: 'failed', reason: MEMBER_TIMEOUT_REASON }; + } + const status: ResponseTerminalStatus = input.aborted + ? abortReason === 'user_cancel' || abortReason === 'cancel_all' + ? 'canceled' + : 'interrupted' + : input.failed + ? 'failed' + : 'completed'; + if (status === 'completed') return { status }; + return { + status, + reason: nonEmpty(input.errorCode) ?? abortReason ?? (status === 'failed' ? 'provider_error' : status), + }; +} + +/** Whether a member's run was stopped by its output timeout rather than by a person. */ +export function stoppedByMemberTimeout(signal: AbortSignal | undefined): boolean { + return signal?.aborted === true && signal.reason === MEMBER_TIMEOUT_REASON; +} diff --git a/packages/api/src/domains/cats/services/agents/routing/route-helpers.ts b/packages/api/src/domains/cats/services/agents/routing/route-helpers.ts index 687a08e8e1..55529a02ba 100644 --- a/packages/api/src/domains/cats/services/agents/routing/route-helpers.ts +++ b/packages/api/src/domains/cats/services/agents/routing/route-helpers.ts @@ -288,6 +288,8 @@ export interface RouteOptions { * canceling one concurrent cat does not abort its siblings (并发取消误伤根因修复). * Absent → fall back to the shared `signal` (route-serial / legacy callers). */ signalForCat?: ((catId: CatId) => AbortSignal | undefined) | undefined; + /** F117 KD-22 (J4): Queue-owned stop for one member whose output timeout fired. Absent → no timeout. */ + stopMember?: import('../invocation/member-output-timeout.js').MemberTimeoutStop | undefined; promptTags?: readonly string[] | undefined; /** Trusted server-owned Cue seeds supplied by connector/workflow ingress. */ memoryCueOpportunitySeeds?: readonly MemoryCueOpportunitySeed[] | undefined; diff --git a/packages/api/src/domains/cats/services/agents/routing/route-parallel.ts b/packages/api/src/domains/cats/services/agents/routing/route-parallel.ts index 6a4cd8251b..5494fa6065 100644 --- a/packages/api/src/domains/cats/services/agents/routing/route-parallel.ts +++ b/packages/api/src/domains/cats/services/agents/routing/route-parallel.ts @@ -89,6 +89,7 @@ import { } from '../invocation/invocation-capacity-snapshot.js'; import { type InvocationParams, invokeSingleCat } from '../invocation/invoke-single-cat.js'; import { buildMcpCallbackInstructions, needsMcpInjection } from '../invocation/McpPromptInjector.js'; +import { type MemberTimeoutEvent, memberTimeoutErrorText } from '../invocation/member-output-timeout.js'; import { getRichBlockBuffer } from '../invocation/RichBlockBuffer.js'; import { recordTurnOutputVerdict, requireTurnOutputAllowed } from '../invocation/response-draft-settlement.js'; import { resolveManagedSessionPolicySnapshot } from '../invocation/session-policy-snapshot.js'; @@ -101,6 +102,7 @@ import { type ContextEvalInput, extractContextEvalSignals } from './context-eval import { buildBriefingMessage } from './format-briefing.js'; import { isDirectOwnerDispositionOrigin } from './human-disposition-invocation-origin.js'; import { persistUserFacingSystemInfoNotices } from './persist-system-info-warnings.js'; +import { resolveResponseTerminal, stoppedByMemberTimeout } from './response-terminal.js'; import { extractRichFromText, isValidRichBlock } from './rich-block-extract.js'; import type { RouteOptions, RouteStrategyDeps } from './route-helpers.js'; import { @@ -462,6 +464,15 @@ export async function* routeParallel( const catToolNames = new Map(); const catCoverageMap = new Map(); const unavailableCats = new Set(); + // F117 KD-22: kept per cat when its output timeout fires, before the cat is stopped. + const catMemberTimeout = new Map(); + const onMemberTimeoutFor = (catId: CatId) => + options.stopMember + ? (timeout: MemberTimeoutEvent): void => { + catMemberTimeout.set(catId as string, timeout); + options.stopMember?.(catId as string, timeout.executionId); + } + : undefined; const streams = await Promise.all( targetCats.map(async (catId) => { @@ -986,6 +997,7 @@ export async function* routeParallel( if (catSignal?.aborted) { return (async function* skipCancelledCat(): AsyncGenerator {})(); } + const onMemberTimeout = onMemberTimeoutFor(catId); const invocationStream = invokeSingleCat(deps.invocationDeps, { ...(options.routeIntent ? { routeIntent: options.routeIntent } : {}), ...(options.routingContextIntent ? { routingContextIntent: options.routingContextIntent } : {}), @@ -1005,6 +1017,7 @@ export async function* routeParallel( ...(targetContentBlocks ? { contentBlocks: targetContentBlocks } : {}), ...(targetUploadDir ? { uploadDir: targetUploadDir } : {}), ...(catSignal ? { signal: catSignal } : {}), + ...(onMemberTimeout ? { onMemberTimeout } : {}), ...(staticIdentity ? { systemPrompt: staticIdentity } : {}), // F194 Phase Z2 (砚砚 catch 2026-05-09)：parallel route 必须传 parentInvocationId， // 与 route-serial.ts:725 对齐。否则 child registry record 缺 parentInvocationId → @@ -1624,28 +1637,28 @@ export async function* routeParallel( } const lifecycleAdmission = catLifecycleResponse.get(msg.catId); const completedSignal = signalForCat?.(msg.catId as CatId) ?? signal; - const abortReason = completedSignal?.reason; - const lifecycleTerminalStatus: 'completed' | 'failed' | 'canceled' | 'interrupted' = !actionOutputCommitAllowed - ? 'interrupted' - : completedSignal?.aborted - ? abortReason === 'user_cancel' || abortReason === 'cancel_all' - ? 'canceled' - : 'interrupted' - : catHadProviderError.has(msg.catId) || (typeof msg.errorCode === 'string' && msg.errorCode.length > 0) - ? 'failed' - : 'completed'; - const lifecycleTerminalReason = - lifecycleTerminalStatus === 'completed' - ? undefined - : !actionOutputCommitAllowed - ? 'output_commit_rejected' - : typeof msg.errorCode === 'string' && msg.errorCode.length > 0 - ? msg.errorCode - : typeof abortReason === 'string' && abortReason.length > 0 - ? abortReason - : lifecycleTerminalStatus === 'failed' - ? 'provider_error' - : lifecycleTerminalStatus; + // F117 KD-22: a member stopped by its output timeout failed, like a provider failure. Its own + // error after the stop was dropped above, so its failure text and diagnostics come from what + // its timer kept before stopping it. + const memberTimeout = stoppedByMemberTimeout(completedSignal) ? catMemberTimeout.get(msg.catId) : undefined; + if (memberTimeout) { + catHadError.add(msg.catId); + catHadProviderError.add(msg.catId); + const priorErrorText = catErrorText.get(msg.catId) ?? ''; + const timeoutText = memberTimeoutErrorText(memberTimeout.diagnostics); + catErrorText.set(msg.catId, `${priorErrorText}${priorErrorText ? '\n' : ''}${timeoutText}`); + catMeta.set(msg.catId, { + ...(catMeta.get(msg.catId) ?? { provider: '', model: '' }), + timeoutDiagnostics: memberTimeout.diagnostics, + }); + } + const { status: lifecycleTerminalStatus, reason: lifecycleTerminalReason } = resolveResponseTerminal({ + aborted: completedSignal?.aborted === true, + abortReason: completedSignal?.reason, + failed: catHadProviderError.has(msg.catId) || (typeof msg.errorCode === 'string' && msg.errorCode.length > 0), + errorCode: msg.errorCode, + outputCommitRejected: !actionOutputCommitAllowed, + }); const lifecycleResponse = lifecycleAdmission && ownInvId ? { @@ -2435,6 +2448,18 @@ export async function* routeParallel( // invocationId → downstream broadcaster falls back to parent → bubble // identity / liveness wrongly attached to parent (instead of own turn). const stampedDone = ownInvId && !msg.invocationId ? { ...msg, invocationId: ownInvId } : msg; + // F117 KD-22: the response is committed; report the timeout before the done, so the Queue + // counts this member failed (not cancelled) and releases its slot. + if (memberTimeout) { + yield { + type: 'error' as const, + catId: msg.catId, + error: memberTimeoutErrorText(memberTimeout.diagnostics), + metadata: { provider: '', model: '', timeoutDiagnostics: memberTimeout.diagnostics }, + ...(ownInvId ? { invocationId: ownInvId } : {}), + timestamp: Date.now(), + }; + } yield projectLiveTurnExecution( { ...stampedDone, diff --git a/packages/api/src/domains/cats/services/agents/routing/route-serial.ts b/packages/api/src/domains/cats/services/agents/routing/route-serial.ts index be1a3862e7..3f079fa1dd 100644 --- a/packages/api/src/domains/cats/services/agents/routing/route-serial.ts +++ b/packages/api/src/domains/cats/services/agents/routing/route-serial.ts @@ -183,6 +183,11 @@ import { } from '../invocation/invocation-capacity-snapshot.js'; import { type InvocationParams, invokeSingleCat } from '../invocation/invoke-single-cat.js'; import { buildMcpCallbackInstructions, needsMcpInjection } from '../invocation/McpPromptInjector.js'; +import { + MEMBER_TIMEOUT_REASON, + type MemberTimeoutEvent, + memberTimeoutErrorText, +} from '../invocation/member-output-timeout.js'; import { getRichBlockBuffer } from '../invocation/RichBlockBuffer.js'; import { recordTurnOutputVerdict, requireTurnOutputAllowed } from '../invocation/response-draft-settlement.js'; import { resolveManagedSessionPolicySnapshot } from '../invocation/session-policy-snapshot.js'; @@ -199,6 +204,7 @@ import { buildBriefingMessage } from './format-briefing.js'; import { resolveEventBackedRoutingExit } from './guards/event-backed-routing-exit.js'; import { isDirectOwnerDispositionOrigin } from './human-disposition-invocation-origin.js'; import { persistUserFacingSystemInfoNotices } from './persist-system-info-warnings.js'; +import { resolveResponseTerminal, stoppedByMemberTimeout } from './response-terminal.js'; import { extractRichFromText, isValidRichBlock } from './rich-block-extract.js'; import type { RouteOptions, RouteStrategyDeps } from './route-helpers.js'; import { @@ -1526,6 +1532,14 @@ export async function* routeSerial( let hadProviderError = false; // Collect error text separately for system-message persistence (F5 reload) let collectedErrorText = ''; + // F117 KD-22: kept when this member's output timeout fires, before it is stopped. + let memberTimeout: MemberTimeoutEvent | undefined; + const onMemberTimeout = options.stopMember + ? (timeout: MemberTimeoutEvent): void => { + memberTimeout = timeout; + options.stopMember?.(catId as string, timeout.executionId); + } + : undefined; // F212 Phase B (云端 codex P2-8 2026-05-27): persist Phase A's structured // cliDiagnostics alongside the error text so cold hydration (F5 reload) can // restore the folded panel — without this, only the legacy red-pill survives. @@ -1867,6 +1881,7 @@ export async function* routeSerial( ...(targetContentBlocks ? { contentBlocks: targetContentBlocks } : {}), ...(targetUploadDir ? { uploadDir: targetUploadDir } : {}), ...(catSignal ? { signal: catSignal } : {}), + ...(onMemberTimeout ? { onMemberTimeout } : {}), ...(staticIdentity ? { systemPrompt: staticIdentity } : {}), ...(options.parentInvocationId ? { parentInvocationId: options.parentInvocationId } : {}), continuityCapsule, @@ -2267,6 +2282,19 @@ export async function* routeSerial( } } + // F117 KD-22: a member stopped by its output timeout failed, like a provider failure. The + // events it produced after the stop were dropped above, so its failure text and diagnostics + // come from what its timer kept before stopping it. + if (stoppedByMemberTimeout(catSignal) && memberTimeout) { + hadError = true; + hadProviderError = true; + collectedErrorText += `${collectedErrorText ? '\n' : ''}${memberTimeoutErrorText(memberTimeout.diagnostics)}`; + persistedMetadata = { + ...(persistedMetadata ?? { provider: '', model: '' }), + timeoutDiagnostics: memberTimeout.diagnostics, + }; + } + // F167 Phase S: this is the single route-side visibility barrier. The // callback records the holder outcome with the durable CAS; nothing below // may enqueue, persist, mutate thread state, synthesize visible output, or @@ -2623,6 +2651,7 @@ export async function* routeSerial( invocationOrigin: resolveInvocationOrigin(options.humanDispositionInvocationOrigin), routeTopology: 'serial', ...(catSignal ? { signal: catSignal } : {}), + ...(onMemberTimeout ? { onMemberTimeout } : {}), ...(staticIdentity ? { systemPrompt: staticIdentity } : {}), ...(options.parentInvocationId ? { parentInvocationId: options.parentInvocationId } : {}), continuityCapsule, @@ -3426,24 +3455,12 @@ export async function* routeSerial( ...(doneMsg?.tracing ? { tracing: doneMsg.tracing } : {}), }, }; - const abortReason = catSignal?.reason; - const lifecycleTerminalStatus: 'completed' | 'failed' | 'canceled' | 'interrupted' = catSignal?.aborted - ? abortReason === 'user_cancel' || abortReason === 'cancel_all' - ? 'canceled' - : 'interrupted' - : hadProviderError - ? 'failed' - : 'completed'; - const lifecycleTerminalReason = - lifecycleTerminalStatus === 'completed' - ? undefined - : typeof doneMsg?.errorCode === 'string' && doneMsg.errorCode.length > 0 - ? doneMsg.errorCode - : typeof abortReason === 'string' && abortReason.length > 0 - ? abortReason - : lifecycleTerminalStatus === 'failed' - ? 'provider_error' - : lifecycleTerminalStatus; + const { status: lifecycleTerminalStatus, reason: lifecycleTerminalReason } = resolveResponseTerminal({ + aborted: catSignal?.aborted === true, + abortReason: catSignal?.reason, + failed: hadProviderError, + errorCode: doneMsg?.errorCode, + }); const lifecycleResponse = lifecycleResponseMessageId && lifecyclePriorFrontierMessageId !== undefined && ownInvocationId ? { @@ -3742,24 +3759,12 @@ export async function* routeSerial( ...(doneMsg?.tracing ? { tracing: doneMsg.tracing } : {}), }, }; - const abortReason = catSignal?.reason; - const lifecycleTerminalStatus: 'completed' | 'failed' | 'canceled' | 'interrupted' = catSignal?.aborted - ? abortReason === 'user_cancel' || abortReason === 'cancel_all' - ? 'canceled' - : 'interrupted' - : hadProviderError - ? 'failed' - : 'completed'; - const lifecycleTerminalReason = - lifecycleTerminalStatus === 'completed' - ? undefined - : typeof doneMsg?.errorCode === 'string' && doneMsg.errorCode.length > 0 - ? doneMsg.errorCode - : typeof abortReason === 'string' && abortReason.length > 0 - ? abortReason - : lifecycleTerminalStatus === 'failed' - ? 'provider_error' - : lifecycleTerminalStatus; + const { status: lifecycleTerminalStatus, reason: lifecycleTerminalReason } = resolveResponseTerminal({ + aborted: catSignal?.aborted === true, + abortReason: catSignal?.reason, + failed: hadProviderError, + errorCode: doneMsg?.errorCode, + }); const lifecycleResponse = lifecycleResponseMessageId && lifecyclePriorFrontierMessageId !== undefined && ownInvocationId ? { @@ -3909,11 +3914,14 @@ export async function* routeSerial( }; let storedErrorTools; if (lifecycleResponseMessageId && ownInvocationId) { + // F117 KD-22: a member stopped by its output timeout failed; any other stop interrupted it. + const timedOut = stoppedByMemberTimeout(catSignal); const terminal = { - status: catSignal?.aborted ? ('interrupted' as const) : ('failed' as const), + status: catSignal?.aborted && !timedOut ? ('interrupted' as const) : ('failed' as const), completedAt: Math.max(Date.now(), invocationStartedAt), - reason: - typeof doneMsg?.errorCode === 'string' && doneMsg.errorCode.length > 0 + reason: timedOut + ? MEMBER_TIMEOUT_REASON + : typeof doneMsg?.errorCode === 'string' && doneMsg.errorCode.length > 0 ? doneMsg.errorCode : 'provider_error', }; @@ -4035,6 +4043,18 @@ export async function* routeSerial( } } + // F117 KD-22: the response is committed; now report the timeout like a provider failure, so + // the Queue counts this member failed (not cancelled) and releases its slot. + if (stoppedByMemberTimeout(catSignal) && memberTimeout) { + yield { + type: 'error' as const, + catId, + error: memberTimeoutErrorText(memberTimeout.diagnostics), + metadata: { provider: '', model: '', timeoutDiagnostics: memberTimeout.diagnostics }, + timestamp: Date.now(), + }; + } + // F222: Frustration auto-issue — detect CLI error + cancel burst signals. // Non-blocking: errors in frustration detection must not interrupt the route pipeline. // F222 P1: Skip for A2A/connector origins — only detect frustration on user-driven routes. diff --git a/packages/api/src/domains/cats/services/freshness/FreshnessInvocationStateStore.ts b/packages/api/src/domains/cats/services/freshness/FreshnessInvocationStateStore.ts index 72588fe6c9..2851cb9f77 100644 --- a/packages/api/src/domains/cats/services/freshness/FreshnessInvocationStateStore.ts +++ b/packages/api/src/domains/cats/services/freshness/FreshnessInvocationStateStore.ts @@ -5,8 +5,8 @@ * Consumed by B1 (notice frequency gating) and B3 (re-invoke trigger). * * Uses Redis HASH per invocation (key: freshness:state:{invocationId}). - * TTL = 30 minutes (invocation timeout, auto-cleanup — not permanent like - * BallCustodyEventLog). This is operational state, not user-visible data. + * TTL = 30 minutes (auto-cleanup — not permanent like BallCustodyEventLog; unrelated to + * CLI_TIMEOUT_MS, which defaults to 0). This is operational state, not user-visible data. * * Design decision (opus-47 insight, KD-7): hot path counters are separated * from the cold path event log (FreshnessAttentionEventLog). B3 re-invoke diff --git a/packages/api/src/domains/routing-context/RoutingDispatchSignalContract.ts b/packages/api/src/domains/routing-context/RoutingDispatchSignalContract.ts index 3b61d1727a..f53cdf81e2 100644 --- a/packages/api/src/domains/routing-context/RoutingDispatchSignalContract.ts +++ b/packages/api/src/domains/routing-context/RoutingDispatchSignalContract.ts @@ -119,6 +119,9 @@ const PROVIDER_FAILURE_CLASSES: Readonly> = { + // F117 KD-22: a member stopped by its output timeout. + timeout: 'provider_timeout', + // Written by the outer invocation timer before KD-22; still read from existing records. invocation_timeout: 'provider_timeout', }; diff --git a/packages/api/src/domains/terminal/tmux-agent-spawner.ts b/packages/api/src/domains/terminal/tmux-agent-spawner.ts index d8ac1a5590..f2adca3064 100644 --- a/packages/api/src/domains/terminal/tmux-agent-spawner.ts +++ b/packages/api/src/domains/terminal/tmux-agent-spawner.ts @@ -17,7 +17,6 @@ import { createModuleLogger } from '../../infrastructure/logger.js'; import { buildCliDiagnostics } from '../../utils/cli-diagnostics.js'; import { withCatCliProcessContext } from '../../utils/cli-process-environment.js'; import { maybeCollectStreamError } from '../../utils/cli-spawn.js'; -import { resolveCliTimeoutMs } from '../../utils/cli-timeout.js'; import type { CliSpawnOptions } from '../../utils/cli-types.js'; // parseNDJSON not used directly — we create readline inline for killability. import type { SpawnCliOverride } from '../cats/services/types.js'; @@ -94,7 +93,10 @@ export async function* spawnCliInTmux( ): AsyncGenerator { options.signal?.throwIfAborted(); const { tmuxGateway } = deps; - const idleTimeoutMs = resolveCliTimeoutMs(options.timeoutMs); + // Only a caller that asks for an idle timeout gets one. F117 KD-22: a dispatched member's timeout + // is its invocation's, driven by CLI_TIMEOUT_MS; the first-event watchdog below still guards a + // member that never comes up. + const idleTimeoutMs = options.timeoutMs ?? 0; const firstEventTimeoutMs = options.firstEventTimeoutMs ?? (options.outputMode === 'plainText' ? idleTimeoutMs : DEFAULT_FIRST_EVENT_TIMEOUT_MS); diff --git a/packages/api/src/utils/ProcessLivenessProbe.ts b/packages/api/src/utils/ProcessLivenessProbe.ts index 638562774a..5d520a362f 100644 --- a/packages/api/src/utils/ProcessLivenessProbe.ts +++ b/packages/api/src/utils/ProcessLivenessProbe.ts @@ -10,6 +10,7 @@ */ import { execFile } from 'node:child_process'; +import type { ProcessActivity } from './process-activity-registry.js'; export type LivenessState = 'active' | 'busy-silent' | 'idle-silent' | 'dead'; @@ -119,6 +120,15 @@ export class ProcessLivenessProbe { } } + /** + * F117 KD-22: whether the process (with its direct children) is still using CPU, whatever it + * printed recently — the member output timeout defers only for this, never for mere stdout. + */ + activity(): ProcessActivity { + if (!this.pidAlive) return 'dead'; + return this.cpuGrowing ? 'busy' : 'idle'; + } + /** Whether bounded extension applies (busy-silent) */ shouldExtendTimeout(): boolean { return this.getState() === 'busy-silent'; diff --git a/packages/api/src/utils/cli-spawn.ts b/packages/api/src/utils/cli-spawn.ts index 3f83977abf..597dd72453 100644 --- a/packages/api/src/utils/cli-spawn.ts +++ b/packages/api/src/utils/cli-spawn.ts @@ -29,7 +29,7 @@ import { CLI_EXECUTION_ID_ENV, CLI_EXECUTION_OWNER_BINDING_ENV } from './cli-pro import { invalidateCliCommand } from './cli-resolve.js'; import { resolveWindowsSpawnPlan } from './cli-spawn-win.js'; import { buildUnixSupervisedSpawnPlan } from './cli-supervised-process.js'; -import { resolveCliTimeoutMs } from './cli-timeout.js'; +import { registerProcessActivity } from './process-activity-registry.js'; export { resolveCliSupervisorNodeArgs } from './cli-supervised-process.js'; @@ -341,8 +341,10 @@ export async function* spawnCli( ): AsyncGenerator { const doSpawn: SpawnFn = deps?.spawnFn ?? defaultSpawn; const livenessWarningDrainIntervalMs = deps?.livenessWarningDrainIntervalMs ?? LIVENESS_WARNING_DRAIN_INTERVAL_MS; - // Default timeout is configurable via CLI_TIMEOUT_MS env var; 0 disables timeout. - const timeoutMs = resolveCliTimeoutMs(options.timeoutMs); + // Only a caller that asks for a response timeout gets one (probes and other one-off commands). + // F117 KD-22: a dispatched member's timeout is its invocation's, driven by CLI_TIMEOUT_MS — never + // a second one here. + const timeoutMs = options.timeoutMs ?? 0; // Log only flag names (--foo) and arg count — never raw values. // Multiple providers pass prompt text via different shapes (positional, @@ -588,13 +590,17 @@ export async function* spawnCli( let localFinalTerminal: 'completed' | 'failed' | null = null; // F118 Phase B: Initialize liveness probe + let unregisterProcessActivity: (() => void) | undefined; if (options.livenessProbe && child.pid !== undefined) { probe = new ProcessLivenessProbe(child.pid, options.livenessProbe); probe.start(); // F152: Register probe for OTel agentLiveness gauge if (options.invocationId) { const catId = options.env?.CAT_CAFE_CAT_ID ?? 'unknown'; - registerLivenessProbe(options.invocationId, catId, () => probe!.getState()); + const liveProbe = probe; + registerLivenessProbe(options.invocationId, catId, () => liveProbe.getState()); + // F117 KD-22: the member output timeout defers while this process is still using CPU. + unregisterProcessActivity = registerProcessActivity(options.invocationId, () => liveProbe.activity()); } } @@ -1081,6 +1087,7 @@ export async function* spawnCli( probe?.stop(); // F152: Unregister probe from OTel gauge if (options.invocationId) unregisterLivenessProbe(options.invocationId); + unregisterProcessActivity?.(); killChild(); // F153 Phase B: End CLI session span with appropriate status diff --git a/packages/api/src/utils/cli-types.ts b/packages/api/src/utils/cli-types.ts index 141c1525ab..115815c3b6 100644 --- a/packages/api/src/utils/cli-types.ts +++ b/packages/api/src/utils/cli-types.ts @@ -26,7 +26,10 @@ export interface CliSpawnOptions { outputMode?: 'ndjson' | 'plainText'; /** Working directory for the process */ cwd?: string; - /** Opt-in timeout in milliseconds before automatic termination (default: 0 = manual cancel only) */ + /** + * Opt-in response timeout in milliseconds for probes and other one-off commands (default: 0 = + * none). A dispatched member never sets it: its one timeout is its invocation's (F117 KD-22). + */ timeoutMs?: number; /** AbortSignal to cancel the process externally */ signal?: AbortSignal; diff --git a/packages/api/src/utils/process-activity-registry.ts b/packages/api/src/utils/process-activity-registry.ts new file mode 100644 index 0000000000..b6f5a4ab58 --- /dev/null +++ b/packages/api/src/utils/process-activity-registry.ts @@ -0,0 +1,20 @@ +/** + * F117 KD-22 (J4): what the process behind an invocation is doing, read by the member output timeout + * to decide whether a silent member is still working. cli-spawn registers each process it probes + * under the invocation it serves; a carrier without a probed process reads as `absent`. + */ +export type ProcessActivity = 'busy' | 'idle' | 'dead'; + +const readers = new Map ProcessActivity>(); + +/** Register the process serving `invocationId`; returns the unregister call for its teardown. */ +export function registerProcessActivity(invocationId: string, read: () => ProcessActivity): () => void { + readers.set(invocationId, read); + return () => { + if (readers.get(invocationId) === read) readers.delete(invocationId); + }; +} + +export function readProcessActivity(invocationId: string): ProcessActivity | 'absent' { + return readers.get(invocationId)?.() ?? 'absent'; +} diff --git a/packages/api/test/1398-phase4-member-output-timeout.test.js b/packages/api/test/1398-phase4-member-output-timeout.test.js new file mode 100644 index 0000000000..fcf2509116 --- /dev/null +++ b/packages/api/test/1398-phase4-member-output-timeout.test.js @@ -0,0 +1,126 @@ +// F117 KD-22 (Phase J, J4): every dispatched member has one timeout, driven by CLI_TIMEOUT_MS — no +// real output for that long. Only output the model produced restarts it; heartbeats, status and +// diagnostics do not. While the member's process is using CPU it gets more time, but never more +// than 2 × CLI_TIMEOUT_MS since its last output. +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +const { MemberOutputTimeout, isMemberOutput } = await import( + '../dist/domains/cats/services/agents/invocation/member-output-timeout.js' +); + +const T = 200; +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); +const text = { type: 'text', catId: 'opus', content: 'hi', timestamp: 0 }; + +function timeout(overrides = {}) { + const fired = []; + const timer = new MemberOutputTimeout({ + timeoutMs: T, + probeProcess: () => 'absent', + onTimeout: (diagnostics) => fired.push(diagnostics), + invocationId: 'inv-child', + ...overrides, + }); + return { timer, fired }; +} + +describe('F117 J4: what counts as a member output', () => { + it('counts text, tool calls and their results, thinking and rich blocks', () => { + assert.equal(isMemberOutput(text), true); + assert.equal(isMemberOutput({ type: 'tool_use', toolName: 'Bash' }), true); + assert.equal(isMemberOutput({ type: 'tool_result', content: 'ok' }), true); + assert.equal( + isMemberOutput({ type: 'system_info', content: JSON.stringify({ type: 'thinking', text: '…' }) }), + true, + ); + assert.equal( + isMemberOutput({ type: 'system_info', content: JSON.stringify({ type: 'rich_block', block: {} }) }), + true, + ); + }); + + it('does not count heartbeats, status, diagnostics or bookkeeping', () => { + for (const message of [ + { type: 'provider_signal', content: 'retrying' }, + { type: 'liveness_signal', content: 'still here' }, + { type: 'status', content: 'working' }, + { type: 'system_info', content: JSON.stringify({ type: 'liveness_warning' }) }, + { type: 'system_info', content: JSON.stringify({ type: 'timeout_diagnostics', silenceDurationMs: 1 }) }, + { type: 'system_info', content: JSON.stringify({ type: 'invocation_created' }) }, + { type: 'system_info', content: 'not json' }, + { type: 'session_init', sessionId: 's' }, + { type: 'agent_loop' }, + ]) { + assert.equal(isMemberOutput(message), false, `${message.type} ${message.content ?? ''}`); + } + }); +}); + +describe('F117 J4: the member output timeout', () => { + it('fires once when the member produces no output for CLI_TIMEOUT_MS', async () => { + const { timer, fired } = timeout(); + await sleep(T * 3); + timer.close(); + assert.equal(fired.length, 1); + assert.equal(fired[0].processAlive, true, 'a carrier without a probed process is still running'); + assert.equal(fired[0].invocationId, 'inv-child'); + assert.ok(fired[0].silenceDurationMs >= T); + }); + + it('restarts on output but not on heartbeats or status', async () => { + const { timer, fired } = timeout(); + await sleep(T * 0.6); + timer.observe(text); + await sleep(T * 0.6); + timer.observe({ type: 'liveness_signal', content: 'ping' }); + timer.observe({ type: 'status', content: 'still working' }); + assert.equal(fired.length, 0, 'output restarted the clock'); + await sleep(T * 0.8); + timer.close(); + assert.equal(fired.length, 1, 'heartbeats and status did not'); + assert.equal(fired[0].lastEventType, 'text'); + assert.ok(fired[0].lastEventAt >= fired[0].firstEventAt); + }); + + it('gives a member whose process is using CPU more time, up to 2 × CLI_TIMEOUT_MS since its last output', async () => { + const startedAt = Date.now(); + let firedAt; + const { timer, fired } = timeout({ + probeProcess: () => 'busy', + onTimeout: (diagnostics) => { + firedAt = Date.now(); + fired.push(diagnostics); + }, + }); + await sleep(T * 1.5); + assert.equal(fired.length, 0, 'still busy at CLI_TIMEOUT_MS: deferred'); + await sleep(T * 1.5); + timer.close(); + assert.equal(fired.length, 1, 'the deferral is capped'); + assert.ok(firedAt - startedAt >= 2 * T - 5 && firedAt - startedAt < 3 * T, `fired after ${firedAt - startedAt}ms`); + assert.ok(fired[0].silenceDurationMs >= 2 * T - 5); + }); + + it('does not defer a member whose process is idle, and reports a dead one as not alive', async () => { + const idle = timeout({ probeProcess: () => 'idle' }); + const dead = timeout({ probeProcess: () => 'dead' }); + await sleep(T * 1.5); + idle.timer.close(); + dead.timer.close(); + assert.equal(idle.fired.length, 1); + assert.equal(idle.fired[0].processAlive, true); + assert.equal(dead.fired.length, 1); + assert.equal(dead.fired[0].processAlive, false); + }); + + it('never fires when CLI_TIMEOUT_MS is 0, or after it was closed', async () => { + const off = timeout({ timeoutMs: 0 }); + const closed = timeout(); + closed.timer.close(); + await sleep(T * 2); + off.timer.close(); + assert.equal(off.fired.length, 0); + assert.equal(closed.fired.length, 0); + }); +}); diff --git a/packages/api/test/1398-phase4-member-timeout-stop.test.js b/packages/api/test/1398-phase4-member-timeout-stop.test.js new file mode 100644 index 0000000000..ab50cd25ef --- /dev/null +++ b/packages/api/test/1398-phase4-member-timeout-stop.test.js @@ -0,0 +1,261 @@ +// F117 KD-22 (Phase J, J4): when a member's output timeout fires it is stopped the way Stop stops it, +// with reason `timeout`, and only that member: its siblings in the same turn run on. The stop goes +// through the Queue's slot, keyed by the parent execution id, so a timer left over from an earlier +// execution never stops the one that took the slot after it. A timed-out member failed — its +// response, its turn execution and the Queue's aggregate say so; nobody cancelled it. +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; + +process.env.CLI_TIMEOUT_MS = '1000'; + +const { routeParallel } = await import('../dist/domains/cats/services/agents/routing/route-parallel.js'); +const { routeSerial } = await import('../dist/domains/cats/services/agents/routing/route-serial.js'); +const { InvocationTracker } = await import('../dist/domains/cats/services/agents/invocation/InvocationTracker.js'); +const { createMemberTimeoutStop, MEMBER_TIMEOUT_REASON } = await import( + '../dist/domains/cats/services/agents/invocation/member-output-timeout.js' +); +const { resolveResponseTerminal } = await import('../dist/domains/cats/services/agents/routing/response-terminal.js'); +const { classifyRoutingDispatchFailure } = await import( + '../dist/domains/routing-context/RoutingDispatchSignalContract.js' +); +const { isCliStartupTimeoutError } = await import('../dist/domains/cats/services/agents/invocation/invoke-helpers.js'); + +const quietLog = { info() {} }; + +/** A member that sends a heartbeat and then never produces output until it is stopped. */ +function silentService(catId) { + return { + async *invoke(_prompt, options) { + yield { type: 'status', catId, content: 'still thinking', timestamp: Date.now() }; + await new Promise((resolve) => { + if (options?.signal?.aborted) return resolve(); + options?.signal?.addEventListener('abort', resolve, { once: true }); + }); + }, + }; +} + +/** A member that answers after `delayMs`, longer than CLI_TIMEOUT_MS would allow in silence. */ +function answeringService(catId, delayMs) { + return { + async *invoke() { + for (let at = 0; at < delayMs; at += 40) { + await new Promise((resolve) => setTimeout(resolve, 40)); + yield { type: 'text', catId, content: '.', timestamp: Date.now() }; + } + yield { type: 'done', catId, timestamp: Date.now() }; + }, + }; +} + +function turnExecutionStore() { + const records = new Map(); + return { + records, + async createRunning(input) { + const record = { ...input, status: 'running' }; + records.set(input.invocationId, record); + return { outcome: 'created', record }; + }, + async transitionTerminal(invocationId, terminal) { + const record = { ...records.get(invocationId), ...terminal }; + records.set(invocationId, record); + return { outcome: 'transitioned', record }; + }, + async get(invocationId) { + return records.get(invocationId) ?? null; + }, + async bindCoveredMessageIds() { + return { outcome: 'bound' }; + }, + }; +} + +function routeDeps(services, turnStore) { + let invocationSeq = 0; + let messageSeq = 0; + const storedById = new Map(); + return { + services, + toolEventLog: { append: async () => {}, updateSummary: async () => {} }, + invocationDeps: { + registry: { + create: () => ({ invocationId: `child-${++invocationSeq}`, callbackToken: `tok-${invocationSeq}` }), + verify: () => ({ ok: false, reason: 'unknown_invocation' }), + }, + sessionManager: { + get: async () => null, + getOrCreate: async () => ({}), + resolveWorkingDirectory: () => '/tmp/test', + }, + threadStore: { + get: async () => null, + getParticipantsWithActivity: async () => [], + updateParticipantActivity: async () => {}, + consumeMentionRoutingFeedback: async () => null, + }, + turnExecutionStore: turnStore, + apiUrl: 'http://127.0.0.1:3004', + }, + messageStore: { + append: async (msg) => { + const stored = { id: `msg-${++messageSeq}`, ...msg, threadId: msg.threadId ?? 'default' }; + storedById.set(stored.id, stored); + return stored; + }, + getById: async (id) => storedById.get(id) ?? null, + getRecent: () => [], + getMentionsFor: () => [], + getRecentMentionsFor: () => [], + getBefore: () => [], + getByThread: () => [], + getByThreadAfter: () => [], + getByThreadBefore: () => [], + }, + draftStore: { delete: () => Promise.resolve(), touch: () => Promise.resolve(), upsert: () => Promise.resolve() }, + socketManager: { broadcastToRoom: () => {} }, + }; +} + +/** A Queue execution: one slot per target under the parent execution id, routed with the stop hook. */ +async function dispatch(route, targets, services) { + const tracker = new InvocationTracker(); + const executionId = 'parent-exec-1'; + tracker.startAll('t1', targets, 'user1', executionId); + const turnStore = turnExecutionStore(); + const events = []; + for await (const event of route(routeDeps(services, turnStore), targets, 'msg', 'user1', 't1', { + signalForCat: (catId) => tracker.getController('t1', catId)?.signal, + parentInvocationId: executionId, + stopMember: createMemberTimeoutStop({ + invocationTracker: tracker, + threadId: 't1', + ownerUserId: 'user1', + log: quietLog, + }), + })) { + events.push(event); + } + const terminalOf = (catId) => [...turnStore.records.values()].find((record) => record.catId === catId); + return { tracker, events, terminalOf }; +} + +function assertTimedOutMember(result, catId) { + assert.equal(result.tracker.isTimedOut('t1', catId), true, `${catId} was stopped with reason timeout`); + const failure = result.events.find((event) => event.type === 'error' && event.catId === catId); + assert.ok(failure, `${catId}'s route reports the timeout`); + assert.match(failure.error, /响应超时/); + assert.ok(failure.metadata.timeoutDiagnostics.silenceDurationMs >= 1000, 'diagnostics kept from before the stop'); + const terminal = result.terminalOf(catId); + assert.equal(terminal.status, 'failed'); + assert.equal(terminal.terminalReason, MEMBER_TIMEOUT_REASON); + assert.equal(terminal.parentInvocationId, 'parent-exec-1', 'the timer keyed the stop by the parent execution'); + assert.notEqual(terminal.invocationId, terminal.parentInvocationId, 'while the member ran as its own child turn'); +} + +function assertFinishedMember(result, catId) { + assert.equal(result.tracker.getSlotState('t1', catId), 'active', `${catId} was not stopped`); + assert.equal(result.tracker.getController('t1', catId).signal.aborted, false); + assert.equal( + result.events.some((event) => event.type === 'error' && event.catId === catId), + false, + ); + assert.equal(result.terminalOf(catId).status, 'succeeded'); +} + +describe('F117 J4: a timed-out member is stopped like Stop, alone', () => { + it('parallel: stops the silent member and lets its sibling finish', async () => { + const result = await dispatch(routeParallel, ['opus', 'codex'], { + opus: silentService('opus'), + codex: answeringService('codex', 1600), + }); + assertTimedOutMember(result, 'opus'); + assertFinishedMember(result, 'codex'); + const opusEvents = result.events.filter((event) => event.catId === 'opus').map((event) => event.type); + assert.ok(opusEvents.indexOf('error') < opusEvents.lastIndexOf('done'), 'the failure is reported before its done'); + }); + + it('serial: stops the silent member, then runs the next one', async () => { + const result = await dispatch(routeSerial, ['opus', 'codex'], { + opus: silentService('opus'), + codex: answeringService('codex', 600), + }); + assertTimedOutMember(result, 'opus'); + assertFinishedMember(result, 'codex'); + }); +}); + +describe('F117 J4: the Queue stop for a member timeout', () => { + it('stops only the execution that armed the timer, never the one that took the slot after it', () => { + const tracker = new InvocationTracker(); + const earlier = tracker.start('t1', 'opus', 'user1', ['opus'], 'exec-earlier'); + tracker.complete('t1', 'opus', earlier); + const current = tracker.start('t1', 'opus', 'user1', ['opus'], 'exec-current'); + const stop = createMemberTimeoutStop({ + invocationTracker: tracker, + threadId: 't1', + ownerUserId: 'user1', + log: quietLog, + }); + + assert.equal(stop('opus', 'exec-earlier'), false); + assert.equal(current.signal.aborted, false, 'a leftover timer leaves the new execution alone'); + + assert.equal(stop('opus', 'exec-current'), true); + assert.equal(current.signal.reason, MEMBER_TIMEOUT_REASON); + assert.equal(tracker.isTimedOut('t1', 'opus'), true); + }); + + it('counts a timed-out member as failed, not as cancelled by the user', () => { + const timedOut = new InvocationTracker(); + timedOut.start('t1', 'opus', 'user1', ['opus'], 'exec-1'); + createMemberTimeoutStop({ invocationTracker: timedOut, threadId: 't1', ownerUserId: 'user1', log: quietLog })( + 'opus', + 'exec-1', + ); + assert.equal(timedOut.resolveFinalStatus('t1', ['opus'], { aborted: false }), 'succeeded', 'outcomes decide'); + + const stopped = new InvocationTracker(); + stopped.start('t1', 'opus', 'user1', ['opus'], 'exec-1'); + stopped.cancel('t1', 'opus', 'user1', 'user_cancel'); + assert.equal(stopped.isTimedOut('t1', 'opus'), false); + assert.equal(stopped.resolveFinalStatus('t1', ['opus'], { aborted: false }), 'canceled_by_user'); + }); +}); + +describe('F117 J4: how a stopped member ends', () => { + it('a timeout is a failure with reason timeout; a Stop cancels; other stops interrupt', () => { + assert.deepEqual(resolveResponseTerminal({ aborted: true, abortReason: 'timeout', failed: false }), { + status: 'failed', + reason: 'timeout', + }); + assert.deepEqual(resolveResponseTerminal({ aborted: true, abortReason: 'user_cancel', failed: false }), { + status: 'canceled', + reason: 'user_cancel', + }); + assert.deepEqual(resolveResponseTerminal({ aborted: true, abortReason: 'preempted', failed: true }), { + status: 'interrupted', + reason: 'preempted', + }); + assert.deepEqual(resolveResponseTerminal({ aborted: false, abortReason: undefined, failed: true }), { + status: 'failed', + reason: 'provider_error', + }); + assert.deepEqual( + resolveResponseTerminal({ aborted: false, abortReason: undefined, failed: false, outputCommitRejected: true }), + { status: 'interrupted', reason: 'output_commit_rejected' }, + ); + assert.deepEqual(resolveResponseTerminal({ aborted: false, abortReason: undefined, failed: false }), { + status: 'completed', + }); + }); + + it('routing counts a member timeout as a provider timeout', () => { + assert.equal(classifyRoutingDispatchFailure({ terminalReason: 'timeout' }), 'provider_timeout'); + }); + + it('keeps the session-resume retry for a CLI that never came up, not for silence after it did', () => { + assert.equal(isCliStartupTimeoutError('布偶猫 CLI 响应超时 (30s, 未收到首帧)'), true); + assert.equal(isCliStartupTimeoutError('布偶猫 CLI 响应超时 (1800s)'), false); + }); +}); diff --git a/packages/api/test/f254-provider-native-freshness.test.js b/packages/api/test/f254-provider-native-freshness.test.js index ae3a9bfba0..ac5a334a26 100644 --- a/packages/api/test/f254-provider-native-freshness.test.js +++ b/packages/api/test/f254-provider-native-freshness.test.js @@ -1457,7 +1457,7 @@ describe('F254 D2 provider-native freshness truth', () => { assert.equal(messages.at(-1).type, 'done'); }); - it('honors the global CLI_TIMEOUT_MS opt-in through protocol interruption', async () => { + it('arms no app-server idle interrupt from CLI_TIMEOUT_MS; only a stop interrupts the turn (F117 KD-22)', async () => { const previousTimeout = process.env.CLI_TIMEOUT_MS; process.env.CLI_TIMEOUT_MS = '20'; const wire = new FakeAppServerWire(); @@ -1502,13 +1502,13 @@ describe('F254 D2 provider-native freshness truth', () => { .filter(Boolean); assert.equal( lifecycle.some((event) => event.interruptReason === 'timeout'), - true, - 'the process-level /config timeout must reach the app-server client', + false, + "the member's one timeout is its invocation's, not an app-server idle interrupt", ); assert.equal( lifecycle.some((event) => event.interruptReason === 'user_cancel'), - false, - 'the failsafe should not be the source of interruption', + true, + 'the turn ran until its signal stopped it', ); }); diff --git a/packages/api/test/gemini-agent-service.test.js b/packages/api/test/gemini-agent-service.test.js index a520910530..aefd2bce2a 100644 --- a/packages/api/test/gemini-agent-service.test.js +++ b/packages/api/test/gemini-agent-service.test.js @@ -647,7 +647,7 @@ describe('GeminiAgentService (antigravity-cli adapter)', () => { assert.equal(args[modelIdx + 1], 'Gemini 3.5 Flash (High)'); }); - test('passes an explicitly configured positive CLI timeout to agy print mode', async (t) => { + test('gives agy no timeout of its own even when CLI_TIMEOUT_MS is set (F117 KD-22)', async (t) => { const savedTimeout = process.env.CLI_TIMEOUT_MS; process.env.CLI_TIMEOUT_MS = '1500'; t.after(() => { @@ -669,9 +669,11 @@ describe('GeminiAgentService (antigravity-cli adapter)', () => { await promise; const args = spawnFn.mock.calls[0].arguments[1]; - const timeoutIdx = args.indexOf('--print-timeout'); - assert.ok(timeoutIdx >= 0, 'positive CLI_TIMEOUT_MS must preserve the opt-in AGY timeout'); - assert.equal(args[timeoutIdx + 1], '2s'); + assert.equal( + args.includes('--print-timeout'), + false, + "the member's one timeout is its invocation's; agy runs until its turn completes", + ); }); test('normalizes legacy Gemini model ids before agy --model spawn', async () => { diff --git a/packages/api/test/invocation-timeout-guard.test.js b/packages/api/test/invocation-timeout-guard.test.js index fbd19a0947..0ad0f8001c 100644 --- a/packages/api/test/invocation-timeout-guard.test.js +++ b/packages/api/test/invocation-timeout-guard.test.js @@ -1,8 +1,8 @@ /** - * F089: Invocation-level hard timeout guard - * - * Regression tests for the invocation timeout that prevents "正在回复中" from - * hanging forever when the service generator neither yields done nor throws. + * F089 → F117 KD-22 (J4): a member that stops producing output must not leave "正在回复中" hanging + * forever when its service generator neither yields done nor throws. F089's outer 2× timer is + * gone; the member output timeout (CLI_TIMEOUT_MS of silence) fires once and the Queue stops the + * member through its signal with reason `timeout`, the way Stop does. */ import './helpers/setup-cat-registry.js'; @@ -31,14 +31,13 @@ let invokeSingleCat; let savedAuditLogDir; let savedCliTimeoutMs; -describe('invocation-level hard timeout (F089)', () => { +describe('member output timeout at the invocation (F089 → F117 KD-22)', () => { before(async () => { savedAuditLogDir = process.env.AUDIT_LOG_DIR; savedCliTimeoutMs = process.env.CLI_TIMEOUT_MS; const tempDir = await mkdtemp(join(tmpdir(), 'cat-inv-timeout-')); process.env.AUDIT_LOG_DIR = tempDir; - // Override CLI_TIMEOUT_MS to make invocation timeout short for testing. - // Invocation timeout = CLI_TIMEOUT_MS * 2 = 400ms + // A short member output timeout for testing: 200ms without output. process.env.CLI_TIMEOUT_MS = '200'; const mod = await import('../dist/domains/cats/services/agents/invocation/invoke-single-cat.js'); invokeSingleCat = mod.invokeSingleCat; @@ -70,16 +69,32 @@ describe('invocation-level hard timeout (F089)', () => { }; } - it('service that never yields done converges via invocation timeout', async () => { - // A service that emits one content event then hangs forever. - // Without the invocation timeout, this would block forever. + /** + * The Queue side of a member timeout: stop the member through its own signal with reason + * `timeout`, the way QueueProcessor's stop hook cancels the slot (F117 KD-22). + */ + function stoppedOnTimeout() { + const controller = new AbortController(); + const fired = []; + return { + fired, + signal: controller.signal, + onMemberTimeout: (timeout) => { + fired.push(timeout); + controller.abort('timeout'); + }, + }; + } + + it('a member that never produces output again is stopped by its output timeout', async () => { + // One output, then a stuck provider: without the member timeout this would block forever. const hangingService = { async *invoke() { yield { type: 'text', catId: 'codex', content: 'thinking...', timestamp: Date.now() }; - // Hang indefinitely — simulates a stuck CLI/provider await new Promise(() => {}); }, }; + const stop = stoppedOnTimeout(); const start = Date.now(); const msgs = await withKeepAlive( @@ -90,28 +105,38 @@ describe('invocation-level hard timeout (F089)', () => { prompt: 'test', userId: 'user1', threadId: 'thread-hang', + parentInvocationId: 'parent-exec-hang', isLastCat: true, + signal: stop.signal, + onMemberTimeout: stop.onMemberTimeout, }), ), ); const elapsed = Date.now() - start; - // Should have converged within a reasonable time (invocation timeout = 400ms) assert.ok(elapsed < 5000, `should converge quickly, took ${elapsed}ms`); - - // Must always end with error + done - const hasError = msgs.some((m) => m.type === 'error'); - const hasDone = msgs.some((m) => m.type === 'done'); - assert.ok(hasError, 'timeout should produce an error event'); - assert.ok(hasDone, 'timeout should always produce a done event'); + assert.equal(stop.fired.length, 1, 'the timeout fires once'); + assert.equal(stop.fired[0].executionId, 'parent-exec-hang', 'keyed by the Queue execution, not the child turn'); + assert.ok(stop.fired[0].diagnostics.silenceDurationMs >= 200); + assert.equal(stop.fired[0].diagnostics.lastEventType, 'text'); + // The stop winds the run down like any Stop: always error + done. + assert.ok( + msgs.some((m) => m.type === 'error'), + 'the stop should produce an error event', + ); + assert.ok( + msgs.some((m) => m.type === 'done'), + 'the stop should always produce a done event', + ); }); - it('timeout yields done with isFinal=true for last cat', async () => { + it('a stopped last member still yields done with isFinal=true', async () => { const hangingService = { async *invoke() { await new Promise(() => {}); }, }; + const stop = stoppedOnTimeout(); const msgs = await withKeepAlive( collect( @@ -122,6 +147,8 @@ describe('invocation-level hard timeout (F089)', () => { userId: 'user1', threadId: 'thread-final', isLastCat: true, + signal: stop.signal, + onMemberTimeout: stop.onMemberTimeout, }), ), ); @@ -131,52 +158,47 @@ describe('invocation-level hard timeout (F089)', () => { assert.equal(doneMsg.isFinal, true, 'done should have isFinal=true for last cat'); }); - it('CLI_TIMEOUT_MS=0 does not produce instant invocation timeout', async () => { - // When CLI_TIMEOUT_MS=0 (meaning "disable CLI timeout"), invocation timeout - // must NOT become 0ms. It should fall back to a sane maximum. + it('CLI_TIMEOUT_MS=0 never times a member out', async () => { const savedTimeout = process.env.CLI_TIMEOUT_MS; process.env.CLI_TIMEOUT_MS = '0'; try { - // Re-import to pick up new env value - const freshMod = await import( - `../dist/domains/cats/services/agents/invocation/invoke-single-cat.js?t=${Date.now()}` - ); - const freshInvoke = freshMod.invokeSingleCat; - - // A service that yields one event then completes quickly (50ms). - // If invocation timeout is 0ms, this would be killed before it finishes. - const quickService = { + // Output, then 300ms of silence (> the 200ms the other cases use), then done. + const quietService = { async *invoke() { yield { type: 'text', catId: 'codex', content: 'hello', timestamp: Date.now() }; - await new Promise((r) => setTimeout(r, 50)); + await new Promise((r) => setTimeout(r, 300)); yield { type: 'done', catId: 'codex', isFinal: true, timestamp: Date.now() }; }, }; + const stop = stoppedOnTimeout(); - const msgs = await collect( - freshInvoke(makeDeps(), { - catId: 'codex', - service: quickService, - prompt: 'test', - userId: 'user1', - threadId: 'thread-zero-timeout', - isLastCat: true, - }), + const msgs = await withKeepAlive( + collect( + invokeSingleCat(makeDeps(), { + catId: 'codex', + service: quietService, + prompt: 'test', + userId: 'user1', + threadId: 'thread-zero-timeout', + isLastCat: true, + signal: stop.signal, + onMemberTimeout: stop.onMemberTimeout, + }), + ), ); - // The service should complete normally — no error from invocation timeout - const hasInvocationError = msgs.some((m) => m.type === 'error' && m.error?.includes?.('invocation_timeout')); - assert.ok(!hasInvocationError, 'CLI_TIMEOUT_MS=0 should not produce invocation_timeout error'); - - // Should have the text event from the service - const hasText = msgs.some((m) => m.type === 'text' && m.content === 'hello'); - assert.ok(hasText, 'should receive events from service'); + assert.equal(stop.fired.length, 0, 'CLI_TIMEOUT_MS=0 arms no timeout'); + assert.ok(!msgs.some((m) => m.type === 'error'), 'the member finishes normally'); + assert.ok( + msgs.some((m) => m.type === 'text' && m.content === 'hello'), + 'should receive events from service', + ); } finally { process.env.CLI_TIMEOUT_MS = savedTimeout; } }); - it('user cancel (AbortSignal) still works alongside invocation timeout', async () => { + it('user cancel (AbortSignal) stops a member before its timeout', async () => { const ac = new AbortController(); const hangingService = { async *invoke() { @@ -185,7 +207,7 @@ describe('invocation-level hard timeout (F089)', () => { }, }; - // Cancel after 100ms — should be faster than invocation timeout (400ms) + // Cancel after 100ms — before the 200ms member timeout setTimeout(() => ac.abort(), 100); const msgs = await collect( @@ -200,7 +222,7 @@ describe('invocation-level hard timeout (F089)', () => { }), ); - // Must always end with done — regardless of whether user cancel or invocation timeout triggered + // Must always end with done assert.ok( msgs.some((m) => m.type === 'done'), 'cancel should produce done event', @@ -256,18 +278,18 @@ describe('invocation-level hard timeout (F089)', () => { assert.equal(returnCalls, 1, 'abortableNext rejection must not abandon the provider iterator'); }); - it('active invocations with steady progress should not hit invocation_timeout', async () => { + it('steady output keeps a member alive; heartbeats alone do not', async () => { const progressiveService = { async *invoke() { - yield { type: 'text', catId: 'codex', content: 'tick-1', timestamp: Date.now() }; - await new Promise((r) => setTimeout(r, 150)); - yield { type: 'text', catId: 'codex', content: 'tick-2', timestamp: Date.now() }; - await new Promise((r) => setTimeout(r, 150)); - yield { type: 'text', catId: 'codex', content: 'tick-3', timestamp: Date.now() }; - await new Promise((r) => setTimeout(r, 150)); + for (const tick of ['tick-1', 'tick-2', 'tick-3']) { + yield { type: 'text', catId: 'codex', content: tick, timestamp: Date.now() }; + yield { type: 'status', catId: 'codex', content: 'working', timestamp: Date.now() }; + await new Promise((r) => setTimeout(r, 150)); + } yield { type: 'done', catId: 'codex', isFinal: true, timestamp: Date.now() }; }, }; + const stop = stoppedOnTimeout(); const msgs = await withKeepAlive( collect( @@ -278,13 +300,14 @@ describe('invocation-level hard timeout (F089)', () => { userId: 'user1', threadId: 'thread-progress', isLastCat: true, + signal: stop.signal, + onMemberTimeout: stop.onMemberTimeout, }), ), 2_000, ); - const hasInvocationError = msgs.some((m) => m.type === 'error' && m.error?.includes?.('invocation_timeout')); - assert.ok(!hasInvocationError, 'steady progress should keep invocation alive'); + assert.equal(stop.fired.length, 0, 'output every 150ms stays under the 200ms timeout'); assert.equal(msgs.filter((m) => m.type === 'text').length, 3, 'should receive all progress events before done'); }); });