Merge pull request #140 from phantom5099/steer-userinput

增加回合中用户输入功能

Author: phantom5099

packages/codingcode/src/agent/agent.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/agent/port.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/context/context.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/layer.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 (

packages/codingcode/src/scheduler/scheduler.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/http-error.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/messages.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/sessions.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/server/routes/subagents.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[]