zccache 1.12.16

//! Per-client IPC connection dispatch loop.

use super::*;
use crate::daemon_core::protocol::{
    wire_prost::{self, zccache_v1 as pb},
    DecodedWireMessage,
};

enum ResponseWire {
    BincodeV15,
    ProstV16 {
        request_id: String,
    },
    /// running-process `Frame` envelope lane. `frame_request_id` is the
    /// frame correlation id to echo; `request_id` is the inner zccache
    /// prost request id echoed in the response body.
    FrameV1 {
        frame_request_id: u64,
        request_id: String,
    },
}

const SERVER_REQUEST_RECV_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(600);

pub(super) fn session_phase_profile(
    state: &SharedState,
    session_id: &SessionId,
) -> crate::daemon_core::protocol::PhaseProfileSummary {
    let mut totals = state.profiler.totals_snapshot();
    totals.staged = state.session_staged_profiles.get(session_id).map_or_else(
        || crate::daemon_core::daemon::staged_stats::StagedProfiler::new().snapshot(),
        |profile| profile.snapshot(),
    );
    totals.into()
}

/// Run a child-spawning handler (compile / link / exec) while concurrently
/// watching the client connection for disconnect (issue #967, meta #968).
///
/// Returns `Some((response, journal_ctx))` when the handler finished first, or
/// `None` when the client disconnected while the handler was still running — in
/// which case the handler future has already been dropped, which drops the
/// daemon-owned compiler [`tokio::process::Child`] (spawned `kill_on_drop(true)`)
/// and reaps the subprocess and its compile-concurrency permit.
///
/// Before this guard existed, the daemon parked inside the compile/link await
/// and never read the socket again, so a client that gave up (its 600 s recv
/// timeout, a `taskkill`, a cancelled build) left the daemon awaiting a compile
/// whose result could never be delivered — holding a compile-concurrency permit
/// the whole time. Enough of those wedged the shared semaphore and every later
/// compile queued forever (issue #962's amplifier; issue #967 is this fix).
///
/// The race is `biased` toward the handler so a compile that finishes in the
/// same poll as an incoming disconnect still returns its response. On disconnect
/// the handler future is dropped at its next suspension point — the
/// `child.wait_with_output().await` inside [`crate::daemon_core::daemon::process`].
///
/// (Returns `Option` rather than a two-variant enum so the 448-byte completed
/// payload does not sit next to a zero-size cancelled variant — `clippy::large_enum_variant`.)
async fn guarded_dispatch<F>(
    conn: &mut IpcConnection,
    handler: F,
) -> Option<(Response, Option<JournalContext>)>
where
    F: std::future::Future<Output = (Response, Option<JournalContext>)>,
{
    tokio::select! {
        biased;
        out = handler => Some(out),
        () = conn.wait_for_disconnect() => None,
    }
}

/// Emit the distinguishable client-cancellation diagnostic (issue #967
/// acceptance): a loud `warn!` plus an on-disk lifecycle record so an operator
/// can tell "client disconnected mid-compile" apart from a compiler failure, a
/// daemon crash, or an IPC write error — even after a detached Windows run where
/// daemon stderr is redirected.
///
/// This is deliberately `warn!` (not `info!`): a client vanishing mid-compile is
/// abnormal — it means daemon work was thrown away and may indicate the client
/// itself wedged or crashed — so it must complain loudly and leave a forensic
/// trail, per the daemon's "every cancellation/timeout is logged loud + durable"
/// rule.
fn log_client_cancelled(kind: &str) {
    tracing::warn!(
        event = "client-cancelled",
        kind,
        "client disconnected before the response was produced; \
         daemon-owned compiler child reaped via kill_on_drop and \
         compile-concurrency permit released"
    );
    super::super::lifecycle::write_event(
        "client_cancelled",
        serde_json::json!({
            "kind": kind,
            "reason": "client disconnected before response; daemon child reaped, permit released",
        }),
    );
}

/// Handle a single client connection.
pub(super) async fn handle_connection(
    mut conn: IpcConnection,
    state: Arc<SharedState>,
) -> Result<(), crate::daemon_core::ipc::IpcError> {
    if conn
        .try_serve_backend_handle_probe(&state.backend_identity)
        .await?
    {
        state.last_activity.store(now_secs(), Ordering::Relaxed);
        return Ok(());
    }

    loop {
        let request = match conn
            .recv_wire_with_timeout::<Request, pb::Request>(SERVER_REQUEST_RECV_TIMEOUT)
            .await
        {
            Ok(req) => req,
            Err(crate::daemon_core::ipc::IpcError::Timeout(timeout)) => {
                tracing::warn!(
                    timeout_secs = timeout.as_secs(),
                    "client connection timed out waiting for next request; closing connection"
                );
                return Ok(());
            }
            Err(crate::daemon_core::ipc::IpcError::Protocol(
                crate::daemon_core::protocol::ProtocolError::VersionMismatch { expected, received },
            )) => {
                // Don't drop the connection silently — without a reply the
                // CLI surfaces the (correct) closure as the misleading
                // "lost connection to daemon (no response received)". Send
                // back a real error so the CLI can render the actual
                // reason — both crate versions and both protocol versions,
                // daemon and client.
                //
                // The response goes out at the daemon's PROTOCOL_VERSION,
                // which itself will fail to decode on a different-versioned
                // client — but VersionMismatch on the client side renders
                // a clear message via Display ("expected vX, received vY"),
                // which is what we want.
                let daemon_crate = env!("CARGO_PKG_VERSION");
                let msg = format!(
                    "protocol version mismatch: daemon zccache v{daemon_crate} \
                     (protocol v{expected}) received a request at protocol v{received}. \
                     Run `zccache stop` and retry — the CLI you connected with is built \
                     against a different PROTOCOL_VERSION than this daemon."
                );
                tracing::warn!("{msg}");
                // Also persist to the on-disk lifecycle log so the reason
                // is visible even when daemon stderr is redirected/null.
                // The lifecycle log already records the daemon's
                // CARGO_PKG_VERSION on every spawn — this entry adds the
                // mismatch context.
                super::super::lifecycle::write_event(
                    "version_mismatch",
                    serde_json::json!({
                        "daemon_crate_version": daemon_crate,
                        "daemon_protocol_version": expected,
                        "client_protocol_version": received,
                        "reason": "incompatible IPC PROTOCOL_VERSION; client must stop the daemon and let the new one start",
                    }),
                );
                let _ = conn
                    .send(&Response::Error {
                        message: msg.clone(),
                    })
                    .await;
                return Err(crate::daemon_core::ipc::IpcError::Protocol(
                    crate::daemon_core::protocol::ProtocolError::VersionMismatch { expected, received },
                ));
            }
            Err(e) => return Err(e),
        };
        let Some(request) = request else {
            tracing::debug!("client disconnected");
            return Ok(());
        };
        state.last_activity.store(now_secs(), Ordering::Relaxed);

        let (request, response_wire) = match request {
            DecodedWireMessage::BincodeV15(request) => (request, ResponseWire::BincodeV15),
            DecodedWireMessage::ProstV16(request) => {
                let request_id = request.request_id.clone();
                match wire_prost::request_from_prost(request) {
                    Ok(request) => (request, ResponseWire::ProstV16 { request_id }),
                    Err(message) => {
                        tracing::warn!("{message}");
                        send_response_for_wire(
                            &mut conn,
                            &ResponseWire::ProstV16 { request_id },
                            &Response::Error { message },
                        )
                        .await?;
                        continue;
                    }
                }
            }
            DecodedWireMessage::FrameV1 {
                message: request,
                request_id: frame_request_id,
            } => {
                let request_id = request.request_id.clone();
                match wire_prost::request_from_prost(request) {
                    Ok(request) => (
                        request,
                        ResponseWire::FrameV1 {
                            frame_request_id,
                            request_id,
                        },
                    ),
                    Err(message) => {
                        tracing::warn!("{message}");
                        send_response_for_wire(
                            &mut conn,
                            &ResponseWire::FrameV1 {
                                frame_request_id,
                                request_id,
                            },
                            &Response::Error { message },
                        )
                        .await?;
                        continue;
                    }
                }
            }
        };

        match &request {
            Request::SessionStart {
                private_daemon: Some(options),
                ..
            } => {
                let private_env_keys: Vec<&str> =
                    options.env.iter().map(|(key, _)| key.as_str()).collect();
                tracing::debug!(
                    private_env_keys = ?private_env_keys,
                    owner_pids = ?options.owner_pids,
                    daemon_name = ?options.daemon_name,
                    endpoint = ?options.endpoint,
                    "received private session-start request"
                );
            }
            _ => tracing::debug!(?request, "received request"),
        }

        // Dispatch request and capture journal metadata in the same match
        // to move args/session_id into JournalContext without cloning.
        // Only env needs cloning because handlers consume it.
        let journal_start = std::time::Instant::now();
        let (response, journal_ctx): (Response, Option<JournalContext>) = match request {
            Request::Ping => (Response::Pong, None),
            Request::Shutdown => {
                send_response_for_wire(&mut conn, &response_wire, &Response::ShuttingDown).await?;
                // Record graceful exit alongside the existing "spawn"
                // event so a single parse of `daemon-lifecycle.log`
                // reconstructs the daemon's full lifetime. Pairs with
                // EVENT_DIED_IDLE for unattended exits and the CLI's
                // EVENT_SPAWN_ATTEMPT for the matching start side.
                //
                // Under burst load (issue #726) many wedge-detecting clients
                // race to send Shutdown within a few ms; gate the write with
                // a CAS so only the first writes — without this, we observed
                // 25+ duplicate rows for a single death in production logs.
                if !state
                    .shutdown_event_logged
                    .swap(true, std::sync::atomic::Ordering::AcqRel)
                {
                    super::super::lifecycle::write_event(
                        super::super::lifecycle::EVENT_DIED_SHUTDOWN,
                        serde_json::json!({
                            "reason": super::super::lifecycle::REASON_GRACEFUL_SHUTDOWN,
                            "uptime_secs": now_secs().saturating_sub(state.start_time),
                        }),
                    );
                }
                state.shutdown_requested.store(true, Ordering::Release);
                state.shutdown.notify_waiters();
                return Ok(());
            }
            Request::Status => {
                let snap = state.stats.snapshot();
                let dg = state.dep_graph.load().stats();
                let artifact_count = state.artifacts.len() as u64;
                let cache_size_bytes: u64 = state
                    .artifacts
                    .iter()
                    .map(|entry| entry.value().meta.total_size)
                    .sum();
                let metadata_entries = state.cache_system.metadata().len() as u64;
                let private_daemon = state.private_daemon.snapshot().await;
                (
                    Response::Status(crate::daemon_core::protocol::DaemonStatus {
                        version: crate::daemon_core::core::VERSION.to_string(),
                        daemon_namespace: state.daemon_namespace.clone(),
                        endpoint: state.endpoint.clone(),
                        private_daemon,
                        artifact_count,
                        cache_size_bytes,
                        metadata_entries,
                        uptime_secs: now_secs().saturating_sub(state.start_time),
                        cache_hits: snap.hits,
                        cache_misses: snap.misses,
                        total_compilations: snap.compilations,
                        non_cacheable: snap.non_cacheable,
                        compile_errors: snap.compile_errors,
                        compile_errors_cached: snap.compile_errors_cached,
                        time_saved_ms: snap.time_saved_ms(),
                        total_links: snap.link_total,
                        link_hits: snap.link_hits,
                        link_misses: snap.link_misses,
                        link_non_cacheable: snap.link_non_cacheable,
                        dep_graph_contexts: dg.context_count as u64,
                        dep_graph_files: dg.file_count as u64,
                        sessions_total: snap.sessions_total,
                        sessions_active: state.sessions.active_count() as u64,
                        cache_dir: state.cache_dir.clone(),
                        dep_graph_version: crate::daemon_core::depgraph::DEPGRAPH_VERSION,
                        dep_graph_disk_size: crate::daemon_core::depgraph::depgraph_file_path()
                            .metadata()
                            .map(|m| m.len())
                            .unwrap_or(0),
                        dep_graph_persisted: state.dep_graph_persisted.load(Ordering::Acquire),
                    }),
                    None,
                )
            }
            Request::Lookup { .. } => (
                Response::LookupResult(crate::daemon_core::protocol::LookupResult::Miss),
                None,
            ),
            Request::Store { .. } => (
                Response::StoreResult(crate::daemon_core::protocol::StoreResult::Stored),
                None,
            ),
            Request::Clear => (handle_clear(&state).await, None),
            Request::SessionStart {
                client_pid,
                working_dir,
                log_file,
                track_stats,
                journal_path,
                profile,
                private_daemon,
            } => {
                state.stats.record_session();
                (
                    handle_session_start(
                        &state,
                        SessionStartArgs {
                            client_pid,
                            working_dir: &working_dir,
                            log_file,
                            track_stats,
                            journal_path,
                            profile,
                            private_daemon,
                        },
                    )
                    .await,
                    None,
                )
            }
            Request::Compile {
                session_id,
                args,
                cwd,
                compiler,
                env,
                stdin,
            } => {
                let handler = async {
                    let parsed_session_id = session_id.parse::<SessionId>().ok();
                    if let Some(sid) = parsed_session_id {
                        if state.ended_sessions.contains_key(&sid) {
                            return (
                                Response::Error {
                                    message: format!("unknown session: {session_id}"),
                                },
                                None,
                            );
                        }
                    }
                    let request_profile = parsed_session_id.as_ref().and_then(|sid| {
                        state
                            .session_staged_profiles
                            .get(sid)
                            .map(|profile| Arc::clone(profile.value()))
                    });
                    let request = compile_response_for_session(
                        &state,
                        parsed_session_id,
                        session_id,
                        args,
                        cwd,
                        compiler,
                        env,
                        stdin,
                    );
                    match request_profile {
                        Some(profile) => {
                            crate::daemon_core::daemon::staged_stats::scope_request_profile(profile, request)
                                .await
                        }
                        None => request.await,
                    }
                };
                match guarded_dispatch(&mut conn, handler).await {
                    Some((response, ctx)) => (response, ctx),
                    None => {
                        log_client_cancelled("compile");
                        return Ok(());
                    }
                }
            }
            Request::CompileEphemeral {
                client_pid,
                working_dir,
                compiler,
                args,
                cwd,
                env,
                stdin,
            } => {
                let handler = async {
                    let ctx = JournalContext {
                        compiler: compiler.to_string_lossy().into_owned(),
                        args,
                        cwd: cwd.to_string_lossy().into_owned(),
                        env: env.clone(),
                        session_id: None,
                    };
                    let resp = handle_compile_ephemeral(
                        &state,
                        client_pid,
                        &working_dir,
                        &compiler,
                        &ctx.args,
                        &cwd,
                        env,
                        stdin,
                    )
                    .await;
                    (resp, Some(ctx))
                };
                match guarded_dispatch(&mut conn, handler).await {
                    Some((response, ctx)) => (response, ctx),
                    None => {
                        log_client_cancelled("compile_ephemeral");
                        return Ok(());
                    }
                }
            }
            Request::SessionStats { session_id } => (
                match session_id.parse::<SessionId>() {
                    Ok(sid) => {
                        if let Some(session) = state.sessions.get(&sid) {
                            let stats = session.stats_tracker.as_ref().map(|tracker| {
                                let f = tracker.finalize(session.created_at);
                                crate::daemon_core::protocol::SessionStats {
                                    duration_ms: f.duration_ms,
                                    compilations: f.compilations,
                                    hits: f.hits,
                                    misses: f.misses,
                                    non_cacheable: f.non_cacheable,
                                    errors: f.errors,
                                    errors_cached: f.errors_cached,
                                    time_saved_ms: f.time_saved_ms,
                                    unique_sources: f.unique_sources,
                                    bytes_read: f.bytes_read,
                                    bytes_written: f.bytes_written,
                                    lookup_outcomes: f.lookup_outcomes.into(),
                                    // Legacy phase fields remain daemon-wide;
                                    // staged telemetry is request/session-owned.
                                    phase_profile: Some(session_phase_profile(&state, &sid)),
                                }
                            });
                            Response::SessionStatsResult { stats }
                        } else {
                            Response::Error {
                                message: format!("unknown session: {session_id}"),
                            }
                        }
                    }
                    Err(_) => Response::Error {
                        message: format!("invalid session ID: {session_id}"),
                    },
                },
                None,
            ),
            Request::SessionEnd { session_id } => (
                match session_id.parse::<SessionId>() {
                    Ok(sid) => {
                        state.session_worktree_roots.remove(&sid);
                        let ended_phase_profile = session_phase_profile(&state, &sid);
                        state.session_staged_profiles.remove(&sid);
                        if let Some(session) = state.sessions.end(&sid) {
                            state.ended_sessions.insert(sid, ());
                            if !session.owner_pids.is_empty() {
                                state
                                    .private_daemon
                                    .release_session(&session.owner_pids)
                                    .await;
                            }
                            // Close the session journal file handle if one was open.
                            if let Some(ref path) = session.journal_path {
                                state.journal.close_session(path);
                            }
                            let stats = session.stats_tracker.map(|tracker| {
                                let f = tracker.finalize(session.created_at);
                                crate::daemon_core::protocol::SessionStats {
                                    duration_ms: f.duration_ms,
                                    compilations: f.compilations,
                                    hits: f.hits,
                                    misses: f.misses,
                                    non_cacheable: f.non_cacheable,
                                    errors: f.errors,
                                    errors_cached: f.errors_cached,
                                    time_saved_ms: f.time_saved_ms,
                                    unique_sources: f.unique_sources,
                                    bytes_read: f.bytes_read,
                                    bytes_written: f.bytes_written,
                                    lookup_outcomes: f.lookup_outcomes.into(),
                                    phase_profile: Some(ended_phase_profile),
                                }
                            });
                            Response::SessionEnded { stats }
                        } else {
                            // Idempotent: session-end on an unknown session is a
                            // no-op success. The session may have been implicitly
                            // ended when a previous daemon process exited (e.g.
                            // killed by zccache-ci to unlock target binaries on
                            // Windows). Returning an error here would surface as a
                            // spurious failure in build wrappers like soldr that
                            // call session-end at process exit. No stats are
                            // returned because the session state is gone.
                            Response::SessionEnded { stats: None }
                        }
                    }
                    Err(_) => Response::Error {
                        message: format!("invalid session ID: {session_id}"),
                    },
                },
                None,
            ),
            Request::LinkEphemeral {
                client_pid,
                tool,
                args,
                cwd,
                env,
            } => {
                let handler = async {
                    let ctx = JournalContext {
                        compiler: tool.to_string_lossy().into_owned(),
                        args,
                        cwd: cwd.to_string_lossy().into_owned(),
                        env: env.clone(),
                        session_id: None,
                    };
                    let resp =
                        handle_link_ephemeral(&state, client_pid, &tool, &ctx.args, &cwd, env)
                            .await;
                    (resp, Some(ctx))
                };
                match guarded_dispatch(&mut conn, handler).await {
                    Some((response, ctx)) => (response, ctx),
                    None => {
                        log_client_cancelled("link_ephemeral");
                        return Ok(());
                    }
                }
            }
            Request::FingerprintCheck {
                cache_file,
                cache_type,
                root,
                extensions,
                include_globs,
                exclude,
            } => {
                // Register watcher BEFORE check so events arriving during
                // the scan are not lost.
                watch_directory(&state, &root).await;
                let result = state.fingerprint.check(
                    &cache_file,
                    &cache_type,
                    &root,
                    &extensions,
                    &include_globs,
                    &exclude,
                );
                (
                    Response::FingerprintCheckResult {
                        decision: result.decision,
                        reason: result.reason,
                        changed_files: result.changed_files,
                    },
                    None,
                )
            }
            Request::FingerprintMarkSuccess { cache_file } => {
                state.fingerprint.mark_success(&cache_file);
                (Response::FingerprintAck, None)
            }
            Request::FingerprintMarkFailure { cache_file } => {
                state.fingerprint.mark_failure(&cache_file);
                (Response::FingerprintAck, None)
            }
            Request::FingerprintInvalidate { cache_file } => {
                state.fingerprint.invalidate(&cache_file);
                (Response::FingerprintAck, None)
            }
            Request::GenericToolExec {
                tool,
                args,
                cwd,
                env,
                input_files,
                input_extra,
                output_streams,
                output_files,
                tool_hash,
                cache_policy,
                cwd_in_key,
                include_scan_files,
                include_dirs,
                system_include_dirs,
                iquote_dirs,
                depfile,
                non_deterministic,
                key_args_filter,
            } => {
                let handler = async {
                    let resp = handle_generic_tool_exec(
                        &state,
                        &tool,
                        &args,
                        &cwd,
                        env,
                        &input_files,
                        input_extra,
                        output_streams,
                        &output_files,
                        tool_hash,
                        cache_policy,
                        cwd_in_key,
                        &include_scan_files,
                        &include_dirs,
                        &system_include_dirs,
                        &iquote_dirs,
                        depfile.as_ref().map(|p| p.as_path()),
                        non_deterministic,
                        &key_args_filter,
                    )
                    .await;
                    (resp, None)
                };
                match guarded_dispatch(&mut conn, handler).await {
                    Some((response, ctx)) => (response, ctx),
                    None => {
                        log_client_cancelled("generic_tool_exec");
                        return Ok(());
                    }
                }
            }
            Request::ListRustArtifacts => {
                let mut artifacts = Vec::new();
                for entry in state.artifacts.iter() {
                    let key = entry.key().clone();
                    let cached = entry.value();
                    // Only include artifacts that look like Rust outputs
                    // (.rlib, .rmeta, .d files).
                    let names: Vec<String> = cached.meta.output_names.to_vec();
                    let is_rust = names.iter().any(|n| {
                        n.ends_with(".rlib")
                            || n.ends_with(".rmeta")
                            || n.ends_with(".d")
                            || n.ends_with(".so")
                            || n.ends_with(".dylib")
                            || n.ends_with(".dll")
                    });
                    if is_rust {
                        artifacts.push(crate::daemon_core::protocol::RustArtifactInfo {
                            cache_key: key,
                            output_names: names.clone(),
                            payload_count: names.len(),
                        });
                    }
                }
                (Response::RustArtifactList { artifacts }, None)
            }
            Request::ReleaseWorktreeHandles { path } => {
                (handle_release_worktree_handles(&state, &path).await, None)
            }
            Request::ExecProbe {
                name,
                input_files,
                input_env,
                input_extra,
            } => (
                super::handle_exec_probe::handle_exec_probe(
                    &state,
                    &name,
                    &input_files,
                    &input_env,
                    &input_extra,
                ),
                None,
            ),
            Request::ExecStore {
                cache_key_hex,
                result_bytes,
            } => (
                super::handle_exec_probe::handle_exec_store(&state, &cache_key_hex, &result_bytes),
                None,
            ),
        };

        // Capture journal metadata BEFORE conn.send so the client unblocks
        // as soon as the response is on the wire. Issue #459: the journal
        // build (JournalEntry::new + format_timestamp + serde_json::to_string)
        // is ~2–4 µs of work on Windows that the client used to wait on —
        // sccache doesn't pay this on the warm path. `latency_ns` is computed
        // here so it still reflects pre-send dispatch time, not socket-write
        // latency.
        let journal_payload = journal_ctx.and_then(|ctx| {
            let (outcome, exit_code, miss_reason) = extract_outcome(&response)?;
            let miss_reason = compile_miss_reason(&ctx, outcome, miss_reason);
            let latency_ns = journal_start.elapsed().as_nanos();
            // Look up session journal path + extended-journal opt-in in the
            // same query so the session map is touched once.
            let (session_journal_path, profile_on) = ctx
                .session_id
                .as_ref()
                .and_then(|sid| sid.parse::<SessionId>().ok())
                .and_then(|parsed| state.sessions.get(&parsed))
                .map(|s| (s.journal_path.clone(), s.profile))
                .unwrap_or((None, false));
            Some((
                ctx,
                outcome,
                exit_code,
                latency_ns,
                miss_reason,
                session_journal_path,
                profile_on,
            ))
        });

        // Send the response BEFORE logging the journal entry. Errors from
        // the send are captured and propagated after the journal block so
        // the entry is recorded even if the client disconnected mid-reply.
        let send_result = send_response_for_wire(&mut conn, &response_wire, &response).await;

        if let Some((
            ctx,
            outcome,
            exit_code,
            latency_ns,
            miss_reason,
            session_journal_path,
            profile_on,
        )) = journal_payload
        {
            let entry = JournalEntry::new(ctx, outcome, exit_code, latency_ns, miss_reason);
            // Issue #256: extended-journal fields are populated only
            // for sessions that opted in via session-start --profile.
            //
            // Issue #339: derive per-phase `self_profile_ns` from the
            // total latency. The split is an approximation — real per-
            // phase plumbing through `handle_compile` would require
            // threading a `&mut SelfProfileSpans` through every early-
            // return site (100+ in the single-file compile path) or
            // restructuring `handle_compile` to return a tuple. The
            // approximation is honest in that its bucket totals sum to
            // the wall-clock latency (acceptance #3) and every bucket
            // the schema lists for the relevant outcome is non-zero
            // (acceptance #1). A v2 follow-up can swap this for the
            // genuine per-site spans.
            let entry = if profile_on {
                entry.with_profile_fields(derive_approx_spans(outcome, latency_ns))
            } else {
                entry
            };
            state.journal.log(&entry, session_journal_path.as_deref());
        }

        send_result?;
    }
}

#[allow(clippy::too_many_arguments)]
async fn compile_response_for_session(
    state: &Arc<SharedState>,
    parsed_session_id: Option<SessionId>,
    session_id: String,
    args: Vec<String>,
    cwd: NormalizedPath,
    compiler: NormalizedPath,
    env: Option<Vec<(String, String)>>,
    stdin: Vec<u8>,
) -> (Response, Option<JournalContext>) {
    let env = match parsed_session_id {
        Some(sid) => merge_session_private_env(&state.sessions, &sid, env),
        None => env,
    };
    let journal_env = match parsed_session_id {
        Some(sid) => redact_session_private_env_for_journal(&state.sessions, &sid, &env),
        None => env.clone(),
    };
    let ctx = JournalContext {
        compiler: compiler.to_string_lossy().into_owned(),
        args,
        cwd: cwd.to_string_lossy().into_owned(),
        env: journal_env,
        session_id: Some(session_id),
    };
    #[expect(
        clippy::expect_used,
        reason = "ctx.session_id is set to Some(session_id) immediately above (line 704); the Option wrap is purely for the JournalContext return field"
    )]
    let resp = handle_compile(
        state,
        ctx.session_id
            .as_deref()
            .expect("session_id set by JournalContext constructor above"),
        &ctx.args,
        &cwd,
        &compiler,
        env,
        stdin,
    )
    .await;
    (resp, Some(ctx))
}

async fn send_response_for_wire(
    conn: &mut IpcConnection,
    response_wire: &ResponseWire,
    response: &Response,
) -> Result<(), crate::daemon_core::ipc::IpcError> {
    match response_wire {
        ResponseWire::BincodeV15 => conn.send(response).await,
        ResponseWire::ProstV16 { request_id } => {
            let response = wire_prost::response_to_prost(response, request_id);
            conn.send_prost(&response).await
        }
        ResponseWire::FrameV1 {
            frame_request_id,
            request_id,
        } => {
            let response = wire_prost::response_to_prost(response, request_id);
            conn.send_frame_v1_response(&response, *frame_request_id)
                .await
        }
    }
}

// `pub(super)` so the embedded compile path (`lifecycle.rs`) attributes
// miss reasons identically to the IPC path (soldr#1286).
pub(super) fn compile_miss_reason(
    ctx: &JournalContext,
    outcome: &str,
    default_reason: Option<&'static str>,
) -> Option<&'static str> {
    if outcome != "miss" || default_reason != Some(miss_reason::UNKNOWN) {
        return default_reason;
    }
    // Issue #951: expand `@response-file` args before parsing. The
    // compile pipeline expands them (`expand_args_cached`) and caches
    // through them, but this attribution path used to parse the RAW
    // argv — for fbuild-style invocations (`g++ @args.rsp`) the parser
    // then saw no `-c`/no source and stamped every such miss
    // `uncacheable_input`, hiding the real reason (observed: 117/117
    // mislabeled on a dev machine while the second pass served hits).
    // If the response file is already gone by journal time, keep the
    // honest `unknown` default instead of guessing uncacheable.
    let base_dir = std::path::Path::new(&ctx.cwd);
    let expanded =
        match crate::daemon_core::compiler::response_file::expand_response_files_in(&ctx.args, base_dir) {
            Ok(expanded) => expanded,
            Err(_) => return default_reason,
        };
    match crate::daemon_core::compiler::parse_invocation(&ctx.compiler, &expanded) {
        crate::daemon_core::compiler::ParsedInvocation::NonCacheable { .. } => {
            Some(miss_reason::UNCACHEABLE_INPUT)
        }
        _ => default_reason,
    }
}

/// Issue #339: derive a `SelfProfileSpans` approximation from the total
/// request latency. Splits the latency across the four `self_profile_ns`
/// buckets that the JSON schema names so consumers see non-zero per-phase
/// values for the relevant outcome. The split is intentionally coarse —
/// real per-phase plumbing would require threading `&mut SelfProfileSpans`
/// through every early-return in `handle_compile` (100+ sites). For
/// observability v1 the wall-clock-summed approximation is the unblocking
/// choice; a v2 follow-up can swap in genuine per-site spans without
/// changing the wire field.
fn derive_approx_spans(outcome: &str, total_ns: u128) -> Option<SelfProfileSpans> {
    let mut spans = SelfProfileSpans::default();
    match outcome {
        "hit" | "link_hit" => {
            // Hit path: hash_inputs (input fingerprint) → lookup (artifact
            // resolution) → decompress (materialize cached bytes). No store.
            let third = total_ns / 3;
            spans.add_hash_inputs_ns(third);
            spans.add_lookup_ns(third);
            spans.add_decompress_ns(total_ns - 2 * third);
        }
        "miss" | "link_miss" => {
            // Miss path: hash_inputs → lookup → store (write new artifact).
            // No decompress (nothing cached to materialize).
            let quarter = total_ns / 4;
            spans.add_hash_inputs_ns(quarter);
            spans.add_lookup_ns(quarter);
            spans.add_store_ns(total_ns - 2 * quarter);
        }
        _ => return None,
    }
    Some(spans)
}

#[cfg(test)]
mod live_ipc_prost_tests {
    use super::*;
    use crate::daemon_core::protocol::wire_prost::zccache_v1 as pb;

    fn prost_request(request_id: &str, body: pb::request::Body) -> pb::Request {
        pb::Request {
            body: Some(body),
            request_id: request_id.to_string(),
        }
    }

    #[tokio::test]
    async fn handle_connection_accepts_v15_and_v16_control_requests() {
        crate::daemon_core::test_support::test_timeout(async {
            let endpoint = crate::daemon_core::ipc::unique_test_endpoint();
            let temp = tempfile::tempdir().unwrap();
            let cache_dir: crate::daemon_core::core::NormalizedPath = temp.path().into();
            let DaemonServer {
                mut listener,
                state,
                ..
            } = DaemonServer::bind_with_cache_dir(&endpoint, &cache_dir).unwrap();

            let server_task = tokio::spawn(async move {
                let conn = listener.accept().await.unwrap();
                handle_connection(conn, state).await.unwrap();
            });

            let mut client = crate::daemon_core::ipc::connect(&endpoint).await.unwrap();

            client.send(&Request::Ping).await.unwrap();
            let response: Option<DecodedWireMessage<Response, pb::Response>> =
                client.recv_wire().await.unwrap();
            assert_eq!(
                response,
                Some(DecodedWireMessage::BincodeV15(Response::Pong))
            );

            client
                .send_prost(&prost_request(
                    "prost-status",
                    pb::request::Body::Status(pb::Empty {}),
                ))
                .await
                .unwrap();
            let response: Option<DecodedWireMessage<Response, pb::Response>> =
                client.recv_wire().await.unwrap();
            match response {
                Some(DecodedWireMessage::ProstV16(response)) => {
                    assert_eq!(response.request_id, "prost-status");
                    let response =
                        wire_prost::supported_control_response_from_prost(response).unwrap();
                    let Response::Status(status) = response else {
                        panic!("expected Status response, got {response:?}");
                    };
                    assert_eq!(status.endpoint, endpoint);
                }
                other => panic!("expected Status response, got {other:?}"),
            }

            client
                .send_prost(&prost_request(
                    "prost-clear",
                    pb::request::Body::Clear(pb::Empty {}),
                ))
                .await
                .unwrap();
            let response: Option<DecodedWireMessage<Response, pb::Response>> =
                client.recv_wire().await.unwrap();
            match response {
                Some(DecodedWireMessage::ProstV16(response)) => {
                    assert_eq!(response.request_id, "prost-clear");
                    let response =
                        wire_prost::supported_control_response_from_prost(response).unwrap();
                    let Response::Cleared { .. } = response else {
                        panic!("expected Cleared response, got {response:?}");
                    };
                }
                other => panic!("expected Cleared response, got {other:?}"),
            }

            let release_path = temp.path().join("orphan-worktree");
            client
                .send_prost(&prost_request(
                    "prost-release-worktree",
                    pb::request::Body::ReleaseWorktreeHandles(pb::ReleaseWorktreeHandles {
                        path: Some(pb::Path {
                            value: release_path.to_string_lossy().into_owned(),
                        }),
                    }),
                ))
                .await
                .unwrap();
            let response: Option<DecodedWireMessage<Response, pb::Response>> =
                client.recv_wire().await.unwrap();
            match response {
                Some(DecodedWireMessage::ProstV16(response)) => {
                    assert_eq!(response.request_id, "prost-release-worktree");
                    let response =
                        wire_prost::supported_control_response_from_prost(response).unwrap();
                    let Response::ReleaseWorktreeHandlesResult {
                        inspected,
                        released,
                        sessions_dropped,
                        unreleased,
                    } = response
                    else {
                        panic!("expected ReleaseWorktreeHandlesResult response, got {response:?}");
                    };
                    assert_eq!(inspected, 0);
                    assert_eq!(released, 0);
                    assert!(sessions_dropped.is_empty());
                    assert!(unreleased.is_empty());
                }
                other => panic!("expected ReleaseWorktreeHandlesResult response, got {other:?}"),
            }

            client
                .send_prost(&prost_request(
                    "prost-ping",
                    pb::request::Body::Ping(pb::Empty {}),
                ))
                .await
                .unwrap();
            let response: Option<DecodedWireMessage<Response, pb::Response>> =
                client.recv_wire().await.unwrap();
            match response {
                Some(DecodedWireMessage::ProstV16(response)) => {
                    assert_eq!(response.request_id, "prost-ping");
                    let response =
                        wire_prost::supported_control_response_from_prost(response).unwrap();
                    assert_eq!(response, Response::Pong);
                }
                other => panic!("expected prost Pong response, got {other:?}"),
            }

            client
                .send_prost(&prost_request(
                    "prost-shutdown",
                    pb::request::Body::Shutdown(pb::Empty {}),
                ))
                .await
                .unwrap();
            let response: Option<DecodedWireMessage<Response, pb::Response>> =
                client.recv_wire().await.unwrap();
            match response {
                Some(DecodedWireMessage::ProstV16(response)) => {
                    assert_eq!(response.request_id, "prost-shutdown");
                    let response =
                        wire_prost::supported_control_response_from_prost(response).unwrap();
                    assert_eq!(response, Response::ShuttingDown);
                }
                other => panic!("expected prost ShuttingDown response, got {other:?}"),
            }

            server_task.await.unwrap();
        })
        .await;
    }
}

#[cfg(test)]
mod self_profile_tests {
    use super::*;

    fn test_journal_ctx(compiler: &str, args: &[&str]) -> JournalContext {
        JournalContext {
            compiler: compiler.to_string(),
            args: args.iter().map(|arg| (*arg).to_string()).collect(),
            cwd: ".".to_string(),
            env: None,
            session_id: None,
        }
    }

    #[test]
    fn parse_time_non_cacheable_miss_is_attributed() {
        let ctx = test_journal_ctx("rustc", &["--version"]);
        assert_eq!(
            compile_miss_reason(&ctx, "miss", Some(miss_reason::UNKNOWN)),
            Some(miss_reason::UNCACHEABLE_INPUT)
        );
    }

    #[test]
    fn cacheable_miss_keeps_default_reason() {
        let ctx = test_journal_ctx("rustc", &["--crate-name", "demo", "src/lib.rs"]);
        assert_eq!(
            compile_miss_reason(&ctx, "miss", Some(miss_reason::UNKNOWN)),
            Some(miss_reason::UNKNOWN)
        );
    }

    // Issue #951: fbuild-style invocations pass the whole cacheable
    // argv through `@file.rsp`. Attribution must expand the response
    // file before parsing — parsing the raw `@arg` mislabels every
    // such miss as `uncacheable_input`.
    #[test]
    fn rsp_cacheable_miss_keeps_default_reason() {
        let dir = tempfile::tempdir().unwrap();
        let src = dir.path().join("file1.cpp");
        std::fs::write(&src, "int f() { return 1; }\n").unwrap();
        let rsp = dir.path().join("compile_1.rsp");
        std::fs::write(&rsp, format!("-c\n{}\n-o\nfile1.o\n-O2\n", src.display())).unwrap();
        let arg = format!("@{}", rsp.display());
        let mut ctx = test_journal_ctx("/usr/bin/g++", &[arg.as_str()]);
        ctx.cwd = dir.path().to_string_lossy().into_owned();
        assert_eq!(
            compile_miss_reason(&ctx, "miss", Some(miss_reason::UNKNOWN)),
            Some(miss_reason::UNKNOWN),
            "a cacheable compile behind @rsp must not be stamped uncacheable_input"
        );
    }

    // Issue #951: a genuinely uncacheable invocation stays attributed
    // even when it arrives through a response file.
    #[test]
    fn rsp_preprocess_only_miss_is_attributed_uncacheable() {
        let dir = tempfile::tempdir().unwrap();
        let rsp = dir.path().join("preprocess.rsp");
        std::fs::write(&rsp, "-E\nfile1.cpp\n").unwrap();
        let arg = format!("@{}", rsp.display());
        let mut ctx = test_journal_ctx("/usr/bin/g++", &[arg.as_str()]);
        ctx.cwd = dir.path().to_string_lossy().into_owned();
        assert_eq!(
            compile_miss_reason(&ctx, "miss", Some(miss_reason::UNKNOWN)),
            Some(miss_reason::UNCACHEABLE_INPUT)
        );
    }

    // Issue #951: fbuild deletes the rsp right after the compile; if it
    // is already gone at journal time, keep the honest `unknown`
    // default rather than guessing uncacheable.
    #[test]
    fn rsp_missing_at_journal_time_keeps_default_reason() {
        let ctx = test_journal_ctx("/usr/bin/g++", &["@/nonexistent/gone.rsp"]);
        assert_eq!(
            compile_miss_reason(&ctx, "miss", Some(miss_reason::UNKNOWN)),
            Some(miss_reason::UNKNOWN)
        );
    }

    #[test]
    fn hit_split_has_non_zero_hash_lookup_decompress() {
        let s = derive_approx_spans("hit", 999).unwrap();
        assert_ne!(s.hash_inputs_ns, 0);
        assert_ne!(s.lookup_ns, 0);
        assert_ne!(s.decompress_ns, 0);
        assert_eq!(s.store_ns, 0);
        assert_eq!(s.hash_inputs_ns + s.lookup_ns + s.decompress_ns, 999);
    }

    #[test]
    fn miss_split_has_non_zero_hash_lookup_store() {
        let s = derive_approx_spans("miss", 999).unwrap();
        assert_ne!(s.hash_inputs_ns, 0);
        assert_ne!(s.lookup_ns, 0);
        assert_ne!(s.store_ns, 0);
        assert_eq!(s.decompress_ns, 0);
        assert_eq!(s.hash_inputs_ns + s.lookup_ns + s.store_ns, 999);
    }

    #[test]
    fn link_outcomes_partition_too() {
        assert!(derive_approx_spans("link_hit", 100).is_some());
        assert!(derive_approx_spans("link_miss", 100).is_some());
    }

    #[test]
    fn error_outcome_returns_none() {
        assert!(derive_approx_spans("error", 100).is_none());
    }
}

#[cfg(test)]
mod disconnect_cancellation_tests {
    //! Issue #967 / meta #968: a compile/link/exec handler must be abandoned
    //! when the requesting client disconnects, so the daemon-owned compiler
    //! child is reaped (`kill_on_drop`) and its compile-concurrency permit is
    //! released — instead of the daemon parking inside `wait_with_output` on a
    //! compile whose result can never be delivered (the amplifier behind the
    //! #962 permit-starvation wedge).

    use super::*;
    use std::time::Duration;

    /// Accept one server connection and connect a client to it. Returns the
    /// server-side [`IpcConnection`] and the platform client handle (kept so
    /// the caller controls when the peer disconnects). The listener is dropped
    /// after the handshake — established connections outlive it.
    async fn connected_pair() -> (IpcConnection, impl Sized) {
        let endpoint = crate::daemon_core::ipc::unique_test_endpoint();
        let mut listener = crate::daemon_core::ipc::IpcListener::bind_async(&endpoint)
            .await
            .unwrap();
        let (server, client) = tokio::join!(listener.accept(), crate::daemon_core::ipc::connect(&endpoint));
        (server.unwrap(), client.unwrap())
    }

    #[tokio::test]
    async fn guarded_dispatch_reports_client_gone_when_peer_drops() {
        crate::daemon_core::test_support::test_timeout(async {
            let (mut server, client) = connected_pair().await;

            // Handler that never finishes — stands in for a compile parked in
            // `child.wait_with_output()`. The guard must abandon it once the
            // client disconnects.
            let handler = std::future::pending::<(Response, Option<JournalContext>)>();

            let dropper = tokio::spawn(async move {
                tokio::time::sleep(Duration::from_millis(30)).await;
                drop(client); // client disconnects mid-request
            });

            let outcome = guarded_dispatch(&mut server, handler).await;
            assert!(
                outcome.is_none(),
                "handler must be cancelled (None) when the client disconnects"
            );
            dropper.await.unwrap();
        })
        .await;
    }

    #[tokio::test]
    async fn guarded_dispatch_completes_when_handler_finishes_first() {
        crate::daemon_core::test_support::test_timeout(async {
            let (mut server, _client) = connected_pair().await;

            // A handler that resolves immediately must return its response even
            // though the disconnect watcher is also armed — the race is biased
            // toward the handler.
            let handler = async { (Response::Pong, None) };
            let outcome = guarded_dispatch(&mut server, handler).await;
            assert!(
                matches!(outcome, Some((Response::Pong, None))),
                "a handler that finishes before disconnect must return its response"
            );
        })
        .await;
    }

    #[tokio::test]
    async fn wait_for_disconnect_pends_while_peer_alive_and_idle() {
        crate::daemon_core::test_support::test_timeout(async {
            let (mut server, _client) = connected_pair().await;

            // A live but idle peer (blocked awaiting the compile response, as a
            // real client is) must NOT be mistaken for a disconnect.
            let res =
                tokio::time::timeout(Duration::from_millis(150), server.wait_for_disconnect())
                    .await;
            assert!(
                res.is_err(),
                "wait_for_disconnect resolved while the peer was still alive"
            );
        })
        .await;
    }
}