feat(exec,app-server): a real JSON contract, and steering that does not abort

Three things the last pass left flagged. Two are done; the third turned out to
belong somewhere else and is reported rather than forced in.

## exec --json emits the schema instead of the SDK guessing it

`exec_events.rs` and `experimental_event_processor_with_json_output.rs` had no
`mod` declaration, so 683 lines defining the `thread.*`/`item.*` schema never
compiled. `sdk/typescript/src/events.ts` says "based on event types from
code-rs/exec/src/exec_events.rs" and then rebuilt those events client-side from
raw core events — the contract existed only in TypeScript, and nothing checked
the two halves agreed.

Declared the modules and fixed what drifted while dead: TaskStarted became a
unit variant, WebSearchEnd became WebSearchComplete, StreamError folded into
Error, ExecCommandEndEvent lost `aggregated_output`, print_config_summary lost a
parameter. `--json` now uses it (`--experimental-json` is an alias, matching
upstream), and the old raw-event processor is deleted rather than left as a
second way to do this. thread.started needed a route because exec consumes
SessionConfigured before the event loop, so EventProcessor grew a defaulted
`on_session_configured`.

Making it real immediately found a lie: the SDK's `Usage` declares five fields
and we emitted three. core::TokenUsage had `cache_write_input_tokens` and
`reasoning_output_tokens` all along; the processor dropped them. Against a mock
model:

  {"type":"thread.started","thread_id":"2bd35c77-..."}
  {"type":"turn.started"}
  {"type":"item.completed","item":{"id":"item_0","type":"agent_message","text":"Hello from the model."}}
  {"type":"turn.completed","usage":{"input_tokens":11,"cached_input_tokens":2,
   "cache_write_input_tokens":0,"output_tokens":5,"reasoning_output_tokens":1}}

The SDK now passes those through instead of reconstructing them. Its test suite
did not compile — at HEAD and in just-every upstream: runStreamed.test.ts uses
`path` and `createMockClient` without importing either, testCodex.ts passes
`config` to a type declaring `configOverrides`, and findCodexPath() referenced
four names (`moduleRequire`, `CODEX_NPM_NAME`, `createRequire`,
`PLATFORM_PACKAGE_BY_TARGET`) that were never declared anywhere. Declared them
against @hanzo/dev, added `env` to CodexOptions, pointed the harness at
code-rs/target/debug/dev instead of codex-rs/target/debug/codex.

  pnpm exec jest tests/runStreamed.test.ts --testTimeout=120000
  3 passed  (asserts thread.started, turn.started, item.completed, turn.completed
             and item.type === "agent_message" against the real binary)

## queueUserMessage: steer a turn instead of tearing it

`sendUserMessage` submits Op::UserInput, and core/src/codex/streaming.rs:998
calls `sess.abort()` before spawning a new turn — over app-server you could
interrupt a turn, never redirect one. Op::QueueUserInput folds items into the
running task instead, and only the TUI used it. It is now an app-server method.
Separate method, not a flag: interrupting and steering are different acts and a
caller should not pick the destructive one by accident.

  → {"method":"queueUserMessage","id":3,"params":{"conversationId":"999222eb-…",
     "items":[{"type":"text","data":{"text":"steer me"}}]}}
  ← {"id":3,"jsonrpc":"2.0","result":{}}
  ← {"error":{"code":-32600,"message":"conversation not found: 00000000-…"},"id":4}

## zrok: not ours to build

Assessed and deliberately not implemented here. `hanzo link` already composes
ttyd + share::start + publish_terminal into exactly this feature, and
cli/src/commands/share.rs calls itself "the ONE implementation of 'publish a
local port on the fabric'". `dev` has no framable surface anyway — app-server is
raw JSON-RPC — and no IAM identity to reach POST /v1/share/enable. The gap is in
cloud/cmd/boxd, which already execs `dev`. Details in the report.

Also: git.hanzo.ai release migration precondition written into LLM.md (four
call sites, forge serves /v1/ not /api/v1/, publish to both for a cycle before
moving the updater). SDK package was @openai/codex-sdk; now @hanzo/dev-sdk.

  cargo test -p {code-core,code-protocol,code-cli,code-app-server,
                 code-exec,code-mcp-server,code-tui} -- --test-threads=1
  1311 passed; 1 failed

The failure is git_info::test_collect_git_info_with_remote, still environmental:
the new narrower insteadOf rule only covers hanzoai/*, and the test's URL is
github.com/example/repo, which the broad rule still rewrites.

Version 0.2.189 -> 0.2.190.

Co-authored-by: Hanzo Dev <[email protected]>

LLM.md

@@ -73,6 +73,31 @@ ids answer on chat/completions but return system-prompt text on responses.
 is `npm pack` output that CI regenerates — it is gitignored on purpose; when it
 was committed it was a second, stale home for that download URL.
 
+### Moving releases to git.hanzo.ai — the precondition
+
+Releases are served by GitHub today. Nothing in this repo points at
+`git.hanzo.ai`, and retargeting it blind breaks every installed copy, because
+the updater and the installer both fetch a URL the user already has baked in.
+There are exactly four places, and they must move together:
+
+| What | Where | Currently |
+|---|---|---|
+| Update check | `code-rs/tui/src/updates.rs` `LATEST_RELEASE_URL` | `api.github.com/repos/hanzoai/dev/releases/latest` |
+| Release page | `code-rs/tui/src/updates.rs` `CODE_RELEASE_URL` | `github.com/hanzoai/dev/releases/latest` |
+| Binary download | `codex-cli/postinstall.js` `downloadUrl` | `github.com/hanzoai/dev/releases/download/v${version}/…` |
+| Asset upload | `.github/workflows/release.yml` | GitHub release assets |
+
+The forge serves its API at `/v1/`, **not** `/api/v1/` — so the update check
+becomes `git.hanzo.ai/v1/repos/hanzoai/dev/releases/latest`, and the JSON it
+returns must keep the `tag_name` and `assets[].name` fields `updates.rs` reads.
+
+**The flip is safe only once the forge is serving release assets for a tag that
+already exists on GitHub**, so an old client fetching the GitHub URL and a new
+client fetching the forge URL get the same bytes. Publish to both for one full
+release cycle, then move the updater, then stop publishing to GitHub. Moving the
+updater first strands everyone who has not yet updated — they poll a host that
+has no releases, and the CLI silently stops offering upgrades.
+
 ## Upstream lineage — where the code actually comes from
 
 Read this before attempting a sync. The directory names lie about the topology.

code-rs/Cargo.lock

@@ -1337,6 +1337,7 @@ dependencies = [
  "tokio",
  "tracing",
  "tracing-subscriber",
+ "ts-rs",
  "uuid",
 ]
 

code-rs/app-server-protocol/src/protocol/common.rs

@@ -418,6 +418,11 @@ client_request_definitions! {
         params: v1::SendUserMessageParams,
         response: v1::SendUserMessageResponse,
     },
+    /// Steer the running turn without tearing it down.
+    QueueUserMessage {
+        params: v1::QueueUserMessageParams,
+        response: v1::QueueUserMessageResponse,
+    },
     SendUserTurn {
         params: v1::SendUserTurnParams,
         response: v1::SendUserTurnResponse,

code-rs/app-server-protocol/src/protocol/v1.rs

@@ -408,6 +408,23 @@ pub struct SendUserMessageParams {
     pub items: Vec<InputItem>,
 }
 
+/// Steer a turn that is already running instead of replacing it.
+///
+/// `sendUserMessage` submits `Op::UserInput`, which aborts the running task and
+/// spawns a new one — the caller can interrupt, never redirect. This submits
+/// `Op::QueueUserInput`, which folds the items into the turn already in flight.
+/// With no turn running the two are equivalent.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+pub struct QueueUserMessageParams {
+    pub conversation_id: ThreadId,
+    pub items: Vec<InputItem>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
+#[serde(rename_all = "camelCase")]
+pub struct QueueUserMessageResponse {}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 pub struct SendUserTurnParams {

code-rs/app-server/src/code_message_processor.rs

@@ -100,6 +100,8 @@ use code_protocol::mcp_protocol::RemoveConversationListenerParams;
 use code_protocol::mcp_protocol::RemoveConversationSubscriptionResponse;
 use code_protocol::mcp_protocol::SetDefaultModelParams;
 use code_protocol::mcp_protocol::SetDefaultModelResponse;
+use code_protocol::mcp_protocol::QueueUserMessageParams;
+use code_protocol::mcp_protocol::QueueUserMessageResponse;
 use code_protocol::mcp_protocol::SendUserMessageParams;
 use code_protocol::mcp_protocol::SendUserMessageResponse;
 use code_protocol::mcp_protocol::SendUserTurnParams;
@@ -207,6 +209,9 @@ impl CodexMessageProcessor {
             ClientRequest::SendUserMessage { request_id, params } => {
                 self.send_user_message(request_id, params).await;
             }
+            ClientRequest::QueueUserMessage { request_id, params } => {
+                self.queue_user_message(request_id, params).await;
+            }
             ClientRequest::InterruptConversation { request_id, params } => {
                 self.interrupt_conversation(request_id, params).await;
             }
@@ -576,11 +581,14 @@ impl CodexMessageProcessor {
         }
     }
 
-    async fn send_user_message(&self, request_id: RequestId, params: SendUserMessageParams) {
-        let SendUserMessageParams {
-            conversation_id,
-            items,
-        } = params;
+    /// Resolve the conversation and map wire items to core items. `None` means
+    /// the error response has already been sent.
+    async fn resolve_conversation_items(
+        &self,
+        request_id: RequestId,
+        conversation_id: ConversationId,
+        items: Vec<WireInputItem>,
+    ) -> Option<(Arc<CodexConversation>, Vec<CoreInputItem>)> {
         let Ok(conversation) = self
             .conversation_manager
             .get_conversation(conversation_id)
@@ -592,7 +600,7 @@ impl CodexMessageProcessor {
                 data: None,
             };
             self.outgoing.send_error(request_id, error).await;
-            return;
+            return None;
         };
 
         let mapped_items: Vec<CoreInputItem> = items
@@ -604,7 +612,22 @@ impl CodexMessageProcessor {
             })
             .collect();
 
-        // Submit user input to the conversation.
+        Some((conversation, mapped_items))
+    }
+
+    async fn send_user_message(&self, request_id: RequestId, params: SendUserMessageParams) {
+        let SendUserMessageParams {
+            conversation_id,
+            items,
+        } = params;
+        let Some((conversation, mapped_items)) = self
+            .resolve_conversation_items(request_id.clone(), conversation_id, items)
+            .await
+        else {
+            return;
+        };
+
+        // Op::UserInput aborts any running task and spawns a new turn.
         let _ = conversation
             .submit(Op::UserInput {
                 items: mapped_items,
@@ -618,6 +641,35 @@ impl CodexMessageProcessor {
             .await;
     }
 
+    /// Steer the turn already in flight instead of replacing it.
+    ///
+    /// The difference from `send_user_message` is entirely in the Op:
+    /// `Op::QueueUserInput` folds the items into the running task, where
+    /// `Op::UserInput` calls `sess.abort()` first. With nothing running, core
+    /// spawns a turn either way.
+    async fn queue_user_message(&self, request_id: RequestId, params: QueueUserMessageParams) {
+        let QueueUserMessageParams {
+            conversation_id,
+            items,
+        } = params;
+        let Some((conversation, mapped_items)) = self
+            .resolve_conversation_items(request_id.clone(), conversation_id, items)
+            .await
+        else {
+            return;
+        };
+
+        let _ = conversation
+            .submit(Op::QueueUserInput {
+                items: mapped_items,
+            })
+            .await;
+
+        self.outgoing
+            .send_response(request_id, QueueUserMessageResponse {})
+            .await;
+    }
+
     #[allow(dead_code)]
     async fn send_user_turn(&self, request_id: RequestId, params: SendUserTurnParams) {
         let SendUserTurnParams {

code-rs/exec/Cargo.toml

@@ -47,6 +47,11 @@ tokio = { workspace = true, features = [
     "sync",
 ] }
 tracing = { workspace = true, features = ["log"] }
+ts-rs = { workspace = true, features = [
+    "uuid-impl",
+    "serde-json-impl",
+    "no-serde-warnings",
+] }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 once_cell = { workspace = true }
 

code-rs/exec/src/cli.rs

@@ -86,7 +86,7 @@ pub struct Cli {
     pub color: Color,
 
     /// Print events to stdout as JSONL.
-    #[arg(long = "json", default_value_t = false)]
+    #[arg(long = "json", alias = "experimental-json", default_value_t = false)]
     pub json: bool,
 
     /// Maximum wall-clock time budget (seconds) before aborting the run.

code-rs/exec/src/event_processor.rs

@@ -2,6 +2,7 @@ use std::path::Path;
 
 use code_core::config::Config;
 use code_core::protocol::Event;
+use code_core::protocol::SessionConfiguredEvent;
 
 pub(crate) enum CodexStatus {
     Running,
@@ -13,6 +14,11 @@ pub(crate) trait EventProcessor {
     /// Print summary of effective configuration and user prompt.
     fn print_config_summary(&mut self, config: &Config, prompt: &str);
 
+    /// The session handshake. `exec` consumes `SessionConfigured` before the
+    /// event loop starts, so a processor that needs it — JSON mode, which
+    /// derives `thread.started` from it — has to be handed it here.
+    fn on_session_configured(&mut self, _ev: &SessionConfiguredEvent) {}
+
     /// Handle a single event emitted by the agent.
     fn process_event(&mut self, event: Event) -> CodexStatus;
 

code-rs/exec/src/event_processor_with_json_output.rs

@@ -1,67 +0,0 @@
-use std::collections::HashMap;
-use std::path::PathBuf;
-
-use code_core::config::Config;
-use code_core::protocol::Event;
-use code_core::protocol::EventMsg;
-use code_core::protocol::TaskCompleteEvent;
-use serde_json::json;
-
-use crate::event_processor::CodexStatus;
-use crate::event_processor::EventProcessor;
-use crate::event_processor::handle_last_message;
-use code_common::create_config_summary_entries;
-
-pub(crate) struct EventProcessorWithJsonOutput {
-    last_message_path: Option<PathBuf>,
-    had_error: bool,
-}
-
-impl EventProcessorWithJsonOutput {
-    pub fn new(last_message_path: Option<PathBuf>) -> Self {
-        Self { last_message_path, had_error: false }
-    }
-}
-
-impl EventProcessor for EventProcessorWithJsonOutput {
-    fn print_config_summary(&mut self, config: &Config, prompt: &str) {
-        let entries = create_config_summary_entries(config)
-            .into_iter()
-            .map(|(key, value)| (key.to_string(), value))
-            .collect::<HashMap<String, String>>();
-        #[expect(clippy::expect_used)]
-        let config_json =
-            serde_json::to_string(&entries).expect("Failed to serialize config summary to JSON");
-        println!("{config_json}");
-
-        let prompt_json = json!({
-            "prompt": prompt,
-        });
-        println!("{prompt_json}");
-    }
-
-    fn process_event(&mut self, event: Event) -> CodexStatus {
-        match event.msg {
-            EventMsg::Error(_) => { self.had_error = true; CodexStatus::Running }
-            EventMsg::AgentMessageDelta(_) | EventMsg::AgentReasoningDelta(_) => {
-                // Suppress streaming events in JSON mode.
-                CodexStatus::Running
-            }
-            EventMsg::TaskComplete(TaskCompleteEvent { last_agent_message }) => {
-                if let Some(output_file) = self.last_message_path.as_deref() {
-                    handle_last_message(last_agent_message.as_deref(), output_file);
-                }
-                CodexStatus::InitiateShutdown
-            }
-            EventMsg::ShutdownComplete => CodexStatus::Shutdown,
-            _ => {
-                if let Ok(line) = serde_json::to_string(&event) {
-                    println!("{line}");
-                }
-                CodexStatus::Running
-            }
-        }
-    }
-
-    // exit_code handled by CLI; suppress unused warnings by omitting method.
-}

code-rs/exec/src/exec_events.rs

@@ -60,8 +60,14 @@ pub struct Usage {
     pub input_tokens: u64,
     /// The number of cached input tokens used during the turn.
     pub cached_input_tokens: u64,
+    /// The number of input tokens written to the prompt cache during the turn.
+    #[serde(default)]
+    pub cache_write_input_tokens: u64,
     /// The number of output tokens used during the turn.
     pub output_tokens: u64,
+    /// The number of reasoning output tokens used during the turn.
+    #[serde(default)]
+    pub reasoning_output_tokens: u64,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, TS)]

code-rs/exec/src/experimental_event_processor_with_json_output.rs

@@ -5,7 +5,7 @@ use std::sync::atomic::AtomicU64;
 use crate::event_processor::CodexStatus;
 use crate::event_processor::EventProcessor;
 use crate::event_processor::handle_last_message;
-use crate::exec_events::AssistantMessageItem;
+use crate::exec_events::AgentMessageItem;
 use crate::exec_events::CommandExecutionItem;
 use crate::exec_events::CommandExecutionStatus;
 use crate::exec_events::FileChangeItem;
@@ -46,8 +46,7 @@ use code_core::protocol::PatchApplyBeginEvent;
 use code_core::protocol::PatchApplyEndEvent;
 use code_core::protocol::SessionConfiguredEvent;
 use code_core::protocol::TaskCompleteEvent;
-use code_core::protocol::TaskStartedEvent;
-use code_core::protocol::WebSearchEndEvent;
+use code_core::protocol::WebSearchCompleteEvent;
 use tracing::error;
 use tracing::warn;
 
@@ -109,14 +108,14 @@ impl ExperimentalEventProcessorWithJsonOutput {
             EventMsg::PatchApplyBegin(ev) => self.handle_patch_apply_begin(ev),
             EventMsg::PatchApplyEnd(ev) => self.handle_patch_apply_end(ev),
             EventMsg::WebSearchBegin(_) => Vec::new(),
-            EventMsg::WebSearchEnd(ev) => self.handle_web_search_end(ev),
+            EventMsg::WebSearchComplete(ev) => self.handle_web_search_complete(ev),
             EventMsg::TokenCount(ev) => {
                 if let Some(info) = &ev.info {
                     self.last_total_token_usage = Some(info.total_token_usage.clone());
                 }
                 Vec::new()
             }
-            EventMsg::TaskStarted(ev) => self.handle_task_started(ev),
+            EventMsg::TaskStarted => self.handle_task_started(),
             EventMsg::TaskComplete(_) => self.handle_task_complete(),
             EventMsg::Error(ev) => {
                 let error = ThreadErrorEvent {
@@ -125,9 +124,6 @@ impl ExperimentalEventProcessorWithJsonOutput {
                 self.last_critical_error = Some(error.clone());
                 vec![ThreadEvent::Error(error)]
             }
-            EventMsg::StreamError(ev) => vec![ThreadEvent::Error(ThreadErrorEvent {
-                message: ev.message.clone(),
-            })],
             EventMsg::PlanUpdate(ev) => self.handle_plan_update(ev),
             _ => Vec::new(),
         }
@@ -147,11 +143,11 @@ impl ExperimentalEventProcessorWithJsonOutput {
         })]
     }
 
-    fn handle_web_search_end(&self, ev: &WebSearchEndEvent) -> Vec<ThreadEvent> {
+    fn handle_web_search_complete(&self, ev: &WebSearchCompleteEvent) -> Vec<ThreadEvent> {
         let item = ThreadItem {
             id: self.get_next_item_id(),
             details: ThreadItemDetails::WebSearch(WebSearchItem {
-                query: ev.query.clone(),
+                query: ev.query.clone().unwrap_or_default(),
             }),
         };
 
@@ -162,7 +158,7 @@ impl ExperimentalEventProcessorWithJsonOutput {
         let item = ThreadItem {
             id: self.get_next_item_id(),
 
-            details: ThreadItemDetails::AssistantMessage(AssistantMessageItem {
+            details: ThreadItemDetails::AgentMessage(AgentMessageItem {
                 text: payload.message.clone(),
             }),
         };
@@ -339,7 +335,7 @@ impl ExperimentalEventProcessorWithJsonOutput {
 
             details: ThreadItemDetails::CommandExecution(CommandExecutionItem {
                 command,
-                aggregated_output: ev.aggregated_output.clone(),
+                aggregated_output: format!("{}{}", ev.stdout, ev.stderr),
                 exit_code: Some(ev.exit_code),
                 status,
             }),
@@ -382,7 +378,7 @@ impl ExperimentalEventProcessorWithJsonOutput {
         vec![ThreadEvent::ItemStarted(ItemStartedEvent { item })]
     }
 
-    fn handle_task_started(&mut self, _: &TaskStartedEvent) -> Vec<ThreadEvent> {
+    fn handle_task_started(&mut self) -> Vec<ThreadEvent> {
         self.last_critical_error = None;
         vec![ThreadEvent::TurnStarted(TurnStartedEvent {})]
     }
@@ -392,7 +388,9 @@ impl ExperimentalEventProcessorWithJsonOutput {
             Usage {
                 input_tokens: u.input_tokens,
                 cached_input_tokens: u.cached_input_tokens,
+                cache_write_input_tokens: u.cache_write_input_tokens,
                 output_tokens: u.output_tokens,
+                reasoning_output_tokens: u.reasoning_output_tokens,
             }
         } else {
             Usage::default()
@@ -420,26 +418,29 @@ impl ExperimentalEventProcessorWithJsonOutput {
     }
 }
 
+impl ExperimentalEventProcessorWithJsonOutput {
+    fn emit(&self, events: Vec<ThreadEvent>) {
+        for event in events {
+            match serde_json::to_string(&event) {
+                Ok(line) => println!("{line}"),
+                Err(e) => error!("Failed to serialize event: {e:?}"),
+            }
+        }
+    }
+}
+
 impl EventProcessor for ExperimentalEventProcessorWithJsonOutput {
-    fn print_config_summary(&mut self, _: &Config, _: &str, ev: &SessionConfiguredEvent) {
-        self.process_event(Event {
-            id: "".to_string(),
-            msg: EventMsg::SessionConfigured(ev.clone()),
-        });
+    /// JSON mode emits no human-readable preamble.
+    fn print_config_summary(&mut self, _: &Config, _: &str) {}
+
+    fn on_session_configured(&mut self, ev: &SessionConfiguredEvent) {
+        let started = self.handle_session_configured(ev);
+        self.emit(started);
     }
 
     fn process_event(&mut self, event: Event) -> CodexStatus {
         let aggregated = self.collect_thread_events(&event);
-        for conv_event in aggregated {
-            match serde_json::to_string(&conv_event) {
-                Ok(line) => {
-                    println!("{line}");
-                }
-                Err(e) => {
-                    error!("Failed to serialize event: {e:?}");
-                }
-            }
-        }
+        self.emit(aggregated);
 
         let Event { msg, .. } = event;
 

code-rs/exec/src/lib.rs

@@ -1,7 +1,8 @@
 mod cli;
 mod event_processor;
 mod event_processor_with_human_output;
-mod event_processor_with_json_output;
+pub mod exec_events;
+mod experimental_event_processor_with_json_output;
 mod slash;
 
 pub use cli::Cli;
@@ -48,7 +49,7 @@ use code_protocol::protocol::SessionSource;
 use code_ollama::DEFAULT_OSS_MODEL;
 use code_protocol::config_types::SandboxMode;
 use event_processor_with_human_output::EventProcessorWithHumanOutput;
-use event_processor_with_json_output::EventProcessorWithJsonOutput;
+use experimental_event_processor_with_json_output::ExperimentalEventProcessorWithJsonOutput;
 use event_processor::handle_last_message;
 use code_git_tooling::GhostCommit;
 use code_git_tooling::CreateGhostCommitOptions;
@@ -477,7 +478,9 @@ pub async fn run_main(cli: Cli, code_linux_sandbox_exe: Option<PathBuf>) -> anyh
     }
     let stop_on_task_complete = auto_drive_goal.is_none() && auto_resolve_state.is_none();
     let mut event_processor: Box<dyn EventProcessor> = if json_mode {
-        Box::new(EventProcessorWithJsonOutput::new(last_message_file.clone()))
+        Box::new(ExperimentalEventProcessorWithJsonOutput::new(
+            last_message_file.clone(),
+        ))
     } else {
         Box::new(EventProcessorWithHumanOutput::create_with_ansi(
             stdout_with_ansi,
@@ -542,7 +545,8 @@ pub async fn run_main(cli: Cli, code_linux_sandbox_exe: Option<PathBuf>) -> anyh
     } else {
         event_processor.print_config_summary(&config, &summary_prompt);
     }
-    info!("Codex initialized with event: {session_configured:?}");
+    event_processor.on_session_configured(&session_configured);
+    info!("Hanzo Dev initialized with event: {session_configured:?}");
 
     if let Some(goal) = auto_drive_goal {
         return run_auto_drive_session(

code-rs/protocol/src/mcp_protocol.rs

@@ -156,6 +156,11 @@ pub enum ClientRequest {
         request_id: RequestId,
         params: SendUserMessageParams,
     },
+    QueueUserMessage {
+        #[serde(rename = "id")]
+        request_id: RequestId,
+        params: QueueUserMessageParams,
+    },
     SendUserTurn {
         #[serde(rename = "id")]
         request_id: RequestId,
@@ -639,6 +644,23 @@ pub struct SendUserMessageParams {
     pub items: Vec<InputItem>,
 }
 
+/// Steer a turn that is already running instead of replacing it.
+///
+/// `sendUserMessage` submits `Op::UserInput`, which calls `sess.abort()` and
+/// spawns a fresh turn — the caller can interrupt, never redirect. This submits
+/// `Op::QueueUserInput`, which folds the items into the turn already in flight.
+/// With no turn running, core spawns one either way.
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, TS)]
+#[serde(rename_all = "camelCase")]
+pub struct QueueUserMessageParams {
+    pub conversation_id: ConversationId,
+    pub items: Vec<InputItem>,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, TS)]
+#[serde(rename_all = "camelCase")]
+pub struct QueueUserMessageResponse {}
+
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, TS)]
 #[serde(rename_all = "camelCase")]
 pub struct SendUserTurnParams {
@@ -945,6 +967,33 @@ mod tests {
         Ok(())
     }
 
+    /// Locks the wire shape of the steering method. `sendUserMessage` submits
+    /// `Op::UserInput`, which aborts the running turn; this one submits
+    /// `Op::QueueUserInput`, which does not. They are separate methods
+    /// precisely so a caller cannot pick the destructive one by accident.
+    #[test]
+    fn serialize_queue_user_message() -> Result<()> {
+        let request = ClientRequest::QueueUserMessage {
+            request_id: RequestId::Integer(7),
+            params: QueueUserMessageParams {
+                conversation_id: ConversationId::default(),
+                items: vec![InputItem::Text {
+                    text: "steer".to_string(),
+                }],
+            },
+        };
+        let value = serde_json::to_value(&request)?;
+        assert_eq!(json!("queueUserMessage"), value["method"]);
+        assert_eq!(
+            json!([{ "type": "text", "data": { "text": "steer" } }]),
+            value["params"]["items"],
+        );
+        // And it round-trips back to the same variant.
+        let back: ClientRequest = serde_json::from_value(value)?;
+        assert!(matches!(back, ClientRequest::QueueUserMessage { .. }));
+        Ok(())
+    }
+
     #[test]
     fn test_conversation_id_default_is_not_zeroes() {
         let id = ConversationId::default();

codex-cli/package-lock.json

@@ -1,11 +1,11 @@
 {
   "name": "@hanzo/dev",
-  "version": "0.2.189",
+  "version": "0.2.190",
   "lockfileVersion": 3,
   "packages": {
     "": {
       "name": "@hanzo/dev",
-      "version": "0.2.189",
+      "version": "0.2.190",
       "license": "Apache-2.0",
       "bin": {
         "codex": "bin/codex.js"

codex-cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@hanzo/dev",
-  "version": "0.2.189",
+  "version": "0.2.190",
   "license": "Apache-2.0",
   "description": "Local AI coding agent for your terminal, powered by Hanzo AI",
   "bin": {
@@ -35,10 +35,10 @@
     "prettier": "^3.3.3"
   },
   "optionalDependencies": {
-    "@hanzo/dev-darwin-arm64": "0.2.189",
-    "@hanzo/dev-darwin-x64": "0.2.189",
-    "@hanzo/dev-linux-x64-musl": "0.2.189",
-    "@hanzo/dev-linux-arm64-musl": "0.2.189",
-    "@hanzo/dev-win32-x64": "0.2.189"
+    "@hanzo/dev-darwin-arm64": "0.2.190",
+    "@hanzo/dev-darwin-x64": "0.2.190",
+    "@hanzo/dev-linux-x64-musl": "0.2.190",
+    "@hanzo/dev-linux-arm64-musl": "0.2.190",
+    "@hanzo/dev-win32-x64": "0.2.190"
   }
 }

sdk/typescript/package.json

@@ -1,18 +1,18 @@
 {
-  "name": "@openai/codex-sdk",
+  "name": "@hanzo/dev-sdk",
   "version": "0.0.0-dev",
-  "description": "TypeScript SDK for Codex APIs.",
+  "description": "TypeScript SDK for Hanzo Dev.",
   "repository": {
     "type": "git",
-    "url": "git+https://github.com/openai/codex.git",
+    "url": "git+https://github.com/hanzoai/dev.git",
     "directory": "sdk/typescript"
   },
   "keywords": [
-    "openai",
-    "codex",
+    "hanzo",
+    "dev",
+    "agent",
     "sdk",
-    "typescript",
-    "api"
+    "typescript"
   ],
   "license": "Apache-2.0",
   "type": "module",

sdk/typescript/src/codex.ts

@@ -13,7 +13,7 @@ export class Codex {
   private options: CodexOptions;
 
   constructor(options: CodexOptions = {}) {
-    this.exec = new CodexExec(options.codexPathOverride, options.configOverrides);
+    this.exec = new CodexExec(options.codexPathOverride, options.configOverrides, options.env);
     this.options = options;
   }
 

sdk/typescript/src/codexOptions.ts

@@ -7,4 +7,6 @@ export type CodexOptions = {
   baseUrl?: string;
   apiKey?: string;
   configOverrides?: CodexConfigObject;
+  /** Environment for the agent process. Replaces the inherited environment. */
+  env?: Record<string, string>;
 };

sdk/typescript/src/exec.ts

@@ -5,8 +5,22 @@ import readline from "node:readline";
 import { CodexConfigObject, CodexConfigValue } from "./codexOptions";
 import { SandboxMode } from "./threadOptions";
 import path from "node:path";
+import { createRequire } from "node:module";
 import { fileURLToPath } from "node:url";
 
+const moduleRequire = createRequire(import.meta.url);
+
+/** The npm package that carries the CLI and its per-platform binaries. */
+const CODEX_NPM_NAME = "@hanzo/dev";
+
+const PLATFORM_PACKAGE_BY_TARGET: Record<string, string> = {
+  "aarch64-apple-darwin": "@hanzo/dev-darwin-arm64",
+  "x86_64-apple-darwin": "@hanzo/dev-darwin-x64",
+  "x86_64-unknown-linux-musl": "@hanzo/dev-linux-x64-musl",
+  "aarch64-unknown-linux-musl": "@hanzo/dev-linux-arm64-musl",
+  "x86_64-pc-windows-msvc": "@hanzo/dev-win32-x64",
+};
+
 export type CodexExecArgs = {
   input: string;
 
@@ -26,10 +40,16 @@ export type CodexExecArgs = {
 export class CodexExec {
   private executablePath: string;
   private configOverrides?: CodexConfigObject;
+  private env?: Record<string, string>;
 
-  constructor(executablePath: string | null = null, configOverrides?: CodexConfigObject) {
+  constructor(
+    executablePath: string | null = null,
+    configOverrides?: CodexConfigObject,
+    env?: Record<string, string>,
+  ) {
     this.executablePath = executablePath || findCodexPath();
     this.configOverrides = configOverrides;
+    this.env = env;
   }
 
   async *run(args: CodexExecArgs): AsyncGenerator<string> {
@@ -68,9 +88,9 @@ export class CodexExec {
       commandArgs.push("resume", args.threadId);
     }
 
-    const env = {
-      ...process.env,
-    };
+    const env: Record<string, string | undefined> = this.env
+      ? { ...this.env }
+      : { ...process.env };
     if (args.baseUrl) {
       env.OPENAI_BASE_URL = args.baseUrl;
     }
@@ -293,13 +313,13 @@ function findCodexPath() {
     vendorRoot = path.join(path.dirname(platformPackageJsonPath), "vendor");
   } catch {
     throw new Error(
-      `Unable to locate Codex CLI binaries. Ensure ${CODEX_NPM_NAME} is installed with optional dependencies.`,
+      `Unable to locate the Hanzo Dev binaries. Ensure ${CODEX_NPM_NAME} is installed with optional dependencies.`,
     );
   }
 
   const archRoot = path.join(vendorRoot, targetTriple);
-  const codexBinaryName = process.platform === "win32" ? "codex.exe" : "codex";
-  const binaryPath = path.join(archRoot, "codex", codexBinaryName);
+  const binaryName = process.platform === "win32" ? "dev.exe" : "dev";
+  const binaryPath = path.join(archRoot, "codex", binaryName);
 
   return binaryPath;
 }

sdk/typescript/src/thread.ts

@@ -85,17 +85,15 @@ export class Thread {
         continue;
       }
 
-      const threadEvents = mapCliEventToThreadEvents(parsed, this.id);
-      if (threadEvents.length === 0) {
+      const event = parseThreadEvent(parsed);
+      if (!event) {
         continue;
       }
 
-      for (const event of threadEvents) {
-        if (event.type === "thread.started") {
-          this._id = event.thread_id;
-        }
-        yield event;
+      if (event.type === "thread.started") {
+        this._id = event.thread_id;
       }
+      yield event;
     }
   }
 
@@ -126,145 +124,32 @@ export class Thread {
   }
 }
 
-function mapCliEventToThreadEvents(
-  raw: unknown,
-  currentThreadId: string | null,
-): ThreadEvent[] {
+/**
+ * `dev exec --json` emits the `thread.*` / `item.*` schema directly — the same
+ * shape declared in ./events.ts, which is generated from
+ * code-rs/exec/src/exec_events.rs. This used to reconstruct that schema from
+ * raw core events on this side, which meant the contract only ever existed in
+ * TypeScript and nothing verified the two halves agreed. They are one wire
+ * format now, so this only has to recognise it.
+ */
+const THREAD_EVENT_TYPES = new Set([
+  "thread.started",
+  "turn.started",
+  "turn.completed",
+  "turn.failed",
+  "item.started",
+  "item.updated",
+  "item.completed",
+  "error",
+]);
+
+function parseThreadEvent(raw: unknown): ThreadEvent | null {
   if (typeof raw !== "object" || raw === null) {
-    return [];
+    return null;
   }
-
-  const event = raw as Record<string, unknown>;
-  const msg = event.msg as Record<string, unknown> | undefined;
-  const msgType = typeof msg?.type === "string" ? (msg.type as string) : undefined;
-
-  if (!msgType || !msg) {
-    return [];
-  }
-
-  switch (msgType) {
-    case "session_configured": {
-      const threadId = deriveThreadId(event, msg, currentThreadId);
-      return [
-        { type: "thread.started", thread_id: threadId },
-      ];
-    }
-    case "task_started": {
-      const events: ThreadEvent[] = [];
-      if (!currentThreadId) {
-        const derivedThreadId = deriveThreadId(event, msg, currentThreadId);
-        events.push({ type: "thread.started", thread_id: derivedThreadId });
-      }
-      events.push({ type: "turn.started" });
-      return events;
-    }
-    case "agent_reasoning": {
-      const text = typeof msg?.text === "string" ? (msg.text as string) : "";
-      const reasoningItem: ThreadItem = {
-        id: extractEventId(event),
-        type: "reasoning",
-        text,
-      };
-      return [
-        {
-          type: "item.completed",
-          item: reasoningItem,
-        },
-      ];
-    }
-    case "agent_message": {
-      const message = typeof msg?.message === "string" ? (msg.message as string) : "";
-      const assistantItem: ThreadItem = {
-        id: extractEventId(event),
-        type: "agent_message",
-        text: message,
-      };
-      return [
-        {
-          type: "item.completed",
-          item: assistantItem,
-        },
-      ];
-    }
-    case "token_count": {
-      const info = msg?.info as { total_token_usage?: Record<string, unknown> } | undefined;
-      const usageInfo = info?.total_token_usage ?? {};
-      const usage = normalizeUsage(usageInfo);
-      return [
-        {
-          type: "turn.completed",
-          usage,
-        },
-      ];
-    }
-    case "error": {
-      const message = typeof msg?.message === "string" ? (msg.message as string) : "Unknown error";
-      return [
-        {
-          type: "error",
-          message,
-        },
-      ];
-    }
-    default:
-      return [];
+  const type = (raw as Record<string, unknown>).type;
+  if (typeof type !== "string" || !THREAD_EVENT_TYPES.has(type)) {
+    return null;
   }
-}
-
-function deriveThreadId(
-  event: Record<string, unknown>,
-  msg: Record<string, unknown>,
-  currentThreadId: string | null,
-): string {
-  if (typeof msg.session_id === "string" && msg.session_id.length > 0) {
-    return msg.session_id;
-  }
-
-  const explicitId = event.thread_id;
-  if (typeof explicitId === "string" && explicitId.length > 0) {
-    return explicitId;
-  }
-
-  if (currentThreadId) {
-    return currentThreadId;
-  }
-
-  const eventId = event.id;
-  if (typeof eventId === "string" && eventId.length > 0) {
-    return eventId;
-  }
-
-  return `thread-${Date.now()}`;
-}
-
-function extractEventId(event: Record<string, unknown>): string {
-  const identifier = event.id;
-  if (typeof identifier === "string" && identifier.length > 0) {
-    return identifier;
-  }
-  const orderSequence = typeof event.event_seq === "number" ? event.event_seq : Date.now();
-  return `event-${orderSequence}`;
-}
-
-function normalizeUsage(raw: Record<string, unknown>): {
-  input_tokens: number;
-  cached_input_tokens: number;
-  output_tokens: number;
-} {
-  const toNumber = (value: unknown): number => {
-    if (typeof value === "number" && Number.isFinite(value)) {
-      return value;
-    }
-    if (typeof value === "string") {
-      const parsed = Number(value);
-      return Number.isFinite(parsed) ? parsed : 0;
-    }
-    return 0;
-  };
-
-  return {
-    input_tokens: toNumber(raw.input_tokens),
-    cached_input_tokens: toNumber(raw.cached_input_tokens),
-    output_tokens: toNumber(raw.output_tokens),
-  };
+  return raw as ThreadEvent;
 }

sdk/typescript/tests/runStreamed.test.ts

@@ -9,8 +9,7 @@ import {
   sse,
   startResponsesTestProxy,
 } from "./responsesProxy";
-
-const codexExecPath = path.join(process.cwd(), "..", "..", "code-rs", "target", "debug", "code");
+import { createMockClient } from "./testCodex";
 
 describe("Codex", () => {
   it("returns thread events", async () => {

sdk/typescript/tests/testCodex.ts

@@ -3,9 +3,11 @@ import path from "node:path";
 import { Codex } from "../src/codex";
 import type { CodexConfigObject } from "../src/codexOptions";
 
+// code-rs is the workspace that ships; codex-rs is the read-only upstream
+// mirror and building it produces nothing a user runs.
 export const codexExecPath =
   process.env.CODEX_EXEC_PATH ??
-  path.join(process.cwd(), "..", "..", "codex-rs", "target", "debug", "codex");
+  path.join(process.cwd(), "..", "..", "code-rs", "target", "debug", "dev");
 
 type CreateTestClientOptions = {
   apiKey?: string;
@@ -46,7 +48,7 @@ export function createTestClient(options: CreateTestClientOptions = {}): TestCli
       codexPathOverride: codexExecPath,
       baseUrl: options.baseUrl,
       apiKey: options.apiKey,
-      config: mergeTestConfig(options.baseUrl, options.config),
+      configOverrides: mergeTestConfig(options.baseUrl, options.config),
       env,
     }),
   };