feat: fork a session from an earlier prompt

.changeset/fork-at-turn.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": minor
+---
+
+The `:fork` server endpoint accepts a `prompt_id` to fork the session from an earlier prompt instead of the latest state.

docs/en/reference/server-api.md

@@ -708,16 +708,20 @@ Session actions are dispatched through one route: the path tail is parsed as `{s
 
 #### `POST /api/v1/sessions/{session_id}:fork`
 
-Copies the session — its transcript, agent state, and files — into a new session in the same workspace, and broadcasts `event.session.created`. Forking is rejected while any of the session's agents has an active turn.
+Copies the session — its transcript, agent state, and files — into a new session in the same workspace, and broadcasts `event.session.created`. Pass `prompt_id` to fork from an earlier point instead of the latest state: the copy keeps only the conversation through that prompt's turn segment and discards everything after it.
+
+Forking is rejected while the session has an active or queued turn, or while another operation (compaction, undo, another fork) is in progress; retry once the session is idle.
 
 | Parameter | In | Type | Description |
 | --- | --- | --- | --- |
 | `title` | body | string | Title for the fork (at least 1 character). Default `Fork: <source title>` |
 | `metadata` | body | object | Custom metadata for the fork |
+| `prompt_id` | body | string | Id of the user prompt to fork from. Default: the latest state |
 
 On success, `data` is [the session object](#the-session-object) of the new session.
 
-- `40901`: the session has an active turn and cannot be forked
+- `40001`: `prompt_id` is unknown in the session
+- `40901`: a turn is active or another operation is in progress, so the session cannot be forked right now
 
 #### `POST /api/v1/sessions/{session_id}:compact`
 
@@ -787,7 +791,7 @@ On success, `data` is `{ items, has_more }` where each item is [the session obje
 
 #### `POST /api/v1/sessions/{session_id}/children`
 
-Creates a child session: a fork of this session recorded as its child, so it shows up under `GET /api/v1/sessions/{session_id}/children`. The same active-turn restriction as `:fork` applies.
+Creates a child session: a fork of this session recorded as its child, so it shows up under `GET /api/v1/sessions/{session_id}/children`. The same rules as `:fork` apply: it is rejected while a turn is active or another operation is in progress.
 
 | Parameter | In | Type | Description |
 | --- | --- | --- | --- |
@@ -797,7 +801,7 @@ Creates a child session: a fork of this session recorded as its child, so it sho
 
 On success, `data` is [the session object](#the-session-object) of the new session, and the server broadcasts `event.session.created`.
 
-- `40901`: the session has an active turn and cannot be forked
+- `40901`: a turn is active or another operation is in progress, so the session cannot be forked right now
 
 #### `GET /api/v1/sessions/{session_id}/status`
 

docs/zh/reference/server-api.md

@@ -708,16 +708,20 @@ schema 还接受 `agent_config` 内的 `system_prompt`、`tools`、`mcp_servers`
 
 #### `POST /api/v1/sessions/{session_id}:fork`
 
-将会话——其转录、Agent 状态与文件——复制到同一工作区中的新会话，并广播 `event.session.created`。当会话中任一 Agent 有进行中的轮次时，fork 会被拒绝。
+将会话——其转录、Agent 状态与文件——复制到同一工作区中的新会话，并广播 `event.session.created`。传入 `prompt_id` 可从更早的位置分叉，而不是复制最新状态：副本只保留到该提示词所在轮次段结束为止的对话，其后的内容全部丢弃。
+
+会话有进行中或排队中的轮次，或者有其他操作（压缩、撤销、另一个 fork）在进行时，fork 会被拒绝；等会话空闲后重试。
 
 | 参数 | 位置 | 类型 | 说明 |
 | --- | --- | --- | --- |
 | `title` | body | string | fork 的标题（至少 1 个字符）。默认 `Fork: <source title>` |
 | `metadata` | body | object | fork 的自定义元数据 |
+| `prompt_id` | body | string | 要从中分叉的用户提示词 id。默认为最新状态 |
 
 成功时，`data` 为新会话的 [session 对象](#session-对象)。
 
-- `40901`：会话有进行中的轮次，无法 fork
+- `40001`：`prompt_id` 在会话中不存在
+- `40901`：有轮次正在进行或有其他操作在进行，暂时无法 fork
 
 #### `POST /api/v1/sessions/{session_id}:compact`
 
@@ -787,7 +791,7 @@ schema 还接受 `agent_config` 内的 `system_prompt`、`tools`、`mcp_servers`
 
 #### `POST /api/v1/sessions/{session_id}/children`
 
-创建子会话：fork 当前会话并记录为其子会话，因此会出现在 `GET /api/v1/sessions/{session_id}/children` 下。适用与 `:fork` 相同的进行中轮次限制。
+创建子会话：fork 当前会话并记录为其子会话，因此会出现在 `GET /api/v1/sessions/{session_id}/children` 下。适用与 `:fork` 相同的规则：有轮次正在进行或有其他操作在进行时会被拒绝。
 
 | 参数 | 位置 | 类型 | 说明 |
 | --- | --- | --- | --- |
@@ -797,7 +801,7 @@ schema 还接受 `agent_config` 内的 `system_prompt`、`tools`、`mcp_servers`
 
 成功时，`data` 为新会话的 [session 对象](#session-对象)，并且服务端广播 `event.session.created`。
 
-- `40901`：会话有进行中的轮次，无法 fork
+- `40901`：有轮次正在进行或有其他操作在进行，暂时无法 fork
 
 #### `GET /api/v1/sessions/{session_id}/status`
 

packages/agent-core-v2/src/agent/loop/loopService.ts

@@ -151,6 +151,8 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
   private machineTurnSuppressed = false;
   private readonly settleWaiters: Array<() => void> = [];
   private quiescenceDepth = 0;
+  private pendingTurnCloses = 0;
+  private pendingFinalizations = 0;
   private activeRequestTrace: LLMRequestTrace | undefined;
   private engine: MachineEngine | undefined;
 
@@ -768,7 +770,8 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
       this.quiescenceDepth > 0 ||
       this.active !== undefined ||
       this.hasPendingRequests() ||
-      this.pendingMachineTurn !== undefined
+      this.pendingMachineTurn !== undefined ||
+      this.pendingTurnCloses > 0
     ) {
       return undefined;
     }
@@ -911,7 +914,8 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
     if (
       this.active === undefined &&
       !this.hasPendingRequests() &&
-      this.pendingMachineTurn === undefined
+      this.pendingMachineTurn === undefined &&
+      this.pendingFinalizations === 0
     ) {
       return Promise.resolve();
     }
@@ -924,6 +928,7 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
     if (
       this.active !== undefined ||
       this.pendingMachineTurn !== undefined ||
+      this.pendingFinalizations > 0 ||
       this.hasPendingRequests()
     ) return;
     if (this.settleWaiters.length === 0) return;
@@ -2030,75 +2035,86 @@ export class AgentLoopService extends Disposable implements IAgentLoopService {
   private async endTurn(turn: ActiveTurn, result: TurnResult): Promise<void> {
     if (this.active !== turn) return;
     this.active = undefined;
-    await this.wire.drainPersisted().catch(() => undefined);
-    for (const nudge of this.nudges.slice(this.nudgeCursor)) {
-      if (nudge.turnScoped && !nudge.dropped) {
-        nudge.dropped = true;
-        nudge.onDrop?.();
+    this.pendingTurnCloses += 1;
+    this.pendingFinalizations += 1;
+    try {
+      await this.wire.drainPersisted().catch(() => undefined);
+      for (const nudge of this.nudges.slice(this.nudgeCursor)) {
+        if (nudge.turnScoped && !nudge.dropped) {
+          nudge.dropped = true;
+          nudge.onDrop?.();
+        }
       }
-    }
-    turn.turn.state = result.type;
-    if (!turn.readyResolved) {
-      if (result.type === 'failed') {
-        turn.ready.reject(result.error);
-      } else if (result.type === 'cancelled') {
-        turn.ready.reject(
-          result.reason instanceof Error ? result.reason : abortError('Turn cancelled'),
-        );
-      } else {
-        turn.ready.reject(new Error2(ErrorCodes.INTERNAL, 'Turn ended before first step'));
+      turn.turn.state = result.type;
+      if (!turn.readyResolved) {
+        if (result.type === 'failed') {
+          turn.ready.reject(result.error);
+        } else if (result.type === 'cancelled') {
+          turn.ready.reject(
+            result.reason instanceof Error ? result.reason : abortError('Turn cancelled'),
+          );
+        } else {
+          turn.ready.reject(new Error2(ErrorCodes.INTERNAL, 'Turn ended before first step'));
+        }
       }
-    }
-    const durationMs = Date.now() - turn.startedAt;
-    const traceId =
-      result.type === 'completed' ? this.lastRequestTraceId : this.activeRequestTrace?.traceId;
-    const error = result.type === 'failed' ? toKimiErrorPayload(result.error) : undefined;
-    const interruptReason = result.type === 'completed' ? undefined : interruptReasonFor(result);
-    void this.dispatcher.dispatch(
-      new TurnEnded({
-        agentId: this.scopeContext.agentId,
-        turnId: turn.id,
-        reason: result.type,
-        error,
-        durationMs,
-        interruptReason,
-        stopReason: result.type === 'completed' ? result.stopReason : undefined,
-        traceId,
-      }),
-    );
-    if (error !== undefined) {
+      const durationMs = Date.now() - turn.startedAt;
+      const traceId =
+        result.type === 'completed' ? this.lastRequestTraceId : this.activeRequestTrace?.traceId;
+      const error = result.type === 'failed' ? toKimiErrorPayload(result.error) : undefined;
+      const interruptReason = result.type === 'completed' ? undefined : interruptReasonFor(result);
       void this.dispatcher.dispatch(
-        new AgentErrorEvent({ ...error, agentId: this.scopeContext.agentId }),
+        new TurnEnded({
+          agentId: this.scopeContext.agentId,
+          turnId: turn.id,
+          reason: result.type,
+          error,
+          durationMs,
+          interruptReason,
+          stopReason: result.type === 'completed' ? result.stopReason : undefined,
+          traceId,
+        }),
       );
-    }
-    if (interruptReason !== undefined) {
-      const interrupted: TurnInterruptedEvent = {
+      if (error !== undefined) {
+        void this.dispatcher.dispatch(
+          new AgentErrorEvent({ ...error, agentId: this.scopeContext.agentId }),
+        );
+      }
+      if (interruptReason !== undefined) {
+        const interrupted: TurnInterruptedEvent = {
+          turn_id: turn.id,
+          at_step: result.steps,
+          mode: turn.mode ?? 'agent',
+          interrupt_reason: interruptReason,
+          provider_type: turn.providerType,
+          protocol: turn.protocol,
+          trace_id: traceId,
+        };
+        this.telemetry.track2('turn_interrupted', interrupted);
+      }
+      const ended: TurnEndedTelemetryEvent = {
         turn_id: turn.id,
-        at_step: result.steps,
+        reason: result.type,
+        duration_ms: durationMs,
         mode: turn.mode ?? 'agent',
-        interrupt_reason: interruptReason,
+        error_type: error?.code,
         provider_type: turn.providerType,
         protocol: turn.protocol,
         trace_id: traceId,
+        enabled_plugins: this.plugins.enabledPluginIds()?.join(','),
       };
-      this.telemetry.track2('turn_interrupted', interrupted);
-    }
-    const ended: TurnEndedTelemetryEvent = {
-      turn_id: turn.id,
-      reason: result.type,
-      duration_ms: durationMs,
-      mode: turn.mode ?? 'agent',
-      error_type: error?.code,
-      provider_type: turn.providerType,
-      protocol: turn.protocol,
-      trace_id: traceId,
-      enabled_plugins: this.plugins.enabledPluginIds()?.join(','),
-    };
-    this.telemetry.track2('turn_ended', ended);
-    this.telemetry.setContext({ turn_id: undefined, trace_id: undefined, thinking_effort: undefined });
-    this.activeRequestTrace = undefined;
-    this.lastRequestTraceId = undefined;
-    turn.result.resolve(result);
+      this.telemetry.track2('turn_ended', ended);
+      this.telemetry.setContext({ turn_id: undefined, trace_id: undefined, thinking_effort: undefined });
+      this.activeRequestTrace = undefined;
+      this.lastRequestTraceId = undefined;
+    } finally {
+      this.pendingTurnCloses -= 1;
+    }
+    try {
+      await this.dispatcher.flush().catch(() => undefined);
+      turn.result.resolve(result);
+    } finally {
+      this.pendingFinalizations -= 1;
+    }
     this.maybeSettle();
   }
 

packages/agent-core-v2/src/workspace/sessionLifecycle/internal/forkTurnSlice.ts

@@ -11,6 +11,7 @@ export interface MainTurnSlice {
   readonly records: readonly WireRecord[];
   readonly cutoffTime?: number;
   readonly lastPrompt?: string;
+  readonly lastTurnReason?: 'completed' | 'cancelled' | 'failed';
 }
 
 export function assertForkTurnIndex(turnIndex: number | undefined): void {
@@ -28,10 +29,7 @@ export function sliceMainRecordsAtTurn(
   sourceSessionId: string,
   turnIndex: number,
 ): MainTurnSlice {
-  const turnStarts: number[] = [];
-  for (let index = 0; index < records.length; index += 1) {
-    if (isUserVisibleTurnRecord(records[index]!)) turnStarts.push(index);
-  }
+  const turnStarts = userVisibleTurnStartIndices(records);
   const start = turnStarts[turnIndex];
   if (start === undefined) {
     throw new Error2(
@@ -58,6 +56,7 @@ export function sliceMainRecordsAtTurn(
     records: retained,
     cutoffTime: cutoffTimes.length === 0 ? undefined : Math.max(...cutoffTimes),
     lastPrompt,
+    lastTurnReason: turnOutcomeOfRecords(retained),
   };
 }
 
@@ -77,6 +76,45 @@ export function sliceSubagentRecordsAtTime(
   return records.slice(0, end);
 }
 
+export type ForkPromptResolution =
+  | { readonly status: 'found'; readonly index: number }
+  | { readonly status: 'unknown' }
+  | { readonly status: 'ambiguous' };
+
+export function resolveForkPromptIndex(
+  records: readonly WireRecord[],
+  promptId: string,
+): ForkPromptResolution {
+  const turnStarts = userVisibleTurnStartIndices(records);
+  const targetIds = new Set<string>([promptId]);
+  for (const record of records) {
+    if (record.type !== 'turn.steer') continue;
+    const promptIds = record['promptIds'];
+    if (!Array.isArray(promptIds) || !promptIds.includes(promptId)) continue;
+    const messageId = record['messageId'];
+    if (typeof messageId === 'string' && messageId.length > 0) targetIds.add(messageId);
+  }
+  let found: number | undefined;
+  let ambiguous = false;
+  for (let start = 0; start < turnStarts.length; start += 1) {
+    const message = asRecord(records[turnStarts[start]!]!['message']);
+    const id = message?.['id'];
+    if (typeof id !== 'string' || !targetIds.has(id)) continue;
+    if (found !== undefined && found !== start) ambiguous = true;
+    found = found ?? start;
+  }
+  if (ambiguous) return { status: 'ambiguous' };
+  return found === undefined ? { status: 'unknown' } : { status: 'found', index: found };
+}
+
+function userVisibleTurnStartIndices(records: readonly WireRecord[]): number[] {
+  const turnStarts: number[] = [];
+  for (let index = 0; index < records.length; index += 1) {
+    if (isUserVisibleTurnRecord(records[index]!)) turnStarts.push(index);
+  }
+  return turnStarts;
+}
+
 function isUserVisibleTurnRecord(record: WireRecord): boolean {
   if (record.type !== 'context.append_message') return false;
   const message = asRecord(record['message']);
@@ -203,6 +241,30 @@ function recordTime(record: WireRecord): number | undefined {
   return undefined;
 }
 
+function turnOutcomeOfRecords(
+  records: readonly WireRecord[],
+): 'completed' | 'cancelled' | 'failed' | undefined {
+  for (let index = records.length - 1; index >= 0; index -= 1) {
+    const record = records[index]!;
+    if (record.type === 'turn.ended') {
+      const reason = record['reason'];
+      if (reason === 'completed' || reason === 'cancelled') return reason;
+      if (reason === 'failed' || reason === 'blocked') return 'failed';
+      return undefined;
+    }
+    if (isEngineTurnStartRecord(record) || isUserVisibleTurnRecord(record)) return undefined;
+  }
+  return undefined;
+}
+
+function isEngineTurnStartRecord(record: WireRecord): boolean {
+  return (
+    record.type === 'turn.prompt' ||
+    record.type === 'turn.steer' ||
+    record.type === 'agent.turn.started'
+  );
+}
+
 function promptMetadataFromTurnRecord(record: WireRecord): string | undefined {
   if (record.type !== 'context.append_message') return undefined;
   const message = asRecord(record['message']);

packages/agent-core-v2/src/workspace/sessionLifecycle/sessionLifecycle.ts

@@ -23,6 +23,7 @@ export interface ForkSessionOptions {
   readonly title?: string;
   readonly metadata?: Record<string, unknown>;
   readonly turnIndex?: number;
+  readonly promptId?: string;
 }
 
 export interface ResumeSessionOptions {