From 8493472983274dd56f75f5e44baf2f1493325da3 Mon Sep 17 00:00:00 2001 From: Hare Date: Sun, 30 Aug 2026 12:18:44 +0900 Subject: [PATCH] refactor: require annotated session log history --- crates/protocol/src/lib.rs | 4 +- crates/session-store/src/history.rs | 55 +-- .../session-store/src/legacy_session_log.rs | 188 +++++++++ crates/session-store/src/lib.rs | 7 +- crates/session-store/src/public_snapshot.rs | 60 +-- crates/session-store/src/segment.rs | 84 ++-- crates/session-store/src/segment_log.rs | 219 ++++------- crates/session-store/src/system_item.rs | 2 +- .../session-store/src/worker_session_store.rs | 368 +++++++++--------- crates/session-store/tests/fs_store_test.rs | 40 +- crates/session-store/tests/session_test.rs | 93 +++-- crates/tui/src/app.rs | 82 ++-- crates/tui/src/worker_list.rs | 16 +- crates/worker-runtime/src/worker_backend.rs | 9 +- crates/worker/src/controller.rs | 8 +- .../src/feature/builtin/flow_transition.rs | 5 +- crates/worker/src/hook.rs | 2 +- crates/worker/src/in_flight.rs | 10 +- crates/worker/src/ipc/interceptor.rs | 7 +- crates/worker/src/ipc/notify_buffer.rs | 2 +- crates/worker/src/ipc/protocol_session.rs | 15 +- crates/worker/src/ipc/server.rs | 2 +- crates/worker/src/segment_log_sink.rs | 54 +-- crates/worker/src/session_history.rs | 64 ++- crates/worker/src/spawn/registry.rs | 43 +- crates/worker/src/worker.rs | 93 +++-- crates/worker/tests/compact_events_test.rs | 38 +- crates/worker/tests/controller_test.rs | 18 +- 28 files changed, 872 insertions(+), 716 deletions(-) create mode 100644 crates/session-store/src/legacy_session_log.rs diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index bbda812a..8c498f09 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -470,7 +470,7 @@ pub struct SessionToolAttachment { #[serde(tag = "event", content = "data", rename_all = "snake_case")] pub enum Event { /// A user input message was accepted, persisted as - /// `LogEntry::UserInput`, and is about to start a new turn. + /// `LogEntry::AnnotatedUserInput`, and is about to start a new turn. /// Broadcast to every subscribed client so TUI / GUI instances show /// the same user line that reconnect snapshots would replay from /// history; clients must not synthesize a separate pending/fake @@ -491,7 +491,7 @@ pub enum Event { /// of parsing free-text prefixes like `[Notification] …` or /// `[File: …]`. /// - /// One event per `LogEntry::SystemItem` commit. Disk-side and + /// One event per `LogEntry::AnnotatedSystemItem` commit. Disk-side and /// wire-side are 1:1. SystemItem { #[cfg_attr(feature = "typescript", ts(type = "unknown"))] diff --git a/crates/session-store/src/history.rs b/crates/session-store/src/history.rs index 9558f343..e960bfec 100644 --- a/crates/session-store/src/history.rs +++ b/crates/session-store/src/history.rs @@ -1,9 +1,8 @@ //! Serializable history entries with restore-authoritative logical identity and origin. -use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; use serde::{Deserialize, Serialize}; -use crate::{LoggedItem, SessionId}; +use crate::LoggedItem; /// Stable logical identity of one model-visible history entry. /// @@ -143,12 +142,15 @@ mod tests { #[test] fn annotated_segment_start_is_restore_visible_without_projecting_metadata() { let session_id = uuid::Uuid::now_v7(); - let history_entry = legacy_logged_history(LoggedItem::Message { - role: LoggedRole::Assistant, - content: vec![crate::LoggedContentPart::Text { - text: "answer".into(), - }], - }); + let history_entry = LoggedHistoryEntry { + item: LoggedItem::Message { + role: LoggedRole::Assistant, + content: vec![crate::LoggedContentPart::Text { + text: "answer".into(), + }], + }, + metadata: LoggedSessionHistoryMetadata::legacy_unknown(), + }; let state = crate::collect_state(&[crate::LogEntry::AnnotatedSegmentStart { ts: 1, session_id, @@ -161,40 +163,3 @@ mod tests { assert_eq!(state.history[0].as_text(), Some("answer")); } } - -/// Legacy Session Logs did not persist annotations. Decode helpers explicitly -/// create `LegacyUnknown`; they never infer Human/System authority from role or -/// plaintext. -pub fn legacy_logged_history(item: LoggedItem) -> LoggedHistoryEntry { - LoggedHistoryEntry { - item, - metadata: LoggedSessionHistoryMetadata::legacy_unknown(), - } -} - -pub fn legacy_segment_history( - session_id: SessionId, - items: impl IntoIterator, -) -> Vec { - items - .into_iter() - .enumerate() - .map(|(index, item)| LoggedHistoryEntry { - item, - metadata: LoggedSessionHistoryMetadata { - // Legacy logs have no persisted entry id. Derive one solely from - // durable segment content rather than minting a new random value - // on every restore/read. The explicit LegacyUnknown origin keeps - // this compatibility identity from becoming trust authority. - entry_id: { - let mut identity = Vec::with_capacity(24); - identity.extend_from_slice(session_id.as_bytes()); - identity.extend_from_slice(&(index as u64).to_be_bytes()); - LoggedSessionHistoryEntryId(format!("l-{}", URL_SAFE_NO_PAD.encode(identity))) - }, - origin: LoggedSessionHistoryOrigin::LegacyUnknown, - derivation: None, - }, - }) - .collect() -} diff --git a/crates/session-store/src/legacy_session_log.rs b/crates/session-store/src/legacy_session_log.rs new file mode 100644 index 00000000..40dd0fd5 --- /dev/null +++ b/crates/session-store/src/legacy_session_log.rs @@ -0,0 +1,188 @@ +//! Versioned decoder for Session schemas that predate canonical annotated history. +//! +//! These types are intentionally private to `session-store`. Current writers, +//! replay, and public projections use [`crate::LogEntry`] exclusively; only the +//! Worker Session schema migration is allowed to deserialize these shapes. + +use agen::llm_client::types::RequestConfig; +use protocol::Segment; +use serde::Deserialize; + +use crate::{ + LogEntry, LoggedHistoryEntry, LoggedItem, LoggedSessionHistoryEntryId, + LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry, SegmentId, + SegmentOrigin, SessionExtension, SessionId, SystemItem, +}; + +#[derive(Debug, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum LegacyHistoryLogEntry { + SegmentStart { + ts: u64, + session_id: SessionId, + system_prompt: Option, + config: RequestConfig, + history: Vec, + #[serde(default)] + forked_from: Option, + #[serde(default)] + compacted_from: Option, + }, + UserInput { + ts: u64, + segments: Vec, + #[serde(default)] + extensions: Vec, + }, + AssistantItem { + ts: u64, + item: LoggedItem, + }, + ToolResult { + ts: u64, + item: LoggedItem, + }, + SystemItem { + ts: u64, + item: SystemItem, + }, +} + +/// Schema-v1 decoder. Non-history records already had their current shape, so +/// they pass through `LogEntry`; legacy history records are converted below. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum LegacySessionLogEntryV1 { + History(LegacyHistoryLogEntry), + Current(LogEntry), +} + +/// Schema v2 retained the v1 history shapes while adding non-history records. +/// Keep a distinct type so supported source versions remain explicit rather +/// than turning migration compatibility into the current `LogEntry` contract. +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum LegacySessionLogEntryV2 { + History(LegacyHistoryLogEntry), + Current(LogEntry), +} + +pub(crate) fn decode_entry( + schema_version: u32, + line: &str, + session_id: SessionId, + segment_id: SegmentId, + line_index: usize, +) -> Result { + let entry = match schema_version { + 1 => match serde_json::from_str::(line)? { + LegacySessionLogEntryV1::History(entry) => Entry::History(entry), + LegacySessionLogEntryV1::Current(entry) => Entry::Current(entry), + }, + 2 => match serde_json::from_str::(line)? { + LegacySessionLogEntryV2::History(entry) => Entry::History(entry), + LegacySessionLogEntryV2::Current(entry) => Entry::Current(entry), + }, + _ => unreachable!("legacy decoder called for unsupported schema {schema_version}"), + }; + Ok(match entry { + Entry::History(entry) => { + canonicalize_history_entry(session_id, segment_id, line_index, entry) + } + Entry::Current(entry) => entry, + }) +} + +enum Entry { + History(LegacyHistoryLogEntry), + Current(LogEntry), +} + +fn legacy_metadata( + segment_id: SegmentId, + line_index: usize, + item_index: usize, +) -> LoggedSessionHistoryMetadata { + let mut identity = Vec::with_capacity(32); + identity.extend_from_slice(segment_id.as_bytes()); + identity.extend_from_slice(&(line_index as u64).to_be_bytes()); + identity.extend_from_slice(&(item_index as u64).to_be_bytes()); + LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId(format!( + "l-{}", + base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, identity) + )), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + } +} + +fn canonicalize_history_entry( + _session_id: SessionId, + segment_id: SegmentId, + line_index: usize, + entry: LegacyHistoryLogEntry, +) -> LogEntry { + match entry { + LegacyHistoryLogEntry::SegmentStart { + ts, + session_id, + system_prompt, + config, + history, + forked_from, + compacted_from, + } => LogEntry::AnnotatedSegmentStart { + ts, + session_id, + system_prompt, + config, + history: history + .into_iter() + .enumerate() + .map(|(item_index, item)| LoggedHistoryEntry { + item, + metadata: legacy_metadata(segment_id, line_index, item_index), + }) + .collect(), + forked_from, + compacted_from, + }, + LegacyHistoryLogEntry::UserInput { + ts, + segments, + extensions, + } => LogEntry::AnnotatedUserInput { + ts, + history: vec![LoggedHistoryEntry { + item: LoggedItem::from(agen::Item::user_message(Segment::flatten_to_text( + &segments, + ))), + metadata: legacy_metadata(segment_id, line_index, 0), + }], + segments, + extensions, + }, + LegacyHistoryLogEntry::AssistantItem { ts, item } => LogEntry::AnnotatedAssistantItem { + ts, + entry: LoggedHistoryEntry { + item, + metadata: legacy_metadata(segment_id, line_index, 0), + }, + }, + LegacyHistoryLogEntry::ToolResult { ts, item } => LogEntry::AnnotatedToolResult { + ts, + entry: LoggedHistoryEntry { + item, + metadata: legacy_metadata(segment_id, line_index, 0), + }, + }, + LegacyHistoryLogEntry::SystemItem { ts, item } => LogEntry::AnnotatedSystemItem { + ts, + entry: LoggedSystemHistoryEntry { + item, + metadata: legacy_metadata(segment_id, line_index, 0), + }, + }, + } +} diff --git a/crates/session-store/src/lib.rs b/crates/session-store/src/lib.rs index 07bc0bc9..411a9aa0 100644 --- a/crates/session-store/src/lib.rs +++ b/crates/session-store/src/lib.rs @@ -26,13 +26,14 @@ //! let (session_id, segment_id) = create_segment(&store, SegmentStartState { //! system_prompt: None, //! config: &config, -//! history: &[], +//! history: Vec::new(), //! })?; //! ``` pub mod event_trace; pub mod fs_store; pub mod history; +mod legacy_session_log; pub mod logged_item; pub mod public_snapshot; pub mod segment; @@ -49,11 +50,11 @@ pub use fs_store::FsStore; pub use history::{ LoggedHistoryDerivation, LoggedHistoryEntry, LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry, - LoggedWorkerSubject, legacy_logged_history, legacy_segment_history, + LoggedWorkerSubject, }; pub use logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged}; pub use segment::{ - SegmentStartState, append_entry, append_system_item, classify_history_item, + SegmentStartState, append_entry, append_system_item, classify_logged_history_entry, create_compacted_segment, create_segment, create_segment_with_ids, ensure_head_or_fork, fork, fork_at, restore, restore_by_segment, save_config_changed, save_delta, save_extension, save_run_completed, save_run_errored, save_turn_end, save_usage, save_user_input, diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 07e3a123..3109bbe3 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -12,13 +12,12 @@ use crate::{ LoggedSessionHistoryOrigin, SessionId, SystemItem, }; -/// Project a complete current-segment log. A valid segment always starts with -/// one of the two SegmentStart records; malformed partial input uses the nil -/// session only to keep the public failure projection deterministic. +/// Project a complete current-segment log. A valid segment starts with one +/// canonical annotated SegmentStart record; malformed partial input uses the +/// nil session only to keep the public failure projection deterministic. pub fn project_current_session_snapshot(log: &[LogEntry]) -> SessionSnapshot { let session_id = log.iter().find_map(|entry| match entry { - LogEntry::SegmentStart { session_id, .. } - | LogEntry::AnnotatedSegmentStart { session_id, .. } => Some(*session_id), + LogEntry::AnnotatedSegmentStart { session_id, .. } => Some(*session_id), _ => None, }); project_session_snapshot(session_id.unwrap_or_else(SessionId::nil), log) @@ -32,20 +31,6 @@ pub fn project_session_snapshot(session_id: SessionId, log: &[LogEntry]) -> Sess for (log_index, record) in log.iter().enumerate() { match record { - LogEntry::SegmentStart { - ts, - session_id, - history, - .. - } => { - session_key = *session_id; - entries.clear(); - for (item_index, item) in history.iter().enumerate() { - if let Some(data) = project_item(item) { - entries.push(legacy_entry(&session_key, log_index, item_index, *ts, data)); - } - } - } LogEntry::AnnotatedSegmentStart { ts, session_id, @@ -56,39 +41,18 @@ pub fn project_session_snapshot(session_id: SessionId, log: &[LogEntry]) -> Sess entries.clear(); extend_history(&mut entries, history, None, *ts); } - LogEntry::UserInput { ts, segments, .. } => entries.push(legacy_entry( - &session_key, - log_index, - 0, - *ts, - SessionSnapshotEntryData::UserInput { - segments: segments.clone(), - }, - )), LogEntry::AnnotatedUserInput { ts, segments, history, .. } => extend_history(&mut entries, history, Some(segments), *ts), - LogEntry::AssistantItem { ts, item } | LogEntry::ToolResult { ts, item } => { - if let Some(data) = project_item(item) { - entries.push(legacy_entry(&session_key, log_index, 0, *ts, data)); - } - } LogEntry::AnnotatedAssistantItem { ts, entry } | LogEntry::AnnotatedToolResult { ts, entry } => { if let Some(data) = project_item(&entry.item) { entries.push(history_entry(entry, *ts, data)); } } - LogEntry::SystemItem { ts, item } => entries.push(system_entry( - item, - legacy_entry_id(&session_key, log_index, 0), - *ts, - SessionEntryProvenance::LegacyUnknown, - Vec::new(), - )), LogEntry::AnnotatedSystemItem { ts, entry } => entries.push(system_entry( &entry.item, entry.metadata.entry_id.0.clone(), @@ -332,9 +296,9 @@ mod tests { use crate::{LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedWorkerSubject}; #[test] - fn legacy_projection_is_stable_and_hides_reasoning_and_system_prompts() { + fn current_projection_is_stable_and_hides_reasoning_and_system_prompts() { let session_id = crate::new_session_id(); - let log = vec![LogEntry::SegmentStart { + let log = vec![LogEntry::AnnotatedSegmentStart { ts: 1, session_id, system_prompt: None, @@ -358,7 +322,17 @@ mod tests { text: "visible".into(), }], }, - ], + ] + .into_iter() + .map(|item| LoggedHistoryEntry { + item, + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId::new(), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + }) + .collect(), forked_from: None, compacted_from: None, }]; diff --git a/crates/session-store/src/segment.rs b/crates/session-store/src/segment.rs index 5be5b320..017cb970 100644 --- a/crates/session-store/src/segment.rs +++ b/crates/session-store/src/segment.rs @@ -4,11 +4,9 @@ //! The caller (typically Worker) holds the Engine directly and calls these //! functions after state-mutating operations. -use crate::logged_item::{LoggedItem, to_logged}; use crate::segment_log::{self, LogEntry, SegmentOrigin}; use crate::store::{Store, StoreError}; -use crate::system_item::SystemItem; -use crate::{SegmentId, SessionId}; +use crate::{LoggedHistoryEntry, LoggedSystemHistoryEntry, SegmentId, SessionId}; use agen::EngineResult; use agen::llm_client::RequestConfig; use agen::llm_client::types::Item; @@ -18,7 +16,7 @@ use protocol::Segment; pub struct SegmentStartState<'a> { pub system_prompt: Option<&'a str>, pub config: &'a RequestConfig, - pub history: &'a [Item], + pub history: Vec, } /// Create a new session + initial segment, writing the initial @@ -44,12 +42,12 @@ pub fn create_segment_with_ids( segment_id: SegmentId, state: SegmentStartState<'_>, ) -> Result<(), StoreError> { - let entry = LogEntry::SegmentStart { + let entry = LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id, system_prompt: state.system_prompt.map(String::from), config: state.config.clone(), - history: to_logged(state.history), + history: state.history.to_vec(), forked_from: None, compacted_from: None, }; @@ -70,12 +68,12 @@ pub fn create_compacted_segment( source_turn_count: usize, ) -> Result { let segment_id = crate::new_segment_id(); - let entry = LogEntry::SegmentStart { + let entry = LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id: source_session_id, system_prompt: state.system_prompt.map(String::from), config: state.config.clone(), - history: to_logged(state.history), + history: state.history.to_vec(), forked_from: None, compacted_from: Some(SegmentOrigin { segment_id: source_segment_id, @@ -154,12 +152,12 @@ pub fn ensure_head_or_fork( } let source_segment_id = *segment_id; let fork_id = crate::new_segment_id(); - let entry = LogEntry::SegmentStart { + let entry = LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id, system_prompt: state.system_prompt.map(String::from), config: state.config.clone(), - history: to_logged(state.history), + history: state.history.to_vec(), forked_from: Some(SegmentOrigin { segment_id: source_segment_id, at_turn_index, @@ -183,8 +181,9 @@ pub fn save_user_input( session_id: SessionId, segment_id: SegmentId, segments: Vec, + history: Vec, ) -> Result<(), StoreError> { - save_user_input_with_extensions(store, session_id, segment_id, segments, Vec::new()) + save_user_input_with_extensions(store, session_id, segment_id, segments, history, Vec::new()) } /// Atomically persist one typed user submission and Runtime-owned session @@ -194,15 +193,17 @@ pub fn save_user_input_with_extensions( session_id: SessionId, segment_id: SegmentId, segments: Vec, + history: Vec, extensions: Vec, ) -> Result<(), StoreError> { append_entry( store, session_id, segment_id, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: segment_log::now_millis(), segments, + history, extensions, }, ) @@ -220,64 +221,57 @@ pub fn save_delta( store: &impl Store, session_id: SessionId, segment_id: SegmentId, - new_items: &[Item], + new_items: &[LoggedHistoryEntry], ) -> Result<(), StoreError> { if new_items.is_empty() { return Ok(()); } let ts = segment_log::now_millis(); - for item in new_items { + for entry in new_items { + let item = Item::from(entry.item.clone()); if item.is_user_message() { // Already persisted by save_user_input at submit time. continue; } - let entry = classify_history_item(item, ts); + let entry = classify_logged_history_entry(entry.clone(), ts); append_entry(store, session_id, segment_id, entry)?; } Ok(()) } -/// Map one history item to its singular `LogEntry` form. Used by the -/// fallback `save_delta` path and the controller's worker-callback -/// classifier so write classification lives in one place. -pub fn classify_history_item(item: &Item, ts: u64) -> LogEntry { +/// Map one annotated history entry to its singular `LogEntry` form. Used by +/// the fallback `save_delta` path and the controller's worker-callback +/// classifier so write classification lives in one place without discarding +/// identity or provenance. +/// Map one already-annotated history entry to its singular canonical record +/// without changing its identity or provenance. +pub fn classify_logged_history_entry(entry: LoggedHistoryEntry, ts: u64) -> LogEntry { + let item = Item::from(entry.item.clone()); if item.is_tool_result() { - LogEntry::ToolResult { - ts, - item: LoggedItem::from(item), - } - } else if item.is_assistant_message() || item.is_tool_call() || item.is_reasoning() { - LogEntry::AssistantItem { - ts, - item: LoggedItem::from(item), - } + LogEntry::AnnotatedToolResult { ts, entry } } else { - // Defensive: anything else (future Item kinds) routes through - // AssistantItem rather than getting silently dropped. - LogEntry::AssistantItem { - ts, - item: LoggedItem::from(item), - } + // Assistant messages, tool calls, reasoning, and future non-user + // items all use the assistant-side canonical record. + LogEntry::AnnotatedAssistantItem { ts, entry } } } -/// Append a single typed system item as `LogEntry::SystemItem`. Helper -/// for the Worker-side interceptor commit path; mirrors the per-item -/// commit shape used for assistant / tool result entries. +/// Append one typed system item and its history metadata as a canonical +/// `LogEntry::AnnotatedSystemItem`. pub fn append_system_item( store: &impl Store, session_id: SessionId, segment_id: SegmentId, - item: SystemItem, + entry: LoggedSystemHistoryEntry, ) -> Result<(), StoreError> { append_entry( store, session_id, segment_id, - LogEntry::SystemItem { + LogEntry::AnnotatedSystemItem { ts: segment_log::now_millis(), - item, + entry, }, ) } @@ -430,12 +424,12 @@ pub fn fork( ) -> Result<(SessionId, SegmentId), StoreError> { let session_id = crate::new_session_id(); let fork_id = crate::new_segment_id(); - let entry = LogEntry::SegmentStart { + let entry = LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id, system_prompt: state.system_prompt.map(String::from), config: state.config.clone(), - history: to_logged(state.history), + history: state.history.to_vec(), forked_from: None, compacted_from: None, }; @@ -470,7 +464,7 @@ pub fn fork_at( // segment), before any turn completes. entries .iter() - .position(|e| !matches!(e, LogEntry::SegmentStart { .. })) + .position(|e| !matches!(e, LogEntry::AnnotatedSegmentStart { .. })) .unwrap_or(entries.len()) } else { entries @@ -482,12 +476,12 @@ pub fn fork_at( let state = segment_log::collect_state(&entries[..cut]); let fork_id = crate::new_segment_id(); - let entry = LogEntry::SegmentStart { + let entry = LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id: source_session_id, system_prompt: state.system_prompt, config: state.config, - history: to_logged(&state.history), + history: state.annotated_history, forked_from: Some(SegmentOrigin { segment_id: source_id, at_turn_index, diff --git a/crates/session-store/src/segment_log.rs b/crates/session-store/src/segment_log.rs index b03163d2..73e250ea 100644 --- a/crates/session-store/src/segment_log.rs +++ b/crates/session-store/src/segment_log.rs @@ -16,7 +16,6 @@ use serde::{Deserialize, Serialize}; use crate::history::{LoggedHistoryEntry, LoggedSystemHistoryEntry}; use crate::logged_item::LoggedItem; -use crate::system_item::SystemItem; /// A single segment log entry, serialized as one JSONL line. /// @@ -50,28 +49,7 @@ impl SessionExtension { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum LogEntry { - /// Segment start. Always the first entry in a segment log. - /// For forked segments, `history` contains the seed state from the parent. - SegmentStart { - ts: u64, - /// Session this segment belongs to. Compaction / fork inherits - /// the source segment's session_id; only fresh "new conversation" - /// segments mint a new session_id. - session_id: crate::SessionId, - system_prompt: Option, - config: RequestConfig, - history: Vec, - /// Origin: forked from a sibling segment at a specific turn boundary. - /// The referenced segment is guaranteed to share `session_id`. - #[serde(default, skip_serializing_if = "Option::is_none")] - forked_from: Option, - /// Origin: compacted from a sibling segment at a specific turn boundary. - /// The referenced segment is guaranteed to share `session_id`. - #[serde(default, skip_serializing_if = "Option::is_none")] - compacted_from: Option, - }, - - /// Schema-v2 segment seed. Retained entries keep their stable logical + /// Canonical segment seed. Retained entries keep their stable logical /// identity and origin across fork/compaction/restore. AnnotatedSegmentStart { ts: u64, @@ -105,22 +83,7 @@ pub enum LogEntry { /// restore conservatively instead of re-running a dangling tool call. Invoke { ts: u64, trigger: InvokeKind }, - /// User input accepted at submit time. Carries the original typed - /// `Vec` so clients can re-render typed atoms (paste chips, - /// file refs) on segment restore. - /// Replay flattens these into a `Item::user_message` for the worker - /// history; the worker layer never sees segments directly. - UserInput { - ts: u64, - segments: Vec, - /// Typed durable state committed atomically with this input record. - /// Runtime-owned Flow invocation uses this to avoid a Backend-instance - /// commit that can get ahead of Worker history. - #[serde(default, skip_serializing_if = "Vec::is_empty")] - extensions: Vec, - }, - - /// Schema-v2 user submission with its exact model-visible entries. Typed + /// Canonical user submission with its exact model-visible entries. Typed /// Flow instructions and caller-attributed input remain separate entries. AnnotatedUserInput { ts: u64, @@ -130,35 +93,19 @@ pub enum LogEntry { history: Vec, }, - /// Schema-v2 model output and metadata committed as one journal record. + /// Canonical model output and metadata committed as one journal record. AnnotatedAssistantItem { ts: u64, entry: LoggedHistoryEntry }, - /// One assistant-side item appended to history — assistant message, - /// reasoning, or tool call. Singular: one entry per history item so - /// the wire-side `Event::*` lane and on-disk LogEntry stay 1:1. - AssistantItem { ts: u64, item: LoggedItem }, - - /// Schema-v2 tool output and metadata committed as one journal record. + /// Canonical tool output and metadata committed as one journal record. AnnotatedToolResult { ts: u64, entry: LoggedHistoryEntry }, - /// One tool-execution result appended to history. - ToolResult { ts: u64, item: LoggedItem }, - - /// Schema-v2 typed system event and model-visible metadata committed + /// Canonical typed system event and model-visible metadata committed /// together. AnnotatedSystemItem { ts: u64, entry: LoggedSystemHistoryEntry, }, - /// One typed agent-injected system item: notification, child-Worker - /// lifecycle event, `@` / `/` resolution payload. Each - /// `SystemItem` carries kind metadata that the LLM - /// itself never sees (the LLM gets `Item::system_message` with the - /// item's denormalised `body`), but live clients and replay paths - /// dispatch on `kind` for typed rendering. - SystemItem { ts: u64, item: SystemItem }, - /// Turn boundary. Records the turn count after increment. TurnEnd { ts: u64, turn_count: usize }, @@ -260,6 +207,10 @@ pub struct RestoredState { pub system_prompt: Option, pub config: RequestConfig, pub history: Vec, + /// Canonical persisted history with stable identity and provenance. This is + /// the authority for rewrites, forks, and annotated restore; `history` is + /// retained as the model-facing item projection. + pub annotated_history: Vec, pub turn_count: usize, /// AgentTurns consumed by the active paused/yielded logical run. pub active_run_turn_count: Option, @@ -276,7 +227,7 @@ pub struct RestoredState { /// session-store は domain を不透明扱いし、各ドメインが自前で fold する。 pub extensions: Vec<(String, serde_json::Value)>, /// User submissions in original typed form, in submit order. - /// One entry per `LogEntry::UserInput`; the K-th entry corresponds to + /// One entry per `LogEntry::AnnotatedUserInput`; the K-th entry corresponds to /// the K-th `Item::user_message` derived during replay (modulo /// pre-compaction history seeded via `SegmentStart.history`, whose /// original segments are not preserved). Used by clients to re-render @@ -291,6 +242,7 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState { system_prompt: None, config: RequestConfig::default(), history: Vec::new(), + annotated_history: Vec::new(), turn_count: 0, active_run_turn_count: None, last_run_interrupted: false, @@ -304,18 +256,6 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState { state.entries_count += 1; match entry { - LogEntry::SegmentStart { - session_id, - system_prompt, - config, - history, - .. - } => { - state.session_id = Some(*session_id); - state.system_prompt = system_prompt.clone(); - state.config = config.clone(); - state.history = history.iter().cloned().map(Item::from).collect(); - } LogEntry::AnnotatedSegmentStart { session_id, system_prompt, @@ -326,6 +266,7 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState { state.session_id = Some(*session_id); state.system_prompt = system_prompt.clone(); state.config = config.clone(); + state.annotated_history = history.clone(); state.history = history .iter() .cloned() @@ -338,26 +279,13 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState { state.last_run_interrupted = true; state.active_run_turn_count = Some(0); } - LogEntry::UserInput { - segments, - extensions, - .. - } => { - let text = Segment::flatten_to_text(segments); - state.history.push(Item::user_message(text)); - state.user_segments.push(segments.clone()); - state.extensions.extend( - extensions - .iter() - .map(|extension| (extension.domain.clone(), extension.payload.clone())), - ); - } LogEntry::AnnotatedUserInput { segments, extensions, history, .. } => { + state.annotated_history.extend(history.iter().cloned()); state .history .extend(history.iter().cloned().map(|entry| Item::from(entry.item))); @@ -370,20 +298,16 @@ pub fn collect_state(entries: &[LogEntry]) -> RestoredState { } LogEntry::AnnotatedAssistantItem { entry, .. } | LogEntry::AnnotatedToolResult { entry, .. } => { + state.annotated_history.push(entry.clone()); state.history.push(Item::from(entry.item.clone())); } LogEntry::AnnotatedSystemItem { entry, .. } => { + state.annotated_history.push(LoggedHistoryEntry { + item: LoggedItem::from(entry.item.to_history_item()), + metadata: entry.metadata.clone(), + }); state.history.push(entry.item.to_history_item()); } - LogEntry::AssistantItem { item, .. } => { - state.history.push(Item::from(item.clone())); - } - LogEntry::ToolResult { item, .. } => { - state.history.push(Item::from(item.clone())); - } - LogEntry::SystemItem { item, .. } => { - state.history.push(item.to_history_item()); - } LogEntry::TurnEnd { turn_count, .. } => { if let Some(active_turn_count) = &mut state.active_run_turn_count { *active_turn_count += turn_count.saturating_sub(state.turn_count); @@ -465,6 +389,20 @@ pub fn now_millis() -> u64 { #[cfg(test)] mod tests { use super::*; + use crate::{ + LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, + }; + + fn annotated(item: Item) -> LoggedHistoryEntry { + LoggedHistoryEntry { + item: LoggedItem::from(item), + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId::new(), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } + } #[test] fn replay_empty() { @@ -476,12 +414,12 @@ mod tests { #[test] fn replay_segment_start_sets_initial_state() { - let state = collect_state(&[LogEntry::SegmentStart { + let state = collect_state(&[LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: Some("You are helpful.".into()), config: RequestConfig::default().with_max_tokens(1024), - history: vec![Item::user_message("seed").into()], + history: vec![annotated(Item::user_message("seed"))], forked_from: None, compacted_from: None, }]); @@ -494,7 +432,7 @@ mod tests { #[test] fn replay_full_turn() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -503,14 +441,15 @@ mod tests { forked_from: None, compacted_from: None, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2000, extensions: vec![], segments: vec![Segment::text("Hello")], + history: vec![annotated(Item::user_message("Hello"))], }, - LogEntry::AssistantItem { + LogEntry::AnnotatedAssistantItem { ts: 3000, - item: Item::assistant_message("Hi!").into(), + entry: annotated(Item::assistant_message("Hi!")), }, LogEntry::TurnEnd { ts: 3100, @@ -531,7 +470,7 @@ mod tests { #[test] fn replay_incomplete_invoke_is_interrupted() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -544,14 +483,15 @@ mod tests { ts: 2000, trigger: InvokeKind::UserSend, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2001, extensions: vec![], segments: vec![Segment::text("run a tool")], + history: vec![annotated(Item::user_message("run a tool"))], }, - LogEntry::AssistantItem { + LogEntry::AnnotatedAssistantItem { ts: 3000, - item: Item::tool_call("call_1", "side_effect", "{}").into(), + entry: annotated(Item::tool_call("call_1", "side_effect", "{}")), }, ]); @@ -561,7 +501,7 @@ mod tests { #[test] fn replay_with_tool_calls() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -570,22 +510,27 @@ mod tests { forked_from: None, compacted_from: None, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2000, extensions: vec![], segments: vec![Segment::text("Check weather")], + history: vec![annotated(Item::user_message("Check weather"))], }, - LogEntry::AssistantItem { + LogEntry::AnnotatedAssistantItem { ts: 3000, - item: Item::tool_call("call_1", "get_weather", r#"{"city":"Tokyo"}"#).into(), + entry: annotated(Item::tool_call( + "call_1", + "get_weather", + r#"{"city":"Tokyo"}"#, + )), }, - LogEntry::ToolResult { + LogEntry::AnnotatedToolResult { ts: 3500, - item: Item::tool_result("call_1", "Sunny, 25C").into(), + entry: annotated(Item::tool_result("call_1", "Sunny, 25C")), }, - LogEntry::AssistantItem { + LogEntry::AnnotatedAssistantItem { ts: 4000, - item: Item::assistant_message("It's sunny in Tokyo!").into(), + entry: annotated(Item::assistant_message("It's sunny in Tokyo!")), }, LogEntry::TurnEnd { ts: 4100, @@ -599,9 +544,9 @@ mod tests { #[test] fn replay_restores_durable_tool_image_detail() { - let entry = LogEntry::ToolResult { + let entry = LogEntry::AnnotatedToolResult { ts: 3500, - item: Item::tool_result_item_with_attachments( + entry: annotated(Item::tool_result_item_with_attachments( "call_image", "attached", None, @@ -609,8 +554,7 @@ mod tests { vec![agen::tool::Attachment::Image( agen::tool::ImageAttachment::new("image/png", b"durable-image".to_vec()), )], - ) - .into(), + )), }; let persisted = serde_json::to_string(&entry).unwrap(); let restored_entry: LogEntry = serde_json::from_str(&persisted).unwrap(); @@ -630,7 +574,7 @@ mod tests { #[test] fn replay_config_changed() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -650,7 +594,7 @@ mod tests { #[test] fn replay_llm_usage_appends_to_usage_history() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -659,10 +603,11 @@ mod tests { forked_from: None, compacted_from: None, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2000, extensions: vec![], segments: vec![Segment::text("hi")], + history: vec![annotated(Item::user_message("hi"))], }, LogEntry::LlmUsage { ts: 2100, @@ -672,9 +617,9 @@ mod tests { cache_write_tokens: 0, output_tokens: 10, }, - LogEntry::AssistantItem { + LogEntry::AnnotatedAssistantItem { ts: 2200, - item: Item::assistant_message("yo").into(), + entry: annotated(Item::assistant_message("yo")), }, LogEntry::LlmUsage { ts: 3100, @@ -698,7 +643,7 @@ mod tests { #[test] fn replay_without_llm_usage_keeps_usage_history_empty() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -707,10 +652,11 @@ mod tests { forked_from: None, compacted_from: None, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2000, extensions: vec![], segments: vec![Segment::text("hi")], + history: vec![annotated(Item::user_message("hi"))], }, ]); assert!(state.usage_history.is_empty()); @@ -771,7 +717,7 @@ mod tests { #[test] fn replay_invoke_marker_only_mutates_interrupted_state() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 0, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -784,10 +730,11 @@ mod tests { ts: 100, trigger: InvokeKind::UserSend, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 101, extensions: vec![], segments: vec![Segment::text("hi")], + history: vec![annotated(Item::user_message("hi"))], }, LogEntry::TurnEnd { ts: 200, @@ -806,7 +753,7 @@ mod tests { #[test] fn replay_paused_turn_abandoned_clears_interrupted_marker() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 0, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -830,7 +777,7 @@ mod tests { #[test] fn replay_restores_active_run_budget_across_compaction_checkpoint() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 0, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -861,7 +808,7 @@ mod tests { })) .expect("legacy run-completed entry"); let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 0, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -924,7 +871,7 @@ mod tests { #[test] fn replay_extension_collects_domain_payload_pairs() { let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -983,9 +930,12 @@ mod tests { #[test] fn user_input_extensions_restore_with_the_same_committed_input() { let segments = vec![Segment::text("Flow instructions"), Segment::text("Ticket")]; - let entry = LogEntry::UserInput { + let entry = LogEntry::AnnotatedUserInput { ts: 9999, segments: segments.clone(), + history: vec![annotated(Item::user_message(Segment::flatten_to_text( + &segments, + )))], extensions: vec![SessionExtension::new( "flow.runtime.v1", serde_json::json!({ "state": "implement", "revision": 0 }), @@ -1000,7 +950,7 @@ mod tests { assert_eq!(state.extensions[0].1["state"], "implement"); } - /// Mixed segments survive a JSON round-trip through `LogEntry::UserInput`, + /// Mixed segments survive a JSON round-trip through `LogEntry::AnnotatedUserInput`, /// and `collect_state` derives `Item::user_message` from the flattened /// text while preserving the original segments separately. This covers /// the segments → flatten → Item replay path from the ticket. @@ -1020,16 +970,19 @@ mod tests { path: "src/main.rs".into(), }, ]; - let entry = LogEntry::UserInput { + let entry = LogEntry::AnnotatedUserInput { ts: 4242, extensions: vec![], segments: segments.clone(), + history: vec![annotated(Item::user_message(Segment::flatten_to_text( + &segments, + )))], }; // JSON round-trip preserves the variant byte-for-byte. let json = serde_json::to_string(&entry).unwrap(); let parsed: LogEntry = serde_json::from_str(&json).unwrap(); let state = collect_state(&[ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1, session_id: uuid::Uuid::nil(), system_prompt: None, diff --git a/crates/session-store/src/system_item.rs b/crates/session-store/src/system_item.rs index 8ab59b56..cf04db4d 100644 --- a/crates/session-store/src/system_item.rs +++ b/crates/session-store/src/system_item.rs @@ -8,7 +8,7 @@ //! `kind` instead of parsing text prefixes like `[Notification] …` or //! `[File: …]`. //! -//! Persisted as the payload of [`crate::LogEntry::SystemItem`] (one +//! Persisted as the payload of [`crate::LogEntry::AnnotatedSystemItem`] (one //! entry per item), and broadcast live as the payload of //! `Event::SystemItem` on the wire. //! diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index 3fa0bae6..dcc12820 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -12,11 +12,7 @@ use crate::event_trace::TraceEntry; use crate::segment_log::LogEntry; use crate::store::{Store, StoreError}; -use crate::{ - LoggedHistoryEntry, LoggedItem, LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, - LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry, SegmentId, SessionId, -}; -use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use crate::{SegmentId, SessionId}; use serde::{Deserialize, Serialize}; use std::fs::{self, File, OpenOptions}; use std::io::{Read, Seek, SeekFrom, Write}; @@ -56,7 +52,11 @@ impl WorkerSessionStore { validate_canonical_segment_logs(&root)?; } PREVIOUS_SESSION_SCHEMA_VERSION | LEGACY_SESSION_SCHEMA_VERSION => { - migrate_segment_logs_to_v3(&root, manifest.session_id)?; + migrate_segment_logs_to_v3( + &root, + manifest.session_id, + manifest.schema_version, + )?; manifest.schema_version = SESSION_SCHEMA_VERSION; atomic_write_json(&root.join(SESSION_FILE), &manifest)?; } @@ -151,13 +151,7 @@ impl WorkerSessionStore { .join(format!("{segment_id}.trace.jsonl")) } - fn append_log_entry( - &self, - path: &Path, - session_id: SessionId, - segment_id: SegmentId, - entry: &LogEntry, - ) -> Result<(), StoreError> { + fn append_log_entry(&self, path: &Path, entry: &LogEntry) -> Result<(), StoreError> { let _guard = self .append_lock .lock() @@ -172,9 +166,8 @@ impl WorkerSessionStore { file.seek(SeekFrom::Start(0))?; let mut existing = Vec::new(); file.read_to_end(&mut existing)?; - let line_index = parse_jsonl::(&existing)?.len(); - let entry = canonicalize_log_entry(session_id, segment_id, line_index, entry.clone()); - let line = serde_json::to_string(&entry)?; + parse_jsonl::(&existing)?; + let line = serde_json::to_string(entry)?; let mut record = Vec::with_capacity(line.len() + 1); record.extend_from_slice(line.as_bytes()); record.push(b'\n'); @@ -232,7 +225,7 @@ impl Store for WorkerSessionStore { entry: &LogEntry, ) -> Result<(), StoreError> { self.ensure_session(session_id, true)?; - self.append_log_entry(&self.log_path(segment_id), session_id, segment_id, entry) + self.append_log_entry(&self.log_path(segment_id), entry) } fn read_all( @@ -285,9 +278,8 @@ impl Store for WorkerSessionStore { ) -> Result<(), StoreError> { self.ensure_session(session_id, true)?; let mut content = Vec::new(); - for (line_index, entry) in entries.iter().enumerate() { - let entry = canonicalize_log_entry(session_id, segment_id, line_index, entry.clone()); - serde_json::to_writer(&mut content, &entry)?; + for entry in entries { + serde_json::to_writer(&mut content, entry)?; content.push(b'\n'); } atomic_write_bytes(&self.log_path(segment_id), &content)?; @@ -380,7 +372,11 @@ fn segment_log_paths(root: &Path) -> Result, StoreErro Ok(paths) } -fn migrate_segment_logs_to_v3(root: &Path, session_id: SessionId) -> Result<(), StoreError> { +fn migrate_segment_logs_to_v3( + root: &Path, + session_id: SessionId, + source_schema_version: u32, +) -> Result<(), StoreError> { struct MigrationPlan { path: PathBuf, source: Vec, @@ -392,21 +388,14 @@ fn migrate_segment_logs_to_v3(root: &Path, session_id: SessionId) -> Result<(), let mut plans = Vec::new(); for (segment_id, path) in segment_log_paths(root)? { let source = fs::read(&path)?; - let entries: Vec = parse_jsonl(&source).map_err(|error| StoreError::Corrupt { - line: 0, - message: format!( - "cannot migrate Worker Session log {}: {error}", - path.display() - ), - })?; - let canonical = entries - .into_iter() - .enumerate() - .map(|(line_index, entry)| { - canonicalize_log_entry(session_id, segment_id, line_index, entry) - }) - .collect::>(); - validate_canonical_entries(&path, &canonical)?; + let canonical = parse_legacy_jsonl(source_schema_version, session_id, segment_id, &source) + .map_err(|error| StoreError::Corrupt { + line: 0, + message: format!( + "cannot migrate Worker Session log {}: {error}", + path.display() + ), + })?; let mut output = Vec::new(); for entry in canonical { serde_json::to_writer(&mut output, &entry)?; @@ -442,120 +431,33 @@ fn migrate_segment_logs_to_v3(root: &Path, session_id: SessionId) -> Result<(), fn validate_canonical_segment_logs(root: &Path) -> Result<(), StoreError> { for (_, path) in segment_log_paths(root)? { - let entries: Vec = parse_jsonl(&fs::read(&path)?)?; - validate_canonical_entries(&path, &entries)?; + let _: Vec = parse_jsonl(&fs::read(&path)?)?; } Ok(()) } -fn validate_canonical_entries(path: &Path, entries: &[LogEntry]) -> Result<(), StoreError> { - for (line_index, entry) in entries.iter().enumerate() { - if matches!( - entry, - LogEntry::SegmentStart { .. } - | LogEntry::UserInput { .. } - | LogEntry::AssistantItem { .. } - | LogEntry::ToolResult { .. } - | LogEntry::SystemItem { .. } - ) { - return Err(StoreError::Corrupt { - line: line_index + 1, - message: format!( - "Worker Session schema v3 contains legacy history record in {}", - path.display() - ), - }); - } - } - Ok(()) -} - -fn legacy_metadata( - _session_id: SessionId, - segment_id: SegmentId, - line_index: usize, - item_index: usize, -) -> LoggedSessionHistoryMetadata { - let mut identity = Vec::with_capacity(32); - identity.extend_from_slice(segment_id.as_bytes()); - identity.extend_from_slice(&(line_index as u64).to_be_bytes()); - identity.extend_from_slice(&(item_index as u64).to_be_bytes()); - LoggedSessionHistoryMetadata { - entry_id: LoggedSessionHistoryEntryId(format!("l-{}", URL_SAFE_NO_PAD.encode(identity))), - origin: LoggedSessionHistoryOrigin::LegacyUnknown, - derivation: None, - } -} - -fn canonicalize_log_entry( +fn parse_legacy_jsonl( + schema_version: u32, session_id: SessionId, segment_id: SegmentId, - line_index: usize, - entry: LogEntry, -) -> LogEntry { - match entry { - LogEntry::SegmentStart { - ts, - session_id, - system_prompt, - config, - history, - forked_from, - compacted_from, - } => LogEntry::AnnotatedSegmentStart { - ts, - session_id, - system_prompt, - config, - history: history - .into_iter() - .enumerate() - .map(|(item_index, item)| LoggedHistoryEntry { - item, - metadata: legacy_metadata(session_id, segment_id, line_index, item_index), - }) - .collect(), - forked_from, - compacted_from, - }, - LogEntry::UserInput { - ts, - segments, - extensions, - } => LogEntry::AnnotatedUserInput { - ts, - history: vec![LoggedHistoryEntry { - item: LoggedItem::from(agen::Item::user_message( - protocol::Segment::flatten_to_text(&segments), - )), - metadata: legacy_metadata(session_id, segment_id, line_index, 0), - }], - segments, - extensions, - }, - LogEntry::AssistantItem { ts, item } => LogEntry::AnnotatedAssistantItem { - ts, - entry: LoggedHistoryEntry { - item, - metadata: legacy_metadata(session_id, segment_id, line_index, 0), - }, - }, - LogEntry::ToolResult { ts, item } => LogEntry::AnnotatedToolResult { - ts, - entry: LoggedHistoryEntry { - item, - metadata: legacy_metadata(session_id, segment_id, line_index, 0), - }, - }, - LogEntry::SystemItem { ts, item } => LogEntry::AnnotatedSystemItem { - ts, - entry: LoggedSystemHistoryEntry { - item, - metadata: legacy_metadata(session_id, segment_id, line_index, 0), - }, - }, - canonical => canonical, - } + bytes: &[u8], +) -> Result, serde_json::Error> { + let text = std::str::from_utf8(bytes).map_err(|error| { + serde_json::Error::io(std::io::Error::new(std::io::ErrorKind::InvalidData, error)) + })?; + text.lines() + .enumerate() + .filter(|(_, line)| !line.trim().is_empty()) + .map(|(line_index, line)| { + crate::legacy_session_log::decode_entry( + schema_version, + line, + session_id, + segment_id, + line_index, + ) + }) + .collect() } fn atomic_write_json(path: &Path, value: &T) -> Result<(), StoreError> { @@ -659,7 +561,21 @@ fn truncate_uncommitted_tail(file: &mut File) -> std::io::Result { #[cfg(test)] mod tests { use super::*; - use crate::{Store, new_segment_id, new_session_id}; + use crate::{ + LoggedHistoryEntry, LoggedItem, LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, + LoggedSessionHistoryOrigin, Store, new_segment_id, new_session_id, + }; + + fn annotated(item: agen::Item) -> LoggedHistoryEntry { + LoggedHistoryEntry { + item: LoggedItem::from(item), + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId::new(), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } + } #[test] fn canonical_layout_and_single_session_invariant() { @@ -748,26 +664,27 @@ mod tests { ) .unwrap(); let source = vec![ - LogEntry::SegmentStart { - ts: 1, - session_id, - system_prompt: None, - config: agen::llm_client::RequestConfig::default(), - history: vec![LoggedItem::from(agen::Item::assistant_message("prior"))], - forked_from: None, - compacted_from: None, - }, - LogEntry::UserInput { - ts: 2, - segments: vec![protocol::Segment::Text { - content: "hello".into(), - }], - extensions: Vec::new(), - }, - LogEntry::AssistantItem { - ts: 3, - item: LoggedItem::from(agen::Item::assistant_message("reply")), - }, + serde_json::json!({ + "kind": "segment_start", + "ts": 1, + "session_id": session_id, + "system_prompt": null, + "config": agen::llm_client::RequestConfig::default(), + "history": [LoggedItem::from(agen::Item::assistant_message("prior"))], + "forked_from": null, + "compacted_from": null + }), + serde_json::json!({ + "kind": "user_input", + "ts": 2, + "segments": [{ "kind": "text", "content": "hello" }], + "extensions": [] + }), + serde_json::json!({ + "kind": "assistant_item", + "ts": 3, + "item": LoggedItem::from(agen::Item::assistant_message("reply")) + }), ]; let path = root .path() @@ -829,15 +746,16 @@ mod tests { .path() .join(SEGMENTS_DIR) .join(format!("{valid_segment}.jsonl")); - let valid_entry = LogEntry::SegmentStart { - ts: 1, - session_id, - system_prompt: None, - config: agen::llm_client::RequestConfig::default(), - history: vec![LoggedItem::from(agen::Item::assistant_message("prior"))], - forked_from: None, - compacted_from: None, - }; + let valid_entry = serde_json::json!({ + "kind": "segment_start", + "ts": 1, + "session_id": session_id, + "system_prompt": null, + "config": agen::llm_client::RequestConfig::default(), + "history": [LoggedItem::from(agen::Item::assistant_message("prior"))], + "forked_from": null, + "compacted_from": null + }); let mut valid_bytes = serde_json::to_vec(&valid_entry).unwrap(); valid_bytes.push(b'\n'); fs::write(&valid_path, &valid_bytes).unwrap(); @@ -862,7 +780,7 @@ mod tests { } #[test] - fn schema_v3_rejects_legacy_records_and_new_writes_are_canonical() { + fn current_jsonl_requires_annotations_across_append_rewrite_and_reopen() { let root = tempfile::tempdir().unwrap(); let session_id = new_session_id(); let segment_id = new_segment_id(); @@ -871,12 +789,12 @@ mod tests { .create_segment( session_id, segment_id, - &[LogEntry::SegmentStart { + &[LogEntry::AnnotatedSegmentStart { ts: 1, session_id, system_prompt: None, config: agen::llm_client::RequestConfig::default(), - history: Vec::new(), + history: vec![annotated(agen::Item::user_message("seed"))], forked_from: None, compacted_from: None, }], @@ -886,11 +804,94 @@ mod tests { .append( session_id, segment_id, - &LogEntry::UserInput { + &LogEntry::AnnotatedAssistantItem { + ts: 2, + entry: annotated(agen::Item::assistant_message("reply")), + }, + ) + .unwrap(); + + let before_rewrite = store.read_all(session_id, segment_id).unwrap(); + store + .create_segment(session_id, segment_id, &before_rewrite) + .unwrap(); + drop(store); + + let reopened = WorkerSessionStore::new(root.path()).unwrap(); + let restored = reopened.read_all(session_id, segment_id).unwrap(); + assert_eq!( + serde_json::to_value(&restored).unwrap(), + serde_json::to_value(&before_rewrite).unwrap() + ); + for entry in &restored { + match entry { + LogEntry::AnnotatedSegmentStart { history, .. } => assert!(history.iter().all( + |entry| !entry.metadata.entry_id.0.is_empty() + && matches!( + entry.metadata.origin, + LoggedSessionHistoryOrigin::LegacyUnknown + ) + )), + LogEntry::AnnotatedAssistantItem { entry, .. } => { + assert!(!entry.metadata.entry_id.0.is_empty()); + assert!(matches!( + entry.metadata.origin, + LoggedSessionHistoryOrigin::LegacyUnknown + )); + } + _ => {} + } + } + + let log = fs::read_to_string(reopened.log_path(segment_id)).unwrap(); + for line in log.lines() { + let value: serde_json::Value = serde_json::from_str(line).unwrap(); + let kind = value["kind"].as_str().unwrap(); + assert!( + !matches!( + kind, + "segment_start" + | "user_input" + | "assistant_item" + | "tool_result" + | "system_item" + ), + "current-schema JSONL contains legacy history record: {kind}" + ); + } + } + + #[test] + fn schema_v3_rejects_legacy_records_and_new_writes_are_canonical() { + let root = tempfile::tempdir().unwrap(); + let session_id = new_session_id(); + let segment_id = new_segment_id(); + let store = WorkerSessionStore::new(root.path()).unwrap(); + store + .create_segment( + session_id, + segment_id, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id, + system_prompt: None, + config: agen::llm_client::RequestConfig::default(), + history: vec![annotated(agen::Item::assistant_message("seed"))], + forked_from: None, + compacted_from: None, + }], + ) + .unwrap(); + store + .append( + session_id, + segment_id, + &LogEntry::AnnotatedUserInput { ts: 2, segments: vec![protocol::Segment::Text { content: "new".into(), }], + history: vec![annotated(agen::Item::user_message("new"))], extensions: Vec::new(), }, ) @@ -907,12 +908,11 @@ mod tests { let mut file = OpenOptions::new().append(true).open(path).unwrap(); serde_json::to_writer( &mut file, - &LogEntry::SystemItem { - ts: 3, - item: crate::SystemItem::LegacyIgnored { - slug: "legacy".into(), - }, - }, + &serde_json::json!({ + "kind": "system_item", + "ts": 3, + "item": { "kind": "legacy_ignored", "slug": "legacy" } + }), ) .unwrap(); file.write_all(b"\n").unwrap(); diff --git a/crates/session-store/tests/fs_store_test.rs b/crates/session-store/tests/fs_store_test.rs index 63028d29..430a685b 100644 --- a/crates/session-store/tests/fs_store_test.rs +++ b/crates/session-store/tests/fs_store_test.rs @@ -1,12 +1,25 @@ use agen::EngineResult; use agen::llm_client::types::{Item, RequestConfig}; use session_store::{ - FsStore, LogEntry, Store, TraceEntry, collect_state, new_segment_id, new_session_id, + FsStore, LogEntry, LoggedHistoryEntry, LoggedItem, LoggedSessionHistoryEntryId, + LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, Store, TraceEntry, collect_state, + new_segment_id, new_session_id, }; use std::io::Write; +fn annotated(item: Item) -> LoggedHistoryEntry { + LoggedHistoryEntry { + item: LoggedItem::from(item), + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId::new(), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } +} + fn nil_session_start(ts: u64, session_id: uuid::Uuid) -> LogEntry { - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts, session_id, system_prompt: None, @@ -25,7 +38,7 @@ fn round_trip_write_and_read() { let segid = new_segment_id(); let entries = vec![ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: sid, system_prompt: Some("You are helpful.".into()), @@ -34,14 +47,15 @@ fn round_trip_write_and_read() { forked_from: None, compacted_from: None, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2000, extensions: vec![], segments: vec![protocol::Segment::text("Hello")], + history: vec![annotated(Item::user_message("Hello"))], }, - LogEntry::AssistantItem { + LogEntry::AnnotatedAssistantItem { ts: 3000, - item: Item::assistant_message("Hi there!").into(), + entry: annotated(Item::assistant_message("Hi there!")), }, LogEntry::TurnEnd { ts: 3100, @@ -79,14 +93,14 @@ fn create_segment_writes_all_entries() { let sid = new_session_id(); let segid = new_segment_id(); - let entries = [LogEntry::SegmentStart { + let entries = [LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: sid, system_prompt: None, config: RequestConfig::default(), history: vec![ - Item::user_message("seed").into(), - Item::assistant_message("ok").into(), + annotated(Item::user_message("seed")), + annotated(Item::assistant_message("ok")), ], forked_from: None, compacted_from: None, @@ -205,7 +219,7 @@ fn read_entry_count_matches_append_tally() { let segid = new_segment_id(); let entries = [ - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: 1000, session_id: sid, system_prompt: None, @@ -214,10 +228,11 @@ fn read_entry_count_matches_append_tally() { forked_from: None, compacted_from: None, }, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: 2000, extensions: vec![], segments: vec![protocol::Segment::text("Hello")], + history: vec![annotated(Item::user_message("Hello"))], }, ]; @@ -254,10 +269,11 @@ fn unterminated_utf8_tail_is_ignored_and_replaced_on_append() { assert_eq!(store.read_all(sid, segid).unwrap().len(), 1); assert_eq!(store.read_entry_count(sid, segid).unwrap(), 1); - let next = LogEntry::UserInput { + let next = LogEntry::AnnotatedUserInput { ts: 2, extensions: vec![], segments: vec![protocol::Segment::text("recovered")], + history: vec![annotated(Item::user_message("recovered"))], }; store.append(sid, segid, &next).unwrap(); diff --git a/crates/session-store/tests/session_test.rs b/crates/session-store/tests/session_test.rs index d8e27fe0..898bc4bd 100644 --- a/crates/session-store/tests/session_test.rs +++ b/crates/session-store/tests/session_test.rs @@ -16,6 +16,21 @@ use session_store::{FsStore, LogEntry, SegmentStartState, Store, collect_state}; // Helpers // ============================================================================= +fn annotated(items: &[Item]) -> Vec { + items + .iter() + .cloned() + .map(|item| session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::from(item), + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId::new(), + origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + }) + .collect() +} + fn simple_text_events() -> Vec { vec![ Event::text_block_start(0), @@ -144,6 +159,7 @@ async fn run_and_persist( session_id, segment_id, vec![protocol::Segment::text(input)], + annotated(&[Item::user_message(input)]), ) .unwrap(); @@ -154,8 +170,8 @@ async fn run_and_persist( worker.engine = locked.unlock(); let projected = worker.history(); - let new_items = &projected[history_before..]; - session_store::save_delta(store, session_id, segment_id, new_items).unwrap(); + let new_items = annotated(&projected[history_before..]); + session_store::save_delta(store, session_id, segment_id, &new_items).unwrap(); session_store::save_turn_end(store, session_id, segment_id, worker.turn_count()).unwrap(); match &result { @@ -219,7 +235,7 @@ async fn session_run_logs_entries() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -237,7 +253,10 @@ async fn session_run_logs_entries() { ); // First entry is SegmentStart - assert!(matches!(&entries[0], LogEntry::SegmentStart { .. })); + assert!(matches!( + &entries[0], + LogEntry::AnnotatedSegmentStart { .. } + )); // Has a RunCompleted with Finished let has_finished = entries.iter().any(|e| { @@ -264,7 +283,7 @@ async fn session_restore_round_trip() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -303,7 +322,7 @@ async fn session_run_with_tool_call() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -314,12 +333,12 @@ async fn session_run_with_tool_call() { let has_tool_results = entries .iter() - .any(|e| matches!(e, LogEntry::ToolResult { .. })); + .any(|e| matches!(e, LogEntry::AnnotatedToolResult { .. })); assert!(has_tool_results, "should have ToolResult entry"); let has_assistant = entries .iter() - .any(|e| matches!(e, LogEntry::AssistantItem { .. })); + .any(|e| matches!(e, LogEntry::AnnotatedAssistantItem { .. })); assert!(has_assistant, "should have AssistantItem entry"); } @@ -338,7 +357,7 @@ async fn session_resume_after_pause() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -377,7 +396,7 @@ async fn session_fork_creates_new_session() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -390,7 +409,7 @@ async fn session_fork_creates_new_session() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -399,7 +418,10 @@ async fn session_fork_creates_new_session() { // Fork should have a SegmentStart with the current history let fork_entries = store.read_all(fork_sid, fork_segid).unwrap(); assert_eq!(fork_entries.len(), 1); - assert!(matches!(&fork_entries[0], LogEntry::SegmentStart { .. })); + assert!(matches!( + &fork_entries[0], + LogEntry::AnnotatedSegmentStart { .. } + )); let fork_state = collect_state(&fork_entries); assert_eq!(fork_state.session_id, Some(fork_sid)); @@ -418,7 +440,7 @@ async fn session_fork_at_truncates_within_session() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -445,6 +467,23 @@ async fn session_fork_at_truncates_within_session() { .expect("source segment has the matching TurnEnd"); let source_state_at_fork = collect_state(&all_entries[..=turn_end_pos]); assert_eq!(fork_state.history.len(), source_state_at_fork.history.len()); + assert_eq!( + fork_state.annotated_history, source_state_at_fork.annotated_history, + "fork_at must preserve every retained history entry identity and provenance", + ); + assert!(fork_state.annotated_history.iter().all(|entry| { + !entry.metadata.entry_id.0.is_empty() + && matches!( + entry.metadata.origin, + session_store::LoggedSessionHistoryOrigin::LegacyUnknown + | session_store::LoggedSessionHistoryOrigin::HumanInput { .. } + | session_store::LoggedSessionHistoryOrigin::WorkerInput { .. } + | session_store::LoggedSessionHistoryOrigin::BackendInstruction { .. } + | session_store::LoggedSessionHistoryOrigin::ModelOutput { .. } + | session_store::LoggedSessionHistoryOrigin::ToolOutput { .. } + | session_store::LoggedSessionHistoryOrigin::DerivedSummary + ) + })); // list_segments should show both source and fork in the same Session. let segs = store.list_segments(sid).unwrap(); @@ -463,7 +502,7 @@ async fn session_config_changed_logged() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -496,7 +535,7 @@ async fn session_auto_forks_on_conflict() { SegmentStartState { system_prompt: worker_a.get_system_prompt(), config: worker_a.request_config(), - history: &worker_a.history(), + history: annotated(&worker_a.history()), }, ) .unwrap(); @@ -505,12 +544,14 @@ async fn session_auto_forks_on_conflict() { let mut entries_written: usize = 1; // Simulate another Worker writing to the same segment behind our back. - let extra_entry = LogEntry::UserInput { - ts: 9999, - extensions: vec![], - segments: vec![protocol::Segment::text("Interloper")], - }; - store.append(sid, original_segid, &extra_entry).unwrap(); + session_store::save_user_input( + &store, + sid, + original_segid, + vec![protocol::Segment::text("Interloper")], + annotated(&[Item::user_message("Interloper")]), + ) + .unwrap(); // Now the on-disk count exceeds our tally — ensure_head_or_fork should auto-fork. session_store::ensure_head_or_fork( @@ -522,7 +563,7 @@ async fn session_auto_forks_on_conflict() { SegmentStartState { system_prompt: worker_a.get_system_prompt(), config: worker_a.request_config(), - history: &worker_a.history(), + history: annotated(&worker_a.history()), }, ) .unwrap(); @@ -543,7 +584,7 @@ async fn session_auto_forks_on_conflict() { // The new segment records its lineage forward via forked_from; the // source segment is left immutable (no terminal marker written back). match &fork_entries[0] { - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { forked_from: Some(origin), .. } => { @@ -563,7 +604,7 @@ async fn session_auto_forks_on_conflict() { ); let has_interloper = original_entries .iter() - .any(|e| matches!(e, LogEntry::UserInput { .. })); + .any(|e| matches!(e, LogEntry::AnnotatedUserInput { .. })); assert!(has_interloper); } @@ -581,7 +622,7 @@ async fn nested_past_fork_leaves_ancestors_immutable() { SegmentStartState { system_prompt: worker.get_system_prompt(), config: worker.request_config(), - history: &worker.history(), + history: annotated(&worker.history()), }, ) .unwrap(); @@ -618,7 +659,7 @@ async fn nested_past_fork_leaves_ancestors_immutable() { // fork2's lineage points at fork1, not the root. match &store.read_all(sid, fork2).unwrap()[0] { - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { forked_from: Some(origin), .. } => assert_eq!(origin.segment_id, fork1), diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index ba158e8b..0c620570 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -765,7 +765,7 @@ impl App { fn method_for_run(&mut self, segments: Vec) -> Method { // TurnHeader / UserMessage blocks are pushed only after the Worker - // emits `Event::UserMessage` from a committed `LogEntry::UserInput`. + // emits `Event::UserMessage` from a committed `LogEntry::AnnotatedUserInput`. // Locally we only clear the input buffer and forward the method, // while remembering enough local state to undo the visible submit if // the accepted run produced no assistant output and was rolled back. @@ -2937,6 +2937,17 @@ mod composer_history_persistence_tests { mod completion_flow_tests { use super::*; + fn annotated(item: agen::Item) -> session_store::LoggedHistoryEntry { + session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::from(item), + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId::new(), + origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } + } + #[test] fn typing_at_creates_completion_state_and_emits_query() { let mut app = App::new("test".into()); @@ -3239,7 +3250,7 @@ mod completion_flow_tests { #[test] fn committed_user_message_survives_fresh_segment_rotation() { let mut app = App::new("test".into()); - let start = session_store::LogEntry::SegmentStart { + let start = session_store::LogEntry::AnnotatedSegmentStart { ts: session_store::segment_log::now_millis(), session_id: uuid::Uuid::nil(), system_prompt: None, @@ -3498,14 +3509,14 @@ mod completion_flow_tests { #[test] fn snapshot_excludes_system_prompt_history_from_public_blocks() { let mut app = App::new("test".into()); - let session_start = session_store::LogEntry::SegmentStart { + let session_start = session_store::LogEntry::AnnotatedSegmentStart { ts: 1, session_id: uuid::Uuid::nil(), system_prompt: None, config: Default::default(), - history: vec![session_store::LoggedItem::from( - &agen::Item::system_message("[File: src/main.rs]\nfn main() {}"), - )], + history: vec![annotated(agen::Item::system_message( + "[File: src/main.rs]\nfn main() {}", + ))], forked_from: None, compacted_from: None, }; @@ -3584,7 +3595,7 @@ mod completion_flow_tests { code: ErrorCode::ProviderError, message: "provider unavailable".into(), }); - let segment_start = session_store::LogEntry::SegmentStart { + let segment_start = session_store::LogEntry::AnnotatedSegmentStart { ts: 5, session_id: uuid::Uuid::nil(), system_prompt: None, @@ -4336,36 +4347,33 @@ mod completion_flow_tests { }); let assistant_item_entries = vec![ - serde_json::json!({ - "kind": "assistant_item", - "ts": 1, - "item": { - "kind": "tool_call", - "call_id": "c1", - "name": "TaskCreate", - "arguments": r#"{"subject":"a","description":"A"}"#, - }, - }), - serde_json::json!({ - "kind": "assistant_item", - "ts": 2, - "item": { - "kind": "tool_call", - "call_id": "c2", - "name": "TaskCreate", - "arguments": r#"{"subject":"b","description":"B"}"#, - }, - }), - serde_json::json!({ - "kind": "assistant_item", - "ts": 3, - "item": { - "kind": "tool_call", - "call_id": "u1", - "name": "TaskUpdate", - "arguments": r#"{"taskid":2,"status":"inprogress"}"#, - }, - }), + serde_json::to_value(session_store::LogEntry::AnnotatedAssistantItem { + ts: 1, + entry: annotated(agen::Item::tool_call( + "c1", + "TaskCreate", + r#"{"subject":"a","description":"A"}"#, + )), + }) + .unwrap(), + serde_json::to_value(session_store::LogEntry::AnnotatedAssistantItem { + ts: 2, + entry: annotated(agen::Item::tool_call( + "c2", + "TaskCreate", + r#"{"subject":"b","description":"B"}"#, + )), + }) + .unwrap(), + serde_json::to_value(session_store::LogEntry::AnnotatedAssistantItem { + ts: 3, + entry: annotated(agen::Item::tool_call( + "u1", + "TaskUpdate", + r#"{"taskid":2,"status":"inprogress"}"#, + )), + }) + .unwrap(), ]; app.handle_worker_event(Event::Snapshot { greeting: test_greeting(), diff --git a/crates/tui/src/worker_list.rs b/crates/tui/src/worker_list.rs index fa2d834c..1a6a1be2 100644 --- a/crates/tui/src/worker_list.rs +++ b/crates/tui/src/worker_list.rs @@ -623,6 +623,17 @@ mod tests { const SOURCE: WorkerVisibilitySource = WorkerVisibilitySource::ResumePicker; + fn annotated(item: agen::Item) -> session_store::LoggedHistoryEntry { + session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::from(item), + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId::new(), + origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } + } + #[test] fn stored_metadata_summary_uses_segment_marker_without_reading_session_log() { let dir = tempdir().unwrap(); @@ -1207,7 +1218,7 @@ mod tests { .append( session_id, segment_id, - &LogEntry::SegmentStart { + &LogEntry::AnnotatedSegmentStart { ts, session_id, system_prompt: None, @@ -1231,9 +1242,10 @@ mod tests { .append( session_id, segment_id, - &LogEntry::UserInput { + &LogEntry::AnnotatedUserInput { ts, segments: vec![protocol::Segment::text(text)], + history: vec![annotated(agen::Item::user_message(text))], extensions: vec![], }, ) diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index d112d7df..e52d0bcf 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -67,8 +67,7 @@ const USER_INPUT_COMMIT_TIMEOUT: Duration = Duration::from_secs(9); fn user_input_has_submission(entry: &LogEntry, submission_id: &str) -> bool { let extensions = match entry { - LogEntry::UserInput { extensions, .. } - | LogEntry::AnnotatedUserInput { extensions, .. } => extensions, + LogEntry::AnnotatedUserInput { extensions, .. } => extensions, _ => return false, }; extensions.iter().any(|extension| { @@ -3244,8 +3243,7 @@ mod tests { assert!(entries.iter().any(|entry| { matches!( entry, - LogEntry::UserInput { segments, .. } - | LogEntry::AnnotatedUserInput { segments, .. } + LogEntry::AnnotatedUserInput { segments, .. } if segments == &vec![Segment::text("start the ticket")] ) })); @@ -3253,8 +3251,7 @@ mod tests { .iter() .find_map(|entry| { let extensions = match entry { - LogEntry::UserInput { extensions, .. } - | LogEntry::AnnotatedUserInput { extensions, .. } => extensions, + LogEntry::AnnotatedUserInput { extensions, .. } => extensions, _ => return None, }; extensions diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 4042ab68..41b33f44 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -631,8 +631,8 @@ fn protocol_command_status(status: WorkdirCommandStatus) -> ProtocolCommandStatu /// /// `Worker::wire_history_persistence` is called separately to wire the /// per-item history commit callback so every assistant / tool item -/// landing in `worker.history` becomes a singular `LogEntry::AssistantItem` -/// / `ToolResult` commit through the sync writer. +/// landing in `worker.history` becomes a singular `LogEntry::AnnotatedAssistantItem` +/// / `AnnotatedToolResult` commit through the sync writer. pub(crate) fn wire_event_bridges_on_engine( worker: &mut Worker, event_tx: &broadcast::Sender, @@ -1317,7 +1317,7 @@ async fn controller_loop( } // Stage the run without a speculative user-message echo. // `Worker::run` validates the input, commits - // `LogEntry::UserInput`, and the session-log sink turns that + // `LogEntry::AnnotatedUserInput`, and the session-log sink turns that // committed entry into the live `Event::UserMessage`. That // keeps every client ordered against `SegmentStart` replay and // makes persisted history the single source of visible user @@ -1343,7 +1343,7 @@ async fn controller_loop( Method::Notify { message, auto_run } => { // Client-side live echo is delivered as `Event::SystemItem` // once the interceptor commits the corresponding - // `LogEntry::SystemItem` entry — drained out of the + // `LogEntry::AnnotatedSystemItem` entry — drained out of the // notify buffer + broadcast through the sink. No // separate echo here. worker.push_notify(message, auto_run); diff --git a/crates/worker/src/feature/builtin/flow_transition.rs b/crates/worker/src/feature/builtin/flow_transition.rs index c307b919..163ac892 100644 --- a/crates/worker/src/feature/builtin/flow_transition.rs +++ b/crates/worker/src/feature/builtin/flow_transition.rs @@ -963,9 +963,12 @@ permission = "read" .append( session_id, segment_id, - &LogEntry::UserInput { + &LogEntry::AnnotatedUserInput { ts: 1, extensions: vec![], + history: vec![crate::session_history::test_logged_history_entry( + agen::Item::user_message("verify current Flow conditions"), + )], segments: vec![Segment::Text { content: "verify current Flow conditions".into(), }], diff --git a/crates/worker/src/hook.rs b/crates/worker/src/hook.rs index c5287b2d..d34ec5cc 100644 --- a/crates/worker/src/hook.rs +++ b/crates/worker/src/hook.rs @@ -161,7 +161,7 @@ impl From for TurnEndAction { /// /// Hook code can use this handle only when the Worker host includes it in an /// event-specific context. The handle queues typed requests; the host drains the -/// queue, commits each entry through `LogEntry::SystemItem`, and only then makes +/// queue, commits each entry through `LogEntry::AnnotatedSystemItem`, and only then makes /// the matching system message visible to the model. It deliberately exposes no /// raw `agen::Item`, history writer, event sender, `Worker`, `Engine`, or /// notification buffer. diff --git a/crates/worker/src/in_flight.rs b/crates/worker/src/in_flight.rs index ecf56f75..095fffb8 100644 --- a/crates/worker/src/in_flight.rs +++ b/crates/worker/src/in_flight.rs @@ -539,9 +539,9 @@ mod tests { text: "done".into(), }], }; - let assistant_entry = LogEntry::AssistantItem { + let assistant_entry = LogEntry::AnnotatedAssistantItem { ts: 1, - item: assistant_item.clone(), + entry: crate::session_history::test_logged_history_entry(assistant_item.clone()), }; let in_flight_guard = in_flight.snapshot_guard(); @@ -593,9 +593,9 @@ mod tests { text: "done".into(), }], }; - let assistant_entry = LogEntry::AssistantItem { + let assistant_entry = LogEntry::AnnotatedAssistantItem { ts: 1, - item: assistant_item.clone(), + entry: crate::session_history::test_logged_history_entry(assistant_item.clone()), }; in_flight.clear_for_committed_item_then(&assistant_item, || { @@ -608,7 +608,7 @@ mod tests { assert!(matches!( entries_snapshot.as_slice(), - [LogEntry::AssistantItem { item, .. }] if item == &assistant_item + [LogEntry::AnnotatedAssistantItem { entry, .. }] if entry.item == assistant_item )); assert!(in_flight_snapshot.is_empty()); } diff --git a/crates/worker/src/ipc/interceptor.rs b/crates/worker/src/ipc/interceptor.rs index a6283f25..f82444aa 100644 --- a/crates/worker/src/ipc/interceptor.rs +++ b/crates/worker/src/ipc/interceptor.rs @@ -60,7 +60,7 @@ pub(crate) struct WorkerInterceptor { pending_notifies: NotifyBuffer, /// Submit-scoped stash of resolver-produced typed system items. /// Drained inside `on_prompt_submit`, committed as - /// `LogEntry::SystemItem` entries through `log_writer`, and + /// `LogEntry::AnnotatedSystemItem` entries through `log_writer`, and /// returned to the worker as `Item::system_message` via /// `PromptAction::ContinueWith`. Populated by `Worker::run` /// immediately before handing off to the worker. @@ -71,7 +71,7 @@ pub(crate) struct WorkerInterceptor { /// Workspace scope associated with Prompt projection provenance. prompt_workspace_id: Option, /// Type-erased commit handle. The interceptor uses it to commit - /// `LogEntry::SystemItem` entries directly (sync) before + /// `LogEntry::AnnotatedSystemItem` entries directly (sync) before /// returning the corresponding `Item::system_message`s up to the /// worker. `None` in tests / `Worker::new` paths where no writer is /// attached. @@ -142,7 +142,7 @@ impl WorkerInterceptor { self } - /// Commit each `SystemItem` as its own `LogEntry::SystemItem` + /// Commit each `SystemItem` as its own `LogEntry::AnnotatedSystemItem` /// entry through the attached writer (no-op when no writer is /// wired). Sync — writes complete before the matching /// `Item::system_message`s reach the worker via @@ -540,7 +540,6 @@ mod tests { entry: session_store::LogEntry, ) -> Result<(), session_store::StoreError> { let item = match entry { - session_store::LogEntry::SystemItem { item, .. } => Some(item), session_store::LogEntry::AnnotatedSystemItem { entry, .. } => Some(entry.item), _ => None, }; diff --git a/crates/worker/src/ipc/notify_buffer.rs b/crates/worker/src/ipc/notify_buffer.rs index 0eb26c66..7b97e4ed 100644 --- a/crates/worker/src/ipc/notify_buffer.rs +++ b/crates/worker/src/ipc/notify_buffer.rs @@ -5,7 +5,7 @@ //! `WorkerInterceptor::pending_history_appends`, which the Engine calls //! at the head of each turn loop iteration. The drain renders each //! pending entry into a typed `SystemItem` (with the `notify_wrapper` -//! prompt applied), commits a `LogEntry::SystemItem` per entry through +//! prompt applied), commits a `LogEntry::AnnotatedSystemItem` per entry through //! the session-log sink, and returns the corresponding //! `Item::system_message`s for the worker to append to its //! persistent history. diff --git a/crates/worker/src/ipc/protocol_session.rs b/crates/worker/src/ipc/protocol_session.rs index 5e16069b..4368fa2b 100644 --- a/crates/worker/src/ipc/protocol_session.rs +++ b/crates/worker/src/ipc/protocol_session.rs @@ -29,18 +29,12 @@ pub fn subscribe_worker_protocol_session(handle: &WorkerHandle) -> WorkerProtoco pub fn live_log_entry_event(entry: LogEntry) -> Option { match entry { - entry @ (LogEntry::SegmentStart { .. } | LogEntry::AnnotatedSegmentStart { .. }) => { + entry @ LogEntry::AnnotatedSegmentStart { .. } => { let session = session_store::public_snapshot::project_current_session_snapshot(&[entry]); Some(Event::SegmentRotated { session }) } - LogEntry::UserInput { segments, .. } | LogEntry::AnnotatedUserInput { segments, .. } => { - Some(Event::UserMessage { segments }) - } - LogEntry::SystemItem { item, .. } => { - let value = serde_json::to_value(&item).expect("SystemItem is Serialize"); - Some(Event::SystemItem { item: value }) - } + LogEntry::AnnotatedUserInput { segments, .. } => Some(Event::UserMessage { segments }), LogEntry::AnnotatedSystemItem { entry, .. } => { let value = serde_json::to_value(&entry.item).expect("SystemItem is Serialize"); Some(Event::SystemItem { item: value }) @@ -89,9 +83,12 @@ mod tests { #[test] fn user_input_log_entry_maps_to_user_message_event() { let segments = vec![protocol::Segment::text("hello from log")]; - let event = live_log_entry_event(LogEntry::UserInput { + let event = live_log_entry_event(LogEntry::AnnotatedUserInput { ts: session_store::segment_log::now_millis(), extensions: vec![], + history: vec![crate::session_history::test_logged_history_entry( + agen::Item::user_message("hello from log"), + )], segments: segments.clone(), }) .expect("UserInput must be live-relevant"); diff --git a/crates/worker/src/ipc/server.rs b/crates/worker/src/ipc/server.rs index 62044ecf..88566597 100644 --- a/crates/worker/src/ipc/server.rs +++ b/crates/worker/src/ipc/server.rs @@ -77,7 +77,7 @@ async fn handle_connection(stream: tokio::net::UnixStream, handle: WorkerHandle) let mut writer = JsonLineWriter::new(writer); // Hold the in-flight stream lock while taking the session-log mirror - // snapshot. `LogEntry::AssistantItem` is mirror-only for live clients, + // snapshot. `LogEntry::AnnotatedAssistantItem` is mirror-only for live clients, // so a finalized assistant block must be observed either as an already // committed entry or as the still-present in-flight block. This lock // order matches `append_entry` (in-flight clear before sink publish) and diff --git a/crates/worker/src/segment_log_sink.rs b/crates/worker/src/segment_log_sink.rs index 77ee51eb..38192af7 100644 --- a/crates/worker/src/segment_log_sink.rs +++ b/crates/worker/src/segment_log_sink.rs @@ -50,7 +50,7 @@ struct SinkInner { /// Broadcast channel for live entry updates. The same `Sender` /// survives session swaps so existing subscribers keep their /// receiver — they observe the swap as a freshly broadcast - /// `LogEntry::SegmentStart` and reset their view accordingly. + /// `LogEntry::AnnotatedSegmentStart` and reset their view accordingly. broadcast_tx: broadcast::Sender, } @@ -89,9 +89,9 @@ impl SegmentLogSink { /// /// Live broadcast fires for committed session-log entries that /// socket clients must see in log order: - /// - `LogEntry::SegmentStart` → `Event::SegmentRotated` on the wire. - /// - `LogEntry::UserInput` → `Event::UserMessage`. - /// - `LogEntry::SystemItem` → `Event::SystemItem`. + /// - `LogEntry::AnnotatedSegmentStart` → `Event::SegmentRotated` on the wire. + /// - `LogEntry::AnnotatedUserInput` → `Event::UserMessage`. + /// - `LogEntry::AnnotatedSystemItem` → `Event::SystemItem`. /// - `LogEntry::Invoke` → `Event::InvokeStart`. /// Everything else (AssistantItem, ToolResult, TurnEnd, /// RunCompleted, RunErrored, PausedTurnAbandoned, LlmUsage, Extension, @@ -120,11 +120,8 @@ impl SegmentLogSink { fn is_live_relevant(entry: &LogEntry) -> bool { matches!( entry, - LogEntry::SegmentStart { .. } - | LogEntry::AnnotatedSegmentStart { .. } - | LogEntry::UserInput { .. } + LogEntry::AnnotatedSegmentStart { .. } | LogEntry::AnnotatedUserInput { .. } - | LogEntry::SystemItem { .. } | LogEntry::AnnotatedSystemItem { .. } | LogEntry::Invoke { .. } ) @@ -132,7 +129,7 @@ impl SegmentLogSink { /// Atomically swap the mirror to `[initial]` and broadcast the new /// session-start entry. Used during compaction / fork: the new - /// `LogEntry::SegmentStart` is the first entry of the replacement + /// `LogEntry::AnnotatedSegmentStart` is the first entry of the replacement /// session, and existing subscribers transition by replaying it /// like any other live entry. /// @@ -234,7 +231,7 @@ mod tests { use session_store::segment_log::now_millis; fn session_start() -> LogEntry { - LogEntry::SegmentStart { + LogEntry::AnnotatedSegmentStart { ts: now_millis(), session_id: uuid::Uuid::nil(), system_prompt: None, @@ -253,9 +250,12 @@ mod tests { } fn user_input(text: &str) -> LogEntry { - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: now_millis(), extensions: vec![], + history: vec![crate::session_history::test_logged_history_entry( + agen::Item::user_message(text), + )], segments: vec![protocol::Segment::Text { content: text.to_owned(), }], @@ -270,7 +270,10 @@ mod tests { let (snapshot, mut rx) = sink.subscribe_with_snapshot(); assert_eq!(snapshot.len(), 2); - assert!(matches!(snapshot[0], LogEntry::SegmentStart { .. })); + assert!(matches!( + snapshot[0], + LogEntry::AnnotatedSegmentStart { .. } + )); assert!(matches!( snapshot[1], LogEntry::TurnEnd { turn_count: 1, .. } @@ -279,13 +282,15 @@ mod tests { } fn notification_entry(text: &str) -> LogEntry { - LogEntry::SystemItem { + LogEntry::AnnotatedSystemItem { ts: now_millis(), - item: session_store::SystemItem::Notification { - message: text.to_owned(), - body: format!("[Notification] {text}"), - prompt_provenance: None, - }, + entry: crate::session_history::test_logged_system_entry( + session_store::SystemItem::Notification { + message: text.to_owned(), + body: format!("[Notification] {text}"), + prompt_provenance: None, + }, + ), } } @@ -305,7 +310,7 @@ mod tests { // for Event::UserMessage. sink.publish(user_input("hi from log")); match rx.try_recv() { - Ok(LogEntry::UserInput { segments, .. }) => { + Ok(LogEntry::AnnotatedUserInput { segments, .. }) => { assert_eq!(segments.len(), 1); } other => panic!("expected UserInput, got {other:?}"), @@ -314,7 +319,7 @@ mod tests { // SystemItem is live-relevant. sink.publish(notification_entry("hi")); match rx.try_recv() { - Ok(LogEntry::SystemItem { .. }) => {} + Ok(LogEntry::AnnotatedSystemItem { .. }) => {} other => panic!("expected SystemItem, got {other:?}"), } @@ -332,7 +337,7 @@ mod tests { assert_eq!(snapshot.len(), 1); match rx.try_recv() { - Ok(LogEntry::SystemItem { .. }) => {} + Ok(LogEntry::AnnotatedSystemItem { .. }) => {} other => panic!("unexpected: {other:?}"), } assert!(rx.try_recv().is_err()); @@ -348,13 +353,16 @@ mod tests { sink.reset_with_initial(session_start()); match rx.try_recv() { - Ok(LogEntry::SegmentStart { .. }) => {} + Ok(LogEntry::AnnotatedSegmentStart { .. }) => {} other => panic!("expected SegmentStart broadcast, got {other:?}"), } let (post_snapshot, _) = sink.subscribe_with_snapshot(); assert_eq!(post_snapshot.len(), 1); - assert!(matches!(post_snapshot[0], LogEntry::SegmentStart { .. })); + assert!(matches!( + post_snapshot[0], + LogEntry::AnnotatedSegmentStart { .. } + )); } #[test] diff --git a/crates/worker/src/session_history.rs b/crates/worker/src/session_history.rs index 3772d554..33ac6eed 100644 --- a/crates/worker/src/session_history.rs +++ b/crates/worker/src/session_history.rs @@ -5,7 +5,6 @@ //! retained only as explicit `LegacyUnknown` entries. use agen::{HistoryEntry, Item}; -use protocol::Segment; use session_store::{ LogEntry, LoggedHistoryDerivation, LoggedHistoryEntry, LoggedSessionHistoryEntryId, LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedWorkerSubject, SegmentId, @@ -18,6 +17,32 @@ pub type WorkerHistoryProvenance = LoggedSessionHistoryOrigin; pub type SessionHistoryDerivation = LoggedHistoryDerivation; pub type WorkerSubjectSnapshot = LoggedWorkerSubject; +#[cfg(test)] +pub(crate) fn test_logged_history_entry(item: impl Into) -> LoggedHistoryEntry { + LoggedHistoryEntry { + item: session_store::LoggedItem::from(item.into()), + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId::new(), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } +} + +#[cfg(test)] +pub(crate) fn test_logged_system_entry( + item: session_store::SystemItem, +) -> session_store::LoggedSystemHistoryEntry { + session_store::LoggedSystemHistoryEntry { + item, + metadata: LoggedSessionHistoryMetadata { + entry_id: LoggedSessionHistoryEntryId::new(), + origin: LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } +} + pub(crate) fn worker_subject(session_id: SessionId) -> WorkerSubjectSnapshot { WorkerSubjectSnapshot { workspace_id: None, @@ -53,10 +78,6 @@ pub(crate) fn to_logged_history_entry( } } -fn legacy_entry(item: Item) -> HistoryEntry { - HistoryEntry::new(item, SessionHistoryMetadata::legacy_unknown()) -} - fn from_logged(entry: &LoggedHistoryEntry) -> HistoryEntry { HistoryEntry::new(Item::from(entry.item.clone()), entry.metadata.clone()) } @@ -74,32 +95,15 @@ pub(crate) fn restore_history_entries( LogEntry::AnnotatedSegmentStart { history: seed, .. } => { history = seed.iter().map(from_logged).collect(); } - LogEntry::SegmentStart { history: seed, .. } => { - history = seed - .iter() - .cloned() - .map(Item::from) - .map(legacy_entry) - .collect(); - } LogEntry::AnnotatedUserInput { history: input, .. } => { history.extend(input.iter().map(from_logged)) } - LogEntry::UserInput { segments, .. } => history.push(legacy_entry(Item::user_message( - Segment::flatten_to_text(segments), - ))), LogEntry::AnnotatedAssistantItem { entry, .. } | LogEntry::AnnotatedToolResult { entry, .. } => history.push(from_logged(entry)), - LogEntry::AssistantItem { item, .. } | LogEntry::ToolResult { item, .. } => { - history.push(legacy_entry(Item::from(item.clone()))); - } LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new( entry.item.to_history_item(), entry.metadata.clone(), )), - LogEntry::SystemItem { item, .. } => { - history.push(legacy_entry(item.to_history_item())); - } _ => {} } } @@ -110,23 +114,9 @@ pub(crate) fn restore_history_entries( mod tests { use super::*; use agen::llm_client::RequestConfig; + use protocol::Segment; use session_store::LogEntry; - #[test] - fn legacy_user_role_is_not_inferred_as_human_authority() { - let entries = vec![LogEntry::UserInput { - ts: 1, - segments: vec![Segment::text("legacy")], - extensions: Vec::new(), - }]; - let restored = - restore_history_entries(SessionId::now_v7(), SegmentId::now_v7(), &entries).unwrap(); - assert!(matches!( - restored[0].annotation.origin, - WorkerHistoryProvenance::LegacyUnknown - )); - } - #[test] fn typed_flow_and_unknown_caller_input_round_trip_without_role_inference() { let session_id = SessionId::now_v7(); diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 8e0ea0f1..2c769f6e 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -112,8 +112,12 @@ impl InternalSpawnedWorkerRecord { fn stop_summary(&self) -> SubWorkerStopSummary { let mut counts = BTreeMap::::new(); for entry in self.session.entries() { - if let session_store::LogEntry::AssistantItem { - item: LoggedItem::ToolCall { name, .. }, + if let session_store::LogEntry::AnnotatedAssistantItem { + entry: + session_store::LoggedHistoryEntry { + item: LoggedItem::ToolCall { name, .. }, + .. + }, .. } = entry { @@ -1024,11 +1028,16 @@ mod tests { && worker.parent_session_id.as_deref() == Some("parent-session") && matches!(*event, Event::TextDone { ref text } if text == "answer") )); - record.session.publish_test_entry(LogEntry::UserInput { - ts: 1, - segments: vec![protocol::Segment::text("question")], - extensions: Vec::new(), - }); + record + .session + .publish_test_entry(LogEntry::AnnotatedUserInput { + ts: 1, + segments: vec![protocol::Segment::text("question")], + history: vec![crate::session_history::test_logged_history_entry( + agen::Item::user_message("question"), + )], + extensions: Vec::new(), + }); let committed = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv()) .await .unwrap() @@ -1159,14 +1168,18 @@ mod tests { let (mut record, _events) = record("child", InternalWorkerVisibility::ParentClient).await; record.change_tracker = Some(tracker); for (index, name) in ["Read", "Read", "Grep"].into_iter().enumerate() { - record.session.publish_test_entry(LogEntry::AssistantItem { - ts: index as u64, - item: LoggedItem::ToolCall { - call_id: format!("call-{index}"), - name: name.to_string(), - arguments: "{}".to_string(), - }, - }); + record + .session + .publish_test_entry(LogEntry::AnnotatedAssistantItem { + ts: index as u64, + entry: crate::session_history::test_logged_history_entry( + LoggedItem::ToolCall { + call_id: format!("call-{index}"), + name: name.to_string(), + arguments: "{}".to_string(), + }, + ), + }); } registry.start_protocol_forwarding(record.clone()); install_record(®istry, record); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 28f69623..80dd85b2 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -900,7 +900,6 @@ where self.state.increment_entries(); if let Some(in_flight) = &self.in_flight { let committed_item = match &entry { - LogEntry::AssistantItem { item, .. } => Some(item.clone()), LogEntry::AnnotatedAssistantItem { entry, .. } => Some(entry.item.clone()), _ => None, }; @@ -1207,11 +1206,11 @@ pub struct Worker { memory_task: Option>, /// Typed user submissions in submit order. K-th entry corresponds to /// the K-th `Item::user_message` in `worker.history()` (modulo seed - /// history loaded via `SegmentStart.history`, whose original segments + /// history loaded via `AnnotatedSegmentStart.history`, whose original segments /// are not preserved). Populated from log on `restore_from_manifest`, /// appended after `save_user_input` on each `run`. Pre-`Event::Snapshot` /// this fed `WorkerSharedState.user_segments`; the new wire format - /// carries typed atoms via `LogEntry::UserInput { segments }` so + /// carries typed atoms via `LogEntry::AnnotatedUserInput { segments }` so /// this remains purely an in-memory tracker for compact alignment. user_segments: Vec>, /// Worker-side session-log mirror + broadcast sink. Populated alongside @@ -1221,7 +1220,8 @@ pub struct Worker { sink: SegmentLogSink, /// `true` once `wire_history_persistence` has installed the /// `Engine::on_history_append` callback that commits each appended - /// item as a singular `LogEntry::AssistantItem` / `ToolResult` + /// item as a singular `LogEntry::AnnotatedAssistantItem` / + /// `AnnotatedToolResult` /// directly through the writer. Tests that drive `Worker::new` without /// going through the controller leave this `false`; `persist_turn` /// then walks the post-`history_before` slice inline so entries @@ -1345,15 +1345,16 @@ impl Worker { } /// Wire `Engine::on_history_append` to commit each appended item - /// directly as a singular `LogEntry::AssistantItem` / `ToolResult` + /// directly as a singular `LogEntry::AnnotatedAssistantItem` / + /// `AnnotatedToolResult` /// through the writer. The controller calls this once per spawned /// Worker after the worker is built; tests that drive `Worker::new` may /// opt in to the same wiring or leave it off (in which case /// `persist_turn`'s inline fallback writes entries at turn end). /// /// `user_message` items are skipped because they are committed - /// up-front via `commit_entry(LogEntry::UserInput { segments })`. - /// `role:system` items are committed as typed `LogEntry::SystemItem` + /// up-front via `commit_entry(LogEntry::AnnotatedUserInput { segments })`. + /// `role:system` items are committed as typed `LogEntry::AnnotatedSystemItem` /// entries by their producers (for example `WorkerInterceptor` and /// interrupted-turn prep) before they reach the worker's history, so this /// callback would otherwise double-write them. @@ -1941,8 +1942,7 @@ impl Worker { } let input = match entries.get(target.user_input_entry_index) { - Some(LogEntry::UserInput { segments, .. }) - | Some(LogEntry::AnnotatedUserInput { segments, .. }) => segments.clone(), + Some(LogEntry::AnnotatedUserInput { segments, .. }) => segments.clone(), _ => { return Err(RewindError::Invalid( "rewind target is no longer a user message".into(), @@ -2081,8 +2081,8 @@ impl Worker { /// Cheap clone via `Option`. /// Snapshot of the typed user segments tracked alongside worker /// history. The K-th entry corresponds to the K-th `Item::user_message` - /// derived from `LogEntry::UserInput` entries (post-compaction); seed - /// history loaded via `SegmentStart.history` does not contribute, + /// derived from `LogEntry::AnnotatedUserInput` entries (post-compaction); seed + /// history loaded via `AnnotatedSegmentStart.history` does not contribute, /// which is acceptable because the original segments are unrecoverable. pub fn user_segments(&self) -> &[Vec] { &self.user_segments @@ -3533,12 +3533,13 @@ impl Worker { // slice from `history_before` inline so the test's // `restore`-style assertions still see entries on disk. if !self.history_persistence_wired { - let new_items: Vec = self.session.history().entries()[history_before..] + let new_entries: Vec<_> = self.session.history().entries()[history_before..] .iter() - .map(|entry| entry.item.clone()) + .map(to_logged_history_entry) .collect(); let ts = segment_log::now_millis(); - for item in &new_items { + for history_entry in new_entries { + let item = Item::from(history_entry.item.clone()); if item.is_user_message() { continue; } @@ -3551,7 +3552,7 @@ impl Worker { ) { continue; } - let entry = session_store::classify_history_item(item, ts); + let entry = session_store::classify_logged_history_entry(history_entry, ts); self.commit_entry(entry)?; } } @@ -6257,8 +6258,7 @@ fn build_rewind_targets(segment_id: uuid::Uuid, entries: &[LogEntry]) -> Vec (segments, ts), + LogEntry::AnnotatedUserInput { segments, ts, .. } => (segments, ts), _ => continue, }; turn_index += 1; @@ -6300,8 +6300,7 @@ fn rewind_truncate_entries(entries: &[LogEntry], user_input_entry_index: usize) fn suffix_has_tool_side_effects(entries: &[LogEntry]) -> bool { entries.iter().any(|entry| match entry { - LogEntry::ToolResult { .. } | LogEntry::AnnotatedToolResult { .. } => true, - LogEntry::AssistantItem { item, .. } => logged_item_is_tool_call(item), + LogEntry::AnnotatedToolResult { .. } => true, LogEntry::AnnotatedAssistantItem { entry, .. } => logged_item_is_tool_call(&entry.item), _ => false, }) @@ -7636,7 +7635,7 @@ mod build_summary_prompt_tests { ); assert!(checkpoint.is_none()); - let mut replacement_entries = vec![LogEntry::SegmentStart { + let mut replacement_entries = vec![LogEntry::AnnotatedSegmentStart { ts: segment_log::now_millis(), session_id: uuid::Uuid::nil(), system_prompt: None, @@ -7964,9 +7963,12 @@ mod build_summary_prompt_tests { ); append_test_entry( worker, - LogEntry::UserInput { + LogEntry::AnnotatedUserInput { ts: ts + 1, extensions: vec![], + history: vec![crate::session_history::test_logged_history_entry( + Item::user_message(text), + )], segments: vec![text_segment(text)], }, ); @@ -7986,16 +7988,18 @@ mod build_summary_prompt_tests { append_user_turn(&worker, 20, "second message"); append_test_entry( &worker, - LogEntry::ToolResult { + LogEntry::AnnotatedToolResult { ts: 30, - item: session_store::LoggedItem::ToolResult { - call_id: "call-1".into(), - summary: "wrote a file".into(), - content: None, - attachments: Vec::new(), - disposition: Default::default(), - is_error: false, - }, + entry: crate::session_history::test_logged_history_entry( + session_store::LoggedItem::ToolResult { + call_id: "call-1".into(), + summary: "wrote a file".into(), + content: None, + attachments: Vec::new(), + disposition: Default::default(), + is_error: false, + }, + ), }, ); @@ -8029,16 +8033,18 @@ mod build_summary_prompt_tests { append_user_turn(&worker, 20, "second message"); append_test_entry( &worker, - LogEntry::ToolResult { + LogEntry::AnnotatedToolResult { ts: 30, - item: session_store::LoggedItem::ToolResult { - call_id: "call-1".into(), - summary: "wrote a file".into(), - content: None, - attachments: Vec::new(), - disposition: Default::default(), - is_error: false, - }, + entry: crate::session_history::test_logged_history_entry( + session_store::LoggedItem::ToolResult { + call_id: "call-1".into(), + summary: "wrote a file".into(), + content: None, + attachments: Vec::new(), + disposition: Default::default(), + is_error: false, + }, + ), }, ); let (head_entries, targets) = worker.list_rewind_targets().unwrap(); @@ -8456,9 +8462,9 @@ mod build_summary_prompt_tests { worker.wire_history_persistence(); let dangling_call = Item::tool_call("call-1", "SideEffect", "{}"); worker - .commit_entry(LogEntry::AssistantItem { + .commit_entry(LogEntry::AnnotatedAssistantItem { ts: segment_log::now_millis(), - item: dangling_call.clone().into(), + entry: crate::session_history::test_logged_history_entry(dangling_call.clone()), }) .unwrap(); worker.set_history_for_test(vec![dangling_call]); @@ -8863,9 +8869,12 @@ mod build_summary_prompt_tests { ); worker.set_history_for_test(vec![evidence.clone()]); worker - .commit_entry(LogEntry::UserInput { + .commit_entry(LogEntry::AnnotatedUserInput { ts: segment_log::now_millis(), extensions: vec![], + history: vec![crate::session_history::test_logged_history_entry( + evidence.clone(), + )], segments: vec![text_segment( "The cancellation regression must leave this evidence available for retry.", )], diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 67e81367..e67519fe 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -25,6 +25,17 @@ use worker::{Worker, WorkerController}; type TestStore = CombinedStore; +fn annotated(item: Item) -> session_store::LoggedHistoryEntry { + session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::from(item), + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId::new(), + origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, + derivation: None, + }, + } +} + #[derive(Clone)] struct MockClient { responses: Arc>>, @@ -210,7 +221,6 @@ fn system_texts_in_sink_session_start( .into_iter() .map(|entry| entry.item) .collect::>(), - session_store::LogEntry::SegmentStart { history, .. } => history, _ => continue, }; return history @@ -310,17 +320,14 @@ permission = "write" // Simulate a foreign writer appending to the same segment. This bumps // the on-disk entry count past the Worker's own append tally without // updating the Worker's `entries_written`. - store - .append( - session_id, - source_segment_id, - &LogEntry::UserInput { - ts: 9999, - segments: vec![protocol::Segment::text("interloper")], - extensions: vec![], - }, - ) - .unwrap(); + session_store::save_user_input( + &store, + session_id, + source_segment_id, + vec![protocol::Segment::text("interloper")], + vec![annotated(Item::user_message("interloper"))], + ) + .unwrap(); // Next run triggers ensure_segment_head, which sees the drift. worker.run_text("second").await.unwrap(); @@ -348,11 +355,6 @@ permission = "write" session_id: seg_session, forked_from: Some(origin), .. - } - | LogEntry::SegmentStart { - session_id: seg_session, - forked_from: Some(origin), - .. } => { assert_eq!(*seg_session, session_id); assert_eq!(origin.segment_id, source_segment_id); @@ -366,7 +368,7 @@ permission = "write" assert_eq!(source_after.len(), source_len_before + 1); assert!(matches!( source_after.last(), - Some(LogEntry::UserInput { .. }) + Some(LogEntry::AnnotatedUserInput { .. }) )); } diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 078fdc97..92191f3e 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -35,29 +35,16 @@ fn history_from_sink(handle: &WorkerHandle) -> Vec { LogEntry::AnnotatedSegmentStart { history, .. } => { items.extend(history.into_iter().map(|entry| Item::from(entry.item))); } - LogEntry::SegmentStart { history, .. } => { - items.extend(history.into_iter().map(Item::from)); - } LogEntry::AnnotatedUserInput { history, .. } => { items.extend(history.into_iter().map(|entry| Item::from(entry.item))); } - LogEntry::UserInput { segments, .. } => { - let text = protocol::Segment::flatten_to_text(&segments); - items.push(Item::user_message(text)); - } LogEntry::AnnotatedAssistantItem { entry, .. } | LogEntry::AnnotatedToolResult { entry, .. } => { items.push(Item::from(entry.item)); } - LogEntry::AssistantItem { item, .. } | LogEntry::ToolResult { item, .. } => { - items.push(Item::from(item)); - } LogEntry::AnnotatedSystemItem { entry, .. } => { items.push(entry.item.to_history_item()); } - LogEntry::SystemItem { item, .. } => { - items.push(item.to_history_item()); - } _ => {} } } @@ -67,7 +54,6 @@ fn history_from_sink(handle: &WorkerHandle) -> Vec { fn system_item(entry: &LogEntry) -> Option<&session_store::SystemItem> { match entry { LogEntry::AnnotatedSystemItem { entry, .. } => Some(&entry.item), - LogEntry::SystemItem { item, .. } => Some(item), _ => None, } } @@ -1095,7 +1081,7 @@ async fn run_with_paste_segment_inlines_content_and_emits_typed_user_message() { // Mixed input: plain text + a paste chip + trailing text. Worker must // flatten this into one user-message string (paste content inlined, // no `[Clipboard ...]` label leaking to the LLM); the committed - // `LogEntry::UserInput` must carry the typed segments unchanged so + // `LogEntry::AnnotatedUserInput` must carry the typed segments unchanged so // socket clients can derive `Event::UserMessage` and re-render the chip. let segments = vec![ protocol::Segment::text("see "), @@ -1130,7 +1116,7 @@ async fn run_with_paste_segment_inlines_content_and_emits_typed_user_message() { _ => {} }, entry = entry_rx.recv() => match entry { - Ok(session_store::LogEntry::UserInput { segments, .. } | session_store::LogEntry::AnnotatedUserInput { segments, .. }) => { + Ok(session_store::LogEntry::AnnotatedUserInput { segments, .. }) => { user_input_segments = Some(segments); if saw_turn_end { break;