diff --git a/crates/cm-api/src/lib.rs b/crates/cm-api/src/lib.rs index fcdfc25..29d60e7 100644 --- a/crates/cm-api/src/lib.rs +++ b/crates/cm-api/src/lib.rs @@ -497,6 +497,10 @@ pub fn router(state: AppState) -> Router { "/api/topology-runs/{id}/events", get(routes::topology::run_events_sse), ) + .route( + "/api/topology-runs/{id}/container-log", + get(routes::topology::run_container_log_sse), + ) .route( "/api/topology-runs/{id}/cancel", post(routes::topology::cancel_run), diff --git a/crates/cm-api/src/routes/topology.rs b/crates/cm-api/src/routes/topology.rs index 82c8c53..d3d2da3 100644 --- a/crates/cm-api/src/routes/topology.rs +++ b/crates/cm-api/src/routes/topology.rs @@ -388,3 +388,169 @@ pub async fn get_run( checkpoint: run.checkpoint, })) } + +// ── Phase: live container log tail ───────────────────────────────── + +/// Strip ANSI escape sequences from a line so the browser terminal +/// renders it cleanly. Cheap and allocation-only when a match hits. +fn strip_ansi(input: &str) -> String { + let mut out = String::with_capacity(input.len()); + let bytes = input.as_bytes(); + let mut i = 0; + while i < bytes.len() { + if bytes[i] == 0x1b && i + 1 < bytes.len() && bytes[i + 1] == b'[' { + // Skip until final byte in @-~ range. + i += 2; + while i < bytes.len() && !(bytes[i] >= 0x40 && bytes[i] <= 0x7e) { + i += 1; + } + i += 1; + } else { + out.push(bytes[i] as char); + i += 1; + } + } + out +} + +/// Squeeze a zeroclaw daemon log line into `[bracket] action outcome +/// · trailing message`. Falls back to the ANSI-stripped raw line when +/// the shape isn't recognised so we never lose an interesting line. +fn compact_container_log(line: &str) -> Option { + let stripped = strip_ansi(line); + let trimmed = stripped.trim_end(); + if trimmed.is_empty() { + return None; + } + // Drop pure framing noise: `zeroclaw_scope{...}` continuations + // that carry no zc_action. + let has_action = trimmed.contains("zc_action="); + if !has_action { + // Non-daemon lines (bash echoes, container startup banners, + // panic backtraces) — keep as-is; those are useful too. + if trimmed.contains("zc_") { + return None; // structural framing without action, drop + } + return Some(trimmed.to_string()); + } + let bracket = trimmed + .split_once(']') + .and_then(|(before, _)| before.strip_prefix('[')) + .unwrap_or(""); + let action = trimmed + .split("zc_action=") + .nth(1) + .and_then(|s| s.split_whitespace().next()) + .unwrap_or("?"); + let outcome = trimmed + .split("zc_outcome=") + .nth(1) + .and_then(|s| s.split_whitespace().next()) + .unwrap_or(""); + let msg = trimmed + .rsplit(':') + .next() + .map(str::trim) + .unwrap_or("") + .to_string(); + let tag = if bracket.is_empty() { + "system" + } else { + bracket + }; + Some(if outcome.is_empty() || outcome == "unknown" { + format!("[{tag}] {action} · {msg}") + } else { + format!("[{tag}] {action} ({outcome}) · {msg}") + }) +} + +/// `GET /api/topology-runs/{id}/container-log` — SSE stream of the +/// per-topic team container's daemon log, filtered from the ZeroClaw +/// structural noise into `[actor] action (outcome) · message` lines. +/// Emits a `line` event per surviving line, plus periodic keep-alives. +/// Ends when the container's log stream closes or the client +/// disconnects. Auth: workspace-scoped like `run_events_sse`. +pub async fn run_container_log_sse( + State(state): State, + Authed(user): Authed, + Path(id): Path, +) -> impl IntoResponse { + // All early exits + the live tail funnel through one stream! so + // Sse::new sees a single concrete stream type. + let pool = state.pool.clone(); + let ws = user.workspace_id; + let stream = async_stream::stream! { + use futures::StreamExt; + // 1) Workspace scope + topic id. + let topic_id = match cm_db::repo::topology_runs::status(&pool, id, ws).await { + Ok(_) => match cm_db::repo::topology_runs::research_topic_id(&pool, id).await { + Ok(Some(t)) => t, + _ => { + yield Ok::( + Event::default().event("error").data( + "run has no bound research topic; container log unavailable", + ), + ); + return; + } + }, + Err(_) => { + yield Ok::( + Event::default().event("error").data("run not found"), + ); + return; + } + }; + + // 2) Docker handle. + let container = crate::research_container::container_name_for(topic_id); + let docker = match crate::research_container::connect() { + Ok(d) => d, + Err(e) => { + yield Ok(Event::default() + .event("error") + .data(format!("docker connect failed: {e}"))); + return; + } + }; + + // 3) Tail. + let opts = bollard::query_parameters::LogsOptionsBuilder::default() + .stdout(true) + .stderr(true) + .follow(true) + .tail("200") + .timestamps(false) + .build(); + yield Ok(Event::default() + .event("info") + .data(format!("tailing {container}"))); + let mut log_stream = docker.logs(&container, Some(opts)); + // Line-accumulator so partial chunks don't truncate a log line. + let mut buf = String::new(); + while let Some(chunk) = log_stream.next().await { + let bytes = match chunk { + Ok(bollard::container::LogOutput::StdOut { message }) + | Ok(bollard::container::LogOutput::StdErr { message }) + | Ok(bollard::container::LogOutput::Console { message }) => message, + Ok(_) => continue, + Err(e) => { + yield Ok(Event::default().event("error").data(e.to_string())); + break; + } + }; + let s = String::from_utf8_lossy(&bytes); + buf.push_str(&s); + while let Some(nl) = buf.find('\n') { + let line: String = buf.drain(..=nl).collect(); + if let Some(compact) = compact_container_log(&line) { + yield Ok(Event::default().event("line").data(compact)); + } + } + } + yield Ok(Event::default().event("done").data("stream closed")); + }; + + Sse::new(stream).keep_alive(KeepAlive::default()) +} diff --git a/frontend/src/components/dashboard/LiveRunLogs.tsx b/frontend/src/components/dashboard/LiveRunLogs.tsx index 8605f9a..e1fbb69 100644 --- a/frontend/src/components/dashboard/LiveRunLogs.tsx +++ b/frontend/src/components/dashboard/LiveRunLogs.tsx @@ -23,6 +23,42 @@ type SseEvent = | { kind: "step"; index: number; ts: number; raw: string } | { kind: "done"; ts: number; raw: string; error?: string | null }; +type ContainerLine = { ts: number; kind: "info" | "line" | "error" | "done"; text: string }; + +/** Subscribe to the run's team container log SSE. `active` gates the + * subscription — the Container tab opens/closes the stream. */ +function useContainerLog(runId: string | null, active: boolean) { + const [lines, setLines] = useState([]); + const [status, setStatus] = useState<"idle" | "connecting" | "streaming" | "done" | "error">( + "idle", + ); + const [prevKey, setPrevKey] = useState(""); + const key = `${runId ?? ""}:${active}`; + if (prevKey !== key) { + setPrevKey(key); + setLines([]); + setStatus(runId && active ? "connecting" : "idle"); + } + useEffect(() => { + if (!runId || !active) return; + const es = new EventSource( + `/api/topology-runs/${encodeURIComponent(runId)}/container-log`, + ); + es.onopen = () => setStatus("streaming"); + for (const kind of ["info", "line", "error", "done"] as const) { + es.addEventListener(kind, (e: MessageEvent) => { + const ts = Date.now(); + setLines((prev) => [...prev, { ts, kind, text: e.data as string }]); + if (kind === "done") setStatus("done"); + if (kind === "error") setStatus("error"); + }); + } + es.onerror = () => setStatus((prev) => (prev === "done" ? prev : "error")); + return () => es.close(); + }, [runId, active]); + return { lines, status }; +} + function useRunEvents(runId: string | null, onStep?: (p: StepPulse) => void) { const [events, setEvents] = useState([]); const [status, setStatus] = useState<"connecting" | "streaming" | "done" | "error">( @@ -220,6 +256,9 @@ export function LiveRunLogs({ }, [topicId]); const { events, status } = useRunEvents(activeRun, onStep); + const [tab, setTab] = useState<"steps" | "container">("steps"); + const { lines: containerLines, status: containerStatus } = + useContainerLog(activeRun, tab === "container"); // Autoscroll to newest event when pinned to the bottom. useEffect(() => { @@ -312,6 +351,45 @@ export function LiveRunLogs({ ) : null} + {/* Sub-tabs: Steps (topology step summaries) vs Container + (live daemon log tail from the team runtime). */} +
+ {( + [ + { k: "steps" as const, label: "Steps", count: events.length, s: status }, + { + k: "container" as const, + label: "Container", + count: containerLines.length, + s: containerStatus, + }, + ] + ).map((t) => { + const isActive = tab === t.k; + return ( + + ); + })} +
+
{ @@ -337,45 +415,81 @@ export function LiveRunLogs({ {/* Pre-step setup phases from pipeline-state. Rendered until real step events arrive, then hidden so the terminal doesn't scroll past the actual step timeline. */} - {events.length === 0 && pipeline ? ( + {tab === "steps" ? ( <> -
- {status === "connecting" - ? "connecting…" - : "worker booting — steps journal after each turn completes"} -
- {pipeline.stages.map((s) => ( -
+ {events.length === 0 && pipeline ? ( + <> +
+ {status === "connecting" + ? "connecting…" + : "worker booting — steps journal after each turn completes"} +
+ {pipeline.stages.map((s) => ( +
+ + setup {STAGE_DOT[s.status] ?? "◯"} {s.key.padEnd(9)} + {" "} + {s.label} + {s.detail ? ( + — {s.detail} + ) : null} +
+ ))} + + ) : null} + {events.length === 0 && !pipeline ? ( +
+ {status === "connecting" ? "connecting…" : "waiting for pipeline state…"} +
+ ) : null} + {events.map((e, i) => ( +
- setup {STAGE_DOT[s.status] ?? "◯"} {s.key.padEnd(9)} + {new Date(e.ts).toLocaleTimeString()}{" "} + {e.kind === "step" ? `#${e.index}` : "done"} {" "} - {s.label} - {s.detail ? ( - — {s.detail} - ) : null} + {summarizeStep(e.raw)}
))} - ) : null} - {events.length === 0 && !pipeline ? ( -
- {status === "connecting" ? "connecting…" : "waiting for pipeline state…"} -
- ) : null} - {events.map((e, i) => ( -
- - {new Date(e.ts).toLocaleTimeString()}{" "} - {e.kind === "step" ? `#${e.index}` : "done"} - {" "} - {summarizeStep(e.raw)} -
- ))} + ) : ( + <> + {containerLines.length === 0 ? ( +
+ {containerStatus === "connecting" + ? "connecting to container log…" + : containerStatus === "streaming" + ? "waiting for first line…" + : containerStatus === "error" + ? "container log stream error" + : "idle"} +
+ ) : null} + {containerLines.map((l, i) => ( +
+ + {new Date(l.ts).toLocaleTimeString()} + {" "} + {l.text} +
+ ))} + + )}
{!pinned ? (