ai-dispatch 10.30.0

// Watcher stream handlers for per-line event processing.
// Strips terminal escapes off each line before any event parsing.
// Exports shared streaming helpers used by child-process and PTY watchers.

use anyhow::Result;
use std::sync::Arc;

use crate::agent::Agent;
use crate::rate_limit;
use crate::store::Store;
use crate::delivery_guard::{DeliveryOutcome, DeliveryOutcome::MissingFinalDelivery};
use crate::types::{
    CompletionInfo, DeliveryAssessment, EventKind, TaskEvent, TaskId, TaskStatus,
};

use super::extract::{append_to_broadcast, extract_finding_detail, parse_milestone_event};
use super::{apply_completion_event, SyntheticMilestoneTracker};

pub(crate) fn apply_codex_delivery_guard(
    store: &Arc<Store>,
    task_id: &TaskId,
    status: TaskStatus,
    outcome: DeliveryOutcome,
    exit_code: Option<i32>,
) -> TaskStatus {
    let MissingFinalDelivery {
        last_work_kind,
        last_message_chars,
    } = outcome
    else {
        return status;
    };
    if status != TaskStatus::Done {
        return status;
    }
    let _ = store.update_delivery_assessment(
        task_id.as_str(),
        Some(DeliveryAssessment::MissingFinalDelivery),
    );
    let _ = store.insert_event(&TaskEvent {
        task_id: task_id.clone(),
        timestamp: chrono::Local::now(),
        event_kind: EventKind::Error,
        detail: "Missing final delivery: Codex exited without a final agent message".to_string(),
        metadata: Some(serde_json::json!({
            "delivery_guard": "missing_final_delivery",
            "last_work_kind": last_work_kind,
            "last_message_chars": last_message_chars,
            "exit_code": exit_code,
        })),
    });
    TaskStatus::Failed
}

pub(crate) struct StreamLineContext<'a> {
    pub agent: &'a dyn Agent,
    pub task_id: &'a TaskId,
    pub store: &'a Arc<Store>,
    pub workgroup_id: Option<&'a str>,
    pub synthetic_tracker: &'a mut SyntheticMilestoneTracker,
}

pub(crate) struct EventDetail {
    pub detail: String,
    pub kind: EventKind,
}

pub(crate) fn handle_streaming_line_with_session(
    ctx: StreamLineContext<'_>,
    info: &mut CompletionInfo,
    event_count: &mut u32,
    line: &str,
    session_saved: &mut bool,
) -> Result<Option<EventDetail>> {
    let StreamLineContext {
        agent,
        task_id,
        store,
        workgroup_id,
        synthetic_tracker,
    } = ctx;

    // PTY-attached agents (e.g. droid >=0.159) glue OSC/CSI escapes onto
    // their stream-json lines; strip them before any parser sees the line.
    let cleaned = crate::watcher::strip_terminal_escapes(line);
    let line = cleaned.as_ref();

    if let Some(finding) = extract_finding_detail(line)
        && let Some(group_id) = workgroup_id
    {
        let _ = store.insert_finding(
            group_id,
            &finding,
            Some(task_id.as_str()),
            None,
            None,
            None,
            None,
            None,
            None,
        );
        append_to_broadcast(group_id, task_id.as_str(), &finding);
    }

    if let Some(event) = parse_milestone_event(task_id, line) {
        synthetic_tracker.observe(&event);
        store.insert_event(&event)?;
        *event_count += 1;
        return Ok(Some(EventDetail::from_event(&event)));
    }

    if let Some(event) = agent.parse_event(task_id, line) {
        apply_completion_event(info, &event);
        synthetic_tracker.observe(&event);
        save_session_id(store, task_id, &event, session_saved)?;
        // The raw line, never `event.detail`. An adapter's detail is a rendering
        // aid composed: cursor's is `completed: grep <the pattern the model
        // chose>`, and on 2026-08-07 one of those — an audit's own `grep
        // "you're out of usage|…"` — became a hold on cursor that no clock would
        // release. `event_kind` is used only as a cheap prefilter over lines the
        // adapter already parsed; it can narrow what is looked at and never
        // admit anything, because `quota_channel` decides admission.
        if event.event_kind == EventKind::Error
            && let Some(message) = rate_limit::refusal_on_channel(
                line,
                agent.kind(),
                crate::quota_channel::Channel::CliStream,
            )
        {
            mark_stream_rate_limit(store, task_id, agent, line, &message);
        }
        store.insert_event(&event)?;
        *event_count += 1;
        if let Some(event) = synthetic_tracker.synthetic_event(task_id, &event) {
            store.insert_event(&event)?;
            *event_count += 1;
        }
        return Ok(Some(EventDetail::from_event(&event)));
    }

    Ok(None)
}

fn mark_stream_rate_limit(
    store: &Arc<Store>,
    task_id: &TaskId,
    agent: &dyn Agent,
    line: &str,
    message: &str,
) {
    let model = store
        .get_task(task_id.as_str())
        .ok()
        .flatten()
        .and_then(|task| task.requested_model);
    if let Ok(value) = serde_json::from_str::<serde_json::Value>(line) {
        rate_limit::mark_rate_limited_for_model_value(
            &agent.kind(),
            agent.rate_limit_name(),
            model.as_deref(),
            &value,
            message,
        );
    } else {
        rate_limit::mark_rate_limited_for_model(
            &agent.kind(),
            agent.rate_limit_name(),
            model.as_deref(),
            message,
        );
    }
}

impl EventDetail {
    fn from_event(event: &crate::types::TaskEvent) -> Self {
        Self {
            detail: event.detail.clone(),
            kind: event.event_kind,
        }
    }
}

fn save_session_id(
    store: &Arc<Store>,
    task_id: &TaskId,
    event: &crate::types::TaskEvent,
    session_saved: &mut bool,
) -> Result<()> {
    if *session_saved {
        return Ok(());
    }
    let Some(metadata) = &event.metadata else {
        return Ok(());
    };
    let Some(session_id) = metadata.get("agent_session_id").and_then(|s| s.as_str()) else {
        return Ok(());
    };
    store.update_agent_session_id(task_id.as_str(), session_id)?;
    *session_saved = true;
    Ok(())
}