task: add lifecycle timelines and recovery to the logs console

crates/mesh-llm-host-runtime/src/api/routes/logs/cleanup.rs

@@ -49,6 +49,8 @@ struct CleanupScopeDto {
     #[serde(skip_serializing_if = "Option::is_none")]
     route: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    exclude_route: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     model: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     provider: Option<String>,
@@ -146,6 +148,7 @@ impl CleanupScopeDto {
             from: filters.from().map(str::to_owned),
             to: filters.to().map(str::to_owned),
             route: filters.route().map(str::to_owned),
+            exclude_route: filters.exclude_route().map(str::to_owned),
             model: filters.model().map(str::to_owned),
             provider: filters.provider().map(str::to_owned),
             engine: filters.engine().map(str::to_owned),

crates/mesh-llm-host-runtime/src/api/routes/logs/events/protocol/audit_entry/tests.rs

@@ -24,15 +24,19 @@ fn audit_record(sequence: u64) -> AuditReplayRecord {
     }
 }
 
-fn durable_audit_detail(entry_id: &str, detail: serde_json::Value) -> AuditEntryDetail {
+fn durable_audit_detail(
+    entry_id: &str,
+    source: &str,
+    detail: serde_json::Value,
+) -> AuditEntryDetail {
     let root = tempfile::tempdir().expect("temporary durable audit store");
     let store = LogStore::open(root.path(), Arc::new(RealClock)).expect("open audit store");
     store
         .insert_audit_entry(
             entry_id,
             None,
             "2026-01-01T00:00:00Z",
-            "cli",
+            source,
             "command_completed",
             Some(&detail.to_string()),
         )
@@ -270,6 +274,7 @@ fn live_audit_frame_preserves_valid_command_summary() {
 fn durable_audit_frame_drops_malformed_command_summary() {
     let record = durable_audit_detail(
         "id-9",
+        "cli",
         serde_json::json!({
             "context_version": 1,
             "command_summary": "mesh-llm gpus --draft run-benchmark --backend cuda",
@@ -284,6 +289,7 @@ fn durable_audit_frame_drops_malformed_command_summary() {
 fn durable_audit_frame_drops_deep_malformed_command_summary() {
     let record = durable_audit_detail(
         "id-12",
+        "cli",
         serde_json::json!({
             "context_version": 1,
             "command_summary": "mesh-llm load unload status discover rotate-key setup --port 1234",
@@ -298,6 +304,7 @@ fn durable_audit_frame_drops_deep_malformed_command_summary() {
 fn durable_audit_frame_preserves_valid_command_summary() {
     let record = durable_audit_detail(
         "id-11",
+        "cli",
         serde_json::json!({
             "context_version": 1,
             "command_summary": "mesh-llm runtime guardrails --mode metrics --port 41731 --root-relay [REDACTED]",
@@ -314,6 +321,7 @@ fn durable_audit_frame_preserves_valid_command_summary() {
 fn durable_audit_frame_redacts_unsafe_rest_parity_metadata() {
     let record = durable_audit_detail(
         "id-14",
+        "cli",
         serde_json::json!({
             "context_version": 1,
             "subject_id": "https://alice:subject-secret@example.test/model?api_key=subject-query",
@@ -340,3 +348,22 @@ fn durable_audit_frame_redacts_unsafe_rest_parity_metadata() {
         assert!(!serialized.contains(unsafe_value));
     }
 }
+
+#[test]
+fn durable_audit_frame_projects_legacy_logging_source_as_canonical() {
+    let record = durable_audit_detail(
+        "legacy-logging-entry",
+        "logging-runtime",
+        serde_json::json!({}),
+    );
+
+    let frame = durable_audit_entry_frame(record).expect("durable audit frame");
+    let data = frame_data(&frame);
+
+    assert!(frame.contains("event: audit_entry"));
+    assert!(frame.contains("id: a1:1"));
+    assert_eq!(data["entryId"], "legacy-logging-entry");
+    assert_eq!(data["source"], "logging_service");
+    assert_eq!(data["sequence"], 1);
+    assert!(!frame.contains("logging-runtime"));
+}

crates/mesh-llm-host-runtime/src/api/routes/logs/parse.rs

@@ -133,6 +133,8 @@ struct CleanupPreviewBody {
     #[serde(default)]
     route: Option<String>,
     #[serde(default)]
+    exclude_route: Option<String>,
+    #[serde(default)]
     model: Option<String>,
     #[serde(default)]
     provider: Option<String>,
@@ -236,7 +238,8 @@ pub(super) fn cleanup_preview_request(
             .as_deref()
             .map(mesh_llm_log_store::CleanupOutcome::try_from)
             .transpose()?,
-    )?;
+    )?
+    .with_exclude_route(body.exclude_route)?;
     let scope = mesh_llm_log_store::CleanupScope::new(
         mesh_llm_log_store::MaintenanceTimestamp::try_from(cutoff.as_str())?,
         body.request_limit,
@@ -648,7 +651,7 @@ mod tests {
     fn cleanup_parsing_normalizes_and_rejects_unbounded_input_before_store_access() {
         let operation_id = uuid::Uuid::new_v4();
         let body = format!(
-            r#"{{"operationId":"{operation_id}","cutoffBefore":"2026-08-03T01:00:00+01:00","requestLimit":1,"source":"durable","from":"2026-08-01T01:00:00+01:00","to":"2026-08-03T00:00:00Z","route":"route-a","model":"Qwen/Qwen3","provider":"mesh","engine":"skippy","outcome":"completed","reason":"operator cleanup"}}"#
+            r#"{{"operationId":"{operation_id}","cutoffBefore":"2026-08-03T01:00:00+01:00","requestLimit":1,"source":"durable","from":"2026-08-01T01:00:00+01:00","to":"2026-08-03T00:00:00Z","route":"route-a","excludeRoute":"models","model":"Qwen/Qwen3","provider":"mesh","engine":"skippy","outcome":"completed","reason":"operator cleanup"}}"#
         );
         let preview = cleanup_preview_request("/api/logs/cleanup/preview", &body)
             .expect("bounded cleanup preview");
@@ -667,6 +670,7 @@ mod tests {
             Some("2026-08-03T00:00:00.000000000Z")
         );
         assert_eq!(preview.scope.filters().model(), Some("Qwen/Qwen3"));
+        assert_eq!(preview.scope.filters().exclude_route(), Some("models"));
         assert_eq!(
             preview.scope.filters().outcome(),
             Some(mesh_llm_log_store::CleanupOutcome::Completed)

crates/mesh-llm-host-runtime/src/api/routes/logs/tests/audit.rs

@@ -116,6 +116,53 @@ async fn audit_filters_by_source() {
     assert_eq!(json["items"][0]["source"], "mesh");
 }
 
+#[tokio::test]
+async fn audit_logging_service_filter_includes_legacy_rows_with_canonical_source() {
+    let (_temp, state) = runtime();
+    let store = state.store().expect("store");
+    for (entry_id, occurred_at, source) in [
+        (
+            "00000000-0000-4000-8000-000000000012",
+            "2026-01-01T00:00:00Z",
+            "logging-runtime",
+        ),
+        (
+            "00000000-0000-4000-8000-000000000013",
+            "2026-01-01T00:00:01Z",
+            "logging_service",
+        ),
+        (
+            "00000000-0000-4000-8000-000000000014",
+            "2026-01-01T00:00:02Z",
+            "runtime",
+        ),
+    ] {
+        store
+            .insert_audit_entry(entry_id, None, occurred_at, source, "health_check", None)
+            .expect("seed audit row");
+    }
+
+    let page = list_audits(&state, "/api/logs/audit?source=logging_service&limit=10")
+        .await
+        .expect("filter logging service rows");
+    let json = serde_json::to_value(page).expect("serialize page");
+    let items = json["items"].as_array().expect("items");
+
+    assert_eq!(items.len(), 2);
+    assert_eq!(
+        items
+            .iter()
+            .map(|item| item["entryId"].as_str().expect("entry id"))
+            .collect::<Vec<_>>(),
+        vec![
+            "00000000-0000-4000-8000-000000000013",
+            "00000000-0000-4000-8000-000000000012",
+        ]
+    );
+    assert!(items.iter().all(|item| item["source"] == "logging_service"));
+    assert!(!json.to_string().contains("logging-runtime"));
+}
+
 #[tokio::test]
 async fn audit_filters_by_severity() {
     let (_temp, state) = runtime();

crates/mesh-llm-host-runtime/src/api/tests/logs_api_routes/read_and_export.rs

@@ -322,6 +322,7 @@ async fn cleanup_preview_and_run_share_receipt_and_cascade_only_selected_artifac
         "from": "2026-08-01T00:00:00Z",
         "to": "2026-08-02T00:00:00Z",
         "route": "cleanup-route",
+        "excludeRoute": "models",
         "model": "cleanup-model",
         "provider": "mesh",
         "engine": "skippy",
@@ -364,6 +365,7 @@ async fn cleanup_preview_and_run_share_receipt_and_cascade_only_selected_artifac
             "from": "2026-08-01T00:00:00.000000000Z",
             "to": "2026-08-02T00:00:00.000000000Z",
             "route": "cleanup-route",
+            "excludeRoute": "models",
             "model": "cleanup-model",
             "provider": "mesh",
             "engine": "skippy",

crates/mesh-llm-host-runtime/src/logging/runtime_state.rs

@@ -39,7 +39,7 @@ use super::{
     WebhookDeliveryScheduler, WebhookDeliveryWorker,
 };
 
-const HEALTH_AUDIT_ACTOR: &str = "logging-runtime";
+const HEALTH_AUDIT_ACTOR: &str = "logging_service";
 
 /// Internal capability state for local logging storage.
 ///

crates/mesh-llm-host-runtime/src/logging/runtime_state/tests/artifact_capture.rs

@@ -488,6 +488,15 @@ fn write_time_privacy_failure_publishes_one_marker_and_keeps_metadata_available(
     );
     assert!(state.health().metadata_available);
     assert_eq!(marker_audit_count(&store), 1);
+    let marker_actor: String = store
+        .conn()
+        .query_row(
+            "SELECT actor FROM audit_entries WHERE action = ?",
+            [ARTIFACT_CAPTURE_DISABLED_PRIVACY_UNAVAILABLE],
+            |row| row.get(0),
+        )
+        .expect("query marker audit actor");
+    assert_eq!(marker_actor, "logging_service");
 }
 
 #[test]

crates/mesh-llm-host-runtime/src/network/openai/response/routing.rs

@@ -231,6 +231,8 @@ mod tests {
     use tokio::io::ReadBuf;
     use tokio::net::TcpListener;
     use tokio::sync::Notify;
+    use tokio::sync::oneshot;
+    use tokio::time::{Duration, timeout};
 
     /// A real duplex pipe as the upstream half of `CancelUpstream`, wrapped to
     /// signal a `Notify` the moment a read finds nothing buffered yet.
@@ -325,9 +327,27 @@ mod tests {
 
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
         let address = listener.local_addr().unwrap();
+        let (disconnected_tx, disconnected_rx) = oneshot::channel();
         let task = tokio::spawn(async move {
-            let (mut client, _) = listener.accept().await.unwrap();
-            route_remote_attempt_after_forward(
+            let (client, _) = listener.accept().await.unwrap();
+            let client_std = client.into_std().unwrap();
+            let observer_std = client_std.try_clone().unwrap();
+            let mut client = TcpStream::from_std(client_std).unwrap();
+            let observer = TcpStream::from_std(observer_std).unwrap();
+            let observer_task = tokio::spawn(async move {
+                let mut peeked = [0; 1];
+                loop {
+                    if observer.readable().await.is_err() {
+                        break;
+                    }
+                    match observer.peek(&mut peeked).await {
+                        Ok(0) | Err(_) => break,
+                        Ok(_) => {}
+                    }
+                }
+                let _ = disconnected_tx.send(());
+            });
+            let result = route_remote_attempt_after_forward(
                 &mut client,
                 &mut upstream,
                 host_id,
@@ -336,7 +356,9 @@ mod tests {
                 ResponseAdapter::None,
                 OpenAiRouteObserver::default(),
             )
-            .await
+            .await;
+            observer_task.await.unwrap();
+            result
         });
         let client_socket = TcpStream::connect(address).await.unwrap();
 
@@ -355,13 +377,15 @@ mod tests {
         // it makes once the body arrives lands on an actually disconnected
         // socket, rather than an upstream read failure standing in for one.
         //
-        // A graceful close (a plain `drop`) sends only a FIN; the server's
-        // very next write can still succeed locally before it learns the
-        // peer is gone, which is exactly the kind of timing-dependent gap
-        // this test exists to close. Zero linger forces an RST instead, so
-        // the eventual write fails deterministically.
+        // A graceful close (a plain `drop`) sends only a FIN; wait until the
+        // accepted socket observes EOF/reset before releasing the body. This
+        // closes the propagation window without consuming route data.
         client_socket.set_zero_linger().unwrap();
         drop(client_socket);
+        timeout(Duration::from_secs(5), disconnected_rx)
+            .await
+            .unwrap()
+            .unwrap();
 
         upstream_writer.write_all(body.as_bytes()).await.unwrap();
 

crates/mesh-llm-log-store/src/maintenance/execution.rs

@@ -614,6 +614,10 @@ fn select_targets(
             parameters.push(rusqlite::types::Value::Text(value.to_owned()));
         }
     }
+    if let Some(exclude_route) = filters.exclude_route() {
+        sql.push_str(" AND (route IS NULL OR route != ?)");
+        parameters.push(rusqlite::types::Value::Text(exclude_route.to_owned()));
+    }
     if let Some(outcome) = filters.outcome() {
         sql.push_str(" AND state = ?");
         parameters.push(rusqlite::types::Value::Text(outcome.as_str().to_owned()));
@@ -960,6 +964,7 @@ pub(super) fn selection_fingerprint(
         scope.filters.from(),
         scope.filters.to(),
         scope.filters.route(),
+        scope.filters.exclude_route(),
         scope.filters.model(),
         scope.filters.provider(),
         scope.filters.engine(),

crates/mesh-llm-log-store/src/maintenance/scope_filters.rs

@@ -16,6 +16,7 @@ pub struct CleanupFilters {
     from: Option<String>,
     to: Option<String>,
     route: Option<String>,
+    exclude_route: Option<String>,
     model: Option<String>,
     provider: Option<String>,
     engine: Option<String>,
@@ -37,6 +38,7 @@ impl CleanupFilters {
             from: from.map(|value| value.0),
             to: to.map(|value| value.0),
             route: normalize_scope_filter(route, "route")?,
+            exclude_route: None,
             model: normalize_scope_filter(model, "model")?,
             provider: normalize_scope_filter(provider, "provider")?,
             engine: normalize_scope_filter(engine, "engine")?,
@@ -48,6 +50,14 @@ impl CleanupFilters {
         Ok(filters)
     }
 
+    pub fn with_exclude_route(
+        mut self,
+        exclude_route: Option<String>,
+    ) -> Result<Self, LogStoreError> {
+        self.exclude_route = normalize_scope_filter(exclude_route, "exclude_route")?;
+        Ok(self)
+    }
+
     pub fn from(&self) -> Option<&str> {
         self.from.as_deref()
     }
@@ -57,6 +67,9 @@ impl CleanupFilters {
     pub fn route(&self) -> Option<&str> {
         self.route.as_deref()
     }
+    pub fn exclude_route(&self) -> Option<&str> {
+        self.exclude_route.as_deref()
+    }
     pub fn model(&self) -> Option<&str> {
         self.model.as_deref()
     }