agentwerk 0.1.10

//! Ticket queue and run orchestration. `TicketSystem` owns the shared
//! ticket store, the registered agents, the active policies, the
//! interrupt signal, and the run-time [`Stats`] object.
//! `bind_agent` stamps the ticket Arc, policies, stats, and signal onto
//! each agent at add time; `run` / `run_dry` then drive the bound
//! agents.

use std::collections::HashMap;
use std::fmt;
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, Weak};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use serde::Serialize;

use crate::providers::{AsUserMessage, Message};

use super::agent::Agent;
use super::policy::Policies;
use super::r#loop::run_main_loop;
use super::stats::{Stats, TicketStats};

/// A ticket. Caller-settable fields: `task`, `labels`, `schema`,
/// `labels`. System-managed fields (`key`, `status`, `reporter`,
/// `created_at`, `result`) are stamped at insertion time.
#[derive(Debug, Clone)]
pub struct Ticket {
    pub task: serde_json::Value,
    pub labels: Vec<String>,
    pub schema: Option<crate::schemas::Schema>,
    pub(crate) key: String,
    pub(crate) status: Status,
    pub(crate) reporter: String,
    pub(crate) created_at: u64,
    /// Set when the ticket transitions `Todo → InProgress`. Millis
    /// since epoch.
    pub(crate) started_at: Option<u64>,
    /// Set when the ticket reaches `Status::Done`. Millis since epoch.
    /// Mutually exclusive with `failed_at`.
    pub(crate) finished_at: Option<u64>,
    /// Set when the ticket reaches `Status::Failed`. Millis since
    /// epoch. Mutually exclusive with `finished_at`.
    pub(crate) failed_at: Option<u64>,
    pub(crate) result: Option<TicketResult>,
}

/// Record an agent writes when it finishes a ticket. Carries the source
/// agent name, the ticket key, and the agent's result payload (a string
/// for tickets without a schema, otherwise the validated JSON value).
/// Same shape as the line `WriteResultTool` appends to `results.jsonl`.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct TicketResult {
    pub agent: String,
    pub ticket: String,
    pub result: serde_json::Value,
}

impl TicketResult {
    /// Result rendered as a String: the raw text for string payloads,
    /// canonical JSON for everything else. Lossless re-parsing is not
    /// required.
    pub fn result_string(&self) -> String {
        match &self.result {
            serde_json::Value::String(s) => s.clone(),
            other => other.to_string(),
        }
    }
}

impl Ticket {
    /// New ticket carrying `task` as its body. Use the chainable helpers
    /// (`label`, `labels`, `schema`) to populate caller-settable fields.
    /// System-managed fields are stamped by the ticket system at
    /// insertion time; the placeholders set here are overwritten.
    pub fn new<T: Serialize>(task: T) -> Self {
        let value = serde_json::to_value(task).expect("Ticket::new: value must serialize to JSON");
        Self {
            task: value,
            labels: Vec::new(),
            schema: None,
            key: String::new(),
            status: Status::Todo,
            reporter: String::new(),
            created_at: 0,
            started_at: None,
            finished_at: None,
            failed_at: None,
            result: None,
        }
    }

    /// Add a single label. Use [`Self::labels`] to add several at once.
    pub fn label(mut self, l: impl Into<String>) -> Self {
        self.labels.push(l.into());
        self
    }

    /// Add many labels at once.
    pub fn labels<I, S>(mut self, iter: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.labels.extend(iter.into_iter().map(Into::into));
        self
    }

    pub fn schema(mut self, schema: crate::schemas::Schema) -> Self {
        self.schema = Some(schema);
        self
    }

    // ---- read-only accessors for system-managed fields ----

    pub fn key(&self) -> &str {
        &self.key
    }

    /// Current status as a lowercase string: `"todo"`, `"in_progress"`,
    /// `"done"`, or `"failed"`. Use [`TicketSystem::set_done`] or
    /// [`TicketSystem::set_failed`] to drive transitions.
    pub fn status(&self) -> &'static str {
        self.status.as_str()
    }

    pub fn reporter(&self) -> &str {
        &self.reporter
    }

    pub fn created_at(&self) -> u64 {
        self.created_at
    }

    pub fn started_at(&self) -> Option<u64> {
        self.started_at
    }

    pub fn finished_at(&self) -> Option<u64> {
        self.finished_at
    }

    pub fn failed_at(&self) -> Option<u64> {
        self.failed_at
    }

    /// Elapsed duration from creation to terminal status (Done or
    /// Failed), `None` while the ticket has not yet reached one. Mirrors
    /// the naming convention of [`Stats::elapsed`].
    pub fn elapsed(&self) -> Option<Duration> {
        let terminal = self.finished_at.or(self.failed_at)?;
        Some(Duration::from_millis(
            terminal.saturating_sub(self.created_at),
        ))
    }

    pub fn result(&self) -> Option<&TicketResult> {
        self.result.as_ref()
    }

    /// Result payload rendered as a String, or `None` when the ticket has
    /// no recorded result. Convenience for callers that want a flat
    /// string view of the result regardless of the underlying JSON shape.
    pub fn result_string(&self) -> Option<String> {
        self.result.as_ref().map(TicketResult::result_string)
    }

    pub fn has_label(&self, label: &str) -> bool {
        self.labels.iter().any(|l| l == label)
    }
}

impl AsUserMessage for Ticket {
    fn as_user_message(&self) -> Message {
        let body = match &self.task {
            serde_json::Value::String(s) => s.clone(),
            other => serde_json::to_string_pretty(other).unwrap_or_default(),
        };
        Message::user(body)
    }
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Status {
    Todo,
    InProgress,
    Done,
    Failed,
}

impl Status {
    /// Lowercase wire form: `"todo"`, `"in_progress"`, `"done"`, `"failed"`.
    /// Single source of truth for the string rendering used by
    /// [`Ticket::status`] and the `tickets.jsonl` event log.
    pub fn as_str(self) -> &'static str {
        match self {
            Status::Todo => "todo",
            Status::InProgress => "in_progress",
            Status::Done => "done",
            Status::Failed => "failed",
        }
    }
}

#[derive(Debug)]
pub enum TicketError {
    TicketMissing { key: String },
    TransitionRejected { from: Status, to: Status },
}

impl fmt::Display for TicketError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TicketMissing { key } => write!(f, "Ticket {key} not found"),
            Self::TransitionRejected { from, to } => {
                write!(f, "Illegal transition {from:?} -> {to:?}")
            }
        }
    }
}

impl std::error::Error for TicketError {}

/// Public ticket system. Owns the shared ticket store, the registered
/// agents, the policies, the interrupt signal, and the run stats.
/// Always lives behind `Arc<TicketSystem>` — `new()` returns
/// `Arc<Self>` so each bound `Agent` can hold a `Weak<TicketSystem>`
/// without creating an Arc cycle through the system's `Vec<Agent>`.
pub struct TicketSystem {
    weak_self: Weak<TicketSystem>,
    pub(crate) tickets: Mutex<HashMap<String, Ticket>>,
    agents: Mutex<Vec<Agent>>,
    policies: Mutex<Policies>,
    pub(crate) interrupt_signal: Mutex<Arc<AtomicBool>>,
    pub(crate) stats: Stats,
    dir: Mutex<Option<PathBuf>>,
    tickets_log_lock: Mutex<()>,
}

impl TicketSystem {
    /// Build a fresh `TicketSystem` and return it inside an `Arc`. The
    /// system captures its own `Weak<Self>` via `Arc::new_cyclic` so
    /// `bind_agent` can hand out the back-reference each `Agent` needs
    /// at run time.
    pub fn new() -> Arc<Self> {
        Arc::new_cyclic(|weak| Self {
            weak_self: weak.clone(),
            tickets: Mutex::new(HashMap::new()),
            agents: Mutex::new(Vec::new()),
            policies: Mutex::new(Policies::default()),
            interrupt_signal: Mutex::new(Arc::new(AtomicBool::new(false))),
            stats: Stats::new(),
            dir: Mutex::new(None),
            tickets_log_lock: Mutex::new(()),
        })
    }

    /// Run-time counters. Read after `run` / `run_dry` returns.
    pub fn stats(&self) -> &Stats {
        &self.stats
    }

    // ---- policy builders ----

    pub fn max_steps(self: Arc<Self>, n: u32) -> Arc<Self> {
        self.policies.lock().unwrap().max_steps = Some(n);
        self
    }

    pub fn max_input_tokens(self: Arc<Self>, n: u64) -> Arc<Self> {
        self.policies.lock().unwrap().max_input_tokens = Some(n);
        self
    }

    pub fn max_output_tokens(self: Arc<Self>, n: u64) -> Arc<Self> {
        self.policies.lock().unwrap().max_output_tokens = Some(n);
        self
    }

    pub fn max_request_tokens(self: Arc<Self>, n: u32) -> Arc<Self> {
        self.policies.lock().unwrap().max_request_tokens = Some(n);
        self
    }

    pub fn max_schema_retries(self: Arc<Self>, n: u32) -> Arc<Self> {
        self.policies.lock().unwrap().max_schema_retries = Some(n);
        self
    }

    pub fn max_request_retries(self: Arc<Self>, n: u32) -> Arc<Self> {
        self.policies.lock().unwrap().max_request_retries = n;
        self
    }

    pub fn request_retry_delay(self: Arc<Self>, d: Duration) -> Arc<Self> {
        self.policies.lock().unwrap().request_retry_delay = d;
        self
    }

    /// Maximum elapsed duration `run_dry` will wait before tripping
    /// the interrupt signal and returning. Hitting the cap is a
    /// graceful stop, not a `PolicyViolated` event.
    pub fn max_time(self: Arc<Self>, d: Duration) -> Arc<Self> {
        self.policies.lock().unwrap().max_time = Some(d);
        self
    }

    /// Override the cancel signal. Useful when a caller wants to share
    /// one `Arc<AtomicBool>` across multiple subsystems.
    pub fn interrupt_signal(self: Arc<Self>, signal: Arc<AtomicBool>) -> Arc<Self> {
        *self.interrupt_signal.lock().unwrap() = signal;
        self
    }

    /// Directory under which the system writes `results.jsonl` and
    /// `tickets.jsonl`. Knowledge lives next to them when the caller points
    /// `Knowledge::open` at the same directory. When unset, `WriteResultTool`
    /// falls back to the calling agent's directory and the ticket
    /// event log is skipped entirely.
    pub fn dir(self: Arc<Self>, dir: impl Into<PathBuf>) -> Arc<Self> {
        *self.dir.lock().unwrap() = Some(dir.into());
        self
    }

    pub(crate) fn dir_value(&self) -> Option<PathBuf> {
        self.dir.lock().unwrap().clone()
    }

    // ---- ticket-creation API mirrored on Agent ----

    /// Enqueue a ticket carrying `task` as its body.
    pub fn task<T: Serialize>(&self, task: T) -> &Self {
        self.dispatch(Ticket::new(task));
        self
    }

    /// Enqueue a ticket carrying `task`, attached to `label` for Path B
    /// routing.
    pub fn task_labeled<T: Serialize>(&self, task: T, label: impl Into<String>) -> &Self {
        self.dispatch(Ticket::new(task).label(label));
        self
    }

    /// Enqueue a ticket whose final `done` result must validate against
    /// `schema`.
    pub fn task_schema<T: Serialize>(&self, task: T, schema: crate::schemas::Schema) -> &Self {
        self.dispatch(Ticket::new(task).schema(schema));
        self
    }

    /// `task_schema` + `task_labeled` combined.
    pub fn task_schema_labeled<T: Serialize>(
        &self,
        task: T,
        schema: crate::schemas::Schema,
        label: impl Into<String>,
    ) -> &Self {
        self.dispatch(Ticket::new(task).schema(schema).label(label));
        self
    }

    /// Enqueue a fully-built `Ticket`. System-managed fields (key,
    /// reporter, created_at, status, result) are overwritten. To pin the
    /// ticket to a specific agent, label it with the agent's name.
    pub fn ticket(&self, ticket: Ticket) -> &Self {
        self.dispatch(ticket);
        self
    }

    fn dispatch(&self, ticket: Ticket) {
        self.insert(ticket, "user".to_string());
    }

    // ---- inherent ticket-store methods ----

    /// Insert `ticket`, stamping system fields. The ticket is always born
    /// `Todo`; to pin it to a specific agent, label it with the agent's
    /// name. Returns the inserted ticket's key.
    pub(crate) fn insert(&self, mut ticket: Ticket, reporter: String) -> String {
        let mut store = self.tickets.lock().unwrap();
        let id = store.len() + 1;
        ticket.key = format!("TICKET-{id}");
        ticket.created_at = now_millis();
        ticket.reporter = reporter;
        ticket.result = None;
        ticket.status = Status::Todo;
        let key = ticket.key.clone();
        let labels = ticket.labels.clone();
        let reporter = ticket.reporter.clone();
        let task = ticket.task.clone();
        let created_at = ticket.created_at;
        store.insert(key.clone(), ticket);
        drop(store);
        TicketStats::record_created(&self.stats);
        for l in &labels {
            let slice = self.stats.stats_for_label(l);
            TicketStats::record_created(&*slice);
        }
        self.append_ticket_event(serde_json::json!({
            "event": "created",
            "ts": created_at,
            "key": key,
            "reporter": reporter,
            "labels": labels,
            "task": task,
        }));
        key
    }

    /// Append one JSON line to `<dir>/tickets.jsonl`. Silently no-ops
    /// when no directory is configured. Errors are swallowed: the log is
    /// observational, not load-bearing for run correctness.
    pub(crate) fn append_ticket_event(&self, event: serde_json::Value) {
        let Some(dir) = self.dir_value() else {
            return;
        };
        let _guard = self.tickets_log_lock.lock().unwrap();
        let _ = append_ticket_event_to_dir(&dir, &event);
    }

    /// Returns a clone of the ticket at `key`, if any.
    pub fn get(&self, key: &str) -> Option<Ticket> {
        self.tickets.lock().unwrap().get(key).cloned()
    }

    /// Atomically find a `Todo` ticket matching `predicate`, label it
    /// with `agent_name`, and transition to `InProgress`.
    pub(crate) fn claim<F>(&self, predicate: F, agent_name: &str) -> Option<String>
    where
        F: Fn(&Ticket) -> bool,
    {
        let now = now_millis();
        let (key, prev, durations, labels) = {
            let mut store = self.tickets.lock().unwrap();
            let mut candidates: Vec<&String> = store
                .iter()
                .filter(|(_, t)| predicate(t))
                .map(|(k, _)| k)
                .collect();
            candidates.sort_by_key(|k| {
                let t = &store[k.as_str()];
                (t.created_at, numeric_id(k))
            });
            let key = candidates.into_iter().next()?.clone();
            let ticket = store.get_mut(&key)?;
            if ticket.status != Status::Todo {
                return None;
            }
            if !ticket.labels.iter().any(|l| l == agent_name) {
                ticket.labels.push(agent_name.to_string());
            }
            let prev = ticket.status;
            stamp_transition_timestamps(ticket, Status::InProgress, now);
            ticket.status = Status::InProgress;
            let durations = terminal_durations(ticket);
            let labels = ticket.labels.clone();
            (key, prev, durations, labels)
        };
        self.record_transition(&key, prev, Status::InProgress, now, durations, &labels);
        Some(key)
    }

    /// Transition a ticket to `Done`.
    pub(crate) fn set_done(&self, key: &str) -> Result<(), TicketError> {
        self.set_final_status(key, Status::Done)
    }

    /// Transition a ticket to `Failed`.
    pub(crate) fn set_failed(&self, key: &str) -> Result<(), TicketError> {
        self.set_final_status(key, Status::Failed)
    }

    fn set_final_status(&self, key: &str, status: Status) -> Result<(), TicketError> {
        let now = now_millis();
        let (prev, durations, labels) = {
            let mut store = self.tickets.lock().unwrap();
            let ticket = store
                .get_mut(key)
                .ok_or_else(|| TicketError::TicketMissing {
                    key: key.to_string(),
                })?;
            let prev = ticket.status;
            stamp_transition_timestamps(ticket, status, now);
            ticket.status = status;
            let durations = terminal_durations(ticket);
            let labels = ticket.labels.clone();
            (prev, durations, labels)
        };
        self.record_transition(key, prev, status, now, durations, &labels);
        Ok(())
    }

    /// Fire stats recorders and ticket log after a transition.
    fn record_transition(
        &self,
        key: &str,
        prev: Status,
        next: Status,
        now: u64,
        durations: (Duration, Duration),
        labels: &[String],
    ) {
        fire_transition_recorder(&self.stats, prev, next, now, durations);
        fire_label_transition(&self.stats, next, durations, labels);
        self.log_transition(key, prev, next, now, durations, labels);
    }

    /// Append a `started` / `done` / `failed` line to `tickets.jsonl` if
    /// `prev → next` is observable. No-op when prev == next or when the
    /// transition is not one we surface.
    fn log_transition(
        &self,
        key: &str,
        prev: Status,
        next: Status,
        ts: u64,
        (ticket_duration, work_duration): (Duration, Duration),
        labels: &[String],
    ) {
        if prev == next {
            return;
        }
        if prev == Status::Todo && next == Status::InProgress {
            self.append_ticket_event(serde_json::json!({
                "event": "started",
                "ts": ts,
                "key": key,
                "labels": labels,
            }));
        }
        match next {
            Status::Done | Status::Failed => {
                let event = if next == Status::Done {
                    "done"
                } else {
                    "failed"
                };
                self.append_ticket_event(serde_json::json!({
                    "event": event,
                    "ts": ts,
                    "key": key,
                    "duration_ms": ticket_duration.as_millis() as u64,
                    "work_ms": work_duration.as_millis() as u64,
                }));
            }
            _ => {}
        }
    }

    /// Attach a `TicketResult` to the ticket at `key`.
    pub(crate) fn set_result(&self, key: &str, result: TicketResult) -> Result<(), TicketError> {
        let mut store = self.tickets.lock().unwrap();
        let ticket = store
            .get_mut(key)
            .ok_or_else(|| TicketError::TicketMissing {
                key: key.to_string(),
            })?;
        ticket.result = Some(result);
        Ok(())
    }

    /// Edit caller-settable fields. Each `Some` overwrites; `None`
    /// leaves the field untouched. The `Option<Option<Schema>>` shape on
    /// `schema` lets callers explicitly clear it via `Some(None)`.
    pub(crate) fn edit(
        &self,
        key: &str,
        task: Option<serde_json::Value>,
        labels: Option<Vec<String>>,
        schema: Option<Option<crate::schemas::Schema>>,
    ) -> Result<(), TicketError> {
        let mut store = self.tickets.lock().unwrap();
        let ticket = store
            .get_mut(key)
            .ok_or_else(|| TicketError::TicketMissing {
                key: key.to_string(),
            })?;
        if let Some(t) = task {
            ticket.task = t;
        }
        if let Some(l) = labels {
            ticket.labels = l;
        }
        if let Some(s) = schema {
            ticket.schema = s;
        }
        Ok(())
    }

    /// Snapshot of every ticket, sorted by creation time then numeric key.
    pub fn tickets(&self) -> Vec<Ticket> {
        let tickets = self.tickets.lock().unwrap();
        let mut out: Vec<Ticket> = tickets.values().cloned().collect();
        out.sort_by_key(|t| (t.created_at, numeric_id(&t.key)));
        out
    }

    /// Earliest ticket by creation time, if any.
    pub fn first(&self) -> Option<Ticket> {
        self.tickets().into_iter().next()
    }

    /// Substring search over the task body, case-insensitive.
    pub fn search(&self, query: &str) -> Vec<Ticket> {
        let needle = query.to_lowercase();
        let store = self.tickets.lock().unwrap();
        let mut out: Vec<Ticket> = store
            .values()
            .filter(|t| match &t.task {
                serde_json::Value::String(s) => s.to_lowercase().contains(&needle),
                other => other.to_string().to_lowercase().contains(&needle),
            })
            .cloned()
            .collect();
        out.sort_by_key(|t| (t.created_at, numeric_id(&t.key)));
        out
    }

    /// Tickets matching `predicate`, sorted by creation time then numeric key.
    ///
    /// The predicate runs while `self.tickets` is locked. It MUST NOT call
    /// other `TicketSystem` methods that lock the same `Mutex` — deadlock.
    pub fn filter<F>(&self, predicate: F) -> Vec<Ticket>
    where
        F: Fn(&Ticket) -> bool,
    {
        let store = self.tickets.lock().unwrap();
        let mut out: Vec<Ticket> = store.values().filter(|t| predicate(t)).cloned().collect();
        out.sort_by_key(|t| (t.created_at, numeric_id(&t.key)));
        out
    }

    /// First ticket matching `predicate`, by creation order. Short-circuits.
    ///
    /// The predicate runs while `self.tickets` is locked. It MUST NOT call
    /// other `TicketSystem` methods that lock the same `Mutex` — deadlock.
    pub fn find<F>(&self, predicate: F) -> Option<Ticket>
    where
        F: Fn(&Ticket) -> bool,
    {
        let store = self.tickets.lock().unwrap();
        let mut matching: Vec<&Ticket> = store.values().filter(|t| predicate(t)).collect();
        matching.sort_by_key(|t| (t.created_at, numeric_id(&t.key)));
        matching.into_iter().next().cloned()
    }

    /// Count of tickets matching `predicate`. Does not allocate.
    ///
    /// The predicate runs while `self.tickets` is locked. It MUST NOT call
    /// other `TicketSystem` methods that lock the same `Mutex` — deadlock.
    pub fn count<F>(&self, predicate: F) -> usize
    where
        F: Fn(&Ticket) -> bool,
    {
        self.tickets
            .lock()
            .unwrap()
            .values()
            .filter(|t| predicate(t))
            .count()
    }

    /// Snapshot of the active policies for the loop's per-step guards.
    pub(crate) fn policies(&self) -> Policies {
        self.policies.lock().unwrap().clone()
    }

    /// Wire `agent` to this system. Drains any tickets the agent had
    /// queued in its private default system into this one, then stamps
    /// the system's `Weak<Self>` onto `agent.ticket_system`.
    pub(crate) fn bind_agent(&self, agent: &mut Agent) {
        if let Some(prior) = agent.ticket_system.upgrade() {
            if !Arc::ptr_eq(
                &prior,
                &self
                    .weak_self
                    .upgrade()
                    .expect("self Arc dropped during bind"),
            ) {
                let drained: Vec<Ticket> = {
                    let mut old = prior.tickets.lock().unwrap();
                    std::mem::take(&mut *old).into_values().collect()
                };
                let reporter = agent.name.clone();
                for ticket in drained {
                    self.insert(ticket, reporter.clone());
                }
            }
        }
        agent.ticket_system = self.weak_self.clone();
        agent.ensure_knowledge_bound();
        self.agents.lock().unwrap().push(agent.clone());
    }
}

impl TicketSystem {
    /// Clone of the currently registered agent list. The list is
    /// append-only by invariant: `bind_agent` is the sole mutator and
    /// only calls `push`. `run_main_loop` relies on element indices
    /// being stable across calls. Any new mutator that removes or
    /// reorders entries would silently break late-add detection: route
    /// additions through `bind_agent` only.
    pub(super) fn clone_agents(&self) -> Vec<Agent> {
        self.agents.lock().unwrap().clone()
    }
}

impl TicketSystem {
    /// Bind `agent` to this system: drain any tickets it queued in its
    /// default system into this one and push a clone onto this system's
    /// agents list. Returns the wired agent for chaining (`.task(...)`
    /// etc.).
    ///
    /// May be called before or after `run()` / `run_dry()`. When called
    /// after `run()`, the new agent starts polling for tickets within
    /// roughly one `IDLE_POLL_INTERVAL` (~100 ms).
    pub fn agent(&self, mut agent: Agent) -> Agent {
        self.bind_agent(&mut agent);
        agent
    }

    /// Start the agent loop on a background tokio task and return a
    /// [`Running`] handle. The handle owns the interrupt signal;
    /// tickets queued afterwards are picked up within
    /// ~`IDLE_POLL_INTERVAL`. Finish with [`Running::run_dry`] to wait
    /// for the queue to drain, or [`Running::stop`] +
    /// [`Running::join`] for an abrupt cancel.
    ///
    /// [`Running`]: super::running::Running
    /// [`Running::run_dry`]: super::running::Running::run_dry
    /// [`Running::stop`]: super::running::Running::stop
    /// [`Running::join`]: super::running::Running::join
    pub fn run(&self) -> super::running::Running {
        let signal = Arc::clone(&self.interrupt_signal.lock().unwrap());
        // Reset so a system can be re-run after a previous run_dry left
        // the flag set.
        signal.store(false, Ordering::Relaxed);
        let system = self
            .weak_self
            .upgrade()
            .expect("TicketSystem dropped during run");
        let supervisor_system = Arc::clone(&system);
        let join = tokio::spawn(async move {
            run_main_loop(&supervisor_system).await;
            supervisor_system.stats.mark_finished(now_millis());
        });
        super::running::Running::new(system, signal, join)
    }

    /// Start a background run and wait for the queue to drain.
    /// Returns every finished ticket's [`TicketResult`], in creation
    /// order. Equivalent to `self.run().run_dry().await`.
    pub fn run_dry(&self) -> impl std::future::Future<Output = Vec<TicketResult>> + Send {
        self.run().run_dry()
    }

    /// Every `Done` ticket's `TicketResult`, in ticket creation order.
    /// Tickets that finished without a recorded result are skipped.
    /// Backing helper for `Running::run_dry`.
    pub(crate) fn collect_results(&self) -> Vec<TicketResult> {
        self.filter(|t| t.status == Status::Done)
            .into_iter()
            .filter_map(|t| t.result)
            .collect()
    }
}

/// Whether the run-wide policies have been exceeded by the current
/// stats reading. Used by the `run_dry` watcher and by the per-agent
/// loop's pre-claim check.
pub(crate) fn policy_violated(policies: &Policies, stats: &Stats) -> bool {
    if let Some(limit) = policies.max_steps {
        if stats.steps() >= u64::from(limit) {
            return true;
        }
    }
    if let Some(limit) = policies.max_input_tokens {
        if stats.input_tokens() >= limit {
            return true;
        }
    }
    if let Some(limit) = policies.max_output_tokens {
        if stats.output_tokens() >= limit {
            return true;
        }
    }
    false
}

/// Same as [`policy_violated`] but returns which policy tripped and its
/// configured limit, for the `PolicyViolated` event.
pub(crate) fn policy_violated_kind(
    policies: &Policies,
    stats: &Stats,
) -> Option<(crate::event::PolicyKind, u64)> {
    use crate::event::PolicyKind;
    if let Some(limit) = policies.max_steps {
        if stats.steps() >= u64::from(limit) {
            return Some((PolicyKind::Steps, u64::from(limit)));
        }
    }
    if let Some(limit) = policies.max_input_tokens {
        if stats.input_tokens() >= limit {
            return Some((PolicyKind::InputTokens, limit));
        }
    }
    if let Some(limit) = policies.max_output_tokens {
        if stats.output_tokens() >= limit {
            return Some((PolicyKind::OutputTokens, limit));
        }
    }
    None
}

pub(crate) fn pending_count(ticket_system: &TicketSystem) -> usize {
    ticket_system
        .tickets
        .lock()
        .unwrap()
        .values()
        .filter(|t| matches!(t.status, Status::Todo | Status::InProgress))
        .count()
}

/// Stamp `started_at` / `finished_at` / `failed_at` on a ticket whose
/// status is about to flip. Called inside the locked critical section.
fn stamp_transition_timestamps(ticket: &mut Ticket, next: Status, now: u64) {
    if ticket.status == Status::Todo && next == Status::InProgress {
        ticket.started_at = Some(now);
    }
    match next {
        Status::Done => {
            ticket.finished_at = Some(now);
        }
        Status::Failed => {
            ticket.failed_at = Some(now);
        }
        _ => {}
    }
}

/// Compute (ticket_duration, work_duration) for a ticket that just
/// reached a terminal status. `ticket_duration` is creation→terminal;
/// `work_duration` is started→terminal. Both default to zero if the
/// relevant timestamps aren't both set.
fn terminal_durations(ticket: &Ticket) -> (Duration, Duration) {
    let ticket_duration = ticket.elapsed().unwrap_or_default();
    let work_duration = match (ticket.started_at, ticket.finished_at.or(ticket.failed_at)) {
        (Some(start), Some(end)) => Duration::from_millis(end.saturating_sub(start)),
        _ => Duration::ZERO,
    };
    (ticket_duration, work_duration)
}

/// Fire the appropriate recorder hook for a status transition. Called
/// after the lock is released.
fn fire_transition_recorder(
    stats: &Stats,
    prev: Status,
    next: Status,
    now: u64,
    (ticket_duration, work_duration): (Duration, Duration),
) {
    if prev == next {
        return;
    }
    if prev == Status::Todo && next == Status::InProgress {
        stats.record_started(now);
    }
    match next {
        Status::Done => stats.record_done(ticket_duration, work_duration),
        Status::Failed => stats.record_failed(ticket_duration, work_duration),
        _ => {}
    }
}

/// Mirror a terminal transition onto every per-label slice the ticket
/// carries. `record_started` is intentionally not mirrored: per-label
/// `started_at` stays zero so `elapsed()` reads `None` on a slice.
fn fire_label_transition(
    stats: &Stats,
    next: Status,
    (ticket_duration, work_duration): (Duration, Duration),
    labels: &[String],
) {
    if !matches!(next, Status::Done | Status::Failed) {
        return;
    }
    for l in labels {
        let slice = stats.stats_for_label(l);
        match next {
            Status::Done => slice.record_done(ticket_duration, work_duration),
            Status::Failed => slice.record_failed(ticket_duration, work_duration),
            _ => unreachable!(),
        }
    }
}

pub(crate) fn now_millis() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

fn numeric_id(key: &str) -> u32 {
    key.rsplit('-')
        .next()
        .and_then(|s| s.parse::<u32>().ok())
        .unwrap_or(u32::MAX)
}

const TICKETS_LOG_FILE: &str = "tickets.jsonl";

fn append_ticket_event_to_dir(
    dir: &std::path::Path,
    event: &serde_json::Value,
) -> std::io::Result<()> {
    std::fs::create_dir_all(dir)?;
    let mut line = serde_json::to_string(event).map_err(std::io::Error::other)?;
    line.push('\n');
    let mut file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(dir.join(TICKETS_LOG_FILE))?;
    use std::io::Write as _;
    file.write_all(line.as_bytes())
}

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

    fn task_ticket(label: &str) -> Ticket {
        Ticket::new(format!("body-{label}")).label(label)
    }

    fn attach_done_result(sys: &TicketSystem, key: &str, agent: &str, result: &str) {
        sys.set_result(
            key,
            TicketResult {
                agent: agent.into(),
                ticket: key.into(),
                result: serde_json::Value::String(result.into()),
            },
        )
        .unwrap();
        sys.set_done(key).unwrap();
    }

    #[test]
    fn task_creates_ticket_with_user_reporter() {
        let sys = TicketSystem::new();
        sys.task("hello");
        let t = sys.get("TICKET-1").unwrap();
        assert_eq!(t.task, serde_json::Value::String("hello".into()));
        assert_eq!(t.reporter(), "user");
        assert_eq!(t.status, Status::Todo);
    }

    #[test]
    fn task_labeled_attaches_label_and_leaves_status_todo() {
        let sys = TicketSystem::new();
        sys.task_labeled("hello", "research");
        let t = sys.get("TICKET-1").unwrap();
        assert_eq!(t.labels, vec!["research".to_string()]);
        assert_eq!(t.status, Status::Todo);
    }

    #[test]
    fn create_with_named_label_is_born_todo_and_carries_label() {
        let sys = TicketSystem::new();
        sys.ticket(Ticket::new("specific work for alice").label("alice"));
        let t = sys.get("TICKET-1").unwrap();
        assert!(t.has_label("alice"));
        assert_eq!(t.status, Status::Todo);
    }

    #[test]
    fn create_with_label_and_schema_is_stored_verbatim() {
        let sys = TicketSystem::new();
        let schema = crate::schemas::Schema::parse(serde_json::json!({"type": "string"})).unwrap();
        sys.ticket(Ticket::new("x").label("urgent").schema(schema));
        let t = sys.get("TICKET-1").unwrap();
        assert_eq!(t.labels, vec!["urgent".to_string()]);
        assert!(t.schema.is_some());
    }

    #[test]
    fn ticket_system_handle_is_shared_between_caller_and_added_agent() {
        let sys = TicketSystem::new();
        let alice = sys.agent(Agent::new().name("alice"));
        // Alice's task lands in the same queue.
        alice.task("from alice");
        sys.task("from system");
        let all_keys: Vec<String> = sys
            .filter(|t| t.status == Status::Todo)
            .iter()
            .map(|t| t.key().to_string())
            .collect();
        assert_eq!(all_keys.len(), 2);
    }

    #[test]
    fn agent_must_be_bound_before_task() {
        let alice = Agent::new().name("alice");
        let sys = TicketSystem::new();
        let alice = sys.agent(alice);
        // Bound — task() works, lands in the shared queue.
        alice.task("first").task("second");
        assert_eq!(sys.count(|t| t.status == Status::Todo), 2);
    }

    #[test]
    #[should_panic(expected = "Agent::task requires a bound TicketSystem")]
    fn unbound_agent_task_panics() {
        let alice = Agent::new().name("alice");
        alice.task("never lands");
    }

    #[test]
    fn search_matches_string_task_case_insensitively() {
        let sys = TicketSystem::new();
        sys.task("Fix Login");
        sys.task("Other thing");
        let hits = sys.search("login");
        assert_eq!(hits.len(), 1);
    }

    #[test]
    fn ticket_label_helpers_compose() {
        let t = task_ticket("research").label("urgent");
        assert_eq!(t.labels, vec!["research".to_string(), "urgent".to_string()]);
    }

    #[test]
    fn set_result_updates_ticket() {
        let sys = TicketSystem::new();
        sys.task("hi");
        sys.set_result(
            "TICKET-1",
            TicketResult {
                agent: "tester".into(),
                ticket: "TICKET-1".into(),
                result: serde_json::Value::String("answer".into()),
            },
        )
        .unwrap();
        let stored = sys.get("TICKET-1").unwrap();
        let attached = stored.result().unwrap();
        assert_eq!(attached.agent, "tester");
        assert_eq!(attached.ticket, "TICKET-1");
        assert_eq!(attached.result, serde_json::Value::String("answer".into()));
    }

    #[test]
    fn first_returns_none_on_empty_system() {
        let sys = TicketSystem::new();
        assert!(sys.first().is_none());
        assert!(sys.tickets().is_empty());
    }

    #[test]
    fn first_returns_earliest_ticket_by_creation() {
        let sys = TicketSystem::new();
        sys.task("first").task("second").task("third");
        let first = sys.first().unwrap();
        assert_eq!(first.key(), "TICKET-1");
        assert_eq!(first.task, serde_json::Value::String("first".into()));
    }

    #[test]
    fn tickets_returns_all_in_creation_order() {
        let sys = TicketSystem::new();
        sys.task("a").task("b").task("c");
        let all = sys.tickets();
        assert_eq!(all.len(), 3);
        assert_eq!(all[0].key(), "TICKET-1");
        assert_eq!(all[1].key(), "TICKET-2");
        assert_eq!(all[2].key(), "TICKET-3");
    }

    #[test]
    fn collect_results_returns_results_in_creation_order() {
        let sys = TicketSystem::new();
        sys.task("a").task("b").task("c");
        attach_done_result(&sys, "TICKET-1", "alice", "first");
        attach_done_result(&sys, "TICKET-3", "alice", "third");
        let results = sys.collect_results();
        assert_eq!(results.len(), 2);
        assert_eq!(results[0].ticket, "TICKET-1");
        assert_eq!(results[0].result, serde_json::Value::String("first".into()));
        assert_eq!(results[1].ticket, "TICKET-3");
        assert_eq!(results[1].result, serde_json::Value::String("third".into()));
    }

    #[test]
    fn collect_results_last_is_most_recently_created_done() {
        let sys = TicketSystem::new();
        sys.task("a").task("b");
        attach_done_result(&sys, "TICKET-2", "alice", "second");
        attach_done_result(&sys, "TICKET-1", "alice", "first");
        let last = sys
            .collect_results()
            .last()
            .cloned()
            .expect("expected last result");
        assert_eq!(last.ticket, "TICKET-2");
        assert_eq!(last.result, serde_json::Value::String("second".into()));
    }

    #[test]
    fn collect_results_is_empty_when_nothing_done() {
        let sys = TicketSystem::new();
        sys.task("pending");
        assert!(sys.collect_results().is_empty());
    }

    #[test]
    fn done_and_failed_filter_by_status() {
        let sys = TicketSystem::new();
        sys.task("ok").task("oops").task("pending");
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        sys.set_failed("TICKET-2").unwrap();
        let done = sys.filter(|t| t.status == Status::Done);
        let failed = sys.filter(|t| t.status == Status::Failed);
        assert_eq!(done.len(), 1);
        assert_eq!(done[0].key(), "TICKET-1");
        assert_eq!(failed.len(), 1);
        assert_eq!(failed[0].key(), "TICKET-2");
    }

    #[test]
    fn ticket_status_transitions_record_stats() {
        let sys = TicketSystem::new();
        sys.task("a").task("b").task("c");
        // Created 3 tickets.
        assert_eq!(sys.stats().tickets_created(), 3);
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        sys.claim(|t| t.key() == "TICKET-2", "agent");
        sys.set_failed("TICKET-2").unwrap();
        assert_eq!(sys.stats().tickets_done(), 1);
        assert_eq!(sys.stats().tickets_failed(), 1);
    }

    #[test]
    fn stats_for_label_counts_creation_per_label() {
        let sys = TicketSystem::new();
        sys.ticket(Ticket::new("a").labels(["scan", "high"]));
        sys.ticket(Ticket::new("b").label("scan"));
        sys.ticket(Ticket::new("c"));
        let stats = sys.stats();
        assert_eq!(stats.tickets_created(), 3);
        assert_eq!(stats.stats_for_label("scan").tickets_created(), 2);
        assert_eq!(stats.stats_for_label("high").tickets_created(), 1);
        assert_eq!(stats.stats_for_label("never-used").tickets_created(), 0);
    }

    #[test]
    fn stats_for_label_counts_terminal_transitions_per_label() {
        let sys = TicketSystem::new();
        sys.ticket(Ticket::new("a").labels(["scan", "high"]));
        sys.ticket(Ticket::new("b").label("scan"));
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        sys.claim(|t| t.key() == "TICKET-2", "agent");
        sys.set_failed("TICKET-2").unwrap();
        let stats = sys.stats();
        let scan = stats.stats_for_label("scan");
        let high = stats.stats_for_label("high");
        assert_eq!(scan.tickets_done(), 1);
        assert_eq!(scan.tickets_failed(), 1);
        assert_eq!(scan.tickets_success_rate(), Some(0.5));
        assert_eq!(high.tickets_done(), 1);
        assert_eq!(high.tickets_failed(), 0);
        assert_eq!(high.tickets_success_rate(), Some(1.0));
    }

    #[test]
    fn stats_for_label_set_failed_path_records_per_label() {
        let sys = TicketSystem::new();
        sys.ticket(Ticket::new("a").label("scan"));
        sys.set_failed("TICKET-1").unwrap();
        assert_eq!(sys.stats().stats_for_label("scan").tickets_failed(), 1);
    }

    #[test]
    fn stats_for_label_unaffected_by_no_label_ticket() {
        let sys = TicketSystem::new();
        sys.ticket(Ticket::new("a"));
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        assert_eq!(sys.stats().tickets_done(), 1);
        assert_eq!(sys.stats().stats_for_label("scan").tickets_done(), 0);
        assert_eq!(sys.stats().stats_for_label("scan").tickets_created(), 0);
    }

    fn read_tickets_log(dir: &std::path::Path) -> Vec<serde_json::Value> {
        std::fs::read_to_string(dir.join("tickets.jsonl"))
            .unwrap()
            .lines()
            .filter(|l| !l.trim().is_empty())
            .map(|l| serde_json::from_str(l).unwrap())
            .collect()
    }

    #[test]
    fn workspace_unset_skips_tickets_log() {
        let sys = TicketSystem::new();
        sys.task("hello");
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        // No workspace, no panic, no file: this asserts the no-op path.
        assert!(sys.dir_value().is_none());
    }

    #[test]
    fn workspace_emits_created_started_done_in_order() {
        let dir = tempfile::tempdir().unwrap();
        let sys = TicketSystem::new().dir(dir.path().to_path_buf());
        sys.task("hello");
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        let lines = read_tickets_log(dir.path());
        assert_eq!(lines.len(), 3);
        assert_eq!(lines[0]["event"], "created");
        assert_eq!(lines[0]["key"], "TICKET-1");
        assert_eq!(lines[0]["reporter"], "user");
        assert_eq!(lines[0]["task"], "hello");
        assert_eq!(lines[1]["event"], "started");
        assert_eq!(lines[1]["key"], "TICKET-1");
        assert_eq!(lines[2]["event"], "done");
        assert_eq!(lines[2]["key"], "TICKET-1");
        assert!(lines[2]["duration_ms"].is_u64());
        assert!(lines[2]["work_ms"].is_u64());
    }

    #[test]
    fn workspace_emits_failed_event_on_set_failed() {
        let dir = tempfile::tempdir().unwrap();
        let sys = TicketSystem::new().dir(dir.path().to_path_buf());
        sys.task("hello");
        sys.set_failed("TICKET-1").unwrap();
        let lines = read_tickets_log(dir.path());
        // created + failed (no started since Todo→Failed via set_failed)
        assert_eq!(lines.len(), 2);
        assert_eq!(lines[0]["event"], "created");
        assert_eq!(lines[1]["event"], "failed");
        assert_eq!(lines[1]["key"], "TICKET-1");
    }

    #[test]
    fn workspace_created_event_carries_labels_when_pinned() {
        let dir = tempfile::tempdir().unwrap();
        let sys = TicketSystem::new().dir(dir.path().to_path_buf());
        sys.ticket(Ticket::new("specific").label("alice"));
        let lines = read_tickets_log(dir.path());
        assert_eq!(lines.len(), 1);
        assert_eq!(lines[0]["event"], "created");
        assert_eq!(lines[0]["labels"], serde_json::json!(["alice"]));
    }

    #[test]
    fn workspace_logs_one_line_per_lifecycle_step_for_multiple_tickets() {
        let dir = tempfile::tempdir().unwrap();
        let sys = TicketSystem::new().dir(dir.path().to_path_buf());
        sys.task("a").task("b");
        sys.claim(|t| t.key() == "TICKET-1", "agent");
        sys.set_done("TICKET-1").unwrap();
        sys.claim(|t| t.key() == "TICKET-2", "agent");
        sys.set_failed("TICKET-2").unwrap();
        let lines = read_tickets_log(dir.path());
        // 2 created + 2 started + 1 done + 1 failed
        assert_eq!(lines.len(), 6);
    }

    // ---- claim / set_done / set_failed unit tests ----

    #[test]
    fn claim_transitions_todo_to_in_progress_and_adds_label() {
        let sys = TicketSystem::new();
        sys.task("hello");
        let key = sys.claim(|t| t.status == Status::Todo, "alice").unwrap();
        assert_eq!(key, "TICKET-1");
        let t = sys.get(&key).unwrap();
        assert_eq!(t.status, Status::InProgress);
        assert!(t.has_label("alice"));
        assert!(t.started_at().is_some());
    }

    #[test]
    fn claim_returns_none_when_no_ticket_matches() {
        let sys = TicketSystem::new();
        sys.task("hello");
        assert!(sys.claim(|t| t.has_label("nonexistent"), "alice").is_none());
    }

    #[test]
    fn second_claim_of_same_ticket_returns_none() {
        let sys = TicketSystem::new();
        sys.task("hello");
        let first = sys.claim(|t| t.key() == "TICKET-1", "alice");
        assert!(first.is_some());
        // Second claim: ticket is now InProgress, not Todo.
        let second = sys.claim(|t| t.key() == "TICKET-1", "bob");
        assert!(second.is_none());
    }

    #[test]
    fn claim_picks_earliest_eligible_ticket() {
        let sys = TicketSystem::new();
        sys.task("a").task("b").task("c");
        let key = sys.claim(|t| t.status == Status::Todo, "alice").unwrap();
        assert_eq!(key, "TICKET-1");
    }

    #[test]
    fn claim_emits_started_event_in_workspace_log() {
        let dir = tempfile::tempdir().unwrap();
        let sys = TicketSystem::new().dir(dir.path().to_path_buf());
        sys.task("hello");
        sys.claim(|t| t.status == Status::Todo, "alice");
        let lines = read_tickets_log(dir.path());
        assert_eq!(lines.len(), 2);
        assert_eq!(lines[0]["event"], "created");
        assert_eq!(lines[1]["event"], "started");
        assert_eq!(lines[1]["key"], "TICKET-1");
    }

    #[test]
    fn set_done_transitions_to_done() {
        let sys = TicketSystem::new();
        sys.task("hello");
        sys.claim(|t| t.status == Status::Todo, "alice");
        sys.set_done("TICKET-1").unwrap();
        let t = sys.get("TICKET-1").unwrap();
        assert_eq!(t.status, Status::Done);
        assert!(t.finished_at().is_some());
    }

    #[test]
    fn set_failed_transitions_to_failed() {
        let sys = TicketSystem::new();
        sys.task("hello");
        sys.claim(|t| t.status == Status::Todo, "alice");
        sys.set_failed("TICKET-1").unwrap();
        let t = sys.get("TICKET-1").unwrap();
        assert_eq!(t.status, Status::Failed);
        assert!(t.failed_at().is_some());
    }
}