diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index f7177c39ba..11485af929 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -899,6 +899,10 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { elapsedMs: event.subagent?.elapsed_ms ?? updatedCalls[callIdx].elapsedMs, outputChars: event.subagent?.output_chars ?? updatedCalls[callIdx].outputChars, result: event.output ?? updatedCalls[callIdx].result, + // Carry the structured failure so the child row keeps its "why / next" + // copy live instead of losing it until a snapshot reload (#4459). A + // successful result clears any stale failure on the row. + failure: event.success ? undefined : parseToolFailure(event.failure), }; const next = [...existing]; next[idx] = { ...entry, subagent: { ...entry.subagent, toolCalls: updatedCalls } }; @@ -912,6 +916,7 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { elapsedMs: event.subagent?.elapsed_ms, outputChars: event.subagent?.output_chars, result: event.output, + failure: event.success ? undefined : parseToolFailure(event.failure), }) ); }, diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index 7fbb37663d..cf6eb84ff8 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx @@ -1205,6 +1205,74 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria }); }); + it('stores the structured failure on a failed subagent tool call (#4459)', () => { + const listeners = renderProvider(); + const threadId = 'tsa-fail'; + + act(() => { + listeners.onSubagentSpawned?.({ + thread_id: threadId, + request_id: 'r1', + tool_name: 'researcher', + skill_id: 'sub-1', + message: 'spawned', + round: 1, + subagent: { mode: 'typed', dedicated_thread: false, prompt_chars: 42 }, + }); + listeners.onSubagentToolCall?.({ + thread_id: threadId, + request_id: 'r1', + round: 1, + tool_name: 'shell', + skill_id: 'sub-1', + tool_call_id: 'cc-1', + subagent: { agent_id: 'researcher', task_id: 'sub-1', child_iteration: 1 }, + }); + }); + + act(() => { + listeners.onSubagentToolResult?.({ + thread_id: threadId, + request_id: 'r1', + round: 1, + tool_name: 'shell', + skill_id: 'sub-1', + tool_call_id: 'cc-1', + success: false, + failure: { + class: 'denied', + category: 'user_declined', + cause_plain: 'You declined this action.', + next_action: 'Ask again if you change your mind.', + recoverable: false, + }, + subagent: { agent_id: 'researcher', task_id: 'sub-1', child_iteration: 1, elapsed_ms: 5 }, + }); + }); + + const timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? []; + const call = timeline[0]?.subagent?.toolCalls[0]; + expect(call).toMatchObject({ callId: 'cc-1', status: 'error' }); + // The structured why/next survives live rather than being dropped until a + // snapshot reload (#4459). + expect(call?.failure).toMatchObject({ + class: 'denied', + category: 'user_declined', + recoverable: false, + causePlain: 'You declined this action.', + nextAction: 'Ask again if you change your mind.', + }); + // The rendered live path uses `subagent.transcript`, so the failure must + // also land on the transcript tool item, not just the fallback list. + const transcriptTool = timeline[0]?.subagent?.transcript?.find( + i => i.kind === 'tool' && i.callId === 'cc-1' + ); + expect(transcriptTool).toMatchObject({ + status: 'error', + failure: { class: 'denied', category: 'user_declined' }, + }); + }); + it('ignores subagent_tool_call events that arrive before subagent_spawned', () => { const listeners = renderProvider(); const threadId = 'tsa-orphan'; diff --git a/app/src/services/chatService.ts b/app/src/services/chatService.ts index ddb1e55674..2cdf1e02ec 100644 --- a/app/src/services/chatService.ts +++ b/app/src/services/chatService.ts @@ -427,6 +427,13 @@ export interface ChatSubagentToolResultEvent { * `subagent.elapsed_ms`. */ output?: string; + /** + * Optional structured failure explanation for a FAILED child tool call + * (#4459), present only when `success` is false. Parsed via + * `parseToolFailure` and stored on the child row so the "why / next" copy + * survives live (previously it was dropped until a snapshot reload). + */ + failure?: unknown; subagent?: SubagentProgressDetail; } diff --git a/app/src/store/chatRuntimeSlice.ts b/app/src/store/chatRuntimeSlice.ts index e7ffdf2f03..de2b31874e 100644 --- a/app/src/store/chatRuntimeSlice.ts +++ b/app/src/store/chatRuntimeSlice.ts @@ -151,6 +151,10 @@ export type SubagentTranscriptItem = displayName?: string; /** Server-computed contextual detail (path / recipient / query). */ detail?: string; + /** Plain-language failure explanation for a FAILED child tool call + * (#4459) — kept on the transcript item so the rendered live path (not + * just the fallback `toolCalls` list) shows the why/next copy. */ + failure?: ToolFailureExplanation; }; /** One child tool call performed by a running sub-agent. */ @@ -730,6 +734,7 @@ function subagentTranscriptItemFromPersisted( outputChars: item.outputChars, displayName: item.displayName, detail: item.detail, + failure: item.failure, }; } return { kind: item.kind, iteration: item.iteration, text: item.text }; @@ -1109,9 +1114,11 @@ const chatRuntimeSlice = createSlice({ elapsedMs?: number; outputChars?: number; result?: string; + failure?: ToolFailureExplanation; }> ) => { - const { threadId, rowId, callId, success, elapsedMs, outputChars, result } = action.payload; + const { threadId, rowId, callId, success, elapsedMs, outputChars, result, failure } = + action.payload; const entry = state.toolTimelineByThread[threadId]?.find(e => e.id === rowId); const item = entry?.subagent?.transcript?.find(i => i.kind === 'tool' && i.callId === callId); if (!item || item.kind !== 'tool') return; @@ -1119,6 +1126,9 @@ const chatRuntimeSlice = createSlice({ if (elapsedMs != null) item.elapsedMs = elapsedMs; if (outputChars != null) item.outputChars = outputChars; if (result != null) item.result = result; + // Carry the structured why/next onto the rendered transcript item; a + // successful result clears any stale failure (#4459). + item.failure = success ? undefined : failure; }, setTaskBoardForThread: ( state, diff --git a/app/src/types/turnState.ts b/app/src/types/turnState.ts index 8be9848146..0a21da9d33 100644 --- a/app/src/types/turnState.ts +++ b/app/src/types/turnState.ts @@ -100,6 +100,9 @@ export type PersistedSubagentTranscriptItem = outputChars?: number; displayName?: string; detail?: string; + /** Plain-language failure explanation for a FAILED child tool call + * (#4459); mirrors {@link PersistedSubagentToolCall.failure}. */ + failure?: PersistedToolFailure; }; export interface PersistedSubagentActivity { diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 3133f9cb8d..c1df7d9d33 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -1204,9 +1204,13 @@ fn assemble_turn_harness( // the retained provider handle (the other clone was consumed into the // primary `ProviderModel`); `build_route_models` clones it per route and // skips the turn's own model so we don't shadow the default. - for route in - routes::build_route_models(&summary_provider, temperature, model, max_output_tokens) - { + for route in routes::build_route_models( + &summary_provider, + temperature, + model, + max_output_tokens, + &provider_usage_carry, + ) { let routes::RouteModel { name, model: route_model, diff --git a/src/openhuman/tinyagents/observability.rs b/src/openhuman/tinyagents/observability.rs index 3c14cf5455..6cd3e4c732 100644 --- a/src/openhuman/tinyagents/observability.rs +++ b/src/openhuman/tinyagents/observability.rs @@ -176,6 +176,22 @@ pub(crate) struct OpenhumanEventBridge { /// is recorded exactly once. See [`OpenhumanEventBridge::record_usage`]. recorded_iterations: Mutex>, state: Mutex, + /// Ordered overflow buffer for progress events that hit backpressure + /// (channel `Full`). Once ANY event spills here, `draining` stays set and + /// every subsequent event queues here too — a single spawned forwarder + /// drains them to the channel in FIFO order — so a later fast-path + /// `try_send` can never jump ahead of an earlier spilled event and scramble + /// start/completed ordering (which would leave a tool row stuck `running` + /// when a `ToolCallCompleted` overtakes its `ToolCallStarted`) (#4466). + overflow: Arc>, +} + +/// Backpressure overflow state guarded by a single mutex so the "are we +/// draining?" decision and the queue mutation stay atomic together. +#[derive(Default)] +struct OverflowState { + queue: std::collections::VecDeque, + draining: bool, } impl OpenhumanEventBridge { @@ -221,6 +237,7 @@ impl OpenhumanEventBridge { usage_carry, recorded_iterations: Mutex::new(std::collections::HashSet::new()), state: Mutex::new(BridgeState::default()), + overflow: Arc::default(), }) } @@ -267,17 +284,51 @@ impl OpenhumanEventBridge { let Some(tx) = &self.on_progress else { return; }; + // Hold the overflow lock across the ordering decision so "are we + // draining?" and the queue mutation are atomic (try_send is + // non-blocking, so holding a std mutex across it is fine). + let mut ov = self.overflow.lock().unwrap_or_else(|p| p.into_inner()); + if ov.draining { + // Already spilling: queue in order; the single forwarder delivers it. + ov.queue.push_back(progress); + return; + } match tx.try_send(progress) { Ok(()) => {} Err(TrySendError::Closed(_)) => {} Err(TrySendError::Full(progress)) => { - // Backpressure, not capacity loss: hand the delta to an awaited - // `send()` on a spawned task rather than dropping it. Guard on a - // live runtime so a non-async construction path can't panic. + // Backpressure, not capacity loss. Enter ordered-drain mode: + // queue this event and spawn ONE forwarder that awaits `send()` + // per event in FIFO order. `draining` stays set (so every later + // event also queues here) until the buffer fully drains — that is + // what stops a later `try_send` from overtaking a spilled earlier + // event and scrambling start/completed ordering. if let Ok(handle) = tokio::runtime::Handle::try_current() { + ov.queue.push_back(progress); + ov.draining = true; + let overflow = Arc::clone(&self.overflow); let tx = tx.clone(); + drop(ov); handle.spawn(async move { - let _ = tx.send(progress).await; + loop { + let next = { + let mut ov = overflow.lock().unwrap_or_else(|p| p.into_inner()); + match ov.queue.pop_front() { + Some(item) => item, + None => { + ov.draining = false; + break; + } + } + }; + if tx.send(next).await.is_err() { + // Receiver gone: stop draining, discard the rest. + let mut ov = overflow.lock().unwrap_or_else(|p| p.into_inner()); + ov.queue.clear(); + ov.draining = false; + break; + } + } }); } else { tracing::debug!( @@ -445,8 +496,13 @@ impl OpenhumanEventBridge { input_tokens: usage.input_tokens, output_tokens: usage.output_tokens, cached_input_tokens: usage.cache_read_tokens, - cache_creation_tokens: usage.cache_creation_tokens, - reasoning_tokens: usage.reasoning_tokens, + // Use the carried-or-crate breakdown (computed above): for + // providers where cache-creation/reasoning tokens ride only on + // the carried UsageInfo, the crate `usage.*` fields are 0, which + // would leave per-call progress/Langfuse telemetry at zero even + // though the cost tracker got the right values (#4467). + cache_creation_tokens, + reasoning_tokens, cost_usd: call_cost, }); self.send(AgentProgress::TurnCostUpdated { diff --git a/src/openhuman/tinyagents/routes.rs b/src/openhuman/tinyagents/routes.rs index d35a831cf7..a9d49c5405 100644 --- a/src/openhuman/tinyagents/routes.rs +++ b/src/openhuman/tinyagents/routes.rs @@ -89,6 +89,11 @@ pub(super) fn build_route_models( temperature: f64, skip_model: &str, max_output_tokens: Option, + // Shared provider-usage carry (#4467): fallback route models must drain the + // SAME carry the bridge reads, or a successful fallback call never feeds its + // backend-charged USD / context-window / cache-creation-reasoning breakdown + // to the cost surfaces (it would fall back to a catalogue estimate). + usage_carry: &super::observability::ProviderUsageCarry, ) -> Vec { let mut out = Vec::new(); for &tier in WORKLOAD_ROUTE_TIERS { @@ -108,7 +113,8 @@ pub(super) fn build_route_models( } let mut model = ProviderModel::new(provider.clone(), tier, temperature) .with_vision(vision) - .with_reasoning(reasoning); + .with_reasoning(reasoning) + .with_usage_carry(usage_carry.clone()); if let Some(cap) = max_output_tokens { model = model.with_max_tokens(cap); } diff --git a/src/openhuman/tools/impl/filesystem/update_memory_md.rs b/src/openhuman/tools/impl/filesystem/update_memory_md.rs index 5b5dcf1fa5..098d000318 100644 --- a/src/openhuman/tools/impl/filesystem/update_memory_md.rs +++ b/src/openhuman/tools/impl/filesystem/update_memory_md.rs @@ -45,6 +45,36 @@ fn workspace_write_lock(workspace_dir: &Path) -> Arc> { ) } +/// Acquire an **inter-process** advisory write lock for `workspace_dir` (#4458). +/// +/// [`WORKSPACE_WRITE_LOCKS`] only serializes writers inside this OS process, but +/// cron launches work via separate `tokio::process::Command` subprocesses that +/// don't share that mutex — so two cron runs could still clobber the same +/// `MEMORY.md` mid read-modify-write. This takes an `fs2` exclusive `flock` on a +/// sentinel `.memory-write.lock` file in the workspace; the returned `File` +/// holds the lock until it is dropped (end of the write). `flock` acquisition +/// blocks, so it runs on a blocking thread. +async fn acquire_cross_process_write_lock(workspace_dir: &Path) -> anyhow::Result { + let lock_path = workspace_dir.join(".memory-write.lock"); + tokio::task::spawn_blocking(move || { + use fs2::FileExt; + if let Some(parent) = lock_path.parent() { + let _ = std::fs::create_dir_all(parent); + } + let file = std::fs::OpenOptions::new() + .create(true) + .write(true) + .truncate(false) + .open(&lock_path) + .map_err(|e| anyhow::anyhow!("open workspace lock file {lock_path:?}: {e}"))?; + file.lock_exclusive() + .map_err(|e| anyhow::anyhow!("acquire workspace write flock: {e}"))?; + Ok::(file) + }) + .await + .map_err(|e| anyhow::anyhow!("workspace lock task join failed: {e}"))? +} + /// Atomically replace `path`'s contents with `content`. /// /// Writes to a sibling temp file in the same directory (so the rename stays on @@ -218,9 +248,13 @@ impl Tool for UpdateMemoryMdTool { // write so no interleaving append can be lost. let lock = workspace_write_lock(&workspace_dir); let _guard = lock.lock().await; + // Also take a cross-process advisory lock so cron subprocesses (which + // don't share the in-process mutex above) can't clobber the same file + // mid-RMW. Held across read + atomic write; released on drop. + let _file_lock = acquire_cross_process_write_lock(&workspace_dir).await?; tracing::debug!( workspace = %workspace_dir.display(), - "[update_memory_md] acquired per-workspace write lock" + "[update_memory_md] acquired per-workspace write lock (in-process + cross-process flock)" ); match action {