live-run-logs: add Container tab that tails filtered daemon logs
Steps summaries only fire AFTER each topology step completes — so a
stalled first turn was completely dark. Add a second tab that
streams the team runtime container's daemon log live via a new SSE
endpoint.
Backend
- GET /api/topology-runs/:id/container-log — workspace-scoped SSE
around bollard's docker.logs(follow=true, tail=200). Buffers on
newline so partial mux chunks don't truncate a log line.
- compact_container_log: parse a zeroclaw daemon line
('[actor] ... zc_action=X zc_outcome=Y ... msg') into
'[actor] action (outcome) · msg'. Framing-only continuations are
dropped; non-zc lines (bash echoes, backtraces) pass through as-is
so nothing interesting is lost. ANSI escapes stripped.
- Everything funnels through one async_stream! so early exits
(workspace check / docker connect / no bound topic) yield an
'error' event and return without breaking Sse::new's single stream
type.
Frontend
- LiveRunLogs gets a sub-tabs strip: Steps · Container.
- New useContainerLog(runId, active) hook — gated by tab so we don't
hold two open SSE streams when the operator isn't looking.
- Same terminal widget renders each container line with a level
color (info/done grey, line default, error red). Sub-tab pill
shows count + status live.
This commit is contained in:
@@ -497,6 +497,10 @@ pub fn router(state: AppState) -> Router {
|
|||||||
"/api/topology-runs/{id}/events",
|
"/api/topology-runs/{id}/events",
|
||||||
get(routes::topology::run_events_sse),
|
get(routes::topology::run_events_sse),
|
||||||
)
|
)
|
||||||
|
.route(
|
||||||
|
"/api/topology-runs/{id}/container-log",
|
||||||
|
get(routes::topology::run_container_log_sse),
|
||||||
|
)
|
||||||
.route(
|
.route(
|
||||||
"/api/topology-runs/{id}/cancel",
|
"/api/topology-runs/{id}/cancel",
|
||||||
post(routes::topology::cancel_run),
|
post(routes::topology::cancel_run),
|
||||||
|
|||||||
@@ -388,3 +388,169 @@ pub async fn get_run(
|
|||||||
checkpoint: run.checkpoint,
|
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<String> {
|
||||||
|
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<AppState>,
|
||||||
|
Authed(user): Authed,
|
||||||
|
Path(id): Path<Uuid>,
|
||||||
|
) -> 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, Infallible>(
|
||||||
|
Event::default().event("error").data(
|
||||||
|
"run has no bound research topic; container log unavailable",
|
||||||
|
),
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
Err(_) => {
|
||||||
|
yield Ok::<Event, Infallible>(
|
||||||
|
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())
|
||||||
|
}
|
||||||
|
|||||||
@@ -23,6 +23,42 @@ type SseEvent =
|
|||||||
| { kind: "step"; index: number; ts: number; raw: string }
|
| { kind: "step"; index: number; ts: number; raw: string }
|
||||||
| { kind: "done"; ts: number; raw: string; error?: string | null };
|
| { 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<ContainerLine[]>([]);
|
||||||
|
const [status, setStatus] = useState<"idle" | "connecting" | "streaming" | "done" | "error">(
|
||||||
|
"idle",
|
||||||
|
);
|
||||||
|
const [prevKey, setPrevKey] = useState<string>("");
|
||||||
|
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) {
|
function useRunEvents(runId: string | null, onStep?: (p: StepPulse) => void) {
|
||||||
const [events, setEvents] = useState<SseEvent[]>([]);
|
const [events, setEvents] = useState<SseEvent[]>([]);
|
||||||
const [status, setStatus] = useState<"connecting" | "streaming" | "done" | "error">(
|
const [status, setStatus] = useState<"connecting" | "streaming" | "done" | "error">(
|
||||||
@@ -220,6 +256,9 @@ export function LiveRunLogs({
|
|||||||
}, [topicId]);
|
}, [topicId]);
|
||||||
|
|
||||||
const { events, status } = useRunEvents(activeRun, onStep);
|
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.
|
// Autoscroll to newest event when pinned to the bottom.
|
||||||
useEffect(() => {
|
useEffect(() => {
|
||||||
@@ -312,6 +351,45 @@ export function LiveRunLogs({
|
|||||||
</div>
|
</div>
|
||||||
) : null}
|
) : null}
|
||||||
|
|
||||||
|
{/* Sub-tabs: Steps (topology step summaries) vs Container
|
||||||
|
(live daemon log tail from the team runtime). */}
|
||||||
|
<div style={{ display: "flex", gap: 6 }}>
|
||||||
|
{(
|
||||||
|
[
|
||||||
|
{ 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 (
|
||||||
|
<button
|
||||||
|
key={t.k}
|
||||||
|
type="button"
|
||||||
|
onClick={() => setTab(t.k)}
|
||||||
|
style={{
|
||||||
|
padding: "4px 10px",
|
||||||
|
borderRadius: 999,
|
||||||
|
fontFamily: mono,
|
||||||
|
fontSize: 10,
|
||||||
|
background: isActive ? "rgba(94,200,216,.15)" : "transparent",
|
||||||
|
border: `1px solid ${
|
||||||
|
isActive ? "rgba(94,200,216,.45)" : "rgba(255,255,255,.1)"
|
||||||
|
}`,
|
||||||
|
color: isActive ? "#e5f6fb" : "#8a8a92",
|
||||||
|
cursor: "pointer",
|
||||||
|
}}
|
||||||
|
>
|
||||||
|
{t.label} · {t.count} · {t.s}
|
||||||
|
</button>
|
||||||
|
);
|
||||||
|
})}
|
||||||
|
</div>
|
||||||
|
|
||||||
<div
|
<div
|
||||||
ref={scrollRef}
|
ref={scrollRef}
|
||||||
onScroll={(e) => {
|
onScroll={(e) => {
|
||||||
@@ -337,45 +415,81 @@ export function LiveRunLogs({
|
|||||||
{/* Pre-step setup phases from pipeline-state. Rendered until
|
{/* Pre-step setup phases from pipeline-state. Rendered until
|
||||||
real step events arrive, then hidden so the terminal
|
real step events arrive, then hidden so the terminal
|
||||||
doesn't scroll past the actual step timeline. */}
|
doesn't scroll past the actual step timeline. */}
|
||||||
{events.length === 0 && pipeline ? (
|
{tab === "steps" ? (
|
||||||
<>
|
<>
|
||||||
<div style={{ color: "#6a6a72", marginBottom: 4 }}>
|
{events.length === 0 && pipeline ? (
|
||||||
{status === "connecting"
|
<>
|
||||||
? "connecting…"
|
<div style={{ color: "#6a6a72", marginBottom: 4 }}>
|
||||||
: "worker booting — steps journal after each turn completes"}
|
{status === "connecting"
|
||||||
</div>
|
? "connecting…"
|
||||||
{pipeline.stages.map((s) => (
|
: "worker booting — steps journal after each turn completes"}
|
||||||
<div key={s.key} style={{ color: "#cfcfd5" }}>
|
</div>
|
||||||
|
{pipeline.stages.map((s) => (
|
||||||
|
<div key={s.key} style={{ color: "#cfcfd5" }}>
|
||||||
|
<span style={{ color: "#5a5a62" }}>
|
||||||
|
setup {STAGE_DOT[s.status] ?? "◯"} {s.key.padEnd(9)}
|
||||||
|
</span>{" "}
|
||||||
|
{s.label}
|
||||||
|
{s.detail ? (
|
||||||
|
<span style={{ color: "#ffb0a5" }}> — {s.detail}</span>
|
||||||
|
) : null}
|
||||||
|
</div>
|
||||||
|
))}
|
||||||
|
</>
|
||||||
|
) : null}
|
||||||
|
{events.length === 0 && !pipeline ? (
|
||||||
|
<div style={{ color: "#6a6a72" }}>
|
||||||
|
{status === "connecting" ? "connecting…" : "waiting for pipeline state…"}
|
||||||
|
</div>
|
||||||
|
) : null}
|
||||||
|
{events.map((e, i) => (
|
||||||
|
<div
|
||||||
|
key={`${i}-${e.ts}`}
|
||||||
|
style={{
|
||||||
|
color: e.kind === "done" && e.error ? "#ff8a7a" : "#cfcfd5",
|
||||||
|
}}
|
||||||
|
>
|
||||||
<span style={{ color: "#5a5a62" }}>
|
<span style={{ color: "#5a5a62" }}>
|
||||||
setup {STAGE_DOT[s.status] ?? "◯"} {s.key.padEnd(9)}
|
{new Date(e.ts).toLocaleTimeString()}{" "}
|
||||||
|
{e.kind === "step" ? `#${e.index}` : "done"}
|
||||||
</span>{" "}
|
</span>{" "}
|
||||||
{s.label}
|
{summarizeStep(e.raw)}
|
||||||
{s.detail ? (
|
|
||||||
<span style={{ color: "#ffb0a5" }}> — {s.detail}</span>
|
|
||||||
) : null}
|
|
||||||
</div>
|
</div>
|
||||||
))}
|
))}
|
||||||
</>
|
</>
|
||||||
) : null}
|
) : (
|
||||||
{events.length === 0 && !pipeline ? (
|
<>
|
||||||
<div style={{ color: "#6a6a72" }}>
|
{containerLines.length === 0 ? (
|
||||||
{status === "connecting" ? "connecting…" : "waiting for pipeline state…"}
|
<div style={{ color: "#6a6a72" }}>
|
||||||
</div>
|
{containerStatus === "connecting"
|
||||||
) : null}
|
? "connecting to container log…"
|
||||||
{events.map((e, i) => (
|
: containerStatus === "streaming"
|
||||||
<div
|
? "waiting for first line…"
|
||||||
key={`${i}-${e.ts}`}
|
: containerStatus === "error"
|
||||||
style={{
|
? "container log stream error"
|
||||||
color: e.kind === "done" && e.error ? "#ff8a7a" : "#cfcfd5",
|
: "idle"}
|
||||||
}}
|
</div>
|
||||||
>
|
) : null}
|
||||||
<span style={{ color: "#5a5a62" }}>
|
{containerLines.map((l, i) => (
|
||||||
{new Date(e.ts).toLocaleTimeString()}{" "}
|
<div
|
||||||
{e.kind === "step" ? `#${e.index}` : "done"}
|
key={`${i}-${l.ts}`}
|
||||||
</span>{" "}
|
style={{
|
||||||
{summarizeStep(e.raw)}
|
color:
|
||||||
</div>
|
l.kind === "error"
|
||||||
))}
|
? "#ff8a7a"
|
||||||
|
: l.kind === "info" || l.kind === "done"
|
||||||
|
? "#8a8a92"
|
||||||
|
: "#cfcfd5",
|
||||||
|
}}
|
||||||
|
>
|
||||||
|
<span style={{ color: "#5a5a62" }}>
|
||||||
|
{new Date(l.ts).toLocaleTimeString()}
|
||||||
|
</span>{" "}
|
||||||
|
{l.text}
|
||||||
|
</div>
|
||||||
|
))}
|
||||||
|
</>
|
||||||
|
)}
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
{!pinned ? (
|
{!pinned ? (
|
||||||
|
|||||||
Reference in New Issue
Block a user