use super::model_driver::run_agent_turn_with_retry;
use super::{
AfterClaimContextInput, AgentContent, AgentContentPart, AgentLoopStepExecution,
AgentLoopStepOutput, AgentMessage, AgentToolCall, AgentTurnRequest, AppId,
AppToolExecutionContext, ClaimedRuntimeInput, Context, DashboardActivityEvent,
DashboardActivityHistoryStore, DashboardActivityHistoryWindow, DashboardState, Duration,
EpisodeActionRecord, EventPayload, EventView, HistoryMessage,
MID_TURN_COMPACTION_MAX_RECOVERIES, PreTurnState, RUNTIME_EVENT_CLAIM_BATCH_SIZE,
RUNTIME_HISTORY_MIN_MESSAGES, RUNTIME_HISTORY_SUMMARY_MAX_TOKENS,
RUNTIME_PREFLIGHT_STAGE_TIMEOUT_SECS, Result, RuntimeErrorActionContext, RuntimeErrorCase,
RuntimeErrorCaseParts, RuntimeErrorKind, RuntimeErrorObservation, RuntimeErrorRuntimeContext,
RuntimeErrorTaskContext, RuntimeStatusLevel, RuntimeTurnPhase, SessionActivityEvent,
TelegramLiveDraftSession, TextActivityDescriptor, TokenEstimateBaseline, ToolCallActivityEvent,
ToolExecutionResult, activity_event_from_tool_call_activity_event,
afterclaim_context_input_for_claimed_inputs, append_runtime_error_case, apply_activity_event,
assistant_activity_cell, build_afterclaim_context_text, build_preturn_context_text,
build_runtime_request_envelope, build_runtime_tool_specs, build_tool_call_activity_event,
claim_pending_runtime_inputs, claimed_events_are_terminal,
claimed_events_require_explicit_completion, claimed_runtime_input_fingerprint,
compact_preserved_body_lines, execute_agent_tool_call, execute_pre_turn_runtime_compaction,
finalize_claimed_runtime_events, handle_model_request_failure, handle_runtime_overflow,
is_context_budget_exceeded, json, maybe_compact_runtime_messages, maybe_record_skill_read,
maybe_start_telegram_live_draft_session, miette, record_runtime_history_messages,
record_skill_run_evidence, render_activity_from_messages, render_telegram_tool_result_status,
runtime_request_budget_limits, runtime_work_origin, set_runtime_status,
set_runtime_status_only, summarize_action_from_tool_call, thinking_activity_cell,
user_activity_cell_from_event,
};
use crate::memory::PlanCompactionInput;
use crate::reasoning::prompt_parts::compact_horizontal_whitespace;
use std::path::{Path, PathBuf};
use tracing::warn;
const EMPTY_REASONING_MAX_LOOPS: usize = 2;
const MAIN_EXPLICIT_COMPLETION_MESSAGE: &str = "The claimed event remains unresolved. Do not end by only outputting text; keep calling tools, and explicitly call `finish_and_send` with a non-empty `reply_message` when the final result is ready.";
fn enter_runtime_phase(
context: &mut Context,
tx: Option<&tokio::sync::watch::Sender<DashboardState>>,
phase: RuntimeTurnPhase,
) {
context.set_runtime_phase(Some(phase));
if let Some(tx) = tx {
tx.send_modify(|state| {
state.status_output =
crate::dashboard::render::render_status_command_output_for_dashboard(context, &[]);
state.status_command =
crate::dashboard::render::status_command_snapshot_for_dashboard(context);
});
}
set_runtime_status_only(tx, format!("processing: runtime turn / {}", phase.label()));
}
pub(super) fn clear_runtime_failures_after_model_success(
context: &Context,
fingerprint: Option<&str>,
) {
let Some(fingerprint) = fingerprint else {
return;
};
context.clear_runtime_overflow_failure(fingerprint);
context.clear_model_request_failure(fingerprint);
}
pub(super) fn clear_runtime_overflow_failure_after_compaction(
context: &Context,
fingerprint: Option<&str>,
) {
if let Some(fingerprint) = fingerprint {
context.clear_runtime_overflow_failure(fingerprint);
}
}
struct RuntimeTurnAbort<'a> {
live_draft_session: Option<TelegramLiveDraftSession>,
claimed_event_ids: &'a [String],
observation: String,
description: String,
}
async fn abort_runtime_turn_before_model(
context: &mut Context,
abort: RuntimeTurnAbort<'_>,
) -> AgentLoopStepExecution {
let RuntimeTurnAbort {
live_draft_session,
claimed_event_ids,
observation,
description,
} = abort;
context.set_runtime_phase(None);
if let Some(session) = live_draft_session {
session.shutdown(context).await;
} else {
context.install_live_progress(None);
}
let output = AgentLoopStepOutput {
observation: observation.clone(),
description,
current_doing: "waiting for next tool decision".to_string(),
actions: vec![EpisodeActionRecord {
kind: "runtime_preflight_failed".to_string(),
summary: observation,
}],
};
finalize_claimed_runtime_events(context, claimed_event_ids, &output);
context.claimed_event_ids.clear();
context.current_work_origin = None;
AgentLoopStepExecution
}
fn maybe_build_afterclaim_context_message(
context: &mut Context,
input: &AfterClaimContextInput,
fingerprint: Option<&str>,
force_reinject: bool,
) -> Option<HistoryMessage> {
let fingerprint = fingerprint.unwrap_or("unfingerprinted");
let message = preview_afterclaim_context_message(context, input, fingerprint, force_reinject)?;
context.afterclaim_context_fingerprint = Some(fingerprint.to_string());
Some(message)
}
fn preview_afterclaim_context_message(
context: &Context,
input: &AfterClaimContextInput,
fingerprint: &str,
force_reinject: bool,
) -> Option<HistoryMessage> {
if input.is_empty() {
return None;
}
let already_injected = context.afterclaim_context_fingerprint.as_deref() == Some(fingerprint);
if already_injected && !force_reinject {
return None;
}
let text = build_afterclaim_context_text(context, input);
if text.trim().is_empty() {
return None;
}
Some(HistoryMessage {
message: AgentMessage::user_content(afterclaim_agent_content(text, input)),
activity_event: None,
tool_call_activity_events: Vec::new(),
})
}
fn afterclaim_agent_content(text: String, input: &AfterClaimContextInput) -> AgentContent {
let parts = input
.iter()
.flat_map(|event| match &event.payload {
EventPayload::TelegramIncoming(payload) => payload
.attachments
.iter()
.map(|attachment| match attachment.kind {
crate::events::TelegramIncomingAttachmentKind::Image => {
AgentContentPart::Image {
path: attachment.local_path.clone(),
media_type: attachment.media_type.clone(),
description: attachment.description.clone(),
}
}
})
.collect::<Vec<_>>(),
EventPayload::TerminalIncoming(payload) => payload
.attachments
.iter()
.map(|attachment| match attachment.kind {
crate::events::TerminalIncomingAttachmentKind::Image => {
AgentContentPart::Image {
path: attachment.local_path.clone(),
media_type: attachment.media_type.clone(),
description: attachment.description.clone(),
}
}
})
.collect::<Vec<_>>(),
})
.collect::<Vec<_>>();
if parts.is_empty() {
AgentContent::text(text)
} else {
AgentContent::multimodal(text, parts)
}
}
fn history_has_complete_afterclaim_context(messages: &[HistoryMessage]) -> bool {
messages
.iter()
.filter_map(HistoryMessage::text_content)
.any(is_complete_afterclaim_context_text)
}
fn is_complete_afterclaim_context_text(text: &str) -> bool {
let text = text.trim_start();
text.starts_with("<afterclaim_context") && text.contains("</afterclaim_context>")
}
fn runtime_context_compacted_output(reason: impl Into<String>) -> AgentLoopStepOutput {
let reason = reason.into();
AgentLoopStepOutput {
observation: reason.clone(),
description: "Runtime context was compacted; the current turn ends so claimed work can be re-claimed with freshly injected context."
.to_string(),
current_doing: "waiting for next tool decision".to_string(),
actions: vec![EpisodeActionRecord {
kind: "runtime_context_compacted".to_string(),
summary: reason,
}],
}
}
fn output_is_runtime_context_compaction_boundary(output: &AgentLoopStepOutput) -> bool {
output
.actions
.iter()
.any(|action| action.kind == "runtime_context_compacted")
}
fn should_prepare_coding_project_for_claimed_input(
previous_fingerprint: Option<&str>,
claimed_fingerprint: Option<&str>,
) -> bool {
claimed_fingerprint.is_some_and(|fingerprint| previous_fingerprint != Some(fingerprint))
}
async fn prepare_coding_project_session(context: &mut Context) -> Result<()> {
let Some(project_dir) = context.coding_project_dir.clone() else {
return Ok(());
};
let app_context = AppToolExecutionContext {
execution_cwd: context.execution_cwd.clone(),
sandbox_policy: context.sandbox_policy.clone(),
dashboard_tx: context.dashboard_tx.clone(),
tool_output_max_tokens: context
.config
.main_model_config()
.tool_output_max_tokens
.max(1),
turn_epoch: context.runtime_turn_epoch,
};
prepare_coding_project_app(&mut context.apps, &project_dir, &app_context).await
}
async fn prepare_coding_project_app(
apps: &mut crate::app::AppManager,
project_dir: &Path,
app_context: &AppToolExecutionContext,
) -> Result<()> {
let coding_id = AppId::coding();
if coding_project_root_is_open(apps, project_dir) {
return Ok(());
}
let call = AgentToolCall {
id: format!("auto-open-coding-project-{}", uuid::Uuid::new_v4()),
name: "open_project".to_string(),
arguments: json!({
"project_root": project_dir.display().to_string(),
}),
};
apps.execute_tool_for_app(&coding_id, &call, app_context)
.await?;
Ok(())
}
fn coding_project_root_is_open(apps: &crate::app::AppManager, project_dir: &Path) -> bool {
apps.state_render_for(&AppId::coding())
.and_then(|render| coding_project_root_from_lines(&render.lines))
.is_some_and(|current| current == project_dir)
}
fn coding_project_root_from_lines(lines: &[String]) -> Option<PathBuf> {
lines.iter().find_map(|line| {
let value = line.strip_prefix("project_root=")?.trim();
(!value.is_empty() && value != "none").then(|| PathBuf::from(value))
})
}
pub async fn execute_agent_loop_step(
context: &mut Context,
tx: Option<&tokio::sync::watch::Sender<DashboardState>>,
) -> AgentLoopStepExecution {
let runtime_turn_id = format!("runtime-turn-{}", uuid::Uuid::new_v4());
let claimed_inputs = claim_pending_runtime_inputs(context, RUNTIME_EVENT_CLAIM_BATCH_SIZE);
context.current_work_origin = runtime_work_origin(&claimed_inputs);
let claimed_input_fingerprint = claimed_runtime_input_fingerprint(&claimed_inputs);
let claimed_event_ids = claimed_inputs
.iter()
.map(|input| input.event_id.to_string())
.collect::<Vec<_>>();
context.claimed_event_ids.clone_from(&claimed_event_ids);
append_claimed_input_activity_cells(context, tx, &claimed_inputs);
let preflight_timeout = Duration::from_secs(RUNTIME_PREFLIGHT_STAGE_TIMEOUT_SECS);
let afterclaim_context_input = afterclaim_context_input_for_claimed_inputs(&claimed_inputs);
let claimed_event_views = claimed_inputs.clone();
let live_draft_session = maybe_start_telegram_live_draft_session(context, &claimed_event_views);
enter_runtime_phase(context, tx, RuntimeTurnPhase::PreflightPreTurnContext);
let should_prepare_coding_project = should_prepare_coding_project_for_claimed_input(
context.afterclaim_context_fingerprint.as_deref(),
claimed_input_fingerprint.as_deref(),
);
if should_prepare_coding_project && let Err(err) = prepare_coding_project_session(context).await
{
set_runtime_status(
tx,
RuntimeStatusLevel::Error,
"runtime turn preflight failed: coding project setup".to_string(),
);
return abort_runtime_turn_before_model(
context,
RuntimeTurnAbort {
live_draft_session,
claimed_event_ids: &claimed_event_ids,
observation: format!("runtime preflight failed: auto coding project setup: {err}"),
description: "Failed to prepare the Coding app for the project session."
.to_string(),
},
)
.await;
}
let preturn_started_at = std::time::Instant::now();
tracing::debug!(
"runtime preflight stage started: {}",
RuntimeTurnPhase::PreflightPreTurnContext.label()
);
let preturn_state = if let Ok(preturn_state) =
tokio::time::timeout(preflight_timeout, async { PreTurnState::new(context) }).await
{
tracing::debug!(
elapsed_ms = preturn_started_at.elapsed().as_millis(),
"runtime preflight stage completed: {}",
RuntimeTurnPhase::PreflightPreTurnContext.label()
);
preturn_state
} else {
let err = miette!(
"runtime preflight stage `{}` timed out after {}s",
RuntimeTurnPhase::PreflightPreTurnContext.label(),
preflight_timeout.as_secs()
);
set_runtime_status(
tx,
RuntimeStatusLevel::Error,
format!(
"runtime turn preflight timeout: {}",
RuntimeTurnPhase::PreflightPreTurnContext.label()
),
);
tracing::error!(
elapsed_ms = preturn_started_at.elapsed().as_millis(),
timeout_secs = preflight_timeout.as_secs(),
"runtime preflight stage timed out: {}",
RuntimeTurnPhase::PreflightPreTurnContext.label()
);
return abort_runtime_turn_before_model(
context,
RuntimeTurnAbort {
live_draft_session,
claimed_event_ids: &claimed_event_ids,
observation: format!("runtime preflight failed: {err}"),
description: "Failed to build preturn context.".to_string(),
},
)
.await;
};
let preturn_context_text = build_preturn_context_text(context, &preturn_state);
let runtime_context_text = if afterclaim_context_input.is_empty() {
preturn_context_text.clone()
} else {
format!(
"{}\n\n{}",
build_afterclaim_context_text(context, &afterclaim_context_input),
preturn_context_text
)
};
if let Some(tx) = tx {
tx.send_modify(|state| {
state
.preturn_context_output
.clone_from(&preturn_context_text);
});
}
let request_envelope = build_runtime_request_envelope(context);
let initial_tools = build_runtime_tool_specs(context);
let request_budget_limits = runtime_request_budget_limits(context);
let runtime_conversation_budget =
request_envelope.conversation_budget_tokens(&initial_tools, request_budget_limits);
let mut initial_injected_context_messages = Vec::new();
if let Some(message) = preview_afterclaim_context_message(
context,
&afterclaim_context_input,
claimed_input_fingerprint
.as_deref()
.unwrap_or("unfingerprinted"),
false,
) {
initial_injected_context_messages.push(message);
}
if !preturn_context_text.trim().is_empty() {
initial_injected_context_messages.push(HistoryMessage::user(preturn_context_text.clone()));
}
let runtime_conversation_summary_budget =
RUNTIME_HISTORY_SUMMARY_MAX_TOKENS.min(runtime_conversation_budget);
let pre_turn_compacted = if let Some(plan) = context
.memory
.plan_runtime_conversation_compaction_for_request(PlanCompactionInput {
envelope: &request_envelope,
injected_messages: &initial_injected_context_messages,
tools: &initial_tools,
limits: request_budget_limits,
baseline: &context.token_estimate_baseline,
min_messages: RUNTIME_HISTORY_MIN_MESSAGES,
summary_max_tokens: runtime_conversation_summary_budget,
}) {
const COMPACTION_BACKOFF_TIMEOUTS_SECS: [u64; 3] = [180, 360, 600];
enter_runtime_phase(context, tx, RuntimeTurnPhase::PreflightCompaction);
let compaction_started_at = std::time::Instant::now();
tracing::debug!(
"runtime preflight stage started: {}",
RuntimeTurnPhase::PreflightCompaction.label()
);
let mut outcome = None;
let mut failure = None;
for (attempt, timeout_secs) in COMPACTION_BACKOFF_TIMEOUTS_SECS.iter().enumerate() {
match tokio::time::timeout(
Duration::from_secs(*timeout_secs),
execute_pre_turn_runtime_compaction(context, &plan),
)
.await
{
Ok(Ok(value)) => {
outcome = Some(value);
break;
}
Ok(Err(err)) => {
failure = Some(format!(
"main model compaction attempt {} failed: {err}",
attempt + 1
));
}
Err(_) => {
failure = Some(format!(
"main model compaction attempt {} timed out after {}s",
attempt + 1,
timeout_secs
));
}
}
if attempt + 1 < COMPACTION_BACKOFF_TIMEOUTS_SECS.len() {
tracing::warn!(
attempt = attempt + 1,
timeout_secs = *timeout_secs,
next_timeout_secs = COMPACTION_BACKOFF_TIMEOUTS_SECS[attempt + 1],
error = failure.as_deref().unwrap_or("unknown compaction failure"),
"runtime preflight compaction failed, retrying with longer timeout"
);
}
}
let Some(outcome) = outcome else {
let err = miette!(
"runtime preflight stage `{}` failed after main-model compaction retries: {}",
RuntimeTurnPhase::PreflightCompaction.label(),
failure.unwrap_or_else(|| "unknown compaction failure".to_string())
);
set_runtime_status(
tx,
RuntimeStatusLevel::Error,
format!(
"runtime turn preflight failed: {}",
RuntimeTurnPhase::PreflightCompaction.label()
),
);
tracing::error!(
elapsed_ms = compaction_started_at.elapsed().as_millis(),
error = %err,
"runtime preflight compaction failed without local fallback"
);
return abort_runtime_turn_before_model(
context,
RuntimeTurnAbort {
live_draft_session,
claimed_event_ids: &claimed_event_ids,
observation: format!("runtime preflight failed: {err}"),
description: "Failed to generate a main-model runtime context compaction."
.to_string(),
},
)
.await;
};
tracing::debug!(
elapsed_ms = compaction_started_at.elapsed().as_millis(),
"runtime preflight stage completed: {}",
RuntimeTurnPhase::PreflightCompaction.label()
);
let applied = context
.memory
.apply_runtime_conversation_compaction(plan, outcome)
.await;
if applied {
clear_runtime_overflow_failure_after_compaction(
context,
claimed_input_fingerprint.as_deref(),
);
}
context.delivered_root_instruction_fingerprint = None;
context.visible_source_lines.clear();
true
} else {
false
};
let mut conversation_slice = context.memory.runtime_conversation_slice(
runtime_conversation_budget,
RUNTIME_HISTORY_MIN_MESSAGES,
runtime_conversation_summary_budget,
);
let mut injected_context_messages = Vec::new();
if let Some(message) = maybe_build_afterclaim_context_message(
context,
&afterclaim_context_input,
claimed_input_fingerprint.as_deref(),
pre_turn_compacted && !history_has_complete_afterclaim_context(&conversation_slice),
) {
injected_context_messages.push(message);
}
if !preturn_context_text.trim().is_empty() {
injected_context_messages.push(HistoryMessage::user(preturn_context_text.clone()));
}
conversation_slice.extend(injected_context_messages.iter().cloned());
let mut runtime_step = context
.memory
.begin_runtime_step_from_parts(request_envelope, conversation_slice);
for message in &injected_context_messages {
let committed_cells = render_activity_from_messages(vec![message.clone()]);
runtime_step.push_history_message(message.clone());
append_committed_activity_cells(context, tx, committed_cells);
}
if !injected_context_messages.is_empty() {
context.memory.runtime_conversation_mut().append_turn(
String::new(),
injected_context_messages,
Vec::new(),
);
}
let mut tool_results = Vec::new();
let mut actions = Vec::new();
let mut budget_recoveries = 0usize;
let mut consecutive_empty_reasoning_loops = 0usize;
let output = 'agent_loop: loop {
let tools = build_runtime_tool_specs(context);
match maybe_compact_runtime_messages(context, &mut runtime_step, &tools, false).await {
Ok(true) => {
clear_runtime_overflow_failure_after_compaction(
context,
claimed_input_fingerprint.as_deref(),
);
set_runtime_status_only(tx, "Compacting runtime context");
context.delivered_root_instruction_fingerprint = None;
context.visible_source_lines.clear();
break 'agent_loop runtime_context_compacted_output(
"runtime context compacted before model request; starting a new turn",
);
}
Ok(false) => {}
Err(err) => {
let observation = format!("runtime context compaction failed: {err}");
set_runtime_status(tx, RuntimeStatusLevel::Error, observation.clone());
tracing::error!("{observation}");
break 'agent_loop AgentLoopStepOutput {
observation: observation.clone(),
description: "Failed to generate a main-model runtime context compaction."
.to_string(),
current_doing: "waiting for next tool decision".to_string(),
actions: vec![EpisodeActionRecord {
kind: "runtime_context_compaction_failed".to_string(),
summary: observation,
}],
};
}
}
let request = AgentTurnRequest {
messages: runtime_step.clone_agent_messages(),
tools: tools.clone(),
};
enter_runtime_phase(context, tx, RuntimeTurnPhase::ModelRequest);
context.emit_live_generation_started();
let response = match run_agent_turn_with_retry(context, request, tx).await {
Ok(response) => {
clear_runtime_failures_after_model_success(
context,
claimed_input_fingerprint.as_deref(),
);
response
}
Err(err) => {
let is_overflow = is_context_budget_exceeded(&err);
if is_overflow && budget_recoveries < MID_TURN_COMPACTION_MAX_RECOVERIES {
match maybe_compact_runtime_messages(context, &mut runtime_step, &tools, true)
.await
{
Ok(true) => {
clear_runtime_overflow_failure_after_compaction(
context,
claimed_input_fingerprint.as_deref(),
);
budget_recoveries += 1;
set_runtime_status(
tx,
RuntimeStatusLevel::Warn,
format!(
"Recovering from context overflow ({budget_recoveries}/{MID_TURN_COMPACTION_MAX_RECOVERIES})"
),
);
break 'agent_loop runtime_context_compacted_output(format!(
"runtime context compacted after context overflow recovery ({budget_recoveries}/{MID_TURN_COMPACTION_MAX_RECOVERIES}); starting a new turn"
));
}
Ok(false) => {}
Err(compaction_err) => {
let observation = format!(
"runtime context compaction failed after context overflow: {compaction_err}"
);
set_runtime_status(tx, RuntimeStatusLevel::Error, observation.clone());
tracing::error!("{observation}");
break 'agent_loop AgentLoopStepOutput {
observation: observation.clone(),
description:
"Failed to generate a main-model runtime context compaction after context overflow."
.to_string(),
current_doing: "waiting for next tool decision".to_string(),
actions: vec![EpisodeActionRecord {
kind: "runtime_context_compaction_failed".to_string(),
summary: observation,
}],
};
}
}
}
let overflow_fuse_tripped = if is_overflow {
handle_runtime_overflow(
context,
claimed_input_fingerprint.as_deref(),
&claimed_event_ids,
&err.to_string(),
)
} else {
false
};
if is_overflow {
record_runtime_error_case(
context,
RuntimeErrorRecordInput {
turn_id: &runtime_turn_id,
claimed_inputs: &claimed_inputs,
claimed_event_ids: &claimed_event_ids,
tools: &tools,
context_text: &runtime_context_text,
error_kind: RuntimeErrorKind::ContextOverflowAfterRecovery,
severity: 3,
detected_by: "runtime_model_request",
expected_behavior: "Runtime context compaction should recover enough budget for the model request or terminate claimed inputs through the overflow fuse.",
actual_behavior: "Model request still failed with context overflow after recovery attempts.",
evidence: &err.to_string(),
recoverability: if overflow_fuse_tripped {
"terminated_by_overflow_fuse"
} else {
"requeued_for_retry"
},
retry_count: budget_recoveries,
terminal_status: Some(if overflow_fuse_tripped {
"fuse_tripped"
} else {
"not_terminal"
}),
assistant_text: None,
tool_calls: &[],
tool_results: &tool_results,
actions: &actions,
},
)
.await;
}
if !is_overflow && !overflow_fuse_tripped {
let is_permanent_model_request =
!crate::core::should_retry_agent_turn_error(&err);
let model_request_attempts = 1;
let model_request_fuse_tripped = handle_model_request_failure(
context,
claimed_input_fingerprint.as_deref(),
&claimed_event_ids,
&err.to_string(),
!is_permanent_model_request,
);
if model_request_fuse_tripped {
record_runtime_error_case(
context,
RuntimeErrorRecordInput {
turn_id: &runtime_turn_id,
claimed_inputs: &claimed_inputs,
claimed_event_ids: &claimed_event_ids,
tools: &tools,
context_text: &runtime_context_text,
error_kind: RuntimeErrorKind::ModelRequestRepeatedFailure,
severity: 3,
detected_by: "runtime_model_request",
expected_behavior: "Model request should succeed or recover with adaptive retry. Repeated non-overflow failures should trip a fuse and terminate the claimed inputs instead of requeueing indefinitely.",
actual_behavior: &format!("Model request failed repeatedly: {err}"),
evidence: &err.to_string(),
recoverability: if is_permanent_model_request {
"terminated_by_non_retryable_model_request"
} else {
"terminated_by_model_request_fuse"
},
retry_count: if is_permanent_model_request {
0
} else {
model_request_attempts
},
terminal_status: Some(if is_permanent_model_request {
"non_retryable"
} else {
"fuse_tripped"
}),
assistant_text: None,
tool_calls: &[],
tool_results: &tool_results,
actions: &actions,
},
)
.await;
}
}
let observation = format!("agent turn failed: {err}");
let terminal_action = EpisodeActionRecord {
kind: "agent_turn_failed".to_string(),
summary: observation.clone(),
};
let mut terminal_actions = actions.clone();
terminal_actions.push(terminal_action);
runtime_step.push_history_message(HistoryMessage::assistant(observation.clone()));
if let Some(cell) = assistant_activity_cell(&observation) {
append_committed_activity_cells(context, tx, vec![cell]);
}
break 'agent_loop AgentLoopStepOutput {
observation,
description: "Model request failed.".to_string(),
current_doing: "waiting for next tool decision".to_string(),
actions: terminal_actions,
};
}
};
let response_protocol = response.protocol();
let follow_up_message = response_protocol.follow_up_message(
claimed_events_require_explicit_completion(context, &claimed_event_ids),
MAIN_EXPLICIT_COMPLETION_MESSAGE,
);
let response_tool_calls = response_protocol.tool_calls;
let response_assistant_text = response_protocol.assistant_text;
let response_reasoning_content = response_protocol.reasoning_content;
let response_assistant_content = response_protocol.final_assistant_message;
if let Some(reasoning_content) = response_reasoning_content.as_deref() {
context.emit_live_reasoning_progress(reasoning_content);
if let Some(cell) = thinking_activity_cell(reasoning_content) {
append_committed_activity_cells(context, tx, vec![cell]);
}
}
if let Some(content) = response_assistant_content.as_deref()
&& !content.trim().is_empty()
{
context.emit_live_assistant_progress(content);
}
if !response_tool_calls.is_empty() {
let calls = response_tool_calls;
let assistant_text = response_assistant_text;
let tool_call_previews = calls
.iter()
.map(|call| {
build_tool_call_activity_event(context, call).unwrap_or_else(|_| {
ToolCallActivityEvent::error(
call.name.clone(),
vec![call.arguments.to_string()],
)
})
})
.collect::<Vec<_>>();
let tool_call_activity_events = tool_call_previews
.iter()
.cloned()
.filter_map(activity_event_from_tool_call_activity_event)
.collect::<Vec<_>>();
runtime_step.push_agent_message(
AgentMessage::assistant_tool_call_protocol_with_reasoning(
assistant_text.clone(),
response_reasoning_content.clone(),
calls.clone(),
),
);
if let Some(content) = assistant_text.clone()
&& !content.trim().is_empty()
{
runtime_step.push_history_message(HistoryMessage::assistant(content));
}
runtime_step.push_history_message(HistoryMessage {
message: AgentMessage::assistant_tool_call_protocol_with_reasoning(
None,
response_reasoning_content.clone(),
calls.clone(),
),
activity_event: None,
tool_call_activity_events,
});
let mut committed_cells = Vec::new();
if let Some(content) = assistant_text.clone()
&& let Some(cell) = assistant_activity_cell(&content)
{
committed_cells.push(cell);
}
append_committed_activity_cells(context, tx, committed_cells);
for (call, call_activity_event) in calls.iter().zip(tool_call_previews.iter()) {
let action_record =
summarize_action_from_tool_call(context, call).unwrap_or_else(|_| {
EpisodeActionRecord {
kind: "tool_call".to_string(),
summary: call.name.clone(),
}
});
actions.push(action_record);
enter_runtime_phase(context, tx, RuntimeTurnPhase::ToolExecution);
if let Some(tx) = tx {
match call_activity_event.clone() {
ToolCallActivityEvent::Exec(event) => {
tx.send_modify(|state| {
apply_activity_event(
state,
DashboardActivityEvent::ExecBegin {
key: call.id.clone(),
title: event.title,
call_lines: event.body_lines,
},
);
});
}
ToolCallActivityEvent::Terminal(event)
if matches!(
event.action,
crate::activity_event::TerminalActivityAction::Execute
| crate::activity_event::TerminalActivityAction::Continue
) =>
{
tx.send_modify(|state| {
apply_activity_event(
state,
DashboardActivityEvent::ExecBegin {
key: call.id.clone(),
title: event.title,
call_lines: event.body_lines,
},
);
});
}
ToolCallActivityEvent::Browser(event)
if !matches!(
event.action,
crate::activity_event::BrowserActivityAction::Snapshot
) =>
{
tx.send_modify(|state| {
apply_activity_event(
state,
DashboardActivityEvent::BrowserBegin {
key: call.id.clone(),
event,
},
);
});
}
ToolCallActivityEvent::Plan(event) if !event.steps.is_empty() => {
tx.send_modify(|state| {
apply_activity_event(
state,
DashboardActivityEvent::AppendCommittedCells {
cells: vec![SessionActivityEvent::PlanResult(event.into())],
},
);
});
}
_ => {}
}
}
let mut result = match execute_agent_tool_call(context, call).await {
Ok(result) => result,
Err(err) => {
let error_text = err.to_string();
record_runtime_error_case(
context,
RuntimeErrorRecordInput {
turn_id: &runtime_turn_id,
claimed_inputs: &claimed_inputs,
claimed_event_ids: &claimed_event_ids,
tools: &tools,
context_text: &runtime_context_text,
error_kind: classify_tool_runtime_error(&call.name, &error_text),
severity: 2,
detected_by: "runtime_tool_executor",
expected_behavior: "Tool calls should use available tools with valid arguments and satisfy the tool-specific runtime contract.",
actual_behavior: "The tool executor rejected the tool call.",
evidence: &error_text,
recoverability: "tool_error_returned_to_model",
retry_count: 0,
terminal_status: None,
assistant_text: assistant_text.as_deref(),
tool_calls: std::slice::from_ref(call),
tool_results: &tool_results,
actions: &actions,
},
)
.await;
let display_tool_name =
crate::app::AppId::render_exposed_tool_name(&call.name);
ToolExecutionResult::from_activity_event(
format!("{display_tool_name} failed"),
json!({
"error": error_text,
}),
Some(crate::dashboard::SessionActivityEvent::Error(
TextActivityDescriptor {
title: format!("{display_tool_name} failed"),
body_lines: compact_preserved_body_lines(&error_text, 12),
}
.into(),
)),
)
}
};
maybe_record_skill_read(context, call);
if let Some(status) = render_telegram_tool_result_status(call, &result) {
context.emit_live_telegram_status(status);
}
if let Some(tx) = tx {
tx.send_modify(|state| {
apply_activity_event(
state,
DashboardActivityEvent::ExecEnd {
key: call.id.clone(),
},
);
apply_activity_event(
state,
DashboardActivityEvent::BrowserEnd {
key: call.id.clone(),
},
);
});
}
let model_content = if result.skip_source_elision {
result.model_content()
} else {
super::coding_source_elision::elide_tool_model_content(
&mut context.visible_source_lines,
call,
&result.model_content(),
)
};
let activity_event = result.activity_event.clone();
let history_content = result.history_content_with_budget(
&call.id,
&call.name,
context
.config
.main_model_config()
.tool_output_max_tokens
.max(1),
);
let model_image_parts = std::mem::take(&mut result.model_image_parts);
runtime_step.push_agent_message(AgentMessage::tool(
call.id.clone(),
call.name.clone(),
model_content,
));
if !model_image_parts.is_empty() {
runtime_step.push_agent_message(AgentMessage::user_content(
AgentContent::multimodal(
format!(
"The `{}` tool attached image content for visual inspection.",
call.name
),
model_image_parts,
),
));
}
runtime_step.push_history_message(HistoryMessage::tool(
call.id.clone(),
call.name.clone(),
history_content,
activity_event.clone(),
));
append_committed_activity_cells(context, tx, activity_event.into_iter().collect());
tool_results.push(format!("{} => {}", call.name, result.summary));
if claimed_events_are_terminal(context, &claimed_event_ids) {
let mut terminal_actions = actions.clone();
terminal_actions.push(EpisodeActionRecord {
kind: "finished".to_string(),
summary: "claimed events reached terminal state".to_string(),
});
break 'agent_loop AgentLoopStepOutput {
observation: if tool_results.is_empty() {
"claimed events reached terminal state".to_string()
} else {
tool_results.join("\n")
},
description: "Finished: claimed events for this turn reached a terminal or handoff state."
.to_string(),
current_doing: "waiting for next tool decision".to_string(),
actions: terminal_actions,
};
}
}
continue 'agent_loop;
}
let assistant_text = response_assistant_content.unwrap_or_default();
let is_empty_reasoning = assistant_text.trim().is_empty()
&& response_reasoning_content
.as_deref()
.is_some_and(|reasoning| !reasoning.trim().is_empty());
if is_empty_reasoning {
consecutive_empty_reasoning_loops += 1;
if consecutive_empty_reasoning_loops >= EMPTY_REASONING_MAX_LOOPS {
warn!(
"agent loop returned empty content with reasoning for {count} consecutive iterations; \
breaking the loop",
count = consecutive_empty_reasoning_loops
);
let observation = "agent turn failed: model produced reasoning but no text output for multiple consecutive iterations";
record_runtime_error_case(
context,
RuntimeErrorRecordInput {
turn_id: &runtime_turn_id,
claimed_inputs: &claimed_inputs,
claimed_event_ids: &claimed_event_ids,
tools: &tools,
context_text: &runtime_context_text,
error_kind: RuntimeErrorKind::ModelEmptyReasoningOutput,
severity: 2,
detected_by: "runtime_empty_reasoning_fuse",
expected_behavior: "The model should either call a tool or return user-visible assistant text.",
actual_behavior: "The model repeatedly returned reasoning content without assistant text or tool calls.",
evidence: response_reasoning_content.as_deref().unwrap_or(""),
recoverability: "terminated_by_empty_reasoning_fuse",
retry_count: consecutive_empty_reasoning_loops,
terminal_status: Some("fuse_tripped"),
assistant_text: None,
tool_calls: &[],
tool_results: &tool_results,
actions: &actions,
},
)
.await;
let mut terminal_actions = actions.clone();
terminal_actions.push(EpisodeActionRecord {
kind: "agent_turn_failed".to_string(),
summary: observation.to_string(),
});
runtime_step.push_history_message(HistoryMessage::assistant(observation));
if let Some(cell) = assistant_activity_cell(observation) {
append_committed_activity_cells(context, tx, vec![cell]);
}
break 'agent_loop AgentLoopStepOutput {
observation: observation.to_string(),
description: "Error: model produced reasoning but no text output for multiple consecutive iterations.".to_string(),
current_doing: "idle".to_string(),
actions: terminal_actions,
};
}
continue 'agent_loop;
}
consecutive_empty_reasoning_loops = 0;
if let Some(expected_behavior) = follow_up_message {
if expected_behavior == MAIN_EXPLICIT_COMPLETION_MESSAGE {
record_runtime_error_case(
context,
RuntimeErrorRecordInput {
turn_id: &runtime_turn_id,
claimed_inputs: &claimed_inputs,
claimed_event_ids: &claimed_event_ids,
tools: &tools,
context_text: &runtime_context_text,
error_kind: RuntimeErrorKind::MissingFinishAndSend,
severity: 2,
detected_by: "runtime_follow_up_gate",
expected_behavior,
actual_behavior: "The model returned assistant text without the required completion tool.",
evidence: &assistant_text,
recoverability: "follow_up_message_inserted",
retry_count: 0,
terminal_status: None,
assistant_text: Some(&assistant_text),
tool_calls: &[],
tool_results: &tool_results,
actions: &actions,
},
)
.await;
}
if !assistant_text.trim().is_empty() {
runtime_step.push_agent_message(AgentMessage::assistant(&assistant_text));
runtime_step
.push_history_message(HistoryMessage::assistant(assistant_text.clone()));
}
runtime_step.push_agent_message(AgentMessage::user(expected_behavior));
continue 'agent_loop;
}
let current_doing = assistant_text
.lines()
.next()
.filter(|line| !line.trim().is_empty())
.unwrap_or("waiting for next tool decision")
.to_string();
let assistant_action = EpisodeActionRecord {
kind: "assistant_message".to_string(),
summary: current_doing.clone(),
};
actions.push(assistant_action);
runtime_step.set_current_doing(current_doing.clone());
runtime_step.push_history_message(HistoryMessage::assistant(assistant_text.clone()));
if let Some(cell) = assistant_activity_cell(&assistant_text) {
append_committed_activity_cells(context, tx, vec![cell]);
}
break 'agent_loop AgentLoopStepOutput {
observation: if tool_results.is_empty() {
assistant_text.clone()
} else {
tool_results.join("\n")
},
description: if tool_results.is_empty() {
"The model returned assistant text without calling a tool.".to_string()
} else {
assistant_text
},
current_doing,
actions: actions.clone(),
};
};
runtime_step.set_current_doing(output.current_doing.clone());
context.set_runtime_phase(None);
if let Some(session) = live_draft_session {
session.shutdown(context).await;
} else {
context.install_live_progress(None);
}
let claimed_events_finished =
claimed_event_ids.is_empty() || claimed_events_are_terminal(context, &claimed_event_ids);
finalize_claimed_runtime_events(context, &claimed_event_ids, &output);
if !claimed_event_ids.is_empty()
&& (claimed_events_finished || output_is_runtime_context_compaction_boundary(&output))
{
context.afterclaim_context_fingerprint = None;
context.visible_source_lines.clear();
context.token_estimate_baseline = TokenEstimateBaseline::default();
}
context.claimed_event_ids.clear();
if !runtime_step.is_history_empty() {
record_runtime_history_messages(context, runtime_step.into_turn_draft()).await;
}
record_skill_run_evidence(context, &output).await;
context.current_work_origin = None;
AgentLoopStepExecution
}
struct RuntimeErrorRecordInput<'a> {
turn_id: &'a str,
claimed_inputs: &'a [ClaimedRuntimeInput],
claimed_event_ids: &'a [String],
tools: &'a [crate::reasoning::runtime::AgentToolSpec],
context_text: &'a str,
error_kind: RuntimeErrorKind,
severity: u8,
detected_by: &'a str,
expected_behavior: &'a str,
actual_behavior: &'a str,
evidence: &'a str,
recoverability: &'a str,
retry_count: usize,
terminal_status: Option<&'a str>,
assistant_text: Option<&'a str>,
tool_calls: &'a [crate::reasoning::runtime::AgentToolCall],
tool_results: &'a [String],
actions: &'a [EpisodeActionRecord],
}
async fn record_runtime_error_case(context: &Context, input: RuntimeErrorRecordInput<'_>) {
let case = RuntimeErrorCase::new(RuntimeErrorCaseParts {
turn_id: input.turn_id.to_string(),
error_kind: input.error_kind,
severity: input.severity,
detected_by: input.detected_by.to_string(),
task: RuntimeErrorTaskContext {
origin: context.current_work_origin.clone(),
event_sources: runtime_error_event_sources(input.claimed_inputs),
user_request_summary: runtime_error_user_request_summary(input.claimed_inputs),
claimed_event_ids: input.claimed_event_ids.to_vec(),
},
runtime: RuntimeErrorRuntimeContext {
phase: context
.active_runtime_phase
.map(|phase| phase.label().to_string()),
available_tool_names: input.tools.iter().map(|tool| tool.name.clone()).collect(),
plan_summary: context
.plan
.steps()
.iter()
.take(8)
.map(|step| {
format!(
"{:?}: {}",
step.status,
compact_runtime_error_text(&step.step, 96)
)
})
.collect(),
compact_context_summary: Some(compact_runtime_error_text(input.context_text, 1600)),
},
action: RuntimeErrorActionContext {
assistant_text_summary: input
.assistant_text
.map(|text| compact_runtime_error_text(text, 600)),
tool_call_summaries: input
.tool_calls
.iter()
.map(|call| {
format!(
"{} {}",
call.name,
compact_runtime_error_text(&call.arguments.to_string(), 320)
)
})
.collect(),
tool_result_summaries: input
.tool_results
.iter()
.map(|result| compact_runtime_error_text(result, 320))
.collect(),
previous_action_window: input
.actions
.iter()
.rev()
.take(6)
.map(|action| {
format!(
"{}: {}",
action.kind,
compact_runtime_error_text(&action.summary, 160)
)
})
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect(),
},
observation: RuntimeErrorObservation {
expected_behavior: input.expected_behavior.to_string(),
actual_behavior: input.actual_behavior.to_string(),
evidence: compact_runtime_error_text(input.evidence, 1200),
recoverability: input.recoverability.to_string(),
retry_count: input.retry_count,
terminal_status: input.terminal_status.map(ToString::to_string),
},
contract_refs: runtime_error_contract_refs(input.error_kind),
});
append_runtime_error_case(case).await;
}
fn classify_tool_runtime_error(tool_name: &str, error_text: &str) -> RuntimeErrorKind {
let tool_name = tool_name
.split_once(crate::app::AppId::TOOL_NAME_SEPARATOR)
.map_or(tool_name, |(_, app_tool_name)| app_tool_name);
let lower = error_text.to_ascii_lowercase();
if tool_name == "update_plan"
|| lower.contains("update_plan must contain")
|| lower.contains("update_plan cannot contain")
{
return RuntimeErrorKind::PlanContractViolation;
}
if tool_name == "finish_and_send" && lower.contains("no claimed event") {
return RuntimeErrorKind::EventIdMissingOrStale;
}
if tool_name.starts_with("browser_") && lower.contains("stale") {
return RuntimeErrorKind::StaleBrowserRef;
}
if tool_name.starts_with("terminal_")
&& (lower.contains("session") || lower.contains("stdin"))
&& (lower.contains("missing") || lower.contains("not found") || lower.contains("invalid"))
{
return RuntimeErrorKind::WrongTerminalSessionContinuation;
}
if lower.contains("invalid arguments")
|| lower.contains("missing field")
|| lower.contains("invalid type")
|| lower.contains("requires a non-empty reply_message")
{
return RuntimeErrorKind::InvalidToolArgs;
}
RuntimeErrorKind::ToolSchemaError
}
fn runtime_error_contract_refs(kind: RuntimeErrorKind) -> Vec<String> {
match kind {
RuntimeErrorKind::MissingFinishAndSend
| RuntimeErrorKind::EventIdMissingOrStale
| RuntimeErrorKind::TransportCompletionViolation => {
vec!["event completion contract".to_string()]
}
RuntimeErrorKind::ClaimedInputLeftUnresolved => {
vec!["app notice completion contract".to_string()]
}
RuntimeErrorKind::InvalidToolArgs | RuntimeErrorKind::ToolSchemaError => {
vec!["tool argument contract".to_string()]
}
RuntimeErrorKind::StaleBrowserRef => vec!["browser reference freshness".to_string()],
RuntimeErrorKind::WrongTerminalSessionContinuation => {
vec!["terminal session continuation".to_string()]
}
RuntimeErrorKind::PlanContractViolation => vec!["plan contract".to_string()],
RuntimeErrorKind::RepeatedIdenticalToolError => vec!["tool retry contract".to_string()],
RuntimeErrorKind::ContextOverflowAfterRecovery => {
vec!["context overflow recovery".to_string()]
}
RuntimeErrorKind::ModelRequestRepeatedFailure => {
vec!["model request fuse".to_string()]
}
RuntimeErrorKind::ModelEmptyReasoningOutput => {
vec!["model output contract".to_string()]
}
}
}
fn runtime_error_event_sources(inputs: &[ClaimedRuntimeInput]) -> Vec<String> {
let mut sources = inputs
.iter()
.map(|input| input.source_label().to_string())
.collect::<Vec<_>>();
sources.sort();
sources.dedup();
sources
}
fn runtime_error_user_request_summary(inputs: &[ClaimedRuntimeInput]) -> Option<String> {
let summaries = inputs
.iter()
.map(|input| {
let event = input;
match &event.payload {
EventPayload::TelegramIncoming(payload) => compact_runtime_error_text(
&format!(
"telegram from {}: {}",
payload.sender, payload.incoming_text
),
240,
),
EventPayload::TerminalIncoming(payload) => compact_runtime_error_text(
&format!("terminal {}: {}", payload.origin, payload.incoming_text),
240,
),
}
})
.collect::<Vec<_>>();
if summaries.is_empty() {
None
} else {
Some(summaries.join(" | "))
}
}
fn compact_runtime_error_text(text: &str, max_chars: usize) -> String {
let compact = compact_horizontal_whitespace(text);
let mut chars = compact.chars();
let mut value = chars.by_ref().take(max_chars).collect::<String>();
if chars.next().is_some() {
value.push_str("...");
}
value
}
pub fn append_workflow_activity_event(
context: &Context,
tx: &tokio::sync::watch::Sender<DashboardState>,
result: &crate::workflow::WorkflowInvocationResult,
) {
append_committed_activity_cells(
context,
Some(tx),
vec![crate::dashboard::SessionActivityEvent::Workflow(
crate::dashboard::WorkflowActivityData {
workflow_id: result.workflow_id.clone(),
status: result.status.clone(),
output: result.output.clone(),
message: result.message.clone(),
snapshot: Some(result.snapshot.clone()),
},
)],
);
}
fn append_committed_activity_cells(
context: &Context,
tx: Option<&tokio::sync::watch::Sender<DashboardState>>,
cells: Vec<crate::dashboard::SessionActivityEvent>,
) {
let stable_ids = cells
.iter()
.map(stable_dashboard_activity_id_for_cell)
.collect::<Option<Vec<_>>>();
append_committed_activity_cells_with_ids(context, tx, cells, stable_ids);
}
fn append_committed_activity_cells_with_ids(
context: &Context,
tx: Option<&tokio::sync::watch::Sender<DashboardState>>,
cells: Vec<crate::dashboard::SessionActivityEvent>,
stable_ids: Option<Vec<String>>,
) {
if cells.is_empty() {
return;
}
let history_items = stable_ids.map_or_else(
|| dashboard_activity_items_from_cells(&cells),
|ids| dashboard_activity_items_from_cells_with_ids(&cells, ids),
);
let persisted_window = context.dashboard_history.as_ref().and_then(|history| {
persist_dashboard_activity_items(history, &history_items).map_or_else(
|err| {
tracing::warn!("persist dashboard activity history failed: {err:?}");
None
},
Some,
)
});
if let Some(tx) = tx {
tx.send_modify(|state| {
if let Some(window) = persisted_window.as_ref() {
state.activity_history = window.clone();
} else {
state
.activity_history
.merge_new_items(history_items.clone());
}
apply_activity_event(
state,
DashboardActivityEvent::AppendCommittedCells { cells },
);
});
}
}
fn refresh_pending_user_inputs_for_dashboard(
context: &Context,
tx: Option<&tokio::sync::watch::Sender<DashboardState>>,
) {
let Some(tx) = tx else {
return;
};
tx.send_modify(|state| {
state.pending_user_inputs = crate::dashboard::render::pending_user_inputs_from_sources(
&context.events,
&context.pending_work,
);
});
}
fn append_claimed_input_activity_cells(
context: &Context,
tx: Option<&tokio::sync::watch::Sender<DashboardState>>,
inputs: &[ClaimedRuntimeInput],
) {
if !inputs.is_empty() {
refresh_pending_user_inputs_for_dashboard(context, tx);
}
let mut cells = Vec::new();
let mut stable_ids = Vec::new();
for input in inputs {
if let Some(cell) = user_activity_cell_from_event(input) {
cells.push(cell);
stable_ids.push(dashboard_user_activity_item_id(input));
}
}
append_committed_activity_cells_with_ids(context, tx, cells, Some(stable_ids));
}
fn dashboard_user_activity_item_id(event: &EventView) -> String {
format!("activity-user-event-{}", event.event_id)
}
fn dashboard_activity_items_from_cells(
cells: &[crate::dashboard::SessionActivityEvent],
) -> Vec<crate::dashboard::DashboardActivityHistoryItem> {
dashboard_activity_items_from_cells_with_ids(
cells,
cells
.iter()
.map(|_| {
format!(
"activity-{}-{}",
chrono::Utc::now().timestamp_millis(),
uuid::Uuid::new_v4()
)
})
.collect(),
)
}
fn stable_dashboard_activity_id_for_cell(
cell: &crate::dashboard::SessionActivityEvent,
) -> Option<String> {
match cell {
crate::dashboard::SessionActivityEvent::Explored(group) => {
Some(format!("activity-{}", group.stable_id))
}
crate::dashboard::SessionActivityEvent::CodingEdit(edit) => {
Some(format!("activity-{}", edit.stable_id))
}
crate::dashboard::SessionActivityEvent::Workflow(workflow) => workflow
.snapshot
.as_ref()
.map(|snapshot| format!("activity-workflow-{}", snapshot.run_id)),
_ => None,
}
}
fn dashboard_activity_items_from_cells_with_ids(
cells: &[crate::dashboard::SessionActivityEvent],
item_ids: Vec<String>,
) -> Vec<crate::dashboard::DashboardActivityHistoryItem> {
cells
.iter()
.zip(item_ids)
.map(|(cell, item_id)| {
crate::dashboard::DashboardActivityHistoryItem::from_event_with_id(cell, &item_id)
})
.collect()
}
fn persist_dashboard_activity_items(
history: &DashboardActivityHistoryStore,
items: &[crate::dashboard::DashboardActivityHistoryItem],
) -> miette::Result<DashboardActivityHistoryWindow> {
history.append_items(items)?;
Ok(history.load_initial_window())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn afterclaim_context_completion_detection_requires_matching_close_tag() {
assert!(is_complete_afterclaim_context_text(
"<afterclaim_context>\nclaimed\n</afterclaim_context>"
));
assert!(!is_complete_afterclaim_context_text(
"<afterclaim_context>\nclaimed"
));
}
#[test]
fn afterclaim_agent_content_carries_telegram_image_parts() {
let input: AfterClaimContextInput = vec![crate::events::EventView {
event_id: uuid::Uuid::nil(),
status: crate::events::EventStatus::Claimed,
reply_message: None,
arrived_at_ms: 1,
payload: crate::events::EventPayload::TelegramIncoming(
crate::events::TelegramIncomingEvent {
chat_id: "chat-1".to_string(),
chat_kind: "private".to_string(),
chat_title: "Alice".to_string(),
sender: "alice".to_string(),
incoming_text: "inspect this".to_string(),
telegram_update_id: 10,
telegram_message_id: Some(20),
telegram_message_date: Some(30),
attachments: vec![crate::events::TelegramIncomingAttachment {
kind: crate::events::TelegramIncomingAttachmentKind::Image,
file_id: "file-1".to_string(),
file_unique_id: "unique-1".to_string(),
media_type: "image/png".to_string(),
local_path: "/tmp/image.png".to_string(),
description: Some("telegram photo 512x512".to_string()),
}],
},
),
last_error: None,
}];
let content = afterclaim_agent_content("claimed input".to_string(), &input);
assert_eq!(content.as_text(), "claimed input");
assert_eq!(
content.parts(),
&[AgentContentPart::Image {
path: "/tmp/image.png".to_string(),
media_type: "image/png".to_string(),
description: Some("telegram photo 512x512".to_string()),
}]
);
}
#[test]
fn claimed_event_activity_cells_use_stable_event_ids() {
let first_event_id =
uuid::Uuid::parse_str("11111111-1111-4111-8111-111111111111").expect("valid uuid");
let second_event_id =
uuid::Uuid::parse_str("22222222-2222-4222-8222-222222222222").expect("valid uuid");
let cells = render_activity_from_messages(vec![
HistoryMessage::user("hello"),
HistoryMessage::assistant("ack"),
]);
let items = dashboard_activity_items_from_cells_with_ids(
&cells,
vec![
format!("activity-user-event-{first_event_id}"),
format!("activity-user-event-{second_event_id}"),
],
);
let item_ids = items
.iter()
.map(|item| item.id.as_str())
.collect::<Vec<_>>();
assert_eq!(
item_ids,
vec![
"activity-user-event-11111111-1111-4111-8111-111111111111",
"activity-user-event-22222222-2222-4222-8222-222222222222",
]
);
assert!(matches!(
items[0].event,
crate::dashboard::SessionActivityEvent::User(_)
));
}
#[test]
fn runtime_compaction_boundary_output_is_detected() {
let output = runtime_context_compacted_output("compacted");
assert!(output_is_runtime_context_compaction_boundary(&output));
}
#[test]
fn coding_project_prepare_runs_only_for_new_claimed_input() {
assert!(!should_prepare_coding_project_for_claimed_input(None, None));
assert!(should_prepare_coding_project_for_claimed_input(
None,
Some("event-a")
));
assert!(should_prepare_coding_project_for_claimed_input(
Some("event-a"),
Some("event-b")
));
assert!(!should_prepare_coding_project_for_claimed_input(
Some("event-a"),
Some("event-a")
));
}
#[test]
fn coding_project_root_parser_ignores_none_and_reads_path() {
assert_eq!(
coding_project_root_from_lines(&[
"kind=coding".to_string(),
"project_root=none".to_string()
]),
None
);
let root = PathBuf::from("/tmp/project with spaces");
assert_eq!(
coding_project_root_from_lines(&[
"kind=coding".to_string(),
format!("project_root={}", root.display()),
"pending_review_events=0".to_string()
]),
Some(root)
);
}
#[tokio::test]
async fn coding_project_prepare_opens_project_scope() {
let project = tempfile::tempdir().expect("project dir");
let mut apps =
crate::app::AppManager::new(vec![Box::new(crate::coding_app::CodingApp::new())])
.expect("app manager");
let app_context = AppToolExecutionContext {
execution_cwd: project.path().to_path_buf(),
sandbox_policy: crate::sandbox::RuntimeSandboxPolicy::disabled(),
dashboard_tx: None,
tool_output_max_tokens: 4096,
turn_epoch: 0,
};
prepare_coding_project_app(&mut apps, project.path(), &app_context)
.await
.expect("prepare coding app");
assert!(coding_project_root_is_open(&apps, project.path()));
prepare_coding_project_app(&mut apps, project.path(), &app_context)
.await
.expect("prepare coding app again");
assert!(coding_project_root_is_open(&apps, project.path()));
}
#[tokio::test]
async fn coding_project_prepare_does_not_render_root_instructions_in_app_state() {
let project = tempfile::tempdir().expect("project dir");
std::fs::write(
project.path().join("AGENTS.md"),
"Root instruction marker\n",
)
.expect("write agents");
let mut apps =
crate::app::AppManager::new(vec![Box::new(crate::coding_app::CodingApp::new())])
.expect("app manager");
let app_context = AppToolExecutionContext {
execution_cwd: project.path().to_path_buf(),
sandbox_policy: crate::sandbox::RuntimeSandboxPolicy::disabled(),
dashboard_tx: None,
tool_output_max_tokens: 4096,
turn_epoch: 0,
};
prepare_coding_project_app(&mut apps, project.path(), &app_context)
.await
.expect("prepare coding app");
prepare_coding_project_app(&mut apps, project.path(), &app_context)
.await
.expect("prepare coding app again");
let state = apps
.state_render_for(&AppId::coding())
.expect("coding state");
let rendered_state = state.lines.join("\n");
assert!(!rendered_state.contains("<root_project_instructions>"));
assert!(!rendered_state.contains("Root instruction marker"));
}
}