use harness_codex::{
CodexAgentRuntime, CodexRole, CodexToolPolicy, OPENAI_ROLE_PROJECTIONS, ReasoningEffort,
ReplayDisposition, RuntimeError, RuntimeEvent, TerminalFailureClass, ToolCallStatus,
};
use serde_json::{Value, json};
fn value(line: &str) -> Value {
serde_json::from_str(line.trim_end()).expect("valid outbound JSONL")
}
fn record_turn_usage(
runtime: &mut CodexAgentRuntime,
thread_id: &str,
turn_id: &str,
input_tokens: u64,
output_tokens: u64,
) {
runtime
.ingest_line(
serde_json::to_string(&json!({
"method": "thread/tokenUsage/updated",
"params": {
"threadId": thread_id,
"turnId": turn_id,
"tokenUsage": {
"last": {
"cachedInputTokens": 1,
"inputTokens": input_tokens,
"outputTokens": output_tokens,
"reasoningOutputTokens": 1,
"totalTokens": input_tokens + output_tokens
},
"total": {
"cachedInputTokens": 1001,
"inputTokens": 1002,
"outputTokens": 1003,
"reasoningOutputTokens": 1004,
"totalTokens": 2005
},
"modelContextWindow": 128000
}
}
}))
.unwrap()
.as_bytes(),
)
.expect("one authoritative current-turn usage update");
}
fn prepare_runtime(mut runtime: CodexAgentRuntime) -> CodexAgentRuntime {
let initialize = value(&runtime.initialize_request().expect("initialize request"));
runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": initialize["id"],
"result": {
"userAgent": "steadymux/0.145.0 (fixture)",
"codexHome": "/redacted",
"platformFamily": "unix",
"platformOs": "macos"
}
}))
.unwrap()
.as_bytes(),
)
.expect("initialize response");
let initialized = value(&runtime.initialized_notification().expect("initialized ack"));
assert_eq!(initialized, json!({"jsonrpc":"2.0","method":"initialized"}));
runtime
}
fn ready_runtime() -> CodexAgentRuntime {
prepare_runtime(CodexAgentRuntime::new())
}
fn active_runtime(thread_id: &str, turn_id: &str) -> CodexAgentRuntime {
let mut runtime = ready_runtime();
let models = value(&runtime.model_list_request(None).unwrap());
runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","id":models["id"],"result":{"data":[{
"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"medium","description":"balanced"}]
}],"nextCursor":null}
})).unwrap().as_bytes()).unwrap();
let thread = value(
&runtime
.thread_start_request("gpt-5.6-terra", ReasoningEffort::Medium)
.unwrap(),
);
runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","id":thread["id"],"result":{"thread":{"id":thread_id}}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let turn = value(
&runtime
.turn_start_request(
thread_id,
"client-message",
"fixture prompt",
"gpt-5.6-terra",
ReasoningEffort::Medium,
)
.unwrap(),
);
runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","id":turn["id"],"result":{"turn":{"id":turn_id,"status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
runtime.ingest_line(serde_json::to_string(&json!({
"method":"turn/started","params":{"threadId":thread_id,"turn":{"id":turn_id,"status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
runtime
}
fn active_tool_runtime(thread_id: &str, turn_id: &str) -> CodexAgentRuntime {
let mut runtime = prepare_runtime(CodexAgentRuntime::with_tool_policy(
CodexToolPolicy::CanonicalMcpLifecycle,
));
let models = value(&runtime.model_list_request(None).unwrap());
runtime
.ingest_line(
serde_json::to_string(&json!({
"id":models["id"],"result":{"data":[{
"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]
}],"nextCursor":null}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let thread = value(
&runtime
.thread_start_request("gpt-5.6-terra", ReasoningEffort::Medium)
.unwrap(),
);
runtime
.ingest_line(
serde_json::to_string(&json!({
"id":thread["id"],"result":{"thread":{"id":thread_id}}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let turn = value(
&runtime
.turn_start_request(
thread_id,
"client-tool-message",
"fixture prompt",
"gpt-5.6-terra",
ReasoningEffort::Medium,
)
.unwrap(),
);
runtime.ingest_line(serde_json::to_string(&json!({
"id":turn["id"],"result":{"turn":{"id":turn_id,"status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
runtime.ingest_line(serde_json::to_string(&json!({
"method":"turn/started","params":{"threadId":thread_id,"turn":{"id":turn_id,"status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
runtime
}
#[test]
fn handshake_account_and_role_projection_are_exact_and_redacted() {
let mut runtime = ready_runtime();
let request = value(&runtime.account_read_request().expect("account/read"));
assert_eq!(request["method"], "account/read");
assert_eq!(request["params"], json!({"refreshToken": false}));
let canary = "private@example.test";
let event = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": request["id"],
"result": {
"account": {"type":"chatgpt", "email": canary, "planType":"pro"},
"requiresOpenaiAuth": true
}
}))
.unwrap()
.as_bytes(),
)
.expect("managed account response");
assert!(matches!(event, RuntimeEvent::ManagedAccount(_)));
assert!(!format!("{event:?}").contains(canary));
assert_eq!(OPENAI_ROLE_PROJECTIONS[0].role, CodexRole::Luna);
assert_eq!(
OPENAI_ROLE_PROJECTIONS[0].escalates_to,
Some(CodexRole::Terra)
);
assert_eq!(
OPENAI_ROLE_PROJECTIONS[1].escalates_to,
Some(CodexRole::Sol)
);
assert_eq!(OPENAI_ROLE_PROJECTIONS[2].escalates_to, None);
}
#[test]
fn model_pages_are_bounded_and_effort_is_catalog_validated() {
let mut runtime = ready_runtime();
let first = value(&runtime.model_list_request(None).expect("first page"));
assert_eq!(first["method"], "model/list");
assert_eq!(
first["params"],
json!({"cursor":null,"limit":50,"includeHidden":false})
);
let event = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0",
"id":first["id"],
"result":{
"data":[{
"id":"gpt-5.6-terra",
"displayName":"Terra",
"hidden":false,
"supportedReasoningEfforts":[
{"reasoningEffort":"low","description":"fast"},
{"reasoningEffort":"medium","description":"balanced"}
]
}],
"nextCursor":"page-2"
}
}))
.unwrap()
.as_bytes(),
)
.expect("model page");
assert!(matches!(event, RuntimeEvent::ModelPage(_)));
assert!(
runtime
.thread_start_request("gpt-5.6-terra", ReasoningEffort::Medium)
.is_ok()
);
assert_eq!(
runtime.thread_start_request("gpt-5.6-terra", ReasoningEffort::High),
Err(RuntimeError::UnsupportedEffort)
);
}
#[test]
fn duplicate_model_identity_is_rejected_across_catalog_pages() {
let mut runtime = ready_runtime();
let first = value(&runtime.model_list_request(None).expect("first page"));
runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0",
"id":first["id"],
"result":{
"data":[{
"id":"gpt-5.6-terra",
"displayName":"Terra",
"hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"medium"}]
}],
"nextCursor":"page-2"
}
}))
.unwrap()
.as_bytes(),
)
.expect("first model page");
let second = value(
&runtime
.model_list_request(Some("page-2"))
.expect("second page"),
);
let duplicate = runtime.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0",
"id":second["id"],
"result":{
"data":[{
"id":"gpt-5.6-terra",
"displayName":"Contradictory Terra",
"hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"high"}]
}],
"nextCursor":null
}
}))
.unwrap()
.as_bytes(),
);
assert_eq!(duplicate, Err(RuntimeError::InvalidInput));
assert_eq!(
runtime.thread_start_request("gpt-5.6-terra", ReasoningEffort::High),
Err(RuntimeError::UnsupportedEffort),
"a contradictory page must not overwrite the accepted model"
);
}
#[test]
fn thread_start_instructions_are_exact_bounded_and_content_free_on_error() {
let mut runtime = ready_runtime();
let models = value(&runtime.model_list_request(None).expect("model request"));
runtime
.ingest_line(
serde_json::to_string(&json!({
"id": models["id"],
"result": {
"data": [{
"id": "gpt-5.6-terra",
"displayName": "Terra",
"hidden": false,
"supportedReasoningEfforts": [{"reasoningEffort": "medium"}]
}],
"nextCursor": null
}
}))
.unwrap()
.as_bytes(),
)
.expect("model response");
let base = "private-base-instruction-canary\nsecond line";
let developer = "private-developer-instruction-canary\nsecond line";
let request = value(
&runtime
.thread_start_request_with_instructions(
"gpt-5.6-terra",
ReasoningEffort::Medium,
Some(base),
Some(developer),
)
.expect("bounded thread instructions"),
);
assert_eq!(request["method"], "thread/start");
assert_eq!(
request["params"],
json!({
"model": "gpt-5.6-terra",
"modelProvider": "openai",
"approvalPolicy": "never",
"sandbox": "read-only",
"ephemeral": true,
"baseInstructions": base,
"developerInstructions": developer
})
);
for forbidden in ["effort", "sandboxPolicy", "dynamicTools", "environments"] {
assert!(request["params"].get(forbidden).is_none());
}
let oversized = "x".repeat(64 * 1024 + 1);
assert_eq!(
runtime.thread_start_request_with_instructions(
"gpt-5.6-terra",
ReasoningEffort::Medium,
Some(&oversized),
None,
),
Err(RuntimeError::InvalidInput)
);
assert_eq!(
runtime.thread_start_request_with_instructions(
"gpt-5.6-terra",
ReasoningEffort::Medium,
Some("base\0instruction"),
None,
),
Err(RuntimeError::InvalidInput)
);
let secret = "sk-privateinstructioncanary123456";
let error = runtime
.thread_start_request_with_instructions(
"gpt-5.6-terra",
ReasoningEffort::Medium,
None,
Some(secret),
)
.expect_err("secret-shaped instruction must fail closed");
assert_eq!(error, RuntimeError::SensitiveInputRejected);
assert!(!format!("{error:?} {error} {runtime:?}").contains(secret));
}
#[test]
fn managed_rate_limits_are_requested_and_projected_without_reset_credit_details() {
let mut runtime = ready_runtime();
let request = value(&runtime.rate_limits_read_request().unwrap());
assert_eq!(request["method"], "account/rateLimits/read");
assert_eq!(request["params"], json!({}));
let event = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","id":request["id"],"result":{
"rateLimits":{
"primary":{"usedPercent":25,"windowDurationMins":15,"resetsAt":1730947200},
"secondary":null,"rateLimitReachedType":null
},
"rateLimitResetCredits":{"availableCount":2,"credits":[{"id":"sensitive-credit-detail"}]}
}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
assert!(matches!(event, RuntimeEvent::RateLimits(ref limits) if
limits.primary.as_ref().is_some_and(|window| window.used_percent == 25)
&& limits.secondary.is_none()));
assert!(!format!("{event:?}").contains("sensitive-credit-detail"));
}
#[test]
fn exact_turn_payload_correlates_interleaved_events_and_completed_is_authoritative() {
let mut runtime = ready_runtime();
let model_request = value(&runtime.model_list_request(None).unwrap());
runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","id":model_request["id"],"result":{"data":[{
"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"medium","description":"balanced"}]
}],"nextCursor":null}
})).unwrap().as_bytes()).unwrap();
let thread_request = value(
&runtime
.thread_start_request("gpt-5.6-terra", ReasoningEffort::Medium)
.unwrap(),
);
assert_eq!(
thread_request["params"],
json!({
"model": "gpt-5.6-terra",
"modelProvider": "openai",
"approvalPolicy": "never",
"sandbox": "read-only",
"ephemeral": true
})
);
runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","id":thread_request["id"],"result":{"thread":{"id":"thr-a"}}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let turn_request = value(
&runtime
.turn_start_request(
"thr-a",
"client-a",
"fixture prompt",
"gpt-5.6-terra",
ReasoningEffort::Medium,
)
.unwrap(),
);
assert_eq!(turn_request["params"]["threadId"], "thr-a");
assert_eq!(turn_request["params"]["model"], "gpt-5.6-terra");
assert_eq!(turn_request["params"]["effort"], "medium");
runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","id":turn_request["id"],"result":{"turn":{"id":"turn-a","status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
for message in [
json!({"jsonrpc":"2.0","method":"turn/started","params":{"threadId":"thr-a","turn":{"id":"turn-a","status":"inProgress","items":[],"error":null}}}),
json!({"jsonrpc":"2.0","method":"item/started","params":{"threadId":"thr-a","turnId":"turn-a","item":{"type":"agentMessage","id":"item-a","text":""}}}),
json!({"jsonrpc":"2.0","method":"item/agentMessage/delta","params":{"threadId":"thr-a","turnId":"turn-a","itemId":"item-a","delta":"draft"}}),
json!({"jsonrpc":"2.0","method":"item/completed","params":{"threadId":"thr-a","turnId":"turn-a","item":{"type":"agentMessage","id":"item-a","text":"canonical"}}}),
] {
runtime
.ingest_line(serde_json::to_string(&message).unwrap().as_bytes())
.unwrap();
}
record_turn_usage(&mut runtime, "thr-a", "turn-a", 12, 34);
let terminal = runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"turn/completed","params":{"threadId":"thr-a","turn":{"id":"turn-a","status":"completed","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
assert!(
matches!(terminal, RuntimeEvent::TurnCompleted(ref receipt) if
receipt.output.as_deref() == Some("canonical")
&& receipt.selected_model == "gpt-5.6-terra"
&& receipt.selected_effort == ReasoningEffort::Medium
&& receipt.input_tokens == Some(12)
&& receipt.output_tokens == Some(34)
&& receipt.replay == ReplayDisposition::DeniedVisibleOutput)
);
assert!(!format!("{terminal:?}").contains("canonical"));
}
#[test]
fn reroute_interrupt_side_effects_and_sensitive_canaries_fail_closed() {
let mut runtime = ready_runtime();
let token_canary = "token-canary-value";
let error = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"warning","params":{"accessToken":token_canary}
}))
.unwrap()
.as_bytes(),
)
.expect_err("token fields rejected");
assert_eq!(error, RuntimeError::SensitiveInputRejected);
assert!(!format!("{error:?} {error}").contains(token_canary));
let scalar_secret = "sk-1234567890abcdef1234567890";
assert_eq!(
runtime.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"warning","params":{"message":scalar_secret}
}))
.unwrap()
.as_bytes()
),
Err(RuntimeError::SensitiveInputRejected)
);
let unknown_tool = runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"item/started","params":{"threadId":"thr-x","turnId":"turn-x","item":{"type":"commandExecution","id":"item-x","command":"danger"}}
})).unwrap().as_bytes()).expect_err("side effects fail closed");
assert_eq!(unknown_tool, RuntimeError::SideEffectingItemRejected);
}
#[test]
fn canonical_mcp_tool_call_and_result_are_correlated_redacted_and_no_replay() {
let mut runtime = active_tool_runtime("thr-tool", "turn-tool");
let argument_canary = "argument-payload-canary";
let result_canary = "result-payload-canary";
let started = runtime
.ingest_line(
serde_json::to_string(&json!({
"method":"item/started","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"inProgress",
"arguments":{"query":argument_canary},"appContext":null,
"pluginId":null,"result":null,"error":null,"durationMs":null
}
}
}))
.unwrap()
.as_bytes(),
)
.expect("canonical tool call start");
assert!(matches!(started, RuntimeEvent::ToolCallStarted {
ref thread_id, ref turn_id, ref item_id
} if thread_id == "thr-tool" && turn_id == "turn-tool" && item_id == "call-tool-1"));
assert!(!format!("{started:?}").contains(argument_canary));
let completed = runtime
.ingest_line(
serde_json::to_string(&json!({
"method":"item/completed","params":{
"threadId":"thr-tool","turnId":"turn-tool","completedAtMs":1,
"item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"completed",
"arguments":{"query":argument_canary},"appContext":null,
"pluginId":null,
"result":{"content":[{"type":"text","text":result_canary}]},
"error":null,"durationMs":1
}
}
}))
.unwrap()
.as_bytes(),
)
.expect("authoritative tool result");
assert!(matches!(completed, RuntimeEvent::ToolCallCompleted {
ref thread_id, ref turn_id, ref item_id, status: ToolCallStatus::Completed
} if thread_id == "thr-tool" && turn_id == "turn-tool" && item_id == "call-tool-1"));
assert!(!format!("{completed:?}").contains(argument_canary));
assert!(!format!("{completed:?}").contains(result_canary));
record_turn_usage(&mut runtime, "thr-tool", "turn-tool", 23, 5);
let terminal = runtime.ingest_line(serde_json::to_string(&json!({
"method":"turn/completed","params":{
"threadId":"thr-tool","turn":{"id":"turn-tool","status":"completed","items":[],"error":null}
}
})).unwrap().as_bytes()).unwrap();
assert!(matches!(terminal, RuntimeEvent::TurnCompleted(ref receipt)
if receipt.replay == ReplayDisposition::DeniedToolOrSideEffect
&& receipt.failure.is_none()
&& receipt.input_tokens == Some(23)
&& receipt.output_tokens == Some(5)));
}
#[test]
fn tool_binding_and_lifecycle_order_are_immutable_and_fail_closed() {
let started = json!({
"method":"item/started","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"inProgress",
"arguments":{"query":"original","limit":1},"result":null,"error":null
}
}
});
let completed = |arguments: Value| {
json!({
"method":"item/completed","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"completed","arguments":arguments,
"result":{"content":[]},"error":null
}
}
})
};
let encode = |message: &Value| serde_json::to_string(message).unwrap();
let mut runtime = active_tool_runtime("thr-tool", "turn-tool");
assert_eq!(
runtime.ingest_line(encode(&completed(json!({"query":"original","limit":1}))).as_bytes()),
Err(RuntimeError::CorrelationMismatch),
"completion before start is rejected"
);
runtime.ingest_line(encode(&started).as_bytes()).unwrap();
assert_eq!(
runtime.ingest_line(encode(&started).as_bytes()),
Err(RuntimeError::CorrelationMismatch),
"duplicate start is rejected"
);
assert_eq!(
runtime
.ingest_line(encode(&completed(json!({"query":"substituted","limit":1}))).as_bytes()),
Err(RuntimeError::CorrelationMismatch),
"arguments digest cannot change after start"
);
runtime
.ingest_line(encode(&completed(json!({"limit":1,"query":"original"}))).as_bytes())
.expect("canonical completion with the same arguments");
assert_eq!(
runtime.ingest_line(encode(&completed(json!({"query":"original","limit":1}))).as_bytes()),
Err(RuntimeError::CorrelationMismatch),
"duplicate completion is rejected"
);
record_turn_usage(&mut runtime, "thr-tool", "turn-tool", 17, 6);
let terminal = json!({
"method":"turn/completed","params":{
"threadId":"thr-tool","turn":{"id":"turn-tool","status":"completed","items":[],"error":null}
}
});
runtime.ingest_line(encode(&terminal).as_bytes()).unwrap();
assert_eq!(
runtime.ingest_line(encode(&terminal).as_bytes()),
Err(RuntimeError::CorrelationMismatch),
"duplicate turn terminal is rejected"
);
}
#[test]
fn turn_terminal_before_tool_completion_is_rejected() {
let mut runtime = active_tool_runtime("thr-tool", "turn-tool");
runtime
.ingest_line(
serde_json::to_string(&json!({
"method":"item/started","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"inProgress","arguments":{},
"result":null,"error":null
}
}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
assert_eq!(
runtime.ingest_line(serde_json::to_string(&json!({
"method":"turn/completed","params":{
"threadId":"thr-tool","turn":{"id":"turn-tool","status":"completed","items":[],"error":null}
}
})).unwrap().as_bytes()),
Err(RuntimeError::ProtocolFailure)
);
}
#[test]
fn no_tools_mode_and_malformed_tool_results_fail_closed() {
let tool_started = json!({
"method":"item/started","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"inProgress","arguments":{},
"appContext":null,"pluginId":null,"result":null,"error":null,"durationMs":null
}
}
});
let mut no_tools = active_runtime("thr-tool", "turn-tool");
assert_eq!(
no_tools.ingest_line(serde_json::to_string(&tool_started).unwrap().as_bytes()),
Err(RuntimeError::SideEffectingItemRejected)
);
let mut tool_runtime = active_tool_runtime("thr-tool", "turn-tool");
tool_runtime
.ingest_line(serde_json::to_string(&tool_started).unwrap().as_bytes())
.unwrap();
let missing_result = json!({
"method":"item/completed","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"completed","arguments":{},
"appContext":null,"pluginId":null,"result":null,"error":null,"durationMs":1
}
}
});
assert_eq!(
tool_runtime.ingest_line(serde_json::to_string(&missing_result).unwrap().as_bytes()),
Err(RuntimeError::ProtocolFailure)
);
}
#[test]
fn tool_effect_survives_interruption_and_terminal_failures_are_content_free() {
let mut runtime = active_tool_runtime("thr-tool", "turn-tool");
runtime.ingest_line(serde_json::to_string(&json!({
"method":"item/started","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"inProgress","arguments":{},
"appContext":null,"pluginId":null,"result":null,"error":null,"durationMs":null
}
}
})).unwrap().as_bytes()).unwrap();
runtime
.ingest_line(
serde_json::to_string(&json!({
"method":"item/completed","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"failed","arguments":{},
"appContext":null,"pluginId":null,"result":null,
"error":{"message":"tool-result-canary"},"durationMs":1
}
}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let terminal = runtime.ingest_line(serde_json::to_string(&json!({
"method":"turn/completed","params":{
"threadId":"thr-tool","turn":{
"id":"turn-tool","status":"failed","items":[],
"error":{"message":"terminal-message-canary","codexErrorInfo":"usageLimitExceeded"}
}
}
})).unwrap().as_bytes()).unwrap();
assert!(matches!(terminal, RuntimeEvent::TurnCompleted(ref receipt)
if receipt.replay == ReplayDisposition::DeniedToolOrSideEffect
&& receipt.failure == Some(TerminalFailureClass::UsageLimit)));
assert!(!format!("{terminal:?}").contains("tool-result-canary"));
assert!(!format!("{terminal:?}").contains("terminal-message-canary"));
}
#[test]
fn interruption_after_tool_result_remains_correlated_and_never_replayable() {
let mut runtime = active_tool_runtime("thr-tool", "turn-tool");
for message in [
json!({
"method":"item/started","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"inProgress","arguments":{},
"appContext":null,"pluginId":null,"result":null,"error":null,"durationMs":null
}
}
}),
json!({
"method":"item/completed","params":{
"threadId":"thr-tool","turnId":"turn-tool","item":{
"type":"mcpToolCall","id":"call-tool-1","server":"fixture-server",
"tool":"fixture-read","status":"completed","arguments":{},
"appContext":null,"pluginId":null,"result":{"content":[]},
"error":null,"durationMs":1
}
}
}),
] {
runtime
.ingest_line(serde_json::to_string(&message).unwrap().as_bytes())
.unwrap();
}
let interrupt = value(
&runtime
.turn_interrupt_request("thr-tool", "turn-tool")
.unwrap(),
);
runtime
.ingest_line(
serde_json::to_string(&json!({
"id":interrupt["id"],"result":{}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let terminal = runtime.ingest_line(serde_json::to_string(&json!({
"method":"turn/completed","params":{
"threadId":"thr-tool","turn":{"id":"turn-tool","status":"interrupted","items":[],"error":null}
}
})).unwrap().as_bytes()).unwrap();
assert!(matches!(terminal, RuntimeEvent::TurnCompleted(ref receipt)
if receipt.status == harness_codex::TerminalStatus::Interrupted
&& receipt.replay == ReplayDisposition::DeniedToolOrSideEffect));
}
#[test]
fn interrupt_state_changes_only_after_a_request_is_emitted() {
let mut runtime = active_runtime("thr-capacity", "turn-capacity");
for _ in 0..64 {
runtime
.rate_limits_read_request()
.expect("fill pending capacity");
}
assert_eq!(
runtime.turn_interrupt_request("thr-capacity", "turn-capacity"),
Err(RuntimeError::PendingCapacityExceeded)
);
assert_eq!(
runtime.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"turn/completed","params":{
"threadId":"thr-capacity","turn":{"id":"turn-capacity","status":"interrupted","items":[],"error":null}
}
}))
.unwrap()
.as_bytes()
),
Err(RuntimeError::CorrelationMismatch)
);
}
#[test]
fn duplicate_items_and_aggregate_output_overflow_fail_closed() {
let mut runtime = active_runtime("thr-bounds", "turn-bounds");
let started = json!({
"jsonrpc":"2.0","method":"item/started","params":{
"threadId":"thr-bounds","turnId":"turn-bounds","item":{"type":"agentMessage","id":"item-1","text":""}
}
});
runtime
.ingest_line(serde_json::to_string(&started).unwrap().as_bytes())
.unwrap();
assert_eq!(
runtime.ingest_line(serde_json::to_string(&started).unwrap().as_bytes()),
Err(RuntimeError::CorrelationMismatch)
);
let chunk = "x".repeat(64 * 1024);
for _ in 0..4 {
runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"item/agentMessage/delta","params":{
"threadId":"thr-bounds","turnId":"turn-bounds","itemId":"item-1","delta":chunk
}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
}
assert_eq!(
runtime.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"item/agentMessage/delta","params":{
"threadId":"thr-bounds","turnId":"turn-bounds","itemId":"item-1","delta":"x"
}
}))
.unwrap()
.as_bytes()
),
Err(RuntimeError::OutputLimitExceeded)
);
}
#[test]
fn model_reroute_is_terminally_failed_and_never_replayable() {
let mut runtime = active_runtime("thr-reroute", "turn-reroute");
let event = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"model/rerouted","params":{
"threadId":"thr-reroute","turnId":"turn-reroute",
"fromModel":"gpt-5.6-terra","toModel":"unrequested-model","reason":"fixture"
}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
assert!(matches!(event, RuntimeEvent::ModelReroutedFailure { .. }));
let after_reroute = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"item/started","params":{
"threadId":"thr-reroute","turnId":"turn-reroute",
"item":{"type":"agentMessage","id":"late-item","text":""}
}
}))
.unwrap()
.as_bytes(),
)
.expect_err("rerouted turn cannot continue");
assert_eq!(after_reroute, RuntimeError::ProtocolFailure);
let terminal = runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"turn/completed","params":{
"threadId":"thr-reroute","turn":{"id":"turn-reroute","status":"failed","items":[],"error":{"message":"redacted"}}
}
})).unwrap().as_bytes()).unwrap();
assert!(
matches!(terminal, RuntimeEvent::TurnCompleted(ref receipt) if receipt.replay == ReplayDisposition::DeniedModelReroute)
);
}
#[test]
fn interrupt_ack_is_not_terminal_and_terminal_interrupted_is_correlated() {
let mut runtime = active_runtime("thr-interrupt", "turn-interrupt");
let request = value(
&runtime
.turn_interrupt_request("thr-interrupt", "turn-interrupt")
.unwrap(),
);
assert_eq!(request["method"], "turn/interrupt");
assert_eq!(
request["params"],
json!({"threadId":"thr-interrupt","turnId":"turn-interrupt"})
);
let ack = runtime
.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","id":request["id"],"result":{}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
assert!(matches!(ack, RuntimeEvent::InterruptAcknowledged { .. }));
let terminal = runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"turn/completed","params":{
"threadId":"thr-interrupt","turn":{"id":"turn-interrupt","status":"interrupted","items":[],"error":null}
}
})).unwrap().as_bytes()).unwrap();
assert!(
matches!(terminal, RuntimeEvent::TurnCompleted(ref receipt) if receipt.status == harness_codex::TerminalStatus::Interrupted)
);
}
#[test]
fn pagination_and_interleaved_turns_remain_bounded_and_disjoint() {
let mut runtime = ready_runtime();
let first = value(&runtime.model_list_request(None).unwrap());
runtime.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","id":first["id"],"result":{"data":[{
"id":"gpt-5.6-terra","displayName":"Terra","hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"medium","description":"balanced"}]
}],"nextCursor":"next-page"}
})).unwrap().as_bytes()).unwrap();
let next = value(&runtime.model_list_request(Some("next-page")).unwrap());
assert_eq!(next["params"]["cursor"], "next-page");
let too_many: Vec<Value> = (0..51)
.map(|index| {
json!({
"id":format!("model-{index}"),"displayName":"fixture","hidden":false,
"supportedReasoningEfforts":[{"reasoningEffort":"medium","description":"balanced"}]
})
})
.collect();
assert_eq!(
runtime.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","id":next["id"],"result":{"data":too_many,"nextCursor":null}
}))
.unwrap()
.as_bytes()
),
Err(RuntimeError::ModelPageLimitExceeded)
);
let mut left = active_runtime("thr-left", "turn-left");
let right_thread = value(
&left
.thread_start_request("gpt-5.6-terra", ReasoningEffort::Medium)
.unwrap(),
);
left.ingest_line(
serde_json::to_string(&json!({
"jsonrpc":"2.0","id":right_thread["id"],"result":{"thread":{"id":"thr-right"}}
}))
.unwrap()
.as_bytes(),
)
.unwrap();
let right_turn = value(
&left
.turn_start_request(
"thr-right",
"client-right",
"fixture",
"gpt-5.6-terra",
ReasoningEffort::Medium,
)
.unwrap(),
);
left.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","id":right_turn["id"],"result":{"turn":{"id":"turn-right","status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
left.ingest_line(serde_json::to_string(&json!({
"method":"turn/started","params":{"threadId":"thr-right","turn":{"id":"turn-right","status":"inProgress","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
for (thread, turn, item) in [
("thr-right", "turn-right", "item-right"),
("thr-left", "turn-left", "item-left"),
] {
left.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"item/started","params":{"threadId":thread,"turnId":turn,"item":{"type":"agentMessage","id":item,"text":""}}
})).unwrap().as_bytes()).unwrap();
left.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"item/completed","params":{"threadId":thread,"turnId":turn,"item":{"type":"agentMessage","id":item,"text":thread}}
})).unwrap().as_bytes()).unwrap();
}
for (thread, turn) in [("thr-left", "turn-left"), ("thr-right", "turn-right")] {
record_turn_usage(&mut left, thread, turn, 19, 7);
let event = left.ingest_line(serde_json::to_string(&json!({
"jsonrpc":"2.0","method":"turn/completed","params":{"threadId":thread,"turn":{"id":turn,"status":"completed","items":[],"error":null}}
})).unwrap().as_bytes()).unwrap();
assert!(
matches!(event, RuntimeEvent::TurnCompleted(ref receipt) if receipt.output.as_deref() == Some(thread))
);
}
}