diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts index 31cfb977f1..c4d553f94e 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadForkParams.ts @@ -21,6 +21,11 @@ export type ThreadForkParams = {threadId: string, /** path?: string | null, /** * [UNSTABLE] Fork after the specified historical turn (inclusive). * When omitted, the full thread history is copied. + * + * The value should come from `thread/read(includeTurns=true)` for the same source thread. + * Requests are rejected for legacy histories whose replayed turn ids are not stable, for + * in-progress turns, and for turns that do not yet contain both a user message and an agent + * response. */ forkAfterTurnId?: string | null, /** * Configuration overrides for the forked thread, if any. diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 83378c4f69..78e7e55516 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -1,3 +1,16 @@ +//! Reconstruct app-server turn history from persisted rollout items. +//! +//! This module translates the low-level rollout log into the higher-level `Turn` / `ThreadItem` +//! structures returned by app-server APIs. The same reducer is used both for replaying stored +//! rollouts and for incrementally rebuilding the current turn from live `EventMsg` values. +//! Callers must feed items in recorded order; reordering lifecycle events can silently attach work +//! to the wrong turn while still producing a superficially valid transcript. +//! +//! Modern rollouts persist explicit turn lifecycle events, which means replay can preserve stable +//! turn ids. Legacy rollouts may only contain message/tool events, so replay has to synthesize +//! turn ids while rebuilding them. Consumers that need durable turn anchors, such as turn-based +//! `thread/fork`, should reject those synthesized ids rather than treating them as stable. + use crate::protocol::v2::CollabAgentState; use crate::protocol::v2::CollabAgentTool; use crate::protocol::v2::CollabAgentToolCallStatus; @@ -54,8 +67,14 @@ use codex_protocol::protocol::ExecCommandStatus as CoreExecCommandStatus; #[cfg(test)] use codex_protocol::protocol::PatchApplyStatus as CorePatchApplyStatus; +/// The replay result produced by [`build_thread_history_from_rollout_items`]. +/// +/// `turns` contains the reconstructed app-server view of the rollout. `has_synthetic_turn_ids` +/// tells callers whether any of those turns only have replay-generated ids, which means those ids +/// are not stable enough to reuse as external anchors across repeated reads. #[derive(Debug, Clone, PartialEq)] pub struct ThreadHistoryBuildResult { + /// Reconstructed turns in replay order. pub turns: Vec, /// True when any turn id had to be synthesized during replay because the /// rollout history lacked explicit turn lifecycle events. @@ -66,7 +85,9 @@ pub struct ThreadHistoryBuildResult { /// annotate whether any turn ids were synthesized during replay. /// /// When available, this uses `TurnContext.turn_id` as the canonical turn id so -/// resumed/rebuilt thread history preserves the original turn identifiers. +/// resumed/rebuilt thread history preserves the original turn identifiers. Callers that need a +/// stable external anchor should prefer this over [`build_turns_from_rollout_items`]; otherwise it +/// is easy to treat a replay-generated id as durable and later fail to find the same turn again. pub fn build_thread_history_from_rollout_items(items: &[RolloutItem]) -> ThreadHistoryBuildResult { let mut builder = ThreadHistoryBuilder::new(); for item in items { @@ -78,11 +99,20 @@ pub fn build_thread_history_from_rollout_items(items: &[RolloutItem]) -> ThreadH /// Convert persisted [`RolloutItem`] entries into a sequence of [`Turn`] values. /// /// When available, this uses `TurnContext.turn_id` as the canonical turn id so -/// resumed/rebuilt thread history preserves the original turn identifiers. +/// resumed/rebuilt thread history preserves the original turn identifiers. Use +/// [`build_thread_history_from_rollout_items`] instead when the caller must distinguish between +/// stable and replay-synthesized turn ids. pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { build_thread_history_from_rollout_items(items).turns } +/// Stateful reducer that groups rollout lifecycle items into app-server turns. +/// +/// The builder owns the in-progress turn boundary while replay is in flight, then emits the +/// finalized `Turn` list once all items have been applied. It intentionally exposes both replay +/// metadata and an active-turn snapshot so app-server can share the same reduction rules for +/// stored history and live notifications. Reusing a builder across unrelated transcripts without +/// calling [`Self::reset`] can leak late events from one transcript into the next. pub struct ThreadHistoryBuilder { turns: Vec, current_turn: Option, @@ -97,6 +127,11 @@ impl Default for ThreadHistoryBuilder { } impl ThreadHistoryBuilder { + /// Create an empty reducer with no active turn. + /// + /// The builder expects to receive the rollout from the beginning of a transcript, in order. + /// Starting midway through a rollout can still produce `Turn` values, but turn boundaries and + /// synthetic-id detection may no longer describe the actual persisted history. pub fn new() -> Self { Self { turns: Vec::new(), @@ -106,14 +141,25 @@ impl ThreadHistoryBuilder { } } + /// Clear all accumulated turns and return the builder to its initial state. pub fn reset(&mut self) { *self = Self::new(); } + /// Finalize replay and return only the reconstructed turns. + /// + /// This is the convenience API for callers that only need the `Turn` list. If a caller needs + /// to know whether any ids were synthesized, using this method would hide that fact and could + /// lead to invalid assumptions about turn-id stability. pub fn finish(self) -> Vec { self.finish_result().turns } + /// Finalize replay and return both the turns and replay metadata. + /// + /// This consumes the builder, closes any active turn, and reports whether replay had to mint + /// any replacement ids. Consumers that derive new API contracts from turn ids, such as + /// turn-based forking, should use this method so they can reject legacy rollouts explicitly. pub fn finish_result(mut self) -> ThreadHistoryBuildResult { self.finish_current_turn(); ThreadHistoryBuildResult { @@ -122,6 +168,12 @@ impl ThreadHistoryBuilder { } } + /// Return the current turn view without consuming the builder. + /// + /// When a turn is actively streaming, this returns that in-progress turn. Otherwise it falls + /// back to the most recently completed turn. This is useful for live UI updates, but a caller + /// that treats the result as committed persisted history can accidentally display incomplete + /// state after a partially applied event stream. pub fn active_turn_snapshot(&self) -> Option { self.current_turn .as_ref() @@ -129,6 +181,7 @@ impl ThreadHistoryBuilder { .or_else(|| self.turns.last().cloned()) } + /// Return whether replay is currently inside an unfinished turn boundary. pub fn has_active_turn(&self) -> bool { self.current_turn.is_some() } @@ -195,6 +248,12 @@ impl ThreadHistoryBuilder { } } + /// Apply one persisted rollout line to the reducer. + /// + /// Unlike [`Self::handle_event`], this accepts the outer `RolloutItem` wrapper so replay can + /// preserve rollback markers and explicit turn boundaries exactly as they were recorded on + /// disk. Feed items in file order; replaying a suffix or shuffling lines can still produce + /// syntactically valid turns while silently attaching items to the wrong exchange. pub fn handle_rollout_item(&mut self, item: &RolloutItem) { match item { RolloutItem::EventMsg(event) => self.handle_event(event), @@ -983,6 +1042,10 @@ fn render_review_output_text(output: &ReviewOutputEvent) -> String { } } +/// Convert core patch-apply change records into the app-server wire format. +/// +/// The returned list is sorted by path so repeated replays of the same patch event stay stable for +/// clients and snapshot tests. pub fn convert_patch_changes( changes: &HashMap, ) -> Vec { diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 00ca885bf4..1838ed49d1 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -1871,6 +1871,11 @@ pub struct ThreadForkParams { /// [UNSTABLE] Fork after the specified historical turn (inclusive). /// When omitted, the full thread history is copied. + /// + /// The value should come from `thread/read(includeTurns=true)` for the same source thread. + /// Requests are rejected for legacy histories whose replayed turn ids are not stable, for + /// in-progress turns, and for turns that do not yet contain both a user message and an agent + /// response. #[experimental("thread/fork.forkAfterTurnId")] #[ts(optional = nullable)] pub fork_after_turn_id: Option, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 44b479ce31..0455191ecb 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -231,6 +231,11 @@ To branch from a stored session, call `thread/fork` with the `thread.id`. This c } } ``` +The `forkAfterTurnId` value should come from `thread/read(includeTurns=true)` for that source +thread. Anchored forks are rejected for legacy histories that only have replay-synthesized turn +ids, for turns that are still in progress, and for turns that do not yet include both the user's +input and the model's response. + Experimental API: `thread/start`, `thread/resume`, and `thread/fork` accept `persistExtendedHistory: true` to persist a richer subset of ThreadItems for non-lossy history when calling `thread/read`, `thread/resume`, and `thread/fork` later. This does not backfill events that were not persisted previously. ### Example: List threads (with pagination & filters) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 818a034daa..d11c2a7d10 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -6921,6 +6921,14 @@ async fn sync_default_client_residency_requirement( } } +/// Return the exact rollout prefix that should seed a turn-anchored `thread/fork`. +/// +/// The selected turn stays in the returned history, while every later turn is removed. The helper +/// validates that the anchor is a stable, already-materialized exchange: legacy replay-synthesized +/// ids are rejected, the turn must not still be in progress, and the turn must include both a +/// user message and an agent response. The cutoff uses both the next explicit `TurnStarted` event +/// and the next user-message boundary so the fork cannot accidentally retain trailing non-user +/// events that conceptually belong to a later turn. fn resolve_thread_fork_history_from_anchor( rollout_items: &[RolloutItem], fork_after_turn_id: &str, @@ -7006,6 +7014,11 @@ fn resolve_thread_fork_history_from_anchor( Ok(rollout_items[..cut_idx].to_vec()) } +/// Return whether this rollout item starts a new user turn for truncation purposes. +/// +/// Anchored forks use this as a fallback boundary when explicit turn lifecycle events are present +/// but a later turn also emitted standalone user-message markers that should not leak into the +/// forked prefix. fn is_user_turn_rollout_boundary(item: &RolloutItem) -> bool { match item { RolloutItem::ResponseItem(ResponseItem::Message { role, .. }) => role == "user",