fix(runtime): preserve prompt cache and durable provider context

crates/astra-cli/src/cli/chat_stream/sse_loop/agentic_loop_turn.rs

@@ -147,9 +147,7 @@ fn message_has_tool_calls(m: &Value) -> bool {
 
 fn retained_history_messages(messages: &[Value]) -> &[Value] {
     match messages.split_last() {
-        Some((last, history)) if last.get("role").and_then(Value::as_str) == Some("user") => {
-            history
-        }
+        Some((last, history)) if astra_turn_types::is_human_user_message(last) => history,
         _ => messages,
     }
 }
@@ -193,19 +191,38 @@ fn project_cross_session_memory_hits(
 fn build_retained_history_turns(
     messages: &[Value],
 ) -> Vec<astra_turn_core::context_assembly_trace::TurnRetention> {
-    let mut turns = Vec::new();
+    let mut turns: Vec<astra_turn_core::context_assembly_trace::TurnRetention> = Vec::new();
 
     for message in messages {
+        let content = msg_content(message);
         let role = message
             .get("role")
             .and_then(Value::as_str)
             .unwrap_or("unknown")
             .to_string();
-        let tokens = prompts::estimate_str_tokens(&msg_content(message)) as u32;
+        let tokens = prompts::estimate_str_tokens(&content) as u32;
         let has_tool_calls = message_has_tool_calls(message);
-        let preview = retained_history_preview(&role, &msg_content(message));
+        if astra_turn_types::is_runtime_owned_message(message) {
+            // Provider occupancy still includes append-only authority, but
+            // user-facing history previews and semantic role summaries must
+            // never expose or attribute runtime control payloads to a human.
+            if let Some(turn) = turns.last_mut() {
+                turn.tokens = turn.tokens.saturating_add(tokens);
+                turn.has_tool_calls |= has_tool_calls;
+            } else {
+                turns.push(astra_turn_core::context_assembly_trace::TurnRetention {
+                    turn_index: 0,
+                    role: "runtime".to_string(),
+                    tokens,
+                    has_tool_calls,
+                    content_preview: String::new(),
+                });
+            }
+            continue;
+        }
+        let preview = retained_history_preview(&role, &content);
 
-        if turns.is_empty() || role == "user" {
+        if turns.is_empty() || astra_turn_types::is_human_user_message(message) {
             turns.push(astra_turn_core::context_assembly_trace::TurnRetention {
                 turn_index: turns.len() as u32,
                 role,
@@ -217,7 +234,7 @@ fn build_retained_history_turns(
         }
 
         if let Some(turn) = turns.last_mut() {
-            turn.tokens += tokens;
+            turn.tokens = turn.tokens.saturating_add(tokens);
             turn.has_tool_calls |= has_tool_calls;
             if retained_turn_role_priority(&role) > retained_turn_role_priority(&turn.role) {
                 turn.role = role;
@@ -2351,6 +2368,7 @@ mod tests {
                 "advisories": [{"kind": "test_signal"}]
             }),
             round_index: 3,
+            attempt_leased: false,
         };
         let required: Vec<String> = Vec::new();
         let volatile_texts: Vec<String> = Vec::new();
@@ -2655,6 +2673,44 @@ mod tests {
         assert!(turns[0].has_tool_calls);
     }
 
+    #[test]
+    fn retained_history_accounts_runtime_tokens_without_exposing_control_preview() {
+        use astra_runtime::prompts;
+
+        let mut authority = json!({
+            "role": "user",
+            "content": "<runtime-authority-frame>\ninternal Work settlement\n</runtime-authority-frame>"
+        });
+        astra_turn_types::mark_append_only_required_context(
+            &mut authority,
+            "final_work_synthesis",
+            astra_turn_types::RuntimeAuthorityLifetime::CurrentUserTurn,
+        );
+        let messages = vec![
+            json!({"role": "user", "content": "real request"}),
+            authority,
+            json!({"role": "assistant", "content": "visible answer"}),
+        ];
+        let expected_tokens = messages
+            .iter()
+            .map(|message| prompts::estimate_str_tokens(&msg_content(message)) as u32)
+            .fold(0_u32, u32::saturating_add);
+
+        let turns = build_retained_history_turns(&messages);
+
+        assert_eq!(turns.len(), 1);
+        assert_eq!(turns[0].tokens, expected_tokens);
+        assert_eq!(turns[0].role, "assistant");
+        assert!(turns[0].content_preview.contains("real request"));
+        assert!(turns[0].content_preview.contains("visible answer"));
+        assert!(!turns[0].content_preview.contains("runtime-authority-frame"));
+        assert!(
+            !turns[0]
+                .content_preview
+                .contains("internal Work settlement")
+        );
+    }
+
     #[test]
     fn retained_history_keeps_system_role_for_system_only_history() {
         let messages = vec![json!({"role": "system", "content": "system note"})];

crates/astra-cli/src/cli/chat_stream/sse_loop/mod.rs

@@ -519,8 +519,7 @@ pub(crate) async fn stream_chat_sse(
     // preceding prompt history is inherited context, not a new conversation
     // item for this run.
     let root_initial_transcript_item = messages.last().and_then(|message| {
-        (message.get("role").and_then(serde_json::Value::as_str) == Some("user"))
-            .then(|| message.clone())
+        astra_turn_types::is_human_user_message(message).then(|| message.clone())
     });
 
     // ─── Context pre-fetch (disabled) ─────────────────────────────────────
@@ -828,6 +827,7 @@ pub(crate) async fn stream_chat_sse(
         current_session_id,
         current_run_id: Some(parent_turn_run_id.clone()),
         current_run_owner_generation: None,
+        provider_canonical_wal_head_transition_id: None,
         inference_purpose: astra_turn_types::InferencePurpose::PrimaryAgent,
         context_manifest_pool: None,
         context_manifest_user_id: persist_session_artifacts.then_some(current_user_id),
@@ -1016,6 +1016,7 @@ pub(crate) async fn stream_chat_sse(
         budget_wrapup_injected: false,
         context_compression_triggered: false,
         canonical_rewrite_state: Default::default(),
+        provider_canonical_wal_base: None,
         budget_wrapup_ignored_rounds: 0,
         compact_tier_applied: astra_turn_core::compaction_types::CompactionTier::Normal,
         skill_produced_output: false,
@@ -1031,8 +1032,8 @@ pub(crate) async fn stream_chat_sse(
         session_facts: Default::default(),
         // Canonical Server execution is the sole per-turn memory producer.
         memory_extraction_service: None,
-        compact_strategy: astra_turn_core::microcompact::CompactStrategy::from_provider_and_model(
-            p.provider, p.model,
+        compact_strategy: astra_turn_core::microcompact::CompactStrategy::from_explicit_or_provider(
+            None, p.provider,
         ),
         approval_overrides: initial_approval_overrides,
         confidence_trend: Default::default(),

crates/astra-cli/src/cli/chat_stream/sse_loop/server_admission_host.rs

@@ -723,7 +723,7 @@ impl AgenticLoopHost for CliServerAdmissionHost<'_> {
         let effective_model_owned = self.model.map(str::to_owned);
         let effective_model = effective_model_owned.as_deref();
         let effective_offering_id = self.offering_id.as_deref();
-        let runtime_volatile_injections = state.take_volatile_pending();
+        let runtime_volatile_injections = state.lease_volatile_pending()?;
         let runtime_volatile_texts = self
             .input_runtime_volatile_texts
             .iter()

crates/astra-cli/src/cli/delegate_subrun.rs

@@ -305,9 +305,9 @@ impl SubRunExecutor for CliDelegateSubRunExecutor {
             .as_deref()
             .map(|model| astra_turn_core::thinking_config::resolve_model_thinking(model).1)
             .unwrap_or_default();
-        let compact_strategy = astra_turn_core::microcompact::CompactStrategy::from_provider_hint(
-            effective_model.as_deref().unwrap_or(""),
-        );
+        // The model alias does not establish a cache protocol. The admitted
+        // server execution owns provider-specific request shaping.
+        let compact_strategy = astra_turn_core::microcompact::CompactStrategy::default();
         // Resolve per-model workflow-guard policy up front; `effective_model`
         // is moved into the SubRunHost below.
         let resolved_tool_policy = astra_config::runtime_config::RuntimeConfig::load()
@@ -496,6 +496,7 @@ impl SubRunExecutor for CliDelegateSubRunExecutor {
             current_session_id: Some(config.session_id.clone()),
             current_run_id: Some(config.run_id.clone()),
             current_run_owner_generation: None,
+            provider_canonical_wal_head_transition_id: None,
             inference_purpose: astra_turn_types::InferencePurpose::SubAgent,
             context_manifest_pool: None,
             context_manifest_user_id: Some(user_id),
@@ -624,6 +625,7 @@ impl SubRunExecutor for CliDelegateSubRunExecutor {
             budget_wrapup_injected: false,
             context_compression_triggered: false,
             canonical_rewrite_state: Default::default(),
+            provider_canonical_wal_base: None,
             budget_wrapup_ignored_rounds: 0,
             compact_tier_applied: astra_turn_core::compaction_types::CompactionTier::Normal,
             skill_produced_output: false,

crates/astra-cli/src/cli/self_command.rs

@@ -991,7 +991,7 @@ fn restored_recent_turn_previews(
     for message in restored.resume_messages() {
         let role = message.get("role").and_then(serde_json::Value::as_str);
         match role {
-            Some("user") => {
+            Some("user") if astra_turn_types::is_human_user_message(message) => {
                 pending_user = extract_text_content(message);
             }
             Some("assistant") => {
@@ -1446,7 +1446,7 @@ mod tests {
         EventPreview, analysis_view_recent_event_previews, build_reflect_response,
         cli_provider_visible_tool_names, event_preview_has_adverse_signal, event_preview_summary,
         execute_self_command, persist_config_override, replace_json_path, resolve_session_id,
-        session_agent_delivery_summary, verify_runtime_config,
+        restored_recent_turn_previews, session_agent_delivery_summary, verify_runtime_config,
     };
     use crate::cli::cli_config::cli_args::{SelfCmd, SelfReflectArgs, SelfSessionArgs};
     use crate::cli::cli_config::cli_utils::{
@@ -1489,6 +1489,43 @@ mod tests {
         }
     }
 
+    #[test]
+    fn restored_preview_does_not_show_runtime_authority_as_user_input() {
+        let mut runtime = serde_json::json!({"role": "user", "content": "runtime control"});
+        astra_turn_types::mark_append_only_required_context(
+            &mut runtime,
+            "final_answer_settlement",
+            astra_turn_types::RuntimeAuthorityLifetime::NextAssistantDecision,
+        );
+        let messages = vec![
+            serde_json::json!({"role": "user", "content": "real request"}),
+            runtime,
+            serde_json::json!({"role": "assistant", "content": "answer"}),
+        ];
+        let artifacts = astra_services::self_surface::LoadedSelfSurfaceArtifacts {
+            session_id: "sid".to_string(),
+            workspace: None,
+            restored: Some(astra_services::session_restore::RestoredSession {
+                session_id: "sid".to_string(),
+                resume_bundle: Some(typed_resume_bundle("sid", 1, messages)),
+                ..Default::default()
+            }),
+            journal_events: Vec::new(),
+            latest_full_context_trace: None,
+        };
+
+        let previews = restored_recent_turn_previews(&artifacts, 4);
+        assert_eq!(previews.len(), 1);
+        assert_eq!(
+            previews[0].user_input_preview.as_deref(),
+            Some("real request")
+        );
+        assert_eq!(
+            previews[0].assistant_output_preview.as_deref(),
+            Some("answer")
+        );
+    }
+
     #[test]
     fn reflect_interruption_preserves_resume_and_stall_causality() {
         let event = EventPreview {

crates/astra-cli/src/cli/skill_subrun.rs

@@ -435,7 +435,7 @@ impl AgenticLoopHost for SubRunHost {
         // Drain runtime volatile as typed edge metadata. Do not splice it into
         // messages[]: that loses producer kind, pollutes prompt-facing history,
         // and makes soft runtime evidence look like user content.
-        let runtime_volatile_injections = state.take_volatile_pending();
+        let runtime_volatile_injections = state.lease_volatile_pending()?;
 
         let effective_model = self.model.as_deref();
         let effective_offering_id = self.offering_id.clone();
@@ -1066,9 +1066,9 @@ impl SkillSubRunExecutor for CliSkillSubRunExecutor {
             .as_deref()
             .map(|model| astra_turn_core::thinking_config::resolve_model_thinking(model).1)
             .unwrap_or_default();
-        let compact_strategy = astra_turn_core::microcompact::CompactStrategy::from_provider_hint(
-            effective_model.as_deref().unwrap_or(""),
-        );
+        // The model alias does not establish a cache protocol. The admitted
+        // server execution owns provider-specific request shaping.
+        let compact_strategy = astra_turn_core::microcompact::CompactStrategy::default();
         // Resolve per-model workflow-guard policy up front; `effective_model`
         // is moved into the SubRunHost below.
         let resolved_tool_policy = astra_config::runtime_config::RuntimeConfig::load()
@@ -1195,6 +1195,7 @@ impl SkillSubRunExecutor for CliSkillSubRunExecutor {
             current_session_id: Some(parent_session_id.to_string()),
             current_run_id: Some(parent_run_id.to_string()),
             current_run_owner_generation: None,
+            provider_canonical_wal_head_transition_id: None,
             inference_purpose: astra_turn_types::InferencePurpose::SubAgent,
             context_manifest_pool: None,
             context_manifest_user_id: Some(user_id),
@@ -1303,6 +1304,7 @@ impl SkillSubRunExecutor for CliSkillSubRunExecutor {
             budget_wrapup_injected: false,
             context_compression_triggered: false,
             canonical_rewrite_state: Default::default(),
+            provider_canonical_wal_base: None,
             budget_wrapup_ignored_rounds: 0,
             compact_tier_applied: astra_turn_core::compaction_types::CompactionTier::Normal,
             skill_produced_output: false,
@@ -1794,6 +1796,7 @@ mod tests {
             kind: astra_runtime::turn::agentic_loop::host::VolatileKind::PolicyAdvisory,
             payload: json!({"signal": "soft subrun evidence"}),
             round_index: 2,
+            attempt_leased: false,
         }];
 
         attach_runtime_volatile_injections(&mut payload, &injections);

crates/astra-cli/src/cli/slash/slash_cache.rs

@@ -305,6 +305,7 @@ mod tests {
             round,
             provider: "openai".into(),
             model: "test-model".into(),
+            cache_capability: None,
             cache_read_tokens,
             cache_creation_tokens: 0,
             tool_count: 0,
@@ -432,6 +433,7 @@ mod tests {
             Some(PromptCacheCapabilityData {
                 protocol: astra_services::PromptCacheProtocolData::OpenAiAutoPrefix,
                 volatile_placement: astra_services::PromptCacheVolatilePlacementData::TailSuffix,
+                volatile_delivery: astra_services::PromptCacheVolatileDeliveryData::All,
                 reuse_scope: Some(PromptCacheReuseScopeData::ConversationTurns),
             }),
             &turns,
@@ -452,6 +454,7 @@ mod tests {
             Some(PromptCacheCapabilityData {
                 protocol: astra_services::PromptCacheProtocolData::OpenAiAutoPrefix,
                 volatile_placement: astra_services::PromptCacheVolatilePlacementData::TailSuffix,
+                volatile_delivery: astra_services::PromptCacheVolatileDeliveryData::All,
                 reuse_scope: Some(PromptCacheReuseScopeData::ConversationTurns),
             }),
             &[],

crates/astra-cli/src/cli/slash/slash_debug.rs

@@ -100,8 +100,7 @@ pub(crate) fn handle_debug_command(arg: &str, state: &SessionState) {
                     .delta
                     .iter()
                     .chain(view.full.iter())
-                    .find(|m| m.get("role").and_then(|v| v.as_str()) == Some("user"))
-                    .and_then(|m| m.get("content").and_then(|v| v.as_str()))
+                    .find_map(human_user_text)
                     .unwrap_or("(unknown)")
                     .to_string(),
                 tokens_in: 0,
@@ -174,6 +173,12 @@ pub(crate) fn handle_debug_command(arg: &str, state: &SessionState) {
     }
 }
 
+fn human_user_text(message: &serde_json::Value) -> Option<&str> {
+    astra_turn_types::is_human_user_message(message)
+        .then(|| message.get("content").and_then(serde_json::Value::as_str))
+        .flatten()
+}
+
 // ── Overview ──────────────────────────────────────────────────────────────────
 
 fn print_overview(session_id: &str, turns: &[TurnSummary], checkpoints: &[PathBuf]) {
@@ -1126,7 +1131,7 @@ fn truncate(s: &str, max: usize) -> String {
 #[cfg(test)]
 mod tests {
     use super::{
-        build_turn_messages_view, list_heavy_checkpoints, load_journal_turns,
+        build_turn_messages_view, human_user_text, list_heavy_checkpoints, load_journal_turns,
         load_messages_from_heavy_path, message_delta, resolve_session_id, truncate,
     };
     use serde_json::json;
@@ -1394,6 +1399,20 @@ mod tests {
         assert_eq!(v.warning, None);
     }
 
+    #[test]
+    fn checkpoint_fallback_preview_excludes_runtime_user_frame() {
+        let human = json!({"role": "user", "content": "real request"});
+        let mut runtime = json!({"role": "user", "content": "runtime control"});
+        astra_turn_types::mark_append_only_required_context(
+            &mut runtime,
+            "final_answer_settlement",
+            astra_turn_types::RuntimeAuthorityLifetime::NextAssistantDecision,
+        );
+
+        assert_eq!(human_user_text(&human), Some("real request"));
+        assert_eq!(human_user_text(&runtime), None);
+    }
+
     // ── format_tool_display_from_preview ──────────────────────────────
 
     #[test]