feat: implement company pipeline orchestration with user commands for full, quick, and skip modes

This commit is contained in:
asepharyana
2026-07-13 05:33:04 +07:00
parent a6eed9e574
commit 2310c2df7f
9 changed files with 269 additions and 94 deletions
+12
View File
@@ -77,6 +77,18 @@ Controller (key input → Action) → Event Loop → LLM stream → Tool executi
- **Tools** — `trait Tool { fn name() -> &str, fn run() -> Result<String> }`, 28 impls, gated by `Harness`. - **Tools** — `trait Tool { fn name() -> &str, fn run() -> Result<String> }`, 28 impls, gated by `Harness`.
- **Shell safety** — `tool/shell_filter/` blocks credential leaks and destructive git commands. - **Shell safety** — `tool/shell_filter/` blocks credential leaks and destructive git commands.
### Company Pipeline (Division Architecture)
- **5 divisions** in `src/app/subagent/division.rs`: Strategy, Engineering, Quality, Security, Documentation.
- **Pipeline orchestrator** in `src/app/workflow/company.rs`: two modes:
- `run_company_pipeline()` — full 5-division pipeline
- `run_company_pipeline_quick()` — 3-division (Strategy → Engineering → Quality)
- **Auto-CEO trigger** in `run_agent_turn()` (`actions/mod.rs`): detects complex requests via `is_complex_request()` heuristics, auto-delegates to pipeline.
- **Override** via `/pipeline full|quick|skip` sets `MiscState::pipeline_override`, consumed on next turn.
- **Live division progress** in TUI panel (`view/workflow.rs`): shows division name + current tool via `AgentStatus::progress`.
- **Auto inline review** after each edit: `src/app/subagent/auto.rs``spawn_quick_review()` injects verdict back into LLM conversation.
- **Background subagents** (test-gen, arch-review, security-review) fire asynchronously at turn end via `TurnEvent::SystemNote`.
## Code Documentation ## Code Documentation
Every function, struct, enum, trait, module, and significant code block must have a doc comment (`///` or `//!`) that explains: Every function, struct, enum, trait, module, and significant code block must have a doc comment (`///` or `//!`) that explains:
+18 -5
View File
@@ -15,7 +15,7 @@ Zesdex is a Rust-powered AI assistant that operates directly in your terminal vi
- **IPC Protocol** — Bidirectional state synchronization between daemon and client processes with diff-based updates. - **IPC Protocol** — Bidirectional state synchronization between daemon and client processes with diff-based updates.
- **Provider Agnostic** — Configurable AI model providers with dynamic model selection, per-role temperature/token limits, and API key management. - **Provider Agnostic** — Configurable AI model providers with dynamic model selection, per-role temperature/token limits, and API key management.
### Tool System (33 built-in tools) ### Tool System (34 built-in tools)
| Category | Tools | | Category | Tools |
|----------|-------| |----------|-------|
@@ -25,15 +25,23 @@ Zesdex is a Rust-powered AI assistant that operates directly in your terminal vi
| **Git** | `git_operator`, `git_worktree`, `git_cred` | | **Git** | `git_operator`, `git_worktree`, `git_cred` |
| **Memory** | `remember`, `recall`, `forget` | | **Memory** | `remember`, `recall`, `forget` |
| **Planning** | `plan_enter`, `plan_ready`, `seqthink` | | **Planning** | `plan_enter`, `plan_ready`, `seqthink` |
| **Workflow** | `workflow_run`, `note_finding` | | **Workflow** | `workflow_run`, `note_finding`, `company_pipeline` |
| **Utility** | `cd`, `dir_list`, `dir_cache_update`, `pong`, `todowrite`, `todofinish` | | **Utility** | `cd`, `dir_list`, `dir_cache_update`, `pong`, `todowrite`, `todofinish` |
| **LSP** | `lsp_connect`, `lsp_diagnostics`, `lsp_hover`, `lsp_completion`, `lsp_definition`, `lsp_references`, `lsp_disconnect` | | **LSP** | `lsp_connect`, `lsp_diagnostics`, `lsp_hover`, `lsp_completion`, `lsp_definition`, `lsp_references`, `lsp_disconnect` |
### Intelligence ### Intelligence
- **Company Pipeline** — Autonomous agent orchestration modeled as a company with specialized divisions. The CEO (main agent) automatically delegates work to 5 divisions in sequence:
```
Strategy → Engineering → Quality → Security → Documentation
```
Each division has a dedicated role, toolset, and system prompt. Controlled via `/pipeline full|quick|skip`.
- **Workflow Engine** — Orchestrate complex multi-step tasks with parallel sub-agents, pipelines, and phased execution. Spawn independent workers that share findings in real-time. - **Workflow Engine** — Orchestrate complex multi-step tasks with parallel sub-agents, pipelines, and phased execution. Spawn independent workers that share findings in real-time.
- **Self-Learning** — Persistent memory system that stores lessons, references, and project knowledge across sessions. Memories include provenance tracking, lifecycle management, and scope isolation. - **Self-Learning** — Persistent memory system that stores lessons, references, and project knowledge across sessions. Memories include provenance tracking, lifecycle management, and scope isolation.
- **Self-Review** — Adaptive quality review system that evaluates completed work against stored lessons and project conventions. - **Self-Review** — Review subagents trigger automatically after each code edit (inline) and at turn completion (background). Three types: code quality, architecture, and security.
- **Self-Healing** — On build/test failures, spawns a sub-agent with the error context to autonomously fix issues before reporting them to the user. - **Self-Healing** — On build/test failures, spawns a sub-agent with the error context to autonomously fix issues before reporting them to the user.
- **MCP Support** — [Model Context Protocol](https://modelcontextprotocol.io/) integration for connecting to external AI tool servers. - **MCP Support** — [Model Context Protocol](https://modelcontextprotocol.io/) integration for connecting to external AI tool servers.
- **Sequential Thinking** — Chain-of-thought reasoning tool for step-by-step problem decomposition. - **Sequential Thinking** — Chain-of-thought reasoning tool for step-by-step problem decomposition.
@@ -85,7 +93,8 @@ src/
│ ├── harness.rs # Tool harness for agent execution │ ├── harness.rs # Tool harness for agent execution
│ ├── workflow/ # Workflow engine │ ├── workflow/ # Workflow engine
│ │ ├── script.rs # Workflow script DSL │ │ ├── script.rs # Workflow script DSL
│ │ ── engine.rs # Workflow executor │ │ ── engine.rs # Workflow executor
│ │ └── company.rs # Company pipeline orchestrator
│ ├── mcp/ # MCP client manager │ ├── mcp/ # MCP client manager
│ │ └── manager.rs # MCP server lifecycle and tool exposure │ │ └── manager.rs # MCP server lifecycle and tool exposure
│ ├── subagent/ # Sub-agent management │ ├── subagent/ # Sub-agent management
@@ -145,7 +154,7 @@ src/
│ ├── manager.rs # OAuth token manager │ ├── manager.rs # OAuth token manager
│ ├── pkce.rs # PKCE code challenge/verifier │ ├── pkce.rs # PKCE code challenge/verifier
│ └── mod.rs │ └── mod.rs
├── tool/ # 33 tool implementations ├── tool/ # 34 tool implementations
│ ├── fs/ # read, write, edit, delete │ ├── fs/ # read, write, edit, delete
│ │ ├── read.rs │ │ ├── read.rs
│ │ ├── write.rs │ │ ├── write.rs
@@ -228,6 +237,10 @@ RUST_LOG=debug zesdex
| `/help` | Show help | | `/help` | Show help |
| `/clear` | Clear transcript | | `/clear` | Clear transcript |
| `/model` | Select AI model provider | | `/model` | Select AI model provider |
| `/pipeline` | Show current pipeline mode |
| `/pipeline full` | Force full company pipeline (5 divisions) on next request |
| `/pipeline quick` | Force quick pipeline (3 divisions) on next request |
| `/pipeline skip` | Skip pipeline — handle next request directly |
| `/exit` | Exit application | | `/exit` | Exit application |
| `/settings` | Open settings | | `/settings` | Open settings |
| `Any text` | Sent to the AI assistant as a prompt | | `Any text` | Sent to the AI assistant as a prompt |
+122 -52
View File
@@ -81,6 +81,10 @@ pub enum Action {
RunWorkflow { RunWorkflow {
script: String, script: String,
}, },
/// User-initiated pipeline via `/pipeline full|quick|skip`.
RunPipeline {
mode: String,
},
} }
/// Apply an `Action` to the application state. /// Apply an `Action` to the application state.
@@ -530,6 +534,9 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
} }
} }
if turn_finished { if turn_finished {
// Consume pipeline override after each turn so it doesn't
// persist across multiple submissions.
state.misc.pipeline_override = None;
maybe_trigger_review(state); maybe_trigger_review(state);
} }
if turn_finished || state.dirty { if turn_finished || state.dirty {
@@ -591,6 +598,30 @@ pub fn apply_action(state: &mut AppStateRest, action: Action) {
state.push_toast(Toast::new(ToastKind::Info, format!("deleted lesson: {}", name))); state.push_toast(Toast::new(ToastKind::Info, format!("deleted lesson: {}", name)));
state.dirty = true; state.dirty = true;
} }
Action::RunPipeline { mode } => {
match mode.as_str() {
"full" => {
state.misc.pipeline_override = Some("full".to_string());
state.push_toast(Toast::new(ToastKind::Info, "Pipeline mode: full (5 divisions) — next request will run Strategy→Engineering→Quality→Security→Documentation".to_string()));
}
"quick" => {
state.misc.pipeline_override = Some("quick".to_string());
state.push_toast(Toast::new(ToastKind::Info, "Pipeline mode: quick (3 divisions) — next request will run Strategy→Engineering→Quality".to_string()));
}
"skip" => {
state.misc.pipeline_override = Some("skip".to_string());
state.push_toast(Toast::new(ToastKind::Info, "Pipeline mode: skip — next request will NOT run the company pipeline".to_string()));
}
"status" => {
let current = state.misc.pipeline_override.as_deref().unwrap_or("auto");
state.push_toast(Toast::new(ToastKind::Info, format!("Pipeline mode: {} (use /pipeline full|quick|skip to change)", current)));
}
_ => {
state.push_toast(Toast::new(ToastKind::Error, format!("Unknown pipeline mode: {} (use: full, quick, skip)", mode)));
}
}
state.dirty = true;
}
Action::RunWorkflow { script } => { Action::RunWorkflow { script } => {
// Open the Workflow overlay so the user can see progress. // Open the Workflow overlay so the user can see progress.
state.misc.overlay = Overlay::Workflow; state.misc.overlay = Overlay::Workflow;
@@ -748,6 +779,7 @@ fn spawn_turn(state: &AppStateRest) {
}) = true; }) = true;
let events_q = turn_events.clone(); let events_q = turn_events.clone();
let pipeline_mode = state.misc.pipeline_override.clone();
std::thread::spawn(move || { std::thread::spawn(move || {
let db = crate::model::msglog::open_or_create(&edit_session_dir) let db = crate::model::msglog::open_or_create(&edit_session_dir)
@@ -767,6 +799,7 @@ fn spawn_turn(state: &AppStateRest) {
temperature, temperature,
max_tokens, max_tokens,
abort_flag, abort_flag,
pipeline_mode,
}; };
let result = run_agent_turn(tc, &messages, &events_q); let result = run_agent_turn(tc, &messages, &events_q);
if let Err(e) = result { if let Err(e) = result {
@@ -795,6 +828,9 @@ struct TurnCtx {
temperature: f32, temperature: f32,
max_tokens: Option<u32>, max_tokens: Option<u32>,
abort_flag: std::sync::Arc<std::sync::atomic::AtomicBool>, abort_flag: std::sync::Arc<std::sync::atomic::AtomicBool>,
/// Pipeline override: None=auto, Some("full"), Some("quick"), Some("skip").
/// Set by the `/pipeline` slash command. Consumed once per turn.
pipeline_mode: Option<String>,
} }
/// Build an ASCII tree of the workspace directory structure for the /// Build an ASCII tree of the workspace directory structure for the
@@ -993,17 +1029,16 @@ fn run_agent_turn(
} }
// ── AUTO CEO PIPELINE ── // ── AUTO CEO PIPELINE ──
// Before the main agent starts working, check if the request is complex // Before the main agent starts working, check if the pipeline should run.
// enough to warrant the full company pipeline. If so, delegate to the // The pipeline mode is determined by:
// divisions (Strategy → Engineering → Quality → Security → Documentation) // 1. User override: `/pipeline full|quick|skip` (consumed once)
// and inject the results before the main agent even starts. // 2. Auto-detect: `is_complex_request()` heuristics
// //
// This only triggers on the first turn of a session (few user messages) // This only triggers on the first turn of a session to avoid re-planning.
// to avoid re-planning mid-conversation.
let user_msg_count = msgs.iter() let user_msg_count = msgs.iter()
.filter(|m| matches!(m.role, crate::dto::chat::message::Role::User)) .filter(|m| matches!(m.role, crate::dto::chat::message::Role::User))
.count(); .count();
if user_msg_count <= 2 { let should_pipeline = if user_msg_count <= 2 {
let user_request = msgs.iter() let user_request = msgs.iter()
.rev() .rev()
.filter(|m| matches!(m.role, crate::dto::chat::message::Role::User)) .filter(|m| matches!(m.role, crate::dto::chat::message::Role::User))
@@ -1011,61 +1046,96 @@ fn run_agent_turn(
.and_then(|m| m.content.as_deref()) .and_then(|m| m.content.as_deref())
.unwrap_or(""); .unwrap_or("");
if !user_request.is_empty() if !user_request.is_empty() {
&& crate::app::workflow::company::is_complex_request(user_request) match tc.pipeline_mode.as_deref() {
{ Some("skip") => {
tracing::info!( tracing::debug!("[ceo] pipeline skipped via /pipeline skip");
"[ceo] complex request detected — delegating to company pipeline" false
); }
Some("full") => true,
// Notify TUI that pipeline is starting Some("quick") => true,
if let Ok(mut q) = events_q.lock() { _ => crate::app::workflow::company::is_complex_request(user_request),
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: "Company pipeline started: Strategy → Engineering → Quality → Security → Documentation".to_string(),
});
} }
} else {
false
}
} else {
false
};
// Run the full company pipeline (blocks this thread — OK since if should_pipeline {
// run_agent_turn already runs on a dedicated thread). let user_request = msgs.iter()
match crate::app::workflow::company::run_company_pipeline( .rev()
.filter(|m| matches!(m.role, crate::dto::chat::message::Role::User))
.next()
.and_then(|m| m.content.as_deref())
.unwrap_or("");
let use_full = tc.pipeline_mode.as_deref() != Some("quick");
let mode_label = if use_full { "full" } else { "quick" };
tracing::info!(
"[ceo] pipeline triggered (mode={}) — delegating to company pipeline",
mode_label
);
if let Ok(mut q) = events_q.lock() {
q.push_back(TurnEvent::SystemNote {
kind: "pipeline".to_string(),
message: format!(
"Company pipeline started ({}): {} → Engineering → Quality{}",
mode_label,
"Strategy",
if use_full { " → Security → Documentation" } else { "" },
),
});
}
let pipeline_result = if use_full {
crate::app::workflow::company::run_company_pipeline(
user_request, user_request,
&tc.edit_log_session_dir, &tc.edit_log_session_dir,
&tc.workspace_roots, &tc.workspace_roots,
Some(events_q), Some(events_q),
) { )
Ok(summary) => { } else {
tracing::info!("[ceo] company pipeline completed successfully"); crate::app::workflow::company::run_company_pipeline_quick(
let pipeline_msg = ChatMessage::system(format!( user_request,
"=== Company Pipeline — Executive Summary ===\n\ &tc.edit_log_session_dir,
The divisions have completed their work.\n\ &tc.workspace_roots,
Review the results below as CEO, then deliver to the user.\n\n\ Some(events_q),
{}", )
summary, };
));
archive_message(&tc.db, &tc.session_id, &pipeline_msg);
msgs.push(pipeline_msg);
if let Ok(mut q) = events_q.lock() { match pipeline_result {
q.push_back(TurnEvent::SystemNote { Ok(summary) => {
kind: "pipeline".to_string(), tracing::info!("[ceo] company pipeline completed successfully");
message: "Company pipeline complete. CEO reviewing results...".to_string(), let pipeline_msg = ChatMessage::system(format!(
}); "[Company Pipeline: {}]\n{}",
} mode_label,
} summary,
Err(e) => { ));
tracing::warn!("[ceo] company pipeline failed: {}", e); archive_message(&tc.db, &tc.session_id, &pipeline_msg);
let fail_msg = ChatMessage::system(format!( msgs.push(pipeline_msg);
"[Pipeline Note] The company pipeline encountered issues: {}.\n\
Proceeding with direct execution as fallback.", if let Ok(mut q) = events_q.lock() {
e, q.push_back(TurnEvent::SystemNote {
)); kind: "pipeline".to_string(),
msgs.push(fail_msg); message: format!("Company pipeline ({}) complete. CEO reviewing results...", mode_label),
});
} }
} }
} else { Err(e) => {
tracing::debug!("[ceo] request not complex — handling directly"); tracing::warn!("[ceo] company pipeline failed: {}", e);
let fail_msg = ChatMessage::system(format!(
"[Pipeline Note] The company pipeline encountered issues: {}.\n\
Proceeding with direct execution as fallback.",
e,
));
msgs.push(fail_msg);
}
} }
} else {
tracing::debug!("[ceo] pipeline not triggered — handling directly");
} }
let mut turn_step = 0usize; let mut turn_step = 0usize;
+3
View File
@@ -69,6 +69,9 @@ pub fn apply_command(command: Command) -> Vec<Action> {
Command::WorkflowRun { script } => { Command::WorkflowRun { script } => {
vec![Action::RunWorkflow { script }] vec![Action::RunWorkflow { script }]
} }
Command::Pipeline { mode } => {
vec![Action::RunPipeline { mode }]
}
Command::Unknown(cmd) => { Command::Unknown(cmd) => {
vec![Action::SystemNote { vec![Action::SystemNote {
kind: "error".to_string(), kind: "error".to_string(),
+12
View File
@@ -90,6 +90,10 @@ const COMMANDS: &[&str] = &[
"/model add", "/model add",
"/workflow", "/workflow",
"/workflow run", "/workflow run",
"/pipeline",
"/pipeline full",
"/pipeline quick",
"/pipeline skip",
"/compact", "/compact",
]; ];
@@ -289,6 +293,13 @@ pub struct MiscState {
pub api_context_length: Option<u32>, pub api_context_length: Option<u32>,
pub tick_count: u64, pub tick_count: u64,
pub todo_content: String, pub todo_content: String,
/// Pipeline mode override set by `/pipeline` command.
/// - `None`: auto-detect (default)
/// - `Some("full")`: force full pipeline
/// - `Some("quick")`: force quick pipeline
/// - `Some("skip")`: skip pipeline, handle directly
/// Consumed on the next agent turn.
pub pipeline_override: Option<String>,
} }
impl MiscState { impl MiscState {
@@ -307,6 +318,7 @@ impl MiscState {
api_context_length: None, api_context_length: None,
tick_count: 0, tick_count: 0,
todo_content: String::new(), todo_content: String::new(),
pipeline_override: None,
} }
} }
+43 -28
View File
@@ -185,7 +185,11 @@ pub fn run_company_pipeline_quick(
Ok(build_executive_summary(user_request, &results, &all_findings, quick_divisions)) Ok(build_executive_summary(user_request, &results, &all_findings, quick_divisions))
} }
/// Build a consolidated executive summary from pipeline results. /// Build a compressed executive summary from pipeline results.
///
/// Keeps output brief to save context window space — just division verdicts
/// and key findings, not full outputs. Full results are accessible to the
/// CEO via the notes/findings that were archived during execution.
fn build_executive_summary( fn build_executive_summary(
request: &str, request: &str,
results: &[String], results: &[String],
@@ -193,41 +197,23 @@ fn build_executive_summary(
divisions: &[division::Division], divisions: &[division::Division],
) -> String { ) -> String {
let mut summary = String::new(); let mut summary = String::new();
summary.push_str(&format!("# Company Pipeline — Executive Summary\n\n")); summary.push_str(&format!("Pipeline for: {}\n", request));
summary.push_str(&format!("**Request**: {}\n\n", request));
summary.push_str("## Division Results\n\n");
for (i, div) in divisions.iter().enumerate() { for (i, div) in divisions.iter().enumerate() {
let result_summary = results.get(i) let verdict = results.get(i)
.map(|r| { .map(|r| {
let first_line = r.lines().next().unwrap_or(r); r.lines().next().unwrap_or(r)
if first_line.len() > 120 { .chars().take(100).collect::<String>()
format!("{}...", &first_line[..117])
} else {
first_line.to_string()
}
}) })
.unwrap_or_else(|| "No output".to_string()); .unwrap_or_else(|| "".to_string());
summary.push_str(&format!("### {} Division\n", div.name)); summary.push_str(&format!(" {}: {}\n", div.name, verdict));
summary.push_str(&format!("- Role: {}\n", div.description));
summary.push_str(&format!("- Result: {}\n\n", result_summary));
} }
if !findings.is_empty() { if !findings.is_empty() {
summary.push_str("## Cross-Division Findings\n\n"); summary.push_str(&format!(" Notes: {} cross-division finding(s)\n", findings.len()));
for (i, f) in findings.iter().enumerate() {
summary.push_str(&format!("{}. {}\n", i + 1, f));
}
summary.push_str("\n");
} }
summary.push_str("---\n");
summary.push_str(&format!(
"Pipeline completed: {} division(s) executed.\n",
divisions.len(),
));
summary summary
} }
@@ -239,14 +225,43 @@ fn build_executive_summary(
/// ///
/// Used by the auto-CEO pipeline trigger in run_agent_turn to decide /// Used by the auto-CEO pipeline trigger in run_agent_turn to decide
/// whether to delegate to the full company pipeline or handle directly. /// whether to delegate to the full company pipeline or handle directly.
///
/// Heuristics:
/// - Very short requests (< 10 chars) are never complex.
/// - Negative keywords (simple/trivial/typo/quick) skip the pipeline.
/// - Positive keywords (refactor/api/implement/architecture) trigger it.
/// - Multi-line or multi-sentence requests are more likely complex.
pub fn is_complex_request(request: &str) -> bool { pub fn is_complex_request(request: &str) -> bool {
let trimmed = request.trim();
// Very short requests are never complex
if trimmed.len() < 10 {
return false;
}
// Single-line simple update patterns
let lower = trimmed.to_lowercase();
let negative_keywords = [
"simple", "trivial", "typo", "just a", "only a", "minor",
"quick", "tiny", "small fix", "rename", "nitpick",
"cosmetic", "formatting", "spelling", "grammar",
"bump", "version bump", "update comment",
];
if negative_keywords.iter().any(|k| lower.contains(k)) {
return false;
}
// Multi-line/multi-sentence → likely complex
let sentences = trimmed.split(|c| c == '.' || c == '!' || c == '?')
.filter(|s| !s.trim().is_empty())
.count();
if sentences >= 3 {
return true;
}
// Positive complexity keywords
let complexity_keywords = [ let complexity_keywords = [
"refactor", "redesign", "architecture", "feature", "implement", "refactor", "redesign", "architecture", "feature", "implement",
"migrate", "restructure", "rewrite", "new module", "new component", "migrate", "restructure", "rewrite", "new module", "new component",
"scaffold", "multi", "multiple files", "api", "endpoint", "scaffold", "multi", "multiple files", "api", "endpoint",
"integration", "system", "workflow", "pipeline", "database", "integration", "system", "workflow", "pipeline", "database",
"authentication", "authorization", "authentication", "authorization", "full stack",
]; ];
let lower = request.to_lowercase();
complexity_keywords.iter().any(|k| lower.contains(k)) complexity_keywords.iter().any(|k| lower.contains(k))
} }
+41 -9
View File
@@ -36,6 +36,9 @@ pub struct AgentStatus {
pub started_at: Option<i64>, pub started_at: Option<i64>,
pub completed_at: Option<i64>, pub completed_at: Option<i64>,
pub error: Option<String>, pub error: Option<String>,
/// Human-readable progress message (e.g. "editing src/main.rs",
/// "running cargo test"). Shown in the TUI panel alongside the state.
pub progress: Option<String>,
} }
/// A single agent tracked within a workflow run. /// A single agent tracked within a workflow run.
@@ -124,6 +127,7 @@ fn spawn_single_agent(
started_at: Some(started_at), started_at: Some(started_at),
completed_at: None, completed_at: None,
error: None, error: None,
progress: None,
}, },
); );
} }
@@ -156,26 +160,52 @@ fn spawn_single_agent(
// long-running agents. No abort mechanism is wired yet at this level; // long-running agents. No abort mechanism is wired yet at this level;
// future work can expose a kill-switch per agent via the live callback. // future work can expose a kill-switch per agent via the live callback.
// Create an mpsc channel and drain events in a background thread so // Create an mpsc channel and drain events in a background thread.
// run_subagent's blocking_send never blocks (previously the _rx was // The drain thread also pushes intra-division progress updates to the
// dropped immediately, which would cause blocking_send to panic/fail // live callback (current tool being executed), so the TUI panel shows
// on a closed channel). // real-time "editing X" or "running build" instead of just "Running…".
let (tx, rx) = tokio::sync::mpsc::channel(64); let (tx, rx) = tokio::sync::mpsc::channel(64);
let drain_agent_id = agent_id.to_string();
let drain_agent_name = agent_name.to_string();
let drain_live = live.cloned();
let drain_started_at = started_at;
let _drain_thread = std::thread::spawn(move || { let _drain_thread = std::thread::spawn(move || {
// Drain all events so run_subagent's blocking_send never blocks.
// Individual SubagentEvent items are not surfaced to the TUI —
// the live state callbacks above handle coarse-grained Running /
// Completed / Failed status. ToolCall / ToolResult / StepCompleted
// events are traced at debug level for observability.
use crate::app::subagent::event::SubagentEvent; use crate::app::subagent::event::SubagentEvent;
let mut rx = rx; let mut rx = rx;
while let Some(event) = rx.blocking_recv() { while let Some(event) = rx.blocking_recv() {
match &event { match &event {
SubagentEvent::ToolCall { _tool, _args } => { SubagentEvent::ToolCall { _tool, _args } => {
tracing::debug!("[subagent] tool call: {}", _tool); tracing::debug!("[subagent] tool call: {}", _tool);
// Push intra-division progress: which tool is running
if let Some(ref f) = drain_live {
f(
drain_agent_id.clone(),
drain_agent_name.clone(),
AgentStatus {
state: AgentState::Running,
started_at: Some(drain_started_at),
completed_at: None,
error: None,
progress: Some(format!("tool: {}", _tool)),
},
);
}
} }
SubagentEvent::ToolResult { _tool, .. } => { SubagentEvent::ToolResult { _tool, .. } => {
tracing::debug!("[subagent] tool result: {}", _tool); tracing::debug!("[subagent] tool result: {}", _tool);
if let Some(ref f) = drain_live {
f(
drain_agent_id.clone(),
drain_agent_name.clone(),
AgentStatus {
state: AgentState::Running,
started_at: Some(drain_started_at),
completed_at: None,
error: None,
progress: Some(format!("done: {}", _tool)),
},
);
}
} }
SubagentEvent::StepCompleted { _step, .. } => { SubagentEvent::StepCompleted { _step, .. } => {
tracing::trace!("[subagent] step {} completed", _step); tracing::trace!("[subagent] step {} completed", _step);
@@ -226,6 +256,7 @@ fn spawn_single_agent(
started_at: Some(started_at), started_at: Some(started_at),
completed_at: Some(completed_at), completed_at: Some(completed_at),
error: None, error: None,
progress: None,
}, },
), ),
Err(e) => f( Err(e) => f(
@@ -236,6 +267,7 @@ fn spawn_single_agent(
started_at: Some(started_at), started_at: Some(started_at),
completed_at: Some(completed_at), completed_at: Some(completed_at),
error: Some(e.to_string()), error: Some(e.to_string()),
progress: None,
}, },
), ),
} }
+13
View File
@@ -22,6 +22,10 @@ pub enum Command {
WorkflowRun { WorkflowRun {
script: String, script: String,
}, },
/// /pipeline full|quick|skip
Pipeline {
mode: String,
},
Unknown(String), Unknown(String),
} }
@@ -75,6 +79,15 @@ pub fn parse_command(text: &str) -> Command {
"/workflow" => Command::WorkflowRun { "/workflow" => Command::WorkflowRun {
script: arg1.to_string(), script: arg1.to_string(),
}, },
"/pipeline" if arg1.is_empty() => Command::Pipeline {
mode: "status".to_string(),
},
"/pipeline" if arg1 == "full" || arg1 == "quick" || arg1 == "skip" => {
Command::Pipeline {
mode: arg1.to_string(),
}
}
"/pipeline" => Command::Unknown(format!("/pipeline {} (use: full|quick|skip)", arg1)),
_ => Command::Unknown(cmd.to_string()), _ => Command::Unknown(cmd.to_string()),
} }
} }
+5
View File
@@ -162,6 +162,11 @@ pub fn draw_workflow_panel(frame: &mut Frame, area: Rect, state: &crate::app::st
), ),
if let Some(ref err) = agent.status.error { if let Some(ref err) = agent.status.error {
Span::styled(format!("{}", err), Style::default().fg(Theme::ERROR)) Span::styled(format!("{}", err), Style::default().fg(Theme::ERROR))
} else if let Some(ref prog) = agent.status.progress {
Span::styled(
format!(" ({})", prog),
Style::default().fg(Theme::DIM),
)
} else { } else {
Span::raw("") Span::raw("")
}, },