Merge remote-tracking branch 'origin/develop' into ticket/t-768-subject-grants

crates/worker/src/controller.rs

@@ -1615,10 +1615,10 @@ where
     if ordinary_subjektiv_features_enabled
         && lifecycle_enabled
         && let Some(host) = subjektiv_host.as_ref()
-        && host.host.journal_startup_granted()
+        && (host.host.journal_startup_granted() || host.host.journal_compaction_granted())
     {
-        worker.install_initial_context_source(Arc::new(
-            crate::feature::builtin::journal::JournalStartupContextSource::new(
+        worker.install_session_context_source(Arc::new(
+            crate::feature::builtin::journal::JournalContextSource::new(
                 host.clone(),
                 worker.prompts(),
             ),

crates/worker/src/feature/builtin.rs

@@ -6,6 +6,16 @@
 
 pub mod backend_job_result;
 pub(crate) mod bash;
+
+/// Host metadata participants are independent of model-facing Tool enablement.
+/// Disabling a Feature must not preserve stale Session-bound authority evidence.
+pub(crate) fn session_metadata_lifecycles()
+-> Vec<std::sync::Arc<dyn crate::metadata_lifecycle::SessionMetadataLifecycle>> {
+    vec![std::sync::Arc::new(
+        crate::subjektiv_metadata::SessionAttributionMetadata,
+    )]
+}
+
 pub mod chat_invocation;
 pub mod drive;
 pub mod flow_transition;

crates/worker/src/feature/builtin/journal.rs

@@ -23,7 +23,7 @@ use crate::hook::{
 use crate::subjektiv::{SubjektivHostConnection, SubjektivHostError};
 use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
 use async_trait::async_trait;
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 use std::sync::Arc;
 use subjektiv::journal::transport::*;
@@ -32,12 +32,73 @@ use subjektiv::journal::{
     MAX_JOURNAL_RESPONSE_BYTES,
 };
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct JournalStartupInput {
+    pub snapshot: subjektiv::journal::startup::JournalStartupSnapshot,
+    pub snapshot_sha256: String,
+    pub prompt_provenance: session_store::PromptRenderProvenance,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+enum JournalReadStatus {
+    Available,
+    Absent,
+    Disabled,
+    Unavailable,
+}
+impl JournalReadStatus {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Available => "available",
+            Self::Absent => "absent",
+            Self::Disabled => "disabled",
+            Self::Unavailable => "unavailable",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct JournalCompactionInput {
+    status: JournalReadStatus,
+    snapshot: Option<subjektiv::journal::startup::JournalStartupSnapshot>,
+    snapshot_sha256: Option<String>,
+    prompt_provenance: session_store::PromptRenderProvenance,
+}
+
+fn journal_input(
+    body: String,
+    payload: session_store::FeatureInputPayload,
+    kind: &str,
+) -> Result<session_store::SystemInputEnvelope, String> {
+    use session_store::*;
+    let public = PublicSystemItem {
+        kind: kind.into(),
+        body: body.clone(),
+        presentation: PublicSystemPresentation::Context,
+    };
+    let metadata = HistoryInputMetadata::new(
+        HistoryInputUsePolicy {
+            recording_material: InputEvidenceUse::Exclude,
+            memory_evidence: InputEvidenceUse::Exclude,
+            observation: InputObservationUse::Hidden,
+            compaction: InputCompactionUse::ReferenceInput,
+        },
+        Some(payload),
+        Some(public),
+    )
+    .map_err(|e| e.to_string())?;
+    SystemInputEnvelope::new(body, metadata).map_err(|e| e.to_string())
+}
+
 /// Subjektiv owns startup projection; Core only commits its typed initial items.
-pub(crate) struct JournalStartupContextSource {
+pub(crate) struct JournalContextSource {
     host: SubjektivHostConnection,
     prompts: Arc<arc_swap::ArcSwap<crate::prompt::catalog::PromptCatalog>>,
 }
-impl JournalStartupContextSource {
+impl JournalContextSource {
     pub(crate) fn new(
         host: SubjektivHostConnection,
         prompts: Arc<arc_swap::ArcSwap<crate::prompt::catalog::PromptCatalog>>,
@@ -46,9 +107,12 @@ impl JournalStartupContextSource {
     }
 }
 #[async_trait]
-impl crate::worker::InitialContextSource for JournalStartupContextSource {
+impl crate::worker::SessionContextSource for JournalContextSource {
     async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
         use sha2::{Digest, Sha256};
+        if !self.host.host.journal_startup_granted() {
+            return Ok(Vec::new());
+        }
         let snapshot = self
             .host
             .journal_startup_context()
@@ -58,8 +122,8 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
         let empty = snapshot.journals.is_empty() && !snapshot.omitted;
         // Hash the same canonical JSON value retained by SystemItem, so replay
         // can verify provenance without depending on domain struct field order.
-        let snapshot = serde_json::to_value(snapshot).map_err(|e| e.to_string())?;
-        let snapshot_json = serde_json::to_string(&snapshot).map_err(|e| e.to_string())?;
+        let snapshot_value = serde_json::to_value(&snapshot).map_err(|e| e.to_string())?;
+        let snapshot_json = serde_json::to_string(&snapshot_value).map_err(|e| e.to_string())?;
         if snapshot_json.len() > subjektiv::journal::startup::STARTUP_MAX_BYTES {
             return Err("Host Journal startup snapshot exceeds independent byte budget".into());
         }
@@ -74,9 +138,10 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
             return Err("Journal startup prompt exceeds rendered byte budget".into());
         }
         let projection = prompts.projection();
-        Ok(vec![session_store::LoggedSystemHistoryEntry {
-            item: session_store::SystemItem::JournalStartup {
-                body,
+        let payload = session_store::FeatureInputPayload::encode(
+            "journal:startup",
+            1,
+            &JournalStartupInput {
                 snapshot,
                 snapshot_sha256: Sha256::digest(snapshot_json.as_bytes())
                     .iter()
@@ -89,11 +154,38 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
                     logical_name: "internal.subjektiv_journal_startup".into(),
                 },
             },
-            metadata: crate::session_history::metadata(
-                session_store::LoggedSessionHistoryOrigin::JournalContext,
-                None,
-            ),
-        }])
+        )
+        .map_err(|e| e.to_string())?;
+        Ok(vec![session_store::LoggedSystemHistoryEntry::from_input(
+            session_store::LoggedSessionHistoryOrigin::Provided {
+                producer: "subjektiv.journal".into(),
+            },
+            None,
+            journal_input(body, payload, "journal_startup")?,
+        )])
+    }
+
+    async fn load_compaction_context(
+        &self,
+        session: session_store::SessionId,
+    ) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
+        // A changed Session must never reuse the old source's Host ownership.
+        let connection = if self.host.context.session_id == session.to_string() {
+            Ok(Some(self.host.clone()))
+        } else {
+            Err(std::io::Error::new(
+                std::io::ErrorKind::PermissionDenied,
+                "context source belongs to a different Session",
+            ))
+        };
+        let input = freeze_compaction_journal(connection, &self.prompts.load_full())?;
+        Ok(vec![session_store::LoggedSystemHistoryEntry::from_input(
+            session_store::LoggedSessionHistoryOrigin::Provided {
+                producer: "subjektiv.journal".into(),
+            },
+            None,
+            input,
+        )])
     }
 }
 
@@ -103,7 +195,8 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
 pub(crate) fn freeze_compaction_journal(
     connection: std::io::Result<Option<SubjektivHostConnection>>,
     prompts: &crate::prompt::catalog::PromptCatalog,
-) -> Result<String, crate::prompt::catalog::CatalogError> {
+) -> Result<session_store::SystemInputEnvelope, String> {
+    use JournalReadStatus::*;
     let workspace_id = connection.as_ref().ok().and_then(|connection| {
         connection
             .as_ref()
@@ -116,24 +209,25 @@ pub(crate) fn freeze_compaction_journal(
         Ok(_) => None,
         Err(_) => Some(Err(())),
     };
+    let snapshot = read.as_ref().and_then(|read| read.as_ref().ok()).cloned();
     let (status, json) = match read {
-        None => ("disabled", String::new()),
+        None => (Disabled, String::new()),
         Some(Ok(snapshot)) => {
             let status = if snapshot.journals.is_empty() {
-                "absent"
+                Absent
             } else {
-                "available"
+                Available
             };
             match serde_json::to_string(&snapshot) {
                 Ok(json) if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES => {
                     (status, json)
                 }
-                _ => ("unavailable", String::new()),
+                _ => (Unavailable, String::new()),
             }
         }
-        Some(Err(())) => ("unavailable", String::new()),
+        Some(Err(())) => (Unavailable, String::new()),
     };
-    if status == "unavailable" {
+    if status == Unavailable {
         tracing::warn!("Journal compaction snapshot unavailable; continuing without Journal");
     }
     use sha2::{Digest, Sha256};
@@ -143,12 +237,35 @@ pub(crate) fn freeze_compaction_journal(
             .map(|byte| format!("{byte:02x}"))
             .collect()
     });
-    prompts.subjektiv_journal_compaction(
-        status,
-        &json,
-        snapshot_sha256.as_deref(),
-        workspace_id.as_deref(),
+    let body = prompts
+        .subjektiv_journal_compaction(
+            status.as_str(),
+            &json,
+            snapshot_sha256.as_deref(),
+            workspace_id.as_deref(),
+        )
+        .map_err(|e| e.to_string())?;
+    let payload = session_store::FeatureInputPayload::encode(
+        "journal:compaction",
+        1,
+        &JournalCompactionInput {
+            status,
+            prompt_provenance: session_store::PromptRenderProvenance {
+                workspace_id,
+                source_digest: prompts.projection().source_digest.clone(),
+                projection_digest: prompts.projection().catalog_digest.clone(),
+                logical_name: "internal.subjektiv_journal_compaction".into(),
+            },
+            snapshot: if status == Unavailable {
+                None
+            } else {
+                snapshot
+            },
+            snapshot_sha256,
+        },
     )
+    .map_err(|e| e.to_string())?;
+    journal_input(body, payload, "journal_compaction")
 }
 
 #[derive(Clone)]

crates/worker/src/feature/builtin/memory.rs

@@ -43,6 +43,80 @@ use crate::worker::{
     WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
 };
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct SubjectBehaviorInput {
+    pub behavior_md: String,
+    pub prompt_provenance: Option<session_store::PromptRenderProvenance>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ResidentSummaryInput {
+    prompt_provenance: Option<session_store::PromptRenderProvenance>,
+}
+
+fn resident_input(
+    body: String,
+    schema: &str,
+    payload: &impl serde::Serialize,
+    kind: &str,
+) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
+    use session_store::*;
+    let input = SystemInputEnvelope::new(
+        body.clone(),
+        HistoryInputMetadata::new(
+            HistoryInputUsePolicy {
+                recording_material: InputEvidenceUse::Exclude,
+                memory_evidence: InputEvidenceUse::Exclude,
+                observation: InputObservationUse::Hidden,
+                compaction: InputCompactionUse::Retain,
+            },
+            Some(FeatureInputPayload::encode(schema, 1, payload)?),
+            Some(PublicSystemItem {
+                kind: kind.into(),
+                body,
+                presentation: PublicSystemPresentation::Context,
+            }),
+        )?,
+    )?;
+    Ok(LoggedSystemHistoryEntry::from_input(
+        LoggedSessionHistoryOrigin::Provided {
+            producer: "subjektiv.resident-context".into(),
+        },
+        None,
+        input,
+    ))
+}
+
+pub(crate) fn subject_behavior_input(
+    body: String,
+    behavior_md: String,
+    prompt_provenance: Option<session_store::PromptRenderProvenance>,
+) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
+    resident_input(
+        body,
+        "subject:behavior",
+        &SubjectBehaviorInput {
+            behavior_md,
+            prompt_provenance,
+        },
+        "subject_behavior_refresh",
+    )
+}
+
+pub(crate) fn resident_summary_input(
+    body: String,
+    prompt_provenance: Option<session_store::PromptRenderProvenance>,
+) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
+    resident_input(
+        body,
+        "memory:resident-summary",
+        &ResidentSummaryInput { prompt_provenance },
+        "resident_summary_refresh",
+    )
+}
+
 #[derive(Clone, Debug)]
 pub struct WorkspaceHttpMemoryBackend {
     client: Arc<dyn WorkspaceClient>,
@@ -651,7 +725,7 @@ struct SubjectBehaviorRefreshState {
 struct WorkspaceSubjektivResidentSummarySource {
     host: SubjektivHostConnection,
     prompts: Arc<ArcSwap<PromptCatalog>>,
-    behavior_state: Mutex<SubjectBehaviorRefreshState>,
+    behavior_state: Arc<Mutex<SubjectBehaviorRefreshState>>,
 }
 
 fn resident_summary_contribution(
@@ -691,6 +765,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> {
@@ -707,7 +788,7 @@ impl WorkspaceSubjektivResidentSummarySource {
     fn render(
         &self,
         output: &server_api::SubjektivResidentContextOutput,
-    ) -> Result<String, String> {
+    ) -> Result<(String, session_store::PromptRenderProvenance), String> {
         use memory::backend::MemoryResidentSummaryAvailability as Availability;
         let availability = match output.memory_surface.availability {
             Availability::Ready => "ready",
@@ -716,53 +797,68 @@ impl WorkspaceSubjektivResidentSummarySource {
             Availability::Failed => "failed",
             Availability::Unavailable => "unavailable",
         };
-        self.prompts
-            .load()
+        let prompts = self.prompts.load_full();
+        let body = prompts
             .subjektiv_resident_context(
                 &output.behavior_md,
                 availability,
                 output.memory_surface.content.as_deref(),
             )
-            .map_err(|error| error.to_string())
-    }
-
-    fn remember_behavior(&self, output: &server_api::SubjektivResidentContextOutput) {
-        let mut state = self
-            .behavior_state
-            .lock()
-            .expect("subject behavior state poisoned");
-        state.applied = Some(output.behavior_md.clone());
+            .map_err(|error| error.to_string())?;
+        let projection = prompts.projection();
+        Ok((
+            body,
+            session_store::PromptRenderProvenance {
+                workspace_id: self.host.host.workspace_id(),
+                source_digest: projection.source_digest.clone(),
+                projection_digest: projection.catalog_digest.clone(),
+                logical_name: "internal.subjektiv_resident_context".into(),
+            },
+        ))
     }
 }
 
 /// Detect behavior changes against the state reflected in history, not the fetched state.
 ///
-/// Initial load builds a prompt contribution; change refresh does not advance applied on fetch/render alone.
-/// The interceptor confirms after durably appending a system item, allowing refetch if append fails.
-/// Invalidate applied before rewrite so even identical bodies can be presented in the new history.
+/// Fetch/render never advances applied state. Initial preparation captures a
+/// source-local completion, invoked only after its prompt and active pointer commit.
+/// Later refreshes are acknowledged by the interceptor after durable append.
+/// Invalidate applied before rewrite so identical bodies can be represented again.
 #[async_trait]
 impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource {
     async fn load(&self) -> SystemPromptContribution {
+        self.load_initial().await.0
+    }
+
+    async fn load_initial(
+        &self,
+    ) -> (
+        SystemPromptContribution,
+        Option<crate::worker::PromptContributionCompletion>,
+    ) {
         match self.fetch().await {
             Ok(output) => match self.render(&output) {
-                Ok(rendered) => {
-                    self.remember_behavior(&output);
-                    SystemPromptContribution::Ready(rendered)
+                Ok((rendered, _)) => {
+                    let state = self.behavior_state.clone();
+                    let behavior = output.behavior_md;
+                    (
+                        SystemPromptContribution::Ready(rendered),
+                        Some(Box::new(move || {
+                            state
+                                .lock()
+                                .expect("subject behavior state poisoned")
+                                .applied = Some(behavior);
+                        })),
+                    )
                 }
                 Err(error) => {
                     tracing::error!(%error, "subject resident context render failed");
-                    SystemPromptContribution::Ready(
-                        "The Host could not render the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory."
-                            .to_string(),
-                    )
+                    (SystemPromptContribution::Ready("The Host could not render the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory.".into()), None)
                 }
             },
             Err(error) => {
                 tracing::error!(%error, "subject resident context fetch failed");
-                SystemPromptContribution::Ready(
-                    "The Host could not fetch the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory."
-                        .to_string(),
-                )
+                (SystemPromptContribution::Ready("The Host could not fetch the Subject resident context. Do not treat this as unset behavior or infer replacement behavior from Memory.".into()), None)
             }
         }
     }
@@ -781,18 +877,31 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource
         if !changed {
             return Ok(None);
         }
-        let rendered = self.render(&output)?;
+        let (rendered, prompt_provenance) = self.render(&output)?;
         Ok(Some(ResidentContextRefresh {
             body: rendered,
             behavior_md: output.behavior_md,
+            prompt_provenance: Some(prompt_provenance),
         }))
     }
 
-    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 Some(input) = metadata
+            .input_metadata
+            .as_ref()
+            .and_then(|input| input.feature_payload())
+            .and_then(|payload| {
+                payload
+                    .decode::<SubjectBehaviorInput>("subject:behavior", 1)
+                    .ok()
+            })
+        {
+            self.confirm_resident_context_behavior(&input.behavior_md);
+        }
     }
 
     fn invalidate_resident_context_representation(&self) {
@@ -944,7 +1053,13 @@ impl Hook<PreLlmRequest> for SubjektivBehaviorRefreshHook {
                     "Subject behavior refresh requires durable session append authority",
                 )
             })?;
-            system_items.append_subject_behavior_refresh(refresh.body, refresh.behavior_md);
+            let input = subject_behavior_input(
+                refresh.body,
+                refresh.behavior_md,
+                refresh.prompt_provenance,
+            )
+            .map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?;
+            system_items.append_input(input);
         }
         Ok(HookPreRequestAction::Continue)
     }
@@ -969,16 +1084,18 @@ impl Hook<WorkerRestored> for SubjektivResidentRestoreRefreshHook {
             .resident_memory_restore_section(surface_ready, summary)
             .map_err(|error| HookError::new(HookErrorCategory::Internal, error.to_string()))?;
         let projection = prompts.projection();
-        context.system_items().append_resident_summary_refresh(
+        let input = resident_summary_input(
             body,
-            session_store::PromptRenderProvenance {
+            Some(session_store::PromptRenderProvenance {
                 workspace_id: self.workspace_id.clone(),
 
                 source_digest: projection.source_digest.clone(),
                 projection_digest: projection.catalog_digest.clone(),
                 logical_name: WorkerPrompt::ResidentMemoryRestoreSection.key().to_string(),
-            },
-        );
+            }),
+        )
+        .map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?;
+        context.system_items().append_input(input);
         Ok(())
     }
 }
@@ -1106,7 +1223,7 @@ pub(crate) fn ordinary_subjektiv_resident_summary_source(
     Ok(Some(Arc::new(WorkspaceSubjektivResidentSummarySource {
         host: host.expect("checked explicit connection"),
         prompts,
-        behavior_state: Mutex::new(SubjectBehaviorRefreshState::default()),
+        behavior_state: Arc::new(Mutex::new(SubjectBehaviorRefreshState::default())),
     })))
 }
 
@@ -1723,8 +1840,16 @@ permission = "write"
             .unwrap()
             .unwrap();
 
-        let initial = source.load().await;
+        let (initial, completion) = source.load_initial().await;
         assert!(matches!(initial, SystemPromptContribution::Ready(_)));
+        assert!(
+            source
+                .load_changed_resident_context()
+                .await
+                .unwrap()
+                .is_some()
+        );
+        completion.expect("successful initial contribution must provide completion")();
         assert!(
             source
                 .load_changed_resident_context()
@@ -1741,7 +1866,13 @@ 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);
+        let input = subject_behavior_input(
+            refresh.body.clone(),
+            refresh.behavior_md.clone(),
+            refresh.prompt_provenance.clone(),
+        )
+        .unwrap();
+        source.system_item_committed(&input.item, &input.metadata);
         assert!(
             source
                 .load_changed_resident_context()
@@ -1758,7 +1889,13 @@ 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);
+        let input = subject_behavior_input(
+            cleared.body.clone(),
+            cleared.behavior_md.clone(),
+            cleared.prompt_provenance.clone(),
+        )
+        .unwrap();
+        source.system_item_committed(&input.item, &input.metadata);
         assert!(
             source
                 .load_changed_resident_context()
@@ -1783,7 +1920,7 @@ permission = "write"
         // target-side bounded resident request contract.
         assert_eq!(
             client.timeouts.lock().unwrap().as_slice(),
-            &[RESIDENT_SUMMARY_REQUEST_TIMEOUT; 7]
+            &[RESIDENT_SUMMARY_REQUEST_TIMEOUT; 8]
         );
     }
 
@@ -1816,7 +1953,13 @@ permission = "write"
             .await
             .unwrap()
             .unwrap();
-        source.confirm_resident_context_behavior(&first.behavior_md);
+        let input = subject_behavior_input(
+            first.body.clone(),
+            first.behavior_md.clone(),
+            first.prompt_provenance.clone(),
+        )
+        .unwrap();
+        source.system_item_committed(&input.item, &input.metadata);
         assert_eq!(
             source
                 .load_changed_resident_context()
@@ -1825,7 +1968,13 @@ permission = "write"
                 .unwrap(),
             second
         );
-        source.confirm_resident_context_behavior(&second.behavior_md);
+        let input = subject_behavior_input(
+            second.body.clone(),
+            second.behavior_md.clone(),
+            second.prompt_provenance.clone(),
+        )
+        .unwrap();
+        source.system_item_committed(&input.item, &input.metadata);
         assert!(
             source
                 .load_changed_resident_context()

crates/worker/src/feature/builtin/session_explore.rs

@@ -371,7 +371,7 @@ fn parse_entry_ref(value: &str) -> Result<SessionEntryRef, ToolError> {
 fn parse_kind(value: &str) -> Result<ReferenceKind, ToolError> {
     ReferenceKind::parse(value).ok_or_else(|| {
         ToolError::InvalidArgument(format!(
-            "invalid kind {value:?}; expected user, assistant, or tool"
+            "invalid kind {value:?}; expected user, assistant, system, or tool"
         ))
     })
 }

crates/worker/src/feature/builtin/task/mod.rs

@@ -276,7 +276,9 @@ mod tests {
     use super::*;
     use crate::hook::{PreRequestInfo, SystemItemAppendHandle};
 
-    fn pre_request_context(pending: Arc<Mutex<Vec<SystemItem>>>) -> PreRequestContext {
+    fn pre_request_context(
+        pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>,
+    ) -> PreRequestContext {
         PreRequestContext::new(
             PreRequestInfo {
                 item_count: 1,
@@ -315,7 +317,7 @@ mod tests {
 
         let queued = pending.lock().expect("pending queue poisoned");
         assert_eq!(queued.len(), 1);
-        let SystemItem::TaskReminder { body, .. } = &queued[0] else {
+        let SystemItem::TaskReminder { body, .. } = &queued[0].item else {
             panic!("unexpected system item: {:?}", queued[0]);
         };
         assert!(body.starts_with("Current session steps are listed below."));
@@ -339,7 +341,7 @@ mod tests {
         }
 
         let queued = pending.lock().expect("pending queue poisoned");
-        let SystemItem::TaskReminder { source, body, .. } = &queued[0] else {
+        let SystemItem::TaskReminder { source, body, .. } = &queued[0].item else {
             panic!("unexpected system item: {:?}", queued[0]);
         };
         assert_eq!(*source, SystemReminderSource::TaskInactivity);
@@ -452,7 +454,7 @@ mod tests {
         }
         let queued = pending.lock().expect("pending queue poisoned");
         assert_eq!(queued.len(), 1);
-        let SystemItem::TaskReminder { body, .. } = &queued[0] else {
+        let SystemItem::TaskReminder { body, .. } = &queued[0].item else {
             panic!("unexpected system item: {:?}", queued[0]);
         };
         assert!(body.contains("taskid 1 (completed) done"));

crates/worker/src/feature/builtin/worker_observation.rs

@@ -813,6 +813,7 @@ mod tests {
                         _ => return None,
                     };
                     Some(protocol::SessionSnapshotEntry {
+                        use_policy: None,
                         entry_id: format!("fake-{index:08}"),
                         timestamp: index as u64,
                         provenance: protocol::SessionEntryProvenance::LegacyUnknown,

crates/worker/src/feature/invocation.rs

@@ -39,6 +39,45 @@ const MAX_INVOCATION_DESCRIPTOR_BYTES: usize =
     protocol::invocation::MAX_FEATURE_INVOCATION_DESCRIPTOR_BYTES;
 const MAX_COMPLETION_RESPONSE_BYTES: usize = 64 * 1024;
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct InvocationResultInput {
+    pub result: FeatureInvocationResult,
+}
+
+pub(crate) fn result_input(
+    result: FeatureInvocationResult,
+    body: String,
+) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
+    use session_store::*;
+    let payload = FeatureInputPayload::encode(
+        "feature:invocation-result",
+        1,
+        &InvocationResultInput { result },
+    )?;
+    let metadata = HistoryInputMetadata::new(
+        HistoryInputUsePolicy {
+            recording_material: InputEvidenceUse::Exclude,
+            memory_evidence: InputEvidenceUse::Exclude,
+            observation: InputObservationUse::Hidden,
+            compaction: InputCompactionUse::Retain,
+        },
+        Some(payload),
+        Some(PublicSystemItem {
+            kind: "feature_invocation_result".into(),
+            body: body.clone(),
+            presentation: PublicSystemPresentation::Context,
+        }),
+    )?;
+    Ok(LoggedSystemHistoryEntry::from_input(
+        LoggedSessionHistoryOrigin::Provided {
+            producer: "feature.invocation".into(),
+        },
+        None,
+        SystemInputEnvelope::new(body, metadata)?,
+    ))
+}
+
 pub(crate) fn invocation_result_is_bounded(result: &FeatureInvocationResult) -> bool {
     result.message.len() <= MAX_INVOCATION_MESSAGE_BYTES
         && result

crates/worker/src/hook.rs

@@ -25,7 +25,9 @@ use agen::tool::{ToolOutput, ToolResult};
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use session_store::{SystemItem, SystemReminder};
+#[cfg(test)]
+use session_store::SystemItem;
+use session_store::SystemReminder;
 use thiserror::Error;
 
 use crate::SessionEntryRef;
@@ -254,7 +256,7 @@ impl From<HookTurnEndAction> for TurnEndAction {
 // Hook context handles
 // =============================================================================
 
-/// Host-created handle for appending approved durable [`SystemItem`] requests.
+/// Host-created handle for appending approved durable [`session_store::SystemItem`] requests.
 ///
 /// 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
@@ -263,11 +265,11 @@ impl From<HookTurnEndAction> for TurnEndAction {
 /// raw `agen::Item`, history writer, event sender, `Worker`, `Engine`, or
 /// notification buffer.
 pub struct SystemItemAppendHandle {
-    pending: Arc<Mutex<Vec<SystemItem>>>,
+    pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>,
 }
 
 impl SystemItemAppendHandle {
-    pub(crate) fn new(pending: Arc<Mutex<Vec<SystemItem>>>) -> Self {
+    pub(crate) fn new(pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>) -> Self {
         Self { pending }
     }
 
@@ -279,53 +281,44 @@ impl SystemItemAppendHandle {
         self.pending
             .lock()
             .expect("system-item append queue poisoned")
-            .push(item);
+            .push(session_store::LoggedSystemHistoryEntry {
+                item,
+                metadata: crate::session_history::metadata(
+                    session_store::LoggedSessionHistoryOrigin::BackendInstruction {
+                        operation_id: None,
+                    },
+                    None,
+                ),
+            });
     }
 
-    /// Queue a Subject behavior refresh with its exact represented behavior. The caller receives this
-    /// narrow operation only at a pre-request boundary, and the host commits it
-    /// before exposing it to the model.
-    pub fn append_subject_behavior_refresh(
-        &self,
-        body: impl Into<String>,
-        behavior_md: impl Into<String>,
-    ) {
+    /// Queue producer-validated System context. Metadata and body are committed together.
+    /// The Host exposes this capability only at a pre-request boundary.
+    pub fn append_input(&self, input: session_store::LoggedSystemHistoryEntry) {
         self.pending
             .lock()
             .expect("system-item append queue poisoned")
-            .push(SystemItem::SubjectBehaviorRefresh {
-                body: body.into(),
-                behavior_md: behavior_md.into(),
-                prompt_provenance: None,
-            });
+            .push(input);
     }
 }
 
 /// Host-created narrow append authority exposed only at a restore lifecycle
-/// boundary. It cannot emit arbitrary system items.
+/// boundary. No model-facing tool receives this capability.
 pub struct RestoreSystemItemAppendHandle {
-    pending: Arc<Mutex<Vec<SystemItem>>>,
+    pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>,
 }
 
 impl RestoreSystemItemAppendHandle {
-    pub(crate) fn new(pending: Arc<Mutex<Vec<SystemItem>>>) -> Self {
+    pub(crate) fn new(pending: Arc<Mutex<Vec<session_store::LoggedSystemHistoryEntry>>>) -> Self {
         Self { pending }
     }
 
-    /// Queue a Feature-rendered resident-summary refresh for durable append at
-    /// the restore lifecycle boundary.
-    pub fn append_resident_summary_refresh(
-        &self,
-        body: impl Into<String>,
-        prompt_provenance: session_store::PromptRenderProvenance,
-    ) {
+    /// Queue producer-validated System context at the Host restore boundary.
+    pub fn append_input(&self, input: session_store::LoggedSystemHistoryEntry) {
         self.pending
             .lock()
             .expect("restore system-item append queue poisoned")
-            .push(SystemItem::ResidentSummaryRefresh {
-                body: body.into(),
-                prompt_provenance: Some(prompt_provenance),
-            });
+            .push(input);
     }
 }
 
@@ -983,7 +976,7 @@ mod tests {
 
         let queued = pending.lock().expect("pending queue poisoned");
         assert_eq!(queued.len(), 1);
-        match &queued[0] {
+        match &queued[0].item {
             SystemItem::TaskReminder { body, .. } => {
                 assert_eq!(body, "remember tasks");
             }
@@ -992,7 +985,7 @@ mod tests {
     }
 
     #[test]
-    fn restore_append_handle_queues_only_resident_refresh_items() {
+    fn restore_append_handle_preserves_producer_metadata() {
         let pending = Arc::new(Mutex::new(Vec::new()));
         let handle = RestoreSystemItemAppendHandle::new(Arc::clone(&pending));
         let provenance = session_store::PromptRenderProvenance {
@@ -1003,15 +996,16 @@ mod tests {
             logical_name: "internal.resident_memory_restore_section".into(),
         };
 
-        handle.append_resident_summary_refresh("current surface", provenance.clone());
-
-        assert!(matches!(
-            pending.lock().unwrap().as_slice(),
-            [SystemItem::ResidentSummaryRefresh {
-                body,
-                prompt_provenance: Some(actual),
-            }] if body == "current surface" && actual == &provenance
-        ));
+        let input = crate::feature::builtin::memory::resident_summary_input(
+            "current surface".into(),
+            Some(provenance),
+        )
+        .unwrap();
+        let expected = serde_json::to_value(&input).unwrap();
+        handle.append_input(input);
+        let queued = pending.lock().unwrap();
+        assert_eq!(queued.len(), 1);
+        assert_eq!(serde_json::to_value(&queued[0]).unwrap(), expected);
     }
 
     #[test]

crates/worker/src/internal_worker.rs

@@ -1499,6 +1499,7 @@ permission = "write"
                     is_error: false,
                 },
                 metadata: session_store::LoggedSessionHistoryMetadata {
+                    input_metadata: None,
                     entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()),
                     origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
                     derivation: None,

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(())
     }
@@ -267,6 +262,31 @@ impl WorkerInterceptor {
         )
     }
 
+    fn commit_system_entries(
+        &self,
+        entries: &[session_store::LoggedSystemHistoryEntry],
+    ) -> Result<(), session_store::StoreError> {
+        let Some(writer) = &self.log_writer else {
+            return Ok(());
+        };
+        for entry in entries {
+            writer.commit_annotated_system_item_then(
+                entry.clone(),
+                Vec::new(),
+                &mut |history| {
+                    self.pending_committed_history
+                        .lock()
+                        .expect("pending committed history poisoned")
+                        .push_back(history.clone());
+                    if let Some(source) = &self.resident_context_source {
+                        source.system_item_committed(&entry.item, &history.annotation);
+                    }
+                },
+            )?;
+        }
+        Ok(())
+    }
+
     fn current_turn_index(&self) -> usize {
         self.next_turn_index
             .load(Ordering::Relaxed)
@@ -392,11 +412,6 @@ impl WorkerInterceptor {
                 } if prompt_provenance.is_none() => {
                     *prompt_provenance = Some(provenance("internal.task_reminder"));
                 }
-                SystemItem::SubjectBehaviorRefresh {
-                    prompt_provenance, ..
-                } if prompt_provenance.is_none() => {
-                    *prompt_provenance = Some(provenance("internal.subjektiv_resident_context"));
-                }
                 SystemItem::Interrupt {
                     prompt_provenance, ..
                 } if prompt_provenance.is_none() => {
@@ -622,15 +637,17 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
             return Ok(PreRequestAction::Yield);
         }
 
-        let mut system_items: Vec<SystemItem> = std::mem::take(
+        let mut system_items: Vec<session_store::LoggedSystemHistoryEntry> = std::mem::take(
             &mut *pending_hook_system_items
                 .lock()
                 .expect("pending hook system-item queue poisoned"),
         );
-        self.attach_prompt_provenance(&mut system_items);
+        for entry in &mut system_items {
+            self.attach_prompt_provenance(std::slice::from_mut(&mut entry.item));
+        }
         let mut appended_items: Vec<Item> = system_items
             .iter()
-            .map(SystemItem::to_history_item)
+            .map(|entry| entry.item.to_history_item())
             .collect();
         let effective_context = if appended_items.is_empty() {
             Cow::Borrowed(context.as_slice())
@@ -644,7 +661,7 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
         let compact_decision =
             self.request_compact_decision(current_tokens, effective_context.as_ref());
         if !matches!(compact_decision, AutomaticCompactDecision::Continue) {
-            if let Err(error) = self.commit_system_items(&system_items) {
+            if let Err(error) = self.commit_system_entries(&system_items) {
                 return Ok(PreRequestAction::Cancel(format!(
                     "session persistence failed: {error}"
                 )));
@@ -688,7 +705,7 @@ impl Interceptor<SessionHistoryMetadata> for WorkerInterceptor {
         if appended_items.is_empty() {
             return Ok(PreRequestAction::Continue);
         }
-        Ok(match self.commit_system_items(&system_items) {
+        Ok(match self.commit_system_entries(&system_items) {
             Ok(()) => PreRequestAction::ContinueWith(appended_items),
             Err(error) => PreRequestAction::Cancel(format!("session persistence failed: {error}")),
         })
@@ -988,7 +1005,14 @@ mod tests {
             input
                 .system_items()
                 .expect("test interceptor has durable append authority")
-                .append_subject_behavior_refresh("current Subject context", "Ask when uncertain.");
+                .append_input(
+                    crate::feature::builtin::memory::subject_behavior_input(
+                        "current Subject context".into(),
+                        "Ask when uncertain.".into(),
+                        None,
+                    )
+                    .unwrap(),
+                );
             Ok(HookPreRequestAction::Continue)
         }
     }
@@ -1003,8 +1027,26 @@ 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 Some(input) = metadata
+                .input_metadata
+                .as_ref()
+                .and_then(|input| input.feature_payload())
+                .and_then(|payload| {
+                    payload
+                        .decode::<crate::feature::builtin::memory::SubjectBehaviorInput>(
+                            "subject:behavior",
+                            1,
+                        )
+                        .ok()
+                })
+            {
+                self.confirmed.lock().unwrap().push(input.behavior_md);
+            }
         }
     }
 
@@ -1548,14 +1590,87 @@ mod tests {
         assert_eq!(*confirmed.lock().unwrap(), ["Ask when uncertain."]);
         assert!(matches!(
             committed.lock().unwrap().as_slice(),
-            [SystemItem::SubjectBehaviorRefresh {
-                behavior_md,
-                prompt_provenance: Some(_),
-                ..
-            }] if behavior_md == "Ask when uncertain."
+            [SystemItem::Input { body }] if body == "current Subject context"
         ));
     }
 
+    #[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()));
@@ -1576,12 +1691,15 @@ mod tests {
             })),
         )
         .with_resident_context_source(Some(source));
-        let items = ["raw behavior", ""].map(|behavior_md| SystemItem::SubjectBehaviorRefresh {
-            body: "same rendered context".into(),
-            behavior_md: behavior_md.into(),
-            prompt_provenance: None,
+        let items = ["raw behavior", ""].map(|behavior_md| {
+            crate::feature::builtin::memory::subject_behavior_input(
+                "same rendered context".into(),
+                behavior_md.into(),
+                None,
+            )
+            .unwrap()
         });
-        interceptor.commit_system_items(&items).unwrap();
+        interceptor.commit_system_entries(&items).unwrap();
         assert_eq!(*confirmed.lock().unwrap(), ["raw behavior", ""]);
         assert_eq!(committed.lock().unwrap().len(), 2);
     }

crates/worker/src/ipc/protocol_session.rs

@@ -90,7 +90,8 @@ pub fn live_log_entry_event(entry: LogEntry) -> Option<Event> {
         }
         LogEntry::AnnotatedSystemItem { entry, .. } => {
             let entry_id = Some(entry.metadata.entry_id.0.clone());
-            let value = serde_json::to_value(&entry.item).expect("SystemItem is Serialize");
+            let value = serde_json::to_value(entry.public_item())
+                .expect("public system input is serializable");
             Some(Event::SystemItem {
                 entry_id,
                 item: value,
@@ -339,21 +340,13 @@ mod tests {
     #[test]
     fn system_item_log_entry_keeps_identity_in_live_event() {
         let entry_id = session_store::LoggedSessionHistoryEntryId::new();
+        let mut input =
+            crate::feature::builtin::memory::resident_summary_input("refresh".into(), None)
+                .unwrap();
+        input.metadata.entry_id = entry_id.clone();
         let event = live_log_entry_event(LogEntry::AnnotatedSystemItem {
             ts: session_store::segment_log::now_millis(),
-            entry: session_store::LoggedSystemHistoryEntry {
-                item: session_store::SystemItem::ResidentSummaryRefresh {
-                    body: "refresh".to_string(),
-                    prompt_provenance: None,
-                },
-                metadata: session_store::LoggedSessionHistoryMetadata {
-                    entry_id: entry_id.clone(),
-                    origin: session_store::LoggedSessionHistoryOrigin::BackendInstruction {
-                        operation_id: None,
-                    },
-                    derivation: None,
-                },
-            },
+            entry: input,
             extensions: Vec::new(),
         })
         .expect("SystemItem must be live-relevant");
@@ -370,6 +363,71 @@ mod tests {
         }
     }
 
+    #[test]
+    fn live_and_snapshot_system_projection_exclude_private_feature_fields() {
+        let provenance = session_store::PromptRenderProvenance {
+            workspace_id: Some("private-workspace".into()),
+            source_digest: "private-source".into(),
+            projection_digest: "private-projection".into(),
+            logical_name: "private-template".into(),
+        };
+        let items = [
+            crate::feature::builtin::memory::subject_behavior_input(
+                "displayed context".into(),
+                "private-raw-behavior".into(),
+                Some(provenance.clone()),
+            )
+            .unwrap(),
+            crate::feature::builtin::memory::resident_summary_input(
+                "displayed reference".into(),
+                Some(provenance),
+            )
+            .unwrap(),
+        ];
+        for input in items {
+            let entry = LogEntry::AnnotatedSystemItem {
+                ts: 7,
+                entry: input,
+                extensions: vec![],
+            };
+            let persisted: LogEntry =
+                serde_json::from_slice(&serde_json::to_vec(&entry).unwrap()).unwrap();
+            let snapshot = session_store::public_snapshot::project_current_session_snapshot(&[
+                persisted.clone(),
+            ]);
+            let Event::SystemItem {
+                entry_id,
+                item: live,
+            } = live_log_entry_event(persisted).unwrap()
+            else {
+                panic!("expected live system input");
+            };
+            let snapshot_entry = &snapshot.entries[0];
+            assert_eq!(entry_id.as_deref(), Some(snapshot_entry.entry_id.as_str()));
+            let protocol::SessionSnapshotEntryData::SystemItem {
+                data: Some(saved),
+                content,
+                ..
+            } = &snapshot_entry.data
+            else {
+                panic!("expected canonical public system input");
+            };
+            assert_eq!(&live, saved);
+            let public: session_store::PublicSystemItem =
+                serde_json::from_value(live.clone()).unwrap();
+            assert_eq!(&public.body, content);
+            assert!(matches!(
+                public.presentation,
+                session_store::PublicSystemPresentation::Context
+            ));
+            let encoded = serde_json::to_string(&snapshot).unwrap();
+            assert!(!encoded.contains("private-"));
+            assert!(live.get("prompt_provenance").is_none());
+            assert!(live.get("snapshot").is_none());
+            assert!(live.get("behavior_md").is_none());
+        }
+    }
+
     #[test]
     fn user_input_log_entry_maps_to_user_message_event() {
         let segments = vec![protocol::Segment::text("hello from log")];

crates/worker/src/lib.rs

@@ -16,6 +16,7 @@ pub mod hook;
 pub(crate) mod in_flight;
 pub mod ipc;
 pub mod job;
+mod metadata_lifecycle;
 pub mod model_client;
 mod paste_artifact_tool;
 pub mod prompt;
@@ -27,6 +28,7 @@ mod session_history;
 pub mod shared_state;
 mod shutdown_after_idle;
 mod subject_release;
+mod subjektiv_metadata;
 pub use subject_release::SubjectReleaseReservation;
 pub mod skill;
 pub mod spawn;

crates/worker/src/metadata_lifecycle.rs

@@ -0,0 +1,69 @@
+//! Atomic metadata transitions with Host-owned Session lifecycle participants.
+//!
+//! Participants inspect domain evidence inside the metadata update, not after the
+//! active pointer has been published. The generic writer never interprets it.
+
+use session_store::worker_metadata::{
+    WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore, WorkerStoreError,
+};
+use std::sync::Arc;
+
+pub(crate) trait SessionMetadataLifecycle: Send + Sync {
+    /// Runs under the metadata update lock. Only compare/transform the supplied
+    /// value: no I/O, reentrant store calls, lock waits, or Feature refresh.
+    fn active_changing(&self, metadata: &mut WorkerMetadata, next: Option<&WorkerActiveSegmentRef>);
+}
+
+pub(crate) type WorkerMetadataWriter =
+    Arc<dyn Fn(WorkerMetadata) -> Result<(), WorkerStoreError> + Send + Sync>;
+pub(crate) type WorkerMetadataSegmentCas = Arc<
+    dyn Fn(&str, &WorkerActiveSegmentRef, WorkerActiveSegmentRef) -> Result<bool, WorkerStoreError>
+        + Send
+        + Sync,
+>;
+
+pub(crate) fn worker_metadata_writer_for_store<St>(store: &St) -> WorkerMetadataWriter
+where
+    St: WorkerMetadataStore + Clone + Send + Sync + 'static,
+{
+    let store = store.clone();
+    let participants = crate::feature::builtin::session_metadata_lifecycles();
+    Arc::new(move |next| {
+        store
+            .update_by_name(&next.worker_name, |metadata| {
+                for participant in &participants {
+                    participant.active_changing(metadata, next.active.as_ref());
+                }
+                metadata.active = next.active;
+                metadata.resolved_manifest_snapshot = next.resolved_manifest_snapshot;
+                if next.workspace_id.is_some() {
+                    metadata.workspace_id = next.workspace_id;
+                }
+                if next.workspace_root.is_some() {
+                    metadata.workspace_root = next.workspace_root;
+                }
+            })
+            .map(|_| ())
+    })
+}
+
+pub(crate) fn worker_metadata_segment_cas_for_store<St>(store: &St) -> WorkerMetadataSegmentCas
+where
+    St: WorkerMetadataStore + Clone + Send + Sync + 'static,
+{
+    let store = store.clone();
+    let participants = crate::feature::builtin::session_metadata_lifecycles();
+    Arc::new(move |worker_name, expected, replacement| {
+        let mut matched = false;
+        store.update_by_name(worker_name, |metadata| {
+            if metadata.active.as_ref() == Some(expected) {
+                for participant in &participants {
+                    participant.active_changing(metadata, Some(&replacement));
+                }
+                metadata.active = Some(replacement);
+                matched = true;
+            }
+        })?;
+        Ok(matched)
+    })
+}

crates/worker/src/session_capture.rs

@@ -74,6 +74,7 @@ impl std::fmt::Display for SessionEntryRef {
 pub(crate) enum ReferenceKind {
     User,
     Assistant,
+    System,
     Tool,
 }
 
@@ -82,6 +83,7 @@ impl ReferenceKind {
         match self {
             Self::User => "user",
             Self::Assistant => "assistant",
+            Self::System => "system",
             Self::Tool => "tool",
         }
     }
@@ -90,6 +92,7 @@ impl ReferenceKind {
         match value {
             "user" => Some(Self::User),
             "assistant" | "agent" => Some(Self::Assistant),
+            "system" => Some(Self::System),
             "tool" => Some(Self::Tool),
             _ => None,
         }
@@ -131,6 +134,7 @@ pub(crate) struct OverviewItem {
 
 #[derive(Debug, Clone)]
 pub(crate) struct ReferenceEntry {
+    use_policy: Option<protocol::SessionEntryUsePolicy>,
     pub id: SessionEntryRef,
     pub origin: SessionEntryProvenance,
     pub entry_range: [u64; 2],
@@ -242,6 +246,7 @@ pub(crate) struct SessionEntryEvidence {
 
 #[derive(Debug, Clone)]
 struct CapturedHistoryEntry {
+    use_policy: Option<protocol::SessionEntryUsePolicy>,
     item: Item,
     entry_id: session_store::LoggedSessionHistoryEntryId,
     origin: SessionEntryProvenance,
@@ -302,10 +307,12 @@ impl SessionCapture {
                         is_error,
                         attachments: _,
                     } => Item::tool_result_item(call_id, summary, content, is_error),
-                    // Observation deliberately excludes system items and
-                    // controller errors from model-visible session evidence.
-                    SessionSnapshotEntryData::SystemItem { .. }
-                    | SessionSnapshotEntryData::Invoke { .. }
+                    SessionSnapshotEntryData::SystemItem { content, .. } => {
+                        entry.use_policy?;
+                        Item::system_message(content)
+                    }
+                    // Controller errors are not model-visible evidence.
+                    SessionSnapshotEntryData::Invoke { .. }
                     | SessionSnapshotEntryData::Usage { .. }
                     | SessionSnapshotEntryData::RunStats { .. }
                     | SessionSnapshotEntryData::RunYielded { .. }
@@ -314,6 +321,7 @@ impl SessionCapture {
                     | SessionSnapshotEntryData::RunError { .. } => return None,
                 };
                 Some(CapturedHistoryEntry {
+                    use_policy: entry.use_policy,
                     item,
                     entry_id: session_store::LoggedSessionHistoryEntryId(entry.entry_id),
                     origin: entry.provenance,
@@ -328,6 +336,7 @@ impl SessionCapture {
             .into_iter()
             .enumerate()
             .map(|(index, item)| CapturedHistoryEntry {
+                use_policy: None,
                 item,
                 entry_id: session_store::LoggedSessionHistoryEntryId(format!("{index:08}")),
                 origin: SessionEntryProvenance::LegacyUnknown,
@@ -342,10 +351,32 @@ impl SessionCapture {
     ) -> Self {
         let entries = entries
             .into_iter()
-            .map(|entry| CapturedHistoryEntry {
-                item: entry.item,
-                entry_id: entry.annotation.entry_id,
-                origin: public_provenance(&entry.annotation.origin),
+            .map(|entry| {
+                let (item, use_policy) = if let Some(input) = &entry.annotation.input_metadata {
+                    let is_system = matches!(
+                        entry.item,
+                        Item::Message {
+                            role: Role::System,
+                            ..
+                        }
+                    );
+                    if is_system {
+                        (
+                            Item::system_message(input.public_projection().body),
+                            Some(input.public_system_use_policy()),
+                        )
+                    } else {
+                        (entry.item, Some(input.public_use_policy()))
+                    }
+                } else {
+                    (entry.item, None)
+                };
+                CapturedHistoryEntry {
+                    item,
+                    use_policy,
+                    entry_id: entry.annotation.entry_id,
+                    origin: public_provenance(&entry.annotation.origin),
+                }
             })
             .collect();
         Self::from_captured_entries(segment_id, entries)
@@ -364,7 +395,8 @@ impl SessionCapture {
             let entry_range = [idx as u64, idx as u64];
             match item {
                 Item::Message { role, content, .. } => {
-                    let Some(kind) = message_reference_kind(&entry.origin, role) else {
+                    let Some(kind) = message_reference_kind(&entry.origin, role, entry.use_policy)
+                    else {
                         continue;
                     };
                     let text = content
@@ -376,6 +408,7 @@ impl SessionCapture {
                     let summary = truncate_chars(&text, 240);
                     let id = SessionEntryRef::from_history_entry_id(&entry.entry_id);
                     index.push(ReferenceEntry {
+                        use_policy: entry.use_policy,
                         id: id.clone(),
                         origin: entry.origin.clone(),
                         entry_range,
@@ -386,7 +419,9 @@ impl SessionCapture {
                         summary: summary.clone(),
                         search_text: text.clone(),
                     });
-                    if matches!(kind, ReferenceKind::User | ReferenceKind::Assistant) {
+                    if entry.use_policy.is_none_or(|p| p.observation_readable)
+                        && matches!(kind, ReferenceKind::User | ReferenceKind::Assistant)
+                    {
                         overview.push(OverviewItem {
                             id: id.clone(),
                             origin: entry.origin.clone(),
@@ -403,6 +438,7 @@ impl SessionCapture {
                 } => {
                     let text = format!("{name}\n{arguments}");
                     index.push(ReferenceEntry {
+                        use_policy: entry.use_policy,
                         id: SessionEntryRef::from_history_entry_id(&entry.entry_id),
                         origin: entry.origin.clone(),
                         entry_range,
@@ -430,6 +466,7 @@ impl SessionCapture {
                         content.as_deref().unwrap_or_default(),
                     );
                     index.push(ReferenceEntry {
+                        use_policy: entry.use_policy,
                         id: SessionEntryRef::from_history_entry_id(&entry.entry_id),
                         origin: entry.origin.clone(),
                         entry_range,
@@ -510,6 +547,9 @@ impl SessionCapture {
         let mut hits = Vec::new();
 
         for entry in &self.index {
+            if entry.use_policy.is_some_and(|p| !p.observation_readable) {
+                continue;
+            }
             if entry.entry_range[0] < min_entry_index || entry.entry_range[0] > max_entry_index {
                 continue;
             }
@@ -581,6 +621,9 @@ impl SessionCapture {
         };
 
         for entry in selected {
+            if entry.use_policy.is_some_and(|p| !p.observation_readable) {
+                continue;
+            }
             if entries.len() >= max_items {
                 truncated = true;
                 break;
@@ -695,21 +738,11 @@ impl SessionCapture {
 
     pub(crate) fn evidence_for(&self, id: &str) -> Option<SessionEntryEvidence> {
         let entry = self.index.iter().find(|entry| entry.id.as_str() == id)?;
-        let excerpt = self
-            .read(
-                ReadSelector::Id(id),
-                ReadOptions {
-                    include_tools: true,
-                    tool_part: ToolPart::Both,
-                    detail: ReadDetail::Compact,
-                    max_items: 1,
-                    max_bytes: 2 * 1024,
-                },
-            )
-            .entries
-            .first()
-            .map(|entry| entry.text.clone())
-            .unwrap_or_else(|| entry.summary.clone());
+        if entry.use_policy.is_some_and(|p| !p.memory_evidence) {
+            return None;
+        }
+        let item = &self.entries.get(entry.entry_range[0] as usize)?.item;
+        let excerpt = render_item(item, entry, ReadDetail::Compact, 2 * 1024);
         Some(SessionEntryEvidence {
             segment_id: self.segment_id.clone(),
             entry_ref: entry.id.clone(),
@@ -731,9 +764,8 @@ fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance
         }
         WorkerHistoryProvenance::WorkerInput { .. } => SessionEntryProvenance::WorkerInput,
         WorkerHistoryProvenance::FlowInstruction { .. } => SessionEntryProvenance::FlowInstruction,
-        WorkerHistoryProvenance::BackendInstruction { .. } => {
-            SessionEntryProvenance::BackendInstruction
-        }
+        WorkerHistoryProvenance::BackendInstruction { .. }
+        | WorkerHistoryProvenance::Provided { .. } => SessionEntryProvenance::BackendInstruction,
         WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
         WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
         WorkerHistoryProvenance::CompactionSummary
@@ -746,7 +778,15 @@ fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance
 fn message_reference_kind(
     origin: &SessionEntryProvenance,
     provider_role: &Role,
+    use_policy: Option<protocol::SessionEntryUsePolicy>,
 ) -> Option<ReferenceKind> {
+    if use_policy.is_some() {
+        return match provider_role {
+            Role::System => Some(ReferenceKind::System),
+            Role::User => Some(ReferenceKind::User),
+            Role::Assistant => Some(ReferenceKind::Assistant),
+        };
+    }
     match origin {
         SessionEntryProvenance::HumanInput | SessionEntryProvenance::WorkerInput => {
             Some(ReferenceKind::User)
@@ -950,6 +990,328 @@ mod tests {
         }
     }
 
+    #[test]
+    fn generic_public_input_observation_and_memory_policies_survive_store_restore_independently() {
+        use session_store::{
+            FeatureInputPayload, FsStore, HistoryInputMetadata, HistoryInputUsePolicy,
+            InputCompactionUse, InputEvidenceUse, InputObservationUse, LogEntry,
+            LoggedSystemHistoryEntry, PublicSystemItem, PublicSystemPresentation, Store,
+            SystemInputEnvelope,
+        };
+        let root = tempfile::tempdir().unwrap();
+        let store = FsStore::new(root.path()).unwrap();
+        let session = session_store::new_session_id();
+        let segment = session_store::new_segment_id();
+        store
+            .create_segment(
+                session,
+                segment,
+                &[LogEntry::AnnotatedSegmentStart {
+                    ts: 1,
+                    session_id: session,
+                    system_prompt: None,
+                    config: Default::default(),
+                    history: vec![],
+                    forked_from: None,
+                    compacted_from: None,
+                }],
+            )
+            .unwrap();
+        let mut inputs = Vec::new();
+        for (producer, observation, memory_evidence, public) in [
+            (
+                "same",
+                InputObservationUse::Readable,
+                InputEvidenceUse::Exclude,
+                true,
+            ),
+            (
+                "same",
+                InputObservationUse::Hidden,
+                InputEvidenceUse::Include,
+                true,
+            ),
+            (
+                "renamed",
+                InputObservationUse::Readable,
+                InputEvidenceUse::Include,
+                true,
+            ),
+            (
+                "renamed",
+                InputObservationUse::Hidden,
+                InputEvidenceUse::Exclude,
+                true,
+            ),
+            (
+                "private",
+                InputObservationUse::Readable,
+                InputEvidenceUse::Include,
+                false,
+            ),
+        ] {
+            let entry = LoggedSystemHistoryEntry::from_input(
+                WorkerHistoryProvenance::Provided {
+                    producer: producer.into(),
+                },
+                None,
+                SystemInputEnvelope::new(
+                    "private model text",
+                    HistoryInputMetadata::new(
+                        HistoryInputUsePolicy {
+                            recording_material: InputEvidenceUse::Exclude,
+                            memory_evidence,
+                            observation,
+                            compaction: InputCompactionUse::ReferenceInput,
+                        },
+                        Some(
+                            FeatureInputPayload::encode(
+                                "future:owner",
+                                99,
+                                &serde_json::json!({ "secret": "private metadata" }),
+                            )
+                            .unwrap(),
+                        ),
+                        public.then(|| PublicSystemItem {
+                            kind: "supplied-context".into(),
+                            body: "public context: do not extract this".into(),
+                            presentation: PublicSystemPresentation::Context,
+                        }),
+                    )
+                    .unwrap(),
+                )
+                .unwrap(),
+            );
+            store
+                .append(
+                    session,
+                    segment,
+                    &LogEntry::AnnotatedSystemItem {
+                        ts: 2,
+                        entry: entry.clone(),
+                        extensions: vec![],
+                    },
+                )
+                .unwrap();
+            inputs.push(entry);
+        }
+        let saved = store.read_all(session, segment).unwrap();
+        let restored =
+            crate::session_history::restore_history_entries(session, segment, &saved).unwrap();
+        let snapshot = session_store::public_snapshot::project_session_snapshot(session, &saved);
+        let snapshot: SessionSnapshot =
+            serde_json::from_value(serde_json::to_value(snapshot).unwrap()).unwrap();
+        for capture in [
+            SessionCapture::from_history_entries(segment.to_string(), restored),
+            SessionCapture::from_session_snapshot(segment.to_string(), snapshot),
+        ] {
+            let hits = capture.search(&SearchOptions {
+                kind: Some(ReferenceKind::System),
+                ..Default::default()
+            });
+            assert_eq!(hits.len(), 2);
+            for (index, input) in inputs.iter().enumerate() {
+                let reference = SessionEntryRef::from_history_entry_id(&input.metadata.entry_id);
+                let read =
+                    capture.read(ReadSelector::Id(reference.as_str()), ReadOptions::default());
+                assert_eq!(!read.entries.is_empty(), matches!(index, 0 | 2));
+                let evidence = capture.evidence_for(reference.as_str());
+                assert_eq!(evidence.is_some(), matches!(index, 1 | 2));
+                for text in read
+                    .entries
+                    .iter()
+                    .map(|entry| entry.text.as_str())
+                    .chain(evidence.iter().map(|entry| entry.excerpt.as_str()))
+                {
+                    assert!(text.contains("public context: do not extract this"));
+                    assert!(!text.contains("private model text"));
+                    assert!(!text.contains("private metadata"));
+                    assert!(!text.contains("future:owner"));
+                }
+            }
+        }
+    }
+
+    // Real FsStore seed/fork/reopen -> ordinary history replay, public snapshot,
+    // capture and both evidence consumers; no Feature or Tool execution occurs.
+    #[test]
+    fn tool_policy_axes_survive_restore_and_fork_without_exposing_derived_context() {
+        use session_store::public_material::{
+            PublicMaterialLimits, PublicMaterialUse, SessionPublicMaterialReader,
+        };
+        use session_store::*;
+        for recording in [InputEvidenceUse::Include, InputEvidenceUse::Exclude] {
+            for memory in [InputEvidenceUse::Include, InputEvidenceUse::Exclude] {
+                for observation in [InputObservationUse::Readable, InputObservationUse::Hidden] {
+                    let root = tempfile::tempdir().unwrap();
+                    let store = FsStore::new(root.path()).unwrap();
+                    let session = new_session_id();
+                    let first = new_segment_id();
+                    let fork = new_segment_id();
+                    let items = vec![
+                        Item::tool_call("original", "ArbitraryTool", "{}"),
+                        Item::tool_result("original", "original result"),
+                        Item::tool_call("derived", "ArbitraryTool", "PRIVATE_SYNTHETIC_CONTEXT"),
+                        Item::tool_result("derived", "PRIVATE_SYNTHETIC_RESULT"),
+                    ];
+                    let mut history: Vec<_> = items
+                        .iter()
+                        .cloned()
+                        .map(|item| LoggedHistoryEntry {
+                            item: LoggedItem::from(item),
+                            metadata: LoggedSessionHistoryMetadata::legacy_unknown(),
+                        })
+                        .collect();
+                    let sources = history[..2]
+                        .iter()
+                        .map(|entry| entry.metadata.entry_id.clone())
+                        .collect::<Vec<_>>();
+                    for entry in &mut history[..2] {
+                        entry.metadata.input_metadata = Some(
+                            HistoryInputMetadata::new(
+                                HistoryInputUsePolicy {
+                                    recording_material: recording,
+                                    memory_evidence: memory,
+                                    observation,
+                                    compaction: InputCompactionUse::Retain,
+                                },
+                                None,
+                                None,
+                            )
+                            .unwrap(),
+                        );
+                    }
+                    for entry in &mut history[2..] {
+                        entry.metadata.derivation = Some(LoggedHistoryDerivation {
+                            sources: sources.clone(),
+                        });
+                        entry.metadata.input_metadata = Some(
+                            HistoryInputMetadata::new(
+                                HistoryInputUsePolicy {
+                                    recording_material: InputEvidenceUse::Exclude,
+                                    memory_evidence: InputEvidenceUse::Exclude,
+                                    observation: InputObservationUse::Hidden,
+                                    compaction: InputCompactionUse::Retain,
+                                },
+                                None,
+                                None,
+                            )
+                            .unwrap(),
+                        );
+                    }
+                    for segment in [first, fork] {
+                        store
+                            .create_segment(
+                                session,
+                                segment,
+                                &[LogEntry::AnnotatedSegmentStart {
+                                    ts: 1,
+                                    session_id: session,
+                                    system_prompt: None,
+                                    config: Default::default(),
+                                    history: history.clone(),
+                                    forked_from: (segment == fork).then_some(SegmentOrigin {
+                                        segment_id: first,
+                                        at_turn_index: 0,
+                                    }),
+                                    compacted_from: None,
+                                }],
+                            )
+                            .unwrap();
+                    }
+                    let reopened = FsStore::new(root.path()).unwrap();
+                    let saved = reopened.read_all(session, fork).unwrap();
+                    let restored =
+                        crate::session_history::restore_history_entries(session, fork, &saved)
+                            .unwrap();
+                    for (index, entry) in restored.iter().enumerate() {
+                        assert_eq!(entry.item, items[index]);
+                        assert_eq!(entry.annotation, history[index].metadata);
+                    }
+                    let capture = SessionCapture::from_history_entries(fork.to_string(), restored);
+                    assert!(
+                        matches!(&capture.entries[2].item, Item::ToolCall { call_id, .. } if call_id == "derived")
+                    );
+                    assert!(
+                        matches!(&capture.entries[3].item, Item::ToolResult { call_id, .. } if call_id == "derived")
+                    );
+                    let snapshot = public_snapshot::project_session_snapshot(session, &saved);
+                    assert!(
+                        !serde_json::to_string(&snapshot)
+                            .unwrap()
+                            .contains("PRIVATE_SYNTHETIC")
+                    );
+                    for capture in [
+                        capture,
+                        SessionCapture::from_session_snapshot(fork.to_string(), snapshot),
+                    ] {
+                        assert_eq!(
+                            capture
+                                .search(&SearchOptions {
+                                    kind: Some(ReferenceKind::Tool),
+                                    ..Default::default()
+                                })
+                                .len(),
+                            if observation == InputObservationUse::Readable {
+                                2
+                            } else {
+                                0
+                            }
+                        );
+                        for (index, entry) in history.iter().enumerate() {
+                            let reference =
+                                SessionEntryRef::from_history_entry_id(&entry.metadata.entry_id);
+                            assert_eq!(
+                                !capture
+                                    .read(
+                                        ReadSelector::Id(reference.as_str()),
+                                        ReadOptions::default()
+                                    )
+                                    .entries
+                                    .is_empty(),
+                                index < 2 && observation == InputObservationUse::Readable
+                            );
+                            assert_eq!(
+                                capture.evidence_for(reference.as_str()).is_some(),
+                                index < 2 && memory == InputEvidenceUse::Include
+                            );
+                        }
+                    }
+                    for purpose in [
+                        PublicMaterialUse::Recording,
+                        PublicMaterialUse::MemoryEvidence,
+                    ] {
+                        let reader = SessionPublicMaterialReader::open_fs_for_use(
+                            root.path(),
+                            session,
+                            fork,
+                            PublicMaterialLimits::default(),
+                            reopened.retention(),
+                            purpose,
+                        )
+                        .unwrap();
+                        let ids = reader
+                            .descriptors()
+                            .map(|entry| entry.identity.entry_ref.clone())
+                            .collect::<Vec<_>>();
+                        let included = match purpose {
+                            PublicMaterialUse::Recording => recording,
+                            PublicMaterialUse::MemoryEvidence => memory,
+                        } == InputEvidenceUse::Include;
+                        assert_eq!(
+                            ids,
+                            sources
+                                .iter()
+                                .filter(|_| included)
+                                .map(|id| format!("E{}", id.0))
+                                .collect::<Vec<_>>()
+                        );
+                    }
+                }
+            }
+        }
+    }
+
     #[test]
     fn overview_contains_user_and_assistant_only() {
         let view = SessionCapture::new(

crates/worker/src/session_history.rs

@@ -24,6 +24,7 @@ pub(crate) fn test_logged_history_entry(item: impl Into<Item>) -> LoggedHistoryE
     LoggedHistoryEntry {
         item: session_store::LoggedItem::from(item.into()),
         metadata: LoggedSessionHistoryMetadata {
+            input_metadata: None,
             entry_id: LoggedSessionHistoryEntryId::new(),
             origin: LoggedSessionHistoryOrigin::LegacyUnknown,
             derivation: None,
@@ -38,6 +39,7 @@ pub(crate) fn test_logged_system_entry(
     session_store::LoggedSystemHistoryEntry {
         item,
         metadata: LoggedSessionHistoryMetadata {
+            input_metadata: None,
             entry_id: LoggedSessionHistoryEntryId::new(),
             origin: LoggedSessionHistoryOrigin::LegacyUnknown,
             derivation: None,
@@ -58,12 +60,23 @@ pub(crate) fn metadata(
     derivation: Option<SessionHistoryDerivation>,
 ) -> SessionHistoryMetadata {
     SessionHistoryMetadata {
+        input_metadata: None,
         entry_id: SessionHistoryEntryId::new(),
         origin,
         derivation,
     }
 }
 
+pub(crate) fn compaction_use(
+    metadata: &SessionHistoryMetadata,
+) -> session_store::InputCompactionUse {
+    metadata
+        .input_metadata
+        .as_ref()
+        .map(|input| input.usage().compaction)
+        .unwrap_or(session_store::InputCompactionUse::Retain)
+}
+
 #[cfg(test)]
 pub(crate) fn history_entry(
     item: Item,
@@ -80,6 +93,7 @@ pub(crate) fn history_entry_with_id(
     HistoryEntry::new(
         item,
         SessionHistoryMetadata {
+            input_metadata: None,
             entry_id,
             origin,
             derivation: None,
@@ -130,22 +144,10 @@ pub(crate) fn restore_history_entries(
                 }
                 *target = from_logged(entry);
             }
-            LogEntry::AnnotatedSystemItem { entry, .. } => {
-                history.push(HistoryEntry::new(entry.item.to_history_item(), {
-                    let mut metadata = entry.metadata.clone();
-                    // Old typed startup records used generic placeholder origins.
-                    // Preserve explicitly supplied metadata on current contributions.
-                    if matches!(
-                        metadata.origin,
-                        WorkerHistoryProvenance::DerivedSummary
-                            | WorkerHistoryProvenance::LegacyUnknown
-                    ) && matches!(entry.item, session_store::SystemItem::JournalStartup { .. })
-                    {
-                        metadata.origin = WorkerHistoryProvenance::JournalContext;
-                    }
-                    metadata
-                }))
-            }
+            LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new(
+                entry.item.to_history_item(),
+                entry.metadata.clone(),
+            )),
             _ => {}
         }
     }

crates/worker/src/skill.rs

@@ -8,6 +8,50 @@ pub use server_api::{
     SkillProjectionStatus, SkillProvenance, SkillResourceRef, SkillSourceKind,
 };
 
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(crate) struct SkillActivationInput {
+    pub name: String,
+}
+
+pub(crate) fn activation_input(
+    activation: &SkillActivationResponse,
+) -> Result<session_store::LoggedSystemHistoryEntry, session_store::InputMetadataError> {
+    use session_store::*;
+    let body = format!(
+        "Agent Skill `{}` activated from {}.\n\n{}",
+        activation.name, activation.provenance.id, activation.body
+    );
+    let payload = FeatureInputPayload::encode(
+        "skill:activation",
+        1,
+        &SkillActivationInput {
+            name: activation.name.clone(),
+        },
+    )?;
+    let metadata = HistoryInputMetadata::new(
+        HistoryInputUsePolicy {
+            recording_material: InputEvidenceUse::Exclude,
+            memory_evidence: InputEvidenceUse::Exclude,
+            observation: InputObservationUse::Hidden,
+            compaction: InputCompactionUse::Retain,
+        },
+        Some(payload),
+        Some(PublicSystemItem {
+            kind: "skill_activation".into(),
+            body: body.clone(),
+            presentation: PublicSystemPresentation::Context,
+        }),
+    )?;
+    Ok(LoggedSystemHistoryEntry::from_input(
+        LoggedSessionHistoryOrigin::Provided {
+            producer: "skill.activation".into(),
+        },
+        None,
+        SystemInputEnvelope::new(body, metadata)?,
+    ))
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum SkillClientError {
     #[error("workspace client is unavailable: {0}")]

crates/worker/src/spawn/comm_tools.rs

@@ -77,6 +77,7 @@ mod tests {
                     .into_iter()
                     .enumerate()
                     .map(|(index, value)| protocol::SessionSnapshotEntry {
+                        use_policy: None,
                         entry_id: format!("test-{index}"),
                         timestamp: index as u64,
                         provenance: protocol::SessionEntryProvenance::LegacyUnknown,

crates/worker/src/spawn/registry.rs

@@ -1358,6 +1358,7 @@ mod tests {
                         is_error: false,
                     },
                     metadata: session_store::LoggedSessionHistoryMetadata {
+                        input_metadata: None,
                         entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()),
                         origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
                         derivation: None,

crates/worker/src/subjektiv.rs

@@ -26,6 +26,30 @@ use server_api::*;
 
 use crate::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceServerOperation};
 
+impl<C: agen::llm_client::client::LlmClient + 'static, St: session_store::Store>
+    crate::Worker<C, St>
+{
+    /// Bind live Host capability and its context provider, not authority inherited
+    /// from a previous binding. Ordinary Feature/lifecycle gates also apply to embedders.
+    pub fn bind_subjektiv_host(&mut self, host: Option<Arc<dyn SubjektivHost>>) {
+        self.set_subjektiv_host(host);
+        if self.manifest_lifecycle_features_enabled()
+            && self.backend_job().is_none()
+            && crate::feature::builtin::memory::ordinary_subjektiv_features_enabled(self.manifest())
+            && let Ok(Some(connection)) = self.subjektiv_host_connection()
+            && (connection.host.journal_startup_granted()
+                || connection.host.journal_compaction_granted())
+        {
+            self.install_session_context_source(Arc::new(
+                crate::feature::builtin::journal::JournalContextSource::new(
+                    connection,
+                    self.prompts(),
+                ),
+            ));
+        }
+    }
+}
+
 /// Host-issued execution context, never deserialized from model arguments.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct SubjektivHostContext {
@@ -1128,7 +1152,12 @@ enabled = true
             let initial = store.read_all(session, segment).unwrap();
             let journal = initial.iter().find_map(|entry| match entry {
                 LogEntry::AnnotatedSystemItem { entry, .. }
-                    if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
+                    if entry
+                        .metadata
+                        .input_metadata
+                        .as_ref()
+                        .and_then(|input| input.feature_payload())
+                        .is_some_and(|payload| payload.schema() == "journal:startup") =>
                 {
                     Some(entry)
                 }
@@ -1138,22 +1167,33 @@ enabled = true
             let body = journal.map(|journal| {
                 assert_eq!(
                     journal.metadata.origin,
-                    LoggedSessionHistoryOrigin::JournalContext
+                    LoggedSessionHistoryOrigin::Provided {
+                        producer: "subjektiv.journal".into()
+                    }
                 );
-                let SystemItem::JournalStartup {
-                    body,
-                    snapshot: saved,
-                    snapshot_sha256,
-                    prompt_provenance,
-                } = &journal.item
-                else {
+                let SystemItem::Input { body } = &journal.item else {
                     unreachable!()
                 };
-                assert_eq!(*saved, serde_json::to_value(&snapshot).unwrap());
+                let payload = journal
+                    .metadata
+                    .input_metadata
+                    .as_ref()
+                    .unwrap()
+                    .feature_payload()
+                    .unwrap()
+                    .decode::<crate::feature::builtin::journal::JournalStartupInput>(
+                        "journal:startup",
+                        1,
+                    )
+                    .unwrap();
+                let saved = serde_json::to_value(&payload.snapshot).unwrap();
+                let snapshot_sha256 = &payload.snapshot_sha256;
+                let prompt_provenance = &payload.prompt_provenance;
+                assert_eq!(saved, serde_json::to_value(&snapshot).unwrap());
                 use sha2::{Digest, Sha256};
                 assert_eq!(
                     *snapshot_sha256,
-                    Sha256::digest(serde_json::to_vec(saved).unwrap())
+                    Sha256::digest(serde_json::to_vec(&saved).unwrap())
                         .iter()
                         .map(|byte| format!("{byte:02x}"))
                         .collect::<String>()
@@ -1166,7 +1206,7 @@ enabled = true
                     body.trim_end(),
                     format!(
                         "## Prior Journal context\n\n```json\n{}\n```",
-                        serde_json::to_string(saved).unwrap()
+                        serde_json::to_string(&saved).unwrap()
                     )
                 );
                 body.as_str()
@@ -1235,7 +1275,7 @@ enabled = true
             }
             assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1);
             let logs = store.read_all(session, segment).unwrap();
-            assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).count(), usize::from(contributes_context));
+            assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if entry.metadata.input_metadata.as_ref().and_then(|input| input.feature_payload()).is_some_and(|payload| payload.schema() == "journal:startup"))).count(), usize::from(contributes_context));
             controller
                 .handle
                 .send(Method::Shutdown {

crates/worker/src/subjektiv_metadata.rs

@@ -0,0 +1,120 @@
+//! Host-owned attribution transitions. Metadata callbacks are synchronous, local
+//! comparisons only: no network I/O, Feature lookup, or nested store access.
+
+use crate::metadata_lifecycle::SessionMetadataLifecycle;
+use session_store::{
+    SessionId, SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata,
+    WorkerMetadataStore,
+};
+
+pub(crate) struct SessionAttributionMetadata;
+
+impl SessionMetadataLifecycle for SessionAttributionMetadata {
+    fn active_changing(
+        &self,
+        metadata: &mut WorkerMetadata,
+        next: Option<&WorkerActiveSegmentRef>,
+    ) {
+        if metadata.active.as_ref().map(|active| active.session_id)
+            != next.map(|active| active.session_id)
+        {
+            metadata.subjektiv_session_attribution = None;
+        }
+    }
+}
+
+pub(crate) fn compare_and_swap_attribution<St: WorkerMetadataStore>(
+    store: &St,
+    worker_name: &str,
+    session_id: SessionId,
+    expected: Option<&SubjektivSessionAttributionState>,
+    replacement: SubjektivSessionAttributionState,
+) -> Result<(), crate::WorkerError> {
+    let mut matched = false;
+    store.update_by_name(worker_name, |metadata| {
+        if metadata.active.as_ref().map(|active| active.session_id) == Some(session_id)
+            && metadata.subjektiv_session_attribution.as_ref() == expected
+            && replacement.session_id() == session_id
+        {
+            metadata.subjektiv_session_attribution = Some(replacement);
+            matched = true;
+        }
+    })?;
+    if !matched {
+        return Err(crate::WorkerError::SubjektivSessionAttribution {
+            state: "attribution_conflict",
+            message: "active Session or observed attribution changed during Host confirmation"
+                .into(),
+        });
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    //! Connect the production metadata writer/CAS and Host lifecycle to a real
+    //! filesystem store; no metadata or persistence boundary is substituted.
+    use super::*;
+    use crate::metadata_lifecycle::{
+        worker_metadata_segment_cas_for_store, worker_metadata_writer_for_store,
+    };
+
+    #[test]
+    fn host_transition_preserves_same_session_and_invalidates_new_session_atomically() {
+        let dir = tempfile::tempdir().unwrap();
+        let store = session_store::FsWorkerStore::new(dir.path()).unwrap();
+        let session = session_store::new_session_id();
+        let mut metadata = WorkerMetadata::new(
+            "body",
+            Some(WorkerActiveSegmentRef::pending_segment(session)),
+        );
+        metadata.workspace_id = Some("workspace".into());
+        metadata.subjektiv_session_attribution =
+            Some(SubjektivSessionAttributionState::Confirmed {
+                session_id: session,
+                subject_id: "subject".into(),
+            });
+        store.write(&metadata).unwrap();
+        let writer = worker_metadata_writer_for_store(&store);
+        let cas = worker_metadata_segment_cas_for_store(&store);
+        let segment =
+            WorkerActiveSegmentRef::active_segment(session, session_store::new_segment_id());
+        let mut next = WorkerMetadata::new("body", Some(segment.clone()));
+        writer(next.clone()).unwrap();
+        let materialized = store.read_by_name("body").unwrap().unwrap();
+        assert_eq!(
+            materialized.subjektiv_session_attribution,
+            metadata.subjektiv_session_attribution
+        );
+        assert_eq!(materialized.workspace_id, metadata.workspace_id);
+        let compacted =
+            WorkerActiveSegmentRef::active_segment(session, session_store::new_segment_id());
+        assert!(cas("body", &segment, compacted.clone()).unwrap());
+        assert_eq!(
+            store
+                .read_by_name("body")
+                .unwrap()
+                .unwrap()
+                .subjektiv_session_attribution,
+            metadata.subjektiv_session_attribution
+        );
+        let new_session = WorkerActiveSegmentRef::pending_segment(session_store::new_session_id());
+        assert!(!cas("body", &segment, new_session.clone()).unwrap());
+        assert!(cas("body", &compacted, new_session.clone()).unwrap());
+        let switched = store.read_by_name("body").unwrap().unwrap();
+        assert_eq!(switched.active, Some(new_session));
+        assert!(switched.subjektiv_session_attribution.is_none());
+        assert_eq!(switched.workspace_id, metadata.workspace_id);
+        // Returning to the old Session cannot recover the invalidated evidence.
+        next.active = metadata.active;
+        writer(next).unwrap();
+        assert!(
+            store
+                .read_by_name("body")
+                .unwrap()
+                .unwrap()
+                .subjektiv_session_attribution
+                .is_none()
+        );
+    }
+}

crates/worker/src/wip.rs

@@ -5566,6 +5566,7 @@ mod tests {
             entry: session_store::LoggedHistoryEntry {
                 item: session_store::LoggedItem::from(saved),
                 metadata: session_store::LoggedSessionHistoryMetadata {
+                    input_metadata: None,
                     entry_id: session_store::LoggedSessionHistoryEntryId::new(),
                     origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
                     derivation: None,

crates/worker/tests/compact_events_test.rs

@@ -119,6 +119,7 @@ fn annotated(item: Item) -> session_store::LoggedHistoryEntry {
     session_store::LoggedHistoryEntry {
         item: session_store::LoggedItem::from(item),
         metadata: session_store::LoggedSessionHistoryMetadata {
+            input_metadata: None,
             entry_id: session_store::LoggedSessionHistoryEntryId::new(),
             origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown,
             derivation: None,
@@ -2212,10 +2213,48 @@ fn summary_entry(
     origin: session_store::LoggedSessionHistoryOrigin,
 ) -> session_store::LoggedHistoryEntry {
     let mut entry = annotated(item);
+    entry.metadata.input_metadata = matches!(
+        &entry.item,
+        session_store::LoggedItem::Message {
+            role: session_store::LoggedRole::System,
+            ..
+        }
+    )
+    .then(|| {
+        session_store::HistoryInputMetadata::new(
+            session_store::HistoryInputUsePolicy {
+                recording_material: session_store::InputEvidenceUse::Exclude,
+                memory_evidence: session_store::InputEvidenceUse::Exclude,
+                observation: session_store::InputObservationUse::Hidden,
+                compaction: match origin {
+                    session_store::LoggedSessionHistoryOrigin::CompactionSummary => {
+                        session_store::InputCompactionUse::OpeningSummary
+                    }
+                    session_store::LoggedSessionHistoryOrigin::JournalContext => {
+                        session_store::InputCompactionUse::ReferenceInput
+                    }
+                    _ => session_store::InputCompactionUse::Retain,
+                },
+            },
+            None,
+            None,
+        )
+        .unwrap()
+    });
     entry.metadata.origin = origin;
     entry
 }
 
+fn compaction_use(
+    metadata: &session_store::LoggedSessionHistoryMetadata,
+) -> session_store::InputCompactionUse {
+    metadata
+        .input_metadata
+        .as_ref()
+        .map(|input| input.usage().compaction)
+        .unwrap_or(session_store::InputCompactionUse::Retain)
+}
+
 fn long_summary(label: &str) -> String {
     format!(
         "## Completed Tasks\n{label}: investigated the transport boundary.\n\n\
@@ -2351,7 +2390,8 @@ async fn full_initial_summary_survives_small_overview_budgets_and_both_retention
             assert_eq!(
                 replacement
                     .iter()
-                    .filter(|entry| entry.metadata.origin == CompactionSummary)
+                    .filter(|entry| compaction_use(&entry.metadata)
+                        == session_store::InputCompactionUse::OpeningSummary)
                     .count(),
                 1
             );
@@ -2596,7 +2636,9 @@ async fn three_chained_compactions_transport_each_complete_predecessor_without_r
         let seed = replacement_seed(&fixture.worker);
         let inherited = seed
             .iter()
-            .filter(|entry| entry.metadata.origin == CompactionSummary)
+            .filter(|entry| {
+                compaction_use(&entry.metadata) == session_store::InputCompactionUse::OpeningSummary
+            })
             .collect::<Vec<_>>();
         assert_eq!(inherited.len(), 1);
         assert_eq!(
@@ -2884,7 +2926,7 @@ impl worker::subjektiv::SubjektivHost for CompactionJournalHost {
     }
 
     fn journal_startup_granted(&self) -> bool {
-        true
+        false
     }
 
     fn journal_startup_context(
@@ -2988,8 +3030,8 @@ fn compaction_snapshot(generation: usize) -> subjektiv::journal::startup::Journa
 fn journal_body(worker: &Worker<MockClient, TestStore>) -> String {
     let seed = replacement_seed(worker);
     assert_eq!(
-        seed[1].metadata.origin,
-        session_store::LoggedSessionHistoryOrigin::JournalContext
+        compaction_use(&seed[1].metadata),
+        session_store::InputCompactionUse::ReferenceInput
     );
     let item = Item::from(seed[1].item.clone());
     assert!(matches!(
@@ -3001,8 +3043,8 @@ fn journal_body(worker: &Worker<MockClient, TestStore>) -> String {
     ));
     assert_eq!(
         seed.iter()
-            .filter(|entry| entry.metadata.origin
-                == session_store::LoggedSessionHistoryOrigin::JournalContext)
+            .filter(|entry| compaction_use(&entry.metadata)
+                == session_store::InputCompactionUse::ReferenceInput)
             .count(),
         1
     );
@@ -3111,20 +3153,34 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
     projection.source_digest = "a".repeat(64);
     let initial_catalog = Arc::new(worker::PromptCatalog::from_projection(projection).unwrap());
     fixture.worker.prompts().store(initial_catalog.clone());
-    let legacy_startup = session_store::LoggedSystemHistoryEntry {
-        item: session_store::SystemItem::JournalStartup {
-            body: "OLD_TYPED_STARTUP_JOURNAL".into(),
-            snapshot: serde_json::json!({"journals": []}),
-            snapshot_sha256: "legacy-startup-digest".into(),
-            prompt_provenance: session_store::PromptRenderProvenance {
-                workspace_id: None,
-                source_digest: "legacy-source".into(),
-                projection_digest: "legacy-projection".into(),
-                logical_name: "internal.subjektiv_journal_startup".into(),
-            },
-        },
-        metadata: summary_entry(Item::system_message(""), DerivedSummary).metadata,
-    };
+    // Already migrated generic input with unrelated provenance/payload; Core
+    // must select it by explicit policy, not identify a Journal Feature.
+    let legacy_startup = session_store::LoggedSystemHistoryEntry::from_input(
+        DerivedSummary,
+        None,
+        session_store::SystemInputEnvelope::new(
+            "OLD_TYPED_STARTUP_JOURNAL",
+            session_store::HistoryInputMetadata::new(
+                session_store::HistoryInputUsePolicy {
+                    recording_material: session_store::InputEvidenceUse::Exclude,
+                    memory_evidence: session_store::InputEvidenceUse::Exclude,
+                    observation: session_store::InputObservationUse::Hidden,
+                    compaction: session_store::InputCompactionUse::ReferenceInput,
+                },
+                Some(
+                    session_store::FeatureInputPayload::encode(
+                        "renamed-owner:reference",
+                        8,
+                        &serde_json::json!({"private": "retained opaque context"}),
+                    )
+                    .unwrap(),
+                ),
+                None,
+            )
+            .unwrap(),
+        )
+        .unwrap(),
+    );
     let legacy_id = legacy_startup.metadata.entry_id.clone();
     fixture
         .restore_seed_with_system(
@@ -3158,7 +3214,7 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
             .unwrap()
             .annotation
             .origin,
-        JournalContext
+        DerivedSummary
     );
     let session = fixture.worker.session_id();
     let original_segment = fixture.worker.segment_id();
@@ -3183,8 +3239,8 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
                     .unwrap()
                     .annotation
                     .origin,
-                JournalContext,
-                "legacy startup classification survives a normal turn"
+                DerivedSummary,
+                "generic source provenance survives a normal turn unchanged"
             );
         }
         assert_eq!(
@@ -3235,7 +3291,7 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
         assert_eq!(host.reads.lock().unwrap().len(), index + 1);
         let history = fixture.worker.history();
         let overview = input
-            .split("## Previous Journal reference (historical, not automatically retained)")
+            .split("## Previous reference input (historical, not automatically retained)")
             .next()
             .unwrap();
         for old in &old_bodies {
@@ -3328,6 +3384,17 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
         .run_text("normal post-compaction turn")
         .await
         .unwrap();
+    let frozen_seed = replacement_seed(&fixture.worker)
+        .into_iter()
+        .filter(|entry| {
+            matches!(
+                compaction_use(&entry.metadata),
+                session_store::InputCompactionUse::ReferenceInput
+                    | session_store::InputCompactionUse::OpeningSummary
+            )
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(frozen_seed.len(), 2);
     let records = fixture
         .worker
         .store()
@@ -3367,6 +3434,20 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai
         .run_text("normal turn after typed history restore")
         .await
         .unwrap();
+    for frozen in frozen_seed {
+        let replayed = fixture
+            .worker
+            .session_history()
+            .entries()
+            .iter()
+            .find(|entry| entry.annotation.entry_id == frozen.metadata.entry_id)
+            .expect("frozen entry identity survives re-restore");
+        assert_eq!(replayed.item, Item::from(frozen.item));
+        assert_eq!(
+            replayed.annotation, frozen.metadata,
+            "origin, derivation, public allowlist, policy and private payload survive atomic seed/restore"
+        );
+    }
     assert_eq!(
         host.reads.lock().unwrap().len(),
         3,
@@ -3437,27 +3518,41 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat
             )
             .await;
         fixture.worker.compact(0).await.unwrap();
-        let body = journal_body(&fixture.worker);
-        let expected = match case {
-            "no-host" | "disabled" => "disabled",
-            "absent" => "absent",
-            _ => "unavailable",
-        };
-        assert!(
-            body.contains(&format!("Journal read status: {expected}.")),
-            "{case}"
-        );
         let input = first_compactor_input(&observed.requests()[1]);
-        assert_frozen_journal_transport(&input, &body);
-        assert_journal_provenance(
-            &body,
-            expected_json.as_deref(),
-            &fixture.worker.prompts().load_full(),
-            (case != "no-host").then_some(COMPACTION_HOST_WORKSPACE_ID),
-        );
+        let body = if matches!(case, "disabled" | "no-host") {
+            assert_eq!(
+                replacement_seed(&fixture.worker)
+                    .iter()
+                    .filter(|entry| compaction_use(&entry.metadata)
+                        == session_store::InputCompactionUse::ReferenceInput)
+                    .count(),
+                0
+            );
+            assert!(!input.contains("Journal reference frozen at this compaction boundary"));
+            String::new()
+        } else {
+            let body = journal_body(&fixture.worker);
+            let expected = if case == "absent" {
+                "absent"
+            } else {
+                "unavailable"
+            };
+            assert!(
+                body.contains(&format!("Journal read status: {expected}.")),
+                "{case}"
+            );
+            assert_frozen_journal_transport(&input, &body);
+            assert_journal_provenance(
+                &body,
+                expected_json.as_deref(),
+                &fixture.worker.prompts().load_full(),
+                Some(COMPACTION_HOST_WORKSPACE_ID),
+            );
+            body
+        };
         assert_full_initial_summary(&input, &inherited);
         assert!(input.contains("UNRECORDED_FALLBACK_GOAL"), "{case}");
-        assert!(input.contains("## Previous Journal reference (historical, not automatically retained)\nPREVIOUS_JOURNAL_DELEGATED_CONTEXT"));
+        assert!(input.contains("## Previous reference input (historical, not automatically retained)\nPREVIOUS_JOURNAL_DELEGATED_CONTEXT"));
         assert!(
             fixture.worker.history().iter().all(|item| !item
                 .as_text()
@@ -3560,3 +3655,132 @@ async fn journal_snapshot_counts_against_compactor_hard_input_cap_without_mutati
         );
     }
 }
+#[tokio::test]
+async fn identical_reference_and_retained_inputs_preserve_distinct_metadata_after_compaction() {
+    use session_store::*;
+    let client = MockClient::new(vec![
+        single_text_events("fixture"),
+        single_text_events("fork checkpoint"),
+        write_summary_tool_use_events("review-summary", "replacement summary"),
+        single_text_events("done"),
+        single_text_events("restore checkpoint"),
+    ]);
+    let mut fixture = SummaryInputFixture::new(MANUAL_ONLY_MANIFEST_TOML, client).await;
+    let make = |producer: &str, compaction| {
+        LoggedSystemHistoryEntry::from_input(
+            LoggedSessionHistoryOrigin::Provided {
+                producer: producer.into(),
+            },
+            None,
+            SystemInputEnvelope::new(
+                "identical model context",
+                HistoryInputMetadata::new(
+                    HistoryInputUsePolicy {
+                        recording_material: InputEvidenceUse::Exclude,
+                        memory_evidence: InputEvidenceUse::Exclude,
+                        observation: InputObservationUse::Hidden,
+                        compaction,
+                    },
+                    Some(
+                        FeatureInputPayload::encode(
+                            "review:owner",
+                            1,
+                            &serde_json::json!({"producer": producer}),
+                        )
+                        .unwrap(),
+                    ),
+                    None,
+                )
+                .unwrap(),
+            )
+            .unwrap(),
+        )
+        .to_history_entry()
+    };
+    let historical = make("old-reference", InputCompactionUse::ReferenceInput);
+    let retained = make("retained-context", InputCompactionUse::Retain);
+    let repeated = make("second-retained-context", InputCompactionUse::Retain);
+    assert_ne!(historical.metadata.entry_id, retained.metadata.entry_id);
+    fixture
+        .restore_seed(
+            vec![historical.clone(), retained.clone(), repeated.clone()],
+            false,
+        )
+        .await;
+    let session = fixture.worker.session_id();
+    let original = fixture.worker.segment_id();
+    fixture
+        .worker
+        .store()
+        .append(
+            session,
+            original,
+            &LogEntry::Extension {
+                ts: 2,
+                domain: "test.fork_drift".into(),
+                payload: serde_json::json!({}),
+            },
+        )
+        .unwrap();
+    fixture.worker.run_text("fork checkpoint").await.unwrap();
+    let fork = fixture.worker.segment_id();
+    assert_ne!(fork, original);
+    let forked = replacement_seed(&fixture.worker);
+    for expected in [&historical, &retained, &repeated] {
+        assert!(forked.contains(expected));
+    }
+    fixture.worker.compact(1_000_000).await.unwrap();
+    let seed = replacement_seed(&fixture.worker);
+    let matching: Vec<_> = seed.iter().filter(|e| e.item == retained.item).collect();
+    assert_eq!(matching.len(), 2);
+    assert_eq!(
+        matching[0].metadata, retained.metadata,
+        "a retained logical entry must not inherit identity/policy/payload from an equal-body reference that was removed"
+    );
+
+    assert_eq!(matching[1].metadata, repeated.metadata);
+    let compacted = fixture.worker.segment_id();
+    let reopened = FsStore::new(fixture._store_dir.path()).unwrap();
+    let records = reopened.read_all(session, compacted).unwrap();
+    assert!(
+        matches!(&records[0], LogEntry::AnnotatedSegmentStart { compacted_from: Some(source), .. } if source.segment_id == fork)
+    );
+    fixture.worker.run_text("restore checkpoint").await.unwrap();
+    let records = fixture.worker.store().read_all(session, compacted).unwrap();
+    let snapshot = public_snapshot::project_current_session_snapshot(&records);
+    let checkpoint = snapshot
+        .entries
+        .iter()
+        .rev()
+        .find(|entry| {
+            matches!(
+                entry.data,
+                protocol::SessionSnapshotEntryData::UserInput { .. }
+            )
+        })
+        .unwrap();
+    fixture
+        .worker
+        .rewind_to(checkpoint.entry_id.clone())
+        .await
+        .unwrap();
+    for expected in [&retained, &repeated] {
+        assert!(
+            fixture
+                .worker
+                .session_history()
+                .entries()
+                .iter()
+                .any(|entry| entry.annotation == expected.metadata
+                    && entry.item == Item::from(expected.item.clone()))
+        );
+    }
+    assert!(
+        !fixture
+            .worker
+            .session_history()
+            .entries()
+            .iter()
+            .any(|entry| entry.annotation.entry_id == historical.metadata.entry_id)
+    );
+}

crates/worker/tests/controller_test.rs

@@ -3091,7 +3091,7 @@ async fn socket_worker_event_turn_ended_while_idle_auto_starts_turn() {
                     Ok(Some(Event::SystemItem { ref item, .. }))
                         if item.get("kind").and_then(|k| k.as_str()) == Some("worker_event")
                             && item
-                                .pointer("/event/worker_name")
+                                .pointer("/presentation/event/worker_name")
                                 .and_then(|v| v.as_str()) == Some("child") =>
                     {
                         saw_worker_event_echo = true;

crates/worker/tests/feature_invocation_test.rs

@@ -14,7 +14,7 @@ use protocol::{
     InvocationArgumentDescriptor, InvocationArgumentType, InvocationArgumentValue,
     InvocationCompletion, InvocationValue, Segment,
 };
-use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry, SystemItem};
+use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry};
 use worker::feature::{
     FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInvocationContext,
     FeatureInvocationHandler, FeatureInvocationHandlerError, FeatureModule, FeatureRegistryBuilder,
@@ -224,10 +224,20 @@ fn results(worker: &Worker<Trace, TestStore>) -> Vec<FeatureInvocationResult> {
         .0
         .into_iter()
         .filter_map(|entry| match entry {
-            LogEntry::AnnotatedSystemItem { entry, .. } => match entry.item {
-                SystemItem::FeatureInvocationResult { result, .. } => Some(result),
-                _ => None,
-            },
+            LogEntry::AnnotatedSystemItem { entry, .. } => {
+                #[derive(serde::Deserialize)]
+                struct SavedResult {
+                    result: FeatureInvocationResult,
+                }
+                entry
+                    .metadata
+                    .input_metadata
+                    .as_ref()?
+                    .feature_payload()?
+                    .decode::<SavedResult>("feature:invocation-result", 1)
+                    .ok()
+                    .map(|input| input.result)
+            }
             _ => None,
         })
         .collect()