Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 21 additions & 4 deletions src/openhuman/inference/provider/claude_code/driver.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,23 @@ use tokio::sync::mpsc;

/// Hard timeout per turn (PLAN §8). If the CLI hangs (network stall,
/// infinite loop, MCP deadlock) we kill the child and surface a timeout.
const TURN_TIMEOUT: Duration = Duration::from_secs(300);
const DEFAULT_TURN_TIMEOUT_SECS: u64 = 900;

/// Hard timeout per turn, overridable with
/// `OPENHUMAN_CLAUDE_CODE_TURN_TIMEOUT_SECS`.
///
/// The default matches the harness's own 900s wall-clock backstop. It used to
/// be 300s, which is shorter than a turn the CLI is *expected* to take once
/// full access lets it run its own tools: the child was killed mid-work and the
/// turn surfaced as a provider timeout rather than a slow answer.
fn turn_timeout() -> Duration {
let secs = std::env::var("OPENHUMAN_CLAUDE_CODE_TURN_TIMEOUT_SECS")
.ok()
.and_then(|raw| raw.trim().parse::<u64>().ok())
.filter(|secs| *secs > 0)
.unwrap_or(DEFAULT_TURN_TIMEOUT_SECS);
Duration::from_secs(secs)
}

use super::event_mapper::EventMapper;
use super::input_builder::build_stdin;
Expand Down Expand Up @@ -431,7 +447,7 @@ pub async fn run_turn(ctx: TurnContext<'_>) -> anyhow::Result<ChatResponse> {

// Wrap the streaming + wait in a timeout so a stuck CLI doesn't
// block this task forever (PLAN §8).
let timed = tokio::time::timeout(TURN_TIMEOUT, async {
let timed = tokio::time::timeout(turn_timeout(), async {
loop {
let n = stdout
.read(&mut buf)
Expand Down Expand Up @@ -468,14 +484,15 @@ pub async fn run_turn(ctx: TurnContext<'_>) -> anyhow::Result<ChatResponse> {
Ok(inner) => inner?,
Err(_elapsed) => {
log::error!(
"[claude-code][driver] turn timeout ({TURN_TIMEOUT:?}) exceeded; killing child"
"[claude-code][driver] turn timeout ({:?}) exceeded; killing child",
turn_timeout()
);
// kill_on_drop handles cleanup, but explicit kill gives us
// a chance to collect stderr.
let _ = child.kill().await;
anyhow::bail!(
"[claude-code][driver] turn timed out after {:?}",
TURN_TIMEOUT
turn_timeout()
);
}
};
Expand Down
55 changes: 33 additions & 22 deletions src/openhuman/inference/provider/claude_code/event_mapper.rs
Original file line number Diff line number Diff line change
Expand Up @@ -172,17 +172,24 @@ impl EventMapper {
}
let call_id = call_id.unwrap();
let tool_name = tool_name.unwrap();
// The block is tracked so its `input_json_delta`s and its stop
// event are swallowed rather than leaking, but it is NOT
// surfaced to OpenHuman's harness — see the note on
// `on_block_stop`.
log::debug!(
"[claude-code][event-mapper] cli-internal tool_use name={tool_name} id={call_id} (not surfaced)"
);
self.blocks.insert(
index,
BlockState {
kind: BlockKind::Tool,
call_id: Some(call_id.clone()),
tool_name: Some(tool_name.clone()),
call_id: Some(call_id),
tool_name: Some(tool_name),
text_accum: String::new(),
input_accum: String::new(),
},
);
vec![ProviderDelta::ToolCallStart { call_id, tool_name }]
Vec::new()
}
_ => Vec::new(),
}
Expand Down Expand Up @@ -224,12 +231,10 @@ impl EventMapper {
.and_then(Value::as_str)
.unwrap_or("")
.to_string();
// Accumulated for the debug log only; the call itself is the
// CLI's to run, so nothing is emitted (see `on_block_stop`).
state.input_accum.push_str(&partial);
let call_id = state.call_id.clone().unwrap_or_default();
vec![ProviderDelta::ToolCallArgsDelta {
call_id,
delta: partial,
}]
Vec::new()
}
_ => Vec::new(),
}
Expand All @@ -240,20 +245,26 @@ impl EventMapper {
return Vec::new();
};
if state.kind == BlockKind::Tool {
let call_id = state.call_id.unwrap_or_default();
let name = state.tool_name.unwrap_or_default();
let arguments = if state.input_accum.trim().is_empty() {
"{}".to_string()
} else {
state.input_accum.clone()
};
self.tool_calls.push(ToolCall {
id: call_id,
name,
arguments,
// Claude Code CLI events carry no OpenAI-compat extra_content.
extra_content: None,
});
// A native `tool_use` block from this CLI is the CLI's OWN call —
// its builtins (Bash / Read / Write / Edit …) or a server from the
// `--mcp-config` we hand it. The CLI executes them itself inside
// its own agentic loop, which is why the matching `tool_result`
// blocks are deliberately dropped in `map_event`.
//
// Surfacing the *call* while dropping its *result* handed
// OpenHuman's harness a tool it does not own and cannot run: with
// `full_access` on (no `--disallowedTools`), a turn that reached
// for `Bash` produced repeated tool failures until the circuit
// breaker halted the run, and the turn then burned its 900s
// wall-clock backstop. So neither half is surfaced, and this
// provider behaves as what it is — a chat model whose tool use is
// internal. OpenHuman's own tools reach it through the prompt
// catalogue, not through native tool calls.
log::debug!(
"[claude-code][event-mapper] dropping cli-internal tool call name={} args_len={}",
state.tool_name.unwrap_or_default(),
state.input_accum.len()
);
}
Vec::new()
}
Expand Down
56 changes: 46 additions & 10 deletions src/openhuman/inference/provider/claude_code/event_mapper_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,22 +25,58 @@ fn text_streams_through() {
assert_eq!(m.final_text, "hello");
}

/// A native `tool_use` block is the CLI's own call — a builtin, or a server
/// from the `--mcp-config` it was handed — and the CLI runs it inside its own
/// loop. The matching `tool_result` was always dropped; surfacing the call
/// while dropping its result handed OpenHuman's harness a tool it cannot run,
/// which failed repeatedly until the circuit breaker halted the whole turn.
/// Neither half is surfaced now.
#[test]
fn tool_call_assembles_input() {
fn cli_internal_tool_calls_are_not_surfaced_to_the_harness() {
let mut m = EventMapper::new();
let start = json!({"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_1","name":"memory_search"}});
let d_args = json!({"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"q\":\"foo\"}"}});
let start = json!({"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_1","name":"Bash"}});
let d_args = json!({"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"command\":\"ls\"}"}});
let stop = json!({"type":"content_block_stop","index":1});
let starts = m.handle(ClaudeCodeEvent::StreamEvent { event: start });

assert!(
m.handle(ClaudeCodeEvent::StreamEvent { event: start })
.is_empty(),
"no ToolCallStart reaches the harness"
);
assert!(
matches!(&starts[0], ProviderDelta::ToolCallStart { tool_name, .. } if tool_name == "memory_search")
m.handle(ClaudeCodeEvent::StreamEvent { event: d_args })
.is_empty(),
"no ToolCallArgsDelta reaches the harness"
);
let args = m.handle(ClaudeCodeEvent::StreamEvent { event: d_args });
assert!(matches!(&args[0], ProviderDelta::ToolCallArgsDelta { .. }));
m.handle(ClaudeCodeEvent::StreamEvent { event: stop });
assert_eq!(m.tool_calls.len(), 1);
assert_eq!(m.tool_calls[0].name, "memory_search");
assert_eq!(m.tool_calls[0].arguments, r#"{"q":"foo"}"#);

assert!(
m.tool_calls.is_empty(),
"the aggregated response carries no tool calls for the harness to execute"
);
}

/// Text in the same turn still streams normally — dropping the tool block must
/// not swallow the CLI's actual answer.
#[test]
fn text_in_a_turn_with_a_cli_tool_call_still_streams() {
let mut m = EventMapper::new();
m.handle(ClaudeCodeEvent::StreamEvent {
event: json!({"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"call_1","name":"Write"}}),
});
m.handle(ClaudeCodeEvent::StreamEvent {
event: json!({"type":"content_block_stop","index":0}),
});
m.handle(ClaudeCodeEvent::StreamEvent {
event: text_block_start(1),
});
let deltas = m.handle(ClaudeCodeEvent::StreamEvent {
event: text_delta(1, "done"),
});

assert!(matches!(&deltas[0], ProviderDelta::TextDelta { delta } if delta == "done"));
assert_eq!(m.final_text, "done");
assert!(m.tool_calls.is_empty());
}

#[test]
Expand Down
Loading