diff --git a/app/src/providers/ChatRuntimeProvider.tsx b/app/src/providers/ChatRuntimeProvider.tsx index 4ccfc75e0..16c5b98e0 100644 --- a/app/src/providers/ChatRuntimeProvider.tsx +++ b/app/src/providers/ChatRuntimeProvider.tsx @@ -922,6 +922,39 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { ); }, onSubagentToolResult: event => { + // Phase 5c: the Flows prompt bar / canvas copilot route to the + // `workflow_builder` specialist via delegation (`build_workflow`), so a + // `propose_workflow`/`revise_workflow` proposal is produced INSIDE the + // delegated worker and arrives here (not on `onToolResult`). This + // extraction must run BEFORE the timeline-entry guards below: under + // the workflow_builder subagent's heavy event volume, the progress + // channel (bounded, `try_send`) can drop earlier events, so the + // timeline row for this call may never have been created — gating + // proposal extraction on finding that row silently drops the + // proposal and the Accept/Reject card never renders (bug). The + // extraction only needs `tool_name`/`success`/`output`, all present + // directly on the event, with no timeline dependency. Surface it on + // the PARENT thread (`event.thread_id`, which the progress bridge + // always stamps with the parent request's thread, not the child's) + // so the same `WorkflowProposalCard` the direct-tool path uses + // renders it. Still validate-only — the card's explicit Save is the + // sole persistence gate. + const subagentProposal = maybeParseWorkflowProposalTool( + event.tool_name, + event.success, + event.output + ); + if (subagentProposal) { + rtLog('workflow proposal parsed (delegated worker)', { + thread: event.thread_id, + tool: event.tool_name, + name: subagentProposal.name, + }); + dispatch( + setWorkflowProposalForThread({ threadId: event.thread_id, proposal: subagentProposal }) + ); + } + const taskId = event.subagent?.task_id ?? event.skill_id; const agentId = event.subagent?.agent_id; if (!agentId) return; @@ -960,30 +993,6 @@ const ChatRuntimeProvider = ({ children }: { children: React.ReactNode }) => { failure: event.success ? undefined : parseToolFailure(event.failure), }) ); - - // Phase 5c: the Flows prompt bar / canvas copilot route to the - // `workflow_builder` specialist via delegation (`build_workflow`), so a - // `propose_workflow`/`revise_workflow` proposal is produced INSIDE the - // delegated worker and arrives here (not on `onToolResult`). Surface it - // on the PARENT thread (`event.thread_id`) so the same - // `WorkflowProposalCard` / copilot the direct-tool path uses renders it. - // Still validate-only — the card's explicit Save is the sole persistence - // gate. - const subagentProposal = maybeParseWorkflowProposalTool( - event.tool_name, - event.success, - event.output - ); - if (subagentProposal) { - rtLog('workflow proposal parsed (delegated worker)', { - thread: event.thread_id, - tool: event.tool_name, - name: subagentProposal.name, - }); - dispatch( - setWorkflowProposalForThread({ threadId: event.thread_id, proposal: subagentProposal }) - ); - } }, onSubagentTextDelta: (event: ChatSubagentTextDeltaEvent) => { const taskId = event.subagent?.task_id; diff --git a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx index a5ec9e92d..03cb0fe29 100644 --- a/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx +++ b/app/src/providers/__tests__/ChatRuntimeProvider.test.tsx @@ -1354,6 +1354,52 @@ describe('ChatRuntimeProvider — dedupe, proactive resolution, mid-turn invaria const timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? []; expect(timeline).toHaveLength(0); }); + + // Regression: the Flows canvas copilot delegates to the `workflow_builder` + // subagent, which can emit 50+ progress events. The progress channel is + // bounded and `try_send`s, so under that volume the `subagent_spawned`/ + // `subagent_tool_call` events that create the timeline row can be dropped + // before `subagent_tool_result` arrives. Proposal extraction must not be + // gated on that timeline row existing, or the Accept/Reject + // `WorkflowProposalCard` silently never renders. + it('surfaces a workflow proposal from a delegated subagent even with no matching timeline row', () => { + const listeners = renderProvider(); + const threadId = 'tsa-no-row'; + + act(() => { + listeners.onSubagentToolResult?.({ + thread_id: threadId, + request_id: 'r1', + round: 1, + tool_name: 'revise_workflow', + skill_id: 'sub-missing', + tool_call_id: 'cc-1', + success: true, + output: JSON.stringify({ + type: 'workflow_proposal', + name: 'Notify on new signup', + graph: { nodes: [], edges: [] }, + require_approval: true, + summary: { trigger: 'signup.created', steps: [] }, + }), + // No `subagent` block and no prior `onSubagentSpawned`/`onSubagentToolCall` + // — so no timeline row exists for this call, mirroring the drop. + }); + }); + + // No timeline row was ever created for this call. + const timeline = store.getState().chatRuntime.toolTimelineByThread[threadId] ?? []; + expect(timeline).toHaveLength(0); + + // The proposal still reaches the parent thread so the Accept/Reject + // card renders. + const proposal = store.getState().chatRuntime.pendingWorkflowProposalsByThread[threadId]; + expect(proposal).toMatchObject({ + name: 'Notify on new signup', + requireApproval: true, + summary: { trigger: 'signup.created' }, + }); + }); }); // Regression: on Windows users report being "locked out" of the composer diff --git a/src/openhuman/agent_orchestration/tools/dispatch.rs b/src/openhuman/agent_orchestration/tools/dispatch.rs index be577eb46..74529cdf9 100644 --- a/src/openhuman/agent_orchestration/tools/dispatch.rs +++ b/src/openhuman/agent_orchestration/tools/dispatch.rs @@ -213,6 +213,30 @@ pub(crate) async fn dispatch_subagent( outcome.output.chars().count(), outcome.iterations, ); + // Also send to the per-request progress sink (mirrors + // `spawn_subagent.rs`) so the web channel bridge emits + // `subagent_done` to the frontend. Without this the delegated + // subagent's timeline row (created on `SubagentSpawned` above) + // stays "running" forever — `publish_subagent_completed` only + // fires the internal DomainEvent bus, not the per-request + // progress channel the UI's timeline is driven from. + if let Some(progress) = current_parent().and_then(|p| p.on_progress.clone()) { + let _ = progress + .send(AgentProgress::SubagentCompleted { + agent_id: outcome.agent_id.clone(), + task_id: outcome.task_id.clone(), + elapsed_ms: outcome.elapsed.as_millis() as u64, + iterations: outcome.iterations as u32, + output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), + // Synchronous delegate dispatch has no worktree + // isolation (that is a `spawn_subagent` concept). + worktree_path: None, + changed_files: Vec::new(), + dirty_status: None, + }) + .await; + } log::info!( "[agent] {} completed via {} iterations={} output_chars={}", agent_id, @@ -236,6 +260,24 @@ pub(crate) async fn dispatch_subagent( outcome.output.chars().count(), outcome.iterations, ); + // Same progress-sink mirror as the `Completed` arm above — + // an incomplete stop is still lifecycle-completed, so the + // timeline row must be released from "running" here too. + if let Some(progress) = current_parent().and_then(|p| p.on_progress.clone()) { + let _ = progress + .send(AgentProgress::SubagentCompleted { + agent_id: outcome.agent_id.clone(), + task_id: outcome.task_id.clone(), + elapsed_ms: outcome.elapsed.as_millis() as u64, + iterations: outcome.iterations as u32, + output_chars: outcome.output.chars().count(), + output: outcome.output.clone(), + worktree_path: None, + changed_files: Vec::new(), + dirty_status: None, + }) + .await; + } log::info!( "[agent] {} stopped incomplete via {} (task_id={}) iterations={} — \ returning partial-progress envelope, not a finished result", diff --git a/src/openhuman/agent_registry/agents/orchestrator/prompt.rs b/src/openhuman/agent_registry/agents/orchestrator/prompt.rs index e39b21a5f..2313b1dbc 100644 --- a/src/openhuman/agent_registry/agents/orchestrator/prompt.rs +++ b/src/openhuman/agent_registry/agents/orchestrator/prompt.rs @@ -102,8 +102,12 @@ fn render_installed_skills(skills: &[Workflow]) -> String { (name the skill and what you want done); it loads and runs the skill in an \ isolated worker and returns only the result, plus a `## Handoff Plan` for any \ step the worker couldn't perform — execute those steps yourself under the \ - approval gate. Use `describe_workflow` for full details. Use \ - `skill_registry_browse` / `skill_registry_search` to find and install new skills.\n\n", + approval gate. Use `describe_workflow` for full details on one of THESE \ + installed skills (it only knows about entries in this list, not Flows \ + automations — do not call it with a Flows `workflow_id`, it will error). Use \ + `skill_registry_browse` / `skill_registry_search` to find and install new skills. \ + For Flows automations (build/inspect/run a tinyflows workflow), use \ + `build_workflow` / the workflow_builder delegate instead.\n\n", ); for skill in skills { let id = if skill.dir_name.is_empty() { diff --git a/src/openhuman/channels/providers/web/run_task.rs b/src/openhuman/channels/providers/web/run_task.rs index 19d756cc7..a5b6c3411 100644 --- a/src/openhuman/channels/providers/web/run_task.rs +++ b/src/openhuman/channels/providers/web/run_task.rs @@ -199,7 +199,14 @@ pub(crate) async fn run_chat_task( } } - let (progress_tx, progress_rx) = tokio::sync::mpsc::channel(64); + // Bounded to 256 (was 64): a heavy-event subagent (e.g. `workflow_builder`, + // 50+ progress events per run) can overflow a smaller buffer, and the + // sender side uses `try_send` — an overflow silently drops progress + // events (including the ones that create a subagent's timeline row), + // which can in turn hide UI state derived from those events. This is + // defense-in-depth; extraction of durable state (like a workflow + // proposal) must not depend on any single progress event surviving. + let (progress_tx, progress_rx) = tokio::sync::mpsc::channel(256); agent.set_on_progress(Some(progress_tx)); agent.set_run_queue(Some(run_queue)); let turn_state_store = TurnStateStore::new(config.workspace_dir.clone());