增加回合中用户输入功能

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<FrameBody>; sessionId: string }
+  | { kind: 'queued'; sessionId: string; turnId: number };
+
 export interface AgentShape {
-  runTurn(
-    input: IncomingPart[],
-    opts: RunTurnOptions
-  ): Effect.Effect<
-    {
-      stream: AsyncGenerator<FrameBody>;
-      sessionId: string;
-    },
-    AgentError
-  >;
+  runTurn(input: IncomingPart[], opts: RunTurnOptions): Effect.Effect<RunTurnResult, AgentError>;
 }
 
 export class AgentService extends Context.Tag('AgentService')<AgentService, AgentShape>() {}

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<string, ContextBuffer>();
@@ -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);
       });

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,14 +32,17 @@ 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))
 );
 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
 );

packages/codingcode/src/scheduler/scheduler.ts

@@ -48,7 +48,7 @@
       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,11 +61,14 @@
             });
           })
         );
+        // 自动化不开已有会话，不可能命中活跃回合
+        if (turnResult.kind === 'queued') return;
+        const { stream, sessionId } = turnResult;
 
         let lastContent = '';
         for await (const body of stream) {
           if (body.family === 'event' && body.event.type === 'text_delta') {
             lastContent += body.event.text;
           } else if (
             body.family === 'transition' &&
             body.transition.to === 'end' &&
@@ -181,7 +184,7 @@
         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 @@
               });
             })
           );
+          // 同上：自动化不开已有会话，queued 不可达
+          if (turnResult.kind === 'queued') return null;
+          const { stream, sessionId } = turnResult;
 
           for await (const body of stream) {
             if (

packages/codingcode/src/server/http-error.ts

@@ -21,6 +21,7 @@ const DOMAIN_STATUS: Record<ErrorCode, number> = {
   EMPTY_RESPONSE: 500,
   SESSION_IO_ERROR: 500,
   INVALID_INPUT: 400,
+  TURN_CONFLICT: 409,
 };
 
 export class NotFoundError extends Error {

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 =>

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 });
 });
 

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 });
 });
 

packages/codingcode/src/session/port.ts

@@ -48,6 +48,11 @@ export interface SessionShape {
   ): Effect.Effect<Map<string, string>, AgentError>;
 
   recordUser(state: SessionStoreState, content: StoredPart[]): Effect.Effect<UserEvent, AgentError>;
+  /** steer 输入：归属当前回合，不递增 turnId、不改标题 */
+  recordUserInput(
+    state: SessionStoreState,
+    content: StoredPart[]
+  ): Effect.Effect<UserEvent, AgentError>;
   recordSystem(
     state: SessionStoreState,
     content: StoredPart[]

packages/codingcode/src/session/session.ts

@@ -191,7 +191,7 @@
 
 export const SessionLayer = Layer.effect(
   SessionService,
   Effect.gen(function* () {
     const assetCache = new Map<string, string>();
 
     const create = (
@@ -260,11 +260,12 @@
 
     const recordUser = (
       state: SessionStoreState,
-      content: StoredPart[]
+      content: StoredPart[],
+      opts?: { steering?: boolean }
     ): Effect.Effect<UserEvent, AgentError> =>
       Effect.try({
         try: () => {
-          state.currentTurnId += 1;
+          if (!opts?.steering) state.currentTurnId += 1;
           const event: UserEvent = {
             type: 'user',
             turnId: state.currentTurnId,
@@ -288,6 +289,11 @@
             : new AgentError('SESSION_IO_ERROR', `Session write failed: ${String(e)}`, e),
       });
 
+    const recordUserInput = (
+      state: SessionStoreState,
+      content: StoredPart[]
+    ): Effect.Effect<UserEvent, AgentError> => recordUser(state, content, { steering: true });
+
     const recordSystem = (
       state: SessionStoreState,
       content: StoredPart[]
@@ -601,6 +607,7 @@
       materializeInput,
       resolveAssets,
       recordUser,
+      recordUserInput,
       recordSystem,
       recordAssistant,
       recordToolResult,

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 {

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;