beam_worker/worker_runtime/

run_loop.rs

use super::*;

/// Returns `Some(("TERM", "xterm-256color"))` when the CLI's spec sets
/// `inject_term_xterm` (codex/traex) and the inherited `TERM` is missing,
/// empty, or `"dumb"`.
///
/// For any other CLI, or when `TERM` is already set to a non-empty value other
/// than `"dumb"`, returns `None` — the environment is never overwritten.
///
/// This is a pure function so it can be tested deterministically without
/// touching the real process environment.
pub(crate) fn maybe_inject_term(
    cli_id: &str,
    current_term: Option<&str>,
) -> Option<(String, String)> {
    let inject = beam_core::cli_specs::cli_spec(cli_id)
        .map(|spec| spec.inject_term_xterm)
        .unwrap_or(false);
    if !inject {
        return None;
    }
    match current_term {
        None | Some("") | Some("dumb") => Some(("TERM".to_string(), "xterm-256color".to_string())),
        _ => None,
    }
}

/// Consecutive dead `is_alive` samples required before emitting `CliExit`.
/// A single false negative (empty `list-panes`, transient pane hide) must
/// not tear the worker down.
pub(crate) const CLI_DEAD_CONFIRM_ATTEMPTS: usize = 3;

/// Count consecutive dead liveness samples. Returns true when the worker
/// should emit `CliExit`.
pub(crate) fn should_emit_cli_exit(consecutive_dead: &mut usize, alive: bool) -> bool {
    if alive {
        *consecutive_dead = 0;
        return false;
    }
    *consecutive_dead = consecutive_dead.saturating_add(1);
    *consecutive_dead >= CLI_DEAD_CONFIRM_ATTEMPTS
}

pub(crate) async fn send_message(
    stdout: &Arc<Mutex<tokio::io::Stdout>>,
    msg: &WorkerToDaemon,
) -> Result<()> {
    let mut out = stdout.lock().await;
    out.write_all(serde_json::to_string(msg)?.as_bytes())
        .await?;
    out.write_all(b"\n").await?;
    out.flush().await?;
    Ok(())
}

pub async fn run(init: InitConfig) -> Result<()> {
    let stdout = Arc::new(Mutex::new(tokio::io::stdout()));
    let session_name = format!("beam-{}", &init.session_id[..8.min(init.session_id.len())]);
    let paths = BeamPaths::discover()?;
    let adapter = Arc::new(Mutex::new(CliAdapter::from_init(&init)?));
    // Backend selection: Herdr adopted sessions observe a user pane; Herdr
    // managed sessions spawn into a labeled workspace; everything else keeps
    // the zellij path unchanged.
    let (backend_impl, attach_context, herdr_handle) =
        crate::backend::select::select_backend(&init, &session_name);
    let spawn_spec = adapter.lock().await.build_spawn_spec(&init);
    let extra_env = maybe_inject_term(&init.cli_id, std::env::var("TERM").ok().as_deref())
        .into_iter()
        .collect();
    let exe_path = std::env::current_exe()
        .ok()
        .map(|p| p.display().to_string())
        .unwrap_or_else(|| "beam".to_string());
    let bindir = std::env::current_exe()
        .ok()
        .and_then(|p| p.parent().map(|v| v.display().to_string()))
        .unwrap_or_default();
    let args = if init.adopted_from.is_some() {
        (spawn_spec.bin, spawn_spec.args)
    } else {
        if super::launch::normalize_cgroup_slice(init.cgroup_slice.as_deref()).is_some()
            && super::launch::current_launch_platform() == super::launch::LaunchPlatform::Linux
            && !super::launch::systemd_run_available()
        {
            anyhow::bail!(
                "cgroupSlice requires systemd-run on Linux (user systemd); install systemd or unset cgroupSlice"
            );
        }
        super::launch::build_launch_spec(
            super::launch::current_launch_platform(),
            super::launch::LaunchInput {
                cgroup_slice: init.cgroup_slice.clone(),
                session_id: init.session_id.clone(),
                beam_home: paths.root().display().to_string(),
                beam_bin: exe_path,
                path_prepend: bindir,
                extra_env,
                spec: spawn_spec,
            },
        )?
    };
    let spawn_opts = SpawnOpts {
        cwd: init.working_dir.clone(),
        cols: DEFAULT_TERMINAL_COLS,
        rows: DEFAULT_TERMINAL_ROWS,
        env: Vec::new(),
    };
    backend_impl
        .spawn(&args.0, &args.1, spawn_opts)
        .await
        .with_context(|| format!("failed to {} session {}", attach_context, init.session_id))?;
    // Shared handle: the backend synchronizes internally per operation, so
    // no outer Mutex is needed and a long write_input() never blocks screen
    // capture, terminal keys, or the screenshot coordinator.
    let backend = backend_impl;
    let (herdr_workspace_id, herdr_pane_id) = crate::backend::select::ready_herdr_identity(
        herdr_handle
            .as_ref()
            .and_then(|h| h.herdr_ids())
            .map(|ids| ids.workspace_pane()),
        init.adopted_from.as_ref(),
    );
    let mut cli_pid_marker = None;
    let child_pid = backend.child_pid().await?;
    adapter.lock().await.on_spawned(child_pid);
    if let Some(pid) = child_pid {
        tokio::fs::create_dir_all(paths.cli_pid_markers_dir()).await?;
        let marker = paths.cli_pid_markers_dir().join(pid.to_string());
        tokio::fs::write(&marker, init.session_id.as_bytes()).await?;
        cli_pid_marker = Some(marker);
    }
    // TUI CLIs drop keystrokes typed before their input UI is up. Wait for the
    // CLI's ready signal (welcome banner, or the composer line appearing)
    // before signaling Ready, so the initial prompt and the first stdin message
    // land on a live input field. Adopted sessions attach to an already-running
    // CLI; there is nothing to wait for.
    let ready_probe = crate::adapters::ready_probe(&init.cli_id);
    if init.adopted_from.is_none() && ready_probe != ReadyProbe::None {
        let ready = wait_for_ready(backend.as_ref(), ready_probe).await;
        if ready {
            info!(session = %init.session_id, adapter = %init.cli_id, probe = ?ready_probe, "CLI TUI ready signal observed");
        } else {
            warn!(session = %init.session_id, adapter = %init.cli_id, probe = ?ready_probe, "CLI TUI ready signal not observed within {}s; typing input anyway", TUI_READY_TIMEOUT.as_secs());
        }
    }
    let latest_screen = Arc::new(RwLock::new(String::new()));
    let latest_raw_screen = Arc::new(RwLock::new(String::new()));
    let display_mode = Arc::new(RwLock::new(DisplayMode::Hidden));
    let analyzer_runtime = Arc::new(RwLock::new(AnalyzerRuntime::default()));
    let usage_limit_tracker = Arc::new(Mutex::new(UsageLimitTracker::default()));
    let current_turn_id = Arc::new(RwLock::new(String::new()));
    let (updates, _) = broadcast::channel::<String>(256);

    send_message(
        &stdout,
        &WorkerToDaemon::Ready {
            zellij_session: session_name.clone(),
            backend_kind: init.backend_kind,
            herdr_workspace_id,
            herdr_pane_id,
        },
    )
    .await?;

    let sample_backend = backend.clone();
    let sample_screen = latest_screen.clone();
    let sample_raw_screen = latest_raw_screen.clone();
    let sample_updates = updates.clone();
    let sample_stdout = stdout.clone();
    let sample_adapter = adapter.clone();
    let sample_display_mode = display_mode.clone();
    let sample_usage_limit_tracker = usage_limit_tracker.clone();
    let sample_current_turn_id = current_turn_id.clone();
    let last_uploaded_hash = Arc::new(Mutex::new(None::<String>));
    let last_broadcast_hash: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
    let sample_last_broadcast_hash = last_broadcast_hash.clone();
    let sample_analyzer_runtime = analyzer_runtime.clone();
    // Channel for trigger events (TurnStarted, PaneUpdate, …) to the screenshot coordinator.
    let (trigger_tx, trigger_rx) = mpsc::channel::<Trigger>(16);
    let screen_capture_task = tokio::spawn(async move {
        let mut last_emitted_status = ScreenStatus::Starting;
        let mut last_emitted_usage_limit: Option<CliUsageLimitState> = None;
        let mut consecutive_dead = 0usize;
        loop {
            let (screen, alive) = {
                let screen = sample_backend.capture_viewport().await.unwrap_or_default();
                // Unknown/error is alive: a wedged probe must not emit CliExit.
                let alive = sample_backend.is_alive().await.unwrap_or(true);
                (screen, alive)
            };

            let hash_changed;

            {
                *sample_raw_screen.write().await = screen.clone();
                let mode = *sample_display_mode.read().await;
                let rendered = render_screen_for_display_mode(&screen, mode);
                let now_ms = now_ms();
                let analyzing = sample_analyzer_runtime.read().await.is_analyzing;
                let base_status = if analyzing {
                    ScreenStatus::Analyzing
                } else {
                    ScreenStatus::Working
                };
                let (status, usage_limit) =
                    sample_usage_limit_tracker
                        .lock()
                        .await
                        .classify(&screen, base_status, now_ms);
                let rendered_hash = lower_hex(&Sha256::digest(rendered.as_bytes()));
                {
                    let guard = sample_last_broadcast_hash.lock().await;
                    hash_changed = guard.as_deref() != Some(&rendered_hash);
                }

                if hash_changed
                    || last_emitted_status != status
                    || last_emitted_usage_limit != usage_limit
                {
                    *sample_last_broadcast_hash.lock().await = Some(rendered_hash.clone());
                    let mut current = sample_screen.write().await;
                    *current = rendered.clone();
                    let _ = sample_updates.send(rendered.clone());
                    let _ = send_message(
                        &sample_stdout,
                        &WorkerToDaemon::ScreenUpdate {
                            content: rendered.clone(),
                            status,
                            usage_limit: usage_limit.clone(),
                        },
                    )
                    .await;
                    last_emitted_status = status;
                    last_emitted_usage_limit = usage_limit.clone();
                }
            }

            if let Ok(poll) = sample_adapter.lock().await.poll() {
                if let Some(cli_session_id) = poll.cli_session_id {
                    let _ = send_message(
                        &sample_stdout,
                        &WorkerToDaemon::CliSessionId { cli_session_id },
                    )
                    .await;
                }
                if let Some((user_text, assistant_text)) = poll.adopt_preamble {
                    let _ = send_message(
                        &sample_stdout,
                        &WorkerToDaemon::AdoptPreamble {
                            user_text,
                            assistant_text,
                        },
                    )
                    .await;
                }
                if let Some(content) = poll.final_output {
                    let turn_id = sample_current_turn_id.read().await.clone();
                    let _ = send_message(
                        &sample_stdout,
                        &WorkerToDaemon::FinalOutput {
                            content,
                            turn_id,
                            kind: poll.final_output_kind,
                            user_text: poll.final_output_user_text,
                        },
                    )
                    .await;
                }
                if poll.prompt_ready {
                    let _ = send_message(&sample_stdout, &WorkerToDaemon::PromptReady).await;
                    let rendered = sample_screen.read().await.clone();
                    let raw = sample_raw_screen.read().await.clone();
                    let now_ms = now_ms();
                    let analyzing = sample_analyzer_runtime.read().await.is_analyzing;
                    let base_status = if analyzing {
                        ScreenStatus::Analyzing
                    } else {
                        ScreenStatus::Idle
                    };
                    let (status, usage_limit) =
                        sample_usage_limit_tracker
                            .lock()
                            .await
                            .classify(&raw, base_status, now_ms);
                    let _ = send_message(
                        &sample_stdout,
                        &WorkerToDaemon::ScreenUpdate {
                            content: rendered,
                            status,
                            usage_limit,
                        },
                    )
                    .await;
                }
            }

            if should_emit_cli_exit(&mut consecutive_dead, alive) {
                warn!(
                    consecutive_dead,
                    "cli liveness dead confirmed, emitting CliExit"
                );
                let _ = send_message(
                    &sample_stdout,
                    &WorkerToDaemon::CliExit {
                        code: Some(0),
                        signal: None,
                    },
                )
                .await;
                break;
            }
            if !alive {
                warn!(
                    consecutive_dead,
                    confirm_after = CLI_DEAD_CONFIRM_ATTEMPTS,
                    "cli liveness dead, waiting for confirmation"
                );
            }

            tokio::time::sleep(Duration::from_millis(5000)).await;
        }
    });
    let mut worker_joins = tokio::task::JoinSet::new();
    worker_joins.spawn(async move {
        let _ = screen_capture_task.await;
    });

    if init.cli_id == "grok" {
        let grok_raw_screen = latest_raw_screen.clone();
        let grok_runtime_state = analyzer_runtime.clone();
        let grok_stdout = stdout.clone();
        worker_joins.spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_millis(800)).await;
                let snapshot = grok_raw_screen.read().await.clone();
                if snapshot.is_empty() {
                    continue;
                }
                let options =
                    crate::worker_runtime::grok_prompts::detect_grok_plan_approval(&snapshot);
                let mut runtime = grok_runtime_state.write().await;
                match options {
                    Some(options) if !runtime.prompt_active => {
                        runtime.prompt_active = true;
                        let _ = send_message(
                            &grok_stdout,
                            &WorkerToDaemon::TuiPrompt {
                                description: "Grok plan approval".to_string(),
                                options,
                                multi_select: false,
                            },
                        )
                        .await;
                    }
                    None if runtime.prompt_active => {
                        runtime.prompt_active = false;
                        let _ = send_message(
                            &grok_stdout,
                            &WorkerToDaemon::TuiPromptResolved {
                                selected_text: None,
                            },
                        )
                        .await;
                    }
                    _ => {}
                }
            }
        });
    }

    if screen_analyzer_enabled(&init.screen_analyzer) {
        let analyzer_cfg = init.screen_analyzer.clone();
        let analyzer_raw_screen = latest_raw_screen.clone();
        let analyzer_runtime_state = analyzer_runtime.clone();
        let analyzer_stdout = stdout.clone();
        let analyzer_cli_id = init.cli_id.clone();
        let analyzer_task = tokio::spawn(async move {
            let client = Client::new();
            loop {
                tokio::time::sleep(Duration::from_millis(analyzer_cfg.interval_ms)).await;
                let snapshot = analyzer_raw_screen.read().await.clone();
                if snapshot.is_empty() {
                    continue;
                }
                let truncated = if snapshot.len() > analyzer_cfg.snapshot_max_chars {
                    snapshot[snapshot.len() - analyzer_cfg.snapshot_max_chars..].to_string()
                } else {
                    snapshot
                };
                let now = now_ms();
                {
                    let mut runtime = analyzer_runtime_state.write().await;
                    if truncated == runtime.last_snapshot {
                        runtime.stable_count = runtime.stable_count.saturating_add(1);
                    } else {
                        runtime.stable_count = 1;
                        runtime.last_snapshot = truncated.clone();
                        if runtime.waiting_for_content_change {
                            runtime.waiting_for_content_change = false;
                        }
                    }
                    if runtime.stable_count < analyzer_cfg.stable_count {
                        continue;
                    }
                    if runtime.waiting_for_content_change
                        && truncated == runtime.last_analyzed_snapshot
                    {
                        continue;
                    }
                    if runtime.cooldown_until_ms > now {
                        continue;
                    }
                    runtime.is_analyzing = true;
                    runtime.last_analyzed_snapshot = truncated.clone();
                }

                if analyzer_cli_id == "grok"
                    && crate::worker_runtime::grok_prompts::detect_grok_plan_approval(&truncated)
                        .is_some()
                {
                    let mut runtime = analyzer_runtime_state.write().await;
                    runtime.is_analyzing = false;
                    continue;
                }

                let result = call_screen_analyzer(&client, &analyzer_cfg, &truncated).await;

                let mut runtime = analyzer_runtime_state.write().await;
                runtime.is_analyzing = false;
                match result {
                    Ok(analysis) => {
                        apply_screen_analyzer_result(
                            &mut runtime,
                            &analysis.check_again_when,
                            now_ms(),
                        );
                        if analysis.needs_interaction && !analysis.options.is_empty() {
                            if !runtime.prompt_active {
                                runtime.prompt_active = true;
                                let _ = send_message(
                                    &analyzer_stdout,
                                    &WorkerToDaemon::TuiPrompt {
                                        description: analysis.description.clone().unwrap_or_else(
                                            || "CLI needs your selection".to_string(),
                                        ),
                                        options: analysis.options.clone(),
                                        multi_select: analysis.multi_select,
                                    },
                                )
                                .await;
                            }
                        } else if runtime.prompt_active {
                            runtime.prompt_active = false;
                            let _ = send_message(
                                &analyzer_stdout,
                                &WorkerToDaemon::TuiPromptResolved {
                                    selected_text: None,
                                },
                            )
                            .await;
                        }
                    }
                    Err(_) => {
                        runtime.waiting_for_content_change = true;
                        runtime.cooldown_until_ms = 0;
                    }
                }
            }
        });
        worker_joins.spawn(async move {
            let _ = analyzer_task.await;
        });
    }

    // Coordinator task — owns the trigger receiver and the 5-second fallback.
    {
        let coord_backend = backend.clone();
        let coord_stdout = stdout.clone();
        let coord_session_id = init.session_id.clone();
        let coord_app_id = init.lark_app_id.clone();
        let coord_app_secret = init.lark_app_secret.clone();
        let coord_display_mode = display_mode.clone();
        let coord_analyzer_runtime = analyzer_runtime.clone();
        let coord_usage_limit_tracker = usage_limit_tracker.clone();
        let coord_last_uploaded_hash = last_uploaded_hash.clone();
        let coord_latest_raw_screen = latest_raw_screen.clone();
        let coord_rx = trigger_rx;
        worker_joins.spawn(async move {
            coordinator_loop(
                coord_backend,
                coord_stdout,
                coord_session_id,
                coord_app_id,
                coord_app_secret,
                coord_display_mode,
                coord_analyzer_runtime,
                coord_usage_limit_tracker,
                coord_last_uploaded_hash,
                coord_latest_raw_screen,
                coord_rx,
            )
            .await;
        });
    }
    // Subscribe task: forward backend pane-update notifications to the screenshot coordinator.
    // Also caches the full viewport ANSI chunk so the coordinator can capture
    // without waiting on a slow backend call inside write_input().
    {
        let sub_backend = backend.clone();
        let sub_trigger_tx = trigger_tx.clone();
        let sub_latest_raw_screen = latest_raw_screen.clone();
        worker_joins.spawn(async move {
            let mut rx = sub_backend.subscribe();
            loop {
                match rx.recv().await {
                    Ok(chunk) => {
                        // latest wins: the chunk is the full viewport (not incremental)
                        *sub_latest_raw_screen.write().await = chunk;
                        if let Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) =
                            sub_trigger_tx.try_send(Trigger::PaneUpdate)
                        {
                            break;
                        } // Ok or Full → discard, keep listening
                    }
                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                }
            }
        });
    }
    // Herdr agent state polling: emit MuxAgentState for side effects only
    // (blocked → attention). v1 uses a bounded `agent get` poll.
    if init.backend_kind == beam_core::BackendKind::Herdr
        && let Some(handle) = herdr_handle.as_ref()
        && let Some(pane_id) = handle.herdr_ids().map(|ids| ids.pane_id)
    {
        let agent_stdout = stdout.clone();
        let agent_pane_id = pane_id.clone();
        worker_joins.spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(2)).await;
                let state = crate::backend::herdr::cli::agent_get(&agent_pane_id)
                    .await
                    .ok()
                    .and_then(|payload| crate::backend::herdr::cli::parse_agent_state(&payload));
                let Some(state) = state else {
                    continue;
                };
                if send_message(
                    &agent_stdout,
                    &WorkerToDaemon::MuxAgentState {
                        state,
                        agent_name: None,
                        pane_id: agent_pane_id.clone(),
                        message: None,
                    },
                )
                .await
                .is_err()
                {
                    break;
                }
            }
        });
    }
    if !init.prompt.is_empty() && !crate::adapters::passes_initial_prompt_via_args(&init.cli_id) {
        usage_limit_tracker.lock().await.begin_turn(
            "",
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap_or_default()
                .as_millis() as u64,
        );
        *current_turn_id.write().await = init
            .prompt_turn_id
            .clone()
            .unwrap_or_else(|| Uuid::new_v4().to_string());
        *last_uploaded_hash.lock().await = None;
        let submit = adapter
            .lock()
            .await
            .write_input(backend.as_ref(), &init.prompt)
            .await?;
        if let Some(cli_session_id) = submit.cli_session_id {
            send_message(&stdout, &WorkerToDaemon::CliSessionId { cli_session_id }).await?;
        }
        if !submit.submitted {
            let message = submit
                .failure_reason
                .unwrap_or_else(|| "CLI submit could not be confirmed".to_string());
            send_message(&stdout, &WorkerToDaemon::UserNotify { message }).await?;
        }
    }

    // Init-time transcript source resolution for adapters with resolvable
    // sources (opencode): resolve cli_session_id before the message loop.
    {
        let resolution = {
            let mut adapter_guard = adapter.lock().await;
            let resolution = adapter_guard
                .resolve_transcript_source(backend.as_ref())
                .await;
            drop(adapter_guard);
            resolution
        };
        if let Some(resolution) = resolution {
            let outcome = resolution.unwrap_or_else(|err| {
                warn!(session = %init.session_id, adapter = %init.cli_id, "resolve_transcript_source error: {:?}", err);
                ResolveOutcome::NotFound {
                    reason: format!("transcript source resolution failed: {}", err),
                }
            });
            match outcome {
                ResolveOutcome::Found(source) => {
                    send_message(
                        &stdout,
                        &WorkerToDaemon::CliSessionId {
                            cli_session_id: source.session_id.clone(),
                        },
                    )
                    .await?;
                    // Also set in adapter state for subsequent poll/write_input.
                    adapter
                        .lock()
                        .await
                        .set_transcript_source(&source.session_id);
                    info!(
                        session = %init.session_id, adapter = %init.cli_id,
                        transcript_session = %source.session_id,
                        "transcript source resolved automatically"
                    );
                }
                ResolveOutcome::Ambiguous { candidates, .. } => {
                    info!(
                        session = %init.session_id, adapter = %init.cli_id,
                        candidate_count = candidates.len(),
                        "transcript source ambiguous, requesting user choice"
                    );
                    let turn_id = Uuid::new_v4().to_string();
                    let choices: Vec<TranscriptChoice> = candidates
                        .iter()
                        .map(|c| TranscriptChoice {
                            session_id: c.session_id.clone(),
                            label: format!("{} ({})", c.session_id, c.db_path.display()),
                        })
                        .collect();
                    send_message(
                        &stdout,
                        &WorkerToDaemon::TranscriptChoices {
                            candidates: choices,
                            turn_id: turn_id.clone(),
                        },
                    )
                    .await?;
                }
                ResolveOutcome::NotFound { reason } => {
                    info!(session = %init.session_id, adapter = %init.cli_id, "transcript source not found: {}", reason);
                    send_message(&stdout, &WorkerToDaemon::UserNotify { message: reason }).await?;
                }
            }
        }
    }

    // Shared "currently processing" stamp: Some((description, start_ms))
    // while the message loop is busy handling one daemon message. Drives
    // both the processing watchdog (WARN past the threshold) and the
    // heartbeat IPC, so the daemon can tell "worker dead" apart from
    // "worker stuck on a message".
    let processing_since: Arc<StdMutex<Option<(String, u64)>>> = Arc::new(StdMutex::new(None));

    // stdin reader runs on a dedicated OS thread: while the message loop is
    // busy handling one message, subsequent daemon messages keep being
    // drained from the pipe into this channel instead of piling up in the
    // kernel pipe buffer.
    let (daemon_msg_tx, mut daemon_msg_rx) = mpsc::channel::<DaemonToWorker>(32);
    std::thread::spawn(move || {
        use std::io::BufRead as _;
        let stdin = std::io::stdin();
        for line in stdin.lock().lines() {
            let line = match line {
                Ok(line) => line,
                Err(_) => break,
            };
            if line.trim().is_empty() {
                continue;
            }
            match serde_json::from_str::<DaemonToWorker>(&line) {
                Ok(msg) => {
                    if daemon_msg_tx.blocking_send(msg).is_err() {
                        break;
                    }
                }
                Err(err) => warn!("failed to parse daemon message: {}", err),
            }
        }
    });

    // Processing watchdog: WARN when one daemon message is being handled
    // longer than the threshold (e.g. a wedged write_input), so the stuck
    // message is directly locatable in the log.
    const MESSAGE_PROCESSING_WARN_THRESHOLD: Duration = Duration::from_secs(120);
    {
        let watchdog_processing = processing_since.clone();
        let watchdog_session_id = init.session_id.clone();
        worker_joins.spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(15)).await;
                let stuck = {
                    let guard = watchdog_processing.lock().unwrap();
                    guard.as_ref().and_then(|(desc, start_ms)| {
                        let elapsed = now_ms().saturating_sub(*start_ms);
                        (elapsed > MESSAGE_PROCESSING_WARN_THRESHOLD.as_millis() as u64)
                            .then(|| (desc.clone(), elapsed))
                    })
                };
                if let Some((desc, elapsed_ms)) = stuck {
                    warn!(
                        session = %watchdog_session_id,
                        message = %desc,
                        elapsed_ms,
                        "daemon message processing exceeded {}s threshold",
                        MESSAGE_PROCESSING_WARN_THRESHOLD.as_secs()
                    );
                }
            }
        });
    }

    // Heartbeat: independent of the message loop; lets the daemon
    // distinguish a dead worker (no heartbeat) from a stuck one (heartbeat
    // keeps coming with processing_since_ms set).
    {
        let hb_stdout = stdout.clone();
        let hb_processing = processing_since.clone();
        worker_joins.spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(10)).await;
                let since = hb_processing
                    .lock()
                    .unwrap()
                    .as_ref()
                    .map(|(_, start)| *start);
                if send_message(
                    &hb_stdout,
                    &WorkerToDaemon::Heartbeat {
                        processing_since_ms: since,
                    },
                )
                .await
                .is_err()
                {
                    break;
                }
            }
        });
    }

    while let Some(msg) = daemon_msg_rx.recv().await {
        *processing_since.lock().unwrap() = Some((daemon_message_desc(&msg), now_ms()));
        match msg {
            DaemonToWorker::Message { content, turn_id } => {
                info!(session = %init.session_id, %turn_id, "Message received, sending TurnStarted to coordinator");
                handle_tui_prompt_override(&stdout, &analyzer_runtime).await;
                let snapshot = latest_raw_screen.read().await.clone();
                usage_limit_tracker.lock().await.begin_turn(
                    &snapshot,
                    SystemTime::now()
                        .duration_since(UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_millis() as u64,
                );
                *current_turn_id.write().await = turn_id.clone();
                *last_uploaded_hash.lock().await = None;
                // Notify coordinator: a new turn has started (best-effort).
                if trigger_tx
                    .send(Trigger::TurnStarted { turn_id })
                    .await
                    .is_err()
                {
                    warn!("coordinator channel closed, TurnStarted not sent");
                }
                let submit = adapter
                    .lock()
                    .await
                    .write_input(backend.as_ref(), &content)
                    .await?;
                if let Some(cli_session_id) = submit.cli_session_id {
                    send_message(&stdout, &WorkerToDaemon::CliSessionId { cli_session_id }).await?;
                }
                if !submit.submitted {
                    let message = submit
                        .failure_reason
                        .unwrap_or_else(|| "CLI submit could not be confirmed".to_string());
                    send_message(&stdout, &WorkerToDaemon::UserNotify { message }).await?;
                }
            }
            DaemonToWorker::RawInput { content, turn_id } => {
                info!(session = %init.session_id, %turn_id, "RawInput received, sending TurnStarted to coordinator");
                handle_tui_prompt_override(&stdout, &analyzer_runtime).await;
                let snapshot = latest_raw_screen.read().await.clone();
                usage_limit_tracker.lock().await.begin_turn(
                    &snapshot,
                    SystemTime::now()
                        .duration_since(UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_millis() as u64,
                );
                *current_turn_id.write().await = turn_id.clone();
                *last_uploaded_hash.lock().await = None;
                // Notify coordinator: a new turn has started (best-effort).
                if trigger_tx
                    .send(Trigger::TurnStarted { turn_id })
                    .await
                    .is_err()
                {
                    warn!("coordinator channel closed, TurnStarted not sent");
                }
                backend.raw_input(&content).await?;
            }
            DaemonToWorker::Close => {
                backend.destroy_session().await?;
                break;
            }
            DaemonToWorker::Restart => {
                backend.destroy_session().await?;
                break;
            }
            DaemonToWorker::RefreshScreen => {
                let screen = backend.capture_viewport().await?;
                *latest_raw_screen.write().await = screen.clone();
                let mode = *display_mode.read().await;
                let rendered = render_screen_for_display_mode(&screen, mode);
                let now_ms = now_ms();
                let analyzing = analyzer_runtime.read().await.is_analyzing;
                let base_status = if analyzing {
                    ScreenStatus::Analyzing
                } else {
                    ScreenStatus::Working
                };
                let (status, usage_limit) =
                    usage_limit_tracker
                        .lock()
                        .await
                        .classify(&screen, base_status, now_ms);
                *latest_screen.write().await = rendered.clone();
                let _ = updates.send(rendered.clone());
                send_message(
                    &stdout,
                    &WorkerToDaemon::ScreenUpdate {
                        content: rendered.clone(),
                        status,
                        usage_limit: usage_limit.clone(),
                    },
                )
                .await?;
                let rendered_hash = lower_hex(&Sha256::digest(rendered.as_bytes()));
                *last_broadcast_hash.lock().await = Some(rendered_hash);
                // Notify coordinator: refresh request (best-effort).
                if trigger_tx.send(Trigger::Refresh).await.is_err() {
                    warn!("coordinator channel closed, Refresh not sent");
                }
            }
            DaemonToWorker::SetDisplayMode { mode } => {
                *display_mode.write().await = mode;
                let raw = latest_raw_screen.read().await.clone();
                let rendered = render_screen_for_display_mode(&raw, mode);
                let now_ms = now_ms();
                let analyzing = analyzer_runtime.read().await.is_analyzing;
                let base_status = if analyzing {
                    ScreenStatus::Analyzing
                } else {
                    ScreenStatus::Working
                };
                let (status, usage_limit) =
                    usage_limit_tracker
                        .lock()
                        .await
                        .classify(&raw, base_status, now_ms);
                *latest_screen.write().await = rendered.clone();
                let _ = updates.send(rendered.clone());
                send_message(
                    &stdout,
                    &WorkerToDaemon::ScreenUpdate {
                        content: rendered,
                        status,
                        usage_limit: usage_limit.clone(),
                    },
                )
                .await?;
                // Notify coordinator: display mode changed (best-effort).
                if trigger_tx
                    .send(Trigger::SetDisplayMode(mode))
                    .await
                    .is_err()
                {
                    warn!("coordinator channel closed, SetDisplayMode not sent");
                }
            }
            DaemonToWorker::TermAction { key } => {
                let keys = term_action_keys(key);
                backend.send_special_keys(&keys).await?;
            }
            DaemonToWorker::SpecialKeys { keys } => {
                backend.send_special_keys(&keys).await?;
            }
            DaemonToWorker::TuiKeys { keys, is_final } => {
                handle_tui_keys(&backend, &analyzer_runtime, &keys, is_final).await?;
            }
            DaemonToWorker::TuiTextInput { keys, text } => {
                handle_tui_text_input(&backend, &adapter, &analyzer_runtime, &keys, &text).await?;
            }
            DaemonToWorker::SetTranscriptSource { cli_session_id } => {
                let applied = adapter.lock().await.set_transcript_source(&cli_session_id);
                if applied {
                    info!(session = %init.session_id, adapter = %init.cli_id, transcript_session = %cli_session_id, "transcript source set by user");
                    send_message(&stdout, &WorkerToDaemon::CliSessionId { cli_session_id }).await?;
                }
            }
            DaemonToWorker::Init(_) => {}
        }
        *processing_since.lock().unwrap() = None;
    }

    worker_joins.abort_all();
    while worker_joins.join_next().await.is_some() {}
    let _ = backend.kill().await;
    if let Some(marker) = cli_pid_marker {
        let _ = tokio::fs::remove_file(marker).await;
    }
    info!("worker exiting");
    Ok(())
}

/// Short human-readable label for a daemon message, used by the processing
/// watchdog and heartbeat stamp (avoids dumping full message contents).
fn daemon_message_desc(msg: &DaemonToWorker) -> String {
    match msg {
        DaemonToWorker::Message { turn_id, .. } => format!("Message(turn_id={})", turn_id),
        DaemonToWorker::RawInput { turn_id, .. } => format!("RawInput(turn_id={})", turn_id),
        DaemonToWorker::Close => "Close".to_string(),
        DaemonToWorker::Restart => "Restart".to_string(),
        DaemonToWorker::SetDisplayMode { .. } => "SetDisplayMode".to_string(),
        DaemonToWorker::TermAction { .. } => "TermAction".to_string(),
        DaemonToWorker::SpecialKeys { .. } => "SpecialKeys".to_string(),
        DaemonToWorker::TuiKeys { .. } => "TuiKeys".to_string(),
        DaemonToWorker::TuiTextInput { .. } => "TuiTextInput".to_string(),
        DaemonToWorker::RefreshScreen => "RefreshScreen".to_string(),
        DaemonToWorker::SetTranscriptSource { .. } => "SetTranscriptSource".to_string(),
        DaemonToWorker::Init(_) => "Init".to_string(),
    }
}

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

    #[test]
    fn daemon_message_desc_labels_variants_without_content() {
        let msg = DaemonToWorker::Message {
            content: "secret prompt".to_string(),
            turn_id: "t-1".to_string(),
        };
        let desc = daemon_message_desc(&msg);
        assert!(desc.contains("Message"));
        assert!(desc.contains("t-1"));
        assert!(!desc.contains("secret prompt"));
        assert_eq!(daemon_message_desc(&DaemonToWorker::Close), "Close");
        assert_eq!(
            daemon_message_desc(&DaemonToWorker::RefreshScreen),
            "RefreshScreen"
        );
    }

    #[test]
    fn cli_exit_requires_consecutive_dead_samples() {
        assert_eq!(CLI_DEAD_CONFIRM_ATTEMPTS, 3);
        let mut dead = 0;
        assert!(!should_emit_cli_exit(&mut dead, false));
        assert_eq!(dead, 1);
        assert!(!should_emit_cli_exit(&mut dead, false));
        assert_eq!(dead, 2);
        assert!(should_emit_cli_exit(&mut dead, false));
        assert_eq!(dead, 3);
        assert!(should_emit_cli_exit(&mut dead, false));
        assert_eq!(dead, 4);
    }

    #[test]
    fn cli_exit_counter_resets_on_alive() {
        let mut dead = 0;
        assert!(!should_emit_cli_exit(&mut dead, false));
        assert!(!should_emit_cli_exit(&mut dead, false));
        assert!(!should_emit_cli_exit(&mut dead, true));
        assert_eq!(dead, 0);
        assert!(!should_emit_cli_exit(&mut dead, false));
        assert_eq!(dead, 1);
    }
}