marver 0.0.27

//! Receiving agent lifecycle events and turning them into state changes.

use std::io::{Read, Write};
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::{Path, PathBuf};
use std::time::Duration;

use chrono::{DateTime, Utc};
use serde_json::{Value, json};

use crate::domain::{BlockedKind, TaskState};
use crate::store::{BlockedInfo, Store, Transition};

#[derive(Debug, thiserror::Error)]
pub enum Error {
    #[error("io error at {path}: {source}")]
    Io {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
    #[error("malformed hook payload: {0}")]
    Malformed(#[source] serde_json::Error),
    /// The connection misbehaved. Concerns that caller only, never the
    /// listener.
    #[error("hook connection rejected: {0}")]
    Rejected(String),
    /// Somebody connected and said nothing — a liveness check, not a hook.
    #[error("liveness probe")]
    Probe,
    /// Another daemon holds the socket. Distinguished from a plain io error
    /// because it is the one failure that is not a fault: the user asked for a
    /// daemon and there is already one, so the answer is to say so and stop,
    /// not to report `Address already in use` and leave them guessing.
    #[error("a marver daemon is already running on {0}")]
    AlreadyRunning(PathBuf),
    /// The socket path exceeded `sockaddr_un`, which is around 100 bytes and
    /// smaller than anything else that accepts a path. Worth its own variant
    /// because the raw message — `path must be shorter than SUN_LEN` — names a
    /// constant from a C header rather than the flag the user passed.
    #[error("the socket path is {len} bytes, past the ~100 a unix socket allows: {path}")]
    SocketPathTooLong { path: PathBuf, len: usize },
    #[error(transparent)]
    Store(#[from] crate::store::Error),
}

pub type Result<T> = std::result::Result<T, Error>;

/// The lifecycle events marver subscribes to.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Kind {
    /// A session began or resumed. Also fires on `/clear` and on a compaction,
    /// so it says a session exists, not that work has just started.
    SessionStart,
    /// The user submitted a prompt. The one event that says a waiting agent has
    /// been answered.
    UserPromptSubmit,
    /// Claude finished responding. The agent's work is done for now.
    Stop,
    /// A turn ended because of an API error rather than completion.
    StopFailure,
    /// Claude wants the user — a permission prompt, or a question.
    Notification,
    /// The session terminated.
    SessionEnd,
    /// Something marver does not model, kept so it still reaches the log.
    Other(String),
}

impl Kind {
    /// The hook event names marver configures, in the spelling Claude Code
    /// uses. Adding an event is a row here, a variant above, and an arm in
    /// [`requested`].
    pub const SUBSCRIBED: &'static [&'static str] = &[
        "SessionStart",
        "UserPromptSubmit",
        "Stop",
        "StopFailure",
        "Notification",
        "SessionEnd",
    ];

    pub fn parse(name: &str) -> Self {
        match name {
            "SessionStart" => Self::SessionStart,
            "UserPromptSubmit" => Self::UserPromptSubmit,
            "Stop" => Self::Stop,
            "StopFailure" => Self::StopFailure,
            "Notification" => Self::Notification,
            "SessionEnd" => Self::SessionEnd,
            other => Self::Other(other.to_string()),
        }
    }

    /// The name in [`Self::SUBSCRIBED`] this parses back from.
    fn name(&self) -> Option<&'static str> {
        Self::SUBSCRIBED
            .iter()
            .copied()
            .find(|name| Self::parse(name) == *self)
    }
}

/// One hook event, with the fields marver reads lifted out of the raw JSON.
#[derive(Debug, Clone, PartialEq)]
pub struct Payload {
    pub kind: Kind,
    pub session_id: Option<String>,
    pub cwd: Option<PathBuf>,
    /// Where Claude Code is writing this session's transcript. The only source
    /// of what the agent has spent — no hook carries usage itself.
    pub transcript_path: Option<PathBuf>,
    /// For `Notification`: which sort it is, e.g. `permission_prompt`.
    pub notification_type: Option<String>,
    /// For `Notification`: the text shown to the user.
    pub message: Option<String>,
    /// For `SessionEnd`: `clear`, `logout`, `prompt_input_exit`, `other`, ...
    pub reason: Option<String>,
    /// For `Stop`: the agent's closing message. Worth keeping — it is the
    /// natural summary to show beside a task waiting for review.
    pub last_assistant_message: Option<String>,
    /// The complete payload, preserved verbatim.
    pub raw: Value,
}

impl Payload {
    pub fn from_json(raw: Value) -> Self {
        let text = |key: &str| raw.get(key).and_then(Value::as_str).map(str::to_string);
        Self {
            kind: Self::kind_of(&raw),
            session_id: text("session_id").or_else(|| text("thread-id")),
            cwd: text("cwd").map(PathBuf::from),
            transcript_path: text("transcript_path").map(PathBuf::from),
            notification_type: text("notification_type"),
            message: text("message"),
            reason: text("reason"),
            last_assistant_message: text("last_assistant_message"),
            raw,
        }
    }

    /// Which event this is, whichever harness is describing it.
    fn kind_of(raw: &Value) -> Kind {
        if let Some(name) = raw.get("hook_event_name").and_then(Value::as_str) {
            return Kind::parse(name);
        }
        match raw.get("type").and_then(Value::as_str) {
            Some("agent-turn-complete") => Kind::Stop,
            Some(other) => Kind::Other(other.to_string()),
            None => Kind::Other(String::new()),
        }
    }

    pub fn parse(bytes: &[u8]) -> Result<Self> {
        serde_json::from_slice(bytes)
            .map(Self::from_json)
            .map_err(Error::Malformed)
    }

    /// Whether a `Notification` means the agent is waiting on the user, and if
    /// so what kind of waiting.
    fn blocked_kind(&self) -> Option<BlockedKind> {
        match self.notification_type.as_deref() {
            Some("permission_prompt") => Some(BlockedKind::PermissionPrompt),
            Some("idle_prompt") => Some(BlockedKind::Silence),
            Some("elicitation_dialog" | "agent_needs_input") => Some(BlockedKind::Question),
            Some(
                "auth_success"
                | "elicitation_complete"
                | "elicitation_response"
                | "agent_completed",
            ) => None,
            _ => Some(BlockedKind::Question),
        }
    }
}

/// One delivery: a payload plus the task it belongs to.
#[derive(Debug, Clone, PartialEq)]
pub struct Delivery {
    pub task_id: i64,
    pub payload: Payload,
}

impl Delivery {
    /// Wire format: the task id and the raw event, one JSON object.
    pub fn to_json(&self) -> Value {
        json!({ "task_id": self.task_id, "payload": self.payload.raw })
    }

    pub fn from_json(value: Value) -> Option<Self> {
        Some(Self {
            task_id: value.get("task_id")?.as_i64()?,
            payload: Payload::from_json(value.get("payload")?.clone()),
        })
    }
}

/// What a hook implies for a task, given where it currently is.
#[derive(Debug, Clone, PartialEq)]
pub enum Outcome {
    /// The task moved.
    Moved { to: TaskState },
    /// The event was recorded but implied no move, or the move was not legal
    /// from the current state.
    Recorded { reason: &'static str },
}

/// The transition a payload asks for, before legality is considered.
fn requested(payload: &Payload, current: TaskState) -> Option<(TaskState, Transition)> {
    match payload.kind {
        // The scheduler is what starts a task; SessionStart only confirms it,
        // and only for one that has not started yet, since it fires on `/clear`
        // and on a compaction too.
        Kind::SessionStart => match current {
            TaskState::Queued => Some((TaskState::Running, Transition::Plain)),
            _ => None,
        },
        // Answering a waiting agent, except a paused one: `p` gave up its slot,
        // and `p` is what takes it back.
        Kind::UserPromptSubmit => match current {
            TaskState::Paused => None,
            _ => Some((TaskState::Running, Transition::Plain)),
        },
        Kind::Stop => Some((TaskState::AwaitingReview, Transition::Plain)),
        Kind::StopFailure => Some((
            TaskState::Failed,
            Transition::Failed(
                payload
                    .message
                    .clone()
                    .unwrap_or_else(|| "the turn ended with an API error".to_string()),
            ),
        )),
        Kind::Notification => {
            let kind = payload.blocked_kind()?;
            let info = match &payload.message {
                Some(message) => BlockedInfo::with_reason(kind, message.clone()),
                None => BlockedInfo::new(kind),
            };
            Some((TaskState::Blocked, Transition::Blocked(info)))
        }
        // A session ending while work is outstanding means the agent went away
        // without finishing.
        Kind::SessionEnd => match current {
            TaskState::Running | TaskState::Blocked => Some((
                TaskState::Failed,
                Transition::Failed("the agent's session ended before it finished".to_string()),
            )),
            _ => None,
        },
        Kind::Other(_) => None,
    }
}

/// Record a hook and apply whatever state change it implies.
pub fn apply(store: &mut Store, delivery: &Delivery, now: DateTime<Utc>) -> Result<Outcome> {
    let task = store.get_task(delivery.task_id)?;

    store.append_event(
        Some(task.id),
        &format!("hook.{}", event_slug(&delivery.payload.kind)),
        &delivery.payload.raw,
        now,
    )?;

    let Some((next, detail)) = requested(&delivery.payload, task.state) else {
        return Ok(Outcome::Recorded {
            reason: "the event implies no state change",
        });
    };

    if next == task.state {
        return Ok(Outcome::Recorded {
            reason: "the task is already in that state",
        });
    }
    if !task.state.can_transition_to(next) {
        return Ok(Outcome::Recorded {
            reason: "the transition is not legal from the current state",
        });
    }

    store.transition(task.id, next, detail, now)?;
    Ok(Outcome::Moved { to: next })
}

/// What a hook is logged as: `SessionStart` becomes `session-start`.
fn event_slug(kind: &Kind) -> String {
    let Some(name) = kind.name() else {
        let Kind::Other(name) = kind else {
            unreachable!("every kind but Other is in SUBSCRIBED");
        };
        return format!("other.{name}");
    };
    let mut slug = String::new();
    for (index, ch) in name.char_indices() {
        if ch.is_ascii_uppercase() && index > 0 {
            slug.push('-');
        }
        slug.push(ch.to_ascii_lowercase());
    }
    slug
}

// ---- transport ---------------------------------------------------------

/// Whether a daemon is listening on this socket.
pub fn is_listening(socket: &Path) -> bool {
    UnixStream::connect(socket).is_ok()
}

/// Send one delivery to the daemon and close.
pub fn send(socket: &Path, delivery: &Delivery) -> Result<()> {
    let mut stream = UnixStream::connect(socket).map_err(|source| Error::Io {
        path: socket.to_path_buf(),
        source,
    })?;
    let body = delivery.to_json().to_string();
    stream
        .write_all(body.as_bytes())
        .and_then(|()| stream.flush())
        .and_then(|()| stream.shutdown(std::net::Shutdown::Write))
        .map_err(|source| Error::Io {
            path: socket.to_path_buf(),
            source,
        })
}

/// Listens for hook deliveries. The largest hook body worth reading.
const MAX_BODY: usize = 1 << 20;

/// How long a connection may take to deliver its body.
const READ_TIMEOUT: Duration = Duration::from_secs(5);

pub struct Receiver {
    listener: UnixListener,
    path: PathBuf,
}

impl Receiver {
    /// Bind the socket, replacing a stale one left by a previous run.
    pub fn bind(path: impl Into<PathBuf>) -> Result<Self> {
        let path = path.into();
        if is_listening(&path) {
            // Checked before the removal below, which is what would otherwise
            // pull the socket out from under a working daemon.
            return Err(Error::AlreadyRunning(path));
        }
        if path.exists() {
            std::fs::remove_file(&path).map_err(|source| Error::Io {
                path: path.clone(),
                source,
            })?;
        }
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).map_err(|source| Error::Io {
                path: parent.to_path_buf(),
                source,
            })?;
        }
        let listener = UnixListener::bind(&path).map_err(|source| match source.kind() {
            // What `sockaddr_un` overflow surfaces as.
            std::io::ErrorKind::InvalidInput => Error::SocketPathTooLong {
                len: path.as_os_str().len(),
                path: path.clone(),
            },
            // Lost a race with another daemon between the check above and
            // here.
            std::io::ErrorKind::AddrInUse => Error::AlreadyRunning(path.clone()),
            _ => Error::Io {
                path: path.clone(),
                source,
            },
        })?;
        Ok(Self { listener, path })
    }

    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Accept one connection and read the delivery it carries.
    pub fn accept(&self) -> Result<Delivery> {
        let (mut stream, _) = self.listener.accept().map_err(|source| Error::Io {
            path: self.path.clone(),
            source,
        })?;
        // A client that connects and says nothing would otherwise hold the
        // accept loop forever, and no other hook could be delivered.
        let _ = stream.set_read_timeout(Some(READ_TIMEOUT));

        let mut body = Vec::new();
        // One byte over the limit is enough to notice; the rest is not read.
        std::io::Read::by_ref(&mut stream)
            .take(MAX_BODY as u64 + 1)
            .read_to_end(&mut body)
            .map_err(|source| Error::Rejected(source.to_string()))?;
        if body.len() > MAX_BODY {
            return Err(Error::Rejected(format!(
                "payload is larger than {MAX_BODY} bytes"
            )));
        }
        // A connection that closes without sending anything is how liveness is
        // tested: see `is_listening`.
        if body.is_empty() {
            return Err(Error::Probe);
        }
        let body = String::from_utf8(body)
            .map_err(|_| Error::Rejected("payload is not valid UTF-8".to_string()))?;

        let value: Value = serde_json::from_str(&body).map_err(Error::Malformed)?;
        Delivery::from_json(value).ok_or_else(|| {
            Error::Malformed(serde::de::Error::custom(
                "missing task_id or payload".to_string(),
            ))
        })
    }
}

impl Drop for Receiver {
    fn drop(&mut self) {
        let _ = std::fs::remove_file(&self.path);
    }
}

// ---- settings generation -----------------------------------------------

/// The Claude Code settings a task's agent runs with.
pub fn settings_for_task(task_id: i64, marver_bin: &Path, socket: &Path) -> Value {
    let entry = json!({
        "hooks": [{
            "type": "command",
            "command": marver_bin.to_string_lossy(),
            "args": [
                "hook",
                "--task", task_id.to_string(),
                "--socket", socket.to_string_lossy(),
            ],
            "timeout": 10
        }]
    });

    let mut hooks = serde_json::Map::new();
    for event in Kind::SUBSCRIBED {
        hooks.insert((*event).to_string(), json!([entry]));
    }
    json!({ "hooks": Value::Object(hooks) })
}

/// Write a task's settings file and return its path, for `claude --settings`.
pub fn write_settings(
    dir: &Path,
    task_id: i64,
    marver_bin: &Path,
    socket: &Path,
) -> Result<PathBuf> {
    std::fs::create_dir_all(dir).map_err(|source| Error::Io {
        path: dir.to_path_buf(),
        source,
    })?;
    let path = dir.join("claude-settings.json");
    let body = serde_json::to_string_pretty(&settings_for_task(task_id, marver_bin, socket))
        .map_err(Error::Malformed)?;
    std::fs::write(&path, body).map_err(|source| Error::Io {
        path: path.clone(),
        source,
    })?;
    Ok(path)
}

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

    fn at(secs: i64) -> DateTime<Utc> {
        DateTime::from_timestamp(secs, 0).expect("valid timestamp")
    }

    fn store_with_task() -> (Store, i64) {
        let mut store = Store::open_in_memory().unwrap();
        let task = store
            .create_task("t", "do it", Path::new("/tmp/tasks"), &[], at(0))
            .unwrap();
        let id = task.id;
        (store, id)
    }

    fn payload(kind: &str, extra: Value) -> Payload {
        let mut raw = json!({
            "session_id": "abc123",
            "transcript_path": "/tmp/t.jsonl",
            "cwd": "/tmp/work",
            "permission_mode": "default",
            "hook_event_name": kind,
        });
        if let (Value::Object(base), Value::Object(more)) = (&mut raw, extra) {
            base.extend(more);
        }
        Payload::from_json(raw)
    }

    fn deliver(store: &mut Store, id: i64, p: Payload, secs: i64) -> Outcome {
        apply(
            store,
            &Delivery {
                task_id: id,
                payload: p,
            },
            at(secs),
        )
        .unwrap()
    }

    fn start(store: &mut Store, id: i64) {
        store
            .transition(id, TaskState::Running, Transition::Plain, at(1))
            .unwrap();
    }

    #[test]
    fn parses_the_documented_common_fields() {
        let p = payload("Stop", json!({}));
        assert_eq!(p.kind, Kind::Stop);
        assert_eq!(p.session_id.as_deref(), Some("abc123"));
        assert_eq!(p.cwd.as_deref(), Some(Path::new("/tmp/work")));
    }

    /// Payloads captured from Claude Code 2.1.221, not invented.
    #[test]
    fn parses_payloads_observed_from_a_real_session() {
        let start = Payload::from_json(json!({
            "cwd": "/tmp/w", "hook_event_name": "SessionStart",
            "session_id": "s1", "source": "startup",
            "transcript_path": "/tmp/t.jsonl"
        }));
        assert_eq!(start.kind, Kind::SessionStart);
        // SessionStart carries no permission_mode; absence must not break it.
        assert_eq!(start.raw["source"], "startup");

        let stop = Payload::from_json(json!({
            "cwd": "/tmp/w", "hook_event_name": "Stop", "session_id": "s1",
            "permission_mode": "auto", "stop_hook_active": false,
            "last_assistant_message": "PONG", "stop_reason": "end_turn"
        }));
        assert_eq!(stop.kind, Kind::Stop);
        assert_eq!(stop.last_assistant_message.as_deref(), Some("PONG"));

        let end = Payload::from_json(json!({
            "cwd": "/tmp/w", "hook_event_name": "SessionEnd",
            "session_id": "s1", "reason": "other"
        }));
        assert_eq!(end.kind, Kind::SessionEnd);
        assert_eq!(end.reason.as_deref(), Some("other"));
    }

    #[test]
    fn an_unmodelled_event_is_kept_not_dropped() {
        let p = payload("PreToolUse", json!({"tool_name": "Bash"}));
        assert_eq!(p.kind, Kind::Other("PreToolUse".into()));
        assert_eq!(p.raw["tool_name"], "Bash");
    }

    #[test]
    fn stop_sends_a_running_task_to_review() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        let outcome = deliver(&mut store, id, payload("Stop", json!({})), 2);
        assert_eq!(
            outcome,
            Outcome::Moved {
                to: TaskState::AwaitingReview
            }
        );
        assert_eq!(store.get_task(id).unwrap().state, TaskState::AwaitingReview);
    }

    #[test]
    fn a_finished_codex_turn_is_a_stop() {
        // Codex names its one event in `type` rather than `hook_event_name`,
        // and a turn ending with the agent waiting is what Stop means here.
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        let raw = json!({
            "type": "agent-turn-complete",
            "thread-id": "0199a5f0",
            "turn-id": "1",
            "cwd": "/tmp/work",
            "input-messages": ["fix the auth flow"],
            "last-assistant-message": "Done — the tests pass.",
        });
        let payload = Payload::from_json(raw.clone());
        assert_eq!(payload.kind, Kind::Stop);
        assert_eq!(payload.session_id.as_deref(), Some("0199a5f0"));

        let outcome = deliver(&mut store, id, payload, 2);

        assert_eq!(
            outcome,
            Outcome::Moved {
                to: TaskState::AwaitingReview
            }
        );
        let logged = store.list_events(id).unwrap();
        assert!(
            logged.iter().any(|event| event.payload == raw),
            "the payload codex sent is what the log keeps: {logged:?}"
        );
    }

    #[test]
    fn an_event_from_a_harness_nobody_taught_marver_is_kept_rather_than_guessed_at() {
        let payload = Payload::from_json(json!({ "type": "something-new" }));
        assert_eq!(payload.kind, Kind::Other("something-new".into()));

        let payload = Payload::from_json(json!({ "no": "names at all" }));
        assert_eq!(payload.kind, Kind::Other(String::new()));
    }

    #[test]
    fn a_permission_prompt_blocks_with_the_right_kind() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload(
                "Notification",
                json!({
                    "notification_type": "permission_prompt",
                    "message": "Claude needs permission to edit src/main.rs"
                }),
            ),
            2,
        );

        let task = store.get_task(id).unwrap();
        assert_eq!(task.state, TaskState::Blocked);
        assert_eq!(task.blocked_kind, Some(BlockedKind::PermissionPrompt));
        assert_eq!(
            task.blocked_reason.as_deref(),
            Some("Claude needs permission to edit src/main.rs")
        );
    }

    #[test]
    fn notification_types_map_to_the_right_blocked_kind() {
        for (notification, expected) in [
            ("permission_prompt", BlockedKind::PermissionPrompt),
            ("idle_prompt", BlockedKind::Silence),
            ("elicitation_dialog", BlockedKind::Question),
            ("agent_needs_input", BlockedKind::Question),
        ] {
            let (mut store, id) = store_with_task();
            start(&mut store, id);
            deliver(
                &mut store,
                id,
                payload("Notification", json!({ "notification_type": notification })),
                2,
            );
            let task = store.get_task(id).unwrap();
            assert_eq!(task.state, TaskState::Blocked, "for {notification}");
            assert_eq!(task.blocked_kind, Some(expected), "for {notification}");
        }
    }

    #[test]
    fn informational_notifications_do_not_block() {
        // The bug this guards: treating every Notification as blocking parks a
        // working task the moment it refreshes credentials.
        for notification in [
            "auth_success",
            "elicitation_complete",
            "elicitation_response",
            "agent_completed",
        ] {
            let (mut store, id) = store_with_task();
            start(&mut store, id);
            let outcome = deliver(
                &mut store,
                id,
                payload("Notification", json!({ "notification_type": notification })),
                2,
            );
            assert!(
                matches!(outcome, Outcome::Recorded { .. }),
                "{notification} should not block"
            );
            assert_eq!(
                store.get_task(id).unwrap().state,
                TaskState::Running,
                "{notification} should leave the task working"
            );
        }
    }

    #[test]
    fn a_task_that_hit_a_permission_prompt_still_reaches_review() {
        // The whole sequence, as it really arrives: the agent asks to run
        // something, the user approves it in the pane — which emits no hook —
        // and the agent then finishes.
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload(
                "Notification",
                json!({"notification_type": "permission_prompt"}),
            ),
            2,
        );
        assert_eq!(store.get_task(id).unwrap().state, TaskState::Blocked);

        let outcome = deliver(&mut store, id, payload("Stop", json!({})), 3);
        assert_eq!(
            outcome,
            Outcome::Moved {
                to: TaskState::AwaitingReview
            },
            "a blocked agent that finishes must offer its diff, not strand"
        );
        let task = store.get_task(id).unwrap();
        assert_eq!(task.state, TaskState::AwaitingReview);
        assert_eq!(task.blocked_kind, None, "the block must be cleared");
    }

    #[test]
    fn an_idle_prompt_arriving_before_stop_does_not_trap_the_task() {
        // `idle_prompt` means "done, waiting for your next prompt", so it is
        // emitted alongside `Stop` at the end of an ordinary turn.
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload("Notification", json!({"notification_type": "idle_prompt"})),
            2,
        );
        deliver(&mut store, id, payload("Stop", json!({})), 3);
        assert_eq!(
            store.get_task(id).unwrap().state,
            TaskState::AwaitingReview,
            "the ordinary end of every turn must not depend on hook ordering"
        );
    }

    #[test]
    fn an_unrecognised_notification_still_blocks() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload(
                "Notification",
                json!({"notification_type": "something_new"}),
            ),
            2,
        );
        let task = store.get_task(id).unwrap();
        assert_eq!(
            task.state,
            TaskState::Blocked,
            "an unknown notification still means the agent wants the user"
        );
        assert_eq!(task.blocked_kind, Some(BlockedKind::Question));
    }

    #[test]
    fn stop_failure_fails_the_task_with_a_reason() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload("StopFailure", json!({"message": "overloaded"})),
            2,
        );
        let task = store.get_task(id).unwrap();
        assert_eq!(task.state, TaskState::Failed);
        assert_eq!(task.failure_reason.as_deref(), Some("overloaded"));
    }

    #[test]
    fn stop_failure_without_a_message_still_records_one() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(&mut store, id, payload("StopFailure", json!({})), 2);
        // The schema requires a reason in the failed state.
        assert!(store.get_task(id).unwrap().failure_reason.is_some());
    }

    #[test]
    fn session_end_fails_a_task_that_was_still_working() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(&mut store, id, payload("SessionEnd", json!({})), 2);
        assert_eq!(store.get_task(id).unwrap().state, TaskState::Failed);
    }

    #[test]
    fn session_end_after_review_started_is_unremarkable() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(&mut store, id, payload("Stop", json!({})), 2);
        let outcome = deliver(&mut store, id, payload("SessionEnd", json!({})), 3);

        assert!(matches!(outcome, Outcome::Recorded { .. }));
        assert_eq!(
            store.get_task(id).unwrap().state,
            TaskState::AwaitingReview,
            "closing a finished session must not fail the task"
        );
    }

    #[test]
    fn replying_to_a_waiting_agent_puts_it_back_to_work() {
        // The event that says so is UserPromptSubmit. SessionStart does not
        // fire when an agent is answered — it fires on startup, `/clear` and
        // compaction — so a task blocked on a question stayed blocked, holding
        // a concurrency slot, until the eventual Stop.
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload(
                "Notification",
                json!({"notification_type": "agent_needs_input"}),
            ),
            2,
        );
        assert_eq!(store.get_task(id).unwrap().state, TaskState::Blocked);

        deliver(&mut store, id, payload("UserPromptSubmit", json!({})), 3);

        let task = store.get_task(id).unwrap();
        assert_eq!(task.state, TaskState::Running);
        assert_eq!(task.blocked_kind, None);
    }

    #[test]
    fn a_prompt_typed_at_a_paused_agent_does_not_resume_it() {
        // `p` set the task aside and gave up its slot. Its session is still
        // there and can still be typed into, but only `p` takes the slot back.
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        store
            .transition(id, TaskState::Paused, Transition::Plain, at(2))
            .unwrap();

        let outcome = deliver(&mut store, id, payload("UserPromptSubmit", json!({})), 3);

        assert!(matches!(outcome, Outcome::Recorded { .. }));
        assert_eq!(store.get_task(id).unwrap().state, TaskState::Paused);
    }

    #[test]
    fn a_compaction_does_not_drag_a_finished_task_back_out_of_review() {
        // SessionStart fires on `/clear` and on a compaction too, and
        // awaiting-review -> running is a legal move, so asking for it
        // unconditionally undid a finished task.
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(&mut store, id, payload("Stop", json!({})), 2);

        let outcome = deliver(
            &mut store,
            id,
            payload("SessionStart", json!({"source": "compact"})),
            3,
        );

        assert!(matches!(outcome, Outcome::Recorded { .. }));
        assert_eq!(
            store.get_task(id).unwrap().state,
            TaskState::AwaitingReview,
            "the diff is still waiting to be looked at"
        );
    }

    #[test]
    fn a_session_starting_under_a_queued_task_confirms_it_is_running() {
        let (mut store, id) = store_with_task();

        deliver(&mut store, id, payload("SessionStart", json!({})), 2);

        assert_eq!(store.get_task(id).unwrap().state, TaskState::Running);
    }

    #[test]
    fn a_late_hook_cannot_resurrect_a_cancelled_task() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        store
            .transition(id, TaskState::Cancelled, Transition::Plain, at(2))
            .unwrap();

        let outcome = deliver(&mut store, id, payload("Stop", json!({})), 3);
        assert!(matches!(outcome, Outcome::Recorded { .. }));
        assert_eq!(store.get_task(id).unwrap().state, TaskState::Cancelled);
    }

    #[test]
    fn a_repeated_hook_is_not_an_error() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(&mut store, id, payload("Stop", json!({})), 2);
        let again = deliver(&mut store, id, payload("Stop", json!({})), 3);
        assert!(matches!(again, Outcome::Recorded { .. }));
    }

    #[test]
    fn every_hook_is_logged_even_when_it_changes_nothing() {
        let (mut store, id) = store_with_task();
        deliver(&mut store, id, payload("PreToolUse", json!({})), 2);
        let kinds: Vec<String> = store
            .list_events(id)
            .unwrap()
            .into_iter()
            .map(|e| e.kind)
            .collect();
        assert!(
            kinds.contains(&"hook.other.PreToolUse".to_string()),
            "got {kinds:?}"
        );
    }

    #[test]
    fn the_raw_payload_survives_into_the_log() {
        let (mut store, id) = store_with_task();
        start(&mut store, id);
        deliver(
            &mut store,
            id,
            payload("Stop", json!({"a_field_marver_ignores": 42})),
            2,
        );
        let events = store.list_events(id).unwrap();
        let hook = events.iter().find(|e| e.kind == "hook.stop").unwrap();
        assert_eq!(hook.payload["a_field_marver_ignores"], 42);
        assert_eq!(hook.payload["session_id"], "abc123");
    }

    // ---- transport ----

    #[test]
    fn a_delivery_survives_the_socket() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let receiver = Receiver::bind(&socket).unwrap();

        let sent = Delivery {
            task_id: 42,
            payload: payload("Stop", json!({"extra": "kept"})),
        };
        let to_send = sent.clone();
        let path = socket.clone();
        let sender = std::thread::spawn(move || send(&path, &to_send).unwrap());

        let received = receiver.accept().unwrap();
        sender.join().unwrap();

        assert_eq!(received.task_id, 42);
        assert_eq!(received.payload.kind, Kind::Stop);
        assert_eq!(received.payload.raw["extra"], "kept");
    }

    /// Connect, write `body` verbatim, and go away.
    fn raw_send(path: &Path, body: &[u8]) {
        let mut stream = UnixStream::connect(path).unwrap();
        stream.write_all(body).unwrap();
    }

    #[test]
    fn one_bad_byte_does_not_stop_the_listener() {
        // `printf '\xff' | nc -U <socket>` used to end hook delivery for good:
        // the read failed as an io error, the daemon treated that as the
        // listener dying, and `Drop` then unlinked the socket file.
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let receiver = Receiver::bind(&socket).unwrap();

        let path = socket.clone();
        std::thread::spawn(move || raw_send(&path, &[0xff]));
        let err = receiver.accept().unwrap_err();
        assert!(
            matches!(err, Error::Rejected(_)),
            "a bad byte concerns that caller only, got {err:?}"
        );

        // And the listener is still good for the next, valid hook.
        let good = Delivery {
            task_id: 7,
            payload: payload("Stop", json!({})),
        };
        let to_send = good.clone();
        let path = socket.clone();
        std::thread::spawn(move || send(&path, &to_send).unwrap());
        assert_eq!(receiver.accept().unwrap().task_id, 7);
    }

    #[test]
    fn an_oversized_payload_is_rejected_rather_than_buffered() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let receiver = Receiver::bind(&socket).unwrap();

        let path = socket.clone();
        std::thread::spawn(move || {
            // Any local process could otherwise drive the daemon to OOM.
            let _ = UnixStream::connect(&path).map(|mut s| s.write_all(&vec![b'x'; MAX_BODY * 2]));
        });
        assert!(matches!(receiver.accept(), Err(Error::Rejected(_))));
    }

    #[test]
    fn a_silent_client_does_not_wedge_the_listener() {
        // One `nc -U` left open used to freeze the daemon's entire event
        // intake: accept reads inline, single-threaded, with no timeout.
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let receiver = Receiver::bind(&socket).unwrap();

        // Connect and hold it open without sending anything.
        let idle = UnixStream::connect(&socket).unwrap();
        let start = std::time::Instant::now();
        assert!(matches!(receiver.accept(), Err(Error::Rejected(_))));
        assert!(
            start.elapsed() < READ_TIMEOUT * 2,
            "the read must time out, not block forever"
        );
        drop(idle);

        let good = Delivery {
            task_id: 3,
            payload: payload("Stop", json!({})),
        };
        let to_send = good.clone();
        let path = socket.clone();
        std::thread::spawn(move || send(&path, &to_send).unwrap());
        assert_eq!(
            receiver.accept().unwrap().task_id,
            3,
            "hooks must flow again once the stalled client is dealt with"
        );
    }

    #[test]
    fn a_stale_socket_does_not_block_startup() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        {
            let _first = Receiver::bind(&socket).unwrap();
            // Leave the file behind as a crash would.
            std::fs::write(&socket, "").ok();
        }
        std::fs::write(&socket, "").ok();
        // A crashed daemon must not make marver permanently unstartable.
        let again = Receiver::bind(&socket);
        assert!(again.is_ok(), "{:?}", again.err());
    }

    #[test]
    fn a_live_socket_is_not_stolen_and_says_why() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let _first = Receiver::bind(&socket).unwrap();

        let Err(err) = Receiver::bind(&socket) else {
            panic!("a second bind must fail");
        };

        // Not merely an error: the one failure here that is not a fault.
        assert!(matches!(err, Error::AlreadyRunning(_)), "got {err:?}");
        assert!(
            err.to_string().contains("already running"),
            "{}",
            err.to_string()
        );
    }

    #[test]
    fn listening_is_answered_by_connecting_not_by_the_file_existing() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        assert!(!is_listening(&socket), "nothing is bound yet");

        {
            let _receiver = Receiver::bind(&socket).unwrap();
            assert!(is_listening(&socket));
        }

        // A crash leaves the file behind.
        std::fs::write(&socket, "").ok();
        assert!(
            !is_listening(&socket),
            "a leftover socket file is not a daemon"
        );
    }

    #[test]
    fn a_probe_is_told_apart_from_a_broken_hook() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let receiver = Receiver::bind(&socket).unwrap();

        let probe = std::thread::spawn(move || {
            // Exactly what `is_listening` does: connect, say nothing, close.
            let _ = UnixStream::connect(&socket);
        });
        let err = receiver.accept().expect_err("a probe carries no delivery");
        probe.join().unwrap();

        // Reported as its own kind so the daemon can stay silent.
        assert!(matches!(err, Error::Probe), "got {err:?}");
    }

    #[test]
    fn the_socket_is_removed_on_shutdown() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        {
            let _receiver = Receiver::bind(&socket).unwrap();
            assert!(socket.exists());
        }
        assert!(!socket.exists());
    }

    #[test]
    fn a_malformed_body_is_rejected_without_killing_the_listener() {
        let tmp = TempDir::new().unwrap();
        let socket = tmp.path().join("hooks.sock");
        let receiver = Receiver::bind(&socket).unwrap();

        let path = socket.clone();
        std::thread::spawn(move || {
            let mut s = UnixStream::connect(&path).unwrap();
            s.write_all(b"not json at all").ok();
        });
        assert!(matches!(receiver.accept(), Err(Error::Malformed(_))));

        // The listener still works afterwards.
        let good = Delivery {
            task_id: 1,
            payload: payload("Stop", json!({})),
        };
        let path = socket.clone();
        let to_send = good.clone();
        std::thread::spawn(move || send(&path, &to_send).unwrap());
        assert_eq!(receiver.accept().unwrap().task_id, 1);
    }

    // ---- settings ----

    #[test]
    fn generated_settings_subscribe_to_every_modelled_event() {
        let settings = settings_for_task(7, Path::new("/usr/local/bin/marver"), Path::new("/s"));
        let hooks = settings["hooks"].as_object().unwrap();
        for event in Kind::SUBSCRIBED {
            assert!(hooks.contains_key(*event), "{event} missing");
        }
        assert_eq!(hooks.len(), Kind::SUBSCRIBED.len());
    }

    #[test]
    fn the_task_id_is_fixed_in_the_argument_vector() {
        let settings = settings_for_task(7, Path::new("/bin/marver"), Path::new("/run/m.sock"));
        let args = &settings["hooks"]["Stop"][0]["hooks"][0]["args"];
        assert_eq!(
            args.as_array().unwrap(),
            &json!(["hook", "--task", "7", "--socket", "/run/m.sock"])
                .as_array()
                .unwrap()
                .clone(),
            "the id must not have to be inferred from cwd"
        );
        // Exec form: args present means no shell splits the path.
        assert_eq!(
            settings["hooks"]["Stop"][0]["hooks"][0]["command"],
            "/bin/marver"
        );
    }

    #[test]
    fn settings_are_written_where_claude_can_be_pointed_at_them() {
        let tmp = TempDir::new().unwrap();
        let path = write_settings(
            tmp.path(),
            3,
            Path::new("/bin/marver"),
            Path::new("/run/m.sock"),
        )
        .unwrap();

        assert!(path.exists());
        let parsed: Value = serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        assert_eq!(
            parsed["hooks"]["Notification"][0]["hooks"][0]["args"][2],
            "3"
        );
    }
}