autofork-daemon 0.26.2

//! Shared daemon state and Claude Code event handling.
//!
//! Since v0.5 the daemon is a pure scheduler: it never spawns fork
//! subprocesses. The asyncRewake Stop hook long-polls via [`handle_stop_wait`];
//! when forks come due the daemon answers with a wake payload the session's own
//! model acts on (spawning `fork` subagents). Fast events (SessionStart,
//! PromptSubmit, SessionEnd) just keep session bookkeeping — and PromptSubmit /
//! SessionEnd cancel any parked stop-wait.

use crate::planner::GateHold;
use autofork_core::config::{load_config_at, Config, Paths};
use autofork_core::moments::{idle_deadlines, resolve_context_window, ForkMoment};
use autofork_core::protocol::{Event, EventKind, ResponseBody};
use autofork_core::store::SessionRow;
use autofork_core::store::{SessionStatus, Store};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicI64, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::sync::oneshot;

/// Cap on the detail one pending external trigger accumulates (bytes): the
/// changed-path list of a `changed:` trigger, or an `emit` payload. A burst
/// that outruns its consumer must not grow a store row without limit; the
/// paths are a hint for the command, not the authoritative diff.
pub const EXTERNAL_DETAIL_CAP: usize = 16_000;

pub fn now() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// Every fork moment that has elapsed for a session by `up_to`: the context
/// gauge (if known, always "elapsed" the instant the turn ended), every idle
/// deadline whose fire time (`base + d`) has passed, and the wall-clock tick
/// `every:` triggers are matched against (always present — per-fork interval
/// math happens in selection, where the fork's last run is known).
/// `pause_started_at` is `None` on a busy (mid-run) poll; on idle polls it
/// gates `every:` to at most one fire per quiet stretch — a fork whose last
/// run is inside the current pause has seen no activity since, so its
/// interval must not turn a quiet session into a periodic cron.
fn elapsed_moments(
    prompt_tokens: Option<u64>,
    max_tokens: u64,
    base: i64,
    deadlines: &[u64],
    up_to: i64,
    pause_started_at: Option<i64>,
) -> Vec<ForkMoment> {
    let mut moments = Vec::new();
    if let Some(pt) = prompt_tokens {
        moments.push(ForkMoment::Context {
            prompt_tokens: pt,
            max_tokens: Some(max_tokens),
        });
    }
    for &d in deadlines {
        if base + d as i64 <= up_to {
            moments.push(ForkMoment::Idle { deadline_secs: d });
        }
    }
    moments.push(ForkMoment::Tick {
        now: up_to,
        pause_started_at,
    });
    moments
}

pub struct Daemon {
    pub paths: Paths,
    pub store: Mutex<Store>,
    /// Per-session cancellation channels for parked stop-wait long polls.
    /// Sending `()` (or dropping) resolves the parked poll as `Waited`.
    pub waits: Mutex<HashMap<String, oneshot::Sender<()>>>,
    /// When we last issued a wake for a session — used to treat an ambiguous
    /// (prompt-less) PromptSubmit shortly after a wake as a non-waking
    /// continuation (the daemon-side belt).
    pub wake_issued_at: Mutex<HashMap<String, i64>>,
    /// When we last processed a chain continue for an opencode session. The
    /// plugin injects the chain report as a real turn and flags it `waking:
    /// false` — but only the instance that injected it. A duplicated event
    /// stream (a second plugin instance, or opencode's own duplicated
    /// session loops) reports the same turn as genuine activity, which resets
    /// the pause counters the chain limit depends on — the observed runaway.
    /// A `waking: true` PromptSubmit inside the grace window after a chain
    /// continue is downgraded to non-waking.
    pub chain_continued_at: Mutex<HashMap<String, i64>>,
    /// Per-session re-evaluation signal. A parked poll normally waits on
    /// timers it computed when it parked — which is right for every trigger
    /// derived from the session's own lifecycle, and wrong for the external
    /// ones: a watched file changing, or `autofork emit`, happens on the
    /// outside world's schedule. Bumping a session's channel wakes its parked
    /// poll to look again, without resolving it (that is `waits`' job).
    /// Senders live in an `Arc` so a parked poll's receiver can never observe
    /// a dropped sender and spin.
    pub nudges: Mutex<HashMap<String, Arc<tokio::sync::watch::Sender<u64>>>>,
    /// Sessions with a currently-parked stop-wait poll (a liveness heartbeat:
    /// the poll's hook subprocess dies with the Claude process). Values are
    /// reference counts, so the entry exists iff a poll is parked.
    pub parked: Mutex<HashMap<String, usize>>,
    /// Sessions with a pending grace-close after a lost poll, keyed to a
    /// generation so any fresh event cancels the close regardless of the
    /// (whole-second) clock granularity.
    pub pending_close: Mutex<HashMap<String, u64>>,
    /// `deliver: context` lifecycle hooks currently running, counted per
    /// spool key. A hook's command runs detached (the event that fired it is
    /// acked immediately), so a client draining the spool right after firing
    /// one would find it empty — the opencode `chat.message` drain waits on
    /// this counter for a bounded moment instead. Only the context lane is
    /// counted: wake blocks leave by another door.
    pub feed_hooks_inflight: Mutex<HashMap<String, usize>>,
    pub close_gen: AtomicU64,
    pub connections: AtomicUsize,
    pub last_busy: AtomicI64,
    /// When this daemon came up. A session with no activity since is one it
    /// inherited from a previous daemon rather than one it watched die.
    pub started_at: i64,
    pub shutdown: tokio::sync::Notify,
}

/// The ceiling on how long a spool drain will wait for in-flight `deliver:
/// context` hooks. A feed's own `timeout:` can be far longer, and a prompt
/// the user already sent must never be held for it: past this, the blocks
/// simply arrive at the next turn, which is where they used to arrive
/// always.
const MAX_FEED_WAIT_MS: u64 = 10_000;

/// How long after issuing a wake an unattributable PromptSubmit — no prompt
/// text, or a task notification the spawn registry can't match — is assumed to
/// be a continuation rather than genuine user activity. Overridable via
/// `AUTOFORK_WAKE_GRACE_SECS` (tests shorten it).
fn wake_grace_secs() -> i64 {
    std::env::var("AUTOFORK_WAKE_GRACE_SECS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(20)
}

/// Clients whose plugin/waiter executes forks natively and delivers reports
/// as turns of the parent session (as opposed to Claude Code, where the
/// session's own model spawns fork subagents and completions arrive as task
/// notifications).
fn is_native_exec_client(client: Option<&str>) -> bool {
    matches!(client, Some("opencode") | Some("codex"))
}

/// How long after a chain continue a `waking: true` PromptSubmit on a
/// native-execution client's session is downgraded to non-waking (the
/// duplicated-event-stream dedupe). Kept short: a genuine user prompt landing
/// inside it merely skips one pause-epoch bump, which the next genuine prompt
/// supplies. Overridable via `AUTOFORK_CHAIN_GRACE_SECS` (tests shorten or
/// zero it).
fn chain_grace_secs() -> i64 {
    std::env::var("AUTOFORK_CHAIN_GRACE_SECS")
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(20)
}

/// After a parked poll drops unanswered, wait this long for a fresh event
/// before closing the session (the Claude process is presumed dead). Overridable
/// via `AUTOFORK_POLL_LOSS_GRACE_MS` (tests shorten it).
fn poll_loss_grace() -> Duration {
    std::env::var("AUTOFORK_POLL_LOSS_GRACE_MS")
        .ok()
        .and_then(|v| v.parse().ok())
        .map(Duration::from_millis)
        .unwrap_or(Duration::from_secs(90))
}

/// RAII marker that a session has a parked stop-wait poll. Increments on
/// creation and decrements on drop — including when the poll future is dropped
/// mid-await (a lost connection), so `parked` stays accurate on every exit path.
pub struct ParkGuard {
    daemon: Arc<Daemon>,
    session_id: String,
}

impl ParkGuard {
    fn new(daemon: &Arc<Daemon>, session_id: &str) -> Self {
        *daemon
            .parked
            .lock()
            .unwrap()
            .entry(session_id.to_string())
            .or_insert(0) += 1;
        Self {
            daemon: daemon.clone(),
            session_id: session_id.to_string(),
        }
    }
}

impl Drop for ParkGuard {
    fn drop(&mut self) {
        let mut parked = self.daemon.parked.lock().unwrap();
        if let Some(c) = parked.get_mut(&self.session_id) {
            *c -= 1;
            if *c == 0 {
                parked.remove(&self.session_id);
            }
        }
    }
}

impl Daemon {
    pub fn new(paths: Paths, store: Store) -> Arc<Self> {
        Arc::new(Self {
            paths,
            store: Mutex::new(store),
            waits: Mutex::new(HashMap::new()),
            nudges: Mutex::new(HashMap::new()),
            wake_issued_at: Mutex::new(HashMap::new()),
            chain_continued_at: Mutex::new(HashMap::new()),
            parked: Mutex::new(HashMap::new()),
            pending_close: Mutex::new(HashMap::new()),
            feed_hooks_inflight: Mutex::new(HashMap::new()),
            close_gen: AtomicU64::new(0),
            connections: AtomicUsize::new(0),
            last_busy: AtomicI64::new(now()),
            started_at: now(),
            shutdown: tokio::sync::Notify::new(),
        })
    }

    pub fn touch_busy(&self) {
        self.last_busy.store(now(), Ordering::SeqCst);
    }

    /// The user-level forks root (`<base>/forks`).
    pub fn user_forks_root(&self) -> PathBuf {
        self.paths.base.join("forks")
    }

    /// The user-level lifecycle-hooks root (`<base>/hooks`).
    pub fn user_hooks_root(&self) -> PathBuf {
        self.paths.base.join("hooks")
    }

    /// The user-level `.claude` directory, whose `forks/` and `skills/`
    /// subdirs are extra discovery roots. `AUTOFORK_CLAUDE_DIR` overrides
    /// (tests use it to keep the real home directory out of fixtures).
    pub fn claude_dir(&self) -> Option<PathBuf> {
        if let Some(dir) = std::env::var_os("AUTOFORK_CLAUDE_DIR") {
            return Some(PathBuf::from(dir));
        }
        std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".claude"))
    }

    /// The user-level `.agents` dir (codex's native skills location; often a
    /// symlink twin of `.claude` — discovery dedupes by canonical path).
    pub fn agents_dir(&self) -> Option<PathBuf> {
        if let Some(dir) = std::env::var_os("AUTOFORK_AGENTS_DIR") {
            return Some(PathBuf::from(dir));
        }
        std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".agents"))
    }

    /// Effective config for a project.
    pub fn cfg_for(&self, project_root: Option<&Path>) -> Config {
        load_config_at(project_root, &self.paths.user_config()).0
    }

    pub fn version() -> &'static str {
        env!("CARGO_PKG_VERSION")
    }

    /// Cancel a parked stop-wait for a session (resolves it as `Waited`).
    /// The session's nudge channel, created on first use.
    fn nudge_channel(&self, session_id: &str) -> Arc<tokio::sync::watch::Sender<u64>> {
        let mut map = self.nudges.lock().unwrap();
        Arc::clone(
            map.entry(session_id.to_string())
                .or_insert_with(|| Arc::new(tokio::sync::watch::channel(0u64).0)),
        )
    }

    /// Ask a session's parked poll (if any) to re-evaluate now: an external
    /// trigger was recorded, or a feed block is waiting to be delivered. A
    /// session with no parked poll needs nothing — the state is in the store,
    /// and its next Stop picks it up.
    pub fn nudge(&self, session_id: &str) {
        if let Some(tx) = self.nudges.lock().unwrap().get(session_id) {
            tx.send_modify(|v| *v = v.wrapping_add(1));
        }
    }

    fn cancel_wait(&self, session_id: &str) {
        if let Some(tx) = self.waits.lock().unwrap().remove(session_id) {
            let _ = tx.send(());
        }
    }

    /// Record that a wake was just issued for a session (grace-window belt).
    pub fn note_wake_issued(&self, session_id: &str) {
        self.wake_issued_at
            .lock()
            .unwrap()
            .insert(session_id.to_string(), now());
    }

    /// Whether a session currently has a parked stop-wait poll.
    pub fn is_parked(&self, session_id: &str) -> bool {
        self.parked.lock().unwrap().contains_key(session_id)
    }

    /// Cancel any pending grace-close for a session (a fresh event proves it is
    /// alive). Called on every event and whenever a new poll parks.
    fn clear_pending_close(&self, session_id: &str) {
        self.pending_close.lock().unwrap().remove(session_id);
    }

    /// A parked poll dropped without the daemon answering it (no Wake, no
    /// Waited): the Claude process likely died. After a grace window, close the
    /// session unless a fresh event cancelled the pending close. A later event
    /// re-opens it via the normal upsert path.
    ///
    /// Note: the asyncRewake hook's own 14400s timeout also drops the poll on a
    /// live-but-long-idle session; the grace-close will close it, and the next
    /// real event re-opens it — acceptable self-correction.
    pub fn on_poll_lost(self: &Arc<Self>, session_id: &str) {
        // No need to wait out the grace when the OS already has the answer:
        // the process the session belongs to is gone.
        if self.harness_gone(session_id) {
            tracing::info!(session = %session_id,
                "stop-wait lost and the client process is gone, closing now");
            self.close_session_firing_hooks(session_id, "gone");
            return;
        }
        let gen = self.close_gen.fetch_add(1, Ordering::SeqCst) + 1;
        self.pending_close
            .lock()
            .unwrap()
            .insert(session_id.to_string(), gen);
        let daemon = self.clone();
        let sid = session_id.to_string();
        let grace = poll_loss_grace();
        tokio::spawn(async move {
            tokio::time::sleep(grace).await;
            // Still the same pending close (no fresh event superseded it)?
            {
                let mut pc = daemon.pending_close.lock().unwrap();
                if pc.get(&sid) != Some(&gen) {
                    return;
                }
                pc.remove(&sid);
            }
            let open = {
                let store = daemon.store.lock().unwrap();
                matches!(store.get_session(&sid), Ok(Some(s)) if s.status == SessionStatus::Open)
            };
            if open {
                tracing::info!(session = %sid, "stop-wait lost, closing session");
                daemon.close_session_firing_hooks(&sid, "lost");
            }
        });
    }

    /// Sessions whose recorded client process no longer exists — the OS-level
    /// answer to "is this session still alive", independent of whether the
    /// client managed to run a `SessionEnd` hook or leave a poll parked.
    /// Sessions with no harness anchor (older clients, opencode) are left to
    /// the poll-loss and timeout paths.
    pub fn dead_harness_sessions(&self) -> Vec<(String, i64)> {
        let store = self.store.lock().unwrap();
        store
            .list_open_sessions()
            .unwrap_or_default()
            .into_iter()
            .filter(|s| s.harness.as_ref().is_some_and(|h| !h.alive()))
            .map(|s| (s.session_id, s.last_activity))
            .collect()
    }

    /// Whether a session's recorded client process is known to be gone.
    pub fn harness_gone(&self, session_id: &str) -> bool {
        let store = self.store.lock().unwrap();
        matches!(store.get_session(session_id), Ok(Some(s))
            if s.harness.as_ref().is_some_and(|h| !h.alive()))
    }

    /// Close a session, firing its `session_end` lifecycle hooks exactly once
    /// (only the call that transitions open → closed fires; racing close
    /// paths — client end, poll loss, harness death, prune, timeout — are
    /// safe). Fork-run sessions close silently. Returns whether this call
    /// closed it.
    ///
    /// `flush_on_close` is honored HERE, not only in the `SessionEnd` hook:
    /// the batch is selected (and stamped) before the close purges the roster,
    /// and handed to a detached end-runner. The hook path stamps the same
    /// forks through `TakeFinalRuns`, so whichever side reaches them first
    /// runs them exactly once — and a close the client never reported still
    /// gets its consolidation forks.
    pub fn close_session_firing_hooks(self: &Arc<Self>, session_id: &str, reason: &str) -> bool {
        self.close_session_with_flush(session_id, reason, true)
    }

    /// [`close_session_firing_hooks`] with the flush made explicit. Passing
    /// `false` closes without running the flush-on-close batch: for a session
    /// this daemon never saw alive (a dead row inherited at startup), the
    /// forks would be consolidating a conversation that ended who-knows-when.
    pub fn close_session_with_flush(
        self: &Arc<Self>,
        session_id: &str,
        reason: &str,
        flush: bool,
    ) -> bool {
        let row_before = {
            let store = self.store.lock().unwrap();
            store.get_session(session_id).ok().flatten()
        };
        let final_runs = match row_before.as_ref() {
            Some(r) if flush && r.status == SessionStatus::Open => {
                crate::flush::take_final_runs_for_close(self, r, reason)
            }
            _ => Vec::new(),
        };
        let (row, transitioned) = {
            let store = self.store.lock().unwrap();
            let row = store.get_session(session_id).ok().flatten();
            let transitioned = store
                .close_session(session_id, reason, now())
                .unwrap_or(false);
            (row, transitioned)
        };
        // Specs we stamped are ours to run even if a racing closer won the
        // transition: the session is closed either way, and dropping them
        // here would silently lose the batch.
        if let Some(r) = row_before.as_ref() {
            crate::flush::spawn_end_runner(self, r, &final_runs);
        }
        if !transitioned {
            return false;
        }
        let Some(row) = row else { return true };
        if self.is_fork_run_session(session_id) {
            return true;
        }
        crate::hooks::fire_matching(
            self,
            &crate::hooks::HookCtx::from_row(&row),
            crate::hooks::HookEvent::SessionEnd { reason },
        );
        true
    }

    /// Whether a wake was issued for this session within the grace window.
    fn recently_woke(&self, session_id: &str, t: i64) -> bool {
        self.wake_issued_at
            .lock()
            .unwrap()
            .get(session_id)
            .is_some_and(|&at| t - at < wake_grace_secs())
    }

    /// Record that a chain continue was just processed for a session (the
    /// duplicated-event-stream dedupe window).
    fn note_chain_continued(&self, session_id: &str) {
        self.chain_continued_at
            .lock()
            .unwrap()
            .insert(session_id.to_string(), now());
    }

    /// Whether a chain continue was processed for this session within the
    /// grace window.
    fn recently_chain_continued(&self, session_id: &str, t: i64) -> bool {
        self.chain_continued_at
            .lock()
            .unwrap()
            .get(session_id)
            .is_some_and(|&at| t - at < chain_grace_secs())
    }

    /// Whether this id names one of our own fork-run sessions. Such ids must
    /// never be registered or scheduled: a fork-run session that slips past
    /// the plugin's eligibility check (lost title marker, duplicate plugin
    /// instance, event race at creation) would otherwise become a scheduled
    /// session whose idle forks fork it again — forks breeding forks.
    pub(crate) fn is_fork_run_session(&self, id: &str) -> bool {
        let store = self.store.lock().unwrap();
        store.is_fork_run_ref(id).unwrap_or(false)
    }

    /// Handle one fast lifecycle event; returns the response body.
    pub async fn handle_event(self: &Arc<Self>, ev: Event) -> ResponseBody {
        self.touch_busy();
        // Lifecycle events for a fork-run session are dropped (SessionEnd is
        // let through: it only closes a row, cleaning up after a session that
        // was registered before its spawn frame landed).
        if ev.event != EventKind::SessionEnd && self.is_fork_run_session(&ev.session_id) {
            tracing::info!(session = %ev.session_id, kind = ?ev.event,
                "ignoring event for a fork-run session");
            return ResponseBody::Ack;
        }
        // A fresh event proves the session is alive: cancel any pending
        // lost-poll close.
        self.clear_pending_close(&ev.session_id);
        let t = now();
        let enable_tags = ev.enable_tags.as_ref().map(|v| v.join(","));
        let disable_tags = ev.disable_tags.as_ref().map(|v| v.join(","));
        match ev.event {
            EventKind::SessionStart => {
                let newly_opened = {
                    let store = self.store.lock().unwrap();
                    let newly = store
                        .upsert_session(
                            &ev.session_id,
                            &ev.project_root,
                            &ev.cwd,
                            ev.transcript_path.as_deref(),
                            ev.model.as_deref(),
                            enable_tags.as_deref(),
                            disable_tags.as_deref(),
                            ev.client.as_deref(),
                            t,
                        )
                        .unwrap_or(false);
                    if let Some(w) = ev.context_window {
                        let _ = store.set_context_window(&ev.session_id, w);
                    }
                    if let Some(h) = ev.harness.as_ref() {
                        let _ = store.set_harness(&ev.session_id, h);
                    }
                    newly
                };
                if newly_opened {
                    crate::hooks::fire_matching(
                        self,
                        &crate::hooks::HookCtx::from_event(&ev),
                        crate::hooks::HookEvent::SessionStart {
                            source: ev.source.as_deref(),
                        },
                    );
                }
                ResponseBody::Ack
            }
            EventKind::PromptSubmit => {
                // Is this genuine user activity, or a non-waking continuation?
                // An asyncRewake wake reminder sniffs on its marker (the CLI's
                // `waking` field). A task notification is only a continuation
                // when it reports one of the daemon's own fork spawns — any
                // other background task finishing is the session picking real
                // work back up, so it must start a new pause (otherwise idle
                // forks stay latched to the old one and never fire again). The
                // post-wake grace window remains the belt for notifications
                // the spawn registry can't vouch for either way (e.g. a fork
                // that completed before its spawn's Stop was ever ingested).
                let waking = if ev.notif_tool_use_id.is_some() || ev.notif_task_id.is_some() {
                    // Refresh the spawn registry from the transcript BEFORE
                    // classifying: the spawn's tool_use is always on disk by
                    // the time its completion notification is delivered, but
                    // the last Stop's ingest may predate it (observed live: a
                    // Stop racing the transcript flush — or no Stop-wait read
                    // at all between spawn and completion — left the registry
                    // empty, misclassified the fork's own completion as
                    // foreign activity, and re-fired the idle fork forever
                    // after, once per fork run).
                    self.ingest_transcript(&ev);
                    let status = ev.notif_status.as_deref().unwrap_or("");
                    let (matched, transitioned) = {
                        let store = self.store.lock().unwrap();
                        if autofork_core::notification::is_terminal_status(status) {
                            store
                                .mark_spawn_terminal(
                                    &ev.session_id,
                                    ev.notif_tool_use_id.as_deref(),
                                    ev.notif_task_id.as_deref(),
                                    status,
                                    t,
                                )
                                .unwrap_or((false, false))
                        } else {
                            let matched = store
                                .is_fork_spawn(
                                    &ev.session_id,
                                    ev.notif_tool_use_id.as_deref(),
                                    ev.notif_task_id.as_deref(),
                                )
                                .unwrap_or(false);
                            (matched, false)
                        }
                    };
                    // One of our own fork runs just settled: handle chain
                    // re-arm (report ended with the sentinel) and gate
                    // release. The `transitioned` edge fires once even though
                    // the same notification is also seen in the transcript
                    // delta.
                    if transitioned {
                        let fork = {
                            let store = self.store.lock().unwrap();
                            store
                                .spawn_fork_name(
                                    &ev.session_id,
                                    ev.notif_tool_use_id.as_deref(),
                                    ev.notif_task_id.as_deref(),
                                )
                                .unwrap_or(None)
                        };
                        if let Some(fork) = fork {
                            self.on_own_fork_terminal(
                                &ev.session_id,
                                &fork,
                                status,
                                ev.notif_continue == Some(true),
                            );
                        }
                    }
                    !matched && !self.recently_woke(&ev.session_id, t)
                } else {
                    ev.waking
                        .unwrap_or_else(|| !self.recently_woke(&ev.session_id, t))
                };
                // Duplicated-event-stream dedupe: the turn a chain report
                // injection starts is flagged non-waking only by the plugin
                // instance that injected it. A second observer of the same
                // session (another plugin instance, or opencode's own
                // duplicated loops) reports that same turn as genuine
                // activity — bumping the pause epoch, which re-arms every
                // idle fork and resets the per-pause chain limit, turning a
                // goal fork into a self-sustaining pump. Any waking
                // PromptSubmit for a native-execution client's session
                // (opencode, codex — never Claude Code, whose completions
                // are task notifications) inside the chain grace window is
                // downgraded to non-waking.
                let waking = if waking
                    && is_native_exec_client(ev.client.as_deref())
                    && self.recently_chain_continued(&ev.session_id, t)
                {
                    tracing::info!(session = %ev.session_id,
                        "waking prompt inside the chain grace window — \
                         treating it as the chain's own injected turn");
                    false
                } else {
                    waking
                };
                let newly_opened = {
                    let store = self.store.lock().unwrap();
                    let newly = store
                        .upsert_session(
                            &ev.session_id,
                            &ev.project_root,
                            &ev.cwd,
                            ev.transcript_path.as_deref(),
                            ev.model.as_deref(),
                            enable_tags.as_deref(),
                            disable_tags.as_deref(),
                            ev.client.as_deref(),
                            t,
                        )
                        .unwrap_or(false);
                    let _ = store.set_last_activity(&ev.session_id, t);
                    if let Some(h) = ev.harness.as_ref() {
                        let _ = store.set_harness(&ev.session_id, h);
                    }
                    // Genuine activity begins a new pause: advance the epoch
                    // (releasing per-pause idle latches), reset the baseline,
                    // and drop any dependents still held for the old moment
                    // (their pause is over; they re-select on the next one).
                    if waking {
                        let _ = store.bump_pause_epoch(&ev.session_id);
                        if let Ok(n) = store.clear_pending_deps(&ev.session_id) {
                            if n > 0 {
                                tracing::info!(
                                    session = %ev.session_id,
                                    dropped = n,
                                    "user activity dropped held dependents"
                                );
                            }
                        }
                    }
                    newly
                };
                let ctx = crate::hooks::HookCtx::from_event(&ev);
                // A session first seen mid-life (daemon restart, wiped state)
                // still gets its session_start edge before the activity one.
                if newly_opened {
                    crate::hooks::fire_matching(
                        self,
                        &ctx,
                        crate::hooks::HookEvent::SessionStart { source: None },
                    );
                }
                if waking {
                    crate::hooks::fire_matching(self, &ctx, crate::hooks::HookEvent::Activity);
                }
                // A turn is in flight either way: cancel any parked stop-wait so
                // no wake fires mid-turn.
                self.cancel_wait(&ev.session_id);
                ResponseBody::Ack
            }
            EventKind::SessionEnd => {
                self.cancel_wait(&ev.session_id);
                self.close_session_firing_hooks(
                    &ev.session_id,
                    ev.reason.as_deref().unwrap_or("ended"),
                );
                ResponseBody::Ack
            }
            // Stop never arrives as a plain event (it is a StopWait long poll).
            EventKind::Stop => ResponseBody::Ack,
        }
    }

    /// The asyncRewake Stop hook's long poll: record activity + the context
    /// gauge, then wait until forks come due (returning a `Wake`) or the wait
    /// is cancelled / the daemon retires (returning `Waited`).
    pub async fn handle_stop_wait(self: &Arc<Self>, ev: Event) -> ResponseBody {
        self.touch_busy();
        // Never park a poll for (or schedule forks on) one of our own
        // fork-run sessions — the breeding-loop guard. Answer Waited so a
        // confused plugin's poll resolves instead of hanging.
        if self.is_fork_run_session(&ev.session_id) {
            tracing::info!(session = %ev.session_id,
                "refusing to schedule a fork-run session");
            return ResponseBody::Waited;
        }
        // A poll from a process whose client is already dead is an orphan: a
        // stop-wait hook that outlived the client that spawned it. Parking it
        // would re-open the session (the upsert below) and keep it "alive"
        // forever on a heartbeat nobody is behind. Close instead, and answer
        // Waited so the orphan exits.
        if let Some(h) = ev.harness.as_ref() {
            if !h.alive() {
                tracing::info!(session = %ev.session_id, pid = h.pid,
                    "stop-wait from an orphaned poll: client process is gone");
                self.close_session_firing_hooks(&ev.session_id, "gone");
                return ResponseBody::Waited;
            }
        }
        // A new poll parking proves the session is alive.
        self.clear_pending_close(&ev.session_id);
        let t = now();
        let enable_tags = ev.enable_tags.as_ref().map(|v| v.join(","));
        let disable_tags = ev.disable_tags.as_ref().map(|v| v.join(","));
        let newly_opened = {
            let store = self.store.lock().unwrap();
            let newly = store
                .upsert_session(
                    &ev.session_id,
                    &ev.project_root,
                    &ev.cwd,
                    ev.transcript_path.as_deref(),
                    ev.model.as_deref(),
                    enable_tags.as_deref(),
                    disable_tags.as_deref(),
                    ev.client.as_deref(),
                    t,
                )
                .unwrap_or(false);
            let _ = store.set_last_activity(&ev.session_id, t);
            if let Some(w) = ev.context_window {
                let _ = store.set_context_window(&ev.session_id, w);
            }
            if let Some(h) = ev.harness.as_ref() {
                let _ = store.set_harness(&ev.session_id, h);
            }
            newly
        };
        // A session first seen at a Stop (daemon restart mid-session) still
        // gets its session_start lifecycle-hook edge.
        if newly_opened {
            crate::hooks::fire_matching(
                self,
                &crate::hooks::HookCtx::from_event(&ev),
                crate::hooks::HookEvent::SessionStart { source: None },
            );
        }
        // Clients that track usage themselves (opencode) report the gauge on
        // the event; otherwise it comes from the transcript delta.
        let prompt_tokens = if let Some(gauge) = ev.context_tokens {
            let store = self.store.lock().unwrap();
            let _ = store.set_prompt_tokens(&ev.session_id, gauge);
            Some(gauge)
        } else {
            self.ingest_transcript(&ev)
        };
        let cfg = self.cfg_for(Some(&ev.project_root));

        // A busy poll must not start a pause or arm idle deadlines — the
        // session isn't pausing, it's working. Two ways to be busy: the
        // client says so (opencode parks a poll mid-run so `every:`/context
        // triggers can still fire), or the session stopped while waiting on
        // background work it started — a `run_in_background` command, a
        // background subagent, a Monitor. The second case is a *stop*, so
        // Claude Code would otherwise call it idle; instead each fork
        // decides what idle means to it (`background_hold:`, else the config
        // default): the ones that wait are held at selection until the last
        // of that work clears (or the hold times out), the ones that don't
        // fire from this stop like any other.
        let client_busy = ev.busy.unwrap_or(false);
        let waiting = !client_busy && self.pending_background(&ev.session_id, &cfg, t);
        if waiting {
            tracing::info!(session = %ev.session_id,
                "stop with background work still running: forks that wait for it are held");
        }
        let busy = client_busy || (waiting && cfg.background_hold);
        // The first Stop of a pause sets the baseline; a wake-turn's own Stop
        // keeps the existing one, so idle deadlines don't reset. A stop that
        // is merely waiting on background work sets it too: forks that don't
        // wait measure from here, and the held ones get a fresh baseline when
        // the work's completion starts a new pause.
        if !client_busy {
            let store = self.store.lock().unwrap();
            let _ = store.set_pause_started_at_if_unset(&ev.session_id, t);
        }

        // Register this wait so PromptSubmit / SessionEnd can cancel it. A
        // stale wait for the same session (if any) is cancelled by the insert.
        let (tx, mut rx) = oneshot::channel::<()>();
        if let Some(old) = self.waits.lock().unwrap().insert(ev.session_id.clone(), tx) {
            let _ = old.send(());
        }
        // Mark the session as having a live parked poll (a liveness heartbeat).
        // The guard is dropped on every exit path, including when this future is
        // dropped because the connection was lost.
        let _park = ParkGuard::new(self, &ev.session_id);

        let Some(session) = ({
            let store = self.store.lock().unwrap();
            store.get_session(&ev.session_id).ok().flatten()
        }) else {
            return ResponseBody::Waited;
        };
        // Feed blocks waiting for this session get the poll first: they are
        // already produced (a command ran, the text exists) and nothing about
        // them needs a moment to be evaluated.
        if let Some(resp) = self.take_feed_delivery(&session) {
            return resp;
        }
        // Held dependents whose predecessors' completions the transcript (or a
        // notification PromptSubmit) just confirmed release right now — this is
        // the Stop that follows the completion's relay turn, so the reports are
        // already in the session's context.
        if let Some((payload, forks)) = crate::planner::release_due(self, &session) {
            return ResponseBody::Wake {
                payload,
                forks: Some(forks),
                feed: None,
            };
        }
        // Idle timing is measured from the pause baseline (the first Stop of
        // this pause), so a wake-turn's own Stop doesn't restart the clock.
        let baseline = session.pause_started_at.unwrap_or(t);
        // Context thresholds are judged against the session's real window: an
        // explicitly reported window (opencode's model catalog) wins; else the
        // hook-reported model id keeps Claude Code's `[1m]` marker (the session
        // row holds the latest non-null value), and an oversized gauge bumps
        // an under-assumed window.
        let max_tokens = resolve_context_window(
            session.model.as_deref(),
            prompt_tokens,
            session.context_window,
        );

        // Idle deadlines (seconds from the baseline) this session's forks
        // need — none on a client-busy poll (the session isn't pausing);
        // while merely waiting on background work every deadline is still
        // armed, and selection holds the forks that wait — plus the
        // absolute instants at which `every:` intervals next elapse.
        let (entries, _) = autofork_core::discovery::discover_forks(
            &session.cwd,
            Some(&self.user_forks_root()),
            self.claude_dir().as_deref(),
            self.agents_dir().as_deref(),
        );
        let deadlines = if client_busy {
            Vec::new()
        } else {
            idle_deadlines(
                entries.iter().map(|e| &e.parsed.def),
                cfg.default_idle_deadline_secs,
            )
        };
        // Idle lifecycle hooks: shell commands that fire once per pause after
        // their deadline, WITHOUT resolving the poll — the session just stays
        // parked (that is their point: "the session went idle but is still
        // open"). None on a busy poll. The gate never holds them: they are
        // infrastructure (leases), not context work.
        let idle_hooks = if busy {
            Vec::new()
        } else {
            let (hook_entries, _) =
                autofork_core::hooks::discover_hooks(&session.cwd, Some(&self.user_hooks_root()));
            crate::hooks::idle_hook_deadlines(&hook_entries, cfg.default_idle_deadline_secs)
        };
        let hook_ctx = crate::hooks::HookCtx::from_row(&session);
        let fire_idle_hooks = |slf: &Arc<Self>, up_to: i64| {
            for (entry, d) in &idle_hooks {
                if baseline + *d as i64 > up_to {
                    continue;
                }
                let fresh = {
                    let store = slf.store.lock().unwrap();
                    store
                        .try_latch_fire(
                            &session.session_id,
                            &format!("hook:{}", entry.name),
                            &format!("hook-idle-pause:{}:{d}", session.pause_epoch),
                            up_to,
                        )
                        .unwrap_or(false)
                };
                if fresh {
                    crate::hooks::execute(
                        slf,
                        &hook_ctx,
                        entry,
                        "idle",
                        vec![("AUTOFORK_IDLE_SECS".to_string(), d.to_string())],
                    );
                }
            }
        };

        let every_times = {
            let ran: std::collections::HashMap<String, Option<i64>> = {
                let store = self.store.lock().unwrap();
                store
                    .roster(&session.session_id)
                    .unwrap_or_default()
                    .into_iter()
                    .map(|e| (e.fork_name, e.ran_at))
                    .collect()
            };
            autofork_core::moments::every_fire_times(
                entries.iter().map(|e| (e.name.as_str(), &e.parsed.def)),
                |name| ran.get(name).copied().flatten(),
                session.created_at,
            )
        };

        // Phase A: find the first instant ≥1 fork is due (read-only eval).
        // Context thresholds are known immediately (the turn just ended);
        // idle forks come due as their deadlines elapse, `every:` intervals
        // at their absolute fire instants.
        // Busy polls carry no pause: `every:` fires freely mid-run. Idle
        // polls carry the pause start, capping `every:` at one fire per
        // quiet stretch.
        let pause_gate = if busy { None } else { Some(baseline) };
        // Everything elapsed right now, the session's own lifecycle moments
        // plus whatever the outside world queued for it since the last look.
        let all_moments = |slf: &Arc<Self>| -> Vec<ForkMoment> {
            let mut moments = elapsed_moments(
                prompt_tokens,
                max_tokens,
                baseline,
                &deadlines,
                now(),
                pause_gate,
            );
            moments.extend(slf.external_moments(&ev.session_id));
            moments
        };
        let due_now = |slf: &Arc<Self>| -> bool {
            let moments = all_moments(slf);
            let mut sel =
                crate::planner::select_forks(slf, &session, &cfg, &moments, GateHold::Apply);
            crate::planner::reserve_fast_path(&session, &mut sel);
            !sel.is_empty()
        };

        let fire_instants: Vec<i64> = {
            let mut v: Vec<i64> = deadlines.iter().map(|&d| baseline + d as i64).collect();
            v.extend(every_times);
            // Idle lifecycle hooks need their own evaluation instants — a
            // hooks-only session would otherwise park with no timer at all.
            v.extend(idle_hooks.iter().map(|(_, d)| baseline + *d as i64));
            // An active gate silences the other idle forks; if its spawn was
            // fumbled, the belt lifts it at issuance + grace — schedule an
            // evaluation there, or a quiet session would never re-check and
            // the held forks would stay silenced for the whole pause.
            if let Some(g) = session.active_gate.as_deref() {
                let issued = {
                    let store = self.store.lock().unwrap();
                    store.last_issued_at(&session.session_id, g).ok().flatten()
                };
                if let Some(at) = issued {
                    v.push(at + crate::planner::gate_grace_secs() + 1);
                }
            }
            // Background work holds the forks that wait for it; if its
            // completion is never observed, each task's hold lapses at
            // `background_hold_timeout` past its start — schedule an
            // evaluation there, or a quiet session would never re-check and
            // the held forks would sleep until the next stop.
            if waiting && cfg.background_hold_timeout_secs > 0 {
                let timeout = cfg.background_hold_timeout_secs as i64;
                let starts = {
                    let store = self.store.lock().unwrap();
                    store
                        .pending_bg_task_starts(&ev.session_id, t - timeout)
                        .unwrap_or_default()
                };
                v.extend(starts.into_iter().map(|at| at + timeout + 1));
            }
            v.sort_unstable();
            v.dedup();
            v
        };
        // Deadlines that elapsed before this poll parked (a re-park after a
        // wake turn) fire their hooks right away; the latch dedupes.
        fire_idle_hooks(self, now());
        let due = due_now(self);
        // External triggers arrive on the outside world's schedule, so the
        // poll can no longer wait only on the timers it computed at park
        // time: it also waits on this session's nudge channel. Holding the
        // sender for the poll's lifetime keeps the receiver from ever seeing
        // a closed channel (which would spin this loop).
        let nudge_tx = self.nudge_channel(&ev.session_id);
        let mut nudge_rx = nudge_tx.subscribe();
        if !due {
            let mut next_instant = 0usize;
            loop {
                // `None` = no deadline left to service; park on the nudge and
                // the cancellations alone, exactly as the old code parked.
                let sleep_for = fire_instants
                    .get(next_instant)
                    .map(|&at| Duration::from_secs((at - now()).max(0) as u64));
                tokio::select! {
                    _ = async {
                        match sleep_for {
                            Some(d) => tokio::time::sleep(d).await,
                            None => std::future::pending::<()>().await,
                        }
                    } => {
                        next_instant += 1;
                        fire_idle_hooks(self, now());
                        if due_now(self) { break; }
                    }
                    _ = nudge_rx.changed() => {
                        if let Some(resp) = self.take_feed_delivery(&session) {
                            return resp;
                        }
                        if due_now(self) { break; }
                    }
                    _ = &mut rx => return ResponseBody::Waited,
                    _ = self.shutdown.notified() => return ResponseBody::Waited,
                }
            }
        }

        // Phase B: debounce so near-simultaneous forks batch into one wake.
        // Cancellation / shutdown during the window wins (nothing is stamped).
        if cfg.wake_debounce_secs > 0 {
            tokio::select! {
                _ = tokio::time::sleep(Duration::from_secs(cfg.wake_debounce_secs)) => {}
                _ = &mut rx => return ResponseBody::Waited,
                _ = self.shutdown.notified() => return ResponseBody::Waited,
            }
        }

        // Phase C: re-evaluate over every moment elapsed by now (deadlines that
        // landed during the debounce join the batch), then issue one wake —
        // stamping throttles and latches at this point.
        let moments = all_moments(self);
        let mut selected =
            crate::planner::select_forks(self, &session, &cfg, &moments, GateHold::Apply);
        crate::planner::reserve_fast_path(&session, &mut selected);
        if let Some((payload, forks)) = crate::planner::build_wake(self, &session, selected) {
            return ResponseBody::Wake {
                payload,
                forks: Some(forks),
                feed: None,
            };
        }
        // Nothing survived re-evaluation; park.
        tokio::select! {
            _ = &mut rx => {}
            _ = self.shutdown.notified() => {}
        }
        ResponseBody::Waited
    }

    /// An opencode fork run started: record it in the spawn registry, keyed by
    /// the fork session id in the `tool_use_id` role. The registry drives
    /// `after`-dependency release and run bookkeeping, same as a Claude Code
    /// spawn observed in the transcript.
    pub fn handle_fork_spawned(
        self: &Arc<Self>,
        session_id: &str,
        fork: &str,
        run_ref: &str,
    ) -> ResponseBody {
        self.touch_busy();
        let store = self.store.lock().unwrap();
        let _ = store.record_spawn(session_id, run_ref, Some(fork), now());
        ResponseBody::Ack
    }

    /// The codex Stop hook's goal fast path: select-and-stamp exactly the
    /// `chain: true` forks due at this pause's first Stop (`idle: 0s`
    /// triggers only) and hand them back for synchronous execution. Non-chain
    /// forks are deliberately not evaluated here — nothing is stamped for
    /// them, so the session's regular parked poll picks them up unchanged.
    /// No debounce: the goal loop wants immediacy.
    pub fn handle_peek_due(self: &Arc<Self>, session_id: &str) -> ResponseBody {
        self.touch_busy();
        let session = {
            let store = self.store.lock().unwrap();
            if store.is_fork_run_ref(session_id).unwrap_or(false) {
                return ResponseBody::Due { forks: Vec::new() };
            }
            store.get_session(session_id).ok().flatten()
        };
        let Some(session) = session else {
            return ResponseBody::Due { forks: Vec::new() };
        };
        let cfg = self.cfg_for(Some(&session.project_root));
        let moments = [autofork_core::moments::ForkMoment::Idle { deadline_secs: 0 }];
        let mut selected =
            crate::planner::select_forks(self, &session, &cfg, &moments, GateHold::Apply);
        selected.retain(|s| s.chain);
        match crate::planner::build_wake(self, &session, selected) {
            Some((_payload, forks)) => ResponseBody::Due { forks },
            None => ResponseBody::Due { forks: Vec::new() },
        }
    }

    /// Spool a headless fork run's report for silent delivery on the
    /// session's next prompt.
    pub fn handle_spool_report(
        self: &Arc<Self>,
        session_id: &str,
        fork: &str,
        text: &str,
    ) -> ResponseBody {
        self.touch_busy();
        let store = self.store.lock().unwrap();
        let _ = store.spool_report(session_id, fork, text, now());
        ResponseBody::Ack
    }

    /// `flush_on_close`: hand the caller every idle fork not yet fired this
    /// pause, stamped, in execution order. Must run BEFORE the session-end
    /// event (close purges the roster).
    pub fn handle_take_final_runs(self: &Arc<Self>, session_id: &str) -> ResponseBody {
        self.touch_busy();
        let session = {
            let store = self.store.lock().unwrap();
            if store.is_fork_run_ref(session_id).unwrap_or(false) {
                return ResponseBody::Due { forks: Vec::new() };
            }
            store.get_session(session_id).ok().flatten()
        };
        let Some(session) = session else {
            return ResponseBody::Due { forks: Vec::new() };
        };
        ResponseBody::Due {
            forks: crate::planner::build_final_runs(self, &session),
        }
    }

    /// Take (and clear) the spooled reports for a session.
    /// The external triggers queued for a session, as moments. Unlike every
    /// other moment these are not computed from the session's state: they
    /// were recorded when the outside world moved and wait in the store until
    /// an evaluation consumes them.
    fn external_moments(&self, session_id: &str) -> Vec<ForkMoment> {
        let store = self.store.lock().unwrap();
        store
            .pending_triggers(session_id)
            .unwrap_or_default()
            .into_iter()
            .filter_map(|(kind, key, _)| {
                autofork_core::moments::ExternalKind::from_label(&kind)
                    .map(|kind| ForkMoment::External { kind, key })
            })
            .collect()
    }

    /// The feed delivery a parked poll should carry right now, if any.
    ///
    /// Two lanes, and which one a client uses is a property of the client,
    /// not of the feed:
    ///
    /// - **wake blocks** (`deliver: wake`) resolve the poll everywhere except
    ///   codex, whose Stop hook runs synchronously and injects them itself
    ///   (block-and-inject beats waking a session that is about to stop).
    /// - **quiet blocks** (`deliver: context`) ride the report spool, which
    ///   Claude Code and codex drain as `additionalContext` at the next
    ///   prompt, and opencode at its next turn (the plugin's `chat.message`
    ///   hook). A session sitting idle has no next turn in sight, so opencode
    ///   also takes them off the poll and injects them as no-reply messages —
    ///   no turn spent either way, and the spool makes either lane deliver
    ///   exactly once.
    fn take_feed_delivery(self: &Arc<Self>, session: &SessionRow) -> Option<ResponseBody> {
        let client = session.client.as_deref();
        let store = self.store.lock().unwrap();
        if client != Some("codex") {
            if let Ok(blocks) = store.take_wake_blocks(&session.session_id) {
                if !blocks.is_empty() {
                    tracing::info!(session = %session.session_id, blocks = blocks.len(),
                        "delivering feed blocks by waking the session");
                    return Some(ResponseBody::Wake {
                        payload: autofork_core::wake::build_feed_wake_payload(&blocks),
                        forks: None,
                        feed: Some(autofork_core::protocol::FeedWake { blocks, wake: true }),
                    });
                }
            }
        }
        if client == Some("opencode") {
            if let Ok(blocks) = store.take_reports(&session.session_id) {
                if !blocks.is_empty() {
                    tracing::info!(session = %session.session_id, blocks = blocks.len(),
                        "handing spooled feed blocks to the opencode plugin");
                    return Some(ResponseBody::Wake {
                        payload: String::new(),
                        forks: None,
                        feed: Some(autofork_core::protocol::FeedWake {
                            blocks,
                            wake: false,
                        }),
                    });
                }
            }
        }
        None
    }

    /// `autofork emit <name>`: record the event for every open session that
    /// could care, fire the hooks listening for it, and nudge the parked
    /// polls so forks waiting on `event: <name>` are evaluated now rather
    /// than at the session's next Stop.
    pub fn handle_emit(
        self: &Arc<Self>,
        name: &str,
        payload: Option<&str>,
        project_root: Option<&Path>,
        session_id: Option<&str>,
    ) -> ResponseBody {
        self.touch_busy();
        let sessions = {
            let store = self.store.lock().unwrap();
            store.list_open_sessions().unwrap_or_default()
        };
        let detail = payload.unwrap_or_default();
        let t = now();
        let mut hit = 0usize;
        for row in sessions {
            if let Some(sid) = session_id {
                if row.session_id != sid {
                    continue;
                }
            }
            if let Some(root) = project_root {
                if !row.project_root.starts_with(root) {
                    continue;
                }
            }
            {
                let store = self.store.lock().unwrap();
                let _ = store.record_pending_trigger(
                    &row.session_id,
                    "event",
                    name,
                    detail,
                    EXTERNAL_DETAIL_CAP,
                    t,
                );
            }
            crate::hooks::fire_external(
                self,
                &crate::hooks::HookCtx::from_row(&row),
                autofork_core::moments::ExternalKind::Event,
                name,
                detail,
            );
            self.nudge(&row.session_id);
            hit += 1;
        }
        tracing::info!(event = %name, sessions = hit, "emit delivered");
        ResponseBody::Emitted { sessions: hit }
    }

    /// Take a session's queued `deliver: wake` blocks (the codex Stop hook's
    /// path — see [`Daemon::take_feed_delivery`]).
    pub fn handle_take_wake_blocks(self: &Arc<Self>, session_id: &str) -> ResponseBody {
        let store = self.store.lock().unwrap();
        let blocks = store.take_wake_blocks(session_id).unwrap_or_default();
        ResponseBody::Reports { blocks }
    }

    pub async fn handle_take_reports(
        self: &Arc<Self>,
        session_id: &str,
        wait_ms: Option<u64>,
    ) -> ResponseBody {
        self.touch_busy();
        if let Some(ms) = wait_ms {
            self.await_feed_hooks(session_id, Duration::from_millis(ms.min(MAX_FEED_WAIT_MS)))
                .await;
        }
        let store = self.store.lock().unwrap();
        ResponseBody::Reports {
            blocks: store.take_reports(session_id).unwrap_or_default(),
        }
    }

    /// Block until no `deliver: context` hook is running for this spool key,
    /// or the budget runs out — whichever comes first. Polled rather than
    /// signalled: in-flight feeds are rare (usually zero, and the loop exits
    /// on its first look), and a poll has no lost-wakeup window to reason
    /// about on a path that must never hang a user's prompt.
    async fn await_feed_hooks(self: &Arc<Self>, key: &str, budget: Duration) {
        let deadline = std::time::Instant::now() + budget;
        loop {
            let running = self
                .feed_hooks_inflight
                .lock()
                .unwrap()
                .get(key)
                .copied()
                .unwrap_or(0);
            if running == 0 {
                return;
            }
            if std::time::Instant::now() >= deadline {
                tracing::debug!(session = %key, running,
                    "feed hooks still running at the drain budget, answering without them");
                return;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
    }

    /// A `deliver: context` hook started for this spool key.
    pub fn feed_hook_started(&self, key: &str) {
        *self
            .feed_hooks_inflight
            .lock()
            .unwrap()
            .entry(key.to_string())
            .or_insert(0) += 1;
    }

    /// ...and finished (or died): drop the count, and the key with it.
    pub fn feed_hook_finished(&self, key: &str) {
        let mut map = self.feed_hooks_inflight.lock().unwrap();
        if let Some(n) = map.get_mut(key) {
            *n = n.saturating_sub(1);
            if *n == 0 {
                map.remove(key);
            }
        }
    }

    /// An opencode fork run finished. Mark it terminal, then nudge the
    /// session's parked stop-wait (resolving it `Waited`): the plugin re-parks
    /// while the session stays idle, and the fresh poll's entry check releases
    /// any `after` dependents this completion unblocked. (Claude Code gets the
    /// same effect from the completion notification's relay turn ending in a
    /// new Stop poll; opencode has no such turn, hence the nudge.)
    ///
    /// `cont`: the run's report ended with the chain sentinel — re-arm the
    /// fork's once-per-pause latch (chain-gated) before the nudge, so the
    /// re-parked poll selects it again.
    pub fn handle_fork_completed(
        self: &Arc<Self>,
        session_id: &str,
        fork: &str,
        run_ref: &str,
        status: &str,
        cont: bool,
    ) -> ResponseBody {
        self.touch_busy();
        let transitioned = {
            let store = self.store.lock().unwrap();
            let (matched, transitioned) = store
                .mark_spawn_terminal(session_id, Some(run_ref), None, status, now())
                .unwrap_or((false, false));
            tracing::debug!(session = %session_id, fork, run_ref, status, matched, cont,
                "opencode fork completion");
            transitioned
        };
        if transitioned {
            self.on_own_fork_terminal(session_id, fork, status, cont);
        }
        self.cancel_wait(session_id);
        ResponseBody::Ack
    }

    /// One of the daemon's own fork runs reached a terminal status (the
    /// `transitioned` edge — callers must dedupe on it, since the same
    /// completion is often seen twice). Two duties:
    ///
    /// **Chain re-arm** — the run completed and its report ended with the
    /// continue sentinel: clear the fork's once-per-pause idle latch so the
    /// next parked poll re-selects it. No epoch bump and no baseline touch,
    /// so every *other* idle fork stays exactly as it was, and the fork's
    /// idle deadline (measured from the pause baseline) has long elapsed —
    /// the re-fire is immediate once the session idles again. Honored only
    /// when the fork's current definition opts in (`chain: true`) and its
    /// wakes this pause stay under the chain limit.
    ///
    /// **Gate release** — the fork holds the session's gate and did NOT
    /// re-arm (chain settled, run failed, or the limit tripped): drop the
    /// gate and clear the pause baseline, so the held idle forks' deadlines
    /// measure from the next Stop — the pause effectively begins now.
    fn on_own_fork_terminal(&self, session_id: &str, fork_name: &str, status: &str, cont: bool) {
        let (session, entry) = {
            let store = self.store.lock().unwrap();
            let session = store.get_session(session_id).ok().flatten();
            let entry = store
                .roster(session_id)
                .ok()
                .and_then(|roster| roster.into_iter().find(|e| e.fork_name == fork_name));
            (session, entry)
        };
        let Some(session) = session else { return };
        let def = entry
            .and_then(|e| std::fs::read_to_string(&e.fork_path).ok())
            .and_then(|content| {
                use autofork_core::frontmatter::ForkParse;
                match autofork_core::frontmatter::parse_fork_file(fork_name, &content) {
                    ForkParse::Fork(parsed) => Some(parsed.def),
                    _ => None,
                }
            });

        let mut rearmed = false;
        if cont && status == "completed" {
            // The client delivers a continuing chain's report as a real turn
            // around the completion frame (opencode injects it, codex queues
            // it); open the dedupe window so duplicated observers of that
            // turn don't classify it as user activity (see the PromptSubmit
            // downgrade).
            if is_native_exec_client(session.client.as_deref()) {
                self.note_chain_continued(session_id);
            }
            match &def {
                Some(def) if def.chain => {
                    let cfg = self.cfg_for(Some(&session.project_root));
                    let limit = def.chain_limit.unwrap_or(cfg.chain_limit) as i64;
                    let since = session.pause_started_at.unwrap_or(session.created_at);
                    let store = self.store.lock().unwrap();
                    let runs = store
                        .count_runs_since(session_id, fork_name, since)
                        .unwrap_or(0);
                    // The per-pause count above resets with the pause; the
                    // wall-clock count cannot — the runaway backstop for
                    // anything that pumps the pause epoch.
                    let window = crate::planner::runaway_window_secs();
                    let hourly = store
                        .count_runs_since(session_id, fork_name, now() - window)
                        .unwrap_or(0);
                    if cfg.runaway_limit > 0 && hourly >= cfg.runaway_limit as i64 {
                        tracing::warn!(session = %session_id, fork = fork_name,
                            runs = hourly, limit = cfg.runaway_limit,
                            "runaway breaker: chain hit its hourly run cap, not re-arming \
                             (raise `runaway_limit` in config if this rate is intended)");
                    } else if runs >= limit {
                        tracing::warn!(session = %session_id, fork = fork_name, runs, limit,
                            "chain limit reached, not re-arming");
                    } else if store
                        .rearm_idle_latch(session_id, fork_name, session.pause_epoch)
                        .unwrap_or(false)
                    {
                        tracing::info!(session = %session_id, fork = fork_name, runs,
                            "chain continue: idle latch re-armed");
                        rearmed = true;
                    }
                }
                _ => {
                    tracing::debug!(session = %session_id, fork = fork_name,
                        "continue sentinel from a fork without chain: true, ignoring");
                }
            }
        }

        // Gate release keys on the *persisted* gate, not the definition —
        // a moved/edited fork file must not leave the gate wedged.
        if !rearmed && session.active_gate.as_deref() == Some(fork_name) {
            let store = self.store.lock().unwrap();
            let _ = store.clear_active_gate(session_id);
            let _ = store.clear_pause_baseline(session_id);
            tracing::info!(session = %session_id, fork = fork_name, status,
                "gate settled: releasing held idle forks, pause restarts");
        }
    }

    /// Whether the session has background work it started still in flight:
    /// a `run_in_background` Bash command, a background subagent or a Monitor
    /// that hasn't reported completion (or been `TaskStop`ped) yet.
    /// autofork's own fork spawns never count (a session isn't busy because
    /// autofork is forking it), and a task older than
    /// `background_hold_timeout` stops counting so work whose completion the
    /// daemon never sees can't silence the session's idle forks for good.
    /// Whether that work makes the session *not idle* is decided per fork
    /// (`background_hold:`, else the config default) at selection.
    pub(crate) fn pending_background(&self, session_id: &str, cfg: &Config, t: i64) -> bool {
        let not_before = match cfg.background_hold_timeout_secs {
            0 => 0,
            secs => t - secs as i64,
        };
        let store = self.store.lock().unwrap();
        store.pending_bg_tasks(session_id, not_before).unwrap_or(0) > 0
    }

    /// Read the transcript delta (updating the stored offset): refresh the
    /// context gauge, record fork spawns and their task ids, and mark spawns
    /// terminal on completion notifications. Returns the session's best-known
    /// prompt token count, or `None` when unavailable.
    fn ingest_transcript(&self, ev: &Event) -> Option<u64> {
        let transcript = ev.transcript_path.as_deref()?;
        let session = {
            let store = self.store.lock().unwrap();
            store.get_session(&ev.session_id).ok().flatten()?
        };
        match crate::transcript::read_delta(transcript, session.transcript_offset) {
            Ok(delta) => {
                let t = now();
                // (fork, status, continue_requested) for spawns this delta
                // flipped terminal — processed after the lock drops, since
                // the terminal handler takes its own locks.
                let mut settled: Vec<(String, String, bool)> = Vec::new();
                {
                    let store = self.store.lock().unwrap();
                    for (tool_use_id, fork_name) in &delta.spawns {
                        tracing::debug!(session = %ev.session_id, tool_use_id, fork = ?fork_name,
                            "fork spawn observed");
                        let _ = store.record_spawn(
                            &ev.session_id,
                            tool_use_id,
                            fork_name.as_deref(),
                            t,
                        );
                    }
                    for (tool_use_id, task_id) in &delta.task_ids {
                        let _ = store.set_spawn_task_id(&ev.session_id, tool_use_id, task_id);
                        // Background work the session itself started (a
                        // `run_in_background` command, a background subagent):
                        // it keeps the session out of the idle state until it
                        // finishes. Own fork spawns are filtered inside.
                        let _ = store.record_bg_task(&ev.session_id, tool_use_id, task_id, t);
                    }
                    // Work the session stopped itself (`TaskStop`) ends
                    // without a notification; the tool use is the only
                    // record that it is over.
                    for task_id in &delta.stopped_tasks {
                        if let Ok(true) =
                            store.mark_bg_task_terminal(&ev.session_id, None, Some(task_id), t)
                        {
                            tracing::debug!(session = %ev.session_id, task_id,
                                "background task stopped by the session; idle clock can start");
                        }
                    }
                    for n in &delta.notifications {
                        let Some(status) = n
                            .status
                            .as_deref()
                            .filter(|s| autofork_core::notification::is_terminal_status(s))
                        else {
                            continue;
                        };
                        if let Ok(true) = store.mark_bg_task_terminal(
                            &ev.session_id,
                            n.tool_use_id.as_deref(),
                            n.task_id.as_deref(),
                            t,
                        ) {
                            tracing::debug!(session = %ev.session_id, status,
                                tool_use_id = ?n.tool_use_id,
                                "background task finished; idle clock can start");
                        }
                        if let Ok((true, transitioned)) = store.mark_spawn_terminal(
                            &ev.session_id,
                            n.tool_use_id.as_deref(),
                            n.task_id.as_deref(),
                            status,
                            t,
                        ) {
                            tracing::debug!(session = %ev.session_id, status,
                                tool_use_id = ?n.tool_use_id, "fork completion observed");
                            if transitioned {
                                if let Ok(Some(fork)) = store.spawn_fork_name(
                                    &ev.session_id,
                                    n.tool_use_id.as_deref(),
                                    n.task_id.as_deref(),
                                ) {
                                    settled.push((fork, status.to_string(), n.continue_requested));
                                }
                            }
                        }
                    }
                    let _ = store.set_transcript_gauge(
                        &ev.session_id,
                        delta.new_offset,
                        delta.prompt_tokens,
                    );
                }
                for (fork, status, cont) in settled {
                    self.on_own_fork_terminal(&ev.session_id, &fork, &status, cont);
                }
                delta.prompt_tokens.or(session.prompt_tokens)
            }
            Err(e) => {
                tracing::debug!(error = %e, "transcript delta unavailable");
                session.prompt_tokens
            }
        }
    }

    /// True when the daemon has nothing to live for right now (no open
    /// connection, which includes any parked stop-wait).
    pub fn is_quiet(&self) -> bool {
        self.connections.load(Ordering::SeqCst) == 0
    }

    /// Exit once quiet for the configured period.
    pub async fn quiet_reaper(self: Arc<Self>) {
        loop {
            tokio::time::sleep(Duration::from_secs(30)).await;
            let quiet_period = self.cfg_for(None).quiet_period_secs as i64;
            let quiet_since = now() - self.last_busy.load(Ordering::SeqCst);
            if self.is_quiet() && quiet_since >= quiet_period {
                tracing::info!("quiet for {quiet_since}s, exiting");
                self.shutdown.notify_waiters();
                return;
            }
        }
    }

    /// Begin shutdown. Parked stop-waits resolve (`Waited`) via the shutdown
    /// notify; `drain` is accepted for wire compatibility but there are no
    /// in-flight runs to drain.
    pub async fn request_shutdown(self: &Arc<Self>, _drain: bool) {
        self.shutdown.notify_waiters();
    }
}