//! Deterministic fake app-server conformance for the K1 Codex adapter family.
use kcode_k1_codex_conversations::State;
use kcode_k1_codex_events::{Inbound, RpcId, decode, rejection_payload};
use kcode_k1_codex_runtime::{
Adapter, Config, DynamicTool, ErrorKind, Event, ToolCall, ToolResult, Turn,
};
use kcode_k1_codex_shim::{
ASYNC_TOOL_ACKNOWLEDGEMENT, BoxCodec, Shim, ShimItem, ToolCallLauncher, ToolLaunchFuture,
};
use serde_json::{Value, json};
use std::{
fmt::Display,
fs::{self, OpenOptions},
io::Write as _,
os::unix::fs::PermissionsExt,
path::{Path, PathBuf},
sync::{
Arc, Mutex,
atomic::{AtomicU64, AtomicUsize, Ordering},
},
time::Duration,
};
const SCRIPT: &str = r#"#!/bin/sh
set -eu
: > transcript.log; : > launch.log
fail() { printf '%s\n' "$1" > fake.fail; exit 1; }
printf '%s\n' "$@" > args.log
[ "$1" = app-server ] || fail subcommand
history=0; effort=0
for arg in "$@"; do case "$arg" in 'history.persistence="none"') history=1 ;; 'model_reasoning_effort="medium"') effort=1 ;; model=*) fail global-model ;; esac; done
[ "$history/$effort" = 1/1 ] || fail global-config
printf 'diagnostic-begin\n' >&2; i=0
while [ "$i" -lt 2048 ]; do printf '0123456789abcdef' >&2; i=$((i + 1)); done
printf '\ndiagnostic-end\n' >&2
init=0; thread=0; turn=0; unsub=0; intr=0; nr=0; sr=0; ur=0; finished=0
out() { printf '%s\n' "$1"; }
rpc_id() { printf '%s\n' "$1" | sed -n 's/.*"id":\([0-9][0-9]*\).*/\1/p'; }
field() { printf '%s\n' "$1" | sed -n "s/.*\"$2\":\"\([^\"]*\)\".*/\1/p"; }
complete_main() {
if [ "$nr$sr$ur$finished" = 1110 ]; then
out '{"method":"turn/completed","params":{"threadId":"th1","turn":{"id":"tu1","status":"completed"}}}'
out '{"method":"turn/completed","params":{"threadId":"th2","turn":{"id":"tu2","status":"completed"}}}'
finished=1
fi
}
while IFS= read -r line; do
printf '%s\n' "$line" >> transcript.log
case "$line" in
*'"method":"initialize"'*)
init=$((init + 1)); [ "$init" -eq 1 ] || fail initialization-count; out '{"id":0,"result":{}}' ;;
*'"method":"initialized"'*) ;;
*'"method":"thread/start"'*)
case "$line" in *'"model":"requested-model"'*) ;; *) fail thread-model ;; esac
case "$line" in *'"dynamicTools":'*'"name":"launch"'*) ;; *) fail tool-registry ;; esac
case "$line" in *ephemeral*) fail ephemeral-thread ;; esac
thread=$((thread + 1)); id=$(rpc_id "$line"); out "{\"id\":$id,\"result\":{\"thread\":{\"id\":\"th$thread\"}}}" ;;
*'"method":"turn/start"'*)
turn=$((turn + 1)); id=$(rpc_id "$line"); th=$(field "$line" threadId); tu="tu$turn"
case "$turn:$th" in 1:th1|2:th2|3:th1|4:th3|5:th4|6:th5) ;; *) fail thread-routing ;; esac
if [ "$turn" -eq 6 ]; then case "$line" in *'H1\nH2\nprompt'*) ;; *) fail shim-history ;; esac; fi
out "{\"id\":$id,\"result\":{\"turn\":{\"id\":\"$tu\"}}}"
case "$turn" in
2)
out '{"method":"future/request","id":"unsupported","params":{}}'; out '{"method":"item/agentMessage/delta","params":{"threadId":"th1","turnId":"tu1","delta":"A"}}'; out '{"method":"item/agentMessage/delta","params":{"threadId":"th2","turnId":"tu2","delta":"B"}}'; out '{"method":"item/tool/call","id":41,"params":{"threadId":"th1","turnId":"tu1","callId":"num","tool":"launch","arguments":{"kind":"number"}}}'; out '{"method":"item/tool/call","id":"41","params":{"threadId":"th2","turnId":"tu2","callId":"str","tool":"launch","arguments":{"kind":"string"}}}' ;;
3)
out "{\"method\":\"item/agentMessage/delta\",\"params\":{\"threadId\":\"$th\",\"turnId\":\"$tu\",\"delta\":\"again\"}}"; out "{\"method\":\"item/tool/call\",\"id\":\"repeat-gate\",\"params\":{\"threadId\":\"$th\",\"turnId\":\"$tu\",\"callId\":\"gate\",\"tool\":\"launch\",\"arguments\":{}}}" ;;
4) out "{\"method\":\"turn/completed\",\"params\":{\"threadId\":\"$th\",\"turn\":{\"id\":\"$tu\",\"status\":\"completed\"}}}" ;;
5) out "{\"method\":\"item/agentMessage/delta\",\"params\":{\"threadId\":\"$th\",\"turnId\":\"$tu\",\"delta\":\"drop-ready\"}}" ;;
6)
out "{\"method\":\"item/agentMessage/delta\",\"params\":{\"threadId\":\"$th\",\"turnId\":\"$tu\",\"delta\":\"L\"}}"; out "{\"method\":\"item/agentMessage/delta\",\"params\":{\"threadId\":\"$th\",\"turnId\":\"$tu\",\"delta\":\"1\"}}"; out "{\"method\":\"item/tool/call\",\"id\":\"wave-1\",\"params\":{\"threadId\":\"$th\",\"turnId\":\"$tu\",\"callId\":\"c1\",\"tool\":\"launch\",\"arguments\":{\"wave\":1}}}" ;;
esac ;;
*'"method":"thread/unsubscribe"'*)
unsub=$((unsub + 1)); id=$(rpc_id "$line"); out "{\"id\":$id,\"result\":{}}" ;;
*'"method":"turn/interrupt"'*)
intr=$((intr + 1)); id=$(rpc_id "$line"); th=$(field "$line" threadId); tu=$(field "$line" turnId)
printf '%s/%s\n' "$th" "$tu" > interrupt.log; out "{\"id\":$id,\"result\":{}}"; out "{\"method\":\"turn/completed\",\"params\":{\"threadId\":\"$th\",\"turn\":{\"id\":\"$tu\",\"status\":\"interrupted\"}}}" ;;
*)
case "$line" in
*'"id":"unsupported"'*)
printf '%s\n' "$line" | grep -q '"code":-32601' || fail unsupported; ur=1 ;;
*'"id":41'*)
printf '%s\n' "$line" | grep -q '"text":"numeric"' || fail numeric-result; nr=1 ;;
*'"id":"41"'*)
printf '%s\n' "$line" | grep -q '"text":"string"' || fail string-result; sr=1 ;;
*'"id":"repeat-gate"'*) out '{"method":"turn/completed","params":{"threadId":"th1","turn":{"id":"tu3","status":"completed"}}}' ;;
*'"id":"wave-1"'*)
grep -qx 'c1:1' launch.log || fail launch-before-first-ack; out '{"method":"item/agentMessage/delta","params":{"threadId":"th5","turnId":"tu6","delta":"M"}}'; out '{"method":"item/agentMessage/delta","params":{"threadId":"th5","turnId":"tu6","delta":"2"}}'; out '{"method":"item/tool/call","id":72,"params":{"threadId":"th5","turnId":"tu6","callId":"c2","tool":"launch","arguments":{"wave":2}}}' ;;
*'"id":72'*)
grep -qx 'c2:2' launch.log || fail launch-before-second-ack; out '{"method":"item/agentMessage/delta","params":{"threadId":"th5","turnId":"tu6","delta":"R"}}'; out '{"method":"item/agentMessage/delta","params":{"threadId":"th5","turnId":"tu6","delta":"3"}}'; out '{"method":"turn/completed","params":{"threadId":"th5","turn":{"id":"tu6","status":"completed"}}}' ;;
esac
complete_main ;;
esac
done
[ "$init/$thread/$turn/$unsub/$intr" = 1/5/6/2/1 ] || fail final-counts
printf stopped > stopped.log
"#;
trait Text<T> {
fn text(self) -> Result<T, String>;
}
impl<T, E: Display> Text<T> for Result<T, E> {
fn text(self) -> Result<T, String> {
self.map_err(|error| error.to_string())
}
}
macro_rules! check {
($condition:expr, $message:literal) => {
if !$condition {
return Err($message.to_owned());
}
};
}
struct App {
directory: PathBuf,
executable: PathBuf,
}
impl App {
fn new() -> Result<Self, String> {
static NEXT: AtomicU64 = AtomicU64::new(0);
let suffix = NEXT.fetch_add(1, Ordering::Relaxed);
let directory =
std::env::temp_dir().join(format!("k1-codex-testkit-{}-{suffix}", std::process::id()));
fs::create_dir(&directory).text()?;
let executable = directory.join("codex");
fs::write(&executable, SCRIPT).text()?;
let mut permissions = fs::metadata(&executable).text()?.permissions();
permissions.set_mode(0o700);
fs::set_permissions(&executable, permissions).text()?;
Ok(Self {
directory,
executable,
})
}
fn path(&self, name: &str) -> PathBuf {
self.directory.join(name)
}
}
impl Drop for App {
fn drop(&mut self) {
let _ = fs::remove_dir_all(&self.directory);
}
}
struct Codec(Arc<AtomicUsize>);
impl BoxCodec for Codec {
type Box = String;
fn tool_call_box(&mut self, call: &ToolCall) -> String {
self.0.fetch_add(1, Ordering::Relaxed);
format!("{}:{}", call.call_id, call.arguments["wave"])
}
fn box_text<'a>(&self, box_: &'a String) -> &'a str {
box_
}
}
struct Launcher {
seen: Arc<Mutex<Vec<String>>>,
log: PathBuf,
}
impl ToolCallLauncher<String> for Launcher {
fn launch<'a>(&'a mut self, box_: &'a String) -> ToolLaunchFuture<'a> {
Box::pin(async move {
self.seen
.lock()
.map_err(|_| "launch record lock poisoned".to_owned())?
.push(box_.clone());
let mut log = OpenOptions::new().append(true).open(&self.log).text()?;
writeln!(log, "{box_}").text()
})
}
}
fn config(app: &App) -> Config {
Config {
executable: app.executable.clone(),
working_directory: app.directory.to_string_lossy().into_owned(),
model: "requested-model".to_owned(),
reasoning_effort: Some("medium".to_owned()),
base_instructions: "dynamic tools only".to_owned(),
tools: vec![DynamicTool {
name: "launch".to_owned(),
description: "launch a deterministic test call".to_owned(),
input_schema: json!({"type": "object"}),
}],
}
}
async fn next(turn: &mut Turn) -> Result<Event, String> {
turn.next_event()
.await
.ok_or_else(|| "turn event stream ended early".to_owned())
}
async fn text(turn: &mut Turn, expected: &str) -> Result<(), String> {
check!(
next(turn).await? == Event::TextDelta(expected.to_owned()),
"text event was wrong"
);
Ok(())
}
async fn done(turn: &mut Turn) -> Result<(), String> {
check!(next(turn).await? == Event::Done, "turn did not complete");
Ok(())
}
fn tool(event: Event, id: &str) -> Result<ToolCall, String> {
match event {
Event::ToolCall(call) if call.call_id == id => Ok(call),
other => Err(format!("expected tool call {id}, got {other:?}")),
}
}
async fn wait_for(mut ready: impl FnMut() -> bool, label: &str) -> Result<(), String> {
tokio::time::timeout(Duration::from_secs(5), async {
while !ready() {
tokio::task::yield_now().await;
}
})
.await
.map_err(|_| format!("timed out waiting for {label}"))
}
fn lower_layer_checks() -> Result<(), String> {
let Inbound::UnsupportedServerRequest(unsupported) =
decode(json!({"method": "future/request", "id": "u"}))
else {
return Err("unsupported request was misclassified".to_owned());
};
check!(
rejection_payload(&unsupported.id, -32601, "unsupported")["id"] == "u",
"unsupported rejection lost its ID"
);
let malformed = json!({
"method": "item/tool/call",
"id": 9,
"params": {"threadId": "t", "turnId": "u", "callId": "c", "tool": "launch"}
});
check!(
matches!(decode(malformed.clone()), Inbound::Malformed(_)),
"malformed call did not fail closed"
);
let malformed_id = RpcId::try_from(&malformed["id"]).text()?;
check!(
rejection_payload(&malformed_id, -32602, "malformed")["id"] == 9,
"malformed rejection lost its ID"
);
let Inbound::ModelReroute(reroute) = decode(json!({"method": "model/rerouted", "id": "r"}))
else {
return Err("model reroute was misclassified".to_owned());
};
let reroute_id = reroute.id.ok_or_else(|| "reroute lost its ID".to_owned())?;
check!(
rejection_payload(&reroute_id, -32602, "reroute")["id"] == "r",
"reroute rejection lost its ID"
);
let mut state = State::<(), ()>::default();
let serial = state.begin_turn("k", ()).text()?;
state.track_tool("k", serial, "call", &json!(7)).text()?;
let duplicate = state
.track_tool("k", serial, "call", &json!(8))
.expect_err("duplicate call must fail");
check!(
duplicate.kind == ErrorKind::Protocol
&& duplicate.message == "duplicate dynamic tool call id",
"duplicate call did not fail closed"
);
let duplicate = state
.track_tool("k", serial, "other", &json!(7))
.expect_err("duplicate RPC ID must fail");
check!(
duplicate.kind == ErrorKind::Protocol
&& duplicate.message == "duplicate app-server request id",
"duplicate RPC ID did not fail closed"
);
Ok(())
}
fn messages(path: &Path) -> Result<Vec<Value>, String> {
fs::read_to_string(path)
.text()?
.lines()
.map(|line| serde_json::from_str(line).text())
.collect()
}
fn response<'a>(messages: &'a [Value], id: &Value) -> Option<&'a Value> {
messages
.iter()
.find(|message| message.get("id") == Some(id) && message.get("result").is_some())
}
fn inspect(app: &App) -> Result<(), String> {
check!(!app.path("fake.fail").exists(), "fake app-server failed");
check!(
fs::read_to_string(app.path("stopped.log")).text()? == "stopped",
"app-server did not stop cleanly"
);
let messages = messages(&app.path("transcript.log"))?;
for (id, output) in [(json!(41), "numeric"), (json!("41"), "string")] {
let response =
response(&messages, &id).ok_or_else(|| "typed tool response missing".to_owned())?;
check!(
response
.pointer("/result/contentItems/0/text")
.and_then(Value::as_str)
== Some(output),
"typed tool response crossed IDs"
);
}
for id in [json!("wave-1"), json!(72)] {
let response =
response(&messages, &id).ok_or_else(|| "shim acknowledgement missing".to_owned())?;
check!(
response.pointer("/result/success").and_then(Value::as_bool) == Some(true)
&& response
.pointer("/result/contentItems/0/text")
.and_then(Value::as_str)
== Some(ASYNC_TOOL_ACKNOWLEDGEMENT),
"shim acknowledgement was not exact"
);
}
check!(
fs::read_to_string(app.path("launch.log")).text()? == "c1:1\nc2:2\n",
"launch order was wrong"
);
Ok(())
}
/// Runs the complete local fake-app-server conformance scenario.
pub async fn run_conformance() -> Result<(), String> {
lower_layer_checks()?;
let app = App::new()?;
let adapter = Adapter::open(config(&app)).await.text()?;
wait_for(
|| {
adapter
.diagnostics()
.windows(b"diagnostic-end".len())
.any(|window| window == b"diagnostic-end")
},
"stderr capture",
)
.await?;
let mut alpha = adapter.start_turn("alpha", "first").await.text()?;
let busy = adapter
.start_turn("alpha", "overlap")
.await
.expect_err("same-key overlap must fail");
check!(
busy.kind == ErrorKind::Busy,
"same-key overlap was not Busy"
);
check!(
busy.diagnostics.len() >= 32_768 && !busy.to_string().contains("diagnostic-begin"),
"diagnostics were truncated or leaked into Display"
);
let mut beta = adapter.start_turn("beta", "second").await.text()?;
text(&mut alpha, "A").await?;
text(&mut beta, "B").await?;
let numeric = tool(next(&mut alpha).await?, "num")?;
let string = tool(next(&mut beta).await?, "str")?;
alpha
.respond(
numeric.call_id,
ToolResult {
success: true,
output: "numeric".to_owned(),
},
)
.await
.text()?;
beta.respond(
string.call_id,
ToolResult {
success: true,
output: "string".to_owned(),
},
)
.await
.text()?;
done(&mut alpha).await?;
done(&mut beta).await?;
let mut repeat = adapter.start_turn("alpha", "again").await.text()?;
text(&mut repeat, "again").await?;
let gate = tool(next(&mut repeat).await?, "gate")?;
let busy = adapter
.close_conversation("alpha")
.await
.expect_err("active close must fail");
check!(busy.kind == ErrorKind::Busy, "active close was not Busy");
repeat
.respond(
gate.call_id,
ToolResult {
success: true,
output: "gate".to_owned(),
},
)
.await
.text()?;
done(&mut repeat).await?;
adapter.close_conversation("alpha").await.text()?;
adapter.close_conversation("alpha").await.text()?;
let mut fresh = adapter.start_turn("alpha", "fresh").await.text()?;
done(&mut fresh).await?;
let mut dropped = adapter.start_turn("drop", "drop").await.text()?;
text(&mut dropped, "drop-ready").await?;
drop(dropped);
wait_for(|| app.path("interrupt.log").exists(), "turn interrupt").await?;
let conversions = Arc::new(AtomicUsize::new(0));
let launches = Arc::new(Mutex::new(Vec::new()));
let launcher = Launcher {
seen: launches.clone(),
log: app.path("launch.log"),
};
let mut shim = Shim::new(
adapter.clone(),
"shim",
Codec(conversions.clone()),
Box::new(launcher),
);
shim.record_boxes(["H1".to_owned(), "H2".to_owned()]);
let output = shim.infer("prompt").await.text()?;
check!(
output.items
== vec![
ShimItem::Text("L1".to_owned()),
ShimItem::Box("c1:1".to_owned()),
ShimItem::Text("M2".to_owned()),
ShimItem::Box("c2:2".to_owned()),
ShimItem::Text("R3".to_owned()),
],
"shim event ordering or text coalescing was wrong"
);
check!(
shim.pending_box_count() == 0 && conversions.load(Ordering::Relaxed) == 2,
"shim conversion or history count was wrong"
);
{
let launches = launches
.lock()
.map_err(|_| "launch record lock poisoned".to_owned())?;
check!(
launches.as_slice() == ["c1:1", "c2:2"],
"shim launch order was wrong"
);
}
shim.close_conversation().await.text()?;
drop(shim);
drop(adapter);
wait_for(|| app.path("stopped.log").exists(), "bounded shutdown").await?;
inspect(&app)
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test(flavor = "current_thread")]
async fn adapter_family_conforms() {
let total = [
include_str!("../Cargo.toml"),
include_str!("../Documentation.md"),
include_str!("lib.rs"),
]
.iter()
.flat_map(|source| source.lines())
.filter(|line| !line.trim().is_empty())
.count();
assert!(total < 500, "package has {total} nonblank lines");
run_conformance().await.expect("adapter family conformance");
}
}