refactor: notify system input producers after durable commit

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<server_api::SubjektivResidentContextOutput, SubjektivHostError> {
@@ -788,11 +795,14 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource
         }))
     }
 
-    fn confirm_resident_context_behavior(&self, behavior_md: &str) {
+    fn system_item_committed(
-        self.behavior_state
+        &self,
-            .lock()
+        item: &session_store::SystemItem,
-            .expect("subject behavior state poisoned")
+        _metadata: &crate::session_history::SessionHistoryMetadata,
-            .applied = Some(behavior_md.to_owned());
+    ) {
+        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()

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) {
+        fn system_item_committed(
-            self.confirmed.lock().unwrap().push(behavior_md.to_owned());
+            &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<Mutex<Vec<session_store::LogEntry>>>);
+        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<Mutex<Vec<session_store::LogEntry>>>,
+            receipts: Arc<Mutex<Vec<crate::session_history::SessionHistoryMetadata>>>,
+        }
+        #[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()));

crates/worker/src/worker.rs

@@ -2733,22 +2733,40 @@ pub trait SystemItemCommitter: Send + Sync {
         item: SystemItem,
         extensions: Vec<SessionExtension>,
         history_provenance: Option<WorkerHistoryProvenance>,
+    ) -> Result<HistoryEntry<SessionHistoryMetadata>, 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<SessionExtension>,
+        history_provenance: Option<WorkerHistoryProvenance>,
+        after_commit: &mut dyn FnMut(&HistoryEntry<SessionHistoryMetadata>),
     ) -> Result<HistoryEntry<SessionHistoryMetadata>, StoreError> {
         let metadata = new_history_metadata(
             history_provenance
                 .unwrap_or(WorkerHistoryProvenance::BackendInstruction { operation_id: None }),
             None,
         );
-        let history_item = item.to_history_item();
+        let history = HistoryEntry::new(item.to_history_item(), metadata.clone());
-        self.commit_log_entry(LogEntry::AnnotatedSystemItem {
+        self.commit_usage_boundary(
-            ts: segment_log::now_millis(),
+            LogEntry::AnnotatedSystemItem {
-            entry: session_store::LoggedSystemHistoryEntry {
+                ts: segment_log::now_millis(),
-                item,
+                entry: session_store::LoggedSystemHistoryEntry { item, metadata },
-                metadata: metadata.clone(),
+                extensions,
             },
-            extensions,
+            &mut || after_commit(&history),
-        })?;
+            &[],
-        Ok(HistoryEntry::new(history_item, metadata))
+        )?;
+        Ok(history)
     }
 }
 
@@ -2937,8 +2955,11 @@ pub(crate) trait SystemPromptContributionSource: Send + Sync {
         Ok(None)
     }
 
-    /// Acknowledges the exact behavior represented by a durably committed refresh.
+    /// Reports a live durable system-input commit to its producer. Interpretation
-    fn confirm_resident_context_behavior(&self, _behavior_md: &str) {}
+    /// 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]