fix(pi): harden gateway failover and command isolation

src-tauri/Cargo.toml

@@ -99,7 +99,9 @@ windows-sys = { version = "0.61", features = [
     "Win32_Globalization",
     "Win32_Security",
     "Win32_Storage_FileSystem",
+    "Win32_System_JobObjects",
     "Win32_System_IO",
+    "Win32_System_Threading",
     "Win32_UI_Shell",
 ] }
 

src-tauri/src/pi_config/gateway.rs

@@ -80,6 +80,15 @@ fn configured_header_class(name: &HeaderName) -> ConfiguredHeaderClass {
     }
 }
 
+/// Whether an inbound client header must be replaced by candidate-local or
+/// gateway-owned transport state before forwarding to an upstream.
+///
+/// Keep this predicate beside configured-header classification so request
+/// filtering and provider validation cannot drift into two deny lists.
+pub(crate) fn gateway_replaces_incoming_header(name: &HeaderName) -> bool {
+    configured_header_class(name) != ConfiguredHeaderClass::CandidateLocal
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub(crate) enum PiGatewayApiFamily {
     AnthropicMessages,
@@ -332,6 +341,44 @@ impl CandidateHeaderPlan {
         Ok(candidate)
     }
 
+    /// Resolve only the fields which define failover protocol identity.
+    ///
+    /// This deliberately does not touch tenant/custom headers and does not
+    /// synthesize outbound authentication. The handler uses it before circuit
+    /// admission so a skipped primary can still constrain compatible
+    /// failovers without executing unrelated credential commands.
+    pub(crate) fn materialize_protocol_identity(
+        &self,
+        resolver: &impl DeferredValueResolver,
+    ) -> Result<Option<(PiGatewayApiFamily, HeaderMap)>, PiGatewayReason> {
+        if !self.protocol_identity_predictable {
+            return Ok(None);
+        }
+
+        let mut protocol_headers = HeaderMap::new();
+        if self.family == PiGatewayApiFamily::AnthropicMessages {
+            // OAuth changes the pinned Anthropic beta contract, so credential
+            // kind is the sole auth detail needed by protocol identity.
+            let credential = self.credential.materialize(resolver, "/apiKey")?;
+            let anthropic_oauth = credential.to_str().is_ok_and(is_anthropic_oauth_credential);
+            protocol_headers.insert(
+                HeaderName::from_static("anthropic-version"),
+                HeaderValue::from_static("2023-06-01"),
+            );
+            protocol_headers.insert(
+                HeaderName::from_static("anthropic-beta"),
+                HeaderValue::from_static(if anthropic_oauth {
+                    "claude-code-20250219,oauth-2025-04-20,interleaved-thinking-2025-05-14"
+                } else {
+                    "interleaved-thinking-2025-05-14"
+                }),
+            );
+        }
+        apply_protocol_headers(&self.provider_headers, resolver, &mut protocol_headers)?;
+        apply_protocol_headers(&self.model_headers, resolver, &mut protocol_headers)?;
+        Ok(Some((self.family, protocol_headers)))
+    }
+
     fn materialize_with_policy(
         &self,
         resolver: &impl DeferredValueResolver,
@@ -543,6 +590,23 @@ fn apply_planned_headers(
     Ok(())
 }
 
+fn apply_protocol_headers(
+    planned: &[PlannedHeader],
+    resolver: &impl DeferredValueResolver,
+    protocol_headers: &mut HeaderMap,
+) -> Result<(), PiGatewayReason> {
+    for entry in planned
+        .iter()
+        .filter(|entry| entry.class == ConfiguredHeaderClass::Protocol)
+    {
+        protocol_headers.insert(
+            entry.name.clone(),
+            entry.value.materialize(resolver, &entry.json_pointer)?,
+        );
+    }
+    Ok(())
+}
+
 impl MaterializedCandidate {
     pub(crate) fn family(&self) -> PiGatewayApiFamily {
         self.family

src-tauri/src/proxy/pi_handler.rs

@@ -12,6 +12,7 @@ use super::server::ProxyState;
 use super::usage::{InputTokenSemantics, TokenUsage, UsageLogger};
 use super::ProxyError;
 use crate::database::PRICING_SOURCE_REQUEST;
+use crate::pi_config::gateway::gateway_replaces_incoming_header;
 use axum::body::Body;
 use axum::extract::{Path, State};
 use axum::response::Response;
@@ -21,7 +22,7 @@ use http::header::{
     AUTHORIZATION, CONNECTION, CONTENT_LENGTH, HOST, PROXY_AUTHENTICATE, PROXY_AUTHORIZATION, TE,
     TRAILER, TRANSFER_ENCODING, UPGRADE,
 };
-use http::{HeaderMap, StatusCode};
+use http::{HeaderMap, HeaderName, StatusCode};
 use http_body_util::BodyExt;
 use serde_json::Value;
 use std::time::{Duration, Instant};
@@ -90,7 +91,7 @@ pub(crate) async fn handle_pi_native(
     let started = Instant::now();
     let session_id =
         crate::proxy::extract_session_id(&incoming_headers, &request_json, "pi").session_id;
-    let mut protocol_anchor = ProtocolAnchor::Unset;
+    let mut protocol_anchor = ProtocolAnchor::for_primary(attempts.first());
     let mut last_error = None;
     record_request_start(&state).await;
 
@@ -115,23 +116,29 @@ pub(crate) async fn handle_pi_native(
         let mut reached_network = false;
         for (offset, candidate) in attempts[index..provider_end].iter().cloned().enumerate() {
             let attempt_index = index + offset;
+            let Some(single_direct_attempt) =
+                begin_protocol_materialization(&mut protocol_anchor, candidate.is_failover)
+            else {
+                continue;
+            };
             let materialized = match materialize_candidate(candidate, path_and_query.clone()).await
             {
                 Ok(candidate) => candidate,
                 Err(error) => {
                     last_error = Some(error.to_string());
-                    continue;
+                    // Deferred materialization is provider-level state, not an
+                    // endpoint health result. Do not execute the same command
+                    // again for every endpoint; move to a compatible provider.
+                    break;
                 }
             };
             let protocol_identity = materialized
                 .transport
                 .failover_protocol_identity()
                 .map(|(family, headers)| (family.as_str().to_string(), headers.clone()));
-            if !protocol_identity_allows_attempt(
-                &mut protocol_anchor,
-                protocol_identity,
-                materialized.is_failover,
-            ) {
+            if !single_direct_attempt
+                && !protocol_identity_allows_attempt(&mut protocol_anchor, protocol_identity)
+            {
                 continue;
             }
 
@@ -261,33 +268,67 @@ pub(crate) async fn handle_pi_native(
     Err(ProxyError::ForwardFailed(error))
 }
 
+/// Return whether this attempt consumes the one direct-only grant, or `None`
+/// when the candidate must not even be materialized.
+fn begin_protocol_materialization(anchor: &mut ProtocolAnchor, is_failover: bool) -> Option<bool> {
+    match anchor {
+        ProtocolAnchor::DirectOnlyPending if !is_failover => {
+            // Consume before materialization so an error in a later
+            // credential/custom header cannot run the protocol command again.
+            *anchor = ProtocolAnchor::Ineligible;
+            Some(true)
+        }
+        ProtocolAnchor::FailoverPending if is_failover => Some(false),
+        ProtocolAnchor::DirectOnlyPending
+        | ProtocolAnchor::FailoverPending
+        | ProtocolAnchor::Ineligible => None,
+        ProtocolAnchor::Unset | ProtocolAnchor::Predictable(_) => Some(false),
+    }
+}
+
 #[derive(Debug)]
 enum ProtocolAnchor {
     Unset,
+    FailoverPending,
     Predictable((String, HeaderMap)),
+    DirectOnlyPending,
     Ineligible,
 }
 
+impl ProtocolAnchor {
+    fn for_primary(primary: Option<&PiRequestCandidate>) -> Self {
+        let Some(primary) = primary else {
+            return Self::Unset;
+        };
+        if !primary.protocol_identity_is_predictable() {
+            return Self::DirectOnlyPending;
+        }
+        match primary.planned_protocol_identity() {
+            Ok(Some(identity)) => Self::Predictable(identity),
+            Ok(None) => Self::DirectOnlyPending,
+            // A provider-level deferred credential/header may be unavailable
+            // while a compatible backup is healthy. Skip every endpoint of
+            // that provider and let the first materialized failover establish
+            // the request's protocol anchor.
+            Err(_) => Self::FailoverPending,
+        }
+    }
+}
+
 fn protocol_identity_allows_attempt(
     anchor: &mut ProtocolAnchor,
     candidate: Option<(String, HeaderMap)>,
-    is_failover: bool,
 ) -> bool {
     match (&*anchor, candidate) {
-        (ProtocolAnchor::Unset, _) if is_failover => false,
-        (ProtocolAnchor::Unset, Some(candidate)) => {
+        (ProtocolAnchor::Unset | ProtocolAnchor::FailoverPending, Some(candidate)) => {
             *anchor = ProtocolAnchor::Predictable(candidate);
             true
         }
-        (ProtocolAnchor::Unset, None) => {
-            // A protocol !command may be used for the first direct attempt,
-            // but its result cannot safely authorize endpoint or provider
-            // replay because the command is evaluated again per attempt.
-            *anchor = ProtocolAnchor::Ineligible;
-            true
-        }
+        (ProtocolAnchor::Unset | ProtocolAnchor::FailoverPending, None) => false,
         (ProtocolAnchor::Predictable(primary), Some(candidate)) => primary == &candidate,
-        (ProtocolAnchor::Predictable(_), None) | (ProtocolAnchor::Ineligible, _) => false,
+        (ProtocolAnchor::Predictable(_), None)
+        | (ProtocolAnchor::DirectOnlyPending, _)
+        | (ProtocolAnchor::Ineligible, _) => false,
     }
 }
 
@@ -476,9 +517,9 @@ fn request_is_streaming(uri: &http::Uri, headers: &HeaderMap, body: &Value) -> b
 }
 
 fn filtered_incoming_headers(headers: &HeaderMap) -> HeaderMap {
+    let connection_named = connection_named_headers(headers);
     let mut filtered = HeaderMap::new();
     for (name, value) in headers {
-        let lower = name.as_str();
         if matches!(
             *name,
             HOST | CONTENT_LENGTH
@@ -490,21 +531,8 @@ fn filtered_incoming_headers(headers: &HeaderMap) -> HeaderMap {
                 | AUTHORIZATION
                 | PROXY_AUTHENTICATE
                 | PROXY_AUTHORIZATION
-        ) || matches!(
-            lower,
-            "x-api-key"
-                | "x-goog-api-key"
-                | "anthropic-version"
-                | "anthropic-beta"
-                | "openai-beta"
-                | "openai-version"
-                | "forwarded"
-                | "x-forwarded-for"
-                | "x-forwarded-host"
-                | "x-forwarded-port"
-                | "x-forwarded-proto"
-                | "x-real-ip"
-        ) || lower.starts_with("proxy-")
+        ) || gateway_replaces_incoming_header(name)
+            || connection_named.contains(name)
         {
             continue;
         }
@@ -878,6 +906,7 @@ async fn prepare_response(
 }
 
 fn filtered_response_headers(headers: &HeaderMap) -> HeaderMap {
+    let connection_named = connection_named_headers(headers);
     let mut filtered = HeaderMap::new();
     for (name, value) in headers {
         if matches!(
@@ -890,7 +919,8 @@ fn filtered_response_headers(headers: &HeaderMap) -> HeaderMap {
                 | UPGRADE
                 | PROXY_AUTHENTICATE
                 | PROXY_AUTHORIZATION
-        ) {
+        ) || connection_named.contains(name)
+        {
             continue;
         }
         filtered.append(name.clone(), value.clone());
@@ -898,6 +928,16 @@ fn filtered_response_headers(headers: &HeaderMap) -> HeaderMap {
     filtered
 }
 
+fn connection_named_headers(headers: &HeaderMap) -> std::collections::HashSet<HeaderName> {
+    headers
+        .get_all(CONNECTION)
+        .iter()
+        .filter_map(|value| value.to_str().ok())
+        .flat_map(|value| value.split(','))
+        .filter_map(|name| HeaderName::from_bytes(name.trim().as_bytes()).ok())
+        .collect()
+}
+
 async fn preflight_sse(
     stream: &mut BoxStream<'static, Result<Bytes, reqwest::Error>>,
     timeout_seconds: u32,
@@ -1230,8 +1270,8 @@ mod tests {
     }
 
     #[test]
-    fn failover_cannot_become_the_protocol_anchor_after_primary_materialization_fails() {
-        let mut anchor = ProtocolAnchor::Unset;
+    fn first_compatible_failover_can_anchor_when_primary_identity_is_unavailable() {
+        let mut anchor = ProtocolAnchor::FailoverPending;
         let mut candidate_headers = HeaderMap::new();
         candidate_headers.insert(
             "anthropic-version",
@@ -1239,54 +1279,84 @@ mod tests {
         );
         let candidate = Some(("anthropic-messages".to_string(), candidate_headers));
 
-        assert!(!protocol_identity_allows_attempt(
-            &mut anchor,
-            candidate,
-            true
-        ));
-        assert!(matches!(anchor, ProtocolAnchor::Unset));
+        assert_eq!(begin_protocol_materialization(&mut anchor, false), None);
+        assert_eq!(
+            begin_protocol_materialization(&mut anchor, true),
+            Some(false)
+        );
+        assert!(protocol_identity_allows_attempt(&mut anchor, candidate));
+        assert!(matches!(anchor, ProtocolAnchor::Predictable(_)));
     }
 
     #[test]
     fn protocol_anchor_blocks_replay_when_identity_is_unpredictable_or_changes() {
-        let mut unpredictable = ProtocolAnchor::Unset;
-        assert!(protocol_identity_allows_attempt(
-            &mut unpredictable,
-            None,
-            false
-        ));
-        assert!(!protocol_identity_allows_attempt(
-            &mut unpredictable,
-            None,
-            false
-        ));
-        assert!(!protocol_identity_allows_attempt(
-            &mut unpredictable,
-            None,
-            true
-        ));
+        let mut unpredictable = ProtocolAnchor::DirectOnlyPending;
+        assert_eq!(
+            begin_protocol_materialization(&mut unpredictable, false),
+            Some(true)
+        );
+        assert_eq!(
+            begin_protocol_materialization(&mut unpredictable, false),
+            None
+        );
+        assert_eq!(
+            begin_protocol_materialization(&mut unpredictable, true),
+            None
+        );
 
         let identity = Some(("openai-responses".to_string(), HeaderMap::new()));
         let mut predictable = ProtocolAnchor::Unset;
         assert!(protocol_identity_allows_attempt(
             &mut predictable,
-            identity.clone(),
-            false
-        ));
-        assert!(protocol_identity_allows_attempt(
-            &mut predictable,
-            identity,
-            false
+            identity.clone()
         ));
+        assert!(protocol_identity_allows_attempt(&mut predictable, identity));
         let mut changed_headers = HeaderMap::new();
         changed_headers.insert("openai-version", http::HeaderValue::from_static("changed"));
         assert!(!protocol_identity_allows_attempt(
             &mut predictable,
-            Some(("openai-responses".to_string(), changed_headers)),
-            false
+            Some(("openai-responses".to_string(), changed_headers))
         ));
     }
 
+    #[test]
+    fn gateway_header_filters_share_protected_and_dynamic_hop_by_hop_rules() {
+        let mut incoming = HeaderMap::new();
+        incoming.insert(
+            CONNECTION,
+            http::HeaderValue::from_static("x-private-hop, x-another-hop"),
+        );
+        incoming.insert(
+            "x-private-hop",
+            http::HeaderValue::from_static("must-not-forward"),
+        );
+        incoming.insert(
+            "x-another-hop",
+            http::HeaderValue::from_static("must-not-forward"),
+        );
+        incoming.insert(
+            "cf-connecting-ip",
+            http::HeaderValue::from_static("203.0.113.5"),
+        );
+        incoming.insert("traceparent", http::HeaderValue::from_static("00-spoofed"));
+        incoming.insert(
+            "x-candidate-local",
+            http::HeaderValue::from_static("preserved"),
+        );
+
+        let request = filtered_incoming_headers(&incoming);
+        assert!(request.get("x-private-hop").is_none());
+        assert!(request.get("x-another-hop").is_none());
+        assert!(request.get("cf-connecting-ip").is_none());
+        assert!(request.get("traceparent").is_none());
+        assert_eq!(request["x-candidate-local"], "preserved");
+
+        let response = filtered_response_headers(&incoming);
+        assert!(response.get("x-private-hop").is_none());
+        assert!(response.get("x-another-hop").is_none());
+        assert_eq!(response["x-candidate-local"], "preserved");
+    }
+
     #[test]
     fn streaming_health_waits_for_the_terminal_outcome() {
         let complete = pi_stream_disposition(

src-tauri/src/proxy/pi_runtime.rs

@@ -25,7 +25,7 @@ use std::io::Read;
 use std::process::{Command, Stdio};
 use std::sync::{
     atomic::{AtomicU64, Ordering},
-    Arc, RwLock,
+    mpsc, Arc, RwLock,
 };
 use std::time::{Duration, Instant};
 use tokio::sync::{OwnedRwLockReadGuard, RwLock as AsyncRwLock};
@@ -218,6 +218,40 @@ impl PiRequestCandidate {
             url,
         })
     }
+
+    pub(crate) fn protocol_identity_is_predictable(&self) -> bool {
+        self.plan.protocol_identity_is_predictable()
+    }
+
+    pub(crate) fn planned_protocol_identity(
+        &self,
+    ) -> Result<Option<(String, http::HeaderMap)>, AppError> {
+        let resolver_failure = std::cell::Cell::new(false);
+        let identity = self
+            .plan
+            .materialize_protocol_identity(&|expression: &str| {
+                // Protocol !commands are marked unpredictable before this
+                // method is called. An auth command must not be run merely to
+                // decide whether a circuit-skipped primary permits failover.
+                if expression.starts_with('!') {
+                    resolver_failure.set(true);
+                    return None;
+                }
+                let resolved = resolve_pi_config_value(expression);
+                resolver_failure.set(resolver_failure.get() || resolved.is_err());
+                resolved.ok()
+            })
+            .map_err(|reason| {
+                if resolver_failure.get() {
+                    AppError::Config(
+                        "failed to pre-resolve Pi primary protocol identity".to_string(),
+                    )
+                } else {
+                    gateway_reason(reason)
+                }
+            })?;
+        Ok(identity.map(|(family, headers)| (family.as_str().to_string(), headers)))
+    }
 }
 
 fn gateway_reason(reason: PiGatewayReason) -> AppError {
@@ -561,6 +595,22 @@ impl PiRuntimeStore {
         odd.saturating_add(1)
     }
 
+    pub(crate) fn next_even_epoch(&self) -> Result<u64, AppError> {
+        let current = self.catalog_epoch.load(Ordering::Acquire);
+        if current % 2 != 0 {
+            return Err(AppError::Conflict(
+                "cannot publish a sorted Pi runtime while catalog admission is fenced".to_string(),
+            ));
+        }
+        let next = current.saturating_add(2);
+        if next % 2 != 0 {
+            return Err(AppError::Config(
+                "Pi catalog epoch overflowed its even publication sequence".to_string(),
+            ));
+        }
+        Ok(next)
+    }
+
     pub(crate) async fn publish(&self, snapshot: Arc<PiRuntimeSnapshot>) -> Result<(), AppError> {
         if snapshot.catalog_epoch % 2 != 0 {
             return Err(AppError::Config(
@@ -626,6 +676,10 @@ impl PiRuntimeStore {
         (self.catalog_epoch.load(Ordering::Acquire) == epoch).then_some(snapshot)
     }
 
+    pub(crate) fn is_admitting(&self, server_generation: u64) -> bool {
+        self.lease(server_generation).is_some()
+    }
+
     pub(crate) async fn admission_guard(
         self: &Arc<Self>,
         server_generation: u64,
@@ -787,6 +841,7 @@ fn execute_config_command(script: &str) -> Result<String, String> {
         .stderr(Stdio::piped())
         .spawn()
         .map_err(|error| format!("failed to start Pi config command: {error}"))?;
+    let command_tree = CommandTree::attach(&mut child)?;
     let stdout = child
         .stdout
         .take()
@@ -795,8 +850,14 @@ fn execute_config_command(script: &str) -> Result<String, String> {
         .stderr
         .take()
         .ok_or_else(|| "failed to capture Pi config command stderr".to_string())?;
-    let stdout_reader = std::thread::spawn(move || read_bounded(stdout));
-    let stderr_reader = std::thread::spawn(move || read_bounded(stderr));
+    let (stdout_sender, stdout_reader) = mpsc::sync_channel(1);
+    let (stderr_sender, stderr_reader) = mpsc::sync_channel(1);
+    std::thread::spawn(move || {
+        let _ = stdout_sender.send(read_bounded(stdout));
+    });
+    std::thread::spawn(move || {
+        let _ = stderr_sender.send(read_bounded(stderr));
+    });
     let started = Instant::now();
     let status = loop {
         match child.try_wait() {
@@ -805,27 +866,32 @@ fn execute_config_command(script: &str) -> Result<String, String> {
                 std::thread::sleep(Duration::from_millis(10));
             }
             Ok(None) => {
-                terminate_command_tree(&mut child);
+                command_tree.terminate(&mut child);
                 let _ = child.wait();
-                let _ = stdout_reader.join();
-                let _ = stderr_reader.join();
                 return Err("Pi config command timed out".to_string());
             }
             Err(error) => {
-                terminate_command_tree(&mut child);
+                command_tree.terminate(&mut child);
                 let _ = child.wait();
-                let _ = stdout_reader.join();
-                let _ = stderr_reader.join();
                 return Err(format!("failed to wait for Pi config command: {error}"));
             }
         }
     };
-    let stdout = stdout_reader
-        .join()
-        .map_err(|_| "Pi config command stdout reader panicked".to_string())??;
-    let stderr = stderr_reader
-        .join()
-        .map_err(|_| "Pi config command stderr reader panicked".to_string())??;
+    // A successful shell may leave descendants holding inherited pipe handles.
+    // Terminate the whole tree before draining, and keep the original deadline
+    // over both process wait and output collection.
+    command_tree.terminate(&mut child);
+    let deadline = started + COMMAND_TIMEOUT;
+    let stdout = receive_command_output(
+        &stdout_reader,
+        deadline,
+        "Pi config command stdout did not close before timeout",
+    )??;
+    let stderr = receive_command_output(
+        &stderr_reader,
+        deadline,
+        "Pi config command stderr did not close before timeout",
+    )??;
     if !status.success() {
         return Err(format!(
             "Pi config command exited unsuccessfully: {}",
@@ -849,14 +915,112 @@ fn read_bounded(reader: impl Read) -> Result<Vec<u8>, String> {
     Ok(output)
 }
 
-fn terminate_command_tree(child: &mut std::process::Child) {
-    #[cfg(unix)]
-    unsafe {
-        let _ = libc::kill(-(child.id() as i32), libc::SIGKILL);
+fn receive_command_output(
+    receiver: &mpsc::Receiver<Result<Vec<u8>, String>>,
+    deadline: Instant,
+    timeout_message: &str,
+) -> Result<Result<Vec<u8>, String>, String> {
+    let remaining = deadline.saturating_duration_since(Instant::now());
+    receiver
+        .recv_timeout(remaining)
+        .map_err(|error| match error {
+            mpsc::RecvTimeoutError::Timeout => timeout_message.to_string(),
+            mpsc::RecvTimeoutError::Disconnected => {
+                "Pi config command output reader stopped unexpectedly".to_string()
+            }
+        })
+}
+
+struct CommandTree {
+    #[cfg(windows)]
+    job: windows_sys::Win32::Foundation::HANDLE,
+}
+
+impl CommandTree {
+    fn attach(child: &mut std::process::Child) -> Result<Self, String> {
+        #[cfg(windows)]
+        {
+            use std::mem::size_of;
+            use std::os::windows::io::AsRawHandle;
+            use windows_sys::Win32::Foundation::CloseHandle;
+            use windows_sys::Win32::System::JobObjects::{
+                AssignProcessToJobObject, CreateJobObjectW, JobObjectExtendedLimitInformation,
+                SetInformationJobObject, JOBOBJECT_EXTENDED_LIMIT_INFORMATION,
+                JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE,
+            };
+
+            // SAFETY: all pointers are either null or point to initialized
+            // values for the duration of their synchronous Win32 calls.
+            unsafe {
+                let job = CreateJobObjectW(std::ptr::null(), std::ptr::null());
+                if job.is_null() {
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    return Err(format!(
+                        "failed to create Pi config command job: {}",
+                        std::io::Error::last_os_error()
+                    ));
+                }
+                let mut limits = JOBOBJECT_EXTENDED_LIMIT_INFORMATION::default();
+                limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
+                if SetInformationJobObject(
+                    job,
+                    JobObjectExtendedLimitInformation,
+                    (&raw const limits).cast(),
+                    size_of::<JOBOBJECT_EXTENDED_LIMIT_INFORMATION>() as u32,
+                ) == 0
+                {
+                    let error = std::io::Error::last_os_error();
+                    CloseHandle(job);
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    return Err(format!(
+                        "failed to configure Pi config command job: {error}"
+                    ));
+                }
+                if AssignProcessToJobObject(job, child.as_raw_handle() as _) == 0 {
+                    let error = std::io::Error::last_os_error();
+                    CloseHandle(job);
+                    let _ = child.kill();
+                    let _ = child.wait();
+                    return Err(format!(
+                        "failed to assign Pi config command to its job: {error}"
+                    ));
+                }
+                return Ok(Self { job });
+            }
+        }
+        #[cfg(not(windows))]
+        {
+            let _ = child;
+            Ok(Self {})
+        }
     }
-    #[cfg(not(unix))]
-    {
-        let _ = child.kill();
+
+    fn terminate(&self, child: &mut std::process::Child) {
+        #[cfg(unix)]
+        unsafe {
+            let _ = libc::kill(-(child.id() as i32), libc::SIGKILL);
+        }
+        #[cfg(windows)]
+        unsafe {
+            let _ = windows_sys::Win32::System::JobObjects::TerminateJobObject(self.job, 1);
+        }
+        #[cfg(not(any(unix, windows)))]
+        {
+            let _ = child.kill();
+        }
+    }
+}
+
+#[cfg(windows)]
+impl Drop for CommandTree {
+    fn drop(&mut self) {
+        // JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE is the final safety net if an
+        // early return occurs before explicit termination.
+        unsafe {
+            let _ = windows_sys::Win32::Foundation::CloseHandle(self.job);
+        }
     }
 }
 
@@ -879,6 +1043,19 @@ mod tests {
         std::env::remove_var("PI_RUNTIME_TEST_VALUE");
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn command_deadline_covers_descendants_holding_output_pipes() {
+        let started = Instant::now();
+        let output =
+            execute_config_command("sleep 30 & printf inherited-pipe").expect("command output");
+        assert_eq!(output, "inherited-pipe");
+        assert!(
+            started.elapsed() < Duration::from_secs(2),
+            "background descendants must be terminated before output drain"
+        );
+    }
+
     #[test]
     fn command_resolution_matches_vendored_transport_oracle() {
         #[cfg(unix)]

src-tauri/src/services/usage_stats.rs

@@ -137,8 +137,9 @@ pub struct RequestLogDetail {
     pub output_tokens: u32,
     pub cache_read_tokens: u32,
     pub cache_creation_tokens: u32,
-    /// Internal storage semantics; omitted from the UI/API payload.
-    #[serde(skip)]
+    /// Persisted request-level semantics used by both pricing and UI cache
+    /// normalization. This must cross IPC; app-type inference is only a legacy
+    /// fallback for rows written before the semantics column existed.
     pub input_token_semantics: i64,
     pub input_cost_usd: String,
     pub output_cost_usd: String,
@@ -1653,10 +1654,10 @@ impl Database {
         let detail_sql = format!(
             "SELECT l.request_id, l.provider_id, {detail_pname} as provider_name, l.app_type, l.model,
                     l.request_model, l.cost_multiplier,
-                    input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
-                    input_cost_usd, output_cost_usd, cache_read_cost_usd, cache_creation_cost_usd, total_cost_usd,
-                    is_streaming, latency_ms, first_token_ms, duration_ms,
-                    status_code, error_message, created_at, l.data_source, l.pricing_model,
+                    l.input_tokens, l.output_tokens, l.cache_read_tokens, l.cache_creation_tokens,
+                    l.input_cost_usd, l.output_cost_usd, l.cache_read_cost_usd, l.cache_creation_cost_usd, l.total_cost_usd,
+                    l.is_streaming, l.latency_ms, l.first_token_ms, l.duration_ms,
+                    l.status_code, l.error_message, l.created_at, l.data_source, l.pricing_model,
                     l.input_token_semantics
              FROM proxy_request_logs l
              LEFT JOIN providers p ON l.provider_id = p.id AND l.app_type = p.app_type
@@ -2406,6 +2407,54 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn paginated_and_detail_ipc_serialize_persisted_input_semantics() -> Result<(), AppError> {
+        let db = Database::memory()?;
+        {
+            let conn = lock_conn!(db.conn);
+            insert_usage_log(
+                &conn,
+                "pi-semantics-ipc",
+                "pi",
+                "pi-provider",
+                "gpt-test",
+                "request",
+                1,
+                1_000,
+                5,
+                800,
+                0,
+                200,
+                "0",
+            )?;
+            conn.execute(
+                "UPDATE proxy_request_logs
+                    SET input_token_semantics = ?1
+                  WHERE request_id = 'pi-semantics-ipc'",
+                [INPUT_TOKEN_SEMANTICS_TOTAL],
+            )?;
+        }
+
+        let page = db.get_request_logs(&LogFilters::default(), 0, 10)?;
+        let page_json =
+            serde_json::to_value(&page).map_err(|error| AppError::Database(error.to_string()))?;
+        assert_eq!(
+            page_json["data"][0]["inputTokenSemantics"],
+            INPUT_TOKEN_SEMANTICS_TOTAL
+        );
+
+        let detail = db
+            .get_request_detail("pi-semantics-ipc")?
+            .expect("request detail");
+        let detail_json =
+            serde_json::to_value(detail).map_err(|error| AppError::Database(error.to_string()))?;
+        assert_eq!(
+            detail_json["inputTokenSemantics"],
+            INPUT_TOKEN_SEMANTICS_TOTAL
+        );
+        Ok(())
+    }
+
     fn create_legacy_nullable_logs_table(conn: &Connection) -> Result<(), AppError> {
         conn.execute(
             "CREATE TABLE proxy_request_logs (