diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index c4973cea5..32155ea04 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -691,6 +691,13 @@ impl SystemPromptContributionSource for WorkspaceResidentSummarySource { } impl WorkspaceSubjektivResidentSummarySource { + fn confirm_resident_context_behavior(&self, behavior_md: &str) { + self.behavior_state + .lock() + .expect("subject behavior state poisoned") + .applied = Some(behavior_md.to_owned()); + } + async fn fetch( &self, ) -> Result { @@ -788,11 +795,14 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource })) } - fn confirm_resident_context_behavior(&self, behavior_md: &str) { - self.behavior_state - .lock() - .expect("subject behavior state poisoned") - .applied = Some(behavior_md.to_owned()); + fn system_item_committed( + &self, + item: &session_store::SystemItem, + _metadata: &crate::session_history::SessionHistoryMetadata, + ) { + if let session_store::SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item { + self.confirm_resident_context_behavior(behavior_md); + } } fn invalidate_resident_context_representation(&self) { @@ -1741,7 +1751,14 @@ permission = "write" .expect("changed behavior must produce one refresh"); assert_eq!(refresh.behavior_md, "Ask when uncertain."); assert!(refresh.body.contains("Ask when uncertain.")); - source.confirm_resident_context_behavior(&refresh.behavior_md); + source.system_item_committed( + &session_store::SystemItem::SubjectBehaviorRefresh { + body: refresh.body.clone(), + behavior_md: refresh.behavior_md.clone(), + prompt_provenance: None, + }, + &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), + ); assert!( source .load_changed_resident_context() @@ -1758,7 +1775,14 @@ permission = "write" .expect("cleared behavior must produce one refresh"); assert_eq!(cleared.behavior_md, ""); assert!(cleared.body.contains("No user-managed behavior is set")); - source.confirm_resident_context_behavior(&cleared.behavior_md); + source.system_item_committed( + &session_store::SystemItem::SubjectBehaviorRefresh { + body: cleared.body.clone(), + behavior_md: cleared.behavior_md.clone(), + prompt_provenance: None, + }, + &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), + ); assert!( source .load_changed_resident_context() @@ -1816,7 +1840,14 @@ permission = "write" .await .unwrap() .unwrap(); - source.confirm_resident_context_behavior(&first.behavior_md); + source.system_item_committed( + &session_store::SystemItem::SubjectBehaviorRefresh { + body: first.body.clone(), + behavior_md: first.behavior_md.clone(), + prompt_provenance: None, + }, + &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), + ); assert_eq!( source .load_changed_resident_context() @@ -1825,7 +1856,14 @@ permission = "write" .unwrap(), second ); - source.confirm_resident_context_behavior(&second.behavior_md); + source.system_item_committed( + &session_store::SystemItem::SubjectBehaviorRefresh { + body: second.body.clone(), + behavior_md: second.behavior_md.clone(), + prompt_provenance: None, + }, + &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), + ); assert!( source .load_changed_resident_context() diff --git a/crates/worker/src/ipc/interceptor.rs b/crates/worker/src/ipc/interceptor.rs index be4f18dda..cbb500796 100644 --- a/crates/worker/src/ipc/interceptor.rs +++ b/crates/worker/src/ipc/interceptor.rs @@ -216,15 +216,6 @@ impl WorkerInterceptor { self } - fn confirm_resident_context_item(&self, item: &SystemItem) { - let Some(source) = self.resident_context_source.as_ref() else { - return; - }; - if let SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item { - source.confirm_resident_context_behavior(behavior_md); - } - } - /// 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 @@ -243,16 +234,20 @@ impl WorkerInterceptor { return Ok(()); }; for (item, extensions, history_provenance) in items { - let entry = writer.commit_system_item_with_extensions( + writer.commit_system_item_with_extensions_then( item.clone(), extensions.clone(), history_provenance.clone(), + &mut |entry| { + self.pending_committed_history + .lock() + .expect("pending committed history poisoned") + .push_back(entry.clone()); + if let Some(source) = self.resident_context_source.as_ref() { + source.system_item_committed(item, &entry.annotation); + } + }, )?; - self.pending_committed_history - .lock() - .expect("pending committed history poisoned") - .push_back(entry); - self.confirm_resident_context_item(item); } Ok(()) } @@ -1003,8 +998,14 @@ mod tests { SystemPromptContribution::Unavailable } - fn confirm_resident_context_behavior(&self, behavior_md: &str) { - self.confirmed.lock().unwrap().push(behavior_md.to_owned()); + fn system_item_committed( + &self, + item: &SystemItem, + _metadata: &crate::session_history::SessionHistoryMetadata, + ) { + if let SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item { + self.confirmed.lock().unwrap().push(behavior_md.to_owned()); + } } } @@ -1556,6 +1557,83 @@ mod tests { )); } + #[test] + fn system_input_completion_reports_only_durable_batch_prefix() { + struct PrefixCommitter(Arc>>); + impl SystemItemCommitter for PrefixCommitter { + fn commit_log_entry( + &self, + entry: session_store::LogEntry, + ) -> Result<(), session_store::StoreError> { + let mut committed = self.0.lock().unwrap(); + if !committed.is_empty() { + return Err(std::io::Error::other("injected second append failure").into()); + } + committed.push(entry); + Ok(()) + } + } + struct Observer { + saved: Arc>>, + receipts: Arc>>, + } + #[async_trait] + impl SystemPromptContributionSource for Observer { + async fn load(&self) -> SystemPromptContribution { + SystemPromptContribution::Unavailable + } + fn system_item_committed( + &self, + item: &SystemItem, + metadata: &crate::session_history::SessionHistoryMetadata, + ) { + let saved = self.saved.lock().unwrap(); + let Some(session_store::LogEntry::AnnotatedSystemItem { entry, .. }) = saved.last() + else { + panic!("completion must follow persistence"); + }; + assert_eq!(&entry.metadata, metadata); + assert_eq!(entry.item.history_text(), item.history_text()); + self.receipts.lock().unwrap().push(metadata.clone()); + } + } + let saved = Arc::new(Mutex::new(Vec::new())); + let receipts = Arc::new(Mutex::new(Vec::new())); + let pending = Arc::new(Mutex::new(Vec::new())); + let interceptor = WorkerInterceptor::new( + Arc::new(HookRegistryBuilder::new().build()), + None, + None, + NotifyBuffer::new(), + pending.clone(), + test_prompts(), + Some(Arc::new(PrefixCommitter(saved.clone()))), + ) + .with_resident_context_source(Some(Arc::new(Observer { + saved: saved.clone(), + receipts: receipts.clone(), + }))); + let items = [ + SystemItem::Notification { + message: "first".into(), + body: "first saved input".into(), + prompt_provenance: None, + }, + SystemItem::FileAttachment { + path: "test.rs".into(), + body: "second unsaved input".into(), + }, + ]; + assert!(interceptor.commit_system_items(&items).is_err()); + assert_eq!(saved.lock().unwrap().len(), 1); + let receipts = receipts.lock().unwrap(); + assert_eq!(receipts.len(), 1); + let pending = interceptor.pending_committed_history.lock().unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].annotation, receipts[0]); + assert_eq!(pending[0].item.as_text(), Some("first saved input")); + } + #[test] fn behavior_ack_uses_raw_content_including_clear_not_rendered_context() { let committed = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 6af8d337b..a60803948 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2733,22 +2733,40 @@ pub trait SystemItemCommitter: Send + Sync { item: SystemItem, extensions: Vec, history_provenance: Option, + ) -> Result, StoreError> { + self.commit_system_item_with_extensions_then( + item, + extensions, + history_provenance, + &mut |_| {}, + ) + } + + /// Notify the producer inside the publication barrier, after the durable + /// append and before a rewrite can invalidate its representation. + fn commit_system_item_with_extensions_then( + &self, + item: SystemItem, + extensions: Vec, + history_provenance: Option, + after_commit: &mut dyn FnMut(&HistoryEntry), ) -> Result, StoreError> { let metadata = new_history_metadata( history_provenance .unwrap_or(WorkerHistoryProvenance::BackendInstruction { operation_id: None }), None, ); - let history_item = item.to_history_item(); - self.commit_log_entry(LogEntry::AnnotatedSystemItem { - ts: segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { - item, - metadata: metadata.clone(), + let history = HistoryEntry::new(item.to_history_item(), metadata.clone()); + self.commit_usage_boundary( + LogEntry::AnnotatedSystemItem { + ts: segment_log::now_millis(), + entry: session_store::LoggedSystemHistoryEntry { item, metadata }, + extensions, }, - extensions, - })?; - Ok(HistoryEntry::new(history_item, metadata)) + &mut || after_commit(&history), + &[], + )?; + Ok(history) } } @@ -2937,8 +2955,11 @@ pub(crate) trait SystemPromptContributionSource: Send + Sync { Ok(None) } - /// Acknowledges the exact behavior represented by a durably committed refresh. - fn confirm_resident_context_behavior(&self, _behavior_md: &str) {} + /// Reports a live durable system-input commit to its producer. Interpretation + /// belongs to the Feature; restoration never invokes this notification. + /// Runs under the publication barrier: only bounded local state updates are + /// allowed, without Host I/O or reentry into history mutation. + fn system_item_committed(&self, _item: &SystemItem, _metadata: &SessionHistoryMetadata) {} /// Invalidates the in-memory representation fence before history may be /// compacted or rewound. @@ -17436,6 +17457,57 @@ permission = "write" } } + #[test] + fn system_input_completion_observes_durable_identity_under_publication_barrier() { + let temp = tempfile::tempdir().unwrap(); + let store = session_store::FsStore::new(temp.path()).unwrap(); + let session_id = SessionId::now_v7(); + let segment_id = SegmentId::now_v7(); + let start = LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id, + system_prompt: None, + config: agen::llm_client::RequestConfig::default(), + history: vec![], + forked_from: None, + compacted_from: None, + }; + store + .create_segment(session_id, segment_id, std::slice::from_ref(&start)) + .unwrap(); + let writer = LogWriterHandle { + store: store.clone(), + state: SegmentState::new(session_id, segment_id, 1), + sink: SegmentLogSink::with_initial(vec![start]), + in_flight: None, + }; + let mut receipt = None; + let committed = writer + .commit_system_item_with_extensions_then( + SystemItem::Notification { + message: "message".into(), + body: "exact committed system body".into(), + prompt_provenance: None, + }, + vec![], + None, + &mut |history| { + assert!(writer.state.append_lock.try_lock().is_err()); + let durable = store.read_all(session_id, segment_id).unwrap(); + let Some(LogEntry::AnnotatedSystemItem { entry, .. }) = durable.last() else { + panic!("completion must follow actual storage append"); + }; + assert_eq!(entry.metadata, history.annotation); + assert_eq!(entry.item.to_history_item(), history.item); + assert_eq!(history.item.as_text(), Some("exact committed system body")); + receipt = Some(history.annotation.entry_id.clone()); + }, + ) + .unwrap(); + assert_eq!(receipt, Some(committed.annotation.entry_id)); + assert!(writer.state.append_lock.try_lock().is_ok()); + } + fn minimal_manifest() -> WorkerManifest { let toml_str = r#" [worker]