harness-codex 0.1.1

// SPDX-License-Identifier: AGPL-3.0-or-later

//! Bounded, content-free S08 owned-child OpenAI subscription probe.
//!
//! This is deliberately a thin caller of the reviewed app-server JSONL
//! protocol. It never accepts prompt content, credentials, or provider URLs.
//! Both the standalone CLI and Desktop seams create fresh owned children from
//! their separately pinned app-server executables.

use crate::{
    AppServerLineTransport, AppServerProcessError, CodexAppServerClient, ReasoningEffort,
    TerminalStatus,
};
use serde::Serialize;
use sha2::{Digest as _, Sha256};

pub const S08_OWNED_CHILD_TURN_SCHEMA: &str = "steadymux.s08-owned-child-turn.v2";
pub const S08_OPENAI_MODEL: &str = "gpt-5.6-terra";
const S08_PROBE_TEXT: &str = "SteadyMux S08 authorized owned-child probe.";

#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
pub struct S08OwnedChildTurnReceipt {
    pub schema: &'static str,
    pub status: &'static str,
    pub correlation: String,
    /// Terminal token counters only. They are bound into `receipt_digest` with
    /// the exact owned-child thread and turn identity; no content is retained.
    pub input_tokens: u64,
    pub output_tokens: u64,
    pub total_tokens: u64,
    pub receipt_digest: String,
}

#[derive(Serialize)]
struct ReceiptDigestInput<'a> {
    schema: &'static str,
    correlation: &'a str,
    thread_id: &'a str,
    turn_id: &'a str,
    model: &'a str,
    effort: ReasoningEffort,
    input_tokens: u64,
    output_tokens: u64,
    total_tokens: u64,
}

/// Runs exactly one fixed, tool-free OpenAI subscription turn through an
/// injected app-server transport. The receipt is content-free and its
/// correlation is the exact client message ID accepted by the protocol.
pub fn run_s08_owned_child_turn<T: AppServerLineTransport>(
    client: &mut CodexAppServerClient<T>,
    correlation: &str,
) -> Result<S08OwnedChildTurnReceipt, AppServerProcessError> {
    if !valid_correlation(correlation) {
        return Err(AppServerProcessError::InvalidConfiguration);
    }
    client.initialize()?;
    let discovery = client.discover()?;
    if discovery.rate_limits.reached_type.is_some()
        || !discovery.account.requires_openai_auth
        || !discovery
            .model_pages
            .iter()
            .flat_map(|page| &page.models)
            .any(|model| {
                model.id == S08_OPENAI_MODEL
                    && model.supported_efforts.contains(&ReasoningEffort::Medium)
            })
    {
        return Err(AppServerProcessError::UnsupportedExecutableVersion);
    }
    let turn = client.start_turn(
        correlation,
        S08_PROBE_TEXT,
        S08_OPENAI_MODEL,
        ReasoningEffort::Medium,
    )?;
    let run = client.wait_for_terminal(&turn)?;
    let shutdown = client.shutdown()?;
    if !shutdown.success
        || run.completion.status != TerminalStatus::Completed
        || run.completion.selected_model != S08_OPENAI_MODEL
        || run.completion.selected_effort != ReasoningEffort::Medium
        || run.completion.output.is_none()
    {
        return Err(AppServerProcessError::UnexpectedTerminalStatus);
    }
    let (Some(input_tokens), Some(output_tokens)) =
        (run.completion.input_tokens, run.completion.output_tokens)
    else {
        return Err(AppServerProcessError::UnexpectedTerminalStatus);
    };
    let total_tokens = input_tokens
        .checked_add(output_tokens)
        .ok_or(AppServerProcessError::UnexpectedTerminalStatus)?;
    let digest_input = ReceiptDigestInput {
        schema: S08_OWNED_CHILD_TURN_SCHEMA,
        correlation,
        thread_id: &turn.thread_id,
        turn_id: &turn.turn_id,
        model: S08_OPENAI_MODEL,
        effort: ReasoningEffort::Medium,
        input_tokens,
        output_tokens,
        total_tokens,
    };
    let bytes =
        serde_json::to_vec(&digest_input).map_err(|_| AppServerProcessError::TransportFailed)?;
    Ok(S08OwnedChildTurnReceipt {
        schema: S08_OWNED_CHILD_TURN_SCHEMA,
        status: "completed",
        correlation: correlation.to_owned(),
        input_tokens,
        output_tokens,
        total_tokens,
        receipt_digest: hex_sha256(&bytes),
    })
}

fn valid_correlation(value: &str) -> bool {
    value.len() <= 128
        && (value.starts_with("s08-openai-") || value.starts_with("s08-desktop-"))
        && value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
}

fn hex_sha256(bytes: &[u8]) -> String {
    let digest = Sha256::digest(bytes);
    let mut output = String::with_capacity(64);
    for byte in digest {
        use std::fmt::Write as _;
        let _ = write!(&mut output, "{byte:02x}");
    }
    output
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{AppServerLineTransport, ShutdownReceipt};
    use serde_json::{Value, json};
    use std::collections::VecDeque;
    use std::time::Duration;

    struct TranscriptTransport {
        inbound: VecDeque<Vec<u8>>,
        outbound: Vec<Value>,
        stopped: bool,
        rate_limit_reached: bool,
        token_usage: (u64, u64),
    }

    impl TranscriptTransport {
        fn new() -> Self {
            Self {
                inbound: VecDeque::new(),
                outbound: Vec::new(),
                stopped: false,
                rate_limit_reached: false,
                token_usage: (2, 1),
            }
        }

        fn rate_limited() -> Self {
            Self {
                rate_limit_reached: true,
                ..Self::new()
            }
        }

        fn overflowing_usage() -> Self {
            Self {
                token_usage: (u64::MAX, 1),
                ..Self::new()
            }
        }

        fn with_usage(input_tokens: u64, output_tokens: u64) -> Self {
            Self {
                token_usage: (input_tokens, output_tokens),
                ..Self::new()
            }
        }

        fn enqueue(&mut self, value: Value) {
            self.inbound
                .push_back(serde_json::to_vec(&value).expect("fixture JSON"));
        }
    }

    impl AppServerLineTransport for TranscriptTransport {
        fn send_line(&mut self, line: &str) -> Result<(), AppServerProcessError> {
            let request: Value =
                serde_json::from_str(line).map_err(|_| AppServerProcessError::TransportFailed)?;
            let method = request["method"]
                .as_str()
                .ok_or(AppServerProcessError::TransportFailed)?;
            self.outbound.push(request.clone());
            let id = request.get("id").cloned();
            match method {
                "initialize" => self.enqueue(json!({"jsonrpc":"2.0","id":id,"result":{"userAgent":format!("steadymux/{}", crate::CURRENT_STANDALONE_CLI_VERSION)}})),
                "initialized" => {}
                "account/read" => self.enqueue(json!({"jsonrpc":"2.0","id":id,"result":{"account":{"type":"chatgpt","planType":"pro"},"requiresOpenaiAuth":true}})),
                "model/list" => self.enqueue(json!({"jsonrpc":"2.0","id":id,"result":{"data":[{"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]}],"nextCursor":null}})),
                "account/rateLimits/read" => {
                    let reached = if self.rate_limit_reached {
                        Value::String("primary".to_owned())
                    } else {
                        Value::Null
                    };
                    self.enqueue(json!({"jsonrpc":"2.0","id":id,"result":{"rateLimits":{"primary":null,"secondary":null,"rateLimitReachedType":reached}}}));
                }
                "thread/start" => self.enqueue(json!({"jsonrpc":"2.0","id":id,"result":{"thread":{"id":"thr-s08"}}})),
                "turn/start" => {
                    let (input_tokens, output_tokens) = self.token_usage;
                    self.enqueue(json!({"jsonrpc":"2.0","id":id,"result":{"turn":{"id":"turn-s08","status":"inProgress","items":[],"error":null}}}));
                    self.enqueue(json!({"jsonrpc":"2.0","method":"turn/started","params":{"threadId":"thr-s08","turn":{"id":"turn-s08","status":"inProgress","items":[],"error":null}}}));
                    self.enqueue(json!({"jsonrpc":"2.0","method":"item/started","params":{"threadId":"thr-s08","turnId":"turn-s08","item":{"type":"agentMessage","id":"item-s08","text":""}}}));
                    self.enqueue(json!({"jsonrpc":"2.0","method":"item/completed","params":{"threadId":"thr-s08","turnId":"turn-s08","item":{"type":"agentMessage","id":"item-s08","text":"fixed output"}}}));
                    let total_tokens = input_tokens.saturating_add(output_tokens);
                    self.enqueue(json!({"jsonrpc":"2.0","method":"thread/tokenUsage/updated","params":{"threadId":"thr-s08","turnId":"turn-s08","tokenUsage":{"last":{"cachedInputTokens":0,"inputTokens":input_tokens,"outputTokens":output_tokens,"reasoningOutputTokens":0,"totalTokens":total_tokens},"total":{"cachedInputTokens":0,"inputTokens":input_tokens,"outputTokens":output_tokens,"reasoningOutputTokens":0,"totalTokens":total_tokens},"modelContextWindow":128000}}}));
                    self.enqueue(json!({"jsonrpc":"2.0","method":"turn/completed","params":{"threadId":"thr-s08","turn":{"id":"turn-s08","status":"completed","items":[],"error":null}}}));
                }
                _ => return Err(AppServerProcessError::TransportFailed),
            }
            Ok(())
        }

        fn recv_line(&mut self, _timeout: Duration) -> Result<Vec<u8>, AppServerProcessError> {
            self.inbound
                .pop_front()
                .ok_or(AppServerProcessError::UnexpectedEof)
        }

        fn shutdown(&mut self) -> Result<ShutdownReceipt, AppServerProcessError> {
            self.stopped = true;
            Ok(ShutdownReceipt {
                forced: false,
                success: true,
            })
        }
    }

    #[test]
    fn transcript_turn_keeps_actual_correlation_and_emits_bound_usage_receipt() {
        let transport = TranscriptTransport::new();
        let mut client =
            CodexAppServerClient::new(transport, Duration::from_secs(1)).expect("bounded client");
        client.require_reviewed_cli_version();
        let receipt =
            run_s08_owned_child_turn(&mut client, "s08-desktop-01").expect("transcript turn");
        assert_eq!(receipt.correlation, "s08-desktop-01");
        assert_eq!(receipt.status, "completed");
        assert_eq!(receipt.input_tokens, 2);
        assert_eq!(receipt.output_tokens, 1);
        assert_eq!(receipt.total_tokens, 3);
        assert_eq!(receipt.receipt_digest.len(), 64);
        assert!(!format!("{receipt:?}").contains("fixed output"));
        let transport = client.into_transport();
        assert!(transport.stopped);
        let turn = transport
            .outbound
            .iter()
            .find(|line| line["method"] == "turn/start")
            .expect("turn start");
        assert_eq!(turn["params"]["input"][0]["text"], S08_PROBE_TEXT);
    }

    #[test]
    fn reached_rate_limit_refuses_before_thread_or_turn_start() {
        let transport = TranscriptTransport::rate_limited();
        let mut client =
            CodexAppServerClient::new(transport, Duration::from_secs(1)).expect("bounded client");
        client.require_reviewed_cli_version();
        assert_eq!(
            run_s08_owned_child_turn(&mut client, "s08-openai-rate-limit"),
            Err(AppServerProcessError::UnsupportedExecutableVersion)
        );
        let transport = client.into_transport();
        assert!(
            transport
                .outbound
                .iter()
                .all(|line| line["method"] != "thread/start" && line["method"] != "turn/start")
        );
    }

    #[test]
    fn terminal_usage_overflow_refuses_without_emitting_a_receipt() {
        let transport = TranscriptTransport::overflowing_usage();
        let mut client =
            CodexAppServerClient::new(transport, Duration::from_secs(1)).expect("bounded client");
        client.require_reviewed_cli_version();
        assert!(run_s08_owned_child_turn(&mut client, "s08-openai-overflow").is_err());
    }

    #[test]
    fn receipt_digest_commits_terminal_usage_for_the_exact_correlation() {
        let mut first = CodexAppServerClient::new(
            TranscriptTransport::with_usage(2, 1),
            Duration::from_secs(1),
        )
        .expect("bounded client");
        first.require_reviewed_cli_version();
        let first = run_s08_owned_child_turn(&mut first, "s08-openai-usage").expect("receipt");

        let mut second = CodexAppServerClient::new(
            TranscriptTransport::with_usage(2, 2),
            Duration::from_secs(1),
        )
        .expect("bounded client");
        second.require_reviewed_cli_version();
        let second = run_s08_owned_child_turn(&mut second, "s08-openai-usage").expect("receipt");

        assert_ne!(first.receipt_digest, second.receipt_digest);
        assert_eq!(first.total_tokens, 3);
        assert_eq!(second.total_tokens, 4);
    }
}