From f8a7c46cf963680b97ce9f39640a21186f993fd2 Mon Sep 17 00:00:00 2001 From: Hare Date: Sun, 30 Aug 2026 09:42:46 +0900 Subject: [PATCH] fix(session): preflight migration and retain timestamps --- crates/protocol/src/lib.rs | 4 + crates/session-store/src/public_snapshot.rs | 45 ++++-- .../session-store/src/worker_session_store.rs | 91 ++++++++++- crates/worker-runtime/src/runtime.rs | 1 + .../src/feature/builtin/memory_extract.rs | 77 +++------- .../src/feature/builtin/worker_observation.rs | 1 + crates/worker/src/session_capture.rs | 143 +++++++++--------- crates/worker/src/spawn/comm_tools.rs | 1 + web/workspace/src/lib/generated/protocol.ts | 6 +- .../src/lib/workspace/console/model.test.ts | 1 + 10 files changed, 223 insertions(+), 147 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 339d1d67..bbda812a 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -394,6 +394,8 @@ pub struct SessionSnapshotEntry { /// Stable identity from durable history metadata, or a deterministic /// identity derived from the legacy segment and log position. pub entry_id: String, + /// Timestamp copied from the durable log record that commits this entry. + pub timestamp: u64, pub provenance: SessionEntryProvenance, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub derived_from: Vec, @@ -1536,6 +1538,7 @@ mod tests { session: SessionSnapshot { entries: vec![SessionSnapshotEntry { entry_id: "entry-1".into(), + timestamp: 1, provenance: SessionEntryProvenance::HumanInput, derived_from: Vec::new(), data: SessionSnapshotEntryData::UserInput { @@ -1565,6 +1568,7 @@ mod tests { parsed["data"]["session"]["entries"][0]["kind"], "user_input" ); + assert_eq!(parsed["data"]["session"]["entries"][0]["timestamp"], 1); assert_eq!(parsed["data"]["greeting"]["worker_name"], "test"); assert_eq!(parsed["data"]["greeting"]["tools"][0], "Read"); assert_eq!(parsed["data"]["greeting"]["context_window"], 200_000); diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index d25feb99..07e3a123 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -33,6 +33,7 @@ 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, .. @@ -41,57 +42,65 @@ pub fn project_session_snapshot(session_id: SessionId, log: &[LogEntry]) -> Sess 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, data)); + entries.push(legacy_entry(&session_key, log_index, item_index, *ts, data)); } } } LogEntry::AnnotatedSegmentStart { + ts, session_id, history, .. } => { session_key = *session_id; entries.clear(); - extend_history(&mut entries, history, None); + extend_history(&mut entries, history, None, *ts); } - LogEntry::UserInput { segments, .. } => entries.push(legacy_entry( + LogEntry::UserInput { ts, segments, .. } => entries.push(legacy_entry( &session_key, log_index, 0, + *ts, SessionSnapshotEntryData::UserInput { segments: segments.clone(), }, )), LogEntry::AnnotatedUserInput { - segments, history, .. - } => extend_history(&mut entries, history, Some(segments)), - LogEntry::AssistantItem { item, .. } | LogEntry::ToolResult { item, .. } => { + 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, data)); + entries.push(legacy_entry(&session_key, log_index, 0, *ts, data)); } } - LogEntry::AnnotatedAssistantItem { entry, .. } - | LogEntry::AnnotatedToolResult { entry, .. } => { + LogEntry::AnnotatedAssistantItem { ts, entry } + | LogEntry::AnnotatedToolResult { ts, entry } => { if let Some(data) = project_item(&entry.item) { - entries.push(history_entry(entry, data)); + entries.push(history_entry(entry, *ts, data)); } } - LogEntry::SystemItem { item, .. } => entries.push(system_entry( + LogEntry::SystemItem { ts, item } => entries.push(system_entry( item, legacy_entry_id(&session_key, log_index, 0), + *ts, SessionEntryProvenance::LegacyUnknown, Vec::new(), )), - LogEntry::AnnotatedSystemItem { entry, .. } => entries.push(system_entry( + LogEntry::AnnotatedSystemItem { ts, entry } => entries.push(system_entry( &entry.item, entry.metadata.entry_id.0.clone(), + *ts, provenance(&entry.metadata.origin), derivation_ids(entry), )), - LogEntry::RunErrored { message, .. } => entries.push(legacy_entry( + LogEntry::RunErrored { ts, message, .. } => entries.push(legacy_entry( &session_key, log_index, 0, + *ts, SessionSnapshotEntryData::RunError { message: message.clone(), }, @@ -116,6 +125,7 @@ fn extend_history( output: &mut Vec, history: &[LoggedHistoryEntry], input_segments: Option<&Vec>, + timestamp: u64, ) { let mut attached_segments = false; for entry in history { @@ -136,16 +146,18 @@ fn extend_history( }; data }; - output.push(history_entry(entry, data)); + output.push(history_entry(entry, timestamp, data)); } } fn history_entry( entry: &LoggedHistoryEntry, + timestamp: u64, data: SessionSnapshotEntryData, ) -> SessionSnapshotEntry { SessionSnapshotEntry { entry_id: entry.metadata.entry_id.0.clone(), + timestamp, provenance: provenance(&entry.metadata.origin), derived_from: entry .metadata @@ -182,10 +194,12 @@ fn legacy_entry( session_key: &SessionId, log_index: usize, item_index: usize, + timestamp: u64, data: SessionSnapshotEntryData, ) -> SessionSnapshotEntry { SessionSnapshotEntry { entry_id: legacy_entry_id(session_key, log_index, item_index), + timestamp, provenance: SessionEntryProvenance::LegacyUnknown, derived_from: Vec::new(), data, @@ -283,6 +297,7 @@ fn project_item(item: &LoggedItem) -> Option { fn system_entry( item: &SystemItem, entry_id: String, + timestamp: u64, provenance: SessionEntryProvenance, derived_from: Vec, ) -> SessionSnapshotEntry { @@ -298,6 +313,7 @@ fn system_entry( .to_owned(); SessionSnapshotEntry { entry_id, + timestamp, provenance, derived_from, data: SessionSnapshotEntryData::SystemItem { @@ -351,6 +367,7 @@ mod tests { let second = project_session_snapshot(session_id, &log); assert_eq!(first, second); assert_eq!(first.entries.len(), 1); + assert_eq!(first.entries[0].timestamp, 1); assert_eq!( first.entries[0].provenance, SessionEntryProvenance::LegacyUnknown diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index ddce9e1b..3fa0bae6 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -381,6 +381,15 @@ fn segment_log_paths(root: &Path) -> Result, StoreErro } fn migrate_segment_logs_to_v3(root: &Path, session_id: SessionId) -> Result<(), StoreError> { + struct MigrationPlan { + path: PathBuf, + source: Vec, + output: Vec, + } + + // Phase 1 is strictly read-only. Every segment must parse and canonicalize + // successfully before the first authoritative byte is replaced. + 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 { @@ -403,19 +412,30 @@ fn migrate_segment_logs_to_v3(root: &Path, session_id: SessionId) -> Result<(), serde_json::to_writer(&mut output, &entry)?; output.push(b'\n'); } + plans.push(MigrationPlan { + path, + source, + output, + }); + } - // Opening a Session is the exclusive restore boundary, but retain an - // unchanged-source fence so a racing writer cannot be silently lost. - if fs::read(&path)? != source { + // Fence the complete preflight snapshot before starting phase 2. Session + // open is the exclusive restore boundary; this additionally fails closed + // if an unexpected writer raced the preflight. + for plan in &plans { + if fs::read(&plan.path)? != plan.source { return Err(StoreError::Corrupt { line: 0, message: format!( "Worker Session segment changed during migration: {}", - path.display() + plan.path.display() ), }); } - atomic_write_bytes(&path, &output)?; + } + + for plan in plans { + atomic_write_bytes(&plan.path, &plan.output)?; } Ok(()) } @@ -774,12 +794,73 @@ mod tests { &reopened.read_all(session_id, segment_id).unwrap(), ); assert_eq!(snapshot.entries.len(), 3); + assert_eq!( + snapshot + .entries + .iter() + .map(|entry| entry.timestamp) + .collect::>(), + vec![1, 2, 3] + ); assert!(snapshot.entries.iter().all(|entry| { entry.provenance == protocol::SessionEntryProvenance::LegacyUnknown && entry.entry_id.len() <= 64 })); } + #[test] + fn schema_v2_preflight_keeps_earlier_segments_unchanged_when_later_is_corrupt() { + let root = tempfile::tempdir().unwrap(); + let session_id = new_session_id(); + let valid_segment = uuid::Uuid::from_u128(1); + let corrupt_segment = uuid::Uuid::from_u128(2); + fs::create_dir_all(root.path().join(SEGMENTS_DIR)).unwrap(); + atomic_write_json( + &root.path().join(SESSION_FILE), + &SessionManifest { + schema_version: PREVIOUS_SESSION_SCHEMA_VERSION, + session_id, + }, + ) + .unwrap(); + let manifest_before = fs::read(root.path().join(SESSION_FILE)).unwrap(); + + let valid_path = root + .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 mut valid_bytes = serde_json::to_vec(&valid_entry).unwrap(); + valid_bytes.push(b'\n'); + fs::write(&valid_path, &valid_bytes).unwrap(); + let corrupt_path = root + .path() + .join(SEGMENTS_DIR) + .join(format!("{corrupt_segment}.jsonl")); + fs::write(&corrupt_path, b"{not-json}\n").unwrap(); + let corrupt_before = fs::read(&corrupt_path).unwrap(); + + let error = match WorkerSessionStore::new(root.path()) { + Ok(_) => panic!("later corrupt segment must fail migration preflight"), + Err(error) => error, + }; + assert!(matches!(error, StoreError::Corrupt { .. })); + assert_eq!(fs::read(&valid_path).unwrap(), valid_bytes); + assert_eq!(fs::read(&corrupt_path).unwrap(), corrupt_before); + assert_eq!( + fs::read(root.path().join(SESSION_FILE)).unwrap(), + manifest_before + ); + } + #[test] fn schema_v3_rejects_legacy_records_and_new_writes_are_canonical() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index a1d2a99d..7ac91a3f 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -4588,6 +4588,7 @@ mod tests { session: protocol::SessionSnapshot { entries: vec![protocol::SessionSnapshotEntry { entry_id: "restored-log-entry".to_owned(), + timestamp: 1, provenance: protocol::SessionEntryProvenance::LegacyUnknown, derived_from: Vec::new(), data: protocol::SessionSnapshotEntryData::RunError { diff --git a/crates/worker/src/feature/builtin/memory_extract.rs b/crates/worker/src/feature/builtin/memory_extract.rs index 165efe6a..023dfab0 100644 --- a/crates/worker/src/feature/builtin/memory_extract.rs +++ b/crates/worker/src/feature/builtin/memory_extract.rs @@ -185,12 +185,9 @@ impl Tool for StageMemoryCandidateTool { })?); } if matches!(params.kind, CandidateKind::Preference) - && entries.iter().any(|entry| { - !matches!( - entry.origin, - crate::WorkerHistoryProvenance::HumanInput { .. } - ) - }) + && entries + .iter() + .any(|entry| !matches!(entry.origin, protocol::SessionEntryProvenance::HumanInput)) { return Err(ToolError::InvalidArgument( "preference candidates require exclusively HumanInput evidence; model, Worker, Flow, backend, derived, and legacy-unknown origins are not preference authority" @@ -324,10 +321,23 @@ fn evidence_kind(entry: &SessionEntryEvidence) -> EvidenceKind { } } -fn evidence_origin(origin: &crate::WorkerHistoryProvenance) -> EvidenceOrigin { - use crate::WorkerHistoryProvenance as Origin; - let mut evidence = EvidenceOrigin { - kind: EvidenceOriginKind::LegacyUnknown, +fn evidence_origin(origin: &protocol::SessionEntryProvenance) -> EvidenceOrigin { + use protocol::SessionEntryProvenance as Origin; + let kind = match origin { + Origin::HumanInput => EvidenceOriginKind::HumanInput, + Origin::WorkerInput => EvidenceOriginKind::WorkerInput, + Origin::FlowInstruction => EvidenceOriginKind::FlowInstruction, + Origin::BackendInstruction => EvidenceOriginKind::BackendInstruction, + Origin::ModelOutput => EvidenceOriginKind::ModelOutput, + Origin::ToolOutput => EvidenceOriginKind::ToolOutput, + Origin::DerivedSummary => EvidenceOriginKind::DerivedSummary, + Origin::LegacyUnknown => EvidenceOriginKind::LegacyUnknown, + }; + EvidenceOrigin { + kind, + // The public SessionSnapshot intentionally excludes account, Worker, + // Runtime, and Flow internals. Preserve the authenticated origin class + // without inventing missing control-plane identity fields. account_id: None, workspace_id: None, runtime_id: None, @@ -335,46 +345,7 @@ fn evidence_origin(origin: &crate::WorkerHistoryProvenance) -> EvidenceOrigin { flow_selector: None, flow_definition_id: None, flow_definition_revision: None, - }; - match origin { - Origin::HumanInput { account_id } => { - evidence.kind = EvidenceOriginKind::HumanInput; - evidence.account_id = Some(account_id.clone()); - } - Origin::WorkerInput { actor } => { - evidence.kind = EvidenceOriginKind::WorkerInput; - evidence.workspace_id = actor.workspace_id.clone(); - evidence.runtime_id = actor.runtime_id.clone(); - evidence.worker_id = Some(actor.worker_id.clone()); - } - Origin::FlowInstruction { - selector, - definition_id, - definition_revision, - .. - } => { - evidence.kind = EvidenceOriginKind::FlowInstruction; - evidence.flow_selector = Some(selector.clone()); - evidence.flow_definition_id = Some(definition_id.clone()); - evidence.flow_definition_revision = Some(*definition_revision); - } - Origin::BackendInstruction { .. } => evidence.kind = EvidenceOriginKind::BackendInstruction, - Origin::ModelOutput { worker } => { - evidence.kind = EvidenceOriginKind::ModelOutput; - evidence.workspace_id = worker.workspace_id.clone(); - evidence.runtime_id = worker.runtime_id.clone(); - evidence.worker_id = Some(worker.worker_id.clone()); - } - Origin::ToolOutput { worker } => { - evidence.kind = EvidenceOriginKind::ToolOutput; - evidence.workspace_id = worker.workspace_id.clone(); - evidence.runtime_id = worker.runtime_id.clone(); - evidence.worker_id = Some(worker.worker_id.clone()); - } - Origin::DerivedSummary => evidence.kind = EvidenceOriginKind::DerivedSummary, - Origin::LegacyUnknown => evidence.kind = EvidenceOriginKind::LegacyUnknown, } - evidence } fn staging_evidence(entry: &SessionEntryEvidence) -> StagingEvidence { @@ -502,12 +473,10 @@ mod tests { } #[test] - fn human_origin_projects_account_authority_into_evidence() { - let origin = evidence_origin(&crate::WorkerHistoryProvenance::HumanInput { - account_id: "account-1".into(), - }); + fn public_human_origin_preserves_class_without_inventing_account_authority() { + let origin = evidence_origin(&protocol::SessionEntryProvenance::HumanInput); assert_eq!(origin.kind, EvidenceOriginKind::HumanInput); - assert_eq!(origin.account_id.as_deref(), Some("account-1")); + assert_eq!(origin.account_id, None); } #[test] diff --git a/crates/worker/src/feature/builtin/worker_observation.rs b/crates/worker/src/feature/builtin/worker_observation.rs index ad59bb3f..2dc059f6 100644 --- a/crates/worker/src/feature/builtin/worker_observation.rs +++ b/crates/worker/src/feature/builtin/worker_observation.rs @@ -794,6 +794,7 @@ mod tests { }; Some(protocol::SessionSnapshotEntry { entry_id: format!("fake-{index:08}"), + timestamp: index as u64, provenance: protocol::SessionEntryProvenance::LegacyUnknown, derived_from: Vec::new(), data, diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index c0be48d0..19026fb8 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -109,7 +109,7 @@ impl ToolPart { #[derive(Debug, Clone)] pub(crate) struct OverviewItem { pub id: SessionEntryRef, - pub origin: WorkerHistoryProvenance, + pub origin: SessionEntryProvenance, pub entry_range: [u64; 2], pub kind: ReferenceKind, pub label: String, @@ -120,7 +120,7 @@ pub(crate) struct OverviewItem { #[derive(Debug, Clone)] pub(crate) struct ReferenceEntry { pub id: SessionEntryRef, - pub origin: WorkerHistoryProvenance, + pub origin: SessionEntryProvenance, pub entry_range: [u64; 2], pub kind: ReferenceKind, pub tool_part: Option, @@ -146,7 +146,7 @@ pub(crate) struct SearchOptions { #[derive(Debug, Clone)] pub(crate) struct SearchHit { pub id: SessionEntryRef, - pub origin: WorkerHistoryProvenance, + pub origin: SessionEntryProvenance, pub kind: ReferenceKind, pub tool_part: Option, pub tool_name: Option, @@ -192,7 +192,7 @@ impl Default for ReadOptions { #[derive(Debug, Clone)] pub(crate) struct ReadEntry { pub id: SessionEntryRef, - pub origin: WorkerHistoryProvenance, + pub origin: SessionEntryProvenance, pub kind: ReferenceKind, pub tool_part: Option, pub tool_name: Option, @@ -211,7 +211,7 @@ pub(crate) struct ReadResult { pub(crate) struct SessionEntryEvidence { pub segment_id: String, pub entry_ref: SessionEntryRef, - pub origin: WorkerHistoryProvenance, + pub origin: SessionEntryProvenance, pub entry_range: [u64; 2], pub kind: ReferenceKind, pub tool_part: Option, @@ -220,10 +220,17 @@ pub(crate) struct SessionEntryEvidence { pub excerpt: String, } +#[derive(Debug, Clone)] +struct CapturedHistoryEntry { + item: Item, + entry_id: session_store::LoggedSessionHistoryEntryId, + origin: SessionEntryProvenance, +} + #[derive(Debug, Clone)] pub(crate) struct SessionCapture { segment_id: String, - entries: Arc>>, + entries: Arc>, overview: Vec, index: Vec, } @@ -280,35 +287,49 @@ impl SessionCapture { SessionSnapshotEntryData::SystemItem { .. } | SessionSnapshotEntryData::RunError { .. } => return None, }; - Some(HistoryEntry::new( + Some(CapturedHistoryEntry { item, - public_snapshot_metadata(entry.entry_id, entry.provenance), - )) + entry_id: session_store::LoggedSessionHistoryEntryId(entry.entry_id), + origin: entry.provenance, + }) }) .collect(); - Self::from_history_entries(segment_id, entries) + Self::from_captured_entries(segment_id, entries) } pub(crate) fn new(segment_id: impl Into, items: Vec) -> Self { let entries = items .into_iter() .enumerate() - .map(|(index, item)| { - let mut metadata = SessionHistoryMetadata::legacy_unknown(); - metadata.entry_id = - session_store::LoggedSessionHistoryEntryId(format!("{index:08}")); - HistoryEntry::new(item, metadata) + .map(|(index, item)| CapturedHistoryEntry { + item, + entry_id: session_store::LoggedSessionHistoryEntryId(format!("{index:08}")), + origin: SessionEntryProvenance::LegacyUnknown, }) .collect(); - Self::from_history_entries(segment_id, entries) + Self::from_captured_entries(segment_id, entries) } pub(crate) fn from_history_entries( segment_id: impl Into, entries: Vec>, + ) -> Self { + let entries = entries + .into_iter() + .map(|entry| CapturedHistoryEntry { + item: entry.item, + entry_id: entry.annotation.entry_id, + origin: public_provenance(&entry.annotation.origin), + }) + .collect(); + Self::from_captured_entries(segment_id, entries) + } + + fn from_captured_entries( + segment_id: impl Into, + entries: Vec, ) -> Self { let segment_id = segment_id.into(); - let entries = Arc::new(entries); let mut overview = Vec::new(); let mut index = Vec::new(); @@ -317,7 +338,7 @@ impl SessionCapture { let entry_range = [idx as u64, idx as u64]; match item { Item::Message { role, content, .. } => { - let Some(kind) = message_reference_kind(&entry.annotation.origin, role) else { + let Some(kind) = message_reference_kind(&entry.origin, role) else { continue; }; let text = content @@ -327,10 +348,10 @@ impl SessionCapture { .join(""); let label = format!("{} message", kind.as_str()); let summary = truncate_chars(&text, 240); - let id = SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id); + let id = SessionEntryRef::from_history_entry_id(&entry.entry_id); index.push(ReferenceEntry { id: id.clone(), - origin: entry.annotation.origin.clone(), + origin: entry.origin.clone(), entry_range, kind, tool_part: None, @@ -342,7 +363,7 @@ impl SessionCapture { if matches!(kind, ReferenceKind::User | ReferenceKind::Assistant) { overview.push(OverviewItem { id: id.clone(), - origin: entry.annotation.origin.clone(), + origin: entry.origin.clone(), entry_range, kind, label, @@ -356,8 +377,8 @@ impl SessionCapture { } => { let text = format!("{name}\n{arguments}"); index.push(ReferenceEntry { - id: SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id), - origin: entry.annotation.origin.clone(), + id: SessionEntryRef::from_history_entry_id(&entry.entry_id), + origin: entry.origin.clone(), entry_range, kind: ReferenceKind::Tool, tool_part: Some(ToolPart::Input), @@ -383,8 +404,8 @@ impl SessionCapture { content.as_deref().unwrap_or_default(), ); index.push(ReferenceEntry { - id: SessionEntryRef::from_history_entry_id(&entry.annotation.entry_id), - origin: entry.annotation.origin.clone(), + id: SessionEntryRef::from_history_entry_id(&entry.entry_id), + origin: entry.origin.clone(), entry_range, kind: ReferenceKind::Tool, tool_part: Some(ToolPart::Output), @@ -424,7 +445,7 @@ impl SessionCapture { Self { segment_id, - entries, + entries: Arc::new(entries), overview, index, } @@ -607,65 +628,41 @@ impl SessionCapture { } } -fn public_snapshot_metadata( - entry_id: String, - provenance: SessionEntryProvenance, -) -> SessionHistoryMetadata { - let worker = session_store::LoggedWorkerSubject { - workspace_id: None, - runtime_id: None, - worker_id: "public-session-snapshot".to_owned(), - }; - let origin = match provenance { - SessionEntryProvenance::HumanInput => WorkerHistoryProvenance::HumanInput { - account_id: "public-session-snapshot".to_owned(), - }, - SessionEntryProvenance::WorkerInput => WorkerHistoryProvenance::WorkerInput { - actor: worker.clone(), - }, - SessionEntryProvenance::FlowInstruction => WorkerHistoryProvenance::FlowInstruction { - selector: "public-session-snapshot".to_owned(), - definition_id: "public-session-snapshot".to_owned(), - definition_revision: 0, - instance_id: "public-session-snapshot".to_owned(), - state_id: "public-session-snapshot".to_owned(), - }, - SessionEntryProvenance::BackendInstruction => { - WorkerHistoryProvenance::BackendInstruction { operation_id: None } +fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance { + match origin { + WorkerHistoryProvenance::HumanInput { .. } => SessionEntryProvenance::HumanInput, + WorkerHistoryProvenance::WorkerInput { .. } => SessionEntryProvenance::WorkerInput, + WorkerHistoryProvenance::FlowInstruction { .. } => SessionEntryProvenance::FlowInstruction, + WorkerHistoryProvenance::BackendInstruction { .. } => { + SessionEntryProvenance::BackendInstruction } - SessionEntryProvenance::ModelOutput => WorkerHistoryProvenance::ModelOutput { - worker: worker.clone(), - }, - SessionEntryProvenance::ToolOutput => WorkerHistoryProvenance::ToolOutput { worker }, - SessionEntryProvenance::DerivedSummary => WorkerHistoryProvenance::DerivedSummary, - SessionEntryProvenance::LegacyUnknown => WorkerHistoryProvenance::LegacyUnknown, - }; - SessionHistoryMetadata { - entry_id: session_store::LoggedSessionHistoryEntryId(entry_id), - origin, - derivation: None, + WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput, + WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput, + WorkerHistoryProvenance::DerivedSummary => SessionEntryProvenance::DerivedSummary, + WorkerHistoryProvenance::LegacyUnknown => SessionEntryProvenance::LegacyUnknown, } } fn message_reference_kind( - origin: &WorkerHistoryProvenance, + origin: &SessionEntryProvenance, provider_role: &Role, ) -> Option { match origin { - WorkerHistoryProvenance::HumanInput { .. } - | WorkerHistoryProvenance::WorkerInput { .. } => Some(ReferenceKind::User), - WorkerHistoryProvenance::ModelOutput { .. } => Some(ReferenceKind::Assistant), - WorkerHistoryProvenance::ToolOutput { .. } => Some(ReferenceKind::Tool), - WorkerHistoryProvenance::LegacyUnknown => match provider_role { + SessionEntryProvenance::HumanInput | SessionEntryProvenance::WorkerInput => { + Some(ReferenceKind::User) + } + SessionEntryProvenance::ModelOutput => Some(ReferenceKind::Assistant), + SessionEntryProvenance::ToolOutput => Some(ReferenceKind::Tool), + SessionEntryProvenance::LegacyUnknown => match provider_role { Role::User => Some(ReferenceKind::User), Role::Assistant => Some(ReferenceKind::Assistant), Role::System => None, }, // Flow/backend/system content remains out of the observation surface // even when represented with a provider user/system role. - WorkerHistoryProvenance::FlowInstruction { .. } - | WorkerHistoryProvenance::BackendInstruction { .. } - | WorkerHistoryProvenance::DerivedSummary => None, + SessionEntryProvenance::FlowInstruction + | SessionEntryProvenance::BackendInstruction + | SessionEntryProvenance::DerivedSummary => None, } } @@ -762,13 +759,13 @@ mod tests { assert_eq!(overview.len(), 1); assert!(matches!( overview[0].origin, - WorkerHistoryProvenance::HumanInput { .. } + SessionEntryProvenance::HumanInput )); let evidence = capture.evidence_for(overview[0].id.as_str()).unwrap(); assert!(evidence.excerpt.ends_with("remember my preference")); assert!(matches!( evidence.origin, - WorkerHistoryProvenance::HumanInput { .. } + SessionEntryProvenance::HumanInput )); } diff --git a/crates/worker/src/spawn/comm_tools.rs b/crates/worker/src/spawn/comm_tools.rs index 9d30d89b..49f9ab78 100644 --- a/crates/worker/src/spawn/comm_tools.rs +++ b/crates/worker/src/spawn/comm_tools.rs @@ -284,6 +284,7 @@ mod tests { .enumerate() .map(|(index, value)| protocol::SessionSnapshotEntry { entry_id: format!("test-{index}"), + timestamp: index as u64, provenance: protocol::SessionEntryProvenance::LegacyUnknown, derived_from: Vec::new(), data: protocol::SessionSnapshotEntryData::RunError { diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 39ab33c2..d25d2eb3 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -93,7 +93,11 @@ export type SessionSnapshotEntry = { * Stable identity from durable history metadata, or a deterministic * identity derived from the legacy segment and log position. */ -entry_id: string, provenance: SessionEntryProvenance, derived_from?: Array, } & ({ "kind": "user_input", segments: Array, } | { "kind": "message", role: SessionMessageRole, content: Array, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_error", message: string, }); +entry_id: string, +/** + * Timestamp copied from the durable log record that commits this entry. + */ +timestamp: number, provenance: SessionEntryProvenance, derived_from?: Array, } & ({ "kind": "user_input", segments: Array, } | { "kind": "message", role: SessionMessageRole, content: Array, } | { "kind": "tool_call", call_id: string, name: string, arguments: string, } | { "kind": "tool_result", call_id: string, summary: string, content?: string | null, is_error: boolean, attachments?: Array, } | { "kind": "system_item", item_kind: string, content: string, data?: unknown, } | { "kind": "run_error", message: string, }); export type SessionSnapshot = { entries: Array, }; diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index abc5c350..94c2c018 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -2108,6 +2108,7 @@ Deno.test("snapshot restores TaskStore state from system history", () => { event.data.session = { entries: [{ entry_id: "task-reminder-1", + timestamp: 1, provenance: "backend_instruction", kind: "system_item", item_kind: "task_reminder",