Skip to content
Merged
Show file tree
Hide file tree
Changes from 8 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1,734 changes: 1,628 additions & 106 deletions workers-dev/Cargo.lock

Large diffs are not rendered by default.

4 changes: 3 additions & 1 deletion workers-dev/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,14 @@ path = "src/main.rs"
[dependencies]
anyhow = "1"
clap = { version = "4", features = ["derive"] }
iii-sdk = "=0.20.0"
crossterm = "0.28"
libc = "0.2"
ratatui = "0.29"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
serde_yaml = "0.9"
tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time", "io-util", "process"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "signal", "time", "io-util", "process", "net"] }
tokio-stream = { version = "0.1", features = ["sync"] }

[dev-dependencies]
Expand Down
16 changes: 12 additions & 4 deletions workers-dev/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,8 @@ Workers are **discovered automatically** from top-level `*/iii.worker.yaml` in t

Only **Rust `deploy: binary`** workers can be started with `cargo run`. Node/bundle workers show as `(iii worker add)` — install them via the iii registry instead.

A worker connected to the engine but not started by this `workers-dev` shows **Process: elsewhere** beside **Engine: connected** (`workers-dev` only tracks processes it spawned itself). Non-Rust workers installed via `iii worker add` show **Process: external**.

Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
## Usage

```bash
Expand Down Expand Up @@ -65,16 +67,22 @@ Use `--color never` or `NO_COLOR=1` to force plain output. Default `--color auto

| Key | Action |
|-----|--------|
| `↑`/`↓` | Select worker (skips group headers) |
| `↑`/`↓` (or `k`/`j`) | Select worker (skips group headers) |
| `s` | Start selected worker |
| `x` | Stop selected worker |
| `r` | Restart selected worker + dependents |
| `l` | Follow logs |
| `r` | Restart selected worker + dependents (confirm shows the blast radius) |
| `f` | Toggle live-follow of the selected worker's logs |
| `PgUp`/`PgDn` | Scroll the log pane (pauses follow; resumes at the bottom) |
| `+`/`-` | Resize the log pane (drags the divider in two columns, the height when stacked) |
| `/` | Filter workers by name (Enter applies, Esc clears) |
| `Ctrl+u` | Start harness stack |
| `Ctrl+a` | Start all managed Rust workers |
| `?` | Toggle the key-reference overlay |
| `q` | Quit |

The log pane shows the last 20 lines for the **selected worker only** (cargo build output + worker logs). Lines are sanitized (no ANSI, no `\r` overwrite garbage).
On a wide terminal the dashboard is a two-column **master/detail** layout: the worker list on the left (sized to fit its columns), the selected worker's logs filling the rest on the right, with `+`/`-` dragging the divider between them. Below ~100 columns the two panes stack vertically instead, and `+`/`-` trade height.

The header shows an at-a-glance health summary (`●` connected, `◐` compiling, `✗` crashed, `○` stopped) and flags the engine as `⚠ unreachable` when a status query fails. The log pane shows the **selected worker only**, scrollable through the full ring buffer, following the live tail by default. Crashed workers show their exit code inline. Lines are sanitized (no ANSI, no `\r` overwrite garbage).

## Config (`workers-dev.yaml`)

Expand Down
22 changes: 11 additions & 11 deletions workers-dev/src/commands/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,10 @@ use crate::orchestrator::Orchestrator;
use crate::status;

pub async fn run_status(orchestrator: &Orchestrator) -> Result<()> {
let views = orchestrator.worker_views().await?;
let (views, engine_error) = orchestrator.dashboard_snapshot().await;
if let Some(err) = engine_error {
eprintln!("warning: engine unreachable: {err}");
}
status::print_status_table(&views);
Ok(())
}
Expand Down Expand Up @@ -49,20 +52,15 @@ pub async fn run_logs(
follow: bool,
lines: usize,
) -> Result<()> {
if !follow {
let tail = orchestrator.logs_tail(&worker, lines).await?;
let color_enabled = orchestrator.config.color_mode.enabled_for_stdout();
for line in tail {
logs::print_colored_line(&line, color_enabled, &mut io::stdout())?;
}
return Ok(());
}

let tail = orchestrator.logs_tail(&worker, lines).await?;
let color_enabled = orchestrator.config.color_mode.enabled_for_stdout();
for line in tail {
logs::print_colored_line(&line, color_enabled, &mut io::stdout())?;
}

if !follow {
return Ok(());
}
io::stdout().flush()?;

let mut rx = orchestrator.subscribe_logs(&worker).await?;
Expand All @@ -76,7 +74,8 @@ pub async fn run_logs(
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
let proc = orchestrator.worker_views().await?;
match proc.iter().find(|v| v.name == worker) {
Some(v) if v.process_status == "running" || v.process_status == "compiling" => {}
Some(v) if v.process_status == "running" || v.process_status == "compiling" => {
}
_ => break,
}
tokio::time::sleep(std::time::Duration::from_millis(200)).await;
Expand All @@ -88,6 +87,7 @@ pub async fn run_logs(
}

pub async fn run_up(orchestrator: Arc<Orchestrator>) -> Result<()> {
orchestrator.ensure_engine().await?;
orchestrator.start_harness_stack(false).await?;
crate::tui::run(orchestrator).await
}
32 changes: 20 additions & 12 deletions workers-dev/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,17 @@ use crate::discover::{discover_repo_workers, harness_stack_names, order_worker_n
pub const DEFAULT_ENGINE_URL: &str = "ws://127.0.0.1:49134";
pub const DEFAULT_POLL_INTERVAL_MS: u64 = 2000;
pub const DEFAULT_CONNECT_TIMEOUT_MS: u64 = 120_000;
/// Per-query timeout for `engine::workers::list`. Kept short so a slow or
/// hung engine can't stall the status poll (and, with off-thread polling,
/// can't freeze the TUI).
pub const ENGINE_QUERY_TIMEOUT_MS: u64 = 3_000;
pub const LOG_RING_CAPACITY: usize = 500;
pub const DEFAULT_LOG_TAIL: usize = 100;
pub const DASHBOARD_LOG_LINES: usize = 20;

#[derive(Debug, Clone)]
pub struct Config {
pub repo_root: PathBuf,
pub engine_url: String,
pub engine_host: String,
pub engine_port: u16,
pub release: bool,
pub poll_interval_ms: u64,
pub connect_timeout_ms: u64,
Expand Down Expand Up @@ -67,10 +68,14 @@ impl Config {
bail!("no workers discovered under {}", repo_root.display());
}

let engine_url = engine_url
let raw_engine_url = engine_url
.or(file_cfg.engine_url)
.unwrap_or_else(|| DEFAULT_ENGINE_URL.to_string());
let (engine_host, engine_port) = parse_engine_url(&engine_url, port)?;
let (engine_host, engine_port) = parse_engine_url(&raw_engine_url, port)?;
// Canonicalize so a `--port` override is reflected in the URL the SDK
// actually connects to. The engine WS endpoint is always `/`, so any
// path in the original URL is intentionally dropped.
let engine_url = format!("ws://{engine_host}:{engine_port}");
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved

let discovered_names = order_worker_names(&worker_specs);
let workers = file_cfg.workers.unwrap_or(discovered_names);
Expand All @@ -83,7 +88,9 @@ impl Config {
.as_deref()
.and_then(|s| {
ColorMode::parse(s).or_else(|| {
eprintln!("warning: invalid color mode {s:?}; using auto");
eprintln!(
"warning: invalid color mode {s:?} (valid: auto, always, never); using auto"
);
None
})
})
Expand All @@ -92,8 +99,6 @@ impl Config {
Ok(Self {
repo_root,
engine_url,
engine_host,
engine_port,
release: release || file_cfg.release.unwrap_or(false),
poll_interval_ms: file_cfg
.poll_interval_ms
Expand Down Expand Up @@ -168,7 +173,10 @@ fn validate_repo_root(path: &Path) -> Result<()> {
if count > 0 {
Ok(())
} else {
bail!("no top-level iii.worker.yaml workers under {}", path.display())
bail!(
"no top-level iii.worker.yaml workers under {}",
path.display()
)
}
}

Expand All @@ -181,9 +189,9 @@ pub fn parse_engine_url(url: &str, port_override: Option<u16>) -> Result<(String
let authority = stripped.split('/').next().unwrap_or(stripped);

let (host, port) = if let Some((host, port_str)) = authority.rsplit_once(':') {
let port: u16 = port_str
.parse()
.with_context(|| format!("invalid port in engine url {url}"))?;
let port: u16 = port_str.parse().with_context(|| {
format!("invalid port in engine url {url} (port must be a number, e.g. 49134)")
})?;
(host.to_string(), port)
} else {
(authority.to_string(), 49134)
Expand Down
30 changes: 12 additions & 18 deletions workers-dev/src/discover.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ pub const HARNESS_STACK: &[&str] = &[
"harness",
];

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum WorkerGroup {
HarnessStack,
Other,
Expand Down Expand Up @@ -72,13 +72,14 @@ pub fn discover_repo_workers(repo_root: &Path) -> Result<Vec<WorkerSpec>> {

let raw = fs::read_to_string(&yaml_path)
.with_context(|| format!("read {}", yaml_path.display()))?;
let parsed: WorkerYaml = serde_yaml::from_str(&raw)
.with_context(|| format!("parse {}", yaml_path.display()))?;
let parsed: WorkerYaml =
serde_yaml::from_str(&raw).with_context(|| format!("parse {}", yaml_path.display()))?;
let name = parsed.name.clone().unwrap_or(folder.clone());
if name != folder {
anyhow::bail!(
"worker folder {folder} has iii.worker.yaml name={name} (mismatch)"
eprintln!(
"warning: skipping worker folder {folder}: iii.worker.yaml name={name} (mismatch)"
);
continue;
Comment thread
andersonleal marked this conversation as resolved.
Resolved
}

let group = if HARNESS_STACK.contains(&name.as_str()) {
Expand All @@ -96,21 +97,10 @@ pub fn discover_repo_workers(repo_root: &Path) -> Result<Vec<WorkerSpec>> {
});
}

specs.sort_by(|a, b| {
group_rank(a.group)
.cmp(&group_rank(b.group))
.then_with(|| a.name.cmp(&b.name))
});
specs.sort_by(|a, b| a.group.cmp(&b.group).then_with(|| a.name.cmp(&b.name)));
Ok(specs)
}

fn group_rank(group: WorkerGroup) -> u8 {
match group {
WorkerGroup::HarnessStack => 0,
WorkerGroup::Other => 1,
}
}

fn classify_spawn(dir: &Path, yaml: &WorkerYaml) -> SpawnKind {
let language = yaml.language.as_deref().unwrap_or("");
let deploy = yaml.deploy.as_deref().unwrap_or("");
Expand Down Expand Up @@ -175,7 +165,11 @@ mod tests {
assert_eq!(specs[2].name, "telegram-bot");
assert_eq!(specs[2].group, WorkerGroup::Other);
assert!(matches!(
specs.iter().find(|s| s.name == "claude-code").unwrap().spawn,
specs
.iter()
.find(|s| s.name == "claude-code")
.unwrap()
.spawn,
SpawnKind::Unsupported { .. }
));
}
Expand Down
34 changes: 15 additions & 19 deletions workers-dev/src/graph.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,9 +30,7 @@ impl WorkerGraph {

if let Some(name) = &parsed.name {
if name != worker {
bail!(
"worker folder {worker} has iii.worker.yaml name={name} (mismatch)"
);
bail!("worker folder {worker} has iii.worker.yaml name={name} (mismatch)");
}
}

Expand All @@ -57,10 +55,7 @@ impl WorkerGraph {
}

pub fn dependencies(&self, worker: &str) -> &[String] {
self.deps
.get(worker)
.map(Vec::as_slice)
.unwrap_or(&[])
self.deps.get(worker).map(Vec::as_slice).unwrap_or(&[])
}

/// Topological start order (dependencies before dependents).
Expand All @@ -86,9 +81,7 @@ impl WorkerGraph {
continue;
}
*in_degree.entry(worker.as_str()).or_insert(0) += 1;
adj.entry(dep.as_str())
.or_default()
.push(worker.as_str());
adj.entry(dep.as_str()).or_default().push(worker.as_str());
}
}

Expand Down Expand Up @@ -139,8 +132,7 @@ impl WorkerGraph {

while let Some(current) = queue.pop_front() {
for candidate in &self.workers {
if self.dependencies(candidate).contains(&current)
&& seen.insert(candidate.clone())
if self.dependencies(candidate).contains(&current) && seen.insert(candidate.clone())
{
queue.push_back(candidate.clone());
}
Expand Down Expand Up @@ -216,13 +208,17 @@ mod tests {
write_worker_yaml(tmp.path(), "provider-anthropic", &["llm-router"]);
write_worker_yaml(tmp.path(), "provider-openai", &["llm-router"]);
write_worker_yaml(tmp.path(), "approval-gate", &["session-manager"]);
write_worker_yaml(tmp.path(), "harness", &[
"session-manager",
"context-manager",
"approval-gate",
"provider-anthropic",
"provider-openai",
]);
write_worker_yaml(
tmp.path(),
"harness",
&[
"session-manager",
"context-manager",
"approval-gate",
"provider-anthropic",
"provider-openai",
],
);
tmp
}

Expand Down
Loading
Loading