Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: stabilize persisted run transition identities

Browse Source
This commit is contained in:
Hare
2026-10-02 14:31:04 +09:00
parent b74f1b495d
commit 138a8f4cec
10 changed files with 635 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/session-store/src/fs_store.rs
+18 -10
View File
@@ -211,7 +211,11 @@ impl FsStore {
}
}
fn parse_jsonl<T: serde::de::DeserializeOwned>(content: &[u8]) -> Result<Vec<T>, StoreError> {
fn parse_jsonl(
content: &[u8],
session_id: SessionId,
segment_id: SegmentId,
) -> Result<Vec<LogEntry>, StoreError> {
let complete = Self::complete_jsonl_prefix(content);
let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt {
line: complete[..error.valid_up_to()]
@@ -222,15 +226,19 @@ impl FsStore {
message: error.to_string(),
})?;
let mut entries = Vec::new();
for (i, line) in content.lines().enumerate() {
if line.trim().is_empty() {
continue;
let mut record_start = 0_u64;
for (index, record) in content.split_inclusive('\n').enumerate() {
let line = record.strip_suffix('\n').unwrap_or(record);
if !line.trim().is_empty() {
let mut entry: LogEntry =
serde_json::from_str(line).map_err(|error| StoreError::Corrupt {
line: index + 1,
message: error.to_string(),
})?;
entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start);
entries.push(entry);
}
let entry: T = serde_json::from_str(line).map_err(|e| StoreError::Corrupt {
line: i + 1,
message: e.to_string(),
})?;
entries.push(entry);
record_start = record_start.saturating_add(record.len() as u64);
}
Ok(entries)
}
@@ -294,7 +302,7 @@ impl Store for FsStore {
return Err(StoreError::NotFound(segment_id));
}
let content = fs::read(&path)?;
Self::parse_jsonl(&content)
Self::parse_jsonl(&content, session_id, segment_id)
}
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
crates/session-store/src/history.rs
+4 -1
View File
@@ -4,7 +4,10 @@ use serde::{Deserialize, Serialize};
use crate::LoggedItem;
/// Stable logical identity of one model-visible history entry.
/// Stable logical identity of one durable public Session entry.
///
/// Model-visible history entries and typed Run transitions share this identity
/// space so every projection can reconcile the same committed record.
///
/// This value is generated at the trusted Worker session boundary and copied
/// unchanged across fork, rewind, compaction retention, restore, and reboot.
crates/session-store/src/public_snapshot.rs
+357 -10
View File
@@ -1041,15 +1041,16 @@ fn project_session_snapshot_for_segment(
)),
LogEntry::RunYielded {
ts,
entry_id,
reason,
active_run_turn_count: count,
} => {
active_run_turn_count = Some(*count);
entries.push(legacy_entry(
entries.push(run_transition_entry(
entry_id.as_ref(),
&session_key,
segment_id.as_ref(),
log_index,
0,
*ts,
SessionSnapshotEntryData::RunYielded {
reason: *reason,
@@ -1059,15 +1060,16 @@ fn project_session_snapshot_for_segment(
}
LogEntry::RunResumed {
ts,
entry_id,
source,
active_run_turn_count: count,
} => {
active_run_turn_count = Some(*count);
entries.push(legacy_entry(
entries.push(run_transition_entry(
entry_id.as_ref(),
&session_key,
segment_id.as_ref(),
log_index,
0,
*ts,
SessionSnapshotEntryData::RunResumed {
source: *source,
@@ -1075,29 +1077,30 @@ fn project_session_snapshot_for_segment(
},
));
}
LogEntry::RunCancelled { ts } => {
LogEntry::RunCancelled { ts, entry_id } => {
active_run_turn_count = None;
entries.push(legacy_entry(
entries.push(run_transition_entry(
entry_id.as_ref(),
&session_key,
segment_id.as_ref(),
log_index,
0,
*ts,
SessionSnapshotEntryData::RunCancelled,
));
}
LogEntry::RunErrored {
ts,
entry_id,
message,
failure,
..
} => {
active_run_turn_count = None;
entries.push(legacy_entry(
entries.push(run_transition_entry(
entry_id.as_ref(),
&session_key,
segment_id.as_ref(),
log_index,
0,
*ts,
SessionSnapshotEntryData::RunError {
message: message.clone(),
@@ -1234,6 +1237,25 @@ fn derivation_ids(entry: &crate::LoggedSystemHistoryEntry) -> Vec<String> {
.unwrap_or_default()
}
fn run_transition_entry(
persisted_entry_id: Option<&crate::LoggedSessionHistoryEntryId>,
session_key: &SessionId,
segment_id: Option<&SegmentId>,
log_index: usize,
timestamp: u64,
data: SessionSnapshotEntryData,
) -> SessionSnapshotEntry {
SessionSnapshotEntry {
entry_id: persisted_entry_id
.map(|entry_id| entry_id.0.clone())
.unwrap_or_else(|| legacy_entry_id(session_key, segment_id, log_index, 0)),
timestamp,
provenance: SessionEntryProvenance::LegacyUnknown,
derived_from: Vec::new(),
data,
}
}
fn legacy_entry(
session_key: &SessionId,
segment_id: Option<&SegmentId>,
@@ -2812,22 +2834,344 @@ mod tests {
));
}
fn run_entry_ids(entries: &[SessionSnapshotEntry]) -> Vec<String> {
entries
.iter()
.filter(|entry| {
matches!(
entry.data,
SessionSnapshotEntryData::RunYielded { .. }
| SessionSnapshotEntryData::RunResumed { .. }
| SessionSnapshotEntryData::RunCancelled
| SessionSnapshotEntryData::RunError { .. }
)
})
.map(|entry| entry.entry_id.clone())
.collect()
}
#[test]
fn saved_run_transition_ids_match_snapshot_retained_and_paged_history() {
let worker_name = "worker-run-identities";
let session_id = crate::new_session_id();
let segment_id = crate::new_segment_id();
let initial = vec![
segment_start(session_id, Vec::new(), None, None),
LogEntry::AnnotatedUserInput {
ts: 2,
segments: vec![Segment::text("compact")],
history: vec![user_message(1)],
extensions: Vec::new(),
},
];
let (_root, aggregate_root) = persist_history_fixture(
worker_name,
session_id,
segment_id,
vec![(segment_id, initial)],
);
let store = WorkerSessionStore::new(aggregate_root.join("session")).unwrap();
let transitions = [
LogEntry::RunYielded {
ts: 10,
entry_id: None,
reason: protocol::RunYieldReason::Compaction,
active_run_turn_count: 2,
},
LogEntry::RunResumed {
ts: 10,
entry_id: None,
source: protocol::RunResumeSource::Compaction,
active_run_turn_count: 2,
},
LogEntry::RunErrored {
ts: 10,
entry_id: None,
interrupted: false,
message: "mid-run compaction failed: unavailable".into(),
failure: Some(protocol::RunFailureKind::Compaction),
},
LogEntry::RunErrored {
ts: 10,
entry_id: None,
interrupted: false,
message: "mid-run compaction failed: unavailable".into(),
failure: Some(protocol::RunFailureKind::Compaction),
},
LogEntry::RunCancelled {
ts: 10,
entry_id: None,
},
];
for entry in transitions {
crate::segment::append_entry(&store, session_id, segment_id, entry).unwrap();
}
let stored = store.read_all(session_id, segment_id).unwrap();
let stored_ids = run_entry_ids(
&project_session_snapshot_for_segment(session_id, Some(segment_id), &stored).entries,
);
assert_eq!(stored_ids.len(), 5);
assert_eq!(stored_ids.iter().collect::<HashSet<_>>().len(), 5);
assert!(
stored_ids
.iter()
.all(|entry_id| !entry_id.starts_with("l-run-"))
);
let retained = read_retained_session_snapshot(
&aggregate_root,
worker_name,
DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
)
.unwrap();
assert_eq!(run_entry_ids(&retained.snapshot.entries), stored_ids);
let page = read_retained_session_history_page(
&aggregate_root,
worker_name,
None,
Some(1),
RetainedHistoryReadLimits::default(),
)
.unwrap();
assert_eq!(page.turns.len(), 1);
assert_eq!(run_entry_ids(&page.turns[0].entries), stored_ids);
let reconnect = read_retained_session_snapshot(
&aggregate_root,
worker_name,
DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
)
.unwrap();
assert_eq!(run_entry_ids(&reconnect.snapshot.entries), stored_ids);
}
#[test]
fn run_transition_ids_survive_segment_lineage_and_history_page_boundaries() {
let worker_name = "worker-run-lineage-identities";
let session_id = crate::new_session_id();
let source_segment = crate::new_segment_id();
let active_segment = crate::new_segment_id();
let source = vec![
segment_start(session_id, Vec::new(), None, None),
LogEntry::AnnotatedUserInput {
ts: 2,
segments: vec![Segment::text("source turn")],
history: vec![user_message(1)],
extensions: Vec::new(),
},
];
let (_root, aggregate_root) = persist_history_fixture(
worker_name,
session_id,
source_segment,
vec![(source_segment, source)],
);
let store = WorkerSessionStore::new(aggregate_root.join("session")).unwrap();
crate::segment::append_entry(
&store,
session_id,
source_segment,
LogEntry::RunYielded {
ts: 3,
entry_id: None,
reason: protocol::RunYieldReason::Compaction,
active_run_turn_count: 1,
},
)
.unwrap();
store
.append(
session_id,
source_segment,
&LogEntry::TurnEnd {
ts: 4,
turn_count: 1,
},
)
.unwrap();
store
.create_segment(
session_id,
active_segment,
&[
segment_start(
session_id,
vec![user_message(1)],
None,
Some(SegmentOrigin {
segment_id: source_segment,
at_turn_index: 1,
}),
),
LogEntry::AnnotatedUserInput {
ts: 5,
segments: vec![Segment::text("active turn")],
history: vec![user_message(2)],
extensions: Vec::new(),
},
],
)
.unwrap();
for entry in [
LogEntry::RunResumed {
ts: 6,
entry_id: None,
source: protocol::RunResumeSource::Compaction,
active_run_turn_count: 1,
},
LogEntry::RunErrored {
ts: 7,
entry_id: None,
interrupted: false,
message: "mid-run compaction failed".into(),
failure: Some(protocol::RunFailureKind::Compaction),
},
] {
crate::segment::append_entry(&store, session_id, active_segment, entry).unwrap();
}
WorkerAggregateStore::new(&aggregate_root, worker_name)
.unwrap()
.write(&crate::WorkerMetadata::new(
worker_name,
Some(crate::WorkerActiveSegmentRef::active_segment(
session_id,
active_segment,
)),
))
.unwrap();
let active_ids = run_entry_ids(
&project_session_snapshot_for_segment(
session_id,
Some(active_segment),
&store.read_all(session_id, active_segment).unwrap(),
)
.entries,
);
let source_ids = run_entry_ids(
&project_session_snapshot_for_segment(
session_id,
Some(source_segment),
&store.read_all(session_id, source_segment).unwrap(),
)
.entries,
);
assert_eq!(active_ids.len(), 2);
assert_eq!(source_ids.len(), 1);
let newest = read_retained_session_history_page(
&aggregate_root,
worker_name,
None,
Some(1),
RetainedHistoryReadLimits::default(),
)
.unwrap();
assert_eq!(run_entry_ids(&newest.turns[0].entries), active_ids);
assert!(newest.has_more);
let older = read_retained_session_history_page(
&aggregate_root,
worker_name,
newest.next_cursor.as_deref(),
Some(1),
RetainedHistoryReadLimits::default(),
)
.unwrap();
assert_eq!(run_entry_ids(&older.turns[0].entries), source_ids);
}
#[test]
fn legacy_run_transition_ids_use_stable_distinct_record_positions() {
let worker_name = "worker-legacy-run-identities";
let session_id = crate::new_session_id();
let segment_id = crate::new_segment_id();
let log = vec![
segment_start(session_id, Vec::new(), None, None),
LogEntry::AnnotatedUserInput {
ts: 2,
segments: vec![Segment::text("legacy")],
history: vec![user_message(1)],
extensions: Vec::new(),
},
LogEntry::RunErrored {
ts: 10,
entry_id: None,
interrupted: false,
message: "same legacy failure".into(),
failure: None,
},
LogEntry::RunErrored {
ts: 10,
entry_id: None,
interrupted: false,
message: "same legacy failure".into(),
failure: None,
},
];
let (_root, aggregate_root) =
persist_history_fixture(worker_name, session_id, segment_id, vec![(segment_id, log)]);
let log_path = aggregate_root
.join("session")
.join("segments")
.join(format!("{segment_id}.jsonl"));
let persisted_before = std::fs::read(&log_path).unwrap();
assert!(!String::from_utf8_lossy(&persisted_before).contains("entry_id\":\"l-run-"));
let retained = read_retained_session_snapshot(
&aggregate_root,
worker_name,
DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
)
.unwrap();
let snapshot_ids = run_entry_ids(&retained.snapshot.entries);
assert_eq!(snapshot_ids.len(), 2);
assert_ne!(snapshot_ids[0], snapshot_ids[1]);
assert!(
snapshot_ids
.iter()
.all(|entry_id| entry_id.starts_with("l-run-"))
);
let page = read_retained_session_history_page(
&aggregate_root,
worker_name,
None,
Some(1),
RetainedHistoryReadLimits::default(),
)
.unwrap();
assert_eq!(run_entry_ids(&page.turns[0].entries), snapshot_ids);
let reread = read_retained_session_snapshot(
&aggregate_root,
worker_name,
DEFAULT_RETAINED_SNAPSHOT_MAX_BYTES,
)
.unwrap();
assert_eq!(run_entry_ids(&reread.snapshot.entries), snapshot_ids);
assert_eq!(std::fs::read(log_path).unwrap(), persisted_before);
}
#[test]
fn run_transitions_project_with_timestamps_and_typed_causes() {
let session_id = crate::new_session_id();
let log = vec![
LogEntry::RunYielded {
ts: 10,
entry_id: None,
reason: protocol::RunYieldReason::Compaction,
active_run_turn_count: 2,
},
LogEntry::RunResumed {
ts: 20,
entry_id: None,
source: protocol::RunResumeSource::Compaction,
active_run_turn_count: 2,
},
LogEntry::RunErrored {
ts: 30,
entry_id: None,
interrupted: false,
message: "mid-run compaction failed: unavailable".into(),
failure: Some(protocol::RunFailureKind::Compaction),
@@ -2869,7 +3213,10 @@ mod tests {
ts: 10,
trigger: protocol::InvokeKind::UserSend,
},
LogEntry::RunCancelled { ts: 20 },
LogEntry::RunCancelled {
ts: 20,
entry_id: None,
},
],
);
crates/session-store/src/segment.rs
+5 -1
View File
@@ -347,6 +347,7 @@ pub fn save_run_yielded(
segment_id,
LogEntry::RunYielded {
ts: segment_log::now_millis(),
entry_id: None,
reason,
active_run_turn_count,
},
@@ -367,6 +368,7 @@ pub fn save_run_resumed(
segment_id,
LogEntry::RunResumed {
ts: segment_log::now_millis(),
entry_id: None,
source,
active_run_turn_count,
},
@@ -390,6 +392,7 @@ pub fn save_run_errored(
segment_id,
LogEntry::RunErrored {
ts: segment_log::now_millis(),
entry_id: None,
interrupted,
message,
failure: Some(RunFailureKind::Engine),
@@ -568,7 +571,8 @@ pub fn append_entry(
store: &impl Store,
session_id: SessionId,
segment_id: SegmentId,
entry: LogEntry,
mut entry: LogEntry,
) -> Result<(), StoreError> {
entry.ensure_session_entry_id();
store.append(session_id, segment_id, &entry)
}
crates/session-store/src/segment_log.rs
+63 -2
View File
@@ -140,6 +140,12 @@ pub enum LogEntry {
/// preceding Invoke and a later RunResumed or terminal record.
RunYielded {
ts: u64,
/// Stable public identity allocated before the record is committed.
/// `None` is accepted only when reading logs written before this field
/// existed; read paths derive the same legacy identity from the durable
/// record position without rewriting the log.
#[serde(default, skip_serializing_if = "Option::is_none")]
entry_id: Option<crate::LoggedSessionHistoryEntryId>,
reason: RunYieldReason,
active_run_turn_count: usize,
},
@@ -149,19 +155,27 @@ pub enum LogEntry {
/// It deliberately does not assert that a subsequent model call occurred.
RunResumed {
ts: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
entry_id: Option<crate::LoggedSessionHistoryEntryId>,
source: RunResumeSource,
active_run_turn_count: usize,
},
/// The logical Run ended because the user intentionally cancelled it.
/// This is terminal, but is not an execution failure.
RunCancelled { ts: u64 },
RunCancelled {
ts: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
entry_id: Option<crate::LoggedSessionHistoryEntryId>,
},
/// `run()` / `resume()` が `EngineError` で終了した。
/// `EngineError` は `Serialize` 不可なので `message` のみ lossy 保持する。
/// Audit-only metadata: replay は `interrupted` のみ反映する。
RunErrored {
ts: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
entry_id: Option<crate::LoggedSessionHistoryEntryId>,
interrupted: bool,
message: String,
/// Older records omit this field. New records distinguish failures in
@@ -236,6 +250,48 @@ pub struct SegmentOrigin {
pub at_turn_index: usize,
}
impl LogEntry {
/// Allocate the durable public identity for a newly committed Run
/// transition. Calling this repeatedly is idempotent so the exact value
/// written to storage is also the value published to live subscribers.
pub fn ensure_session_entry_id(&mut self) {
let entry_id = match self {
Self::RunYielded { entry_id, .. }
| Self::RunResumed { entry_id, .. }
| Self::RunCancelled { entry_id, .. }
| Self::RunErrored { entry_id, .. } => entry_id,
_ => return,
};
if entry_id.is_none() {
*entry_id = Some(crate::LoggedSessionHistoryEntryId::new());
}
}
/// Supply an in-memory compatibility identity for a legacy Run transition
/// that predates persisted entry IDs. The physical record position is
/// stable within the append-only Segment and distinguishes equal payloads.
/// This helper never writes the synthesized value back to the log.
pub(crate) fn ensure_legacy_session_entry_id(
&mut self,
session_id: crate::SessionId,
segment_id: crate::SegmentId,
record_start: u64,
) {
let entry_id = match self {
Self::RunYielded { entry_id, .. }
| Self::RunResumed { entry_id, .. }
| Self::RunCancelled { entry_id, .. }
| Self::RunErrored { entry_id, .. } => entry_id,
_ => return,
};
if entry_id.is_none() {
*entry_id = Some(crate::LoggedSessionHistoryEntryId(format!(
"l-run-{session_id}-{segment_id}-{record_start}"
)));
}
}
}
/// State collected from log entries.
#[derive(Debug, Clone)]
pub struct RestoredState {
@@ -878,6 +934,7 @@ mod tests {
},
LogEntry::RunYielded {
ts: 300,
entry_id: None,
reason: RunYieldReason::Compaction,
active_run_turn_count: 2,
},
@@ -888,6 +945,7 @@ mod tests {
},
LogEntry::RunResumed {
ts: 500,
entry_id: None,
source: RunResumeSource::Compaction,
active_run_turn_count: 2,
},
@@ -909,7 +967,10 @@ mod tests {
ts: 200,
turn_count: 1,
},
LogEntry::RunCancelled { ts: 300 },
LogEntry::RunCancelled {
ts: 300,
entry_id: None,
},
]);
assert!(!state.last_run_interrupted);
crates/session-store/src/worker_session_store.rs
+56 -6
View File
@@ -61,6 +61,8 @@ pub(crate) struct RetainedLogRecord {
pub(crate) struct RetainedSegmentReader {
file: File,
session_id: SessionId,
segment_id: SegmentId,
read_end: u64,
buffer_start: u64,
buffer: Vec<u8>,
@@ -103,7 +105,12 @@ impl RetainedSegmentReader {
message: "empty retained Segment log record".to_string(),
});
}
let entry = serde_json::from_slice(&record)?;
let mut entry: LogEntry = serde_json::from_slice(&record)?;
entry.ensure_legacy_session_entry_id(
self.session_id,
self.segment_id,
record_start,
);
return Ok((
Some(RetainedLogRecord {
start_offset: record_start,
@@ -255,7 +262,10 @@ impl WorkerSessionStore {
if byte_len > max_bytes {
return Err(StoreError::ReadLimitExceeded);
}
Ok((parse_jsonl(&bytes)?, byte_len))
Ok((
parse_session_log_jsonl(&bytes, session_id, segment_id)?,
byte_len,
))
}
pub(crate) fn read_first_log_record_read_only_bounded(
@@ -264,6 +274,7 @@ impl WorkerSessionStore {
segment_id: SegmentId,
max_bytes: u64,
) -> Result<(Option<RetainedLogRecord>, u64, u64), StoreError> {
let record_start = 0;
self.validate_retained_session(session_id)?;
let file = open_retained_file(&self.log_path(segment_id)).map_err(|error| {
if error.kind() == std::io::ErrorKind::NotFound {
@@ -289,7 +300,8 @@ impl WorkerSessionStore {
message: "empty retained Segment log record".to_string(),
});
}
let entry = serde_json::from_slice(&record)?;
let mut entry: LogEntry = serde_json::from_slice(&record)?;
entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start);
Ok((
Some(RetainedLogRecord {
start_offset: 0,
@@ -308,6 +320,7 @@ impl WorkerSessionStore {
start_offset: u64,
max_bytes: u64,
) -> Result<(Option<RetainedLogRecord>, u64), StoreError> {
let record_start = start_offset;
self.validate_retained_session(session_id)?;
let mut file = open_retained_file(&self.log_path(segment_id)).map_err(|error| {
if error.kind() == std::io::ErrorKind::NotFound {
@@ -334,7 +347,8 @@ impl WorkerSessionStore {
message: "empty retained Segment log record".to_string(),
});
}
let entry = serde_json::from_slice(&record)?;
let mut entry: LogEntry = serde_json::from_slice(&record)?;
entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start);
Ok((
Some(RetainedLogRecord {
start_offset,
@@ -353,6 +367,7 @@ impl WorkerSessionStore {
end_offset: u64,
max_bytes: u64,
) -> Result<RetainedLogRecord, StoreError> {
let record_start = start_offset;
self.validate_retained_session(session_id)?;
let record_len =
end_offset
@@ -387,7 +402,8 @@ impl WorkerSessionStore {
});
}
record.pop();
let entry = serde_json::from_slice(&record)?;
let mut entry: LogEntry = serde_json::from_slice(&record)?;
entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start);
Ok(RetainedLogRecord {
start_offset,
end_offset,
@@ -429,6 +445,8 @@ impl WorkerSessionStore {
}
Ok(RetainedSegmentReader {
file,
session_id,
segment_id,
read_end: before_offset,
buffer_start: before_offset,
buffer: Vec::new(),
@@ -645,7 +663,7 @@ impl Store for WorkerSessionStore {
if !path.exists() {
return Err(StoreError::NotFound(segment_id));
}
parse_jsonl(&fs::read(path)?)
parse_session_log_jsonl(&fs::read(path)?, session_id, segment_id)
}
fn list_sessions(&self) -> Result<Vec<SessionId>, StoreError> {
@@ -1154,6 +1172,38 @@ fn complete_jsonl_prefix(content: &[u8]) -> &[u8] {
.unwrap_or(&[])
}
fn parse_session_log_jsonl(
content: &[u8],
session_id: SessionId,
segment_id: SegmentId,
) -> Result<Vec<LogEntry>, StoreError> {
let complete = complete_jsonl_prefix(content);
let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt {
line: complete[..error.valid_up_to()]
.iter()
.filter(|byte| **byte == b'\n')
.count()
+ 1,
message: error.to_string(),
})?;
let mut entries = Vec::new();
let mut record_start = 0_u64;
for (index, record) in content.split_inclusive('\n').enumerate() {
let line = record.strip_suffix('\n').unwrap_or(record);
if !line.trim().is_empty() {
let mut entry: LogEntry =
serde_json::from_str(line).map_err(|error| StoreError::Corrupt {
line: index + 1,
message: error.to_string(),
})?;
entry.ensure_legacy_session_entry_id(session_id, segment_id, record_start);
entries.push(entry);
}
record_start = record_start.saturating_add(record.len() as u64);
}
Ok(entries)
}
fn parse_jsonl<T: serde::de::DeserializeOwned>(content: &[u8]) -> Result<Vec<T>, StoreError> {
let complete = complete_jsonl_prefix(content);
let content = std::str::from_utf8(complete).map_err(|error| StoreError::Corrupt {
crates/tui/src/app.rs
+1
View File
@@ -3800,6 +3800,7 @@ mod completion_flow_tests {
let run_errored = session_store::LogEntry::RunErrored {
ts: 3,
entry_id: None,
interrupted: false,
message: "provider unavailable".into(),
failure: None,
crates/worker/src/ipc/protocol_session.rs
+117
View File
@@ -117,11 +117,128 @@ pub async fn dispatch_worker_protocol_method(
#[cfg(test)]
mod tests {
use super::*;
use session_store::Store;
use crate::segment_log_sink::SegmentLogSink;
use crate::worker::{LogWriterHandle, SegmentState};
fn committed_entry(event: Event) -> protocol::SessionSnapshotEntry {
match event {
Event::SessionEntryCommitted { entry } => entry,
other => panic!("expected SessionEntryCommitted, got {other:?}"),
}
}
#[test]
fn persisted_run_transitions_keep_identity_across_live_and_snapshot_projection() {
let temp = tempfile::tempdir().unwrap();
let store = session_store::FsStore::new(temp.path()).unwrap();
let session_id = session_store::new_session_id();
let segment_id = session_store::new_segment_id();
let start = LogEntry::AnnotatedSegmentStart {
ts: 1,
session_id,
system_prompt: None,
config: agen::llm_client::RequestConfig::default(),
history: Vec::new(),
forked_from: None,
compacted_from: None,
};
store
.create_segment(session_id, segment_id, std::slice::from_ref(&start))
.unwrap();
let sink = SegmentLogSink::with_initial(vec![start]);
let writer = LogWriterHandle {
store: store.clone(),
state: SegmentState::new(session_id, segment_id, 1),
sink: sink.clone(),
in_flight: None,
};
let (_, mut receiver) = sink.subscribe_with_snapshot();
let transitions = [
LogEntry::RunYielded {
ts: 10,
entry_id: None,
reason: protocol::RunYieldReason::Compaction,
active_run_turn_count: 2,
},
LogEntry::RunResumed {
ts: 10,
entry_id: None,
source: protocol::RunResumeSource::Compaction,
active_run_turn_count: 2,
},
LogEntry::RunErrored {
ts: 10,
entry_id: None,
interrupted: false,
message: "same compaction failure".into(),
failure: Some(protocol::RunFailureKind::Compaction),
},
LogEntry::RunErrored {
ts: 10,
entry_id: None,
interrupted: false,
message: "same compaction failure".into(),
failure: Some(protocol::RunFailureKind::Compaction),
},
LogEntry::RunCancelled {
ts: 10,
entry_id: None,
},
];
let mut live_ids = Vec::new();
for transition in transitions {
writer.append_entry(transition).unwrap();
let committed = receiver.try_recv().expect("committed transition broadcast");
live_ids.push(
committed_entry(live_log_entry_event(committed).expect("live projection")).entry_id,
);
}
assert_eq!(
live_ids
.iter()
.collect::<std::collections::HashSet<_>>()
.len(),
5
);
let (mirror, _) = sink.subscribe_with_snapshot();
let snapshot = session_store::public_snapshot::project_current_session_snapshot(&mirror);
let snapshot_ids: Vec<_> = snapshot
.entries
.iter()
.filter(|entry| {
matches!(
entry.data,
protocol::SessionSnapshotEntryData::RunYielded { .. }
| protocol::SessionSnapshotEntryData::RunResumed { .. }
| protocol::SessionSnapshotEntryData::RunError { .. }
| protocol::SessionSnapshotEntryData::RunCancelled
)
})
.map(|entry| entry.entry_id.clone())
.collect();
assert_eq!(snapshot_ids, live_ids);
let persisted = store.read_all(session_id, segment_id).unwrap();
let persisted_snapshot =
session_store::public_snapshot::project_current_session_snapshot(&persisted);
let persisted_ids: Vec<_> = persisted_snapshot
.entries
.iter()
.filter(|entry| snapshot_ids.contains(&entry.entry_id))
.map(|entry| entry.entry_id.clone())
.collect();
assert_eq!(persisted_ids, live_ids);
}
#[test]
fn durable_run_transition_maps_to_session_entry_committed() {
let event = live_log_entry_event(LogEntry::RunYielded {
ts: 42,
entry_id: Some(session_store::LoggedSessionHistoryEntryId::new()),
reason: protocol::RunYieldReason::Compaction,
active_run_turn_count: 3,
})
crates/worker/src/segment_log_sink.rs
+1
View File
@@ -330,6 +330,7 @@ mod tests {
// Durable logical-Run transitions share the committed-entry lane.
sink.publish(LogEntry::RunResumed {
ts: now_millis(),
entry_id: None,
source: protocol::RunResumeSource::Compaction,
active_run_turn_count: 1,
});
crates/worker/src/worker.rs
+13 -2
View File
@@ -1426,7 +1426,8 @@ where
/// Append `entry` to the log: disk write → counter bump → in-memory
/// mirror push → broadcast. The Store owns physical write ordering and
/// partial-write recovery; publication happens only after it returns Ok.
pub fn append_entry(&self, entry: LogEntry) -> Result<(), StoreError> {
pub fn append_entry(&self, mut entry: LogEntry) -> Result<(), StoreError> {
entry.ensure_session_entry_id();
let _append_guard = self
.state
.append_lock
@@ -3055,7 +3056,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
/// Append `entry` to the session log AND publish it through the
/// broadcast sink. The Store is the commit boundary: a failed write is
/// never counted or published.
pub(crate) fn commit_entry(&self, entry: LogEntry) -> Result<(), StoreError> {
pub(crate) fn commit_entry(&self, mut entry: LogEntry) -> Result<(), StoreError> {
entry.ensure_session_entry_id();
let loc = self.segment_state.location();
self.store.append(loc.session_id, loc.segment_id, &entry)?;
self.segment_state.increment_entries();
@@ -4707,6 +4709,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
.expect("pause/yield resume must retain the logical-run turn budget");
self.commit_entry(LogEntry::RunResumed {
ts: segment_log::now_millis(),
entry_id: None,
source,
active_run_turn_count,
})?;
@@ -5024,6 +5027,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// marked as already delivered to clients through the session-log sink.
self.commit_entry(LogEntry::RunErrored {
ts: segment_log::now_millis(),
entry_id: None,
interrupted: false,
message,
failure: Some(RunFailureKind::Compaction),
@@ -5435,6 +5439,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
{
self.commit_entry(LogEntry::RunErrored {
ts: segment_log::now_millis(),
entry_id: None,
interrupted: false,
message: message.to_string(),
failure: Some(RunFailureKind::Compaction),
@@ -5458,6 +5463,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
EngineRunExit::Yielded => {
self.commit_entry(LogEntry::RunYielded {
ts: segment_log::now_millis(),
entry_id: None,
reason: RunYieldReason::Compaction,
active_run_turn_count: active_run_turn_count
.expect("yielded logical run must retain its turn budget"),
@@ -5474,11 +5480,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
EngineRunExit::Interrupted(RunInterruptionReason::Cancelled) => {
self.commit_entry(LogEntry::RunCancelled {
ts: segment_log::now_millis(),
entry_id: None,
})?;
}
EngineRunExit::Interrupted(reason) => {
self.commit_entry(LogEntry::RunErrored {
ts: segment_log::now_millis(),
entry_id: None,
interrupted,
message: run_interruption_reason_message(reason),
failure: Some(RunFailureKind::Engine),
@@ -11442,6 +11450,7 @@ mod build_summary_prompt_tests {
worker
.commit_entry(LogEntry::RunErrored {
ts: segment_log::now_millis(),
entry_id: None,
interrupted: true,
message: "provider failed".to_string(),
failure: Some(RunFailureKind::Engine),
@@ -11472,6 +11481,7 @@ mod build_summary_prompt_tests {
worker
.commit_entry(LogEntry::RunCancelled {
ts: segment_log::now_millis(),
entry_id: None,
})
.unwrap();
assert_eq!(
@@ -11482,6 +11492,7 @@ mod build_summary_prompt_tests {
worker
.commit_entry(LogEntry::RunResumed {
ts: segment_log::now_millis(),
entry_id: None,
source: RunResumeSource::Compaction,
active_run_turn_count: 1,
})