if runtime.last_usage.reported() {

        println!("Usage: {}", runtime.last_usage.line());

    }

    if let Err(error) = revoked {

        eprintln!("oa: the thread was not revoked: {error}");

    // The revocation reply carries the grant's own spend, and dropping it is

    // how the CLI came to print a client-side count nothing could check.

    match revoked {

        Ok(spent) => {

            if let Some(line) = runtime.spend_line(spent) {

                println!("{line}");

            }

        }

        Err(error) => eprintln!("oa: the thread was not revoked: {error}"),

    }

    for failure in &runtime.record_failures {

        eprintln!("oa: {failure}");

    }

    if let Some(path) = coder.export.as_deref() {

    let id = task.id;

    let sink = events.clone();

    let turn = runtime.execute_turn(&task.prompt, move |chunk| {

        let _ = sink.send(ChildEvent::Output {

    let outcome = {

        let turn = runtime.execute_turn(&task.prompt, move |chunk| {

            let _ = sink.send(ChildEvent::Output {

                id,

                text: chunk.to_string(),

            });

        });

        tokio::select! {

            biased;

            _ = cancel.changed() => Err("stopped before finishing".to_string()),

            answered = turn => answered.map_err(|error| error.to_string()),

        }

    };

    // A cancelled child drops its turn mid-flight, so the turn's own failure

    // path never runs and the transcript would simply stop. Say what happened

    // before the thread is revoked, so a stopped child is not left looking

    // like one that finished quietly.

    if let Err(why) = &outcome {

        runtime.note_interruption(why).await;

    }

    // Awaited, on every path. A child used to leave its thread to the `Drop`

    // impl, which spawns a best-effort DELETE the process may never poll — and

    // a thread left open holds its grant's remaining budget. The failure is

    // reported to the parent's event stream rather than to a screen this child

    // does not own.

    if let Err(error) = runtime.close().await {

        let _ = events.send(ChildEvent::Activity {

            id,

            text: chunk.to_string(),

            text: format!("the thread was not revoked: {error}"),

        });

    });

    }

    for failure in &runtime.record_failures {

        let _ = events.send(ChildEvent::Activity {

            id,

            text: failure.clone(),

        });

    }

    tokio::select! {

        biased;

        _ = cancel.changed() => Err(ChildFailure {

            why: "stopped before finishing".to_string(),

            pid: None,

        }),

        answered = turn => match answered {

            Ok(text) => Ok(ChildAnswer { text, pid: None }),

            Err(error) => Err(ChildFailure { why: error.to_string(), pid: None }),

        },

    match outcome {

        Ok(text) => Ok(ChildAnswer { text, pid: None }),

        Err(why) => Err(ChildFailure { why, pid: None }),

    }

}

/// How often the streaming bullet flips.

const PULSE: Duration = Duration::from_millis(400);

/// How long the exit waits for the session to revoke its thread.

///

/// Long enough for a `DELETE` on a working connection, short enough that a

/// reader who has quit does not sit looking at a finished screen.

const THREAD_REVOCATION_GRACE: Duration = Duration::from_secs(15);

/// The diff inspector's state.

struct DiffView {

    files: Vec<FileDiff>,

    }

}

/// Revoke a session's thread and say what it cost, and what went unrecorded.

///

/// The lines land on stdout rather than in a screen because both callers reach

/// here after their screen is gone. Silence when the session held no thread:

/// the local lane has nothing to revoke and nothing was billed.

pub async fn close_and_report(session: &mut CoderRuntimeSession) {

    match session.close().await {

        Ok(spent) => {

            if let Some(line) = session.spend_line(spent) {

                println!("{line}");

            }

        }

        Err(error) => eprintln!("oa: the thread was not revoked: {error}"),

    }

    for failure in &session.record_failures {

        eprintln!("oa: {failure}");

    }

}

/// Own the session and do what the app asks for.

///

/// Returns once the app has dropped its control channel, having revoked the

/// session's thread on the way out. The caller awaits that rather than

/// aborting the task: an aborted actor drops its session, and the `Drop` impl

/// only spawns a revocation that the exiting process may never poll.

pub async fn runtime_actor(

    mut session: CoderRuntimeSession,

    mut control: UnboundedReceiver<Control>,

                    Err(error) => TurnEvent::Failed(error.to_string()),

                };

                if events.send(event).is_err() {

                    return;

                    break;

                }

            }

            Control::Diff(arguments) => {

                    Err(why) => TurnEvent::Notice(why),

                };

                if events.send(event).is_err() {

                    return;

                    break;

                }

            }

            Control::ForeignResume(selection) => {

                        control: session,

                    });

                    if opened.is_err() {

                        return;

                        break;

                    }

                    // Forwarded from a task of its own so this actor stays

                    // able to answer while the program runs.

            },

        }

    }

    // The app has gone. Revoke the thread here, awaited, rather than leaving

    // it to `Drop` in an aborted task.

    close_and_report(&mut session).await;

}

/// Collect the diff `/diff` asked for.

        .await

    };

    runtime.abort();

    // `run_loop` has taken the control sender with it, so the actor's receiver

    // is closed and the actor is on its way out with a revocation to make.

    // Awaited rather than aborted: aborting drops the session inside a dead

    // task, where the `Drop` impl can only spawn a DELETE this process may

    // exit before polling. Bounded, because a turn still streaming would

    // otherwise hold the exit for as long as the model wants — and in that

    // case `Drop`'s best effort is what is left, which is what it is for.

    if tokio::time::timeout(THREAD_REVOCATION_GRACE, runtime)

        .await

        .is_err()

    {

        eprintln!(

            "oa: the session was still working after {}s, so its thread was left to the \

             best-effort revocation.",

            THREAD_REVOCATION_GRACE.as_secs()

        );

    }

    result?;

    if let Some(path) = args.export {

    // streamed, which is how the offline paths still say something.

    let streamed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));

    let saw = std::sync::Arc::clone(&streamed);

    let answer = session

    let answered = session

        .execute_turn(&prompt, move |chunk| {

            use std::io::Write;

            saw.store(true, std::sync::atomic::Ordering::Relaxed);

            let _ = std::io::stdout().flush();

        })

        .await

        .map_err(|e| e.to_string())?;

        .map_err(|e| e.to_string());

    // Awaited, and awaited whether the turn worked or not. This path used to

    // return on the failure and leave the revocation to `Drop`, which spawns a

    // best-effort DELETE that may never be polled before the process exits —

    // and a thread left open holds its grant's remaining budget.

    close_and_report(&mut session).await;

    let answer = answered?;

    if !streamed.load(std::sync::atomic::Ordering::Relaxed) {

        print!("{answer}");

    }

//!

//! ## What a Rust-opened thread replays to

//!

//! [`crate::runtime`] records `thread.opened` and no turn events, so a thread

//! this CLI opened replays to an empty conversation while one `openagents

//! coder` opened replays to its transcript. That is the record, not a parse

//! failure, and the caller says how many messages came back rather than

//! implying a transcript that is not there.

//! Its own conversation. [`crate::runtime`] records `turn.user`, `tool.ran`

//! and `turn.assistant` as a turn runs, which is the vocabulary below, so a

//! thread this CLI opened resumes the same way one `openagents coder` opened

//! does. It did not always: the runtime recorded nothing at all, every thread

//! held one `thread.opened` and no more, and `--resume` on one replayed zero

//! messages. A thread from before that fix still replays to an empty

//! conversation — that is the record, not a parse failure, and the caller says

//! how many messages came back rather than implying a transcript that is not

//! there.

use crate::runtime::ChatMessage;

use serde::Deserialize;

//! request with the sentence `Completed autonomous reasoning turn (offline

//! fallback).` and exit 0. Neither comes back.

//!

//! ## A turn writes itself down before the thread is revoked

//!

//! The thread lane records what it did to `POST /api/v1/threads/{id}/events`

//! as it happens — `turn.user`, `turn.reasoning`, `tool.ran`, `turn.assistant`

//! — which is the vocabulary [`crate::resume`] replays and the one

//! `docs/2026-08-24-coder-account-integration-audit.md` fixes. It used to

//! record nothing at all, so every session reached `DELETE` with a transcript

//! holding one `thread.opened` and nothing else, and the account's history

//! read as a wall of cancellations over runs that had answered correctly.

//!

//! A turn that did not answer records [`ThreadRecord::failed`] instead, and

//! never an answer: the point of writing the turn down is that the record

//! matches what happened, which it does not if a refused proxy, a broken

//! stream, an interrupted session, or an exhausted step budget lands in the

//! transcript looking like an answer.

//!

//! Recording is best effort and does not fail a turn that worked: a transcript

//! that could not be written is kept in [`CoderRuntimeSession::record_failures`]

//! and reported, rather than swallowed or allowed to throw away an answer.

//!

//! ## Model ids are the server's, not this file's

//!

//! The deployment publishes its catalog at `GET /api/v1/models` and refuses

/// A backstop against a model that loops, not a budget.

const MAX_TOOL_STEPS: usize = 30;

/// How many transcript events one append may carry.

///

/// The server's own cap (`OpenAgents.Threads.maximum_event_batch/0`). A longer

/// list is split into several appends rather than refused as one.

const MAX_EVENT_BATCH: usize = 100;

/// The tier names a reader may type, and the catalog id each one opens on.

///

/// A tier is the unit `--lane` deals in: `flash` is the fast lane whatever

    pub model: String,

}

/// One event on a thread's transcript, in the vocabulary `--resume` replays.

///

/// The words are the ones the server's own readers know: `turn.user`,

/// `turn.reasoning`, `tool.ran` and `turn.assistant` are what

/// [`crate::resume::replay_wire`] rebuilds a conversation from and what the

/// WEKA export cuts model calls at. `turn.failed` is deliberately outside that

/// set — a turn that did not answer has no answer to replay, and a reader that

/// does not know the word skips it rather than feeding a failure back to a

/// model as though the model had said it.

#[derive(Debug, Clone, PartialEq, Serialize)]

pub struct ThreadRecord {

    pub event_type: String,

    pub payload: serde_json::Value,

}

impl ThreadRecord {

    fn new(event_type: &str, payload: serde_json::Value) -> Self {

        Self {

            event_type: event_type.to_string(),

            payload,

        }

    }

    /// What the reader asked.

    pub fn user(text: &str) -> Self {

        Self::new("turn.user", serde_json::json!({ "text": text }))

    }

    /// What the model thought before it answered, recorded whole.

    pub fn reasoning(text: &str) -> Self {

        Self::new("turn.reasoning", serde_json::json!({ "text": text }))

    }

    /// One call with its arguments and its result: one fact, one event.

    ///

    /// `arguments` is the raw JSON string the wire carried, not a re-encoding

    /// of it, because that is what goes back to a model on replay.

    pub fn tool_ran(call_id: &str, tool: &str, arguments: &str, output: &str) -> Self {

        Self::new(

            "tool.ran",

            serde_json::json!({

                "call_id": call_id,

                "tool": tool,

                "arguments": arguments,

                "output": output,

            }),

        )

    }

    /// The answer, with what the turn spent reaching it.

    pub fn assistant(text: &str, usage: TurnUsage, calls: usize) -> Self {

        Self::new(

            "turn.assistant",

            serde_json::json!({

                "text": text,

                "usage": usage,

                "calls": calls,

            }),

        )

    }

    /// A turn that produced no answer, and the sentence saying why.

    ///

    /// This is the half of the record that keeps it honest. A session that

    /// failed must not read afterwards as one that succeeded, so every exit

    /// from a turn that is not the model's own answer writes one of these and

    /// no `turn.assistant`.

    pub fn failed(why: &str, usage: TurnUsage, calls: usize) -> Self {

        Self::new(

            "turn.failed",

            serde_json::json!({

                "error": why,

                "usage": usage,

                "calls": calls,

            }),

        )

    }

}

#[derive(Debug, Clone, Serialize, Deserialize)]

pub struct ChatMessage {

    pub role: String,

    pub last_model: Option<String>,

    /// What the last turn spent, summed over its steps.

    pub last_usage: TurnUsage,

    /// How many tools the last turn ran, counted as they were recorded.

    pub last_calls: usize,

    /// What every turn this session ran has spent, summed.

    ///

    /// The figure to hold a server-reported grant spend against: `last_usage`

    /// is one turn, and a session that ran four of them and compared the

    /// fourth against the grant's total would report a divergence on every

    /// multi-turn session and call it a mismatch.

    pub session_usage: TurnUsage,

    /// Transcript appends this session could not make, in the order they failed.

    ///

    /// Recording is best effort — a thread the server would not take an event

    /// for is not a reason to throw away an answer the reader is waiting on —

    /// but a silent best effort is how the record came to disagree with the

    /// session in the first place. The failure is kept here and reported.

    pub record_failures: Vec<String>,

    /// The reasoning the last turn emitted, if the model emits any.

    ///

    /// Kept off the content callback deliberately: `delta.reasoning` and

            last_grant: None,

            last_model: None,

            last_usage: TurnUsage::default(),

            last_calls: 0,

            session_usage: TurnUsage::default(),

            record_failures: Vec::new(),

            last_reasoning: String::new(),

            reasoning: None,

            repository: None,

        }))

    }

    /// What the server billed this session against what this process counted.

    ///

    /// `close` hands back the grant's own `spent`, and the caller used to drop

    /// it, so the CLI printed its client-side accumulation and nothing could

    /// ever notice the two disagreeing. This is the line that notices: the

    /// server's figure, and a second sentence naming the gap when there is one.

    ///

    /// `None` when the server reported no spend — the local lane, or a session

    /// that never opened a thread — because there is nothing to reconcile.

    pub fn spend_line(&self, reported: Option<TurnUsage>) -> Option<String> {

        let reported = reported?;

        let counted = self.session_usage.total_tokens;

        let billed = reported.total_tokens;

        let line = format!("Billed by the server: {billed} tokens");

        if billed == counted {

            return Some(line);

        }

        Some(format!(

            "{line} — this session counted {counted}, a difference of {}. \

             The server's figure is the one the account is charged against.",

            billed.abs_diff(counted)

        ))

    }

    // ───────────────────────────────────────────────────────── the transcript

    /// The thread this session holds, while it holds one.

    pub fn thread(&self) -> Option<&str> {

        self.thread_id.as_deref()

    }

    /// Append events to this session's thread transcript.

    ///

    /// `POST /api/v1/threads/{id}/events`, batched: one round trip lands the

    /// whole list in order or none of it, capped at the server's own batch

    /// maximum so a long step is split rather than refused whole. A session

    /// holding no thread — the local lane, or one already revoked — writes

    /// nothing and answers `Ok(false)`, because nowhere to write is not a

    /// failure to write.

    pub async fn record(&self, events: &[ThreadRecord]) -> Result<bool, Failure> {

        let Some(thread_id) = &self.thread_id else {

            return Ok(false);

        };

        if events.is_empty() {

            return Ok(false);

        }

        let url = format!("{}/threads/{thread_id}/events", self.api_base);

        for batch in events.chunks(MAX_EVENT_BATCH) {

            let mut request = self

                .http

                .post(&url)

                .timeout(Duration::from_secs(30))

                .json(&serde_json::json!({ "events": batch }));

            if let Some(token) = &self.user_token {

                request = request.bearer_auth(token);

            }

            let resp = request.send().await.map_err(|error| -> Failure {

                format!("{url} could not be reached: {error}").into()

            })?;

            if !resp.status().is_success() {

                let status = resp.status();

                let body = resp.text().await.unwrap_or_default();

                return Err(format!(

                    "{url} refused the transcript append: {status} {}",

                    snippet(&body)

                )

                .into());

            }

        }

        Ok(true)

    }

    /// Record, keeping a refusal rather than failing the turn over it.

    pub async fn note(&mut self, events: Vec<ThreadRecord>) {

        if let Err(error) = self.record(&events).await {

            let kinds = events

                .iter()

                .map(|event| event.event_type.as_str())

                .collect::<Vec<_>>()

                .join(", ");

            self.record_failures

                .push(format!("{kinds} were not recorded: {error}"));

        }

    }

    // ─────────────────────────────────────────────────────────── the turn

    pub async fn execute_turn<F>(

        });

        self.last_usage = TurnUsage::default();

        self.last_calls = 0;

        self.last_reasoning.clear();

        if self.lane.is_local() {

        let answered = if self.lane.is_local() {

            self.run_local_turn(&tool_defs, chunk_callback).await

        } else {

            self.run_thread_turn(&tool_defs, chunk_callback).await

        }

            self.run_thread_turn(prompt, &tool_defs, chunk_callback)

                .await

        };

        self.session_usage.add(self.last_usage);

        answered

    }

    async fn run_thread_turn<F>(

        &mut self,

        prompt: &str,

        tool_defs: &[ToolDefinition],

        mut chunk_callback: F,

    ) -> Result<String, Failure>

        };

        self.last_model = Some(grant.model.clone());

        // The transcript opens with what was asked, before a model is reached,

        // so a turn that dies mid-step still leaves the question behind.

        self.note(vec![ThreadRecord::user(prompt)]).await;

        let mut final_answer = String::new();

        // False means the step budget ran out with every step still calling

        // tools, which is not an answer and must not be returned as one.

                Ok(r) => {

                    let status = r.status();

                    let body = r.text().await.unwrap_or_default();

                    return Err(format!(

                    let why = format!(

                        "{} refused the turn: {status} {}",

                        grant.proxy_url,

                        snippet(&body)

                    )

                    .into());

                    );

                    return Err(self.record_failure(why).await);

                }

                Err(error) => {

                    return Err(format!("{} could not be reached: {error}", grant.proxy_url).into())

                    let why = format!("{} could not be reached: {error}", grant.proxy_url);

                    return Err(self.record_failure(why).await);

                }

            };

                let event = match event {

                    Ok(event) => event,

                    Err(error) => {

                        return Err(format!(

                        let why = format!(

                            "the reply from {} stopped mid-stream: {error}",

                            grant.proxy_url

                        )

                        .into())

                        );

                        // What streamed before the break is part of what

                        // happened and is kept, but the turn still records as

                        // failed: an incomplete answer is not an answer.

                        self.last_usage.add(step.usage);

                        self.last_reasoning.push_str(&step.reasoning);

                        return Err(self.record_failure(why).await);

                    }

                };

                if event.data == "[DONE]" {

            self.last_usage.add(step.usage);

            self.last_reasoning.push_str(&step.reasoning);

            // The working comes before whatever it led to, in the order it

            // happened. Recorded whole: it is the largest part of what a

            // session produces and a transcript without it is a summary.

            if !step.reasoning.trim().is_empty() {

                let thought = ThreadRecord::reasoning(&step.reasoning);

                self.note(vec![thought]).await;

            }

            if step.tool_calls.is_empty() {

                final_answer = step.content;

                // The answer joins the transcript. `run_tools` records an

                    tool_calls: None,

                    tool_call_id: None,

                });

                let said = ThreadRecord::assistant(&final_answer, self.last_usage, self.last_calls);

                self.note(vec![said]).await;

                answered = true;

                break;

            }

            self.run_tools(step).await;

            let ran = self.run_tools(step).await;

            self.last_calls += ran.len();

            self.note(ran).await;

        }

        if !answered {

            return Err(format!(

            let why = format!(

                "the turn used all {MAX_TOOL_STEPS} tool steps without producing an answer; \

                 nothing was returned rather than an empty answer that reads as success"

            )

            .into());

            );

            return Err(self.record_failure(why).await);

        }

        Ok(final_answer)

    }

    /// Write a turn's failure to the transcript, then hand back the failure.

    ///

    /// Every exit from a turn that is not the model's own answer goes through

    /// here, so the record cannot say a session succeeded where it did not.

    async fn record_failure(&mut self, why: String) -> Failure {

        let (usage, calls) = (self.last_usage, self.last_calls);

        self.note(vec![ThreadRecord::failed(&why, usage, calls)])

            .await;

        why.into()

    }

    /// Record that this session stopped before the turn in flight answered.

    ///

    /// A dropped turn future never reaches [`Self::record_failure`] — a

    /// cancelled child, a session quit mid-turn — so the caller that dropped

    /// it says so here. Without this an interruption leaves a transcript that

    /// simply stops, which a later reader has no way to tell from a turn that

    /// finished quietly.

    pub async fn note_interruption(&mut self, why: &str) {

        let (usage, calls) = (self.last_usage, self.last_calls);

        self.note(vec![ThreadRecord::failed(why, usage, calls)])

            .await;

    }

    /// Record the assistant's tool calls, run them, and put the results back.

    async fn run_tools(&mut self, step: StepAccumulator) {

    ///

    /// Hands back one `tool.ran` per call for the transcript: the call and its

    /// result are one fact, and the caller has both only once the tool has run.

    async fn run_tools(&mut self, step: StepAccumulator) -> Vec<ThreadRecord> {

        let recorded: Vec<serde_json::Value> = step

            .tool_calls

            .values()

            tool_call_id: None,

        });

        let mut ran = Vec::new();

        for (id, name, args_str) in step.tool_calls.into_values() {

            let arguments: serde_json::Value =

                serde_json::from_str(&args_str).unwrap_or(serde_json::json!({}));

                arguments,

            };

            let result = self.tools.execute_tool(&call).await;

            ran.push(ThreadRecord::tool_ran(

                &id,

                &name,

                &args_str,

                &result.output,

            ));

            self.messages.push(ChatMessage {

                role: "tool".to_string(),

                content: Some(result.output),

                tool_call_id: Some(id),

            });

        }

        ran

    }

    // ────────────────────────────────────────────────────── the local lane

                answered = true;

                break;

            }

            self.run_tools(step).await;

            // The local lane holds no thread of its own, so `note` writes

            // nothing; it is called anyway so a local session that resumed

            // somebody's thread records against it like any other.

            let ran = self.run_tools(step).await;

            self.note(ran).await;

        }

        if !answered {

    assert!(!ChildLane::known("gemni"));

    assert!(!ChildLane::known(""));

}

// ──────────────────────────────────────────────────── the child's own thread

//

// A delegated child opens a thread of its own and used to leave revoking it to

// the `Drop` impl, which spawns a best-effort `DELETE` onto whatever runtime is

// still up. The audit in #89 found two threads still open from long-dead

// sessions; a thread left open holds its grant's remaining budget (issue #107).

/// How long the stub holds a revocation open.

///

/// Long enough to tell an awaited `close()` from the `Drop` impl's spawned

/// best effort, which returns at once and leaves its request in flight.

const REVOCATION_HELD: Duration = Duration::from_millis(600);

/// A stand-in for the account API and the proxy, which records what it took.

struct ProxyStub {

    origin: String,

    requests: std::sync::Arc<std::sync::Mutex<Vec<String>>>,

}

impl ProxyStub {

    fn request_lines(&self) -> Vec<String> {

        self.requests

            .lock()

            .unwrap()

            .iter()

            .map(|request| request.lines().next().unwrap_or_default().to_string())

            .collect()

    }

}

/// Answers a thread open, transcript appends, a revocation, and one turn.

async fn proxy_stub() -> ProxyStub {

    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    let origin = format!("http://127.0.0.1:{port}");

    let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));

    let seen = std::sync::Arc::clone(&requests);

    let grant_url = format!("{origin}/proxy");

    tokio::spawn(async move {

        loop {

            let Ok((mut socket, _)) = listener.accept().await else {

                return;

            };

            let mut request = Vec::new();

            let mut buffer = [0u8; 4096];

            loop {

                let Ok(read) = socket.read(&mut buffer).await else {

                    break;

                };

                if read == 0 {

                    break;

                }

                request.extend_from_slice(&buffer[..read]);

                let text = String::from_utf8_lossy(&request);

                if let Some(end) = text.find("\r\n\r\n") {

                    let length = text

                        .lines()

                        .find_map(|line| {

                            line.strip_prefix("content-length: ")

                                .or_else(|| line.strip_prefix("Content-Length: "))

                        })

                        .and_then(|value| value.trim().parse::<usize>().ok())

                        .unwrap_or(0);

                    if request.len() >= end + 4 + length {

                        break;

                    }

                }

            }

            let request = String::from_utf8_lossy(&request).to_string();

            let line = request.lines().next().unwrap_or_default().to_string();

            seen.lock().unwrap().push(request);

            let (status, content_type, body) = if line.starts_with("POST")

                && line.contains("/events")

            {

                (

                    201,

                    "application/json",

                    r#"{"events":[{"id":1}]}"#.to_string(),

                )

            } else if line.starts_with("POST /api/v1/threads") {

                (

                    200,

                    "application/json",

                    format!(

                        r#"{{"thread":{{"id":"th_child"}},"grant":{{"token":"tok","url":"{grant_url}","model":"ox-alpha"}}}}"#

                    ),

                )

            } else if line.starts_with("DELETE /api/v1/threads/") {

                // Held open, so an awaited revocation is measurably slower

                // than a spawned one. See the test below.

                tokio::time::sleep(REVOCATION_HELD).await;

                (

                    200,

                    "application/json",

                    r#"{"grant":{"status":"revoked","spent":{"calls":1,"total_tokens":12}}}"#

                        .to_string(),

                )

            } else {

                let frame =

                    serde_json::json!({"choices":[{"delta":{"content":"the child answered"}}]});

                let stream = format!("data: {frame}\n\ndata: [DONE]\n\n");

                let head = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\nconnection: close\r\n\r\n";

                let _ = socket.write_all(head.as_bytes()).await;

                let _ = socket.write_all(stream.as_bytes()).await;

                let _ = socket.flush().await;

                continue;

            };

            let head = format!(

                "HTTP/1.1 {status} X\r\ncontent-type: {content_type}\r\ncontent-length: {}\r\nconnection: close\r\n\r\n",

                body.len()

            );

            let _ = socket.write_all(head.as_bytes()).await;

            let _ = socket.write_all(body.as_bytes()).await;

            let _ = socket.flush().await;

        }

    });

    ProxyStub { origin, requests }

}

/// A child on the proxy revokes its own thread, awaited, and writes its turn

/// down on the way.

///

/// A child used to leave the revocation to the `Drop` impl, which spawns a

/// `DELETE` onto whatever runtime is still up and may never be polled. Timed

/// rather than merely observed: the stub holds the revocation open, so a child

/// that finishes faster than that did not wait for it.

#[tokio::test]

async fn a_delegated_child_revokes_its_own_thread() {

    let _guard = exclusive();

    let stub = proxy_stub().await;

    std::env::set_var("OPENAGENTS_API_BASE", format!("{}/api/v1", stub.origin));

    let supervisor = DelegationSupervisor::new(1, "ox-alpha", Some("oat_test".to_string()))

        .with_isolation(Isolation::None)

        .in_directory(Some(std::env::temp_dir()));

    let (results, _events) = run(&supervisor, "say something", None).await;

    std::env::remove_var("OPENAGENTS_API_BASE");

    assert_eq!(results.len(), 1);

    assert!(

        results[0].success,

        "the child failed: {}",

        results[0].output

    );

    let lines = stub.request_lines();

    assert!(

        lines

            .iter()

            .any(|line| line.starts_with("DELETE /api/v1/threads/th_child")),

        "the child left its thread open: {lines:?}"

    );

    assert!(

        lines

            .iter()

            .any(|line| line.starts_with("POST") && line.contains("/events")),

        "the child recorded nothing of what it did: {lines:?}"

    );

    assert!(

        results[0].duration_ms >= REVOCATION_HELD.as_millis(),

        "the child finished in {}ms with the revocation held open for {}ms, so it was \

         spawned and abandoned rather than awaited",

        results[0].duration_ms,

        REVOCATION_HELD.as_millis()

    );

}

    Sse(Vec<String>, Option<(usize, Duration)>),

    /// Newline-delimited JSON, the shape Ollama streams.

    Ndjson(Vec<String>, Option<(usize, Duration)>),

    /// A body sent after a pause. What tells a call that was awaited from one

    /// that was spawned and hoped for: only the first waits.

    Delayed(Duration, u16, &'static str, String),

}

/// A server that records what it was asked and answers from a script.

                    return;

                };

                seen.lock().unwrap().push(request.clone());

                match handler(&request, &origin) {

                let reply = match handler(&request, &origin) {

                    Reply::Delayed(pause, status, content_type, body) => {

                        tokio::time::sleep(pause).await;

                        Reply::Body(status, content_type, body)

                    }

                    other => other,

                };

                match reply {

                    Reply::Delayed(..) => unreachable!("already unwrapped"),

                    Reply::Body(status, content_type, body) => {

                        let head = format!(

                            "HTTP/1.1 {status} X\r\ncontent-type: {content_type}\r\n\

    session.execute_turn("one", |_| {}).await.unwrap();

    session.execute_turn("two", |_| {}).await.unwrap();

    // The open, not the appends: `POST /threads/{id}/events` shares the prefix

    // and a turn now writes several of those.

    let opens = stub

        .request_lines()

        .iter()

        .filter(|line| line.starts_with("POST /api/v1/threads"))

        .filter(|line| line.starts_with("POST /api/v1/threads ") && !line.contains("/events"))

        .count();

    assert_eq!(opens, 1, "each turn opened its own thread");

}

        bodies[1]

    );

}

// ────────────────────────────────────────────────────────────── the record

//

// A session used to reach `DELETE` having recorded nothing at all, so the

// server's only terminal act was a cancellation and 31 of the 50 most recent

// threads on one account read `error_code: cancelled` over runs that had

// answered correctly (issue #106). These prove the turn writes itself down

// first, and — the half that matters more — that a turn which did *not* answer

// writes down that it did not.

/// The `event_type` of every transcript event the stub was posted, in order.

fn recorded(stub: &Stub) -> Vec<serde_json::Value> {

    stub.requests()

        .iter()

        .filter(|request| {

            request

                .lines()

                .next()

                .is_some_and(|line| line.starts_with("POST") && line.contains("/events"))

        })

        .filter_map(|request| {

            request

                .split_once("\r\n\r\n")

                .map(|(_, body)| body.to_string())

        })

        .filter_map(|body| serde_json::from_str::<serde_json::Value>(&body).ok())

        .flat_map(|body| {

            body.get("events")

                .and_then(|events| events.as_array())

                .cloned()

                .unwrap_or_default()

        })

        .collect()

}

fn kinds(events: &[serde_json::Value]) -> Vec<String> {

    events

        .iter()

        .map(|event| event["event_type"].as_str().unwrap_or("?").to_string())

        .collect()

}

/// The 201 the record route answers an append with.

fn appended() -> Reply {

    Reply::Body(

        201,

        "application/json",

        r#"{"events":[{"id":7}],"thread":{"id":"th_test","event_count":7}}"#.to_string(),

    )

}

fn revoked(total_tokens: u64) -> Reply {

    Reply::Body(

        200,

        "application/json",

        format!(

            r#"{{"grant":{{"status":"revoked","spent":{{"calls":1,"total_tokens":{total_tokens}}}}},

                "thread":{{"id":"th_test","status":"cancelled"}}}}"#

        ),

    )

}

/// A stub that answers a thread open, transcript appends, a revocation, and a

/// two-step turn: reasoning and a `shell` call, then the answer.

fn recording_stub() -> Stub {

    start(|request, origin| {

        let line = request.lines().next().unwrap_or_default().to_string();

        if line.starts_with("POST") && line.contains("/events") {

            return appended();

        }

        if line.starts_with("POST /api/v1/threads") {

            return Reply::Body(200, "application/json", grant_body(origin, "ox-alpha"));

        }

        if line.starts_with("DELETE /api/v1/threads/") {

            return revoked(116);

        }

        // The second call carries the tool result back, which is how a

        // stateless stub tells the steps of one turn apart.

        if request.contains(r#""role":"tool""#) {

            return Reply::Sse(

                vec![

                    frame(serde_json::json!({"choices":[{"delta":{"content":"It said hello."}}]})),

                    frame(serde_json::json!({

                        "choices": [],

                        "usage": {"prompt_tokens": 99, "completion_tokens": 17, "total_tokens": 116}

                    })),

                ],