From 16514586c57d7e57ff4d6ecafe6936ff1f5029c0 Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Mon, 10 Aug 2026 21:46:50 +0800 Subject: [PATCH 1/3] fix(cli): wait for detached run terminal state --- docs/architecture/local-control-plane/spec.md | 10 +- docs/guides/cli.md | 6 +- src/cli/format.ts | 3 +- .../agent/deepchat/runtime/sessionUpdates.ts | 63 ++- src/main/app/composition.ts | 8 + src/main/cli/index.ts | 2 +- src/main/cli/routes.ts | 2 +- src/main/cli/runService.ts | 38 +- src/main/cli/server.ts | 8 +- src/main/cli/surface.ts | 8 +- src/shared/contracts/localControl.ts | 2 +- src/shared/contracts/routes/ocr.routes.ts | 5 +- src/shared/contracts/routes/runs.routes.ts | 3 + test/main/cli/client.test.ts | 26 +- test/main/cli/inputCapabilityServices.test.ts | 2 +- test/main/cli/runService.test.ts | 384 ++++++++++++++++-- test/main/cli/server.test.ts | 9 +- test/main/cli/surface.test.ts | 10 +- 18 files changed, 508 insertions(+), 81 deletions(-) diff --git a/docs/architecture/local-control-plane/spec.md b/docs/architecture/local-control-plane/spec.md index 66e3877be..62f39eddc 100644 --- a/docs/architecture/local-control-plane/spec.md +++ b/docs/architecture/local-control-plane/spec.md @@ -148,7 +148,7 @@ Main atomically replaces a descriptor with this public shape: ```ts type LocalControlDescriptorV1 = { protocolVersion: 1 - surfaceVersion: 1 + surfaceVersion: 2 appVersion: string endpoint: { kind: 'unix'; path: string } | { kind: 'pipe'; name: string } pid: number @@ -215,7 +215,7 @@ expose those temporary paths. ## Contract Ownership and Surface -`CLI_SURFACE_V1` is a readonly registry whose entries reference the same `RouteContract` objects used +`CLI_SURFACE_V2` is a readonly registry whose entries reference the same `RouteContract` objects used by renderer IPC. A surface entry adds only transport and policy metadata: ```ts @@ -540,7 +540,11 @@ existing lifecycle, then starts the initial turn. It returns a durable run/sessi streaming. Disconnect does not destroy a detached run; status/messages can be recovered from session state and event cursors. If initial-turn startup fails, the response and run event retain that durable identity but expose only a stable failure message; upstream error text is excluded from public output -and logs. `runs.cancel` is idempotent and ownership checked. +and logs. `runs.get` projects `running`, `awaiting_interaction`, or `terminal` separately from the +Session status so a valid permission or question pause remains observable. `run watch` treats +provider-round `chat.stream.completed` and `chat.stream.failed` events as progress only; it exits +after the root Session reaches `idle` or `error`, never when a descendant Session or one provider +round finishes. `runs.cancel` is idempotent and ownership checked. `events.subscribe` is human-only in V1. An Agent can own only its currently executing conversation, so waiting on that run from its bash tool would deadlock the run on itself. Agent callers may use the diff --git a/docs/guides/cli.md b/docs/guides/cli.md index 53d7fc6f1..895e4a3df 100644 --- a/docs/guides/cli.md +++ b/docs/guides/cli.md @@ -181,7 +181,9 @@ deepchat run cancel --run --json ``` `agent run` 先创建 durable detached Session，再启动首轮。CLI 断开不会删除 run；可以通过 -`run get` 恢复消息，human caller 可通过 cursor 续接 `run watch`。Agent caller 自身不能递归执行 +`run get` 恢复消息和 `running | awaiting_interaction | terminal` phase，human caller 可通过 cursor +续接 `run watch`。一次 provider stream 完成或失败不会提前结束 watch；只有 root Session 进入 +`idle`/`error` 才是整个 detached run 的终态。Agent caller 自身不能递归执行 `agent run`，也不能等待当前正在执行的自身 run；Agent 仅可使用非阻塞的 `run get` 与幂等的 `run cancel`。 @@ -218,7 +220,7 @@ Agent 调用额外经过以下控制： 1. shell command permission； 2. main 签发的短期 scoped token； -3. deny-by-default `CLI_SURFACE_V1` caller/scope policy； +3. deny-by-default `CLI_SURFACE_V2` caller/scope policy； 4. effect policy 与 renderer-only approval； 5. ownership、rate、call/byte quota 与脱敏审计。 diff --git a/src/cli/format.ts b/src/cli/format.ts index c60ffe793..5ec48257f 100644 --- a/src/cli/format.ts +++ b/src/cli/format.ts @@ -93,8 +93,9 @@ export function formatHumanResult( } case 'runs.get': { const result = contract.output.parse(value) + const interaction = result.phase === 'awaiting_interaction' ? ', awaiting interaction' : '' return [ - `Run ${result.runId} (${result.status})`, + `Run ${result.runId} (${result.status}${interaction})`, `Agent: ${result.agentId}`, `Model: ${result.providerId}/${result.modelId}`, ...result.messages.flatMap((message) => [ diff --git a/src/main/agent/deepchat/runtime/sessionUpdates.ts b/src/main/agent/deepchat/runtime/sessionUpdates.ts index c50fed6fa..59496f91d 100644 --- a/src/main/agent/deepchat/runtime/sessionUpdates.ts +++ b/src/main/agent/deepchat/runtime/sessionUpdates.ts @@ -73,26 +73,69 @@ export const buildAssistantDeliverySegments = ( blocks: AssistantMessageBlock[] ): AssistantDeliverySegment[] => buildDeliverySegments(messageId, blocks) +function isWaitingInteractionBlock(block: AssistantMessageBlock): boolean { + return ( + block.type === 'action' && + block.status === 'pending' && + block.extra?.needsUserAction !== false && + Boolean(block.tool_call?.id) && + (block.action_type === 'tool_call_permission' || block.action_type === 'question_request') + ) +} + +function hasSubagentWaitingInteraction(block: AssistantMessageBlock): boolean { + if (block.type !== 'tool_call' || block.tool_call?.name !== 'subagent_orchestrator') return false + const rawProgress = block.extra?.subagentProgress + if (typeof rawProgress !== 'string' || !rawProgress.trim()) return false + try { + const progress = JSON.parse(rawProgress) as { tasks?: unknown } + if (!Array.isArray(progress?.tasks)) return false + return progress.tasks.some((task) => { + if (!task || typeof task !== 'object' || Array.isArray(task)) return false + const candidate = task as { sessionId?: unknown; waitingInteraction?: unknown } + if (typeof candidate.sessionId !== 'string' || !candidate.sessionId) return false + const waiting = candidate.waitingInteraction + if (!waiting || typeof waiting !== 'object' || Array.isArray(waiting)) return false + const interaction = waiting as { + type?: unknown + messageId?: unknown + toolCallId?: unknown + actionBlock?: unknown + } + return ( + (interaction.type === 'permission' || interaction.type === 'question') && + typeof interaction.messageId === 'string' && + Boolean(interaction.messageId) && + typeof interaction.toolCallId === 'string' && + Boolean(interaction.toolCallId) && + Boolean(interaction.actionBlock) && + typeof interaction.actionBlock === 'object' && + !Array.isArray(interaction.actionBlock) + ) + }) + } catch { + return false + } +} + +export const hasWaitingInteraction = (blocks: AssistantMessageBlock[]): boolean => + blocks.some((block) => isWaitingInteractionBlock(block) || hasSubagentWaitingInteraction(block)) + export const extractWaitingInteraction = ( blocks: AssistantMessageBlock[], messageId: string ): DeepChatInternalSessionWaitingInteraction | null => { for (let index = 0; index < blocks.length; index += 1) { const block = blocks[index] - if ( - block.type !== 'action' || - block.status !== 'pending' || - block.extra?.needsUserAction !== true || - !block.tool_call?.id - ) { - continue - } + if (!isWaitingInteractionBlock(block)) continue + const toolCallId = block.tool_call?.id + if (!toolCallId) continue if (block.action_type === 'tool_call_permission') { return { type: 'permission', messageId, - toolCallId: block.tool_call.id, + toolCallId, actionBlock: JSON.parse(JSON.stringify(block)) as AssistantMessageBlock } } @@ -101,7 +144,7 @@ export const extractWaitingInteraction = ( return { type: 'question', messageId, - toolCallId: block.tool_call.id, + toolCallId, actionBlock: JSON.parse(JSON.stringify(block)) as AssistantMessageBlock } } diff --git a/src/main/app/composition.ts b/src/main/app/composition.ts index 6758642ce..aa0365081 100644 --- a/src/main/app/composition.ts +++ b/src/main/app/composition.ts @@ -1728,6 +1728,14 @@ export async function createMainProcessControl(dependencies: { turn: sessionTurn, projection: sessionQuery, sessions: appSessionService, + hasWaitingDescendantInteraction: (runId) => + liveDelegationRepository + .listActiveTurns() + .some( + ({ delegation, turn }) => + (turn.status === 'waiting_permission' || turn.status === 'waiting_question') && + resolveSessionRunId(delegation.parentSessionId) === runId + ), eventHub: typedEventHub, log: logger }) diff --git a/src/main/cli/index.ts b/src/main/cli/index.ts index ddebae5e5..ca12d5b95 100644 --- a/src/main/cli/index.ts +++ b/src/main/cli/index.ts @@ -35,7 +35,7 @@ export { createCliProviderModelAdminRoutes, type CliProviderModelAdminDependencies } from './providerModelAdminRoutes' -export { CLI_SURFACE_V1, getCliSurfaceEntry, listCliSurfaceCapabilities } from './surface' +export { CLI_SURFACE_V2, getCliSurfaceEntry, listCliSurfaceCapabilities } from './surface' export { CliMutationGuard, type CliApprovalPresentationPort, diff --git a/src/main/cli/routes.ts b/src/main/cli/routes.ts index 791a93f59..f435a517f 100644 --- a/src/main/cli/routes.ts +++ b/src/main/cli/routes.ts @@ -83,7 +83,7 @@ export function createCliRoutes(deps: { { id: 'surface' as const, status: capabilities.length > 0 ? ('ok' as const) : ('error' as const), - message: `${capabilities.length} V1 methods are registered` + message: `${capabilities.length} surface V${LOCAL_CONTROL_SURFACE_VERSION} methods are registered` }, { id: 'renderer' as const, diff --git a/src/main/cli/runService.ts b/src/main/cli/runService.ts index f51d22947..63d8594d0 100644 --- a/src/main/cli/runService.ts +++ b/src/main/cli/runService.ts @@ -27,7 +27,10 @@ import { runsTurnFailedEvent } from '@shared/contracts/events' import { extractUserMessageInput } from '@/session/data/userMessageContent' -import { buildAssistantResponseMarkdown } from '@/agent/deepchat/runtime/sessionUpdates' +import { + buildAssistantResponseMarkdown, + hasWaitingInteraction +} from '@/agent/deepchat/runtime/sessionUpdates' import type { AssistantMessageBlock } from '@shared/types/agent-interface' import { createRouteMap, @@ -59,6 +62,7 @@ type RunTurnPort = Readonly<{ type RunProjectionPort = Readonly<{ getSession(sessionId: string): Promise + getMessages(sessionId: string): Promise listMessagesPage( sessionId: string, options?: { limit?: number; cursor?: MessagePageCursor | null } @@ -74,6 +78,7 @@ export type CliRunServiceOptions = Readonly<{ turn: RunTurnPort projection: RunProjectionPort sessions: RunSessionStorePort + hasWaitingDescendantInteraction(runId: string): boolean eventHub: TypedEventHub now?: () => number log?: Pick @@ -111,6 +116,27 @@ function messageText(message: ChatMessageRecord): string { } } +function runPhase( + status: SessionWithState['status'], + messages: readonly ChatMessageRecord[], + hasWaitingDescendantInteraction: boolean +): PublicRunSnapshot['phase'] { + if (status !== 'generating') return 'terminal' + if (hasWaitingDescendantInteraction) return 'awaiting_interaction' + for (const message of messages) { + if (message.role !== 'assistant') continue + try { + const blocks = JSON.parse(message.content) as AssistantMessageBlock[] + if (Array.isArray(blocks) && hasWaitingInteraction(blocks)) { + return 'awaiting_interaction' + } + } catch { + // A malformed transcript cannot manufacture an interaction wait. + } + } + return 'running' +} + function toPublicMessage(message: ChatMessageRecord): PublicRunMessage { const text = truncateUtf8(messageText(message), RUN_MESSAGE_MAX_TEXT_BYTES) return { @@ -412,7 +438,6 @@ export class CliRunService { const payload = data as { runId?: unknown; sessionId?: unknown; status?: unknown } if (event === runsTurnFailedEvent.name) return payload.runId === runId if (payload.sessionId !== runId) return false - if (event === 'chat.stream.completed' || event === 'chat.stream.failed') return true if (event !== 'sessions.status.changed') return false const status = payload.status return status === 'idle' || status === 'error' @@ -450,6 +475,10 @@ export class CliRunService { this.requireRunSnapshot(runId), this.options.projection.listMessagesPage(runId, { limit, cursor: cursor ?? null }) ]) + const phaseMessages = + session.status === 'generating' && (cursor != null || page.hasMore) + ? await this.options.projection.getMessages(runId) + : page.messages const projectedPage = projectMessagePage(page) return runsGetRoute.output.parse({ runId, @@ -457,6 +486,11 @@ export class CliRunService { agentId: session.agentId, title: session.title, status: session.status, + phase: runPhase( + session.status, + phaseMessages, + session.status === 'generating' && this.options.hasWaitingDescendantInteraction(runId) + ), providerId: session.providerId, modelId: session.modelId, createdAt: session.createdAt, diff --git a/src/main/cli/server.ts b/src/main/cli/server.ts index 475e9de78..c2df8daf6 100644 --- a/src/main/cli/server.ts +++ b/src/main/cli/server.ts @@ -52,7 +52,7 @@ import { type CliControlLayout } from './descriptor' import { CliRequestError } from './errors' -import { CLI_SURFACE_V1 } from './surface' +import { CLI_SURFACE_V2 } from './surface' import type { CliSurfaceEntry } from './surface' import type { CliRuntimeStatus } from './routes' import type { ArtifactSpool } from './artifactSpool' @@ -287,7 +287,7 @@ export class CliServer { this.platform = dependencies.platform ?? process.platform this.pid = dependencies.pid ?? process.pid this.log = dependencies.log ?? console - this.surface = new Map(dependencies.surface ?? CLI_SURFACE_V1) + this.surface = new Map(dependencies.surface ?? CLI_SURFACE_V2) } getStatus(): CliRuntimeStatus { @@ -658,7 +658,7 @@ export class CliServer { routeMethod = declaredMethod entry = this.surface.get(declaredMethod) if (!entry || entry.transport !== requestTransport) { - throw new CliRequestError('not_found', 'Method is not exposed by CLI surface V1', { + throw new CliRequestError('not_found', 'Method is not exposed by the CLI surface', { httpStatus: 404 }) } @@ -711,7 +711,7 @@ export class CliServer { } entry ??= this.surface.get(rpcRequest.method) if (!entry || entry.transport !== requestTransport) { - throw new CliRequestError('not_found', 'Method is not exposed by CLI surface V1', { + throw new CliRequestError('not_found', 'Method is not exposed by the CLI surface', { httpStatus: 404 }) } diff --git a/src/main/cli/surface.ts b/src/main/cli/surface.ts index c8eb44fa6..f8d8c4a14 100644 --- a/src/main/cli/surface.ts +++ b/src/main/cli/surface.ts @@ -426,7 +426,7 @@ const mediaEntry = (contract: RouteContract): CliSurfaceEntry => ({ } }) -const CLI_SURFACE_V1_ENTRIES = [ +const CLI_SURFACE_V2_ENTRIES = [ { contract: modelsInvokeRoute, effect: 'compute', @@ -1002,14 +1002,14 @@ function createSurfaceRegistry( return registry } -export const CLI_SURFACE_V1 = createSurfaceRegistry(CLI_SURFACE_V1_ENTRIES) +export const CLI_SURFACE_V2 = createSurfaceRegistry(CLI_SURFACE_V2_ENTRIES) export function getCliSurfaceEntry(method: string): CliSurfaceEntry | undefined { - return CLI_SURFACE_V1.get(method) + return CLI_SURFACE_V2.get(method) } export function listCliSurfaceCapabilities(): CliCapability[] { - return Array.from(CLI_SURFACE_V1.values(), (entry) => ({ + return Array.from(CLI_SURFACE_V2.values(), (entry) => ({ method: entry.contract.name, possibleEffects: [...listCliSurfaceEffects(entry)], callers: [...entry.callers], diff --git a/src/shared/contracts/localControl.ts b/src/shared/contracts/localControl.ts index 3c35b0d64..5b883d733 100644 --- a/src/shared/contracts/localControl.ts +++ b/src/shared/contracts/localControl.ts @@ -2,7 +2,7 @@ import { z } from 'zod' import { JsonValueSchema, TimestampMsSchema, type JsonValue } from './json' export const LOCAL_CONTROL_PROTOCOL_VERSION = 1 as const -export const LOCAL_CONTROL_SURFACE_VERSION = 1 as const +export const LOCAL_CONTROL_SURFACE_VERSION = 2 as const export const LOCAL_CONTROL_MAX_REQUEST_TIMEOUT_MS = 30 * 60_000 export const LOCAL_CONTROL_MAX_JSON_RESPONSE_BYTES = 16 * 1024 * 1024 export const LOCAL_CONTROL_MAX_STREAM_RECORD_BYTES = 20 * 1024 * 1024 diff --git a/src/shared/contracts/routes/ocr.routes.ts b/src/shared/contracts/routes/ocr.routes.ts index 505499767..6fa1d4e65 100644 --- a/src/shared/contracts/routes/ocr.routes.ts +++ b/src/shared/contracts/routes/ocr.routes.ts @@ -1,5 +1,6 @@ import { z } from 'zod' import { defineRouteContract } from '../common' +import { LOCAL_CONTROL_PROTOCOL_VERSION, LOCAL_CONTROL_SURFACE_VERSION } from '../localControl' import { ATTACHMENT_OCR_MAX_TEXT_CHARACTERS, ATTACHMENT_PDF_OCR_MAX_PAGE_SPANS, @@ -138,8 +139,8 @@ export const OcrBenchmarkSchema = z inputBytes: z.number().int().positive().max(OCR_EXTRACTION_MAX_INPUT_BYTES), durationMs: OcrPublicTimingSchema, appVersion: z.string().min(1).max(128), - protocolVersion: z.literal(1), - surfaceVersion: z.literal(1) + protocolVersion: z.literal(LOCAL_CONTROL_PROTOCOL_VERSION), + surfaceVersion: z.literal(LOCAL_CONTROL_SURFACE_VERSION) }) .strict() .superRefine((benchmark, context) => { diff --git a/src/shared/contracts/routes/runs.routes.ts b/src/shared/contracts/routes/runs.routes.ts index 142b11e1e..937712ae6 100644 --- a/src/shared/contracts/routes/runs.routes.ts +++ b/src/shared/contracts/routes/runs.routes.ts @@ -15,6 +15,7 @@ export const RUN_MAX_MESSAGE_PAGE_SIZE = 100 export const RunIdSchema = EntityIdSchema.max(128) export const RunEventCursorSchema = LocalControlEventCursorSchema +export const PublicRunPhaseSchema = z.enum(['running', 'awaiting_interaction', 'terminal']) const BoundedIdentifierSchema = z.string().trim().min(1).max(256) const PublicRunMessageTextSchema = z @@ -54,6 +55,7 @@ export const PublicRunSnapshotSchema = z agentId: EntityIdSchema, title: z.string(), status: SessionStatusSchema, + phase: PublicRunPhaseSchema, providerId: z.string(), modelId: z.string(), createdAt: TimestampMsSchema, @@ -144,6 +146,7 @@ export const eventsSubscribeRoute = defineRouteContract({ }) export type PublicRunMessage = z.infer +export type PublicRunPhase = z.infer export type PublicRunSnapshot = z.infer export type RunDetachedInput = z.infer export type RunDetachedOutput = z.infer diff --git a/test/main/cli/client.test.ts b/test/main/cli/client.test.ts index e4e8c13cd..5b84b0658 100644 --- a/test/main/cli/client.test.ts +++ b/test/main/cli/client.test.ts @@ -22,7 +22,7 @@ const temporaryDirectories: string[] = [] const testDescriptor: LocalControlDescriptor = { protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, appVersion: '9.8.7', endpoint: { kind: 'unix', path: '/tmp/deepchat-test.sock' }, pid: 1, @@ -189,7 +189,7 @@ describe('bundled CLI client', () => { await expect(invocation.result).resolves.toBe(0) expect(invocation.stdout.read()).toContain('DeepChat 9.8.7') - expect(invocation.stdout.read()).toContain('Protocol 1, surface 1') + expect(invocation.stdout.read()).toContain('Protocol 1, surface 2') expect(invocation.stderr.read()).toBe('') expect(dispatch).toHaveBeenCalledOnce() }) @@ -315,7 +315,7 @@ describe('bundled CLI client', () => { const invokeRpc = vi.fn(async (invocation) => LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { @@ -363,7 +363,7 @@ describe('bundled CLI client', () => { await onEvent( LocalControlEventEnvelopeSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, sequence: 0, timestamp: 1_000, requestId: invocation.id, @@ -380,7 +380,7 @@ describe('bundled CLI client', () => { ) return LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { runId: 'run-1', lastCursor: 'epoch-1:7' } @@ -423,7 +423,7 @@ describe('bundled CLI client', () => { await onEvent( LocalControlEventEnvelopeSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, sequence: 0, timestamp: 1_000, requestId: invocation.id, @@ -466,7 +466,7 @@ describe('bundled CLI client', () => { await onEvent( LocalControlEventEnvelopeSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, sequence: 0, timestamp: 1_000, requestId: invocation.id, @@ -504,7 +504,7 @@ describe('bundled CLI client', () => { await onEvent( LocalControlEventEnvelopeSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, sequence: 0, timestamp: 1_000, requestId: invocation.id, @@ -514,7 +514,7 @@ describe('bundled CLI client', () => { ) return LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { @@ -671,7 +671,7 @@ describe('bundled CLI client', () => { const invokeUpload = vi.fn(async (invocation) => LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { @@ -756,7 +756,7 @@ describe('bundled CLI client', () => { const invokeRpc = vi.fn(async (invocation) => LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { @@ -810,7 +810,7 @@ describe('bundled CLI client', () => { const invokeRpc = vi.fn(async (invocation) => LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { config } @@ -851,7 +851,7 @@ describe('bundled CLI client', () => { const invokeRpc = vi.fn(async (invocation) => LocalControlRpcResponseSchema.parse({ protocolVersion: 1, - surfaceVersion: 1, + surfaceVersion: 2, id: invocation.id, ok: true, result: { diff --git a/test/main/cli/inputCapabilityServices.test.ts b/test/main/cli/inputCapabilityServices.test.ts index ce0c32054..7257a99fa 100644 --- a/test/main/cli/inputCapabilityServices.test.ts +++ b/test/main/cli/inputCapabilityServices.test.ts @@ -270,7 +270,7 @@ describe('CLI audio transcription and OCR services', () => { inputBytes: bytes.length, appVersion: '1.2.3', protocolVersion: 1, - surfaceVersion: 1 + surfaceVersion: 2 }, engine: { requestedBackend: 'auto', diff --git a/test/main/cli/runService.test.ts b/test/main/cli/runService.test.ts index e907ccd8f..5b33e6601 100644 --- a/test/main/cli/runService.test.ts +++ b/test/main/cli/runService.test.ts @@ -72,11 +72,79 @@ function createMessage(overrides: Partial): ChatMessageRecord } } +function createInteractionMessage( + actionType: 'tool_call_permission' | 'question_request', + needsUserAction: boolean | undefined = true +): ChatMessageRecord { + return createMessage({ + id: `message-${actionType}`, + orderSeq: 2, + role: 'assistant', + status: 'pending', + content: JSON.stringify([ + { + type: 'action', + action_type: actionType, + status: 'pending', + timestamp: 100, + tool_call: { id: `tool-${actionType}`, name: 'test_tool', params: '{}' }, + extra: needsUserAction === undefined ? {} : { needsUserAction } + } + ]) + }) +} + +function createSubagentInteractionMessage( + type: 'permission' | 'question', + waiting = true, + progress?: string +): ChatMessageRecord { + return createMessage({ + id: `message-subagent-${type}`, + orderSeq: 2, + role: 'assistant', + status: 'pending', + content: JSON.stringify([ + { + type: 'tool_call', + status: 'loading', + timestamp: 100, + tool_call: { id: 'subagent-orchestrator', name: 'subagent_orchestrator', params: '{}' }, + extra: { + subagentProgress: + progress ?? + JSON.stringify({ + tasks: [ + { + sessionId: 'child-session', + waitingInteraction: waiting + ? { + type, + messageId: 'child-message', + toolCallId: 'child-tool', + actionBlock: { + type: 'action', + status: 'pending', + action_type: + type === 'question' ? 'question_request' : 'tool_call_permission' + } + } + : null + } + ] + }) + } + } + ]) + }) +} + function createHarness( overrides: { session?: SessionWithState storedSession?: SessionRecord | null messages?: ChatMessageRecord[] + hasWaitingDescendantInteraction?: boolean } = {} ): { service: CliRunService @@ -97,6 +165,7 @@ function createHarness( } const projection = { getSession: vi.fn(async () => session), + getMessages: vi.fn(async () => overrides.messages ?? []), listMessagesPage: vi.fn(async () => ({ messages: overrides.messages ?? [], nextCursor: null, @@ -120,6 +189,9 @@ function createHarness( turn, projection, sessions, + hasWaitingDescendantInteraction: vi.fn( + () => overrides.hasWaitingDescendantInteraction ?? false + ), eventHub: hub, now: () => 200, log @@ -150,6 +222,43 @@ async function nextEvent(events: AsyncIterable): Promise +}> { + const emitted: string[] = [] + let snapshotEmitted!: () => void + const snapshotReady = new Promise((resolve) => { + snapshotEmitted = resolve + }) + const result = service.dispatchStream( + eventsSubscribeRoute.name, + { runId: 'run-1' }, + humanCaller, + new AbortController().signal, + async (event) => { + emitted.push(event) + if (event === 'runs.snapshot') snapshotEmitted() + } + ) + await snapshotReady + return { emitted, result } +} + +async function expectWatcherPending(result: Promise): Promise { + let settled = false + void result.then( + () => { + settled = true + }, + () => { + settled = true + } + ) + await Promise.resolve() + expect(settled).toBe(false) +} + describe('CliRunService', () => { it('creates a durable default-permission session before starting its initial turn', async () => { const { service, hub, lifecycle, turn } = createHarness() @@ -257,12 +366,109 @@ describe('CliRunService', () => { expect(result.messages[0]).toMatchObject({ role: 'user', text: 'hello' }) expect(result.messages[1].textTruncated).toBe(true) + expect(result.phase).toBe('running') expect(Buffer.byteLength(result.messages[1].text, 'utf8')).toBeLessThanOrEqual( RUN_MESSAGE_MAX_TEXT_BYTES ) expect(JSON.stringify(result)).not.toContain('private-provider-detail') }) + it.each([ + ['permission', 'tool_call_permission', true, 'awaiting_interaction'], + ['question', 'question_request', true, 'awaiting_interaction'], + ['recovered question', 'question_request', undefined, 'awaiting_interaction'], + ['resolved permission', 'tool_call_permission', false, 'running'], + ['resolved question', 'question_request', false, 'running'] + ] as const)( + 'projects a %s with the correct phase', + async (_kind, actionType, needsAction, phase) => { + const { service } = createHarness({ + messages: [createInteractionMessage(actionType, needsAction)] + }) + + await expect( + invokeRoute(service, runsGetRoute.name, { runId: 'run-1' }) + ).resolves.toMatchObject({ + phase, + status: 'generating' + }) + } + ) + + it.each([ + ['permission', true, 'awaiting_interaction'], + ['question', true, 'awaiting_interaction'], + ['cleared interaction', false, 'running'] + ] as const)( + 'projects a legacy subagent %s with the correct phase', + async (_kind, waiting, phase) => { + const type = _kind === 'question' ? 'question' : 'permission' + const { service } = createHarness({ + messages: [createSubagentInteractionMessage(type, waiting)] + }) + + await expect( + invokeRoute(service, runsGetRoute.name, { runId: 'run-1' }) + ).resolves.toMatchObject({ phase }) + } + ) + + it('projects a current live-delegation wait as awaiting_interaction', async () => { + const { service } = createHarness({ hasWaitingDescendantInteraction: true }) + + await expect( + invokeRoute(service, runsGetRoute.name, { runId: 'run-1' }) + ).resolves.toMatchObject({ phase: 'awaiting_interaction' }) + }) + + it('ignores malformed subagent progress when projecting phase', async () => { + const { service } = createHarness({ + messages: [createSubagentInteractionMessage('permission', true, '{')] + }) + + await expect( + invokeRoute(service, runsGetRoute.name, { runId: 'run-1' }) + ).resolves.toMatchObject({ phase: 'running' }) + }) + + it('derives phase from the latest message when reading an older transcript page', async () => { + const { service, projection } = createHarness() + vi.mocked(projection.listMessagesPage).mockResolvedValueOnce({ + messages: [createMessage({})], + nextCursor: null, + hasMore: false + }) + vi.mocked(projection.getMessages).mockResolvedValueOnce([ + createInteractionMessage('question_request') + ]) + + await expect( + invokeRoute(service, runsGetRoute.name, { + runId: 'run-1', + cursor: { orderSeq: 2, id: 'message-2' } + }) + ).resolves.toMatchObject({ phase: 'awaiting_interaction' }) + expect(projection.getMessages).toHaveBeenCalledWith('run-1') + }) + + it('derives phase independently of a limited transcript page', async () => { + const { service, projection } = createHarness() + vi.mocked(projection.listMessagesPage).mockResolvedValueOnce({ + messages: [createMessage({})], + nextCursor: { orderSeq: 1, id: 'message-1' }, + hasMore: true + }) + vi.mocked(projection.getMessages).mockResolvedValueOnce([ + createMessage({}), + createInteractionMessage('tool_call_permission') + ]) + + await expect( + invokeRoute(service, runsGetRoute.name, { runId: 'run-1', limit: 1 }) + ).resolves.toMatchObject({ phase: 'awaiting_interaction' }) + expect(projection.getMessages).toHaveBeenCalledWith('run-1') + }) + it('enforces the public message limit in UTF-8 bytes', () => { const message = { id: 'message-1', @@ -368,8 +574,10 @@ describe('CliRunService', () => { }) }) - it('emits a recovery snapshot and then terminates on a targeted completion event', async () => { - const { service, hub } = createHarness() + it('keeps watching after a provider round and terminates on the root Session status', async () => { + const { service, hub } = createHarness({ + messages: [createInteractionMessage('tool_call_permission')] + }) const emitted: Array<{ event: string; data: unknown; context: unknown }> = [] let snapshotEmitted!: () => void const snapshotReady = new Promise((resolve) => { @@ -398,62 +606,177 @@ describe('CliRunService', () => { }, { kind: 'run', runId: 'run-1' } ) + await vi.waitFor(() => + expect(emitted.map((entry) => entry.event)).toContain('chat.stream.completed') + ) + let settled = false + void result.then( + () => { + settled = true + }, + () => { + settled = true + } + ) + await Promise.resolve() + expect(settled).toBe(false) + + hub.publish( + 'sessions.status.changed', + { sessionId: 'run-1', status: 'idle', version: 301 }, + { kind: 'run', runId: 'run-1' } + ) - await expect(result).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:1' }) - expect(emitted.map((entry) => entry.event)).toEqual(['runs.snapshot', 'chat.stream.completed']) + await expect(result).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:2' }) + expect(emitted.map((entry) => entry.event)).toEqual([ + 'runs.snapshot', + 'chat.stream.completed', + 'sessions.status.changed' + ]) + expect(emitted[0].data).toMatchObject({ + run: { phase: 'awaiting_interaction' } + }) expect(emitted[0].context).toEqual({ runId: 'run-1', cursor: 'test-epoch_1:0' }) }) - it('does not terminate a root run watcher when a descendant session completes', async () => { + it('keeps watching after a provider failure until the root Session enters error', async () => { + const { service, hub } = createHarness() + const { emitted, result } = await startRunWatcher(service) + + hub.publish( + 'chat.stream.failed', + { + requestId: 'request-1', + sessionId: 'run-1', + messageId: 'message-2', + failedAt: 300, + error: 'provider round failed' + }, + { kind: 'run', runId: 'run-1' } + ) + await vi.waitFor(() => expect(emitted).toContain('chat.stream.failed')) + await expectWatcherPending(result) + + hub.publish( + 'sessions.status.changed', + { sessionId: 'run-1', status: 'error', version: 301 }, + { kind: 'run', runId: 'run-1' } + ) + + await expect(result).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:2' }) + expect(emitted).toEqual(['runs.snapshot', 'chat.stream.failed', 'sessions.status.changed']) + }) + + it('keeps watching after replaying a provider terminal event from a cursor', async () => { const { service, hub } = createHarness() + hub.publish( + 'chat.stream.completed', + { + requestId: 'request-1', + sessionId: 'run-1', + messageId: 'message-2', + completedAt: 300 + }, + { kind: 'run', runId: 'run-1' } + ) const emitted: string[] = [] - let snapshotEmitted!: () => void - const snapshotReady = new Promise((resolve) => { - snapshotEmitted = resolve - }) + const controller = new AbortController() const result = service.dispatchStream( eventsSubscribeRoute.name, - { runId: 'run-1' }, + { runId: 'run-1', cursor: 'test-epoch_1:0' }, humanCaller, - new AbortController().signal, + controller.signal, async (event) => { emitted.push(event) - if (event === 'runs.snapshot') snapshotEmitted() } ) - await snapshotReady + await vi.waitFor(() => expect(emitted).toEqual(['chat.stream.completed'])) + await expectWatcherPending(result) + + hub.publish( + 'sessions.status.changed', + { sessionId: 'run-1', status: 'idle', version: 301 }, + { kind: 'run', runId: 'run-1' } + ) + + await expect(result).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:2' }) + expect(emitted).toEqual(['chat.stream.completed', 'sessions.status.changed']) + }) + + it('finishes when cursor catch-up includes an already-terminal root Session', async () => { + const idleSession = { ...baseSession, status: 'idle' as const } + const { service, hub } = createHarness({ session: idleSession }) hub.publish( 'chat.stream.completed', { - requestId: 'request-child', - sessionId: 'child-session', - messageId: 'message-child', + requestId: 'request-1', + sessionId: 'run-1', + messageId: 'message-2', completedAt: 300 }, { kind: 'run', runId: 'run-1' } ) - await vi.waitFor(() => expect(emitted).toContain('chat.stream.completed')) - let settled = false - void result.finally(() => { - settled = true - }) - await Promise.resolve() - expect(settled).toBe(false) + hub.publish( + 'sessions.status.changed', + { sessionId: 'run-1', status: 'idle', version: 301 }, + { kind: 'run', runId: 'run-1' } + ) + const emitted: string[] = [] + + await expect( + service.dispatchStream( + eventsSubscribeRoute.name, + { runId: 'run-1', cursor: 'test-epoch_1:0' }, + humanCaller, + new AbortController().signal, + async (event) => { + emitted.push(event) + } + ) + ).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:2' }) + expect(emitted).toEqual(['chat.stream.completed', 'sessions.status.changed']) + }) + + it('terminates a watcher when detached initial-turn startup fails', async () => { + const { service, hub } = createHarness() + const { emitted, result } = await startRunWatcher(service) hub.publish( - 'chat.stream.completed', + 'runs.turn.failed', { - requestId: 'request-root', + runId: 'run-1', sessionId: 'run-1', - messageId: 'message-root', - completedAt: 301 + failedAt: 300, + error: 'Detached Agent run could not start' }, { kind: 'run', runId: 'run-1' } ) + await expect(result).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:1' }) + expect(emitted).toEqual(['runs.snapshot', 'runs.turn.failed']) + }) + + it('does not terminate a root run watcher when a descendant Session becomes terminal', async () => { + const { service, hub } = createHarness() + const { emitted, result } = await startRunWatcher(service) + + hub.publish( + 'sessions.status.changed', + { sessionId: 'child-session', status: 'idle', version: 300 }, + { kind: 'run', runId: 'run-1' } + ) + await vi.waitFor(() => expect(emitted).toContain('sessions.status.changed')) + await expectWatcherPending(result) + + hub.publish( + 'sessions.status.changed', + { sessionId: 'run-1', status: 'idle', version: 301 }, + { kind: 'run', runId: 'run-1' } + ) + await expect(result).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:2' }) - expect(emitted).toEqual(['runs.snapshot', 'chat.stream.completed', 'chat.stream.completed']) + expect(emitted).toEqual(['runs.snapshot', 'sessions.status.changed', 'sessions.status.changed']) }) it('returns immediately after recovering an already-terminal run', async () => { @@ -472,7 +795,10 @@ describe('CliRunService', () => { ).resolves.toEqual({ runId: 'run-1', lastCursor: 'test-epoch_1:0' }) expect(emit).toHaveBeenCalledWith( 'runs.snapshot', - expect.objectContaining({ recoveryReason: 'cursor_missing' }), + expect.objectContaining({ + recoveryReason: 'cursor_missing', + run: expect.objectContaining({ phase: 'terminal' }) + }), { runId: 'run-1', cursor: 'test-epoch_1:0' } ) }) diff --git a/test/main/cli/server.test.ts b/test/main/cli/server.test.ts index 03bbec94a..7863bfc67 100644 --- a/test/main/cli/server.test.ts +++ b/test/main/cli/server.test.ts @@ -442,10 +442,15 @@ describe('CLI local transport', () => { it('enforces protocol versions and the explicit surface before dispatch', async () => { const { descriptor, dispatch } = await createTestServer() - const incompatible = await rpcRequest(descriptor, { protocolVersion: 2 }) + const incompatibleProtocol = await rpcRequest(descriptor, { protocolVersion: 2 }) + const legacySurface = await rpcRequest(descriptor, { surfaceVersion: 1 }) const hidden = await rpcRequest(descriptor, { method: 'settings.getSnapshot' }) - expect(incompatible).toMatchObject({ + expect(incompatibleProtocol).toMatchObject({ + status: 409, + body: { ok: false, error: { code: 'unsupported_version' } } + }) + expect(legacySurface).toMatchObject({ status: 409, body: { ok: false, error: { code: 'unsupported_version' } } }) diff --git a/test/main/cli/surface.test.ts b/test/main/cli/surface.test.ts index 0a3f7bba9..e05ee2fee 100644 --- a/test/main/cli/surface.test.ts +++ b/test/main/cli/surface.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from 'vitest' import { DEEPCHAT_ROUTE_CATALOG } from '@shared/contracts/routes' import { - CLI_SURFACE_V1, + CLI_SURFACE_V2, getCliSurfaceEntry, listCliSurfaceCapabilities, resolveCliSurfaceEffect @@ -10,9 +10,9 @@ import { const humanApprovalCaller = { principal: 'human' } as const const agentApprovalCaller = { principal: 'agent' } as const -describe('CLI surface V1', () => { +describe('CLI surface V2', () => { it('contains only explicit canonical route contracts', () => { - const methods = Array.from(CLI_SURFACE_V1.keys()).sort() + const methods = Array.from(CLI_SURFACE_V2.keys()).sort() expect(methods).toEqual([ 'artifacts.delete', @@ -62,7 +62,7 @@ describe('CLI surface V1', () => { 'speech.generate', 'videos.generate' ]) - for (const [method, entry] of CLI_SURFACE_V1) { + for (const [method, entry] of CLI_SURFACE_V2) { expect(entry.contract).toBe( DEEPCHAT_ROUTE_CATALOG[method as keyof typeof DEEPCHAT_ROUTE_CATALOG] ) @@ -79,7 +79,7 @@ describe('CLI surface V1', () => { }) it('keeps Agent mutation policy as an explicit operation opt-in', () => { - const policies = Array.from(CLI_SURFACE_V1, ([method, entry]) => ({ method, entry })).filter( + const policies = Array.from(CLI_SURFACE_V2, ([method, entry]) => ({ method, entry })).filter( ({ entry }) => entry.agentPolicy !== undefined ) From b87426a1cf409fc5f868d8a4bb24f3a6ead7e82d Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Mon, 10 Aug 2026 22:28:16 +0800 Subject: [PATCH 2/3] fix(cli): avoid full transcript phase scans --- docs/architecture/local-control-plane/spec.md | 2 +- src/main/app/composition.ts | 2 + src/main/cli/runService.ts | 4 +- .../session/data/tables/deepchatMessages.ts | 22 ++++ src/main/session/data/transcript.ts | 5 + test/main/cli/runService.test.ts | 60 +++++----- .../data/tables/deepchatMessagesTable.test.ts | 104 +++++++++++++++++- test/main/session/data/transcript.test.ts | 22 ++++ 8 files changed, 179 insertions(+), 42 deletions(-) diff --git a/docs/architecture/local-control-plane/spec.md b/docs/architecture/local-control-plane/spec.md index 62f39eddc..cb20df852 100644 --- a/docs/architecture/local-control-plane/spec.md +++ b/docs/architecture/local-control-plane/spec.md @@ -241,7 +241,7 @@ Public method names describe their domain (`models.invoke`, `providers.listPubli `sessions.runDetached`). The `cli.*` namespace is reserved for behavior that exists only to operate or diagnose the bundled CLI. -### V1 Capability Matrix +### V2 Capability Matrix `H` means an authenticated human CLI connection. `A` means a short-lived Agent connection with the listed scope. “Policy” means the renderer-only effect policy may be required; it never means a CLI diff --git a/src/main/app/composition.ts b/src/main/app/composition.ts index aa0365081..6388c3fb0 100644 --- a/src/main/app/composition.ts +++ b/src/main/app/composition.ts @@ -1728,6 +1728,8 @@ export async function createMainProcessControl(dependencies: { turn: sessionTurn, projection: sessionQuery, sessions: appSessionService, + getPendingAssistantMessages: (runId) => + sessionData.transcript.getPendingAssistantMessages(runId), hasWaitingDescendantInteraction: (runId) => liveDelegationRepository .listActiveTurns() diff --git a/src/main/cli/runService.ts b/src/main/cli/runService.ts index f00fb7ca9..b5fce851e 100644 --- a/src/main/cli/runService.ts +++ b/src/main/cli/runService.ts @@ -62,7 +62,6 @@ type RunTurnPort = Readonly<{ type RunProjectionPort = Readonly<{ getSession(sessionId: string): Promise - getMessages(sessionId: string): Promise listMessagesPage( sessionId: string, options?: { limit?: number; cursor?: MessagePageCursor | null } @@ -78,6 +77,7 @@ export type CliRunServiceOptions = Readonly<{ turn: RunTurnPort projection: RunProjectionPort sessions: RunSessionStorePort + getPendingAssistantMessages(runId: string): ChatMessageRecord[] hasWaitingDescendantInteraction(runId: string): boolean eventHub: TypedEventHub now?: () => number @@ -483,7 +483,7 @@ export class CliRunService { ]) const phaseMessages = session.status === 'generating' && (cursor != null || page.hasMore) - ? await this.options.projection.getMessages(runId) + ? this.options.getPendingAssistantMessages(runId) : page.messages const projectedPage = projectMessagePage(page) return runsGetRoute.output.parse({ diff --git a/src/main/session/data/tables/deepchatMessages.ts b/src/main/session/data/tables/deepchatMessages.ts index e7dd3eefd..41bf281a8 100644 --- a/src/main/session/data/tables/deepchatMessages.ts +++ b/src/main/session/data/tables/deepchatMessages.ts @@ -32,11 +32,22 @@ export interface DeepChatAssistantMessageIdentityRow { updated_at: number } +const PENDING_ASSISTANT_INDEX_SQL = ` + CREATE INDEX IF NOT EXISTS idx_deepchat_messages_pending_assistant + ON deepchat_messages(session_id, order_seq, id) + WHERE role = 'assistant' AND status = 'pending'; +` + export class DeepChatMessagesTable extends BaseTable { constructor(db: Database.Database) { super(db, 'deepchat_messages') } + override createTable(): void { + super.createTable() + this.db.exec(PENDING_ASSISTANT_INDEX_SQL) + } + getCreateTableSQL(): string { return ` CREATE TABLE IF NOT EXISTS deepchat_messages ( @@ -52,6 +63,7 @@ export class DeepChatMessagesTable extends BaseTable { updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_deepchat_messages_session ON deepchat_messages(session_id, order_seq); + ${PENDING_ASSISTANT_INDEX_SQL} ` } @@ -158,6 +170,16 @@ export class DeepChatMessagesTable extends BaseTable { .all(sessionId) as DeepChatMessageRow[] } + getPendingAssistantBySession(sessionId: string): DeepChatMessageRow[] { + return this.db + .prepare( + `SELECT * FROM deepchat_messages + WHERE session_id = ? AND role = 'assistant' AND status = 'pending' + ORDER BY order_seq, id` + ) + .all(sessionId) as DeepChatMessageRow[] + } + hasBySession(sessionId: string): boolean { return Boolean( this.db.prepare('SELECT 1 FROM deepchat_messages WHERE session_id = ? LIMIT 1').get(sessionId) diff --git a/src/main/session/data/transcript.ts b/src/main/session/data/transcript.ts index d57d97049..00edeb15d 100644 --- a/src/main/session/data/transcript.ts +++ b/src/main/session/data/transcript.ts @@ -445,6 +445,11 @@ export class SessionTranscript { return this.toRecords(rows) } + getPendingAssistantMessages(sessionId: string): ChatMessageRecord[] { + const rows = this.database.deepchatMessagesTable.getPendingAssistantBySession(sessionId) + return this.toRecords(rows) + } + hasMessages(sessionId: string): boolean { return this.database.deepchatMessagesTable.hasBySession(sessionId) } diff --git a/test/main/cli/runService.test.ts b/test/main/cli/runService.test.ts index f1b8d9527..a0673ceed 100644 --- a/test/main/cli/runService.test.ts +++ b/test/main/cli/runService.test.ts @@ -153,6 +153,7 @@ function createHarness( turn: CliRunServiceOptions['turn'] projection: CliRunServiceOptions['projection'] sessions: CliRunServiceOptions['sessions'] + getPendingAssistantMessages: ReturnType log: { warn: ReturnType } } { const session = overrides.session ?? baseSession @@ -165,13 +166,17 @@ function createHarness( } const projection = { getSession: vi.fn(async () => session), - getMessages: vi.fn(async () => overrides.messages ?? []), listMessagesPage: vi.fn(async () => ({ messages: overrides.messages ?? [], nextCursor: null, hasMore: false })) } + const getPendingAssistantMessages = vi.fn(() => + (overrides.messages ?? []).filter( + (message) => message.role === 'assistant' && message.status === 'pending' + ) + ) const sessions = { get: vi.fn(() => overrides.storedSession === undefined ? (session as SessionRecord) : overrides.storedSession @@ -189,6 +194,7 @@ function createHarness( turn, projection, sessions, + getPendingAssistantMessages, hasWaitingDescendantInteraction: vi.fn( () => overrides.hasWaitingDescendantInteraction ?? false ), @@ -201,6 +207,7 @@ function createHarness( turn, projection, sessions, + getPendingAssistantMessages, log } } @@ -246,17 +253,14 @@ async function startRunWatcher(service: CliRunService): Promise<{ } async function expectWatcherPending(result: Promise): Promise { - let settled = false - void result.then( - () => { - settled = true - }, - () => { - settled = true - } - ) - await Promise.resolve() - expect(settled).toBe(false) + const outcome = await Promise.race([ + result.then( + () => 'settled' as const, + () => 'settled' as const + ), + new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), 10)) + ]) + expect(outcome).toBe('pending') } describe('CliRunService', () => { @@ -433,16 +437,17 @@ describe('CliRunService', () => { ).resolves.toMatchObject({ phase: 'running' }) }) - it('derives phase from the latest message when reading an older transcript page', async () => { - const { service, projection } = createHarness() + it.each([ + ['direct question', createInteractionMessage('question_request')], + ['legacy subagent wait', createSubagentInteractionMessage('permission')] + ])('derives phase from a pending %s outside an older transcript page', async (_kind, message) => { + const { service, projection, getPendingAssistantMessages } = createHarness() vi.mocked(projection.listMessagesPage).mockResolvedValueOnce({ messages: [createMessage({})], nextCursor: null, hasMore: false }) - vi.mocked(projection.getMessages).mockResolvedValueOnce([ - createInteractionMessage('question_request') - ]) + getPendingAssistantMessages.mockReturnValueOnce([message]) await expect( invokeRoute(service, runsGetRoute.name, { @@ -450,25 +455,24 @@ describe('CliRunService', () => { cursor: { orderSeq: 2, id: 'message-2' } }) ).resolves.toMatchObject({ phase: 'awaiting_interaction' }) - expect(projection.getMessages).toHaveBeenCalledWith('run-1') + expect(getPendingAssistantMessages).toHaveBeenCalledWith('run-1') }) it('derives phase independently of a limited transcript page', async () => { - const { service, projection } = createHarness() + const { service, projection, getPendingAssistantMessages } = createHarness() vi.mocked(projection.listMessagesPage).mockResolvedValueOnce({ messages: [createMessage({})], nextCursor: { orderSeq: 1, id: 'message-1' }, hasMore: true }) - vi.mocked(projection.getMessages).mockResolvedValueOnce([ - createMessage({}), + getPendingAssistantMessages.mockReturnValueOnce([ createInteractionMessage('tool_call_permission') ]) await expect( invokeRoute(service, runsGetRoute.name, { runId: 'run-1', limit: 1 }) ).resolves.toMatchObject({ phase: 'awaiting_interaction' }) - expect(projection.getMessages).toHaveBeenCalledWith('run-1') + expect(getPendingAssistantMessages).toHaveBeenCalledWith('run-1') }) it('returns only the final assistant answer without exposing process blocks', async () => { @@ -813,17 +817,7 @@ describe('CliRunService', () => { await vi.waitFor(() => expect(emitted.map((entry) => entry.event)).toContain('chat.stream.completed') ) - let settled = false - void result.then( - () => { - settled = true - }, - () => { - settled = true - } - ) - await Promise.resolve() - expect(settled).toBe(false) + await expectWatcherPending(result) hub.publish( 'sessions.status.changed', diff --git a/test/main/session/data/tables/deepchatMessagesTable.test.ts b/test/main/session/data/tables/deepchatMessagesTable.test.ts index 7166c4000..5fd831842 100644 --- a/test/main/session/data/tables/deepchatMessagesTable.test.ts +++ b/test/main/session/data/tables/deepchatMessagesTable.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it, vi } from 'vitest' import { DeepChatAssistantBlocksTable } from '@/session/data/tables/deepchatAssistantBlocks' -import { DeepChatMessagesTable } from '@/session/data/tables/deepchatMessages' +import { + DeepChatMessagesTable, + type DeepChatMessageRow +} from '@/session/data/tables/deepchatMessages' import { DeepChatMessageTracesTable } from '@/session/data/tables/deepchatMessageTraces' import type { AssistantMessageBlock } from '@shared/types/agent-interface' import { Database, nativeSqliteDescribeIf } from '../../../nativeSqliteHarness' @@ -8,14 +11,14 @@ import { Database, nativeSqliteDescribeIf } from '../../../nativeSqliteHarness' const DatabaseCtor = Database! const describeIfNativeSqlite = nativeSqliteDescribeIf() -function createMessageRow(orderSeq: number) { +function createMessageRow(orderSeq: number): DeepChatMessageRow { return { id: `m${orderSeq}`, session_id: 's1', order_seq: orderSeq, - role: 'user' as const, + role: 'user', content: '{}', - status: 'sent' as const, + status: 'sent', is_context_edge: 0, metadata: '{}', created_at: orderSeq, @@ -24,9 +27,25 @@ function createMessageRow(orderSeq: number) { } } -function createMockDb(rows: ReturnType[]) { +function createMockDb(rows: DeepChatMessageRow[]) { return { prepare: vi.fn((sql: string) => { + if (sql.includes("role = 'assistant'") && sql.includes("status = 'pending'")) { + return { + all: (sessionId: string) => + rows + .filter( + (row) => + row.session_id === sessionId && + row.role === 'assistant' && + row.status === 'pending' + ) + .sort( + (left, right) => left.order_seq - right.order_seq || left.id.localeCompare(right.id) + ) + } + } + if (sql.includes('FROM deepchat_messages m') && sql.includes('ORDER BY m.order_seq DESC')) { return { all: ( @@ -81,6 +100,25 @@ describe('DeepChatMessagesTable', () => { expect(page[0]?.order_seq).toBe(502) expect(page[500]?.order_seq).toBe(2) }) + + it('queries only pending assistant messages for the requested session', () => { + const rows = [ + { ...createMessageRow(2), role: 'assistant', status: 'pending' }, + { ...createMessageRow(1), role: 'assistant', status: 'pending' }, + { ...createMessageRow(3), role: 'assistant', status: 'sent' }, + { ...createMessageRow(4), status: 'pending' }, + { + ...createMessageRow(5), + id: 'other', + session_id: 's2', + role: 'assistant', + status: 'pending' + } + ] satisfies DeepChatMessageRow[] + const table = new DeepChatMessagesTable(createMockDb(rows)) + + expect(table.getPendingAssistantBySession('s1').map((row) => row.id)).toEqual(['m1', 'm2']) + }) }) describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { @@ -148,6 +186,48 @@ describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { } }) + it('loads only pending assistant messages for one session', () => { + const { db, table } = createTable() + try { + for (const row of [ + { id: 'pending-2', sessionId: 's1', orderSeq: 2, role: 'assistant', status: 'pending' }, + { id: 'pending-1', sessionId: 's1', orderSeq: 1, role: 'assistant', status: 'pending' }, + { id: 'sent', sessionId: 's1', orderSeq: 3, role: 'assistant', status: 'sent' }, + { id: 'user', sessionId: 's1', orderSeq: 4, role: 'user', status: 'pending' }, + { id: 'other', sessionId: 's2', orderSeq: 1, role: 'assistant', status: 'pending' } + ] as const) { + table.insert({ ...row, content: '[]' }) + } + + expect(table.getPendingAssistantBySession('s1').map((row) => row.id)).toEqual([ + 'pending-1', + 'pending-2' + ]) + } finally { + db.close() + } + }) + + it('installs the pending assistant index for an existing message table', () => { + const { db, table } = createTable() + try { + db.exec('DROP INDEX idx_deepchat_messages_pending_assistant') + + table.createTable() + + expect( + db + .prepare( + `SELECT name FROM sqlite_master + WHERE type = 'index' AND name = 'idx_deepchat_messages_pending_assistant'` + ) + .get() + ).toEqual({ name: 'idx_deepchat_messages_pending_assistant' }) + } finally { + db.close() + } + }) + it('projects only assistant identity and result text for delegated result reads', () => { const { db, table } = createTable() try { @@ -266,7 +346,7 @@ describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { }) it('uses the existing session index', () => { - const { db } = createTable() + const { db, table } = createTable() try { const plan = db .prepare( @@ -279,6 +359,18 @@ describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { expect(plan.some((row) => /deepchat_message_traces|materialize/i.test(row.detail))).toBe( false ) + + const pendingPlan = db + .prepare( + `EXPLAIN QUERY PLAN SELECT * FROM deepchat_messages + WHERE session_id = ? AND role = 'assistant' AND status = 'pending' + ORDER BY order_seq, id` + ) + .all('s1') as Array<{ detail: string }> + expect( + pendingPlan.some((row) => /idx_deepchat_messages_pending_assistant/i.test(row.detail)) + ).toBe(true) + expect(table.getPendingAssistantBySession('s1')).toEqual([]) } finally { db.close() } diff --git a/test/main/session/data/transcript.test.ts b/test/main/session/data/transcript.test.ts index 961b96e23..581d7040d 100644 --- a/test/main/session/data/transcript.test.ts +++ b/test/main/session/data/transcript.test.ts @@ -25,6 +25,7 @@ function createMockSqlitePresenter() { updateMetadata: vi.fn(), updateContentAndStatus: vi.fn(), getBySession: vi.fn().mockReturnValue([]), + getPendingAssistantBySession: vi.fn().mockReturnValue([]), hasBySession: vi.fn().mockReturnValue(false), getByStatus: vi.fn().mockReturnValue([]), getIdsBySession: vi.fn().mockReturnValue([]), @@ -734,6 +735,27 @@ describe('SessionTranscript', () => { }) }) + describe('getPendingAssistantMessages', () => { + it('materializes only the pending assistant rows selected by the table', () => { + sqlitePresenter.deepchatMessagesTable.getPendingAssistantBySession.mockReturnValue([ + createMessageRow({ role: 'assistant', status: 'pending', content: '[]' }) + ]) + + expect(store.getPendingAssistantMessages('s1')).toEqual([ + expect.objectContaining({ + id: 'm1', + sessionId: 's1', + role: 'assistant', + status: 'pending', + content: '[]' + }) + ]) + expect( + sqlitePresenter.deepchatMessagesTable.getPendingAssistantBySession + ).toHaveBeenCalledWith('s1') + }) + }) + describe('hasMessages', () => { it('uses the table existence query without loading message rows', () => { sqlitePresenter.deepchatMessagesTable.hasBySession.mockReturnValue(true) From e65da8c1cde636a134c815d60ae0ac25036f7174 Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Tue, 11 Aug 2026 10:38:17 +0800 Subject: [PATCH 3/3] perf(session): avoid startup index build --- .../session/data/tables/deepchatMessages.ts | 12 ------ .../data/tables/deepchatMessagesTable.test.ts | 38 ++----------------- 2 files changed, 4 insertions(+), 46 deletions(-) diff --git a/src/main/session/data/tables/deepchatMessages.ts b/src/main/session/data/tables/deepchatMessages.ts index 41bf281a8..2016ef9fc 100644 --- a/src/main/session/data/tables/deepchatMessages.ts +++ b/src/main/session/data/tables/deepchatMessages.ts @@ -32,22 +32,11 @@ export interface DeepChatAssistantMessageIdentityRow { updated_at: number } -const PENDING_ASSISTANT_INDEX_SQL = ` - CREATE INDEX IF NOT EXISTS idx_deepchat_messages_pending_assistant - ON deepchat_messages(session_id, order_seq, id) - WHERE role = 'assistant' AND status = 'pending'; -` - export class DeepChatMessagesTable extends BaseTable { constructor(db: Database.Database) { super(db, 'deepchat_messages') } - override createTable(): void { - super.createTable() - this.db.exec(PENDING_ASSISTANT_INDEX_SQL) - } - getCreateTableSQL(): string { return ` CREATE TABLE IF NOT EXISTS deepchat_messages ( @@ -63,7 +52,6 @@ export class DeepChatMessagesTable extends BaseTable { updated_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_deepchat_messages_session ON deepchat_messages(session_id, order_seq); - ${PENDING_ASSISTANT_INDEX_SQL} ` } diff --git a/test/main/session/data/tables/deepchatMessagesTable.test.ts b/test/main/session/data/tables/deepchatMessagesTable.test.ts index 5fd831842..6e3653d51 100644 --- a/test/main/session/data/tables/deepchatMessagesTable.test.ts +++ b/test/main/session/data/tables/deepchatMessagesTable.test.ts @@ -208,26 +208,6 @@ describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { } }) - it('installs the pending assistant index for an existing message table', () => { - const { db, table } = createTable() - try { - db.exec('DROP INDEX idx_deepchat_messages_pending_assistant') - - table.createTable() - - expect( - db - .prepare( - `SELECT name FROM sqlite_master - WHERE type = 'index' AND name = 'idx_deepchat_messages_pending_assistant'` - ) - .get() - ).toEqual({ name: 'idx_deepchat_messages_pending_assistant' }) - } finally { - db.close() - } - }) - it('projects only assistant identity and result text for delegated result reads', () => { const { db, table } = createTable() try { @@ -346,11 +326,13 @@ describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { }) it('uses the existing session index', () => { - const { db, table } = createTable() + const { db } = createTable() try { const plan = db .prepare( - 'EXPLAIN QUERY PLAN SELECT * FROM deepchat_messages WHERE session_id = ? ORDER BY order_seq' + `EXPLAIN QUERY PLAN SELECT * FROM deepchat_messages + WHERE session_id = ? AND role = 'assistant' AND status = 'pending' + ORDER BY order_seq, id` ) .all('s1') as Array<{ detail: string }> @@ -359,18 +341,6 @@ describeIfNativeSqlite('DeepChatMessagesTable runtime projection', () => { expect(plan.some((row) => /deepchat_message_traces|materialize/i.test(row.detail))).toBe( false ) - - const pendingPlan = db - .prepare( - `EXPLAIN QUERY PLAN SELECT * FROM deepchat_messages - WHERE session_id = ? AND role = 'assistant' AND status = 'pending' - ORDER BY order_seq, id` - ) - .all('s1') as Array<{ detail: string }> - expect( - pendingPlan.some((row) => /idx_deepchat_messages_pending_assistant/i.test(row.detail)) - ).toBe(true) - expect(table.getPendingAssistantBySession('s1')).toEqual([]) } finally { db.close() }