refactor(runtime): retire unowned recovery and judgment protocols

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

@@ -129,17 +129,6 @@ pub(crate) fn resolved_tool_metrics<I>(
     fallback_count: u32,
     fallback_tools: I,
     tool_call_records: &[ToolCallRecord],
-) -> (u32, Vec<String>)
-where
-    I: IntoIterator<Item = String>,
-{
-    resolved_tool_metrics_with_authority(fallback_count, fallback_tools, tool_call_records, false)
-}
-
-fn resolved_tool_metrics_with_authority<I>(
-    fallback_count: u32,
-    fallback_tools: I,
-    tool_call_records: &[ToolCallRecord],
     aggregate_authoritative: bool,
 ) -> (u32, Vec<String>)
 where
@@ -266,11 +255,11 @@ pub(crate) fn build_stream_result(ctx: StreamResultBuild<'_>) -> StreamResult {
         run_transcript_messages,
         applied_user_intents,
     } = ctx;
-    let (_, tools_used) = resolved_tool_metrics_with_authority(
+    let (_, tools_used) = resolved_tool_metrics(
         tool_calls_count,
         tools_used,
         &tool_call_records,
-        tool_record_coverage_partial,
+        server_terminal_authoritative || tool_record_coverage_partial,
     );
     let mut interruption = interruption;
     let mut server_terminal_unverified = server_terminal_unverified;
@@ -665,6 +654,7 @@ mod tests {
                 String::new(),
             ],
             &[],
+            false,
         );
 
         assert_eq!(count, 4);
@@ -680,7 +670,7 @@ mod tests {
             tool_record(" read_file", true, Some("contents")),
         ];
 
-        let (count, tools) = resolved_tool_metrics(0, std::iter::empty(), &records);
+        let (count, tools) = resolved_tool_metrics(0, std::iter::empty(), &records, false);
 
         assert_eq!(count, 3);
         assert_eq!(tools, vec!["bash".to_string(), "read_file".to_string()]);
@@ -695,7 +685,9 @@ mod tests {
         ctx.tool_ledger_aggregate = succeeded_aggregate(34);
         ctx.tools_used = HashSet::from(["bash".to_string(), "write_file".to_string()]);
         ctx.server_terminal_authoritative = true;
-        ctx.tool_record_coverage_partial = true;
+        // A complete Server ledger still does not make the edge-only local
+        // records a complete projection of all executed tools.
+        ctx.tool_record_coverage_partial = false;
         ctx.tool_call_records = vec![tool_record("bash", false, Some("first run failed"))];
 
         let result = build_stream_result(ctx);

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

@@ -1053,6 +1053,7 @@ pub(crate) async fn stream_chat_sse(
             state.total_tool_calls,
             state.telemetry.all_tools_used.iter().cloned(),
             &state.stall.tool_call_records,
+            !state.telemetry.server_summary_run_usage.is_empty(),
         );
         let tool_outcomes = astra_services::session_journal::ToolOutcomeSummary::from_records(
             &state.stall.tool_call_records,

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

@@ -2152,16 +2152,12 @@ impl AgenticLoopHost for CliServerAdmissionHost<'_> {
                 })
                 .cloned()
                 .collect::<Vec<_>>();
-            let mut tools_used = if tool_records.is_empty() {
-                astra_core::canonical_names::normalize_name_list(
-                    state.telemetry.all_tools_used.iter().cloned(),
-                )
-            } else {
-                astra_core::canonical_names::normalize_name_list(
-                    tool_records.iter().map(|record| record.name.clone()),
-                )
-            };
-            tools_used.sort_unstable();
+            let (_, tools_used) = super::agentic_sse_loop::resolved_tool_metrics(
+                state.total_tool_calls,
+                state.telemetry.all_tools_used.iter().cloned(),
+                &tool_records,
+                !state.telemetry.server_summary_run_usage.is_empty(),
+            );
             inc.replace_tool_records(tool_records);
             // Retained records may be filtered (policy/synthetic entries) and
             // may overlap live edge records already counted incrementally.

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

@@ -214,7 +214,7 @@ fn tool_request_for_run(
         "turn_chain_id": turn_chain_id,
         "request_id": call_id,
         "schema_admitted_by_server": true,
-        "execution_timeout_ms": 300_000,
+        "execution_timeout_ms": 300_000, "command_timeout_cap_ms": 30_000,
         "execution_deadline_unix_ms": 4_102_444_800_000_u64,
         "tool": tool,
         "args": args,

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

@@ -5347,8 +5347,8 @@ async fn apply_restored_session(
             }
         }
     };
-    // Crash recovery has one current state-machine protocol. Missing crash
-    // state means there is nothing to replay; malformed state fails closed.
+    // Local recovery validates checkpoint and journal facts; it does not
+    // re-execute tools. Missing state needs no restore; malformed state fails closed.
     let step_restored = if !local_checkpoint_is_admissible {
         tracing::warn!(
             session_id = %restored.session_id,
@@ -5361,13 +5361,12 @@ async fn apply_restored_session(
                 restored: cr_restored,
                 ..
             })) => {
-                tracing::info!("crash recovery: auto-recovered via state machine");
+                tracing::info!("crash recovery: restored validated checkpoint");
                 Some(cr_restored)
             }
             Ok(Some(astra_pipeline::crash_recovery::RecoveryOutcome::RequiresUserInput {
                 pending_decisions,
                 restored: cr_restored,
-                mut manager,
                 ..
             })) => {
                 tracing::info!(
@@ -5388,30 +5387,18 @@ async fn apply_restored_session(
                         tool_name.clone().bold()
                     );
 
-                    let reason = match decision {
-                        astra_pipeline::crash_recovery::ToolReplayDecision::RequireUserInput {
-                            reason,
-                        } => reason.clone(),
-                        astra_pipeline::crash_recovery::ToolReplayDecision::InFlightAtCrash {
-                            tool_name: tn,
-                        } => {
-                            format!("Tool '{}' was in-flight when crash occurred", tn)
-                        }
-                        _ => "Unknown".to_string(),
-                    };
-                    eprintln!("      Reason: {}", reason.dim());
+                    eprintln!("      Reason: {}", decision.as_str().dim());
                 }
 
                 eprintln!();
                 eprintln!("  {}", "Options:".bold());
-                eprintln!("    [r] Replay - Re-execute the tool");
-                eprintln!("    [s] Skip - Skip this tool (use cached result if available)");
+                eprintln!(
+                    "    [c] Continue - Accept unknown tool effects and resume the conversation"
+                );
+                eprintln!("        No tool is re-executed or skipped by recovery");
                 eprintln!("    [a] Abort - Abort recovery and fail");
                 eprintln!();
-                eprint!(
-                    "  {} ",
-                    "Choose action for all pending tools (r/s/a):".bold()
-                );
+                eprint!("  {} ", "Choose whether to continue or abort (c/a):".bold());
                 let _ = std::io::stderr().flush();
 
                 let mut input = String::new();
@@ -5421,21 +5408,11 @@ async fn apply_restored_session(
 
                 let choice = input.trim().to_lowercase();
                 match choice.as_str() {
-                    "r" | "replay" | "s" | "skip" => {
-                        let label = if choice.starts_with('r') {
-                            "replay"
-                        } else {
-                            "skip"
-                        };
+                    "c" | "continue" => {
                         eprintln!(
-                            "  {} User chose to {} all pending tools",
-                            "✓".green(),
-                            label
+                            "  {} Continuing with acknowledged unknown tool effects",
+                            "✓".green()
                         );
-                        // force_complete transitions Replaying -> Recovered, bypassing pending checks
-                        if let Err(e) = manager.force_complete() {
-                            return Err(format!("Failed to apply {} decisions: {}", label, e));
-                        }
                         Some(cr_restored)
                     }
                     "a" | "abort" => {

crates/astra-cli/src/cli/stream/mcp_result_tests.rs

@@ -105,6 +105,7 @@ async fn callback_through_pipeline(
             tool: tool.clone(),
             args: args.clone(),
             execution_timeout_ms: 30_000,
+            command_timeout_cap_ms: None,
             execution_deadline_unix_ms: u64::MAX,
         }])
         .await;