use std::{collections::HashMap, path::Path, sync::Arc};
use nanocodex_core::{
AgentEventKind, EventSink, MODEL, ModelConfig, Prompt, ResponseItem, ToolDefinition, Usage,
responses::RequestProfile,
};
use nanocodex_service::{
CodeCall, CodeCallKind, ResponsesAttempt, ResponsesAttemptFactory, ResponsesClient,
ResponsesOutput, ResponsesServiceResponse, TRANSPORT, TransportStats, TurnResult,
};
use serde_json::value::RawValue;
use tokio::sync::watch;
use tower::Service;
use tracing::{Instrument, info, info_span};
use web_time::Instant;
use super::{
CompactionCompleted, CompactionFailed, CompactionStarted, ModelCallCompleted, ModelCallFailed,
ModelCallStarted, RunError, RunStarted, RunStats, RunSteered, ToolCallArguments, ToolCallEvent,
ToolResultEvent, WarmupCompleted, WarmupFailed, WarmupStarted,
agents_md::load_project_instructions,
compaction,
context_manager::ContextManager,
display_endpoint, elapsed_ns,
input::{
custom_tool_notification, custom_tool_output, function_tool_output, task_context,
task_input, turn_aborted,
},
resolve_workspace, terminal_payload,
};
use crate::{NanocodexError, Result};
use nanocodex_tools::{
ImageGenerationConfig, NestedToolCall, OwnedToolContext, ToolContext, ToolOutputBody,
ToolRuntime, ToolRuntimeControl, Tools, WebSearchConfig, prepare_output_images,
prepare_user_input,
};
pub(crate) struct ModelRun<S> {
events: EventSink,
config: Arc<ModelConfig>,
client: ResponsesClient<S>,
transport_stats: Arc<TransportStats>,
started_at: Instant,
stats: RunStats,
server_reasoning_included: bool,
session: Option<ModelSessionState>,
active_tools: Option<ToolRuntimeControl>,
active_tool_call: Option<ActiveToolCall>,
tool_call_indices: HashMap<Box<str>, u32>,
tools: Tools,
lineage_id: Arc<str>,
}
pub(crate) enum ModelTurnOutcome {
Completed(CompletedModelTurn),
Cancelled(ModelCheckpoint),
}
pub(crate) struct CompletedModelTurn {
pub(crate) final_message: String,
pub(crate) checkpoint: ModelCheckpoint,
}
#[derive(Clone)]
pub(crate) struct ModelCheckpoint {
workspace: String,
conversation: ConversationState,
preserve_inherited_delta: bool,
}
impl ModelCheckpoint {
pub(crate) fn workspace(&self) -> &str {
&self.workspace
}
}
#[cfg(not(target_family = "wasm"))]
pub(crate) trait AgentSend: Send {}
#[cfg(not(target_family = "wasm"))]
impl<T: Send> AgentSend for T {}
#[cfg(target_family = "wasm")]
pub(crate) trait AgentSend {}
#[cfg(target_family = "wasm")]
impl<T> AgentSend for T {}
struct ModelSessionState {
workspace: String,
tools: ToolRuntime,
factory: ResponsesAttemptFactory,
conversation: ConversationState,
preserve_inherited_delta: bool,
}
struct ActiveToolCall {
call_id: String,
name: String,
kind: CodeCallKind,
started_at: Instant,
}
enum ModelTaskOutcome {
Completed(String),
Cancelled,
}
#[derive(Clone)]
struct ConversationState {
canonical_context: Arc<ResponseItem>,
context: ContextManager,
delta_start: usize,
previous_response_id: Option<String>,
}
impl ConversationState {
fn new(history: Vec<ResponseItem>) -> Result<Self> {
let canonical_context =
history
.first()
.cloned()
.ok_or(NanocodexError::MalformedResponse {
detail: "task input did not include initial context",
})?;
Ok(Self {
canonical_context: Arc::new(canonical_context),
context: ContextManager::new(history),
delta_start: 0,
previous_response_id: None,
})
}
fn flattened_history(&self) -> Vec<ResponseItem> {
self.context.flattened_items()
}
fn clear_delta(&mut self) {
self.delta_start = self.context.len();
}
fn append(&mut self, items: impl IntoIterator<Item = ResponseItem>) {
self.context.record_items(items);
}
fn update_token_info(&mut self, usage: Option<&Usage>) {
self.context.update_token_info(usage);
}
fn active_context_tokens(&self, server_reasoning_included: bool) -> u64 {
self.context
.active_context_tokens(server_reasoning_included)
}
fn prompt_history(&self) -> nanocodex_core::responses::ResponseHistory {
self.context.prompt_items()
}
fn shared_history(&self) -> nanocodex_core::responses::ResponseHistory {
self.context.shared_items()
}
fn install_compaction(
&mut self,
item: ResponseItem,
canonical_context: ResponseItem,
request_prefix: &[ResponseItem],
) {
let history =
compaction::install_history(&self.context.flattened_items(), &canonical_context, item);
self.canonical_context = Arc::new(canonical_context);
self.context.replace_and_recompute(history, request_prefix);
self.delta_start = 0;
self.previous_response_id = None;
}
fn reset_for_full_request(&mut self) {
self.delta_start = 0;
self.previous_response_id = None;
}
fn commit(&mut self) -> Result<()> {
if self.previous_response_id.is_none() {
return Err(NanocodexError::MalformedResponse {
detail: "completed turn did not have a response ID",
});
}
self.context.commit_tail();
self.delta_start = self.context.len();
Ok(())
}
fn commit_interrupted(&mut self) {
self.reset_for_full_request();
self.context.commit_tail();
}
}
impl<S> ModelRun<S> {
pub(crate) fn new(
events: EventSink,
config: Arc<ModelConfig>,
client: ResponsesClient<S>,
transport_stats: Arc<TransportStats>,
tools: Tools,
lineage_id: Arc<str>,
) -> Self {
Self {
events,
config,
client,
transport_stats,
started_at: Instant::now(),
stats: RunStats::default(),
server_reasoning_included: false,
session: None,
active_tools: None,
active_tool_call: None,
tool_call_indices: HashMap::new(),
tools,
lineage_id,
}
}
pub(crate) fn from_checkpoint(
events: EventSink,
config: Arc<ModelConfig>,
client: ResponsesClient<S>,
transport_stats: Arc<TransportStats>,
tools: Tools,
lineage_id: Arc<str>,
checkpoint: ModelCheckpoint,
) -> Self {
let runtime = tool_runtime(&checkpoint.workspace, &config, &tools);
let active_tools = runtime.control();
let factory = ResponsesAttemptFactory::new(
request_profile(
events.request_id(),
&lineage_id,
runtime.model_specs(),
config.system_prompt(),
),
events.clone(),
Arc::clone(&transport_stats),
);
Self {
events,
config,
client,
transport_stats,
started_at: Instant::now(),
stats: RunStats::default(),
server_reasoning_included: false,
session: Some(ModelSessionState {
workspace: checkpoint.workspace,
tools: runtime,
factory,
conversation: checkpoint.conversation,
preserve_inherited_delta: checkpoint.preserve_inherited_delta,
}),
active_tools: Some(active_tools),
active_tool_call: None,
tool_call_indices: HashMap::new(),
tools,
lineage_id,
}
}
}
impl<S> ModelRun<S>
where
S: Service<ResponsesAttempt, Response = ResponsesServiceResponse> + AgentSend + 'static,
S::Error: Into<NanocodexError>,
S::Future: AgentSend,
{
pub(crate) fn emit_cancelled_before_start(
&mut self,
task: &Prompt,
workspace: Option<&str>,
) -> Result<()> {
self.started_at = Instant::now();
self.stats = RunStats::default();
self.events.emit(
AgentEventKind::RunStarted,
RunStarted {
mode: "openai_model",
model: MODEL,
effort: self.config.thinking.as_str(),
transport: TRANSPORT,
orchestration: ModelConfig::orchestration(),
websocket_url: display_endpoint(&self.config.websocket_url),
workspace,
instruction_bytes: task.instruction.text_bytes(),
},
)?;
let error = NanocodexError::TurnCancelled;
let message = error.to_string();
self.events
.emit(AgentEventKind::RunError, RunError { message: &message })?;
self.events.emit(
AgentEventKind::RunFailed,
terminal_payload(
"cancelled",
self.started_at.elapsed(),
&self.config,
&self.stats,
),
)?;
Ok(())
}
pub(crate) async fn execute(
&mut self,
task: Prompt,
workspace: Option<Arc<str>>,
steers: tokio::sync::mpsc::Receiver<Prompt>,
mut cancel: tokio::sync::oneshot::Receiver<()>,
fork_snapshots: watch::Sender<Option<ModelCheckpoint>>,
) -> Result<ModelTurnOutcome> {
self.started_at = Instant::now();
self.stats = RunStats::default();
let transport_before = self.transport_stats.snapshot();
self.events.emit(
AgentEventKind::RunStarted,
RunStarted {
mode: "openai_model",
model: MODEL,
effort: self.config.thinking.as_str(),
transport: TRANSPORT,
orchestration: ModelConfig::orchestration(),
websocket_url: display_endpoint(&self.config.websocket_url),
workspace: workspace.as_deref(),
instruction_bytes: task.instruction.text_bytes(),
},
)?;
let outcome = self
.execute_task(task, workspace, steers, &mut cancel, &fork_snapshots)
.await;
let elapsed = self.started_at.elapsed();
match outcome {
Ok(ModelTaskOutcome::Completed(message)) => {
self.stats
.apply_transport(self.transport_stats.since(transport_before));
self.events.emit(
AgentEventKind::RunCompleted,
terminal_payload("completed", elapsed, &self.config, &self.stats),
)?;
let checkpoint = self.commit_checkpoint()?;
Ok(ModelTurnOutcome::Completed(CompletedModelTurn {
final_message: message,
checkpoint,
}))
}
Ok(ModelTaskOutcome::Cancelled) => {
if let Some(tools) = &self.active_tools {
tools.cancel().await;
}
let checkpoint = self.commit_interrupted_checkpoint()?;
let elapsed = self.started_at.elapsed();
let error = NanocodexError::TurnCancelled;
let message = error.to_string();
self.events
.emit(AgentEventKind::RunError, RunError { message: &message })?;
self.stats
.apply_transport(self.transport_stats.since(transport_before));
self.events.emit(
AgentEventKind::RunFailed,
terminal_payload("cancelled", elapsed, &self.config, &self.stats),
)?;
Ok(ModelTurnOutcome::Cancelled(checkpoint))
}
Err(error) => {
let message = error.to_string();
self.events
.emit(AgentEventKind::RunError, RunError { message: &message })?;
self.stats
.apply_transport(self.transport_stats.since(transport_before));
self.events.emit(
AgentEventKind::RunFailed,
terminal_payload("failed", elapsed, &self.config, &self.stats),
)?;
Err(error)
}
}
}
async fn execute_task(
&mut self,
task: Prompt,
requested_workspace: Option<Arc<str>>,
steers: tokio::sync::mpsc::Receiver<Prompt>,
cancel: &mut tokio::sync::oneshot::Receiver<()>,
fork_snapshots: &watch::Sender<Option<ModelCheckpoint>>,
) -> Result<ModelTaskOutcome> {
let mut session = if let Some(mut session) = self.session.take() {
if let Some(requested) = requested_workspace.as_deref() {
let resolved = match resolve_workspace(Some(requested)) {
Ok(resolved) => resolved,
Err(error) => {
self.session = Some(session);
return Err(error);
}
};
if resolved != session.workspace {
let current = session.workspace.clone();
self.session = Some(session);
return Err(NanocodexError::WorkspaceChanged {
current,
requested: resolved,
});
}
}
let user_content = prepare_user_input(&task.instruction).await;
if session.preserve_inherited_delta {
session.preserve_inherited_delta = false;
} else {
session.conversation.clear_delta();
}
session.conversation.append([ResponseItem::message(
nanocodex_core::MessageRole::User,
user_content,
)]);
session
} else {
let workspace = resolve_workspace(requested_workspace.as_deref())?;
let project_instructions = load_project_instructions(Path::new(&workspace))?;
let tools = tool_runtime(&workspace, &self.config, &self.tools);
self.active_tools = Some(tools.control());
let factory = ResponsesAttemptFactory::new(
request_profile(
self.events.request_id(),
&self.lineage_id,
tools.model_specs(),
self.config.system_prompt(),
),
self.events.clone(),
Arc::clone(&self.transport_stats),
);
let user_content = prepare_user_input(&task.instruction).await;
let history = task_input(
user_content,
&workspace,
tools.default_shell_name(),
project_instructions.as_deref(),
);
let conversation = ConversationState::new(history)?;
let mut session = ModelSessionState {
workspace,
tools,
factory,
conversation,
preserve_inherited_delta: false,
};
Self::publish_fork_snapshot(&mut session, fork_snapshots);
let warmup = {
let warmup = self.perform_warmup(&session.factory);
tokio::pin!(warmup);
tokio::select! {
biased;
_ = &mut *cancel => None,
outcome = &mut warmup => Some(outcome),
}
};
let Some(warmup) = warmup else {
self.session = Some(session);
return Ok(ModelTaskOutcome::Cancelled);
};
match warmup {
Ok(response_id) => session.conversation.previous_response_id = Some(response_id),
Err(error) if error.responses_error().is_some() => {
session.conversation.reset_for_full_request();
self.stats.last_response_id = None;
}
Err(error) => return Err(error),
}
session
};
let outcome = {
let task = self.drive_session(&mut session, steers, fork_snapshots);
tokio::pin!(task);
tokio::select! {
biased;
_ = &mut *cancel => None,
outcome = &mut task => Some(outcome),
}
};
self.session = Some(session);
match outcome {
Some(outcome) => outcome.map(ModelTaskOutcome::Completed),
None => Ok(ModelTaskOutcome::Cancelled),
}
}
fn commit_checkpoint(&mut self) -> Result<ModelCheckpoint> {
let session = self
.session
.as_mut()
.ok_or(NanocodexError::InvalidAttemptState {
detail: "completed turn did not have a model session",
})?;
session.conversation.commit()?;
Ok(ModelCheckpoint {
workspace: session.workspace.clone(),
conversation: session.conversation.clone(),
preserve_inherited_delta: false,
})
}
fn commit_interrupted_checkpoint(&mut self) -> Result<ModelCheckpoint> {
let aborted_output = if let Some(call) = self.active_tool_call.take() {
let duration_ns = elapsed_ns(call.started_at);
let output = ToolOutputBody::Text(format!(
"Wall time: {:.3} seconds\naborted by user",
call.started_at.elapsed().as_secs_f64()
));
self.stats.tool_wall_duration_ns += duration_ns;
self.events.emit(
AgentEventKind::ToolResult,
ToolResultEvent {
call_id: &call.call_id,
tool: &call.name,
status: "cancelled",
duration_ns,
result: &output,
metadata: None,
},
)?;
Some(match call.kind {
CodeCallKind::Custom => custom_tool_output(call.call_id, output),
CodeCallKind::Function => function_tool_output(call.call_id, output),
})
} else {
None
};
let session = self
.session
.as_mut()
.ok_or(NanocodexError::InvalidAttemptState {
detail: "cancelled turn did not have a model session",
})?;
session.conversation.append(aborted_output);
session.conversation.append([turn_aborted()]);
session.conversation.commit_interrupted();
Ok(ModelCheckpoint {
workspace: session.workspace.clone(),
conversation: session.conversation.clone(),
preserve_inherited_delta: false,
})
}
fn publish_fork_snapshot(
session: &mut ModelSessionState,
snapshots: &watch::Sender<Option<ModelCheckpoint>>,
) {
session.conversation.context.commit_tail();
snapshots.send_replace(Some(ModelCheckpoint {
workspace: session.workspace.clone(),
conversation: session.conversation.clone(),
preserve_inherited_delta: true,
}));
}
async fn drive_session(
&mut self,
session: &mut ModelSessionState,
mut steers: tokio::sync::mpsc::Receiver<Prompt>,
fork_snapshots: &watch::Sender<Option<ModelCheckpoint>>,
) -> Result<String> {
let mut can_drain_steers = false;
loop {
if can_drain_steers {
self.drain_steers(&mut session.conversation, &mut steers)
.await?;
}
Self::publish_fork_snapshot(session, fork_snapshots);
let call_index = self.stats.model_calls + 1;
let response = self
.perform_model_call(call_index, &session.conversation, &session.factory)
.await?;
session
.conversation
.update_token_info(response.usage.as_ref());
session.conversation.previous_response_id = Some(response.id.clone());
let end_turn = response.end_turn;
let final_message = response.final_message;
let code_calls = response.code_calls;
session
.conversation
.append(response.output_items.into_iter().map(strip_item_id));
can_drain_steers = true;
if code_calls.is_empty() {
if end_turn == Some(false) {
session.conversation.clear_delta();
self.maybe_compact(
call_index,
&mut session.conversation,
&session.factory,
&session.workspace,
session.tools.default_shell_name(),
)
.await?;
continue;
}
if !steers.is_empty() {
session.conversation.clear_delta();
continue;
}
if let Some(message) = final_message {
return Ok(if message.trim().is_empty() {
"The model completed without emitting assistant text.".to_owned()
} else {
message
});
}
return Err(NanocodexError::MalformedResponse {
detail: "model completed without a final message or exec call",
});
}
session.conversation.clear_delta();
for call in code_calls {
let history = (call.name == "exec")
.then(|| Arc::new(session.conversation.flattened_history()));
let output = self
.execute_model_tool(&session.tools, call_index, call, history)
.await?;
session.conversation.append(output);
}
self.maybe_compact(
call_index,
&mut session.conversation,
&session.factory,
&session.workspace,
session.tools.default_shell_name(),
)
.await?;
}
}
async fn drain_steers(
&mut self,
conversation: &mut ConversationState,
steers: &mut tokio::sync::mpsc::Receiver<Prompt>,
) -> Result<()> {
while let Ok(steer) = steers.try_recv() {
if let Ok(content) = serde_json::to_string(&steer) {
info!(
target: "nanocodex",
content_kind = "steer",
content = content.as_str(),
"turn content"
);
}
let instruction_bytes = steer.instruction.text_bytes();
let user_content = prepare_user_input(&steer.instruction).await;
conversation.append([ResponseItem::message(
nanocodex_core::MessageRole::User,
user_content,
)]);
self.stats.steers += 1;
self.events.emit(
AgentEventKind::RunSteered,
RunSteered {
steer_index: self.stats.steers,
instruction_bytes,
},
)?;
}
Ok(())
}
async fn execute_model_tool(
&mut self,
tools: &ToolRuntime,
call_index: u32,
call: CodeCall,
history: Option<Arc<Vec<ResponseItem>>>,
) -> Result<Vec<ResponseItem>> {
self.tool_call_indices
.insert(call.call_id.clone().into_boxed_str(), call_index);
let arguments = if call.name == "exec" {
ToolCallArguments::Text(&call.input)
} else {
serde_json::from_str::<&RawValue>(&call.input)
.map_or(ToolCallArguments::Text(&call.input), ToolCallArguments::Raw)
};
self.events.emit(
AgentEventKind::ToolCall,
ToolCallEvent {
call_id: &call.call_id,
tool: &call.name,
arguments,
model_call_index: call_index,
},
)?;
self.stats.tool_calls += 1;
if let Some(message) = unsupported_tool_message(&call) {
let output = ToolOutputBody::Text(message);
self.events.emit(
AgentEventKind::ToolResult,
ToolResultEvent {
call_id: &call.call_id,
tool: &call.name,
status: "failed",
duration_ns: 0,
result: &output,
metadata: None,
},
)?;
return Ok(vec![match call.kind {
CodeCallKind::Custom => custom_tool_output(call.call_id, output),
CodeCallKind::Function => function_tool_output(call.call_id, output),
}]);
}
let owned_context = owned_code_context(&call, history, self.events.request_id())?;
let started_at = self.track_active_tool_call(&call);
let tool_span = model_tool_span(&call, call_index);
record_span_content(&tool_span, "tool.arguments", &call.input);
let mut execution = if let Some(context) = owned_context {
tools
.execute_code_owned(&call.input, context)
.instrument(tool_span.clone())
.await
} else {
let context = ToolContext {
model: MODEL,
session_id: self.events.request_id(),
call_id: &call.call_id,
history: &[],
output_token_budget: nanocodex_tools::DEFAULT_TOOL_OUTPUT_TOKENS,
};
tools
.wait_for_code(&call.input, context)
.instrument(tool_span.clone())
.await
};
prepare_output_images(&mut execution.output).await;
if let Ok(content) = serde_json::to_string(&execution.output) {
record_span_content(&tool_span, "tool.output", &content);
}
self.active_tool_call = None;
let duration_ns = elapsed_ns(started_at);
tool_span.record("status", status(execution.success));
tool_span.record("otel.status_code", otel_status(execution.success));
tool_span.record("duration_ns", duration_ns);
self.stats.tool_wall_duration_ns += duration_ns;
for nested in &execution.nested_calls {
self.emit_nested_tool(call_index, &call.call_id, nested)?;
}
self.events.emit(
AgentEventKind::ToolResult,
ToolResultEvent {
call_id: &call.call_id,
tool: &call.name,
status: status(execution.success),
duration_ns,
result: &execution.output,
metadata: None,
},
)?;
let output = match call.kind {
CodeCallKind::Custom => custom_tool_output(call.call_id.clone(), execution.output),
CodeCallKind::Function => function_tool_output(call.call_id.clone(), execution.output),
};
let mut outputs = Vec::with_capacity(execution.notifications.len() + 1);
outputs.push(output);
outputs.extend(
execution.notifications.into_iter().map(|notification| {
custom_tool_notification(notification.call_id, notification.text)
}),
);
Ok(outputs)
}
fn track_active_tool_call(&mut self, call: &CodeCall) -> Instant {
let started_at = Instant::now();
self.active_tool_call = Some(ActiveToolCall {
call_id: call.call_id.clone(),
name: call.name.clone(),
kind: call.kind,
started_at,
});
started_at
}
async fn maybe_compact(
&mut self,
after_model_call_index: u32,
conversation: &mut ConversationState,
factory: &ResponsesAttemptFactory,
workspace: &str,
shell: &str,
) -> Result<()> {
let Some(auto_compact_token_limit) = compaction::auto_compact_token_limit(MODEL) else {
return Ok(());
};
let active_context_tokens =
conversation.active_context_tokens(self.server_reasoning_included);
if active_context_tokens < auto_compact_token_limit {
return Ok(());
}
let previous_response_id = conversation.previous_response_id.as_deref().ok_or(
NanocodexError::MalformedResponse {
detail: "compaction did not have a previous response ID",
},
)?;
let (item, _usage) = self
.perform_compaction(
after_model_call_index,
conversation.prompt_history(),
conversation.delta_start,
previous_response_id,
active_context_tokens,
auto_compact_token_limit,
factory,
)
.await?;
let project_instructions = load_project_instructions(Path::new(workspace))?;
let canonical_context = task_context(workspace, shell, project_instructions.as_deref());
conversation.install_compaction(item, canonical_context, factory.profile().prefix());
Ok(())
}
fn emit_nested_tool(
&mut self,
call_index: u32,
parent_call_id: &str,
call: &NestedToolCall,
) -> Result<()> {
let embedded_parent = call.call_id.rsplit_once("/code-").map(|(parent, _)| parent);
let original_parent = embedded_parent.unwrap_or(parent_call_id);
let call_id = embedded_parent.map_or_else(
|| format!("{parent_call_id}/{}", call.call_id),
|_| call.call_id.clone(),
);
let call_index = self
.tool_call_indices
.get(original_parent)
.copied()
.unwrap_or(call_index);
self.events.emit(
AgentEventKind::ToolCall,
ToolCallEvent {
call_id: &call_id,
tool: &call.name,
arguments: &call.input,
model_call_index: call_index,
},
)?;
self.events.emit(
AgentEventKind::ToolResult,
ToolResultEvent {
call_id: &call_id,
tool: &call.name,
status: status(call.success),
duration_ns: call.duration_ns,
result: &call.output,
metadata: call.metadata.as_deref(),
},
)?;
self.stats.tool_calls += 1;
self.stats.tool_work_duration_ns += call.duration_ns;
Ok(())
}
async fn perform_warmup(&mut self, factory: &ResponsesAttemptFactory) -> Result<String> {
let started_at = Instant::now();
self.events.emit(
AgentEventKind::ModelWarmupStarted,
WarmupStarted {
model: MODEL,
prompt_cache_key: factory.profile().prompt_cache_key(),
},
)?;
let span = info_span!(
target: "nanocodex",
"model.warmup",
otel.kind = "internal",
otel.status_code = tracing::field::Empty,
model = MODEL,
system_prompt.bytes = self.config.system_prompt().len(),
status = tracing::field::Empty,
duration_ns = tracing::field::Empty,
);
if let Ok(content) = serde_json::to_string(factory.profile().prefix()) {
record_span_content(&span, "model.input", &content);
}
let success = match self
.client
.execute(factory.warmup())
.instrument(span.clone())
.await
{
Ok(success) => success,
Err(error) => {
span.record("status", "failed");
span.record("otel.status_code", "ERROR");
span.record("duration_ns", elapsed_ns(started_at));
return self.warmup_failed(started_at, error.into());
}
};
let attempt = success.attempt();
let connection_generation = success.connection_generation();
self.server_reasoning_included |= success.server_reasoning_included();
let ResponsesOutput::Warmup(response) = success.into_output() else {
span.record("status", "failed");
span.record("otel.status_code", "ERROR");
return Err(NanocodexError::InvalidAttemptState {
detail: "warmup returned a non-warmup response",
});
};
let duration_ns = elapsed_ns(started_at);
span.record("status", "completed");
span.record("otel.status_code", "OK");
span.record("duration_ns", duration_ns);
self.stats.warmup_duration_ns += duration_ns;
if let Some(usage) = &response.usage {
self.stats.warmup_usage.add(usage);
}
self.stats.last_response_id = Some(response.id.clone());
self.events.emit(
AgentEventKind::ModelWarmupCompleted,
WarmupCompleted {
response_id: &response.id,
attempt,
connection_generation,
duration_ns,
usage: response.usage.as_ref(),
},
)?;
Ok(response.id)
}
fn warmup_failed<T>(&mut self, started_at: Instant, error: NanocodexError) -> Result<T> {
let duration_ns = elapsed_ns(started_at);
self.stats.warmup_duration_ns += duration_ns;
let message = error.to_string();
self.events.emit(
AgentEventKind::ModelWarmupFailed,
WarmupFailed {
duration_ns,
error: &message,
},
)?;
Err(error)
}
async fn perform_model_call(
&mut self,
call_index: u32,
conversation: &ConversationState,
factory: &ResponsesAttemptFactory,
) -> Result<TurnResult> {
let previous_response_id = conversation.previous_response_id.as_deref();
let started_at = Instant::now();
self.stats.model_calls += 1;
self.events.emit(
AgentEventKind::ModelCallStarted,
ModelCallStarted {
call_index,
model: MODEL,
effort: self.config.thinking.as_str(),
previous_response_id,
},
)?;
let request = factory.generation(
call_index,
conversation.prompt_history(),
conversation.shared_history(),
conversation.delta_start,
previous_response_id,
);
let (input_item_count, input_bytes, input_content) = trace_model_input(&request);
let span = model_call_span(
call_index,
previous_response_id.is_some(),
input_item_count,
input_bytes,
);
if let Some(input_content) = &input_content {
record_span_content(&span, "model.input", input_content);
}
let success = match self.client.execute(request).instrument(span.clone()).await {
Ok(success) => success,
Err(error) => {
span.record("status", "failed");
span.record("otel.status_code", "ERROR");
span.record("duration_ns", elapsed_ns(started_at));
return self.model_call_failed(call_index, started_at, error.into());
}
};
let attempt = success.attempt();
let connection_generation = success.connection_generation();
self.server_reasoning_included |= success.server_reasoning_included();
let ResponsesOutput::Generation(response) = success.into_output() else {
span.record("status", "failed");
span.record("otel.status_code", "ERROR");
return Err(NanocodexError::InvalidAttemptState {
detail: "generation returned a non-generation response",
});
};
let duration_ns = elapsed_ns(started_at);
record_model_response(&span, &response);
span.record("status", "completed");
span.record("otel.status_code", "OK");
span.record("duration_ns", duration_ns);
if let Some(usage) = &response.usage {
span.record("input_tokens", usage.input_tokens);
span.record(
"cached_input_tokens",
usage
.input_tokens_details
.as_ref()
.map_or(0, |details| details.cached_tokens),
);
span.record("output_tokens", usage.output_tokens);
}
self.stats.model_duration_ns += duration_ns;
if let Some(usage) = &response.usage {
self.stats.usage.add(usage);
}
self.stats.last_response_id = Some(response.id.clone());
self.events.emit(
AgentEventKind::ModelCallCompleted,
ModelCallCompleted {
call_index,
model: MODEL,
response_id: &response.id,
attempt,
connection_generation,
status: &response.status,
duration_ns,
time_to_first_event_ns: response.time_to_first_event_ns,
time_to_first_output_ns: response.time_to_first_output_ns,
tool_calls: response.code_calls.len(),
usage: response.usage.as_ref(),
},
)?;
Ok(response)
}
#[allow(clippy::too_many_arguments)]
async fn perform_compaction(
&mut self,
after_model_call_index: u32,
history: nanocodex_core::responses::ResponseHistory,
incremental_start: usize,
previous_response_id: &str,
active_context_tokens: u64,
auto_compact_token_limit: u64,
factory: &ResponsesAttemptFactory,
) -> Result<(ResponseItem, Option<Usage>)> {
let trigger = compaction::trigger();
let mut history: Vec<_> = history.iter().cloned().collect();
compaction::trim_tool_outputs_to_fit_context_window(&mut history, active_context_tokens);
let history = nanocodex_core::responses::ResponseHistory::new(history);
let started_at = Instant::now();
self.stats.compactions += 1;
self.events.emit(
AgentEventKind::ModelCompactionStarted,
CompactionStarted {
after_model_call_index,
active_context_tokens,
auto_compact_token_limit,
previous_response_id,
},
)?;
let request = factory.compaction(
after_model_call_index,
history.clone(),
history,
incremental_start,
previous_response_id,
trigger,
);
let (input_item_count, input_bytes, input_content) = trace_model_input(&request);
let span = info_span!(
target: "nanocodex",
"model.compaction",
otel.kind = "internal",
otel.status_code = tracing::field::Empty,
after_model_call_index,
model.input.item_count = input_item_count,
model.input.bytes = input_bytes,
model.response.id = tracing::field::Empty,
status = tracing::field::Empty,
duration_ns = tracing::field::Empty,
);
if let Some(input_content) = &input_content {
record_span_content(&span, "model.input", input_content);
}
let success = match self.client.execute(request).instrument(span.clone()).await {
Ok(success) => success,
Err(error) => {
span.record("status", "failed");
span.record("otel.status_code", "ERROR");
span.record("duration_ns", elapsed_ns(started_at));
return self.compaction_failed(after_model_call_index, started_at, error.into());
}
};
let attempt = success.attempt();
let connection_generation = success.connection_generation();
self.server_reasoning_included |= success.server_reasoning_included();
let ResponsesOutput::Compaction(response) = success.into_output() else {
span.record("status", "failed");
span.record("otel.status_code", "ERROR");
return Err(NanocodexError::InvalidAttemptState {
detail: "compaction returned a non-compaction response",
});
};
let duration_ns = elapsed_ns(started_at);
span.record("model.response.id", response.id.as_str());
if let Ok(content) = serde_json::to_string(&response.item) {
record_span_content(&span, "model.output_item", &content);
}
span.record("status", "completed");
span.record("otel.status_code", "OK");
span.record("duration_ns", duration_ns);
self.stats.model_duration_ns += duration_ns;
if let Some(usage) = &response.usage {
self.stats.usage.add(usage);
}
self.stats.last_response_id = Some(response.id.clone());
self.events.emit(
AgentEventKind::ModelCompactionCompleted,
CompactionCompleted {
after_model_call_index,
response_id: &response.id,
attempt,
connection_generation,
status: &response.status,
duration_ns,
time_to_first_event_ns: response.time_to_first_event_ns,
time_to_first_output_ns: response.time_to_first_output_ns,
usage: response.usage.as_ref(),
},
)?;
Ok((response.item, response.usage))
}
fn compaction_failed<T>(
&mut self,
after_model_call_index: u32,
started_at: Instant,
error: crate::NanocodexError,
) -> Result<T> {
let duration_ns = elapsed_ns(started_at);
self.stats.model_duration_ns += duration_ns;
let message = error.to_string();
self.events.emit(
AgentEventKind::ModelCompactionFailed,
CompactionFailed {
after_model_call_index,
duration_ns,
error: &message,
},
)?;
Err(error)
}
fn model_call_failed<T>(
&mut self,
call_index: u32,
started_at: Instant,
error: crate::NanocodexError,
) -> Result<T> {
let duration_ns = elapsed_ns(started_at);
self.stats.model_duration_ns += duration_ns;
let message = error.to_string();
self.events.emit(
AgentEventKind::ModelCallFailed,
ModelCallFailed {
call_index,
model: MODEL,
duration_ns,
error: &message,
},
)?;
Err(error)
}
}
fn unsupported_tool_message(call: &CodeCall) -> Option<String> {
if call.namespace.is_none() && matches!(call.name.as_str(), "exec" | "wait") {
return None;
}
let qualified_name = format!("{}{}", call.namespace.as_deref().unwrap_or(""), call.name);
Some(match &call.kind {
CodeCallKind::Custom => format!("unsupported custom tool call: {qualified_name}"),
CodeCallKind::Function => format!("unsupported call: {qualified_name}"),
})
}
fn trace_model_input(request: &ResponsesAttempt) -> (usize, usize, Option<String>) {
let item_count = request.input_item_count();
let items = request.input_items().collect::<Vec<_>>();
let content = serde_json::to_string(&items).ok();
let bytes = content.as_ref().map_or(0, String::len);
(item_count, bytes, content)
}
fn model_tool_span(call: &CodeCall, call_index: u32) -> tracing::Span {
info_span!(
target: "nanocodex",
"tool.call",
otel.kind = "internal",
otel.status_code = tracing::field::Empty,
tool.name = %call.name,
tool.call_id = %call.call_id,
tool.arguments.bytes = call.input.len(),
model.call_index = call_index,
status = tracing::field::Empty,
duration_ns = tracing::field::Empty,
)
}
fn owned_code_context(
call: &CodeCall,
history: Option<Arc<Vec<ResponseItem>>>,
session_id: &str,
) -> Result<Option<OwnedToolContext>> {
if call.name != "exec" {
return Ok(None);
}
let history = history.ok_or(NanocodexError::MalformedResponse {
detail: "exec call did not have an owned history snapshot",
})?;
Ok(Some(OwnedToolContext::new(
MODEL,
session_id,
&call.call_id,
history,
nanocodex_tools::DEFAULT_TOOL_OUTPUT_TOKENS,
)))
}
fn record_span_content(span: &tracing::Span, kind: &'static str, content: &str) {
span.in_scope(|| {
info!(
target: "nanocodex",
content_kind = kind,
content,
"trace content"
);
});
}
fn record_indexed_span_content(
span: &tracing::Span,
kind: &'static str,
index: usize,
content: &str,
) {
span.in_scope(|| {
info!(
target: "nanocodex",
content_kind = kind,
output.index = index,
content,
"trace content"
);
});
}
fn model_call_span(
call_index: u32,
previous_response: bool,
input_item_count: usize,
input_bytes: usize,
) -> tracing::Span {
info_span!(
target: "nanocodex",
"model.call",
otel.kind = "internal",
otel.status_code = tracing::field::Empty,
model = MODEL,
model.call_index = call_index,
previous_response,
model.input.item_count = input_item_count,
model.input.bytes = input_bytes,
model.response.id = tracing::field::Empty,
model.response.status = tracing::field::Empty,
model.response.end_turn = tracing::field::Empty,
model.output.item_count = tracing::field::Empty,
model.output.bytes = tracing::field::Empty,
model.tool_call_count = tracing::field::Empty,
assistant.output.bytes = tracing::field::Empty,
status = tracing::field::Empty,
duration_ns = tracing::field::Empty,
input_tokens = tracing::field::Empty,
cached_input_tokens = tracing::field::Empty,
output_tokens = tracing::field::Empty,
reasoning.summary_count = tracing::field::Empty,
)
}
fn record_model_response(span: &tracing::Span, response: &TurnResult) {
span.record("model.response.id", response.id.as_str());
span.record("model.response.status", response.status.as_str());
if let Some(end_turn) = response.end_turn {
span.record("model.response.end_turn", end_turn);
}
span.record("model.output.item_count", response.output_items.len());
span.record("model.tool_call_count", response.code_calls.len());
let output_content = serde_json::to_string(&response.output_items).ok();
let output_bytes = output_content.as_ref().map_or(0, String::len);
span.record("model.output.bytes", output_bytes);
let mut summary_count = 0_usize;
for (index, item) in response.output_items.iter().enumerate() {
let kind = if let ResponseItem::Reasoning { summary, .. } = item {
summary_count = summary_count.saturating_add(summary.len());
"reasoning"
} else {
"model.output_item"
};
if let Ok(content) = serde_json::to_string(item) {
record_indexed_span_content(span, kind, index, &content);
}
}
if let Some(message) = &response.final_message {
span.record("assistant.output.bytes", message.len());
}
span.record("reasoning.summary_count", summary_count);
}
fn strip_item_id(mut item: ResponseItem) -> ResponseItem {
item.strip_id();
item
}
fn request_profile(
session_id: &str,
lineage_id: &str,
tool_specs: Vec<ToolDefinition>,
system_prompt: &str,
) -> RequestProfile {
RequestProfile::new(
session_id,
lineage_id,
Arc::from([
ResponseItem::additional_tools(tool_specs),
ResponseItem::message(
nanocodex_core::MessageRole::Developer,
[nanocodex_core::ContentItem::InputText {
text: system_prompt.into(),
}],
),
]),
)
}
fn tool_runtime(workspace: &str, config: &ModelConfig, tools: &Tools) -> ToolRuntime {
ToolRuntime::new(
workspace,
tools.web_search_enabled().then(|| WebSearchConfig {
endpoint: config.search_endpoint(),
auth: config.auth.clone(),
}),
tools
.image_generation_enabled()
.then(|| ImageGenerationConfig {
api_base_url: config.api_base_url.clone(),
auth: config.auth.clone(),
save_root: Path::new(workspace).to_path_buf(),
}),
)
.with_tools(tools)
}
const fn status(success: bool) -> &'static str {
if success { "completed" } else { "failed" }
}
const fn otel_status(success: bool) -> &'static str {
if success { "OK" } else { "ERROR" }
}
#[cfg(test)]
#[path = "agent_tests.rs"]
mod tests;