diff --git a/packages/codingcode/src/agent/agent.ts b/packages/codingcode/src/agent/agent.ts index 4c1c700..fdb71d6 100644 --- a/packages/codingcode/src/agent/agent.ts +++ b/packages/codingcode/src/agent/agent.ts @@ -1,12 +1,13 @@ -import { Effect, Either, Queue, Stream, Fiber, Layer } from 'effect'; +import { Effect, Either, Stream, Fiber, Layer } from 'effect'; import { AgentError } from '../util/error.js'; import { Result } from '../util/result.js'; import { AgentService, ToolEnvPort } from './port.js'; -import type { RunTurnOptions, ToolEnv } from './port.js'; +import type { RunTurnOptions } from './port.js'; import { ApprovalService, getToolNames } from '../approval/port.js'; import { CheckpointService } from '../checkpoint/port.js'; import { ContextService } from '../context/port.js'; import { EventSinkService } from '../sink/port.js'; +import { TurnRegistryService } from '../turn/port.js'; import { MailboxService } from '../session/mailbox.js'; import { HookService } from '../hooks/port.js'; import { LLMService } from '../llm/port.js'; @@ -19,7 +20,6 @@ import { TodoService } from '../todo/port.js'; import { ToolExecutorService } from '../tools/port.js'; import { buildSystemPrompt, renderSkillBlock } from './prompt.js'; import type { - EndTransition, FrameBody, FrameError, ResponseMeta, @@ -92,6 +92,7 @@ export const AgentLayer = Layer.effect( const mcp = yield* McpService; const context = yield* ContextService; const sink = yield* EventSinkService; + const turn = yield* TurnRegistryService; const mailbox = yield* MailboxService; const memory = yield* MemoryService; const llm = yield* LLMService; @@ -117,6 +118,18 @@ export const AgentLayer = Layer.effect( Effect.gen(function* () { const normalizedCwd = normalizePath(opts.cwd); + // 提交判定：命中活跃回合 → 输入进状态表，直接返回，不开回合 + if (opts.sessionId) { + const verdict = yield* turn.submit(opts.sessionId, { id: opts.inputId!, parts: input }); + if (verdict.kind === 'attached') { + return { + kind: 'queued' as const, + sessionId: opts.sessionId, + turnId: verdict.turnId, + }; + } + } + yield* rules.evictProjectRules(normalizedCwd); yield* hooks.reloadUserHooks(normalizedCwd); yield* hooks.emit('agent.turn.start', { sessionId: '', projectPath: normalizedCwd }); @@ -149,129 +162,128 @@ export const AgentLayer = Layer.effect( } const state = yield* session.load(normalizedCwd, sessionId, parentSessionId); + const sid: string = sessionId; + const turnId = state.currentTurnId + 1; - const profileName = opts.activeProfile ?? state.activeProfile; - const effectivePerm = opts.permissionMode ?? state.permissionMode; - if (opts.permissionMode) { - yield* session.setPermissionMode( - normalizedCwd, - sessionId, - opts.permissionMode, - parentSessionId - ); - } - if (opts.activeProfile) { - yield* session.setActiveProfile( - normalizedCwd, - sessionId, - opts.activeProfile, - parentSessionId - ); + // 先于任何落盘：409 的 loser 不写 transcript、不建 checkpoint + if ( + !(yield* turn.claim(sid, { + turnId, + parentSessionId: state.parentSessionId, + agentName: state.agentName, + })) + ) { + return yield* Effect.fail(new AgentError('TURN_CONFLICT', 'concurrent turn on session')); } - state.memorySnapshot = yield* memory.loadMemoryForPrompt(state.cwd); + return yield* Effect.gen(function* () { + const profileName = opts.activeProfile ?? state.activeProfile; + const effectivePerm = opts.permissionMode ?? state.permissionMode; + if (opts.permissionMode) { + yield* session.setPermissionMode(normalizedCwd, sid, opts.permissionMode, parentSessionId); + } + if (opts.activeProfile) { + yield* session.setActiveProfile(normalizedCwd, sid, opts.activeProfile, parentSessionId); + } - const profile: AgentProfile | undefined = profileName - ? resolveProfile(profileName) - : undefined; + state.memorySnapshot = yield* memory.loadMemoryForPrompt(state.cwd); - const mcpTools = yield* mcp.listProjectMcpTools(normalizedCwd); - const catalog = yield* executor.prepare(getToolNames(profileName), mcpTools); + const profile: AgentProfile | undefined = profileName + ? resolveProfile(profileName) + : undefined; - const toolEnv = yield* toolEnvPort.getToolEnv(); + const mcpTools = yield* mcp.listProjectMcpTools(normalizedCwd); + const catalog = yield* executor.prepare(getToolNames(profileName), mcpTools); - yield* assertMediaAllowed(input, model); - const parts = yield* session.materializeInput(state, input); - const turnId = (yield* session.recordUser(state, parts)).turnId; + const toolEnv = yield* toolEnvPort.getToolEnv(); - // 用户显式 @ 的 skill：按 path 回查权威数据，正文拼块后作为同回合的第二条 user 事件 - if (opts.skills?.length) { - const all = yield* skills.getAll(state.cwd); - const chosen = all.filter((s) => opts.skills!.some((m) => m.path === s.skillPath)); - if (chosen.length) { - const entries = yield* Effect.forEach(chosen, (s) => - skills.readContent(s.skillPath).pipe(Effect.map((body) => ({ skill: s, body }))) - ); - yield* session.recordSystem(state, [textPart(renderSkillBlock(entries))]); + yield* assertMediaAllowed(input, model); + const parts = yield* session.materializeInput(state, input); + yield* session.recordUser(state, parts); + + // 用户显式 @ 的 skill：按 path 回查权威数据，正文拼块后作为同回合的第二条 user 事件 + if (opts.skills?.length) { + const all = yield* skills.getAll(state.cwd); + const chosen = all.filter((s) => opts.skills!.some((m) => m.path === s.skillPath)); + if (chosen.length) { + const entries = yield* Effect.forEach(chosen, (s) => + skills.readContent(s.skillPath).pipe(Effect.map((body) => ({ skill: s, body }))) + ); + yield* session.recordSystem(state, [textPart(renderSkillBlock(entries))]); + } } - } - // checkpoint baseline - yield* checkpoint.snapshotBaseline(state.cwd, sessionId, turnId); + // checkpoint baseline + yield* checkpoint.snapshotBaseline(state.cwd, sid, turnId); - // get rules text - const rulesText = yield* rules.getAllRules(state.cwd); + const rulesText = yield* rules.getAllRules(state.cwd); - // run agent loop：出站队列挂在 sink 上，本回合是它的唯一读者 - const q = yield* sink.attach(sessionId); - const emit = (body: FrameBody) => Effect.runSync(sink.emit(sessionId, body)); - const stream = runAgentLoop( - { - state, - model, - profile, - catalog, - systemPrompt: opts.systemPrompt, - toolEnv, - abortSignal: opts.signal, - rulesText, - sid: sessionId, - projectPath: state.cwd, - permissionMode: effectivePerm, - }, - { q, emit, onEnd: () => Effect.runSync(sink.detach(sessionId)) } - ); + // 出站队列挂在 sink 上，本回合是它的唯一读者 + const q = yield* sink.attach(sid); + const emit = (body: FrameBody) => Effect.runSync(sink.emit(sid, body)); - return { stream, sessionId }; - }); + // 出生帧须在 attach 之后 + yield* turn.transition(sid, { kind: 'start' }); - function runAgentLoop( - opts: { - state: any; - model: string; - profile: AgentProfile | undefined; - abortSignal: AbortSignal | undefined; - catalog: ToolCatalog; - toolEnv: ToolEnv; - rulesText: string; - sid: string; - projectPath: string; - permissionMode: PermissionMode; - systemPrompt?: string; - }, - out: { - q: Queue.Queue; - emit: (body: FrameBody) => void; - onEnd: () => void; - } - ): AsyncGenerator { - const program = agentLoopInternal(opts, out.emit); - - return (async function* () { - const fiber = Effect.runFork(opts.toolEnv.provide(program)); - if (opts.abortSignal) { - opts.abortSignal.addEventListener( - 'abort', - () => { - Effect.runFork(Fiber.interrupt(fiber)); + // 回合出生：eager 段的最后一句 + const program = agentLoopInternal( + { + state, + model, + profile, + catalog, + rulesText, + abortSignal: opts.signal, + sid, + projectPath: state.cwd, + permissionMode: effectivePerm, + systemPrompt: opts.systemPrompt, }, - { once: true } + emit ); - if (opts.abortSignal.aborted) Effect.runFork(Fiber.interrupt(fiber)); - } - - try { - const stream = Stream.fromQueue(out.q).pipe( - Stream.takeUntil((body: FrameBody) => isTurnEnd(body)) - ); - for await (const body of Stream.toAsyncIterable(stream) as AsyncIterable) { - yield body; + const fiber = Effect.runFork(toolEnv.provide(program)); + if (opts.signal) { + opts.signal.addEventListener('abort', () => Effect.runFork(Fiber.interrupt(fiber)), { + once: true, + }); + if (opts.signal.aborted) Effect.runFork(Fiber.interrupt(fiber)); } - } finally { - out.onEnd(); - } - })(); - } + yield* turn.arm(sid, () => Effect.runFork(Fiber.interrupt(fiber))); + + // 搬帧器：只搬帧，不负责启动回合 + const stream = (async function* () { + try { + for await (const body of Stream.toAsyncIterable( + Stream.fromQueue(q).pipe(Stream.takeUntil((b: FrameBody) => isTurnEnd(b))) + ) as AsyncIterable) { + yield body; + } + } finally { + Effect.runSync(sink.detach(sid)); + } + })(); + + return { kind: 'turn' as const, stream, sessionId: sid }; + }).pipe( + // 出生前失败 → 落 error 终态（settle 幂等） + Effect.onError((cause) => + turn + .transition(sid, { + kind: 'fail', + error: toFrameError(cause as unknown as AgentError), + }) + .pipe(Effect.ignore) + ), + Effect.onInterrupt(() => + turn + .transition(sid, { + kind: 'fail', + error: { message: 'turn aborted before start', code: 'AGENT_TERMINATED' }, + }) + .pipe(Effect.ignore) + ) + ); + }); function agentLoopInternal( opts: { @@ -301,15 +313,6 @@ export const AgentLayer = Layer.effect( } = opts; const { tools, lookup: toolLookup } = catalog; - let ended = false; - let deliveryPhase: 'currentTurn' | 'nextTurn' = 'currentTurn'; - const offerEnd = (transition: EndTransition) => - Effect.sync(() => { - if (ended) return; - ended = true; - emit({ family: 'transition', transition }); - }); - return Effect.gen(function* () { const basePrompt = buildSystemPrompt({ cwd: projectPath, @@ -328,8 +331,33 @@ export const AgentLayer = Layer.effect( let lastResult: Result | null = null; + const sessionRef: SessionRef = { + cwd: state.cwd, + sessionId: state.sessionId, + parentSessionId: state.parentSessionId, + currentTurnId: state.currentTurnId, + }; + + /** 吸收本回合待投递的 steer 输入；返回条数 */ + const absorbPendingInputs = () => + Effect.gen(function* () { + let n = 0; + // steer 输入先于子代理结果，否则"用户指令"会排在"子代理汇报"之后 + for (const ui of yield* turn.drain(sid)) { + // 与首输入同一条准入：不合法即抛错，整批输入随该 step 失败 + yield* assertMediaAllowed(ui.parts, model); + const stored = yield* session.materializeInput(state, ui.parts); + + const ev = yield* session.recordUserInput(state, stored); + yield* context.absorb(sessionRef, [ev]); + emit({ family: 'event', event: { type: 'user_input', id: ui.id, parts: stored } }); + n++; + } + return n; + }); + yield* hooks.emit('agent.turn.start', { sessionId: sid, projectPath }); - emit({ family: 'transition', transition: { to: 'start', turnId: state.currentTurnId } }); + // 出生帧已在 eager 段投过，这里不再重复 for (let step = 0; step < maxSteps; step++) { yield* hooks.emitDecision('agent.step.before', { @@ -339,19 +367,12 @@ export const AgentLayer = Layer.effect( }); if (step === 0) { - emit({ family: 'transition', transition: { to: 'executing' } }); + yield* turn.transition(sid, { kind: 'running' }); } - const sessionRef: SessionRef = { - cwd: state.cwd, - sessionId: state.sessionId, - parentSessionId: state.parentSessionId, - currentTurnId: state.currentTurnId, - }; + let drainedCount = yield* absorbPendingInputs(); - const mayDrain = deliveryPhase === 'currentTurn' && step > 0; - let drainedCount = 0; - if (mayDrain) { + if (step > 0) { for (const item of yield* mailbox.drain(state.sessionId)) { const ev = yield* session.recordSubagentResult(state, item); yield* context.absorb(sessionRef, [ev]); @@ -361,7 +382,10 @@ export const AgentLayer = Layer.effect( const history = yield* Effect.either(context.getHistory(sessionRef, model)); if (Either.isLeft(history)) { - yield* offerEnd({ to: 'end', reason: 'error', error: toFrameError(history.left) }); + yield* turn.transition(sid, { + kind: 'fail', + error: toFrameError(history.left), + }); yield* hooks.emit('agent.turn.end', { sessionId: sid, turnId: state.currentTurnId, @@ -407,10 +431,6 @@ export const AgentLayer = Layer.effect( }); } else { responded = part.usage ? { usage: part.usage } : {}; - emit({ - family: 'transition', - transition: { to: 'executing', responded }, - }); } } }, @@ -418,7 +438,7 @@ export const AgentLayer = Layer.effect( }) ); if (Either.isLeft(streamed)) { - yield* offerEnd({ to: 'end', reason: 'error', error: toFrameError(streamed.left) }); + yield* turn.transition(sid, { kind: 'fail', error: toFrameError(streamed.left) }); yield* hooks.emit('agent.turn.end', { sessionId: sid, turnId: state.currentTurnId, @@ -428,6 +448,9 @@ export const AgentLayer = Layer.effect( return Result.err(streamed.left); } + // 流结束帧（携带 usage）：在流外统一投递 + yield* turn.transition(sid, { kind: 'running', responded }); + if (toolCalls.length === 0) { if (content.trim() === '' && !abortSignal?.aborted) { const detail = @@ -436,7 +459,7 @@ export const AgentLayer = Layer.effect( 'EMPTY_RESPONSE', `model returned an empty response${detail}` ); - yield* offerEnd({ to: 'end', reason: 'error', error: toFrameError(emptyErr) }); + yield* turn.transition(sid, { kind: 'fail', error: toFrameError(emptyErr) }); yield* hooks.emit('agent.turn.end', { sessionId: sid, turnId: state.currentTurnId, @@ -449,7 +472,6 @@ export const AgentLayer = Layer.effect( const assistantEv = yield* session.recordAssistant(state, content, [], responded.usage); yield* context.absorb(sessionRef, [assistantEv]); - deliveryPhase = 'nextTurn'; const stopDecision = yield* hooks.emitDecision('agent.turn.stop', { sessionId: sid, content, @@ -463,7 +485,7 @@ export const AgentLayer = Layer.effect( 'AGENT_LOOP_DETECTED', 'max stop continuations exceeded' ); - yield* offerEnd({ to: 'end', reason: 'error', error: toFrameError(loopErr) }); + yield* turn.transition(sid, { kind: 'fail', error: toFrameError(loopErr) }); yield* hooks.emit('agent.turn.end', { sessionId: sid, turnId: state.currentTurnId, @@ -480,7 +502,10 @@ export const AgentLayer = Layer.effect( continue; } - yield* offerEnd({ to: 'end', reason: 'done' }); + // 终稿后有 steer 输入 → 吸收进上下文并续跑，turnId 不变、不重发 start 帧 + if ((yield* absorbPendingInputs()) > 0) continue; + + yield* turn.transition(sid, { kind: 'complete', reason: 'done' }); lastResult = Result.ok(content); yield* hooks.emit('agent.turn.end', { sessionId: sid, @@ -574,7 +599,7 @@ export const AgentLayer = Layer.effect( if (lastResult) return lastResult; const maxErr = AgentError.maxStepsReached(maxSteps); - yield* offerEnd({ to: 'end', reason: 'maxSteps' }); + yield* turn.transition(sid, { kind: 'complete', reason: 'maxSteps' }); yield* hooks.emit('agent.turn.end', { sessionId: sid, turnId: state.currentTurnId, @@ -586,28 +611,29 @@ export const AgentLayer = Layer.effect( Effect.interruptible, Effect.onInterrupt(() => Effect.gen(function* () { - yield* offerEnd({ to: 'end', reason: 'aborted' }); + yield* turn.transition(sid, { kind: 'interrupt' }); yield* hooks .emit('agent.turn.end', { - sessionId: opts.sid, - turnId: opts.state.currentTurnId, + sessionId: sid, + turnId: state.currentTurnId, status: 'aborted', - projectPath: opts.projectPath, + projectPath, }) .pipe(Effect.ignore); }) ), Effect.ensuring( Effect.gen(function* () { - yield* offerEnd({ - to: 'end', - reason: 'error', - error: { message: 'agent terminated without end frame', code: 'AGENT_TERMINATED' }, - }); + yield* turn + .transition(sid, { + kind: 'fail', + error: { message: 'agent terminated without end frame', code: 'AGENT_TERMINATED' }, + }) + .pipe(Effect.ignore); yield* checkpoint - .snapshotFinal(opts.projectPath, opts.sid, opts.state.currentTurnId) + .snapshotFinal(state.cwd, sid, state.currentTurnId) .pipe(Effect.ignore); - yield* flushMemoryInBackground(opts.state.sessionId, opts.model, opts.projectPath); + yield* flushMemoryInBackground(state.sessionId, model, state.cwd); }) ) ); diff --git a/packages/codingcode/src/agent/port.ts b/packages/codingcode/src/agent/port.ts index 924a36c..e498ba9 100644 --- a/packages/codingcode/src/agent/port.ts +++ b/packages/codingcode/src/agent/port.ts @@ -17,19 +17,16 @@ export interface RunTurnOptions { agentName?: string; systemPrompt?: string; skills?: ReadonlyArray<{ name: string; path: string }>; + /** steer 路径下由前端提供，作为 `PendingUserInput.id` 原样透传 */ + inputId?: string; } +export type RunTurnResult = + | { kind: 'turn'; stream: AsyncGenerator; sessionId: string } + | { kind: 'queued'; sessionId: string; turnId: number }; + export interface AgentShape { - runTurn( - input: IncomingPart[], - opts: RunTurnOptions - ): Effect.Effect< - { - stream: AsyncGenerator; - sessionId: string; - }, - AgentError - >; + runTurn(input: IncomingPart[], opts: RunTurnOptions): Effect.Effect; } export class AgentService extends Context.Tag('AgentService')() {} diff --git a/packages/codingcode/src/context/context.ts b/packages/codingcode/src/context/context.ts index 02b9b97..caedd1f 100644 --- a/packages/codingcode/src/context/context.ts +++ b/packages/codingcode/src/context/context.ts @@ -21,7 +21,7 @@ import { COMPACTION_SYSTEM_PROMPT } from './compaction-prompt.js'; import { AgentError } from '../util/error.js'; import { ContextService } from './port.js'; import type { CompressResult } from './port.js'; -import { EventSinkService } from '../sink/port.js'; +import { TurnRegistryService } from '../turn/port.js'; export function transcriptPathFor(ref: SessionRef): string { return transcriptPathOf(ref.cwd, ref.sessionId, ref.parentSessionId); @@ -211,7 +211,7 @@ export const ContextLayer = Layer.effect( Effect.gen(function* () { const session = yield* SessionService; const llm = yield* LLMService; - const sink = yield* EventSinkService; + const turn = yield* TurnRegistryService; // 回合内内存态：键 = sessionId，仅在换回合（turnId 变化）时重建 const buffers = new Map(); @@ -429,13 +429,10 @@ export const ContextLayer = Layer.effect( const contextWindow = contextWindowOf(model); yield* runMicroCompact(buf, contextWindow); if (needsCompaction(buf, contextWindow)) { - // 压缩判定与压缩帧都归 context，agent 不参与 - yield* sink.emit(ref.sessionId, { family: 'transition', transition: { to: 'compress' } }); + // 压缩判定归 context，压缩相位帧由状态表投递 + yield* turn.transition(ref.sessionId, { kind: 'compressing' }); yield* summarizeToFit(buf, contextWindow, model); - yield* sink.emit(ref.sessionId, { - family: 'transition', - transition: { to: 'executing' }, - }); + yield* turn.transition(ref.sessionId, { kind: 'running' }); } return yield* resolveMediaParts(buildContextMessages(buf.events, buf.compactedTurnIds), ref.cwd); }); diff --git a/packages/codingcode/src/layer.ts b/packages/codingcode/src/layer.ts index 2e07205..08657a6 100644 --- a/packages/codingcode/src/layer.ts +++ b/packages/codingcode/src/layer.ts @@ -17,6 +17,7 @@ import { AgentLayer } from './agent/agent.js'; import { ToolEnvLayer } from './agent/tool-env.js'; import { SubagentRunnerLayer } from './subagent/subagent.js'; import { SubagentRunRegistryLayer } from './subagent/registry.js'; +import { TurnRegistryLayer } from './turn/registry.js'; import { MailboxLayer } from './session/mailbox.js'; import { SchedulerLayer } from './scheduler/scheduler.js'; @@ -31,6 +32,9 @@ const InfraLayer = Layer.mergeAll( TodoLayer ); +// 状态帧由 turn 投递 ⇒ turn 要 sink； +const TurnWithDeps = TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)); + const ApprovalWithDeps = ApprovalLayer.pipe( Layer.provide(Layer.mergeAll(HookLayer, EventSinkLayer, ApprovalWaitLayer)) ); @@ -38,7 +42,7 @@ const ToolExecutorWithDeps = ToolExecutorLayer.pipe( Layer.provide(Layer.mergeAll(HookLayer, ApprovalWithDeps)) ); const ContextWithDeps = ContextLayer.pipe( - Layer.provide(Layer.mergeAll(SessionLayer, LlmLayer, EventSinkLayer)) + Layer.provide(Layer.mergeAll(SessionLayer, LlmLayer, EventSinkLayer, TurnWithDeps)) ); const MemoryWithDeps = MemoryLayer.pipe(Layer.provide(LlmLayer)); @@ -52,7 +56,8 @@ const AgentServiceLayers = Layer.mergeAll( ContextWithDeps, MemoryWithDeps, CheckpointLayer, - LlmLayer + LlmLayer, + TurnWithDeps ); // agent with deps @@ -65,8 +70,9 @@ const SubagentWithDeps = SubagentRunnerLayer.pipe(Layer.provide(AgentWithDeps)); // 运行注册表：要 runner 起子代理、要 mailbox 投递终态、要 sink 发 subagent_event 帧、 // 要 hooks 在子代理终态时触发 agent.subagent.complete。 +// 终态归属已并入状态表（turn 已在 AgentWithDeps 链路里，Layer 按引用去重 ⇒ 全进程一份 records）。 const SubagentRunRegistryWithDeps = SubagentRunRegistryLayer.pipe( - Layer.provide(Layer.mergeAll(SubagentWithDeps, MailboxLayer, EventSinkLayer, HookLayer)) + Layer.provide(Layer.mergeAll(SubagentWithDeps, MailboxLayer, EventSinkLayer, HookLayer, TurnWithDeps)) ); export const AppLayer = Layer.mergeAll( @@ -82,6 +88,7 @@ export const AppLayer = Layer.mergeAll( AgentWithDeps, SubagentWithDeps, SubagentRunRegistryWithDeps, + TurnWithDeps, SchedulerLayer, EventSinkLayer ); diff --git a/packages/codingcode/src/scheduler/scheduler.ts b/packages/codingcode/src/scheduler/scheduler.ts index 15dd6b3..3c4e42d 100644 --- a/packages/codingcode/src/scheduler/scheduler.ts +++ b/packages/codingcode/src/scheduler/scheduler.ts @@ -48,7 +48,7 @@ export const SchedulerLayer = Layer.effect( const timeout = setTimeout(() => controller.abort(), TIMEOUT_MS); try { - const { stream, sessionId } = await _rt.runPromise( + const turnResult = await _rt.runPromise( Effect.gen(function* () { const agent = yield* AgentService; return yield* agent.runTurn([textPart(auto.description)], { @@ -61,6 +61,9 @@ export const SchedulerLayer = Layer.effect( }); }) ); + // 自动化不开已有会话，不可能命中活跃回合 + if (turnResult.kind === 'queued') return; + const { stream, sessionId } = turnResult; let lastContent = ''; for await (const body of stream) { @@ -181,7 +184,7 @@ export const SchedulerLayer = Layer.effect( const timeout = setTimeout(() => controller.abort(), TIMEOUT_MS); try { - const { stream, sessionId } = await _rt.runPromise( + const turnResult = await _rt.runPromise( Effect.gen(function* () { const agent = yield* AgentService; return yield* agent.runTurn([textPart(auto.description)], { @@ -193,6 +196,9 @@ export const SchedulerLayer = Layer.effect( }); }) ); + // 同上：自动化不开已有会话，queued 不可达 + if (turnResult.kind === 'queued') return null; + const { stream, sessionId } = turnResult; for await (const body of stream) { if ( diff --git a/packages/codingcode/src/server/http-error.ts b/packages/codingcode/src/server/http-error.ts index c6cbacf..6be2d8c 100644 --- a/packages/codingcode/src/server/http-error.ts +++ b/packages/codingcode/src/server/http-error.ts @@ -21,6 +21,7 @@ const DOMAIN_STATUS: Record = { EMPTY_RESPONSE: 500, SESSION_IO_ERROR: 500, INVALID_INPUT: 400, + TURN_CONFLICT: 409, }; export class NotFoundError extends Error { diff --git a/packages/codingcode/src/server/routes/messages.ts b/packages/codingcode/src/server/routes/messages.ts index c708611..7560f79 100644 --- a/packages/codingcode/src/server/routes/messages.ts +++ b/packages/codingcode/src/server/routes/messages.ts @@ -10,6 +10,7 @@ import { ASK_BEFORE_EXEC_PERMISSION_MODE, BUILD_PROFILE_NAME } from '../../util/ import type { IncomingMedia, IncomingPart } from '../../llm/types.js'; import { frameStream, + json, pathParams, readJsonFrom, sseResponse, @@ -28,6 +29,8 @@ type MessageBody = { input: WirePart[]; cwd: string; model?: string; + /** 前端队列项 id：命中活跃回合时原样回显在 user_input 帧上 */ + inputId?: string; skills?: Array<{ name: string; path: string }>; }; @@ -94,6 +97,7 @@ const sendMessage: Handler = Effect.gen(function* () { cwd, signal: web.signal, model: body.model, + inputId: body.inputId, skills: body.skills, }; if (isNew) { @@ -107,12 +111,16 @@ const sendMessage: Handler = Effect.gen(function* () { } const agent = yield* AgentService; - const { stream, sessionId: actualSid } = yield* agent.runTurn(input, { + const result = yield* agent.runTurn(input, { sessionId: isNew ? undefined : sessionId, ...runOpts, } as never); - return yield* sseResponse(frameStream(stream), { sessionId: actualSid }); + // 命中活跃回合：输入已进状态表，不开新流，回 202 + if (result.kind === 'queued') { + return json({ queued: true, sessionId: result.sessionId, turnId: result.turnId }, 202); + } + return yield* sseResponse(frameStream(result.stream), { sessionId: result.sessionId }); }); export const addMessagesRoutes = (router: Router): Router => diff --git a/packages/codingcode/src/server/routes/sessions.ts b/packages/codingcode/src/server/routes/sessions.ts index e76e199..2677517 100644 --- a/packages/codingcode/src/server/routes/sessions.ts +++ b/packages/codingcode/src/server/routes/sessions.ts @@ -13,6 +13,7 @@ import { SessionService } from '../../session/port.js'; import { computePaths } from '../../session/paths.js'; import { ContextService } from '../../context/port.js'; import { MailboxService } from '../../session/mailbox.js'; +import { TurnRegistryService } from '../../turn/port.js'; import { estimatePromptTokensFrom } from '../../context/context.js'; import { CheckpointService } from '../../checkpoint/port.js'; import { activeModelId, setGlobalActive } from '../../infra/models.js'; @@ -102,9 +103,14 @@ const deleteSession: Handler = Effect.gen(function* () { const session = yield* SessionService; const context = yield* ContextService; const mailbox = yield* MailboxService; - yield* session.deleteSession(id ?? '', cwd); - yield* context.dispose(id ?? ''); - yield* mailbox.dispose(id ?? ''); + const turn = yield* TurnRegistryService; + const sessionId = id ?? ''; + // 先请求停止：子代理纤维仍在跑的话，dispose 后它会成为孤儿（还会 mailbox.offer 重建队列） + yield* turn.stopChildren(sessionId); + yield* session.deleteSession(sessionId, cwd); + yield* context.dispose(sessionId); + yield* mailbox.dispose(sessionId); + yield* turn.dispose(sessionId); return json({ ok: true }); }); diff --git a/packages/codingcode/src/server/routes/subagents.ts b/packages/codingcode/src/server/routes/subagents.ts index 57ae5c3..187e392 100644 --- a/packages/codingcode/src/server/routes/subagents.ts +++ b/packages/codingcode/src/server/routes/subagents.ts @@ -1,12 +1,12 @@ import * as HttpRouter from '@effect/platform/HttpRouter'; import { Effect } from 'effect'; -import { SubagentRunRegistryService } from '../../subagent/registry.js'; +import { TurnRegistryService } from '../../turn/port.js'; import { json, pathParams, type Handler, type Router } from '../handler.js'; const stopAll: Handler = Effect.gen(function* () { const { id: sessionId } = yield* pathParams; - const registry = yield* SubagentRunRegistryService; - const stopped = yield* registry.stopAll(sessionId ?? ''); + const turn = yield* TurnRegistryService; + const stopped = yield* turn.stopChildren(sessionId ?? ''); return json({ stopped }); }); diff --git a/packages/codingcode/src/session/port.ts b/packages/codingcode/src/session/port.ts index b656d34..95f8117 100644 --- a/packages/codingcode/src/session/port.ts +++ b/packages/codingcode/src/session/port.ts @@ -48,6 +48,11 @@ export interface SessionShape { ): Effect.Effect, AgentError>; recordUser(state: SessionStoreState, content: StoredPart[]): Effect.Effect; + /** steer 输入：归属当前回合，不递增 turnId、不改标题 */ + recordUserInput( + state: SessionStoreState, + content: StoredPart[] + ): Effect.Effect; recordSystem( state: SessionStoreState, content: StoredPart[] diff --git a/packages/codingcode/src/session/session.ts b/packages/codingcode/src/session/session.ts index c6bef86..ee27a70 100644 --- a/packages/codingcode/src/session/session.ts +++ b/packages/codingcode/src/session/session.ts @@ -260,11 +260,12 @@ export const SessionLayer = Layer.effect( const recordUser = ( state: SessionStoreState, - content: StoredPart[] + content: StoredPart[], + opts?: { steering?: boolean } ): Effect.Effect => Effect.try({ try: () => { - state.currentTurnId += 1; + if (!opts?.steering) state.currentTurnId += 1; const event: UserEvent = { type: 'user', turnId: state.currentTurnId, @@ -288,6 +289,11 @@ export const SessionLayer = Layer.effect( : new AgentError('SESSION_IO_ERROR', `Session write failed: ${String(e)}`, e), }); + const recordUserInput = ( + state: SessionStoreState, + content: StoredPart[] + ): Effect.Effect => recordUser(state, content, { steering: true }); + const recordSystem = ( state: SessionStoreState, content: StoredPart[] @@ -601,6 +607,7 @@ export const SessionLayer = Layer.effect( materializeInput, resolveAssets, recordUser, + recordUserInput, recordSystem, recordAssistant, recordToolResult, diff --git a/packages/codingcode/src/sink/types.ts b/packages/codingcode/src/sink/types.ts index c14e4c0..7f45263 100644 --- a/packages/codingcode/src/sink/types.ts +++ b/packages/codingcode/src/sink/types.ts @@ -1,5 +1,6 @@ import type { TodoItem } from '../todo/types.js'; import type { TokenUsage } from '../llm/types.js'; +import type { StoredPart } from '../session/types.js'; export interface FrameError { readonly message: string; @@ -52,6 +53,11 @@ export type RuntimeEvent = readonly sessionId: string; readonly agentName: string; readonly status: 'spawned' | 'completed' | 'failed'; + } + | { + readonly type: 'user_input'; + readonly id: string; + readonly parts: readonly StoredPart[]; }; export interface Fatal { diff --git a/packages/codingcode/src/subagent/port.ts b/packages/codingcode/src/subagent/port.ts index b5eda2b..059bc77 100644 --- a/packages/codingcode/src/subagent/port.ts +++ b/packages/codingcode/src/subagent/port.ts @@ -9,7 +9,6 @@ import type { IncomingPart } from '../llm/types.js'; export interface RunSubagentOptions { sessionId?: string; cwd: string; - signal?: AbortSignal; activeProfile?: ProfileName; permissionMode?: PermissionMode; model: string; diff --git a/packages/codingcode/src/subagent/registry.ts b/packages/codingcode/src/subagent/registry.ts index 864b501..4263bfe 100644 --- a/packages/codingcode/src/subagent/registry.ts +++ b/packages/codingcode/src/subagent/registry.ts @@ -1,4 +1,4 @@ -import { Context, Effect, Fiber, Layer, Option, Stream, SubscriptionRef } from 'effect'; +import { Context, Effect, Layer } from 'effect'; import { AgentError } from '../util/error.js'; import { isTurnEnd } from '../sink/types.js'; import type { EndTransition, FrameBody } from '../sink/types.js'; @@ -11,19 +11,9 @@ import { loadConfig } from '../infra/config.js'; import { MailboxService } from '../session/mailbox.js'; import { EventSinkService } from '../sink/port.js'; import { HookService } from '../hooks/port.js'; +import { TurnRegistryService } from '../turn/port.js'; import { SubagentRunnerService } from './port.js'; -export type SubagentRunStatus = - | { readonly kind: 'running' } - | { readonly kind: 'ended'; readonly end: EndTransition }; - -/** wait 只回信号、不回内容 —— 内容已由父回合 drain 时写进父 transcript */ -export type WaitOutcome = 'completed' | 'failed' | 'timeout'; - -export const SUBAGENT_WAIT_MIN_MS = 10_000; -export const SUBAGENT_WAIT_DEFAULT_MS = 30_000; -export const SUBAGENT_WAIT_MAX_MS = 3_600_000; - export interface SpawnOptions { prompt: string; agentName: string; @@ -35,10 +25,9 @@ export interface SpawnOptions { systemPrompt?: string; } +/** 终态由子会话自己的 finish 写入；本表只负责派发与投递 */ export interface SubagentRunRegistryShape { spawn(opts: SpawnOptions): Effect.Effect<{ sessionId: string; agentName: string }, AgentError>; - wait(sessionId: string, timeoutMs: number): Effect.Effect; - stopAll(parentSessionId: string): Effect.Effect; } export class SubagentRunRegistryService extends Context.Tag('SubagentRunRegistry')< @@ -51,9 +40,6 @@ interface SubagentRun { readonly parentSessionId: string; readonly parentCwd: string; readonly agentName: string; - readonly status: SubscriptionRef.SubscriptionRef; - readonly abort: AbortController; - fiber?: Fiber.Fiber; } const BODY_TOKEN_BUDGET = 900; @@ -128,7 +114,10 @@ export const SubagentRunRegistryLayer = Layer.scoped( const runner = yield* SubagentRunnerService; const sink = yield* EventSinkService; const hooks = yield* HookService; - const runs = new Map(); // 键 = 子会话 sessionId；parentSessionId 只是条目上的字段 + const turn = yield* TurnRegistryService; + + /** 曾派发过子会话的父会话 id，仅用于 layer 终结时请求停止 */ + const parents = new Set(); /** 帧直投父会话的出站队列：EventSink 的键就是收件人会话，不需要任何回调透传 */ const emitSubagent = ( @@ -142,18 +131,6 @@ export const SubagentRunRegistryLayer = Layer.scoped( event: { type: 'subagent_event', sessionId, agentName, status }, }); - /** 同步读当前值：SubscriptionRef 的 get 随时可读、读不走 */ - const currentStatus = (run: SubagentRun): SubagentRunStatus => - Effect.runSync(SubscriptionRef.get(run.status)); - - const countRunning = (parentSessionId: string): number => { - let n = 0; - for (const run of runs.values()) { - if (run.parentSessionId === parentSessionId && currentStatus(run).kind === 'running') n++; - } - return n; - }; - // 只入队，不写盘：写盘由父回合在自己的 drain 点做（父回合循环手里才有 state） const drainRun = (run: SubagentRun, stream: AsyncGenerator) => Effect.gen(function* () { @@ -172,7 +149,9 @@ export const SubagentRunRegistryLayer = Layer.scoped( }) .pipe(Effect.ignore); - yield* SubscriptionRef.set(run.status, { kind: 'ended', end: outcome.end }); + // 终态由子会话自己的 finish 写入（turn.transition）；这里只补投递信号， + // 让 wait 在"结果已进父会话 mailbox"之后才返回 + yield* turn.markDelivered(run.sessionId); yield* emitSubagent( run.parentSessionId, run.sessionId, @@ -189,16 +168,15 @@ export const SubagentRunRegistryLayer = Layer.scoped( const spawn = (opts: SpawnOptions) => Effect.gen(function* () { - if (countRunning(opts.parentSessionId) >= loadConfig().subagent.maxBackground) { + if ((yield* turn.runningChildren(opts.parentSessionId)) >= loadConfig().subagent.maxBackground) { return yield* Effect.fail( new AgentError('TOOL_EXECUTION_FAILED', 'Concurrent subagent limit reached') ); } - const abort = new AbortController(); + // 子会话的可停性由 runTurn 的 arm 登记的 Fiber.interrupt 承担 const { stream, sessionId } = yield* runner.runSubagent([textPart(opts.prompt)], { cwd: opts.parentCwd, - signal: abort.signal, // 子代理自己的 signal，与父回合无关 activeProfile: opts.parentProfile, permissionMode: BYPASS_PERMISSION_MODE, parentSessionId: opts.parentSessionId, @@ -207,71 +185,27 @@ export const SubagentRunRegistryLayer = Layer.scoped( systemPrompt: opts.systemPrompt, }); - const status = yield* SubscriptionRef.make({ kind: 'running' }); const run: SubagentRun = { sessionId, parentSessionId: opts.parentSessionId, parentCwd: opts.parentCwd, agentName: opts.agentName, - status, - abort, }; - runs.set(sessionId, run); + parents.add(opts.parentSessionId); yield* emitSubagent(opts.parentSessionId, sessionId, opts.agentName, 'spawned'); - run.fiber = yield* Effect.forkDaemon(drainRun(run, stream)); + yield* Effect.forkDaemon(drainRun(run, stream)); return { sessionId, agentName: opts.agentName }; }); - const wait = (sessionId: string, timeoutMs: number): Effect.Effect => - Effect.gen(function* () { - const run = runs.get(sessionId); - if (!run) { - return yield* Effect.fail( - new AgentError('TOOL_EXECUTION_FAILED', `Unknown subagent: ${sessionId}`) - ); - } - // 等状态通道的后续值；changes 的首次发射即当前值 ⇒ 已终态的 run 立即返回 - const settled = run.status.changes.pipe( - Stream.filterMap((s) => - s.kind === 'ended' ? Option.some(s.end) : Option.none() - ), - Stream.runHead, - Effect.map( - (opt): WaitOutcome => - Option.isNone(opt) ? 'timeout' : opt.value.reason === 'done' ? 'completed' : 'failed' - ) - ); - - return yield* Effect.race( - Effect.sleep(timeoutMs).pipe(Effect.as('timeout')), - settled - ); - }); - - const stopAll = (parentSessionId: string): Effect.Effect => - Effect.sync(() => { - let stopped = 0; - for (const run of runs.values()) { - if (run.parentSessionId !== parentSessionId) continue; - if (run.abort.signal.aborted) continue; // 已请求过停止，不重复计数 - if (currentStatus(run).kind !== 'running') continue; - run.abort.abort(); - stopped++; - } - return stopped; - }); - yield* Effect.addFinalizer(() => Effect.sync(() => { - for (const run of runs.values()) { - run.abort.abort(); - if (run.fiber) Effect.runFork(Fiber.interrupt(run.fiber)); - } - runs.clear(); + // 终态归属由 turn 的记录承担；这里只请求停止本层仍在跑的派生会话 + for (const parentSessionId of parents) void turn.stopChildren(parentSessionId); + parents.clear(); }) ); - return { spawn, wait, stopAll }; + return { spawn }; }) ); diff --git a/packages/codingcode/src/subagent/subagent.ts b/packages/codingcode/src/subagent/subagent.ts index 63505ee..20e2c9c 100644 --- a/packages/codingcode/src/subagent/subagent.ts +++ b/packages/codingcode/src/subagent/subagent.ts @@ -3,6 +3,7 @@ import { SubagentRunnerService } from './port.js'; import type { RunSubagentOptions } from './port.js'; import { AgentService } from '../agent/port.js'; import type { FrameBody } from '../sink/types.js'; +import { AgentError } from '../util/error.js'; import type { Result } from '../util/result.js'; import { BYPASS_PERMISSION_MODE } from '../util/enums.js'; import type { IncomingPart } from '../llm/types.js'; @@ -17,7 +18,6 @@ export const SubagentRunnerLayer = Layer.effect( const result = yield* agent.runTurn(input, { sessionId: opts.sessionId, cwd: opts.cwd, - signal: opts.signal, activeProfile: opts.activeProfile, // 子代理不经审批：调用点未给定时固定 bypass，避免继承父会话的审批链路 permissionMode: opts.permissionMode ?? BYPASS_PERMISSION_MODE, @@ -26,6 +26,12 @@ export const SubagentRunnerLayer = Layer.effect( agentName: opts.agentName, systemPrompt: opts.systemPrompt, }); + // 子代理委派不开新会话也不可能命中活跃回合：queued 分支不可达 + if (result.kind === 'queued') { + return yield* Effect.fail( + new AgentError('TURN_CONFLICT', `concurrent turn on subagent session ${result.sessionId}`) + ); + } return { stream: result.stream as AsyncGenerator, unknown>, sessionId: result.sessionId, diff --git a/packages/codingcode/src/tools/domains/subagent/wait.ts b/packages/codingcode/src/tools/domains/subagent/wait.ts index bd54b4f..f06cfc9 100644 --- a/packages/codingcode/src/tools/domains/subagent/wait.ts +++ b/packages/codingcode/src/tools/domains/subagent/wait.ts @@ -1,14 +1,14 @@ import { z } from 'zod'; import { Effect } from 'effect'; import type { ToolDefinition } from '../../types.js'; -import { - SubagentRunRegistryService, - SUBAGENT_WAIT_DEFAULT_MS, - SUBAGENT_WAIT_MIN_MS, - SUBAGENT_WAIT_MAX_MS, -} from '../../../subagent/registry.js'; +import { AgentError } from '../../../util/error.js'; +import { TurnRegistryService } from '../../../turn/port.js'; -export const waitAgentTool: ToolDefinition = { +export const SUBAGENT_WAIT_MIN_MS = 10_000; +export const SUBAGENT_WAIT_DEFAULT_MS = 30_000; +export const SUBAGENT_WAIT_MAX_MS = 3_600_000; + +export const waitAgentTool: ToolDefinition = { name: 'wait_agent', concurrencySafe: true, description: @@ -26,13 +26,18 @@ export const waitAgentTool: ToolDefinition = { }), execute: (args, _ctx) => Effect.gen(function* () { - const registry = yield* SubagentRunRegistryService; + const turn = yield* TurnRegistryService; const { sessionId, timeoutMs } = args as { sessionId: string; timeoutMs?: number }; const clamped = Math.min( Math.max(timeoutMs ?? SUBAGENT_WAIT_DEFAULT_MS, SUBAGENT_WAIT_MIN_MS), SUBAGENT_WAIT_MAX_MS ); - const outcome = yield* registry.wait(sessionId, clamped); + const outcome = yield* turn.wait(sessionId, clamped); + if (outcome === undefined) { + return yield* Effect.fail( + new AgentError('TOOL_EXECUTION_FAILED', `Unknown subagent: ${sessionId}`) + ); + } return outcome; }), }; diff --git a/packages/codingcode/src/turn/port.ts b/packages/codingcode/src/turn/port.ts new file mode 100644 index 0000000..851bf99 --- /dev/null +++ b/packages/codingcode/src/turn/port.ts @@ -0,0 +1,44 @@ +import { Context } from 'effect'; +import type { Effect } from 'effect'; +import type { + PendingUserInput, + SubmitOutcome, + TurnClaim, + TurnCommand, + WaitOutcome, +} from './types.js'; + +export interface TurnRegistryShape { + // —— 出生 —— + /** 出生登记：已有 running 记录 → false */ + claim(sessionId: string, claim: TurnClaim): Effect.Effect; + + // —— 转移（状态机本体：一次调用写表并投帧）—— + /** `kind: 'start'` 须在 sink.attach 之后 */ + transition(sessionId: string, cmd: TurnCommand): Effect.Effect; + + // —— steer 输入槽 —— + /** 提交判定：running 则入槽；`id` 由前端提供，回显在 `user_input` 帧上 */ + submit(sessionId: string, input: PendingUserInput): Effect.Effect; + /** 取走全部待投递输入 */ + drain(sessionId: string): Effect.Effect>; + + // —— 查询与停止 —— + /** 等终态；未知 id → undefined，超时 → 'timeout' */ + wait(sessionId: string, timeoutMs: number): Effect.Effect; + /** 结果已进父会话 mailbox，唤醒 wait */ + markDelivered(sessionId: string): Effect.Effect; + /** 登记停止句柄（runFork 之后紧邻调用） */ + arm(sessionId: string, stop: () => void): Effect.Effect; + /** 该父会话下仍在跑的派生会话数 */ + runningChildren(parentSessionId: string): Effect.Effect; + /** 停掉该父会话下所有派生会话 */ + stopChildren(parentSessionId: string): Effect.Effect; + /** 会话删除时清空（含派生条目） */ + dispose(sessionId: string): Effect.Effect; +} + +export class TurnRegistryService extends Context.Tag('TurnRegistry')< + TurnRegistryService, + TurnRegistryShape +>() {} diff --git a/packages/codingcode/src/turn/registry.ts b/packages/codingcode/src/turn/registry.ts new file mode 100644 index 0000000..fc60aab --- /dev/null +++ b/packages/codingcode/src/turn/registry.ts @@ -0,0 +1,206 @@ +import { Deferred, Effect, Layer } from 'effect'; +import { EventSinkService } from '../sink/port.js'; +import { TurnRegistryService } from './port.js'; +import type { + EndReason, + PendingUserInput, + TurnClaim, + TurnCommand, + TurnState, + WaitOutcome, +} from './types.js'; +import type { EndTransition, Transition } from '../sink/types.js'; + +interface TurnRecord { + turnId: number; + state: TurnState; + /** 终态时写入 */ + endReason?: EndReason; + /** 派生关系，子代理并发闸与 stopChildren 按它过滤 */ + parentSessionId?: string; + agentName?: string; + /** 回合级 steer 槽；被 drain 取空 */ + pendingUserInputs: PendingUserInput[]; + /** 终态信号，由 settle 完成 */ + ended: Deferred.Deferred; + /** 投递完成信号（仅委派记录），由 markDelivered 完成 */ + delivered: Deferred.Deferred; + stop?: () => void; + /** stopChildren 已请求过停止，避免重复计数 */ + stopping?: boolean; +} + +export const TurnRegistryLayer = Layer.effect( + TurnRegistryService, + Effect.gen(function* () { + const sink = yield* EventSinkService; + const records = new Map(); // 键 = 会话 sessionId + + /** 相位帧的唯一投递口 */ + const frame = (sessionId: string, transition: Transition) => + sink.emit(sessionId, { family: 'transition', transition }); + + const project = (rec: TurnRecord): WaitOutcome => + rec.state === 'complete' ? 'completed' : 'failed'; + + /** 终结：写状态、投 end 帧、唤醒 wait、回收派生条目 */ + const settle = ( + sessionId: string, + state: TurnState, + endReason: EndReason, + end: EndTransition + ) => + Effect.sync(() => { + const rec = records.get(sessionId); + if (!rec || rec.state !== 'running') return; // 幂等 + rec.state = state; + rec.endReason = endReason; + + Effect.runSync(frame(sessionId, end)); + Effect.runSync(Deferred.succeed(rec.ended, void 0)); + + // 回收本回合派生的终态条目 + for (const [id, child] of records) + if (child.parentSessionId === sessionId && child.state !== 'running') records.delete(id); + }); + + /** 转移：状态机的全部出口都在这一个 switch 里 */ + const transition = (sessionId: string, cmd: TurnCommand): Effect.Effect => + Effect.gen(function* () { + switch (cmd.kind) { + case 'start': { + const rec = records.get(sessionId); + if (rec) yield* frame(sessionId, { to: 'start', turnId: rec.turnId }); + return; + } + case 'running': + yield* frame(sessionId, { + to: 'executing', + ...(cmd.responded ? { responded: cmd.responded } : {}), + }); + return; + case 'compressing': + yield* frame(sessionId, { to: 'compress' }); + return; + case 'complete': + yield* settle(sessionId, 'complete', { kind: cmd.reason }, { + to: 'end', + reason: cmd.reason, + }); + return; + case 'interrupt': + yield* settle(sessionId, 'interrupt', { kind: 'aborted' }, { + to: 'end', + reason: 'aborted', + }); + return; + case 'fail': + yield* settle( + sessionId, + 'error', + { kind: 'error', error: cmd.error }, + { to: 'end', reason: 'error', error: cmd.error } + ); + return; + default: { + // 新增 kind 会在这里编译报错 + const _exhaustive: never = cmd; + return _exhaustive; + } + } + }); + + return { + claim: (sessionId, c: TurnClaim) => + Effect.gen(function* () { + // 只有 running 算活跃；终态记录可被新回合覆盖 + if (records.get(sessionId)?.state === 'running') return false; + records.set(sessionId, { + turnId: c.turnId, + state: 'running', + parentSessionId: c.parentSessionId, + agentName: c.agentName, + pendingUserInputs: [], + ended: yield* Deferred.make(), + delivered: yield* Deferred.make(), + }); + return true; + }), + + transition, + + submit: (sessionId, input: PendingUserInput) => + Effect.sync(() => { + const rec = records.get(sessionId); + if (!rec || rec.state !== 'running') return { kind: 'no-active-turn' as const }; + rec.pendingUserInputs.push(input); + return { kind: 'attached' as const, turnId: rec.turnId }; + }), + + drain: (sessionId) => + Effect.sync(() => { + const rec = records.get(sessionId); + if (!rec) return []; + const out = rec.pendingUserInputs; + rec.pendingUserInputs = []; + return out; + }), + + wait: (sessionId, timeoutMs) => + Effect.gen(function* () { + const rec = records.get(sessionId); + if (!rec) return undefined; + const converged = Effect.gen(function* () { + yield* Deferred.await(rec.ended); // 已终态则立即通过 + // 委派记录：等结果进父会话 mailbox + if (rec.parentSessionId) yield* Deferred.await(rec.delivered); + return project(rec); + }); + return yield* Effect.race( + Effect.sleep(timeoutMs).pipe(Effect.as('timeout')), + converged + ); + }), + + markDelivered: (sessionId) => + Effect.sync(() => { + const rec = records.get(sessionId); + if (rec) Effect.runSync(Deferred.succeed(rec.delivered, void 0)); + }), + + arm: (sessionId, stop) => + Effect.sync(() => { + const rec = records.get(sessionId); + if (rec) rec.stop = stop; + }), + + runningChildren: (parentSessionId) => + Effect.sync(() => { + let n = 0; + for (const rec of records.values()) + if (rec.parentSessionId === parentSessionId && rec.state === 'running') n++; + return n; + }), + + stopChildren: (parentSessionId) => + Effect.sync(() => { + let stopped = 0; + for (const rec of records.values()) { + if (rec.parentSessionId !== parentSessionId || rec.state !== 'running') continue; + if (!rec.stop || rec.stopping) continue; // 未 arm / 已请求过 + rec.stopping = true; + rec.stop(); + stopped++; + } + return stopped; + }), + + dispose: (sessionId) => + Effect.sync(() => { + records.delete(sessionId); + for (const [id, rec] of records) + if (rec.parentSessionId === sessionId) records.delete(id); + }), + }; + }) +); diff --git a/packages/codingcode/src/turn/types.ts b/packages/codingcode/src/turn/types.ts new file mode 100644 index 0000000..930b003 --- /dev/null +++ b/packages/codingcode/src/turn/types.ts @@ -0,0 +1,42 @@ +import type { FrameError, ResponseMeta } from '../sink/types.js'; +import type { IncomingPart } from '../llm/types.js'; + +/** 状态轴：判活只看这一轴 */ +export type TurnState = 'running' | 'complete' | 'interrupt' | 'error'; + +/** 原因轴：终态上的属性，不参与判活 */ +export type EndReason = + | { readonly kind: 'done' } + | { readonly kind: 'maxSteps' } + | { readonly kind: 'aborted' } + | { readonly kind: 'error'; readonly error: FrameError }; + +/** 转移集：状态机本体 */ +export type TurnCommand = + | { readonly kind: 'start' } + | { readonly kind: 'running'; readonly responded?: ResponseMeta } + | { readonly kind: 'compressing' } + | { readonly kind: 'complete'; readonly reason: 'done' | 'maxSteps' } + | { readonly kind: 'interrupt' } + | { readonly kind: 'fail'; readonly error: FrameError }; + +/** 回合级待投递用户输入；`id` 由前端在入队时生成，drain 吸收时回显在 `user_input` 帧上 */ +export interface PendingUserInput { + readonly id: string; + readonly parts: IncomingPart[]; +} + +/** 提交判定：并入当前回合 / 无活跃回合（由调用方开新回合） */ +export type SubmitOutcome = + | { readonly kind: 'attached'; readonly turnId: number } + | { readonly kind: 'no-active-turn' }; + +/** wait 的对外摘要：状态轴的有损投影 + 等待动作的超时 */ +export type WaitOutcome = 'completed' | 'failed' | 'timeout'; + +/** claim 时登记的附加元数据；键是第一个参数 sessionId。派生关系随记录走，父回合为 undefined */ +export interface TurnClaim { + readonly turnId: number; + readonly parentSessionId?: string; + readonly agentName?: string; +} diff --git a/packages/codingcode/src/util/error.ts b/packages/codingcode/src/util/error.ts index 1418241..0489192 100644 --- a/packages/codingcode/src/util/error.ts +++ b/packages/codingcode/src/util/error.ts @@ -16,7 +16,8 @@ export type ErrorCode = | 'AGENT_LOOP_DETECTED' | 'EMPTY_RESPONSE' | 'SESSION_IO_ERROR' - | 'INVALID_INPUT'; + | 'INVALID_INPUT' + | 'TURN_CONFLICT'; export class AgentError extends Error { readonly _tag = 'AgentError'; diff --git a/packages/codingcode/test/context/budget-integration.test.ts b/packages/codingcode/test/context/budget-integration.test.ts index f9569af..3dc6d2b 100644 --- a/packages/codingcode/test/context/budget-integration.test.ts +++ b/packages/codingcode/test/context/budget-integration.test.ts @@ -11,16 +11,20 @@ import type { SessionRef } from '../../src/session/types.js'; import { useTempProjectBase } from '../helpers/project-base.js'; import { ContextLayer, transcriptPathFor } from '../../src/context/context.js'; import { EventSinkLayer } from '../../src/sink/sink.js'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; useTempProjectBase(); +const TurnWithDeps = TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)); + const TestLayer = Layer.mergeAll( SessionLayer, Layer.succeed(LLMService, { complete: () => Effect.fail(new Error('no llm')), completeStream: () => (async function* () {})(), } as any), - EventSinkLayer + EventSinkLayer, + TurnWithDeps ); async function getCtxService(): Promise { diff --git a/packages/codingcode/test/context/compressor/behavior.test.ts b/packages/codingcode/test/context/compressor/behavior.test.ts index 280c68a..2a00fb7 100644 --- a/packages/codingcode/test/context/compressor/behavior.test.ts +++ b/packages/codingcode/test/context/compressor/behavior.test.ts @@ -20,6 +20,7 @@ import { estimateTokens } from '../../../src/context/tokens.js'; import { useTempProjectBase } from '../../helpers/project-base.js'; import { ContextLayer } from '../../../src/context/context.js'; import { EventSinkLayer } from '../../../src/sink/sink.js'; +import { TurnRegistryLayer } from '../../../src/turn/registry.js'; // 上下文窗口现在由 catalog 按模型值现取，测试里钉死成一个可控值 const windowState = vi.hoisted(() => ({ value: 128000 })); @@ -113,7 +114,12 @@ const FailingLLM = { } as any; function makeTestLayer(llm: unknown) { - return Layer.mergeAll(SessionLayer, Layer.succeed(LLMService, llm as any), EventSinkLayer); + return Layer.mergeAll( + SessionLayer, + Layer.succeed(LLMService, llm as any), + EventSinkLayer, + TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)) + ); } async function getCtxService(llm: unknown): Promise { diff --git a/packages/codingcode/test/context/get-history-media.test.ts b/packages/codingcode/test/context/get-history-media.test.ts index 60967ed..c4f91cc 100644 --- a/packages/codingcode/test/context/get-history-media.test.ts +++ b/packages/codingcode/test/context/get-history-media.test.ts @@ -13,16 +13,20 @@ import { assetNameFor, writeAsset } from '../../src/session/assets.js'; import { useTempProjectBase } from '../helpers/project-base.js'; import { ContextLayer, transcriptPathFor } from '../../src/context/context.js'; import { EventSinkLayer } from '../../src/sink/sink.js'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; useTempProjectBase(); +const TurnWithDeps = TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)); + const TestLayer = Layer.mergeAll( SessionLayer, Layer.succeed(LLMService, { complete: () => Effect.fail(new Error('no llm')), completeStream: () => (async function* () {})(), } as any), - EventSinkLayer + EventSinkLayer, + TurnWithDeps ); const CWD = '/tmp/test'; diff --git a/packages/codingcode/test/context/memory-buffer.test.ts b/packages/codingcode/test/context/memory-buffer.test.ts index f1768d4..116f481 100644 --- a/packages/codingcode/test/context/memory-buffer.test.ts +++ b/packages/codingcode/test/context/memory-buffer.test.ts @@ -6,6 +6,7 @@ import { ContextLayer } from '../../src/context/context.js'; import { SessionService } from '../../src/session/port.js'; import { LLMService } from '../../src/llm/port.js'; import { EventSinkLayer } from '../../src/sink/sink.js'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; import type { SessionEvent } from '../../src/session/types.js'; import type { SessionRef } from '../../src/session/types.js'; import { text } from '../helpers/parts.js'; @@ -44,7 +45,16 @@ function makeLayer(counting: ReturnType, summary = S complete: () => Effect.succeed({ content: summary }), completeStream: () => (async function* () {})(), } as any); - return ContextLayer.pipe(Layer.provide(Layer.mergeAll(sessionLayer, llmLayer, EventSinkLayer))); + return ContextLayer.pipe( + Layer.provide( + Layer.mergeAll( + sessionLayer, + llmLayer, + EventSinkLayer, + TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)) + ) + ) + ); } async function getCtx(counting: ReturnType): Promise { diff --git a/packages/codingcode/test/helpers/agent-harness.ts b/packages/codingcode/test/helpers/agent-harness.ts index 1ef7f43..2e511c7 100644 --- a/packages/codingcode/test/helpers/agent-harness.ts +++ b/packages/codingcode/test/helpers/agent-harness.ts @@ -19,6 +19,8 @@ import { SkillService } from '../../src/skills/port.js'; import { SubagentRunnerService } from '../../src/subagent/port.js'; import { TodoService } from '../../src/todo/port.js'; import { ToolExecutorService } from '../../src/tools/port.js'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; +import { TurnRegistryService } from '../../src/turn/port.js'; import type { FrameBody, RuntimeEvent, Transition } from '../../src/sink/types.js'; import type { TokenUsage, LLMStreamPart } from '../../src/llm/types.js'; import type { SessionStoreState, SessionRef } from '../../src/session/types.js'; @@ -267,6 +269,7 @@ export function makeAgentLayer(mocks: HarnessMocks): Layer.Layer { ), resolveAssets: () => Effect.succeed(new Map()), recordUser: () => Effect.succeed({ turnId: 1 }), + recordUserInput: () => Effect.succeed({ turnId: 1 }), recordSystem: () => Effect.succeed({}), recordAssistant: () => Effect.succeed({}), recordToolResult: () => Effect.succeed({}), @@ -320,6 +323,9 @@ export function makeAgentLayer(mocks: HarnessMocks): Layer.Layer { listProjectMcpTools: () => Effect.succeed([]), } as any); + // 状态帧由 turn 投递 ⇒ turn 要 sink；无环：turn → sink + const TurnWithDeps = TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)); + const services = Layer.mergeAll( mocks.sessionLayer ?? Layer.succeed(SessionService, session as any), Layer.succeed(ToolExecutorService, executor as any), @@ -333,6 +339,7 @@ export function makeAgentLayer(mocks: HarnessMocks): Layer.Layer { mocks.skillLayer ?? Layer.succeed(SkillService, skills as any), ContextMockLayer, EventSinkLayer, + TurnWithDeps, MailboxLayer, Layer.succeed(MemoryService, memory as any), Layer.succeed(LLMService, { @@ -403,9 +410,17 @@ export async function runAgentTurn( if (opts.skills) runOpts.skills = opts.skills; return yield* agent.runTurn(opts.input ?? incomingText('test'), runOpts); }); - let runRes: { stream: AsyncGenerator; sessionId: string }; + let runRes: { kind: 'turn'; stream: AsyncGenerator; sessionId: string }; try { - runRes = await Effect.runPromise(Effect.provide(program, appLayer) as any); + const result = (await Effect.runPromise( + Effect.provide(program, appLayer) as any + )) as + | { kind: 'turn'; stream: AsyncGenerator; sessionId: string } + | { kind: 'queued'; sessionId: string; turnId: number }; + if (result.kind === 'queued') { + throw new Error(`harness: runTurn unexpectedly returned queued (turnId ${result.turnId})`); + } + runRes = result; } catch (err) { console.error('HARNESS-RUN-ERROR', err); throw err; diff --git a/packages/codingcode/test/server/subagents-route.test.ts b/packages/codingcode/test/server/subagents-route.test.ts index ba85115..0f92ede 100644 --- a/packages/codingcode/test/server/subagents-route.test.ts +++ b/packages/codingcode/test/server/subagents-route.test.ts @@ -1,10 +1,10 @@ import { describe, it, expect } from 'vitest'; import { Effect, Layer, ManagedRuntime } from 'effect'; import { createServer } from '../../src/server/index.js'; -import { SubagentRunRegistryService } from '../../src/subagent/registry.js'; +import { TurnRegistryService } from '../../src/turn/port.js'; -async function makeApp(stopAll: (sessionId: string) => Effect.Effect) { - const layer = Layer.succeed(SubagentRunRegistryService, { stopAll } as any); +async function makeApp(stopChildren: (sessionId: string) => Effect.Effect) { + const layer = Layer.succeed(TurnRegistryService, { stopChildren } as any); return createServer(ManagedRuntime.make(layer)); } @@ -16,7 +16,7 @@ describe('POST /api/sessions/:id/subagents/stop', () => { expect(await res.json()).toEqual({ stopped: 3 }); }); - it('把路径里的 sessionId 原样透给注册表', async () => { + it('把路径里的 sessionId 原样透给回合注册表', async () => { const seen: string[] = []; const app = await makeApp((sessionId) => Effect.sync(() => { diff --git a/packages/codingcode/test/session/record-user-input.test.ts b/packages/codingcode/test/session/record-user-input.test.ts new file mode 100644 index 0000000..47efc51 --- /dev/null +++ b/packages/codingcode/test/session/record-user-input.test.ts @@ -0,0 +1,162 @@ +import { describe, it, expect } from 'vitest'; +import { mkdirSync, rmSync } from 'fs'; +import { join } from 'path'; +import { randomUUID } from 'crypto'; +import { Effect } from 'effect'; +import { SessionService } from '../../src/session/port.js'; +import { SessionLayer } from '../../src/session/session.js'; +import { encodeProjectPath } from '../../src/util/path.js'; +import { computePaths } from '../../src/session/paths.js'; +import { readSessionMeta } from '../../src/session/file-ops.js'; +import { useTempProjectBase } from '../helpers/project-base.js'; +import { text } from '../helpers/parts.js'; + +const base = useTempProjectBase(); + +function run(eff: Effect.Effect): Promise { + return Effect.runPromise(eff.pipe(Effect.provide(SessionLayer) as any)); +} + +function cleanup(dir: string) { + rmSync(join(base.dir, encodeProjectPath(dir)), { recursive: true, force: true }); + rmSync(dir, { recursive: true, force: true }); +} + +function newDir(): string { + const dir = join(base.dir, randomUUID()); + mkdirSync(dir, { recursive: true }); + return dir; +} + +const create = (dir: string) => + Effect.gen(function* () { + const svc = yield* SessionService; + return yield* svc.create(dir, { + model: 'test-model', + activeProfile: 'build', + permissionMode: 'askBeforeExec', + }); + }); + +describe('recordUserInput — steer input lands under the current turnId', () => { + it('persists a source:"user" event without bumping turnId', async () => { + const dir = newDir(); + try { + const created = await run(create(dir)); + const res = await run( + Effect.gen(function* () { + const svc = yield* SessionService; + const state = yield* svc.load(dir, created.sessionId); + // 先落一条常规用户消息，建立 turnId=1 + yield* svc.recordUser(state, text('first turn')); + const before = state.currentTurnId; + const ev = yield* svc.recordUserInput(state, text('steer me')); + return { before, after: state.currentTurnId, ev }; + }) + ); + + expect(res.ev.source).toBe('user'); + expect(res.ev.turnId).toBe(res.before); + expect(res.after).toBe(res.before); + expect(res.ev.content).toEqual(text('steer me')); + } finally { + cleanup(dir); + } + }); + + it('does not overwrite the title derived from the first message', async () => { + const dir = newDir(); + try { + const created = await run(create(dir)); + await run( + Effect.gen(function* () { + const svc = yield* SessionService; + const state = yield* svc.load(dir, created.sessionId); + yield* svc.recordUser(state, text('fix the login bug')); + yield* svc.recordUserInput(state, text('a steer that must not become the title')); + }) + ); + + expect(readSessionMeta(computePaths(dir, created.sessionId).transcriptPath)?.title).toBe( + 'fix the login bug' + ); + } finally { + cleanup(dir); + } + }); + + it('derives a title from the steer input when none was set yet', async () => { + const dir = newDir(); + try { + const created = await run(create(dir)); + await run( + Effect.gen(function* () { + const svc = yield* SessionService; + const state = yield* svc.load(dir, created.sessionId); + yield* svc.recordUserInput(state, text('first thing said is a steer')); + }) + ); + + expect(readSessionMeta(computePaths(dir, created.sessionId).transcriptPath)?.title).toBe( + 'first thing said is a steer' + ); + } finally { + cleanup(dir); + } + }); + + it('recordUser still bumps turnId for the next turn after a steer', async () => { + const dir = newDir(); + try { + const created = await run(create(dir)); + const res = await run( + Effect.gen(function* () { + const svc = yield* SessionService; + const state = yield* svc.load(dir, created.sessionId); + yield* svc.recordUser(state, text('turn one')); + const afterOne = state.currentTurnId; + yield* svc.recordUserInput(state, text('steer')); + const afterSteer = state.currentTurnId; + yield* svc.recordUser(state, text('turn two')); + return { afterOne, afterSteer, afterTwo: state.currentTurnId }; + }) + ); + + expect(res.afterSteer).toBe(res.afterOne); + expect(res.afterTwo).toBe(res.afterOne + 1); + } finally { + cleanup(dir); + } + }); + + it('the steer event is visible via readUITurns with content matching the parts', async () => { + const dir = newDir(); + try { + const created = await run(create(dir)); + const parts = text('steer with parts'); + await run( + Effect.gen(function* () { + const svc = yield* SessionService; + const state = yield* svc.load(dir, created.sessionId); + yield* svc.recordUser(state, text('turn one')); + yield* svc.recordUserInput(state, parts); + }) + ); + + const turns = await run( + Effect.gen(function* () { + const svc = yield* SessionService; + return yield* svc.readUITurns(created.sessionId, dir); + }) + ); + + const items = turns.flatMap((t) => t.items); + const steer = items.filter((i) => i.type === 'message' && i.role === 'user').at(-1)!; + expect(steer.type).toBe('message'); + if (steer.type !== 'message') throw new Error('expected message'); + expect(steer.parts).toEqual(parts); + } finally { + cleanup(dir); + } + }); +}); diff --git a/packages/codingcode/test/subagent/registry.test.ts b/packages/codingcode/test/subagent/registry.test.ts index ef8eec9..95e2ebf 100644 --- a/packages/codingcode/test/subagent/registry.test.ts +++ b/packages/codingcode/test/subagent/registry.test.ts @@ -1,13 +1,12 @@ import { describe, it, expect } from 'vitest'; import { Effect, Layer, Queue } from 'effect'; -import { - SubagentRunRegistryLayer, - SubagentRunRegistryService, -} from '../../src/subagent/registry.js'; +import { SubagentRunRegistryLayer, SubagentRunRegistryService } from '../../src/subagent/registry.js'; import { SubagentRunnerService } from '../../src/subagent/port.js'; import { MailboxLayer, MailboxService } from '../../src/session/mailbox.js'; import { EventSinkService } from '../../src/sink/port.js'; import { HookService } from '../../src/hooks/port.js'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; +import { TurnRegistryService } from '../../src/turn/port.js'; import type { FrameBody } from '../../src/sink/types.js'; const spawnOpts = { @@ -27,7 +26,7 @@ function doneStream(text = 'child-done'): AsyncGenerator { })(); } -/** 永不结束：用来制造 wait 超时 */ +/** 永不结束：用来制造 hang */ function hangingStream(): AsyncGenerator { return (async function* () { await new Promise((r) => setTimeout(r, 30_000)); @@ -47,15 +46,10 @@ function failStream(): AsyncGenerator { function makeHarness(makeStream: () => AsyncGenerator) { const emitted: Array<{ sessionId: string; body: FrameBody }> = []; - const signals: AbortSignal[] = []; - // 每次 spawn 给一条独立流与递增的 sessionId（配额用例会连续 spawn 多次） let n = 0; const runner = Layer.succeed(SubagentRunnerService, { - runSubagent: (_prompt: string, opts: any) => - Effect.sync(() => { - signals.push(opts.signal); - return { stream: makeStream(), sessionId: `child-${++n}` }; - }), + runSubagent: (_prompt: string, _opts: any) => + Effect.sync(() => ({ stream: makeStream(), sessionId: `child-${++n}` })), } as any); const sink = Layer.succeed(EventSinkService, { attach: () => Effect.succeed(Effect.runSync(Queue.unbounded())), @@ -70,12 +64,12 @@ function makeHarness(makeStream: () => AsyncGenerator) { emitDecision: () => Effect.succeed(null), reloadUserHooks: () => Effect.void, } as any); - // provideMerge：MailboxService 既注入注册表，也保留在输出里供断言使用 - // （同一次 Layer 构建 ⇒ 同一实例） + + const turn = TurnRegistryLayer.pipe(Layer.provide(sink)); const layers = SubagentRunRegistryLayer.pipe( - Layer.provideMerge(Layer.mergeAll(runner, MailboxLayer, sink, hooks)) + Layer.provideMerge(Layer.mergeAll(runner, MailboxLayer, sink, hooks, turn)) ); - return { emitted, signals, layers }; + return { emitted, layers }; } const run = (layers: Layer.Layer, eff: Effect.Effect): Promise => @@ -146,21 +140,7 @@ describe('subagent run registry', () => { expect(statuses).toEqual(['spawned', 'completed']); }); - it('wait 对已终态返回 completed（不阻塞）', async () => { - const { layers } = makeHarness(doneStream); - const outcome = await run( - layers, - Effect.gen(function* () { - const reg = yield* SubagentRunRegistryService; - yield* reg.spawn(spawnOpts); - yield* Effect.sleep(50); - return yield* reg.wait('child-1', 10_000); - }) - ); - expect(outcome).toBe('completed'); - }); - - it('error 终态：mailbox 正文含失败原因，wait 返回 failed', async () => { + it('error 终态：mailbox 正文含失败原因', async () => { const { layers } = makeHarness(failStream); const result = await run( layers, @@ -169,109 +149,56 @@ describe('subagent run registry', () => { const mb = yield* MailboxService; yield* reg.spawn(spawnOpts); yield* Effect.sleep(50); - const items = yield* mb.drain('parent-1'); - const outcome = yield* reg.wait('child-1', 10_000); - return { items, outcome }; - }) - ); - expect(result.outcome).toBe('failed'); - expect(result.items[0]!.content).toContain('boom'); - expect(result.items[0]!.content).toContain('did not finish'); - }); - - it('wait 超时返回 timeout，且不取消子代理', async () => { - const { layers } = makeHarness(hangingStream); - const outcome = await run( - layers, - Effect.gen(function* () { - const reg = yield* SubagentRunRegistryService; - yield* reg.spawn(spawnOpts); - return yield* reg.wait('child-1', 60); - }) - ); - expect(outcome).toBe('timeout'); - }); - - it('wait 未知 sessionId 报 Unknown subagent', async () => { - const { layers } = makeHarness(doneStream); - const error = await run( - layers, - Effect.gen(function* () { - const reg = yield* SubagentRunRegistryService; - return yield* Effect.either(reg.wait('nope', 60)); + return yield* mb.drain('parent-1'); }) ); - expect(error._tag).toBe('Left'); - expect(String((error as any).left.message)).toMatch(/Unknown subagent/); + expect(result[0]!.content).toContain('boom'); + expect(result[0]!.content).toContain('did not finish'); }); - it('配额：超过 maxBackground 时拒绝新的 spawn', async () => { + it('配额：超过 maxBackground 时拒绝新的 spawn（并发闸走状态表）', async () => { const { layers } = makeHarness(hangingStream); const result = await run( layers, Effect.gen(function* () { const reg = yield* SubagentRunRegistryService; + const turn = yield* TurnRegistryService; const outcomes: string[] = []; - for (let i = 0; i < 6; i++) { + // 并发闸先判、再 spawn、成功后才 claim 占位（与 runTurn 的真实顺序一致） + for (let i = 1; i <= 6; i++) { const r = yield* Effect.either(reg.spawn(spawnOpts)); outcomes.push(r._tag); + if (r._tag === 'Right') { + yield* turn.claim(r.right.sessionId, { turnId: 1, parentSessionId: 'parent-1' }); + } } return outcomes; }) ); - // 默认上限 4：前四个成功，之后失败 - expect(result.slice(0, 4)).toEqual(['Right', 'Right', 'Right', 'Right']); - expect(result.slice(4)).toEqual(['Left', 'Left']); + // 默认上限 4：前四个成功，之后被拒 + expect(result).toEqual(['Right', 'Right', 'Right', 'Right', 'Left', 'Left']); }); - it('stopAll 只停仍在跑的，返回停掉的数量；重复调用不重复计数', async () => { - const { layers, signals } = makeHarness(hangingStream); + it('stopChildren 只停仍在跑的派生会话（句柄由 arm 登记）', async () => { + const { layers } = makeHarness(hangingStream); const result = await run( layers, Effect.gen(function* () { const reg = yield* SubagentRunRegistryService; - yield* reg.spawn(spawnOpts); - yield* reg.spawn(spawnOpts); - const first = yield* reg.stopAll('parent-1'); - const second = yield* reg.stopAll('parent-1'); + const turn = yield* TurnRegistryService; + for (let i = 0; i < 2; i++) { + const sid = `child-${i + 1}`; + yield* turn.claim(sid, { turnId: 1, parentSessionId: 'parent-1' }); + const stopped: string[] = []; + yield* turn.arm(sid, () => stopped.push(sid)); + yield* reg.spawn(spawnOpts); + } + const first = yield* turn.stopChildren('parent-1'); + const second = yield* turn.stopChildren('parent-1'); return { first, second }; }) ); expect(result.first).toBe(2); expect(result.second).toBe(0); - expect(signals.every((s) => s.aborted)).toBe(true); - }); - - it('stopAll 不碰已终态的 run，也不误伤别的父会话', async () => { - const { layers } = makeHarness(doneStream); - const result = await run( - layers, - Effect.gen(function* () { - const reg = yield* SubagentRunRegistryService; - yield* reg.spawn(spawnOpts); - yield* reg.spawn({ ...spawnOpts, parentSessionId: 'parent-2' }); - yield* Effect.sleep(50); - return { - done: yield* reg.stopAll('parent-1'), - other: yield* reg.stopAll('parent-2'), - }; - }) - ); - expect(result.done).toBe(0); - expect(result.other).toBe(0); - }); - - it('scope 结束（dispose）时 abort 掉所有仍活着的子代理', async () => { - const { layers, signals } = makeHarness(hangingStream); - await run( - layers, - Effect.gen(function* () { - const reg = yield* SubagentRunRegistryService; - yield* reg.spawn(spawnOpts); - yield* reg.spawn(spawnOpts); - }) - ); - expect(signals).toHaveLength(2); - expect(signals.every((s) => s.aborted)).toBe(true); }); }); diff --git a/packages/codingcode/test/subagent/runner-wiring.test.ts b/packages/codingcode/test/subagent/runner-wiring.test.ts index 6c788ff..a3b8cd2 100644 --- a/packages/codingcode/test/subagent/runner-wiring.test.ts +++ b/packages/codingcode/test/subagent/runner-wiring.test.ts @@ -13,6 +13,7 @@ import { CheckpointService } from '../../src/checkpoint/port.js'; import { ContextService } from '../../src/context/port.js'; import { EventSinkLayer } from '../../src/sink/sink.js'; import { MailboxLayer } from '../../src/session/mailbox.js'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; import { LLMService } from '../../src/llm/port.js'; import { AgentError } from '../../src/util/error.js'; import { MemoryService } from '../../src/memory/port.js'; @@ -126,6 +127,8 @@ const TodoMock = Layer.succeed(TodoService, { reset: () => {}, } as any); +const TurnWithDeps = TurnRegistryLayer.pipe(Layer.provide(EventSinkLayer)); + const AgentDeps = Layer.mergeAll( SessionLayer, Layer.succeed(ToolExecutorService, { @@ -150,6 +153,7 @@ const AgentDeps = Layer.mergeAll( dispose: () => Effect.void, } as any), EventSinkLayer, + TurnWithDeps, MailboxLayer, Layer.succeed(MemoryService, { loadMemoryForPrompt: () => Effect.succeed(''), diff --git a/packages/codingcode/test/turn/registry.test.ts b/packages/codingcode/test/turn/registry.test.ts new file mode 100644 index 0000000..daafc3a --- /dev/null +++ b/packages/codingcode/test/turn/registry.test.ts @@ -0,0 +1,280 @@ +import { describe, it, expect } from 'vitest'; +import { Effect, Layer, Queue } from 'effect'; +import { TurnRegistryLayer } from '../../src/turn/registry.js'; +import { TurnRegistryService } from '../../src/turn/port.js'; +import type { PendingUserInput, TurnCommand } from '../../src/turn/types.js'; +import { EventSinkService } from '../../src/sink/port.js'; +import type { FrameBody } from '../../src/sink/types.js'; +import { textPart } from '../../src/llm/types.js'; + +function makeHarness() { + const emitted: Array<{ sessionId: string; body: FrameBody }> = []; + const sink = Layer.succeed(EventSinkService, { + attach: () => Effect.succeed(Effect.runSync(Queue.unbounded())), + detach: () => Effect.void, + emit: (sessionId: string, body: FrameBody) => + Effect.sync(() => { + emitted.push({ sessionId, body }); + }), + } as any); + const layers = TurnRegistryLayer.pipe(Layer.provide(sink)); + /** 投给某会话的帧序列投影 */ + const framesFor = (sessionId: string) => + emitted.filter((e) => e.sessionId === sessionId).map((e) => e.body); + return { emitted, layers, framesFor }; +} + +const run = (layers: Layer.Layer, eff: Effect.Effect): Promise => + Effect.runPromise(eff.pipe(Effect.provide(layers)) as unknown as Effect.Effect); + +const input = (id: string): PendingUserInput => ({ id, parts: [textPart('hi')] }); + +describe('turn registry', () => { + it('claim 在已有 running 记录时返回 false 且不覆盖 turnId', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + const first = yield* t.claim('s1', { turnId: 1 }); + const second = yield* t.claim('s1', { turnId: 2 }); + return { first, second }; + }) + ); + expect(result.first).toBe(true); + expect(result.second).toBe(false); + }); + + it('记录终态后 claim 成功并换成新 turnId', async () => { + const { layers, framesFor } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('s1', { turnId: 1 }); + yield* t.transition('s1', { kind: 'complete', reason: 'done' }); + const re = yield* t.claim('s1', { turnId: 5 }); + yield* t.transition('s1', { kind: 'start' }); + return re; + }) + ); + expect(result).toBe(true); + // 新回合的 start 帧带新 turnId + const start = framesFor('s1').find( + (f) => f.family === 'transition' && f.transition.to === 'start' + ) as any; + expect(start.transition.turnId).toBe(5); + }); + + it('submit 无记录 / 终态记录返回 no-active-turn，running 时返回 attached 且 parts 原样', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + const none = yield* t.submit('s1', input('a')); + yield* t.claim('s1', { turnId: 7 }); + const attached = yield* t.submit('s1', input('b')); + const drained = yield* t.drain('s1'); + yield* t.transition('s1', { kind: 'complete', reason: 'done' }); + const afterEnd = yield* t.submit('s1', input('c')); + return { none, attached, drained, afterEnd }; + }) + ); + expect(result.none.kind).toBe('no-active-turn'); + expect(result.attached).toEqual({ kind: 'attached', turnId: 7 }); + expect(result.drained.map((d) => d.id)).toEqual(['b']); + expect(result.afterEnd.kind).toBe('no-active-turn'); + }); + + it('drain 取走后槽为空（第二次返回空数组）', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('s1', { turnId: 1 }); + yield* t.submit('s1', input('a')); + const first = yield* t.drain('s1'); + const second = yield* t.drain('s1'); + return { first: first.length, second: second.length }; + }) + ); + expect(result.first).toBe(1); + expect(result.second).toBe(0); + }); + + it.each([ + { kind: 'complete', reason: 'done' }, + { kind: 'complete', reason: 'maxSteps' }, + { kind: 'interrupt' }, + { kind: 'fail', error: { message: 'boom', code: 'X' } }, + ])('终态 kind %o 投出对应 end 帧，且重复只投一帧', async (cmd) => { + const { layers, framesFor } = makeHarness(); + await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('s1', { turnId: 1 }); + yield* t.transition('s1', cmd); + yield* t.transition('s1', cmd); + }) + ); + const ends = framesFor('s1').filter( + (f) => f.family === 'transition' && f.transition.to === 'end' + ); + expect(ends).toHaveLength(1); + }); + + it('start / running / compressing 投帧但不改变 running 状态', async () => { + const { layers, framesFor } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('s1', { turnId: 3 }); + yield* t.transition('s1', { kind: 'start' }); + yield* t.transition('s1', { kind: 'running' }); + yield* t.transition('s1', { kind: 'compressing' }); + // 仍在 running ⇒ submit 命中 + return yield* t.submit('s1', input('x')); + }) + ); + expect(result.kind).toBe('attached'); + const tos = framesFor('s1').map((f) => (f.family === 'transition' ? f.transition.to : f.family)); + expect(tos).toEqual(['start', 'executing', 'compress']); + }); + + it('wait：running 阻塞到终结并投影 completed', async () => { + const { layers } = makeHarness(); + const outcome = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('s1', { turnId: 1 }); + // 稍后终结 + yield* Effect.forkDaemon( + Effect.gen(function* () { + yield* Effect.sleep(30); + yield* t.transition('s1', { kind: 'complete', reason: 'done' }); + }) + ); + return yield* t.wait('s1', 5_000); + }) + ); + expect(outcome).toBe('completed'); + }); + + it('wait：已终态立即返回且可重复 wait', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('s1', { turnId: 1 }); + yield* t.transition('s1', { kind: 'interrupt' }); + return { a: yield* t.wait('s1', 5_000), b: yield* t.wait('s1', 5_000) }; + }) + ); + expect(result.a).toBe('failed'); + expect(result.b).toBe('failed'); + }); + + it('wait：未知 id 返回 undefined；超时返回 timeout', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + const unknown = yield* t.wait('nope', 10); + yield* t.claim('s1', { turnId: 1 }); + const timed = yield* t.wait('s1', 20); + return { unknown, timed }; + }) + ); + expect(result.unknown).toBeUndefined(); + expect(result.timed).toBe('timeout'); + }); + + it('有 parentSessionId 的记录在 markDelivered 之前不返回', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('child', { turnId: 1, parentSessionId: 'parent' }); + yield* t.transition('child', { kind: 'complete', reason: 'done' }); + const before = yield* t.wait('child', 30); + yield* t.markDelivered('child'); + const after = yield* t.wait('child', 5_000); + return { before, after }; + }) + ); + expect(result.before).toBe('timeout'); + expect(result.after).toBe('completed'); + }); + + it('settle 归属回收：父终结时清掉终态子记录、保留 running 子记录', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('parent', { turnId: 1 }); + yield* t.claim('done-child', { turnId: 1, parentSessionId: 'parent' }); + yield* t.transition('done-child', { kind: 'complete', reason: 'done' }); + yield* t.claim('live-child', { turnId: 1, parentSessionId: 'parent' }); + yield* t.transition('parent', { kind: 'complete', reason: 'done' }); + // done-child 已被回收 ⇒ wait 返回 undefined + return { + doneChild: yield* t.wait('done-child', 10), + liveChild: yield* t.wait('live-child', 10), + }; + }) + ); + expect(result.doneChild).toBeUndefined(); + expect(result.liveChild).toBe('timeout'); + }); + + it('runningChildren / stopChildren 只命中匹配父会话且 running 的记录', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + const stopped: string[] = []; + yield* t.claim('c1', { turnId: 1, parentSessionId: 'p1' }); + yield* t.claim('c2', { turnId: 1, parentSessionId: 'p1' }); + yield* t.claim('c3', { turnId: 1, parentSessionId: 'p2' }); + yield* t.arm('c1', () => stopped.push('c1')); + yield* t.arm('c2', () => stopped.push('c2')); + // 未 arm 的 c3 会被跳过 + const before = yield* t.runningChildren('p1'); + const stoppedCount = yield* t.stopChildren('p1'); + return { before, stoppedCount, stopped }; + }) + ); + expect(result.before).toBe(2); + expect(result.stoppedCount).toBe(2); + expect(result.stopped).toEqual(['c1', 'c2']); + }); + + it('dispose 同时清掉 parentSessionId 指向该会话的派生条目', async () => { + const { layers } = makeHarness(); + const result = await run( + layers, + Effect.gen(function* () { + const t = yield* TurnRegistryService; + yield* t.claim('parent', { turnId: 1 }); + yield* t.claim('child', { turnId: 1, parentSessionId: 'parent' }); + yield* t.dispose('parent'); + return { + parent: yield* t.wait('parent', 10), + child: yield* t.wait('child', 10), + }; + }) + ); + expect(result.parent).toBeUndefined(); + expect(result.child).toBeUndefined(); + }); +}); diff --git a/packages/desktop/src/App.tsx b/packages/desktop/src/App.tsx index 2389be0..2bda5e1 100644 --- a/packages/desktop/src/App.tsx +++ b/packages/desktop/src/App.tsx @@ -12,7 +12,7 @@ import TitleBar from './TitleBar'; import ErrorBoundary from './shared/ErrorBoundary'; export default function App() { - const { sendMessage, abort } = useAgentCore(); + const { sendMessage, sendQueuedInput, abort } = useAgentCore(); const theme = useUIStore((s) => s.theme); const view = useUIStore((s) => s.view); const rootPath = useWorkspaceStore((s) => s.rootPath); @@ -49,7 +49,7 @@ export default function App() {
- +
); } diff --git a/packages/desktop/src/agent/AgentWorkspace.tsx b/packages/desktop/src/agent/AgentWorkspace.tsx index b3bfc65..dcc9eab 100644 --- a/packages/desktop/src/agent/AgentWorkspace.tsx +++ b/packages/desktop/src/agent/AgentWorkspace.tsx @@ -1,7 +1,18 @@ import { useState, useRef, useCallback, useLayoutEffect, useEffect } from 'react'; import { createPortal } from 'react-dom'; -import { Send, Square, ShieldCheck, Shield, FileText, Paperclip, X } from 'lucide-react'; -import { useAgentStore } from '../stores/agent.store'; +import { + Send, + Square, + ShieldCheck, + Shield, + FileText, + Paperclip, + X, + GripVertical, + Pencil, +} from 'lucide-react'; +import { useShallow } from 'zustand/react/shallow'; +import { useAgentStore, type QueuedInput } from '../stores/agent.store'; import { useWorkspaceStore } from '../stores/workspace.store'; import { compactSession, @@ -17,7 +28,7 @@ import ApprovalPanel from './ApprovalPanel'; import ProfileIndicator from './ProfileIndicator'; import PlanPanel from '../shared/PlanPanel'; import MediaView from '../shared/MediaView'; -import { reachableMimes, attachmentHint, type ContentPart } from '@shared/parts'; +import { reachableMimes, attachmentHint, isLocalMedia, textOf, type ContentPart } from '@shared/parts'; import { MAX_MEDIA_BYTES, type PermissionMode } from '@codingcode/sdk'; /** 附件条上限：单文件字节上限用 sdk 的 `MAX_MEDIA_BYTES`，与服务端闸门同值，前端只做即时提示。 */ @@ -247,11 +258,13 @@ interface SkillRef { function InputBox({ centered, sendMessage, + sendQueuedInput, abort, onOpenPlanPanel, }: { centered?: boolean; sendMessage: (parts: ContentPart[], cwd?: string, skills?: SkillRef[]) => Promise; + sendQueuedInput: (threadId: string, item: QueuedInput) => Promise; abort: () => void; onOpenPlanPanel?: () => void; }) { @@ -275,6 +288,17 @@ function InputBox({ }); const storePermissionMode = useAgentStore((s) => s.permissionMode); const setPermissionMode = useAgentStore((s) => s.setPermissionMode); + const queueInput = useAgentStore((s) => s.queueInput); + const removeQueuedInput = useAgentStore((s) => s.removeQueuedInput); + const reorderQueuedInputs = useAgentStore((s) => s.reorderQueuedInputs); + // useShallow：selector 在无队列 / 无会话时会返回新的 `[]` 字面量， + // 默认的 Object.is 比较会让每次 render 都判定「值变了」→ 自激更新循环。 + // 浅比较按元素引用判等，彻底消除「新数组 identity」问题。 + const queuedInputs = useAgentStore( + useShallow((s) => + s.currentThreadId ? (s.queuedInputsByThreadId[s.currentThreadId] ?? []) : [] + ) + ); const workspace = useWorkspaceStore(); const pendingInput = useAgentStore((s) => s.pendingInput); const setPendingInput = useAgentStore((s) => s.setPendingInput); @@ -448,7 +472,7 @@ function InputBox({ const handleSend = useCallback(() => { const trimmed = text.trim(); - if ((!trimmed && attachments.length === 0) || isStreaming) return; + if (!trimmed && attachments.length === 0) return; const skills: SkillRef[] = []; const seen = new Set(); for (const m of trimmed.matchAll(/@([a-zA-Z0-9-]+)/g)) { @@ -472,20 +496,99 @@ function InputBox({ setNotice(null); setPickedPaths({}); setSkillMenu(null); + if (isStreaming) { + // 活跃回合：只入前端队列，不发请求（等点击发送或回合结束后补发） + const threadId = currentThreadId; + if (threadId) queueInput(threadId, parts); + return; + } sendMessage(parts, workspace.rootPath || undefined, skills); }, [ text, attachments, isStreaming, + currentThreadId, + queueInput, sendMessage, workspace.rootPath, pickedPaths, skillOptions, ]); + /** 编辑队列项：文本回输入框、本地媒体回附件条、该项消失 */ + const editQueuedInput = useCallback( + (item: QueuedInput) => { + const threadId = currentThreadId; + if (!threadId) return; + const restoredAttachments: Attachment[] = []; + for (const p of item.parts) { + if (p.type === 'text') { + setText((prev) => (prev ? `${prev}\n${p.text}` : p.text)); + } else if (isLocalMedia(p)) { + restoredAttachments.push({ + id: newId(), + dataUrl: p.dataUrl, + mimeType: p.mimeType, + filename: p.filename ?? 'attachment', + }); + } + } + if (restoredAttachments.length > 0) { + setAttachments((prev) => [...prev, ...restoredAttachments]); + } + removeQueuedInput(threadId, item.id); + }, + [currentThreadId, removeQueuedInput] + ); + return (
+ {queuedInputs.length > 0 && currentThreadId && ( +
+ {queuedInputs.map((item, index) => ( +
+ + {index + 1} + + {textOf(item.parts) || '（空）'} + + + + +
+ ))} +
+ )}
{ e.preventDefault(); @@ -565,7 +668,6 @@ function InputBox({ } }} placeholder="可向 AI 询问任何事" - disabled={isStreaming} rows={3} className="flex-1 bg-transparent px-5 pt-4 pb-3 text-[15px] text-[var(--text-primary)] placeholder-[var(--text-disabled)] resize-none outline-none leading-relaxed disabled:opacity-50" /> @@ -591,7 +693,6 @@ function InputBox({ } fileInputRef.current?.click(); }} - disabled={isStreaming} aria-label="添加图片、音频或 PDF" title={capsHint ?? '添加图片、音频或 PDF'} className="w-8 h-8 shrink-0 flex items-center justify-center rounded-full text-[var(--text-placeholder)] hover:text-[var(--text-secondary)] hover:bg-[var(--bg-hover)] disabled:opacity-40 transition-colors" @@ -750,10 +851,11 @@ function InputBox({ interface AgentWorkspaceProps { sendMessage: (parts: ContentPart[], cwd?: string, skills?: SkillRef[]) => Promise; + sendQueuedInput: (threadId: string, item: QueuedInput) => Promise; abort: () => void; } -export default function AgentWorkspace({ sendMessage, abort }: AgentWorkspaceProps) { +export default function AgentWorkspace({ sendMessage, sendQueuedInput, abort }: AgentWorkspaceProps) { const currentThreadId = useAgentStore((s) => s.currentThreadId); const isCompressing = useAgentStore((s) => s.isCompressing); const workspace = useWorkspaceStore(); @@ -769,7 +871,7 @@ export default function AgentWorkspace({ sendMessage, abort }: AgentWorkspacePro {' '} 中构建什么？ - +
); } @@ -789,6 +891,7 @@ export default function AgentWorkspace({ sendMessage, abort }: AgentWorkspacePro
setPlanPanelOpen(true)} /> diff --git a/packages/desktop/src/agent/MessageStream.tsx b/packages/desktop/src/agent/MessageStream.tsx index f37b9e6..0bad42f 100644 --- a/packages/desktop/src/agent/MessageStream.tsx +++ b/packages/desktop/src/agent/MessageStream.tsx @@ -1,6 +1,7 @@ import { useEffect, useLayoutEffect, useRef, useState, useCallback, useMemo } from 'react'; import { Copy, Check } from 'lucide-react'; import { useVirtualizer } from '@tanstack/react-virtual'; +import { useShallow } from 'zustand/react/shallow'; import { useAgentStore } from '../stores/agent.store'; import { useRollbackStore } from '../stores/rollback.store'; import MessageItem from '../shared/MessageItem'; @@ -216,7 +217,11 @@ function TurnDiffPanel({ } export default function MessageStream({ threadId }: MessageStreamProps) { - const turns = useAgentStore((s) => s.threads[threadId]?.turns ?? []); + // useShallow：`?? []` 分支每次 render 都会新建数组，默认 Object.is 比较会 + // 判为「值变了」→ 流式高频渲染时形成自激更新循环。浅比较按元素引用判等。 + const turns = useAgentStore( + useShallow((s) => s.threads[threadId]?.turns ?? []) + ); /** 资产地址按项目定位，气泡里的媒体直接指向服务端的资产路由 */ const threadCwd = useAgentStore((s) => s.threads[threadId]?.cwd ?? ''); const setCurrentThread = useAgentStore((s) => s.setCurrentThread); diff --git a/packages/desktop/src/hooks/useAgent.ts b/packages/desktop/src/hooks/useAgent.ts index 6a40136..6d6f461 100644 --- a/packages/desktop/src/hooks/useAgent.ts +++ b/packages/desktop/src/hooks/useAgent.ts @@ -1,5 +1,5 @@ import { useEffect, useCallback, useRef } from 'react'; -import { useAgentStore, type ModelEntry } from '../stores/agent.store'; +import { useAgentStore, type ModelEntry, type QueuedInput } from '../stores/agent.store'; import { useWorkspaceStore } from '../stores/workspace.store'; import { useRollbackStore } from '../stores/rollback.store'; import { agentClient } from '../lib/core-api'; @@ -86,6 +86,8 @@ function registerInflight(threadId: string, controller: AbortController): void { export function useAgentCore() { const lastRootRef = useRef(null); + /** 回合结束后自动补发的稳定入口（在 flushQueued 定义后赋值） */ + const sendQueuedInputRef = useRef<((threadId: string) => Promise) | null>(null); const startTurn = useAgentStore((s) => s.startTurn); const applyChunk = useAgentStore((s) => s.applyChunk); const updateTurnId = useAgentStore((s) => s.updateTurnId); @@ -111,6 +113,9 @@ export function useAgentCore() { const setProfile = useAgentStore((s) => s.setProfile); const setPermissionMode = useAgentStore((s) => s.setPermissionMode); const modelId = useAgentStore((s) => s.model); + const queueInput = useAgentStore((s) => s.queueInput); + const removeQueuedInput = useAgentStore((s) => s.removeQueuedInput); + const markQueueSending = useAgentStore((s) => s.markQueueSending); // Abort all in-flight streams when the workspace root changes (project switch). useEffect(() => { @@ -195,6 +200,96 @@ export function useAgentCore() { }); }, [currentThreadId, setThreadTurns]); + /** + * 建乐观 Turn + 消费帧流。sendMessage 与 sendQueuedInput 共用； + * `preRegistered` 为 sendQueuedInput 已登记的 controller，避免重复登记。 + */ + const runTurnStream = useCallback( + async ( + threadId: string, + parts: ContentPart[], + opts: { + cwd: string; + model: string; + skills?: Array<{ name: string; path: string }>; + preRegistered?: AbortController; + } + ) => { + let activeTurnId = randomId(); + const userItem: Item = { id: randomId(), type: 'message', role: 'user', parts }; + const turn: Turn = { id: activeTurnId, items: [userItem], status: 'running' }; + startTurn(threadId, turn, { + cwd: opts.cwd, + title: textOf(parts).slice(0, 60), + model: opts.model, + }); + + const controller = opts.preRegistered ?? new AbortController(); + if (!opts.preRegistered) registerInflight(threadId, controller); + + const state = createStreamState(randomId()); + const fx: StreamEffects = { + applyItem: (item) => applyChunk(threadId, activeTurnId, item), + applyTodo: (items) => applyTodoUpdate(threadId, items), + setUsage: (usage) => { + setThreadUsage(threadId, usage); + const s = useAgentStore.getState(); + const model = s.models.find((m) => m.id === s.model); + if (model) setContextUsage({ used: usage.prompt, contextWindow: model.context_window }); + }, + setCompacted: () => { + useAgentStore.getState().clearThreadUsage(threadId); + }, + syncTurnId: (turnId) => { + const next = String(turnId); + updateTurnId(threadId, activeTurnId, next); + activeTurnId = next; + }, + newId: () => randomId(), + onInputDelivered: (id) => removeQueuedInput(threadId, id), + }; + + let failed = false; + try { + const result = await agentClient.submitInput(toWireParts(parts), { + sessionId: threadId, + cwd: opts.cwd, + model: opts.model, + signal: controller.signal, + skills: opts.skills, + }); + if (result.kind === 'turn') { + for await (const frame of result.stream) { + reduceFrame(frame, state, fx); + } + } + } catch (err: any) { + failed = true; + const msg = err instanceof ApiError ? (err.body?.message ?? err.message) : String(err); + applyChunk(threadId, activeTurnId, { id: randomId(), type: 'error', message: msg }); + } finally { + completeTurn(threadId, activeTurnId, failed || state.hasError ? 'error' : 'completed'); + if (!failed && !state.hasError && state.planTitle !== null) { + setPendingPlan(threadId, { sessionId: threadId, title: state.planTitle }); + } + abortAndClear(threadId); + // 自动补发：回合结束后按顺序把仍处于 queued 的项逐条发出 + void sendQueuedInputRef.current?.(threadId); + } + }, + [ + startTurn, + applyChunk, + applyTodoUpdate, + completeTurn, + setPendingPlan, + updateTurnId, + setThreadUsage, + setContextUsage, + removeQueuedInput, + ] + ); + const sendMessage = useCallback( async (parts: ContentPart[], cwd?: string, skills?: Array<{ name: string; path: string }>) => { const effectiveCwd = cwd || workspace.rootPath || ''; @@ -235,95 +330,106 @@ export function useAgentCore() { } const threadId: string = resolvedThreadId; + // 首次发送的防重：已有在飞流时直接忽略 if (inflightControllers.has(threadId)) return; clearPendingPlan(threadId); - let activeTurnId = randomId(); - const userItem: Item = { id: randomId(), type: 'message', role: 'user', parts }; - const turn: Turn = { id: activeTurnId, items: [userItem], status: 'running' }; - startTurn(threadId, turn, { - cwd: effectiveCwd, - title: textOf(parts).slice(0, 60), - model, - }); + await runTurnStream(threadId, parts, { cwd: effectiveCwd, model, skills }); + }, + [ + runTurnStream, + setCurrentThreadWithProfile, + clearPendingPlan, + workspace.rootPath, + storeProfile, + storePermissionMode, + modelId, + currentThreadId, + setActiveModel, + setModels, + ] + ); + /** + * 队列项投递：命中活跃回合 → 202，等运行中的 SSE 送来 user_input 帧； + * 无活跃回合 → 开新回合（自动补发即走此分支）。 + */ + const sendQueuedInput = useCallback( + async (threadId: string, item: QueuedInput) => { + markQueueSending(threadId, item.id); + const thread = useAgentStore.getState().threads[threadId]; + const effectiveCwd = thread?.cwd || workspace.rootPath || ''; + const model = thread?.model || modelId; const controller = new AbortController(); - registerInflight(threadId, controller); - - const state = createStreamState(randomId()); - const fx: StreamEffects = { - applyItem: (item) => applyChunk(threadId, activeTurnId, item), - applyTodo: (items) => applyTodoUpdate(threadId, items), - setUsage: (usage) => { - setThreadUsage(threadId, usage); - const s = useAgentStore.getState(); - const model = s.models.find((m) => m.id === s.model); - if (model) setContextUsage({ used: usage.prompt, contextWindow: model.context_window }); - }, - setCompacted: () => { - useAgentStore.getState().clearThreadUsage(threadId); - }, - syncTurnId: (turnId) => { - const next = String(turnId); - updateTurnId(threadId, activeTurnId, next); - activeTurnId = next; - }, - newId: () => randomId(), - }; - + let result; try { - const stream = agentClient.sendMessage(toWireParts(parts), { + result = await agentClient.submitInput(toWireParts(item.parts), { sessionId: threadId, + inputId: item.id, cwd: effectiveCwd, model, signal: controller.signal, - skills, }); - - for await (const frame of stream) { - reduceFrame(frame, state, fx); - } - - completeTurn(threadId, activeTurnId, state.hasError ? 'error' : 'completed'); - if (!state.hasError && state.planTitle !== null) { - setPendingPlan(threadId, { sessionId: threadId, title: state.planTitle }); - } - } catch (err: any) { + } catch (err) { + // 提交失败：移除该项并提示 + removeQueuedInput(threadId, item.id); const msg = err instanceof ApiError ? (err.body?.message ?? err.message) : String(err); - applyChunk(threadId, activeTurnId, { id: randomId(), type: 'error', message: msg }); - completeTurn(threadId, activeTurnId, 'error'); - } finally { - abortAndClear(threadId); + const running = useAgentStore + .getState() + .threads[threadId]?.turns.find((t) => t.status === 'running'); + if (running) applyChunk(threadId, running.id, { id: randomId(), type: 'error', message: msg }); + return; } + if (result.kind === 'queued') return; // 等运行中的 SSE 送来 user_input 帧 + + // 无活跃回合：开新回合；乐观 Turn 用 item.parts + removeQueuedInput(threadId, item.id); + if (inflightControllers.has(threadId)) return; + clearPendingPlan(threadId); + registerInflight(threadId, controller); + await runTurnStream(threadId, item.parts, { + cwd: effectiveCwd, + model, + skills: undefined, + preRegistered: controller, + }); }, [ - startTurn, - setCurrentThreadWithProfile, + runTurnStream, + markQueueSending, + removeQueuedInput, applyChunk, - applyTodoUpdate, - completeTurn, - setPendingPlan, clearPendingPlan, - updateTurnId, - setThreadUsage, - setContextUsage, workspace.rootPath, - storeProfile, - storePermissionMode, modelId, - currentThreadId, - setActiveModel, - setModels, ] ); + /** 回合结束后：按顺序补发仍 queued 的队列项（每项都是一次 sendQueuedInput） */ + const flushQueued = useCallback( + async (threadId: string) => { + const pending = (useAgentStore.getState().queuedInputsByThreadId[threadId] ?? []).filter( + (q) => q.status === 'queued' + ); + for (const item of pending) { + const current = useAgentStore.getState().queuedInputsByThreadId[threadId] ?? []; + if (!current.some((q) => q.id === item.id && q.status === 'queued')) continue; + await sendQueuedInput(threadId, item); + } + }, + [sendQueuedInput] + ); + + // runTurnStream 的 finally 需要一个稳定引用，避免与 flushQueued 互相依赖 + sendQueuedInputRef.current = flushQueued; + const abort = useCallback(() => { const threadId = currentThreadId; if (!threadId) return; abortAndClear(threadId); }, [currentThreadId]); - return { sendMessage, abort }; + return { sendMessage, sendQueuedInput, queueInput, abort }; } // ---- useAgentApproval: approveTool + rejectTool ---- diff --git a/packages/desktop/src/lib/frame-reducer.ts b/packages/desktop/src/lib/frame-reducer.ts index ac95d45..0e309ac 100644 --- a/packages/desktop/src/lib/frame-reducer.ts +++ b/packages/desktop/src/lib/frame-reducer.ts @@ -28,6 +28,8 @@ export interface StreamEffects { setCompacted(): void; syncTurnId(turnId: number): void; newId(): string; + /** steer 输入已被服务端投递（user_input 帧到达）：从队列区移除对应项 */ + onInputDelivered?(id: string): void; } function toResultItem(outcome: ToolOutcome): { output: string; exitCode: number } { @@ -125,6 +127,10 @@ export function reduceFrame(frame: Frame, state: StreamState, fx: StreamEffects) status: e.status === 'spawned' ? 'running' : e.status, }); return; + case 'user_input': + fx.applyItem({ id: `steer-${e.id}`, type: 'message', role: 'user', parts: [...e.parts] }); + fx.onInputDelivered?.(e.id); + return; case 'tool_result': { if (e.name === 'submit_plan' && e.outcome.status !== 'ok') state.planTitle = null; if (e.outcome.status === 'denied') { diff --git a/packages/desktop/src/stores/agent.store.ts b/packages/desktop/src/stores/agent.store.ts index 2e3695a..6b3f86d 100644 --- a/packages/desktop/src/stores/agent.store.ts +++ b/packages/desktop/src/stores/agent.store.ts @@ -24,6 +24,10 @@ function appendText(parts: ContentPart[], text: string): void { else parts.push({ type: 'text', text }); } +function randomId(): string { + return crypto.randomUUID ? crypto.randomUUID() : Math.random().toString(36).slice(2, 11); +} + interface TodoPanelState { items: TodoItem[]; hasSeenNonEmptyTodo: boolean; @@ -35,6 +39,13 @@ export interface PendingPlan { title: string; } +export interface QueuedInput { + id: string; + /** 本地形态：本地媒体带 dataUrl，提交时才转线上形状 */ + parts: ContentPart[]; + status: 'queued' | 'sending'; +} + export interface StoredProfile { activeProfile: ProfileName; permissionMode: PermissionMode; @@ -54,6 +65,7 @@ interface AgentState { todoByThreadId: Record; pendingInput: string | null; pendingPlanByThreadId: Record; + queuedInputsByThreadId: Record; usageByThreadId: Record; profileByThreadId: Record; isCompressing: boolean; @@ -112,6 +124,10 @@ interface AgentActions { updateTurnId: (threadId: string, oldTurnId: string, newTurnId: string) => void; completeTurn: (threadId: string, turnId: string, status: 'completed' | 'error') => void; setPendingInput: (input: string | null) => void; + queueInput: (threadId: string, parts: ContentPart[]) => void; + removeQueuedInput: (threadId: string, id: string) => void; + markQueueSending: (threadId: string, id: string) => void; + reorderQueuedInputs: (threadId: string, from: number, to: number) => void; clearRunningTurns: (threadId: string) => void; applyTodoUpdate: (threadId: string, items: TodoItem[]) => void; toggleTodoCollapsed: (threadId: string) => void; @@ -133,6 +149,7 @@ export const useAgentStore = create()( todoByThreadId: {}, pendingInput: null, pendingPlanByThreadId: {}, + queuedInputsByThreadId: {}, usageByThreadId: {}, profileByThreadId: {}, isCompressing: false, @@ -211,6 +228,7 @@ export const useAgentStore = create()( delete s.todoByThreadId[id]; delete s.pendingPlanByThreadId[id]; delete s.profileByThreadId[id]; + delete s.queuedInputsByThreadId[id]; }), upsertThread: (thread) => @@ -315,6 +333,11 @@ export const useAgentStore = create()( delete s.profileByThreadId[id]; } } + for (const id of Object.keys(s.queuedInputsByThreadId)) { + if (!incomingIds.has(id)) { + delete s.queuedInputsByThreadId[id]; + } + } }); useRollbackStore.getState().cleanupDeletedThreads(incomingIds); }, @@ -471,6 +494,35 @@ export const useAgentStore = create()( s.pendingInput = input; }), + queueInput: (threadId, parts) => + set((s) => { + const list = s.queuedInputsByThreadId[threadId] ?? []; + list.push({ id: randomId(), parts, status: 'queued' }); + s.queuedInputsByThreadId[threadId] = list; + }), + + removeQueuedInput: (threadId, id) => + set((s) => { + const list = s.queuedInputsByThreadId[threadId]; + if (!list) return; + s.queuedInputsByThreadId[threadId] = list.filter((q) => q.id !== id); + }), + + markQueueSending: (threadId, id) => + set((s) => { + const item = s.queuedInputsByThreadId[threadId]?.find((q) => q.id === id); + if (item) item.status = 'sending'; + }), + + reorderQueuedInputs: (threadId, from, to) => + set((s) => { + const list = s.queuedInputsByThreadId[threadId]; + if (!list) return; + if (from < 0 || from >= list.length || to < 0 || to >= list.length || from === to) return; + const [moved] = list.splice(from, 1); + if (moved) list.splice(to, 0, moved); + }), + setPendingPlan: (threadId, plan) => set((s) => { if (plan === null) { diff --git a/packages/desktop/test/input-box-plan-button.test.tsx b/packages/desktop/test/input-box-plan-button.test.tsx index 5b81eca..2ad8c32 100644 --- a/packages/desktop/test/input-box-plan-button.test.tsx +++ b/packages/desktop/test/input-box-plan-button.test.tsx @@ -10,7 +10,7 @@ describe('Desktop: InputBox "查看计划" button', () => { const agentWorkspaceSource = sourceContent('agent/AgentWorkspace.tsx'); it('imports FileText from lucide-react', () => { - expect(agentWorkspaceSource).toMatch(/FileText.*from\s+['"]lucide-react['"]/); + expect(agentWorkspaceSource).toMatch(/FileText[\s\S]*?from\s+['"]lucide-react['"]/); }); it('derives planExists from the agent store', () => { diff --git a/packages/desktop/test/selector-stable-reference.test.ts b/packages/desktop/test/selector-stable-reference.test.ts new file mode 100644 index 0000000..9dd6ada --- /dev/null +++ b/packages/desktop/test/selector-stable-reference.test.ts @@ -0,0 +1,85 @@ +/** + * @vitest-environment jsdom + * + * 回归：React「Maximum update depth exceeded」自激更新循环。 + * + * 根因：zustand selector 里出现 `?? []` / `: []` 这类字面量 fallback 时， + * 每次 render 都会返回一个【新数组引用】。zustand 默认用 Object.is 比较 + * selector 的返回值，于是每次 render 都判定「值变了」→ 触发订阅更新 → + * 又 render → 又新引用 …… 形成无限循环。 + * + * 修法：用 `useShallow` 包住 selector，按【元素引用】浅比较。 + * 本文件从两个层面锁住： + * 1. 源码层面 —— 相关 selector 确实被 useShallow 包裹（防回退）； + * 2. 语义层面 —— 空数组 fallback 在浅比较下判为「相等」，不再触发更新。 + */ +import { describe, it, expect } from 'vitest'; +import { readFileSync } from 'fs'; +import { resolve } from 'path'; + +function sourceOf(relativePath: string): string { + return readFileSync(resolve(__dirname, '..', 'src', relativePath), 'utf-8'); +} + +describe('zustand selector 自激更新循环回归', () => { + describe('源码层面：文本量 fallback 的 selector 必须被 useShallow 包裹', () => { + const cases: Array<{ file: string; selectorMarker: RegExp; label: string }> = [ + { + file: 'agent/AgentWorkspace.tsx', + selectorMarker: /queuedInputsByThreadId\[s\.currentThreadId\]\s*\?\?\s*\[\]/, + label: 'AgentWorkspace 的 queuedInputs', + }, + { + file: 'agent/MessageStream.tsx', + selectorMarker: /s\.threads\[threadId\]\?\.turns\s*\?\?\s*\[\]/, + label: 'MessageStream 的 turns', + }, + ]; + + for (const { file, selectorMarker, label } of cases) { + it(`${label} 使用 ?? [] 且已被 useShallow 包裹`, () => { + const src = sourceOf(file); + // 前提：确实是「字面量 fallback」型 selector（否则本用例失去意义） + expect(src).toMatch(selectorMarker); + // 约束：该 selector 被 useShallow 包裹 + expect(src).toMatch(/useShallow\(\s*\(s\)\s*=>/); + // 约束：从 zustand/react/shallow 引入 + expect(src).toMatch(/from\s+['"]zustand\/react\/shallow['"]/); + }); + } + }); + + describe('语义层面：浅比较让「内容相同的空数组」判等', () => { + // 复刻 zustand 默认比较（Object.is） + const defaultEquals = Object.is; + // 复刻 useShallow 的比较语义：逐元素 === + function shallowEqualsArray(a: readonly unknown[], b: readonly unknown[]): boolean { + if (a === b) return true; + if (a.length !== b.length) return false; + for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false; + return true; + } + + it('两个独立的 [] 字面量在 Object.is 下【不相等】—— 这正是循环的起因', () => { + expect(defaultEquals([], [])).toBe(false); + }); + + it('两个独立的 [] 字面量在浅比较下【相等】—— 不再触发更新', () => { + expect(shallowEqualsArray([], [])).toBe(true); + }); + + it('内容相同但引用不同的元素数组，浅比较判等', () => { + const item = { id: 'q1', status: 'queued' }; + const a = [item]; + const b = [item]; // 不同数组，同一元素引用 + expect(defaultEquals(a, b)).toBe(false); + expect(shallowEqualsArray(a, b)).toBe(true); + }); + + it('元素发生真实增删时，浅比较仍能判出不相等（不会漏更新）', () => { + const item = { id: 'q1', status: 'queued' }; + expect(shallowEqualsArray([item], [])).toBe(false); + expect(shallowEqualsArray([], [item])).toBe(false); + }); + }); +}); diff --git a/packages/sdk/src/contracts.ts b/packages/sdk/src/contracts.ts index 0f8fa30..4c0454e 100644 --- a/packages/sdk/src/contracts.ts +++ b/packages/sdk/src/contracts.ts @@ -131,17 +131,28 @@ export type InputPart = export const MAX_MEDIA_BYTES = 10 * 1024 * 1024; export interface AgentRuntimeClient { - sendMessage( + /** + * 提交一轮输入。 + * + * 命中活跃回合时服务端把输入并入当前回合（202），返回 `{ kind: 'queued' }`； + * 否则开新回合并返回 `{ kind: 'turn', stream }`。 + */ + submitInput( input: InputPart[], options: { sessionId?: string; + /** 前端队列项 id：命中活跃回合时原样回显在 user_input 帧上 */ + inputId?: string; cwd: string; model?: string; signal?: AbortSignal; /** 用户在输入框显式 @ 的 skill，name 供对齐、path 为唯一查找键 */ skills?: Array<{ name: string; path: string }>; } - ): AsyncGenerator; + ): Promise< + | { kind: 'turn'; stream: AsyncGenerator } + | { kind: 'queued'; sessionId: string; turnId: number } + >; sendApprovalResponse(input: { sessionId: string; diff --git a/packages/sdk/src/decode.ts b/packages/sdk/src/decode.ts index 9b3e0eb..d9fe985 100644 --- a/packages/sdk/src/decode.ts +++ b/packages/sdk/src/decode.ts @@ -11,7 +11,14 @@ export type DecodeResult = | { readonly ok: false; readonly reason: DecodeFailureReason; readonly raw: unknown }; const TRANSITIONS = new Set(['start', 'executing', 'compress', 'end']); -const EVENTS = new Set(['text_delta', 'tool_call', 'tool_result', 'approval_request', 'subagent_event']); +const EVENTS = new Set([ + 'text_delta', + 'tool_call', + 'tool_result', + 'approval_request', + 'subagent_event', + 'user_input', +]); const END_REASONS = new Set(['done', 'error', 'maxSteps', 'aborted']); const OUTCOMES = new Set(['ok', 'error', 'denied']); @@ -100,6 +107,11 @@ export function decodeFrame(raw: unknown): DecodeResult { return { ok: false, reason: 'shape', raw }; } } + if (e.type === 'user_input') { + if (typeof e.id !== 'string' || !Array.isArray(e.parts)) { + return { ok: false, reason: 'shape', raw }; + } + } return { ok: true, frame: raw as unknown as Frame }; } diff --git a/packages/sdk/src/http/agent-runtime.ts b/packages/sdk/src/http/agent-runtime.ts index 9cd7fff..1f9e380 100644 --- a/packages/sdk/src/http/agent-runtime.ts +++ b/packages/sdk/src/http/agent-runtime.ts @@ -12,11 +12,11 @@ export function createHttpAgentClient( const { apiPost } = request; return { - async *sendMessage(input, { sessionId, cwd, model, signal, skills }) { + async submitInput(input, { sessionId, inputId, cwd, model, signal, skills }) { const path = `/api/sessions/${sessionId || '_'}/messages`; const response = await fetch(`${baseUrl}${path}`, { method: 'POST', - body: JSON.stringify({ input, cwd, model, skills }), + body: JSON.stringify({ input, inputId, cwd, model, skills }), headers: { 'Content-Type': 'application/json' }, signal, }); @@ -24,14 +24,24 @@ export function createHttpAgentClient( throw new ApiError(response.status, path, await parseErrorBody(response)); } - for await (const data of parseSseStream(response)) { - const decoded = decodeFrame(data); - if (!decoded.ok) { - console.warn(`[agent-runtime] dropped frame (${decoded.reason})`, decoded.raw); - continue; - } - yield decoded.frame; + // 202：输入已并入活跃回合，没有流可读 + const contentType = response.headers.get('content-type') ?? ''; + if (!contentType.includes('text/event-stream')) { + const body = (await response.json()) as { sessionId: string; turnId: number }; + return { kind: 'queued', sessionId: body.sessionId, turnId: body.turnId }; } + + const stream = (async function* () { + for await (const data of parseSseStream(response)) { + const decoded = decodeFrame(data); + if (!decoded.ok) { + console.warn(`[agent-runtime] dropped frame (${decoded.reason})`, decoded.raw); + continue; + } + yield decoded.frame; + } + })(); + return { kind: 'turn', stream }; }, async sendApprovalResponse({ sessionId, approvalId, response }) { diff --git a/packages/sdk/src/protocol.ts b/packages/sdk/src/protocol.ts index baf2693..5a359ba 100644 --- a/packages/sdk/src/protocol.ts +++ b/packages/sdk/src/protocol.ts @@ -1,4 +1,5 @@ import type { TodoItem, TokenUsage } from './types.js'; +import type { StoredContentPart } from './dto.js'; export type EndReason = 'done' | 'error' | 'maxSteps' | 'aborted'; @@ -58,6 +59,11 @@ export type RuntimeEvent = readonly sessionId: string; readonly agentName: string; readonly status: 'spawned' | 'completed' | 'failed'; + } + | { + readonly type: 'user_input'; + readonly id: string; + readonly parts: readonly StoredContentPart[]; }; export interface Fatal { diff --git a/packages/sdk/test/contracts.test.ts b/packages/sdk/test/contracts.test.ts index ccacf54..7dcb845 100644 --- a/packages/sdk/test/contracts.test.ts +++ b/packages/sdk/test/contracts.test.ts @@ -34,11 +34,11 @@ type _ForkTurnsIsUITurn = ForkResult['turns'] extends UITurn[] ? true : never; type _ContextTurnsIsUITurn = RollbackContextResult['turns'] extends UITurn[] ? true : never; describe('client contracts', () => { - it('http agent 只暴露 sendMessage / sendApprovalResponse / compact', () => { + it('http agent 只暴露 submitInput / sendApprovalResponse / compact', () => { expect(Object.keys(httpAgent).sort()).toEqual([ 'compact', 'sendApprovalResponse', - 'sendMessage', + 'submitInput', ]); }); diff --git a/packages/sdk/test/http/agent-runtime.test.ts b/packages/sdk/test/http/agent-runtime.test.ts index d9c0dc9..7a64ebb 100644 --- a/packages/sdk/test/http/agent-runtime.test.ts +++ b/packages/sdk/test/http/agent-runtime.test.ts @@ -54,15 +54,19 @@ const STREAM: Frame[] = [ { ...ENVELOPE, seq: 7, family: 'transition', transition: { to: 'end', reason: 'done' } }, ]; -describe('createHttpAgentClient.sendMessage', () => { +describe('createHttpAgentClient.submitInput', () => { it('decodes frame envelopes from the SSE stream', async () => { const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(createSseResponse(STREAM)); const request = createRequestHelpers('http://localhost:8080'); const client = createHttpAgentClient('http://localhost:8080', request); + const result = await client.submitInput(input('hi'), { sessionId: 'sess-123', cwd: '/tmp' }); + expect(result.kind).toBe('turn'); + if (result.kind !== 'turn') throw new Error('expected turn'); + const frames: Frame[] = []; - for await (const frame of client.sendMessage(input('hi'), { sessionId: 'sess-123', cwd: '/tmp' })) { + for await (const frame of result.stream) { frames.push(frame); } @@ -90,8 +94,11 @@ describe('createHttpAgentClient.sendMessage', () => { const request = createRequestHelpers('http://localhost:8080'); const client = createHttpAgentClient('http://localhost:8080', request); + const result = await client.submitInput(input('hi'), { sessionId: 'sess-123', cwd: '/tmp' }); + if (result.kind !== 'turn') throw new Error('expected turn'); + const frames: Frame[] = []; - for await (const frame of client.sendMessage(input('hi'), { sessionId: 'sess-123', cwd: '/tmp' })) { + for await (const frame of result.stream) { frames.push(frame); } @@ -111,8 +118,11 @@ describe('createHttpAgentClient.sendMessage', () => { const request = createRequestHelpers('http://localhost:8080'); const client = createHttpAgentClient('http://localhost:8080', request); + const result = await client.submitInput(input('hi'), { cwd: '/tmp' }); + if (result.kind !== 'turn') throw new Error('expected turn'); + const frames: Frame[] = []; - for await (const frame of client.sendMessage(input('hi'), { cwd: '/tmp' })) { + for await (const frame of result.stream) { frames.push(frame); } @@ -142,8 +152,11 @@ describe('createHttpAgentClient.sendMessage', () => { const request = createRequestHelpers('http://localhost:8080'); const client = createHttpAgentClient('http://localhost:8080', request); + const result = await client.submitInput(input('hi'), { sessionId: 's', cwd: '/tmp' }); + if (result.kind !== 'turn') throw new Error('expected turn'); + const frames: Frame[] = []; - for await (const c of client.sendMessage(input('hi'), { sessionId: 's', cwd: '/tmp' })) { + for await (const c of result.stream) { frames.push(c); }