Merge pull request #648 from paperclipai/paperclip-nicer-runlogs-formats
Humanize run transcripts and polish transcript UX
This commit is contained in:
Dotta
@@ -1,191 +1,19 @@
|
||||
import { useEffect, useMemo, useRef, useState, type MutableRefObject } from "react";
|
||||
import { useMemo } from "react";
|
||||
import { Link } from "@/lib/router";
|
||||
import { useQuery } from "@tanstack/react-query";
|
||||
import type { Issue, LiveEvent } from "@paperclipai/shared";
|
||||
import type { Issue } from "@paperclipai/shared";
|
||||
import { heartbeatsApi, type LiveRunForIssue } from "../api/heartbeats";
|
||||
import { issuesApi } from "../api/issues";
|
||||
import { getUIAdapter } from "../adapters";
|
||||
import type { TranscriptEntry } from "../adapters";
|
||||
import { queryKeys } from "../lib/queryKeys";
|
||||
import { cn, relativeTime } from "../lib/utils";
|
||||
import { ExternalLink } from "lucide-react";
|
||||
import { Identity } from "./Identity";
|
||||
import { RunTranscriptView } from "./transcript/RunTranscriptView";
|
||||
import { useLiveRunTranscripts } from "./transcript/useLiveRunTranscripts";
|
||||
|
||||
type FeedTone = "info" | "warn" | "error" | "assistant" | "tool";
|
||||
|
||||
interface FeedItem {
|
||||
id: string;
|
||||
ts: string;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
agentName: string;
|
||||
text: string;
|
||||
tone: FeedTone;
|
||||
dedupeKey: string;
|
||||
streamingKind?: "assistant" | "thinking";
|
||||
}
|
||||
|
||||
const MAX_FEED_ITEMS = 40;
|
||||
const MAX_FEED_TEXT_LENGTH = 220;
|
||||
const MAX_STREAMING_TEXT_LENGTH = 4000;
|
||||
const MIN_DASHBOARD_RUNS = 4;
|
||||
|
||||
function readString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
function summarizeEntry(entry: TranscriptEntry): { text: string; tone: FeedTone } | null {
|
||||
if (entry.kind === "assistant") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "assistant" } : null;
|
||||
}
|
||||
if (entry.kind === "thinking") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text: `[thinking] ${text}`, tone: "info" } : null;
|
||||
}
|
||||
if (entry.kind === "tool_call") {
|
||||
return { text: `tool ${entry.name}`, tone: "tool" };
|
||||
}
|
||||
if (entry.kind === "tool_result") {
|
||||
const base = entry.content.trim();
|
||||
return {
|
||||
text: entry.isError ? `tool error: ${base}` : `tool result: ${base}`,
|
||||
tone: entry.isError ? "error" : "tool",
|
||||
};
|
||||
}
|
||||
if (entry.kind === "stderr") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "error" } : null;
|
||||
}
|
||||
if (entry.kind === "system") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "warn" } : null;
|
||||
}
|
||||
if (entry.kind === "stdout") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "info" } : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function createFeedItem(
|
||||
run: LiveRunForIssue,
|
||||
ts: string,
|
||||
text: string,
|
||||
tone: FeedTone,
|
||||
nextId: number,
|
||||
options?: {
|
||||
streamingKind?: "assistant" | "thinking";
|
||||
preserveWhitespace?: boolean;
|
||||
},
|
||||
): FeedItem | null {
|
||||
if (!text.trim()) return null;
|
||||
const base = options?.preserveWhitespace ? text : text.trim();
|
||||
const maxLength = options?.streamingKind ? MAX_STREAMING_TEXT_LENGTH : MAX_FEED_TEXT_LENGTH;
|
||||
const normalized = base.length > maxLength ? base.slice(-maxLength) : base;
|
||||
return {
|
||||
id: `${run.id}:${nextId}`,
|
||||
ts,
|
||||
runId: run.id,
|
||||
agentId: run.agentId,
|
||||
agentName: run.agentName,
|
||||
text: normalized,
|
||||
tone,
|
||||
dedupeKey: `feed:${run.id}:${ts}:${tone}:${normalized}`,
|
||||
streamingKind: options?.streamingKind,
|
||||
};
|
||||
}
|
||||
|
||||
function parseStdoutChunk(
|
||||
run: LiveRunForIssue,
|
||||
chunk: string,
|
||||
ts: string,
|
||||
pendingByRun: Map<string, string>,
|
||||
nextIdRef: MutableRefObject<number>,
|
||||
): FeedItem[] {
|
||||
const pendingKey = `${run.id}:stdout`;
|
||||
const combined = `${pendingByRun.get(pendingKey) ?? ""}${chunk}`;
|
||||
const split = combined.split(/\r?\n/);
|
||||
pendingByRun.set(pendingKey, split.pop() ?? "");
|
||||
const adapter = getUIAdapter(run.adapterType);
|
||||
|
||||
const summarized: Array<{ text: string; tone: FeedTone; streamingKind?: "assistant" | "thinking" }> = [];
|
||||
const appendSummary = (entry: TranscriptEntry) => {
|
||||
if (entry.kind === "assistant" && entry.delta) {
|
||||
const text = entry.text;
|
||||
if (!text.trim()) return;
|
||||
const last = summarized[summarized.length - 1];
|
||||
if (last && last.streamingKind === "assistant") {
|
||||
last.text += text;
|
||||
} else {
|
||||
summarized.push({ text, tone: "assistant", streamingKind: "assistant" });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (entry.kind === "thinking" && entry.delta) {
|
||||
const text = entry.text;
|
||||
if (!text.trim()) return;
|
||||
const last = summarized[summarized.length - 1];
|
||||
if (last && last.streamingKind === "thinking") {
|
||||
last.text += text;
|
||||
} else {
|
||||
summarized.push({ text: `[thinking] ${text}`, tone: "info", streamingKind: "thinking" });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const summary = summarizeEntry(entry);
|
||||
if (!summary) return;
|
||||
summarized.push({ text: summary.text, tone: summary.tone });
|
||||
};
|
||||
|
||||
const items: FeedItem[] = [];
|
||||
for (const line of split.slice(-8)) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
const parsed = adapter.parseStdoutLine(trimmed, ts);
|
||||
if (parsed.length === 0) {
|
||||
const fallback = createFeedItem(run, ts, trimmed, "info", nextIdRef.current++);
|
||||
if (fallback) items.push(fallback);
|
||||
continue;
|
||||
}
|
||||
for (const entry of parsed) {
|
||||
appendSummary(entry);
|
||||
}
|
||||
}
|
||||
|
||||
for (const summary of summarized) {
|
||||
const item = createFeedItem(run, ts, summary.text, summary.tone, nextIdRef.current++, {
|
||||
streamingKind: summary.streamingKind,
|
||||
preserveWhitespace: !!summary.streamingKind,
|
||||
});
|
||||
if (item) items.push(item);
|
||||
}
|
||||
|
||||
return items;
|
||||
}
|
||||
|
||||
function parseStderrChunk(
|
||||
run: LiveRunForIssue,
|
||||
chunk: string,
|
||||
ts: string,
|
||||
pendingByRun: Map<string, string>,
|
||||
nextIdRef: MutableRefObject<number>,
|
||||
): FeedItem[] {
|
||||
const pendingKey = `${run.id}:stderr`;
|
||||
const combined = `${pendingByRun.get(pendingKey) ?? ""}${chunk}`;
|
||||
const split = combined.split(/\r?\n/);
|
||||
pendingByRun.set(pendingKey, split.pop() ?? "");
|
||||
|
||||
const items: FeedItem[] = [];
|
||||
for (const line of split.slice(-8)) {
|
||||
const item = createFeedItem(run, ts, line, "error", nextIdRef.current++);
|
||||
if (item) items.push(item);
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
function isRunActive(run: LiveRunForIssue): boolean {
|
||||
return run.status === "queued" || run.status === "running";
|
||||
}
|
||||
@@ -195,11 +23,6 @@ interface ActiveAgentsPanelProps {
|
||||
}
|
||||
|
||||
export function ActiveAgentsPanel({ companyId }: ActiveAgentsPanelProps) {
|
||||
const [feedByRun, setFeedByRun] = useState<Map<string, FeedItem[]>>(new Map());
|
||||
const seenKeysRef = useRef(new Set<string>());
|
||||
const pendingByRunRef = useRef(new Map<string, string>());
|
||||
const nextIdRef = useRef(1);
|
||||
|
||||
const { data: liveRuns } = useQuery({
|
||||
queryKey: [...queryKeys.liveRuns(companyId), "dashboard"],
|
||||
queryFn: () => heartbeatsApi.liveRunsForCompany(companyId, MIN_DASHBOARD_RUNS),
|
||||
@@ -220,179 +43,30 @@ export function ActiveAgentsPanel({ companyId }: ActiveAgentsPanelProps) {
|
||||
return map;
|
||||
}, [issues]);
|
||||
|
||||
const runById = useMemo(() => new Map(runs.map((r) => [r.id, r])), [runs]);
|
||||
const activeRunIds = useMemo(() => new Set(runs.filter(isRunActive).map((r) => r.id)), [runs]);
|
||||
|
||||
// Clean up pending buffers for runs that ended
|
||||
useEffect(() => {
|
||||
const stillActive = new Set<string>();
|
||||
for (const runId of activeRunIds) {
|
||||
stillActive.add(`${runId}:stdout`);
|
||||
stillActive.add(`${runId}:stderr`);
|
||||
}
|
||||
for (const key of pendingByRunRef.current.keys()) {
|
||||
if (!stillActive.has(key)) {
|
||||
pendingByRunRef.current.delete(key);
|
||||
}
|
||||
}
|
||||
}, [activeRunIds]);
|
||||
|
||||
// WebSocket connection for streaming
|
||||
useEffect(() => {
|
||||
if (activeRunIds.size === 0) return;
|
||||
|
||||
let closed = false;
|
||||
let reconnectTimer: number | null = null;
|
||||
let socket: WebSocket | null = null;
|
||||
|
||||
const appendItems = (runId: string, items: FeedItem[]) => {
|
||||
if (items.length === 0) return;
|
||||
setFeedByRun((prev) => {
|
||||
const next = new Map(prev);
|
||||
const existing = [...(next.get(runId) ?? [])];
|
||||
for (const item of items) {
|
||||
if (seenKeysRef.current.has(item.dedupeKey)) continue;
|
||||
seenKeysRef.current.add(item.dedupeKey);
|
||||
|
||||
const last = existing[existing.length - 1];
|
||||
if (
|
||||
item.streamingKind &&
|
||||
last &&
|
||||
last.runId === item.runId &&
|
||||
last.streamingKind === item.streamingKind
|
||||
) {
|
||||
const mergedText = `${last.text}${item.text}`;
|
||||
const nextText =
|
||||
mergedText.length > MAX_STREAMING_TEXT_LENGTH
|
||||
? mergedText.slice(-MAX_STREAMING_TEXT_LENGTH)
|
||||
: mergedText;
|
||||
existing[existing.length - 1] = {
|
||||
...last,
|
||||
ts: item.ts,
|
||||
text: nextText,
|
||||
dedupeKey: last.dedupeKey,
|
||||
};
|
||||
continue;
|
||||
}
|
||||
|
||||
existing.push(item);
|
||||
}
|
||||
if (seenKeysRef.current.size > 6000) {
|
||||
seenKeysRef.current.clear();
|
||||
}
|
||||
next.set(runId, existing.slice(-MAX_FEED_ITEMS));
|
||||
return next;
|
||||
});
|
||||
};
|
||||
|
||||
const scheduleReconnect = () => {
|
||||
if (closed) return;
|
||||
reconnectTimer = window.setTimeout(connect, 1500);
|
||||
};
|
||||
|
||||
const connect = () => {
|
||||
if (closed) return;
|
||||
const protocol = window.location.protocol === "https:" ? "wss" : "ws";
|
||||
const url = `${protocol}://${window.location.host}/api/companies/${encodeURIComponent(companyId)}/events/ws`;
|
||||
socket = new WebSocket(url);
|
||||
|
||||
socket.onmessage = (message) => {
|
||||
const raw = typeof message.data === "string" ? message.data : "";
|
||||
if (!raw) return;
|
||||
|
||||
let event: LiveEvent;
|
||||
try {
|
||||
event = JSON.parse(raw) as LiveEvent;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.companyId !== companyId) return;
|
||||
const payload = event.payload ?? {};
|
||||
const runId = readString(payload["runId"]);
|
||||
if (!runId || !activeRunIds.has(runId)) return;
|
||||
|
||||
const run = runById.get(runId);
|
||||
if (!run) return;
|
||||
|
||||
if (event.type === "heartbeat.run.event") {
|
||||
const seq = typeof payload["seq"] === "number" ? payload["seq"] : null;
|
||||
const eventType = readString(payload["eventType"]) ?? "event";
|
||||
const messageText = readString(payload["message"]) ?? eventType;
|
||||
const dedupeKey = `${runId}:event:${seq ?? `${eventType}:${messageText}:${event.createdAt}`}`;
|
||||
if (seenKeysRef.current.has(dedupeKey)) return;
|
||||
seenKeysRef.current.add(dedupeKey);
|
||||
if (seenKeysRef.current.size > 6000) seenKeysRef.current.clear();
|
||||
const tone = eventType === "error" ? "error" : eventType === "lifecycle" ? "warn" : "info";
|
||||
const item = createFeedItem(run, event.createdAt, messageText, tone, nextIdRef.current++);
|
||||
if (item) appendItems(run.id, [item]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "heartbeat.run.status") {
|
||||
const status = readString(payload["status"]) ?? "updated";
|
||||
const dedupeKey = `${runId}:status:${status}:${readString(payload["finishedAt"]) ?? ""}`;
|
||||
if (seenKeysRef.current.has(dedupeKey)) return;
|
||||
seenKeysRef.current.add(dedupeKey);
|
||||
if (seenKeysRef.current.size > 6000) seenKeysRef.current.clear();
|
||||
const tone = status === "failed" || status === "timed_out" ? "error" : "warn";
|
||||
const item = createFeedItem(run, event.createdAt, `run ${status}`, tone, nextIdRef.current++);
|
||||
if (item) appendItems(run.id, [item]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "heartbeat.run.log") {
|
||||
const chunk = readString(payload["chunk"]);
|
||||
if (!chunk) return;
|
||||
const stream = readString(payload["stream"]) === "stderr" ? "stderr" : "stdout";
|
||||
if (stream === "stderr") {
|
||||
appendItems(run.id, parseStderrChunk(run, chunk, event.createdAt, pendingByRunRef.current, nextIdRef));
|
||||
return;
|
||||
}
|
||||
appendItems(run.id, parseStdoutChunk(run, chunk, event.createdAt, pendingByRunRef.current, nextIdRef));
|
||||
}
|
||||
};
|
||||
|
||||
socket.onerror = () => {
|
||||
socket?.close();
|
||||
};
|
||||
|
||||
socket.onclose = () => {
|
||||
scheduleReconnect();
|
||||
};
|
||||
};
|
||||
|
||||
connect();
|
||||
|
||||
return () => {
|
||||
closed = true;
|
||||
if (reconnectTimer !== null) window.clearTimeout(reconnectTimer);
|
||||
if (socket) {
|
||||
socket.onmessage = null;
|
||||
socket.onerror = null;
|
||||
socket.onclose = null;
|
||||
socket.close(1000, "active_agents_panel_unmount");
|
||||
}
|
||||
};
|
||||
}, [activeRunIds, companyId, runById]);
|
||||
const { transcriptByRun, hasOutputForRun } = useLiveRunTranscripts({
|
||||
runs,
|
||||
companyId,
|
||||
maxChunksPerRun: 120,
|
||||
});
|
||||
|
||||
return (
|
||||
<div>
|
||||
<h3 className="text-sm font-semibold text-muted-foreground uppercase tracking-wide mb-3">
|
||||
<h3 className="mb-3 text-sm font-semibold uppercase tracking-wide text-muted-foreground">
|
||||
Agents
|
||||
</h3>
|
||||
{runs.length === 0 ? (
|
||||
<div className="border border-border rounded-lg p-4">
|
||||
<div className="rounded-xl border border-border p-4">
|
||||
<p className="text-sm text-muted-foreground">No recent agent runs.</p>
|
||||
</div>
|
||||
) : (
|
||||
<div className="grid grid-cols-1 sm:grid-cols-2 xl:grid-cols-4 gap-2 sm:gap-4">
|
||||
<div className="grid grid-cols-1 gap-2 sm:grid-cols-2 sm:gap-4 xl:grid-cols-4">
|
||||
{runs.map((run) => (
|
||||
<AgentRunCard
|
||||
key={run.id}
|
||||
run={run}
|
||||
issue={run.issueId ? issueById.get(run.issueId) : undefined}
|
||||
feed={feedByRun.get(run.id) ?? []}
|
||||
transcript={transcriptByRun.get(run.id) ?? []}
|
||||
hasOutput={hasOutputForRun(run.id)}
|
||||
isActive={isRunActive(run)}
|
||||
/>
|
||||
))}
|
||||
@@ -405,104 +79,76 @@ export function ActiveAgentsPanel({ companyId }: ActiveAgentsPanelProps) {
|
||||
function AgentRunCard({
|
||||
run,
|
||||
issue,
|
||||
feed,
|
||||
transcript,
|
||||
hasOutput,
|
||||
isActive,
|
||||
}: {
|
||||
run: LiveRunForIssue;
|
||||
issue?: Issue;
|
||||
feed: FeedItem[];
|
||||
transcript: TranscriptEntry[];
|
||||
hasOutput: boolean;
|
||||
isActive: boolean;
|
||||
}) {
|
||||
const bodyRef = useRef<HTMLDivElement>(null);
|
||||
const recent = feed.slice(-20);
|
||||
|
||||
useEffect(() => {
|
||||
const body = bodyRef.current;
|
||||
if (!body) return;
|
||||
body.scrollTo({ top: body.scrollHeight, behavior: "smooth" });
|
||||
}, [feed.length]);
|
||||
|
||||
return (
|
||||
<div className={cn(
|
||||
"flex flex-col rounded-lg border overflow-hidden min-h-[200px]",
|
||||
"flex h-[320px] flex-col overflow-hidden rounded-xl border shadow-sm",
|
||||
isActive
|
||||
? "border-blue-500/30 bg-background/80 shadow-[0_0_12px_rgba(59,130,246,0.08)]"
|
||||
: "border-border bg-background/50",
|
||||
? "border-cyan-500/25 bg-cyan-500/[0.04] shadow-[0_16px_40px_rgba(6,182,212,0.08)]"
|
||||
: "border-border bg-background/70",
|
||||
)}>
|
||||
{/* Header */}
|
||||
<div className="flex items-center justify-between px-3 py-2 border-b border-border/50">
|
||||
<div className="flex items-center gap-2 min-w-0">
|
||||
{isActive ? (
|
||||
<span className="relative flex h-2 w-2 shrink-0">
|
||||
<span className="animate-pulse absolute inline-flex h-full w-full rounded-full bg-blue-400 opacity-75" />
|
||||
<span className="relative inline-flex rounded-full h-2 w-2 bg-blue-500" />
|
||||
</span>
|
||||
) : (
|
||||
<span className="flex h-2 w-2 shrink-0">
|
||||
<span className="inline-flex rounded-full h-2 w-2 bg-muted-foreground/40" />
|
||||
</span>
|
||||
)}
|
||||
<Identity name={run.agentName} size="sm" />
|
||||
{isActive && (
|
||||
<span className="text-[11px] font-medium text-blue-600 dark:text-blue-400">Live</span>
|
||||
)}
|
||||
</div>
|
||||
<Link
|
||||
to={`/agents/${run.agentId}/runs/${run.id}`}
|
||||
className="inline-flex items-center gap-1 text-[10px] text-muted-foreground hover:text-foreground shrink-0"
|
||||
>
|
||||
<ExternalLink className="h-2.5 w-2.5" />
|
||||
</Link>
|
||||
</div>
|
||||
<div className="border-b border-border/60 px-3 py-3">
|
||||
<div className="flex items-start justify-between gap-2">
|
||||
<div className="min-w-0">
|
||||
<div className="flex items-center gap-2">
|
||||
{isActive ? (
|
||||
<span className="relative flex h-2.5 w-2.5 shrink-0">
|
||||
<span className="absolute inline-flex h-full w-full animate-ping rounded-full bg-cyan-400 opacity-70" />
|
||||
<span className="relative inline-flex h-2.5 w-2.5 rounded-full bg-cyan-500" />
|
||||
</span>
|
||||
) : (
|
||||
<span className="inline-flex h-2.5 w-2.5 rounded-full bg-muted-foreground/35" />
|
||||
)}
|
||||
<Identity name={run.agentName} size="sm" />
|
||||
</div>
|
||||
<div className="mt-2 flex items-center gap-2 text-[11px] text-muted-foreground">
|
||||
<span>{isActive ? "Live now" : run.finishedAt ? `Finished ${relativeTime(run.finishedAt)}` : `Started ${relativeTime(run.createdAt)}`}</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
{/* Issue context */}
|
||||
{run.issueId && (
|
||||
<div className="px-3 py-1.5 border-b border-border/40 text-xs flex items-center gap-1 min-w-0">
|
||||
<Link
|
||||
to={`/issues/${issue?.identifier ?? run.issueId}`}
|
||||
className={cn(
|
||||
"hover:underline min-w-0 line-clamp-2 min-h-[2rem]",
|
||||
isActive ? "text-blue-600 hover:text-blue-500 dark:text-blue-400 dark:hover:text-blue-300" : "text-muted-foreground hover:text-foreground",
|
||||
)}
|
||||
title={issue?.title ? `${issue?.identifier ?? run.issueId.slice(0, 8)} - ${issue.title}` : issue?.identifier ?? run.issueId.slice(0, 8)}
|
||||
to={`/agents/${run.agentId}/runs/${run.id}`}
|
||||
className="inline-flex items-center gap-1 rounded-full border border-border/70 bg-background/70 px-2 py-1 text-[10px] text-muted-foreground transition-colors hover:text-foreground"
|
||||
>
|
||||
{issue?.identifier ?? run.issueId.slice(0, 8)}
|
||||
{issue?.title ? ` - ${issue.title}` : ""}
|
||||
<ExternalLink className="h-2.5 w-2.5" />
|
||||
</Link>
|
||||
</div>
|
||||
)}
|
||||
|
||||
{/* Feed body */}
|
||||
<div ref={bodyRef} className="flex-1 max-h-[140px] overflow-y-auto p-2 font-mono text-[11px] space-y-1">
|
||||
{isActive && recent.length === 0 && (
|
||||
<div className="text-xs text-muted-foreground">Waiting for output...</div>
|
||||
)}
|
||||
{!isActive && recent.length === 0 && (
|
||||
<div className="text-xs text-muted-foreground">
|
||||
{run.finishedAt ? `Finished ${relativeTime(run.finishedAt)}` : `Started ${relativeTime(run.createdAt)}`}
|
||||
{run.issueId && (
|
||||
<div className="mt-3 rounded-lg border border-border/60 bg-background/60 px-2.5 py-2 text-xs">
|
||||
<Link
|
||||
to={`/issues/${issue?.identifier ?? run.issueId}`}
|
||||
className={cn(
|
||||
"line-clamp-2 hover:underline",
|
||||
isActive ? "text-cyan-700 dark:text-cyan-300" : "text-muted-foreground hover:text-foreground",
|
||||
)}
|
||||
title={issue?.title ? `${issue?.identifier ?? run.issueId.slice(0, 8)} - ${issue.title}` : issue?.identifier ?? run.issueId.slice(0, 8)}
|
||||
>
|
||||
{issue?.identifier ?? run.issueId.slice(0, 8)}
|
||||
{issue?.title ? ` - ${issue.title}` : ""}
|
||||
</Link>
|
||||
</div>
|
||||
)}
|
||||
{recent.map((item, index) => (
|
||||
<div
|
||||
key={item.id}
|
||||
className={cn(
|
||||
"flex gap-2 items-start",
|
||||
index === recent.length - 1 && isActive && "animate-in fade-in slide-in-from-bottom-1 duration-300",
|
||||
)}
|
||||
>
|
||||
<span className="text-[10px] text-muted-foreground shrink-0">{relativeTime(item.ts)}</span>
|
||||
<span className={cn(
|
||||
"min-w-0 break-words",
|
||||
item.tone === "error" && "text-red-600 dark:text-red-300",
|
||||
item.tone === "warn" && "text-amber-600 dark:text-amber-300",
|
||||
item.tone === "assistant" && "text-emerald-700 dark:text-emerald-200",
|
||||
item.tone === "tool" && "text-cyan-600 dark:text-cyan-300",
|
||||
item.tone === "info" && "text-foreground/80",
|
||||
)}>
|
||||
{item.text}
|
||||
</span>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
|
||||
<div className="min-h-0 flex-1 overflow-y-auto p-3">
|
||||
<RunTranscriptView
|
||||
entries={transcript}
|
||||
density="compact"
|
||||
limit={5}
|
||||
streaming={isActive}
|
||||
collapseStdout
|
||||
emptyMessage={hasOutput ? "Waiting for transcript parsing..." : isActive ? "Waiting for output..." : "No transcript captured."}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
|
||||
@@ -1,262 +1,32 @@
|
||||
import { useEffect, useMemo, useRef, useState, type MutableRefObject } from "react";
|
||||
import { useMemo, useState } from "react";
|
||||
import { Link } from "@/lib/router";
|
||||
import { useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { heartbeatsApi, type LiveRunForIssue } from "../api/heartbeats";
|
||||
import { getUIAdapter } from "../adapters";
|
||||
import type { TranscriptEntry } from "../adapters";
|
||||
import { queryKeys } from "../lib/queryKeys";
|
||||
import { cn, relativeTime, formatDateTime } from "../lib/utils";
|
||||
import { formatDateTime } from "../lib/utils";
|
||||
import { ExternalLink, Square } from "lucide-react";
|
||||
import { Identity } from "./Identity";
|
||||
import { StatusBadge } from "./StatusBadge";
|
||||
import { RunTranscriptView } from "./transcript/RunTranscriptView";
|
||||
import { useLiveRunTranscripts } from "./transcript/useLiveRunTranscripts";
|
||||
|
||||
interface LiveRunWidgetProps {
|
||||
issueId: string;
|
||||
companyId?: string | null;
|
||||
}
|
||||
|
||||
type FeedTone = "info" | "warn" | "error" | "assistant" | "tool";
|
||||
|
||||
interface FeedItem {
|
||||
id: string;
|
||||
ts: string;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
agentName: string;
|
||||
text: string;
|
||||
tone: FeedTone;
|
||||
dedupeKey: string;
|
||||
streamingKind?: "assistant" | "thinking";
|
||||
}
|
||||
|
||||
const MAX_FEED_ITEMS = 80;
|
||||
const MAX_FEED_TEXT_LENGTH = 220;
|
||||
const MAX_STREAMING_TEXT_LENGTH = 4000;
|
||||
const LOG_POLL_INTERVAL_MS = 2000;
|
||||
const LOG_READ_LIMIT_BYTES = 256_000;
|
||||
|
||||
function readString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
function toIsoString(value: string | Date | null | undefined): string | null {
|
||||
if (!value) return null;
|
||||
return typeof value === "string" ? value : value.toISOString();
|
||||
}
|
||||
|
||||
function summarizeEntry(entry: TranscriptEntry): { text: string; tone: FeedTone } | null {
|
||||
if (entry.kind === "assistant") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "assistant" } : null;
|
||||
}
|
||||
if (entry.kind === "thinking") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text: `[thinking] ${text}`, tone: "info" } : null;
|
||||
}
|
||||
if (entry.kind === "tool_call") {
|
||||
return { text: `tool ${entry.name}`, tone: "tool" };
|
||||
}
|
||||
if (entry.kind === "tool_result") {
|
||||
const base = entry.content.trim();
|
||||
return {
|
||||
text: entry.isError ? `tool error: ${base}` : `tool result: ${base}`,
|
||||
tone: entry.isError ? "error" : "tool",
|
||||
};
|
||||
}
|
||||
if (entry.kind === "stderr") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "error" } : null;
|
||||
}
|
||||
if (entry.kind === "system") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "warn" } : null;
|
||||
}
|
||||
if (entry.kind === "stdout") {
|
||||
const text = entry.text.trim();
|
||||
return text ? { text, tone: "info" } : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function createFeedItem(
|
||||
run: LiveRunForIssue,
|
||||
ts: string,
|
||||
text: string,
|
||||
tone: FeedTone,
|
||||
nextId: number,
|
||||
options?: {
|
||||
streamingKind?: "assistant" | "thinking";
|
||||
preserveWhitespace?: boolean;
|
||||
},
|
||||
): FeedItem | null {
|
||||
if (!text.trim()) return null;
|
||||
const base = options?.preserveWhitespace ? text : text.trim();
|
||||
const maxLength = options?.streamingKind ? MAX_STREAMING_TEXT_LENGTH : MAX_FEED_TEXT_LENGTH;
|
||||
const normalized = base.length > maxLength ? base.slice(-maxLength) : base;
|
||||
return {
|
||||
id: `${run.id}:${nextId}`,
|
||||
ts,
|
||||
runId: run.id,
|
||||
agentId: run.agentId,
|
||||
agentName: run.agentName,
|
||||
text: normalized,
|
||||
tone,
|
||||
dedupeKey: `feed:${run.id}:${ts}:${tone}:${normalized}`,
|
||||
streamingKind: options?.streamingKind,
|
||||
};
|
||||
}
|
||||
|
||||
function parseStdoutChunk(
|
||||
run: LiveRunForIssue,
|
||||
chunk: string,
|
||||
ts: string,
|
||||
pendingByRun: Map<string, string>,
|
||||
nextIdRef: MutableRefObject<number>,
|
||||
): FeedItem[] {
|
||||
const pendingKey = `${run.id}:stdout`;
|
||||
const combined = `${pendingByRun.get(pendingKey) ?? ""}${chunk}`;
|
||||
const split = combined.split(/\r?\n/);
|
||||
pendingByRun.set(pendingKey, split.pop() ?? "");
|
||||
const adapter = getUIAdapter(run.adapterType);
|
||||
|
||||
const summarized: Array<{ text: string; tone: FeedTone; streamingKind?: "assistant" | "thinking" }> = [];
|
||||
const appendSummary = (entry: TranscriptEntry) => {
|
||||
if (entry.kind === "assistant" && entry.delta) {
|
||||
const text = entry.text;
|
||||
if (!text.trim()) return;
|
||||
const last = summarized[summarized.length - 1];
|
||||
if (last && last.streamingKind === "assistant") {
|
||||
last.text += text;
|
||||
} else {
|
||||
summarized.push({ text, tone: "assistant", streamingKind: "assistant" });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (entry.kind === "thinking" && entry.delta) {
|
||||
const text = entry.text;
|
||||
if (!text.trim()) return;
|
||||
const last = summarized[summarized.length - 1];
|
||||
if (last && last.streamingKind === "thinking") {
|
||||
last.text += text;
|
||||
} else {
|
||||
summarized.push({ text: `[thinking] ${text}`, tone: "info", streamingKind: "thinking" });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const summary = summarizeEntry(entry);
|
||||
if (!summary) return;
|
||||
summarized.push({ text: summary.text, tone: summary.tone });
|
||||
};
|
||||
|
||||
const items: FeedItem[] = [];
|
||||
for (const line of split.slice(-8)) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
const parsed = adapter.parseStdoutLine(trimmed, ts);
|
||||
if (parsed.length === 0) {
|
||||
if (run.adapterType === "openclaw_gateway") {
|
||||
continue;
|
||||
}
|
||||
const fallback = createFeedItem(run, ts, trimmed, "info", nextIdRef.current++);
|
||||
if (fallback) items.push(fallback);
|
||||
continue;
|
||||
}
|
||||
for (const entry of parsed) {
|
||||
appendSummary(entry);
|
||||
}
|
||||
}
|
||||
|
||||
for (const summary of summarized) {
|
||||
const item = createFeedItem(run, ts, summary.text, summary.tone, nextIdRef.current++, {
|
||||
streamingKind: summary.streamingKind,
|
||||
preserveWhitespace: !!summary.streamingKind,
|
||||
});
|
||||
if (item) items.push(item);
|
||||
}
|
||||
|
||||
return items;
|
||||
}
|
||||
|
||||
function parseStderrChunk(
|
||||
run: LiveRunForIssue,
|
||||
chunk: string,
|
||||
ts: string,
|
||||
pendingByRun: Map<string, string>,
|
||||
nextIdRef: MutableRefObject<number>,
|
||||
): FeedItem[] {
|
||||
const pendingKey = `${run.id}:stderr`;
|
||||
const combined = `${pendingByRun.get(pendingKey) ?? ""}${chunk}`;
|
||||
const split = combined.split(/\r?\n/);
|
||||
pendingByRun.set(pendingKey, split.pop() ?? "");
|
||||
|
||||
const items: FeedItem[] = [];
|
||||
for (const line of split.slice(-8)) {
|
||||
const item = createFeedItem(run, ts, line, "error", nextIdRef.current++);
|
||||
if (item) items.push(item);
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
function parsePersistedLogContent(
|
||||
runId: string,
|
||||
content: string,
|
||||
pendingByRun: Map<string, string>,
|
||||
): Array<{ ts: string; stream: "stdout" | "stderr" | "system"; chunk: string }> {
|
||||
if (!content) return [];
|
||||
|
||||
const pendingKey = `${runId}:records`;
|
||||
const combined = `${pendingByRun.get(pendingKey) ?? ""}${content}`;
|
||||
const split = combined.split("\n");
|
||||
pendingByRun.set(pendingKey, split.pop() ?? "");
|
||||
|
||||
const parsed: Array<{ ts: string; stream: "stdout" | "stderr" | "system"; chunk: string }> = [];
|
||||
for (const line of split) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
try {
|
||||
const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown };
|
||||
const stream = raw.stream === "stderr" || raw.stream === "system" ? raw.stream : "stdout";
|
||||
const chunk = typeof raw.chunk === "string" ? raw.chunk : "";
|
||||
const ts = typeof raw.ts === "string" ? raw.ts : new Date().toISOString();
|
||||
if (!chunk) continue;
|
||||
parsed.push({ ts, stream, chunk });
|
||||
} catch {
|
||||
// Ignore malformed log rows.
|
||||
}
|
||||
}
|
||||
|
||||
return parsed;
|
||||
function isRunActive(status: string): boolean {
|
||||
return status === "queued" || status === "running";
|
||||
}
|
||||
|
||||
export function LiveRunWidget({ issueId, companyId }: LiveRunWidgetProps) {
|
||||
const queryClient = useQueryClient();
|
||||
const [feed, setFeed] = useState<FeedItem[]>([]);
|
||||
const [cancellingRunIds, setCancellingRunIds] = useState(new Set<string>());
|
||||
const seenKeysRef = useRef(new Set<string>());
|
||||
const pendingByRunRef = useRef(new Map<string, string>());
|
||||
const pendingLogRowsByRunRef = useRef(new Map<string, string>());
|
||||
const logOffsetByRunRef = useRef(new Map<string, number>());
|
||||
const runMetaByIdRef = useRef(new Map<string, { agentId: string; agentName: string }>());
|
||||
const nextIdRef = useRef(1);
|
||||
const bodyRef = useRef<HTMLDivElement>(null);
|
||||
|
||||
const handleCancelRun = async (runId: string) => {
|
||||
setCancellingRunIds((prev) => new Set(prev).add(runId));
|
||||
try {
|
||||
await heartbeatsApi.cancel(runId);
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.issues.liveRuns(issueId) });
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.issues.activeRun(issueId) });
|
||||
} finally {
|
||||
setCancellingRunIds((prev) => {
|
||||
const next = new Set(prev);
|
||||
next.delete(runId);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
const { data: liveRuns } = useQuery({
|
||||
queryKey: queryKeys.issues.liveRuns(issueId),
|
||||
@@ -297,329 +67,94 @@ export function LiveRunWidget({ issueId, companyId }: LiveRunWidgetProps) {
|
||||
);
|
||||
}, [activeRun, issueId, liveRuns]);
|
||||
|
||||
const runById = useMemo(() => new Map(runs.map((run) => [run.id, run])), [runs]);
|
||||
const activeRunIds = useMemo(() => new Set(runs.map((run) => run.id)), [runs]);
|
||||
const runIdsKey = useMemo(
|
||||
() => runs.map((run) => run.id).sort((a, b) => a.localeCompare(b)).join(","),
|
||||
[runs],
|
||||
);
|
||||
const appendItems = (items: FeedItem[]) => {
|
||||
if (items.length === 0) return;
|
||||
setFeed((prev) => {
|
||||
const next = [...prev];
|
||||
for (const item of items) {
|
||||
if (seenKeysRef.current.has(item.dedupeKey)) continue;
|
||||
seenKeysRef.current.add(item.dedupeKey);
|
||||
const { transcriptByRun, hasOutputForRun } = useLiveRunTranscripts({ runs, companyId });
|
||||
|
||||
const last = next[next.length - 1];
|
||||
if (
|
||||
item.streamingKind &&
|
||||
last &&
|
||||
last.runId === item.runId &&
|
||||
last.streamingKind === item.streamingKind
|
||||
) {
|
||||
const mergedText = `${last.text}${item.text}`;
|
||||
const nextText =
|
||||
mergedText.length > MAX_STREAMING_TEXT_LENGTH
|
||||
? mergedText.slice(-MAX_STREAMING_TEXT_LENGTH)
|
||||
: mergedText;
|
||||
next[next.length - 1] = {
|
||||
...last,
|
||||
ts: item.ts,
|
||||
text: nextText,
|
||||
dedupeKey: last.dedupeKey,
|
||||
};
|
||||
continue;
|
||||
}
|
||||
|
||||
next.push(item);
|
||||
}
|
||||
if (seenKeysRef.current.size > 6000) {
|
||||
seenKeysRef.current.clear();
|
||||
}
|
||||
if (next.length === prev.length) return prev;
|
||||
return next.slice(-MAX_FEED_ITEMS);
|
||||
});
|
||||
const handleCancelRun = async (runId: string) => {
|
||||
setCancellingRunIds((prev) => new Set(prev).add(runId));
|
||||
try {
|
||||
await heartbeatsApi.cancel(runId);
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.issues.liveRuns(issueId) });
|
||||
queryClient.invalidateQueries({ queryKey: queryKeys.issues.activeRun(issueId) });
|
||||
} finally {
|
||||
setCancellingRunIds((prev) => {
|
||||
const next = new Set(prev);
|
||||
next.delete(runId);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
const body = bodyRef.current;
|
||||
if (!body) return;
|
||||
body.scrollTo({ top: body.scrollHeight, behavior: "smooth" });
|
||||
}, [feed.length]);
|
||||
|
||||
useEffect(() => {
|
||||
for (const run of runs) {
|
||||
runMetaByIdRef.current.set(run.id, { agentId: run.agentId, agentName: run.agentName });
|
||||
}
|
||||
}, [runs]);
|
||||
|
||||
useEffect(() => {
|
||||
const stillActive = new Set<string>();
|
||||
for (const runId of activeRunIds) {
|
||||
stillActive.add(`${runId}:stdout`);
|
||||
stillActive.add(`${runId}:stderr`);
|
||||
}
|
||||
for (const key of pendingByRunRef.current.keys()) {
|
||||
if (!stillActive.has(key)) {
|
||||
pendingByRunRef.current.delete(key);
|
||||
}
|
||||
}
|
||||
const liveRunIds = new Set(activeRunIds);
|
||||
for (const key of pendingLogRowsByRunRef.current.keys()) {
|
||||
const runId = key.replace(/:records$/, "");
|
||||
if (!liveRunIds.has(runId)) {
|
||||
pendingLogRowsByRunRef.current.delete(key);
|
||||
}
|
||||
}
|
||||
for (const runId of logOffsetByRunRef.current.keys()) {
|
||||
if (!liveRunIds.has(runId)) {
|
||||
logOffsetByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
}, [activeRunIds]);
|
||||
|
||||
useEffect(() => {
|
||||
if (runs.length === 0) return;
|
||||
|
||||
let cancelled = false;
|
||||
|
||||
const readRunLog = async (run: LiveRunForIssue) => {
|
||||
const offset = logOffsetByRunRef.current.get(run.id) ?? 0;
|
||||
try {
|
||||
const result = await heartbeatsApi.log(run.id, offset, LOG_READ_LIMIT_BYTES);
|
||||
if (cancelled) return;
|
||||
|
||||
const rows = parsePersistedLogContent(run.id, result.content, pendingLogRowsByRunRef.current);
|
||||
const items: FeedItem[] = [];
|
||||
for (const row of rows) {
|
||||
if (row.stream === "stderr") {
|
||||
items.push(
|
||||
...parseStderrChunk(run, row.chunk, row.ts, pendingByRunRef.current, nextIdRef),
|
||||
);
|
||||
continue;
|
||||
}
|
||||
if (row.stream === "system") {
|
||||
const item = createFeedItem(run, row.ts, row.chunk, "warn", nextIdRef.current++);
|
||||
if (item) items.push(item);
|
||||
continue;
|
||||
}
|
||||
items.push(
|
||||
...parseStdoutChunk(run, row.chunk, row.ts, pendingByRunRef.current, nextIdRef),
|
||||
);
|
||||
}
|
||||
appendItems(items);
|
||||
|
||||
if (result.nextOffset !== undefined) {
|
||||
logOffsetByRunRef.current.set(run.id, result.nextOffset);
|
||||
return;
|
||||
}
|
||||
if (result.content.length > 0) {
|
||||
logOffsetByRunRef.current.set(run.id, offset + result.content.length);
|
||||
}
|
||||
} catch {
|
||||
// Ignore log read errors while run output is initializing.
|
||||
}
|
||||
};
|
||||
|
||||
const readAll = async () => {
|
||||
await Promise.all(runs.map((run) => readRunLog(run)));
|
||||
};
|
||||
|
||||
void readAll();
|
||||
const interval = window.setInterval(() => {
|
||||
void readAll();
|
||||
}, LOG_POLL_INTERVAL_MS);
|
||||
|
||||
return () => {
|
||||
cancelled = true;
|
||||
window.clearInterval(interval);
|
||||
};
|
||||
}, [runIdsKey, runs]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!companyId || activeRunIds.size === 0) return;
|
||||
|
||||
let closed = false;
|
||||
let reconnectTimer: number | null = null;
|
||||
let socket: WebSocket | null = null;
|
||||
|
||||
const scheduleReconnect = () => {
|
||||
if (closed) return;
|
||||
reconnectTimer = window.setTimeout(connect, 1500);
|
||||
};
|
||||
|
||||
const connect = () => {
|
||||
if (closed) return;
|
||||
const protocol = window.location.protocol === "https:" ? "wss" : "ws";
|
||||
const url = `${protocol}://${window.location.host}/api/companies/${encodeURIComponent(companyId)}/events/ws`;
|
||||
socket = new WebSocket(url);
|
||||
|
||||
socket.onmessage = (message) => {
|
||||
const raw = typeof message.data === "string" ? message.data : "";
|
||||
if (!raw) return;
|
||||
|
||||
let event: LiveEvent;
|
||||
try {
|
||||
event = JSON.parse(raw) as LiveEvent;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.companyId !== companyId) return;
|
||||
const payload = event.payload ?? {};
|
||||
const runId = readString(payload["runId"]);
|
||||
if (!runId || !activeRunIds.has(runId)) return;
|
||||
|
||||
const run = runById.get(runId);
|
||||
if (!run) return;
|
||||
|
||||
if (event.type === "heartbeat.run.event") {
|
||||
const seq = typeof payload["seq"] === "number" ? payload["seq"] : null;
|
||||
const eventType = readString(payload["eventType"]) ?? "event";
|
||||
const messageText = readString(payload["message"]) ?? eventType;
|
||||
const dedupeKey = `${runId}:event:${seq ?? `${eventType}:${messageText}:${event.createdAt}`}`;
|
||||
if (seenKeysRef.current.has(dedupeKey)) return;
|
||||
seenKeysRef.current.add(dedupeKey);
|
||||
if (seenKeysRef.current.size > 2000) {
|
||||
seenKeysRef.current.clear();
|
||||
}
|
||||
const tone = eventType === "error" ? "error" : eventType === "lifecycle" ? "warn" : "info";
|
||||
const item = createFeedItem(run, event.createdAt, messageText, tone, nextIdRef.current++);
|
||||
if (item) appendItems([item]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "heartbeat.run.status") {
|
||||
const status = readString(payload["status"]) ?? "updated";
|
||||
const dedupeKey = `${runId}:status:${status}:${readString(payload["finishedAt"]) ?? ""}`;
|
||||
if (seenKeysRef.current.has(dedupeKey)) return;
|
||||
seenKeysRef.current.add(dedupeKey);
|
||||
if (seenKeysRef.current.size > 2000) {
|
||||
seenKeysRef.current.clear();
|
||||
}
|
||||
const tone = status === "failed" || status === "timed_out" ? "error" : "warn";
|
||||
const item = createFeedItem(run, event.createdAt, `run ${status}`, tone, nextIdRef.current++);
|
||||
if (item) appendItems([item]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "heartbeat.run.log") {
|
||||
const chunk = readString(payload["chunk"]);
|
||||
if (!chunk) return;
|
||||
const stream = readString(payload["stream"]) === "stderr" ? "stderr" : "stdout";
|
||||
if (stream === "stderr") {
|
||||
appendItems(parseStderrChunk(run, chunk, event.createdAt, pendingByRunRef.current, nextIdRef));
|
||||
return;
|
||||
}
|
||||
appendItems(parseStdoutChunk(run, chunk, event.createdAt, pendingByRunRef.current, nextIdRef));
|
||||
}
|
||||
};
|
||||
|
||||
socket.onerror = () => {
|
||||
socket?.close();
|
||||
};
|
||||
|
||||
socket.onclose = () => {
|
||||
scheduleReconnect();
|
||||
};
|
||||
};
|
||||
|
||||
connect();
|
||||
|
||||
return () => {
|
||||
closed = true;
|
||||
if (reconnectTimer !== null) window.clearTimeout(reconnectTimer);
|
||||
if (socket) {
|
||||
socket.onmessage = null;
|
||||
socket.onerror = null;
|
||||
socket.onclose = null;
|
||||
socket.close(1000, "issue_live_widget_unmount");
|
||||
}
|
||||
};
|
||||
}, [activeRunIds, companyId, runById]);
|
||||
|
||||
if (runs.length === 0 && feed.length === 0) return null;
|
||||
|
||||
const recent = feed.slice(-25);
|
||||
if (runs.length === 0) return null;
|
||||
|
||||
return (
|
||||
<div className="rounded-lg border border-cyan-500/30 bg-background/80 overflow-hidden shadow-[0_0_12px_rgba(6,182,212,0.08)]">
|
||||
{runs.length > 0 ? (
|
||||
runs.map((run) => (
|
||||
<div key={run.id} className="px-3 py-2 border-b border-border/50">
|
||||
<div className="flex items-center justify-between mb-2">
|
||||
<Link to={`/agents/${run.agentId}`} className="hover:underline">
|
||||
<Identity name={run.agentName} size="sm" />
|
||||
</Link>
|
||||
<span className="text-xs text-muted-foreground">
|
||||
{formatDateTime(run.startedAt ?? run.createdAt)}
|
||||
</span>
|
||||
</div>
|
||||
<div className="flex items-center gap-2 text-xs">
|
||||
<span className="text-muted-foreground">Run</span>
|
||||
<Link
|
||||
to={`/agents/${run.agentId}/runs/${run.id}`}
|
||||
className="inline-flex items-center rounded-md border border-border bg-accent/40 px-2 py-1 font-mono text-muted-foreground hover:text-foreground hover:bg-accent/60 transition-colors"
|
||||
>
|
||||
{run.id.slice(0, 8)}
|
||||
</Link>
|
||||
<StatusBadge status={run.status} />
|
||||
<div className="ml-auto flex items-center gap-2">
|
||||
<button
|
||||
onClick={() => handleCancelRun(run.id)}
|
||||
disabled={cancellingRunIds.has(run.id)}
|
||||
className="inline-flex items-center gap-1 text-[10px] text-red-600 hover:text-red-500 dark:text-red-400 dark:hover:text-red-300 disabled:opacity-50"
|
||||
>
|
||||
<Square className="h-2 w-2" fill="currentColor" />
|
||||
{cancellingRunIds.has(run.id) ? "Stopping…" : "Stop"}
|
||||
</button>
|
||||
<Link
|
||||
to={`/agents/${run.agentId}/runs/${run.id}`}
|
||||
className="inline-flex items-center gap-1 text-[10px] text-cyan-600 hover:text-cyan-500 dark:text-cyan-300 dark:hover:text-cyan-200"
|
||||
>
|
||||
Open run
|
||||
<ExternalLink className="h-2.5 w-2.5" />
|
||||
</Link>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
))
|
||||
) : (
|
||||
<div className="flex items-center px-3 py-2 border-b border-border/50">
|
||||
<span className="text-xs font-medium text-muted-foreground">Recent run updates</span>
|
||||
<div className="overflow-hidden rounded-xl border border-cyan-500/25 bg-background/80 shadow-[0_18px_50px_rgba(6,182,212,0.08)]">
|
||||
<div className="border-b border-border/60 bg-cyan-500/[0.04] px-4 py-3">
|
||||
<div className="text-xs font-semibold uppercase tracking-[0.18em] text-cyan-700 dark:text-cyan-300">
|
||||
Live Runs
|
||||
</div>
|
||||
<div className="mt-1 text-xs text-muted-foreground">
|
||||
Streamed with the same transcript UI used on the full run detail page.
|
||||
</div>
|
||||
)}
|
||||
|
||||
<div ref={bodyRef} className="max-h-[220px] overflow-y-auto p-2 font-mono text-[11px] space-y-1">
|
||||
{recent.length === 0 && (
|
||||
<div className="text-xs text-muted-foreground">Waiting for run output...</div>
|
||||
)}
|
||||
{recent.map((item, index) => (
|
||||
<div
|
||||
key={item.id}
|
||||
className={cn(
|
||||
"grid grid-cols-[auto_1fr] gap-2 items-start",
|
||||
index === recent.length - 1 && "animate-in fade-in slide-in-from-bottom-1 duration-300",
|
||||
)}
|
||||
>
|
||||
<span className="text-[10px] text-muted-foreground">{relativeTime(item.ts)}</span>
|
||||
<div className={cn(
|
||||
"min-w-0",
|
||||
item.tone === "error" && "text-red-600 dark:text-red-300",
|
||||
item.tone === "warn" && "text-amber-600 dark:text-amber-300",
|
||||
item.tone === "assistant" && "text-emerald-700 dark:text-emerald-200",
|
||||
item.tone === "tool" && "text-cyan-600 dark:text-cyan-300",
|
||||
item.tone === "info" && "text-foreground/80",
|
||||
)}>
|
||||
<Identity name={item.agentName} size="sm" className="text-cyan-600 dark:text-cyan-400" />
|
||||
<span className="text-muted-foreground"> [{item.runId.slice(0, 8)}] </span>
|
||||
<span className="break-words">{item.text}</span>
|
||||
</div>
|
||||
</div>
|
||||
))}
|
||||
</div>
|
||||
|
||||
<div className="divide-y divide-border/60">
|
||||
{runs.map((run) => {
|
||||
const isActive = isRunActive(run.status);
|
||||
const transcript = transcriptByRun.get(run.id) ?? [];
|
||||
return (
|
||||
<section key={run.id} className="px-4 py-4">
|
||||
<div className="mb-3 flex flex-col gap-3 sm:flex-row sm:items-start sm:justify-between">
|
||||
<div className="min-w-0">
|
||||
<Link to={`/agents/${run.agentId}`} className="inline-flex hover:underline">
|
||||
<Identity name={run.agentName} size="sm" />
|
||||
</Link>
|
||||
<div className="mt-2 flex flex-wrap items-center gap-2 text-xs text-muted-foreground">
|
||||
<Link
|
||||
to={`/agents/${run.agentId}/runs/${run.id}`}
|
||||
className="inline-flex items-center rounded-full border border-border/70 bg-background/70 px-2 py-1 font-mono hover:border-cyan-500/30 hover:text-foreground"
|
||||
>
|
||||
{run.id.slice(0, 8)}
|
||||
</Link>
|
||||
<StatusBadge status={run.status} />
|
||||
<span>{formatDateTime(run.startedAt ?? run.createdAt)}</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="flex items-center gap-2">
|
||||
{isActive && (
|
||||
<button
|
||||
onClick={() => handleCancelRun(run.id)}
|
||||
disabled={cancellingRunIds.has(run.id)}
|
||||
className="inline-flex items-center gap-1 rounded-full border border-red-500/20 bg-red-500/[0.06] px-2.5 py-1 text-[11px] font-medium text-red-700 transition-colors hover:bg-red-500/[0.12] dark:text-red-300 disabled:opacity-50"
|
||||
>
|
||||
<Square className="h-2.5 w-2.5" fill="currentColor" />
|
||||
{cancellingRunIds.has(run.id) ? "Stopping…" : "Stop"}
|
||||
</button>
|
||||
)}
|
||||
<Link
|
||||
to={`/agents/${run.agentId}/runs/${run.id}`}
|
||||
className="inline-flex items-center gap-1 rounded-full border border-border/70 bg-background/70 px-2.5 py-1 text-[11px] font-medium text-cyan-700 transition-colors hover:border-cyan-500/30 hover:text-cyan-600 dark:text-cyan-300"
|
||||
>
|
||||
Open run
|
||||
<ExternalLink className="h-3 w-3" />
|
||||
</Link>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="max-h-[320px] overflow-y-auto pr-1">
|
||||
<RunTranscriptView
|
||||
entries={transcript}
|
||||
density="compact"
|
||||
limit={8}
|
||||
streaming={isActive}
|
||||
collapseStdout
|
||||
emptyMessage={hasOutputForRun(run.id) ? "Waiting for transcript parsing..." : "Waiting for run output..."}
|
||||
/>
|
||||
</div>
|
||||
</section>
|
||||
);
|
||||
})}
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
1000
ui/src/components/transcript/RunTranscriptView.tsx
Normal file
1000
ui/src/components/transcript/RunTranscriptView.tsx
Normal file
File diff suppressed because it is too large
Load Diff
283
ui/src/components/transcript/useLiveRunTranscripts.ts
Normal file
283
ui/src/components/transcript/useLiveRunTranscripts.ts
Normal file
@@ -0,0 +1,283 @@
|
||||
import { useEffect, useMemo, useRef, useState } from "react";
|
||||
import type { LiveEvent } from "@paperclipai/shared";
|
||||
import { heartbeatsApi, type LiveRunForIssue } from "../../api/heartbeats";
|
||||
import { buildTranscript, getUIAdapter, type RunLogChunk, type TranscriptEntry } from "../../adapters";
|
||||
|
||||
const LOG_POLL_INTERVAL_MS = 2000;
|
||||
const LOG_READ_LIMIT_BYTES = 256_000;
|
||||
|
||||
interface UseLiveRunTranscriptsOptions {
|
||||
runs: LiveRunForIssue[];
|
||||
companyId?: string | null;
|
||||
maxChunksPerRun?: number;
|
||||
}
|
||||
|
||||
function readString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
function isTerminalStatus(status: string): boolean {
|
||||
return status === "failed" || status === "timed_out" || status === "cancelled" || status === "succeeded";
|
||||
}
|
||||
|
||||
function parsePersistedLogContent(
|
||||
runId: string,
|
||||
content: string,
|
||||
pendingByRun: Map<string, string>,
|
||||
): Array<RunLogChunk & { dedupeKey: string }> {
|
||||
if (!content) return [];
|
||||
|
||||
const pendingKey = `${runId}:records`;
|
||||
const combined = `${pendingByRun.get(pendingKey) ?? ""}${content}`;
|
||||
const split = combined.split("\n");
|
||||
pendingByRun.set(pendingKey, split.pop() ?? "");
|
||||
|
||||
const parsed: Array<RunLogChunk & { dedupeKey: string }> = [];
|
||||
for (const line of split) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
try {
|
||||
const raw = JSON.parse(trimmed) as { ts?: unknown; stream?: unknown; chunk?: unknown };
|
||||
const stream = raw.stream === "stderr" || raw.stream === "system" ? raw.stream : "stdout";
|
||||
const chunk = typeof raw.chunk === "string" ? raw.chunk : "";
|
||||
const ts = typeof raw.ts === "string" ? raw.ts : new Date().toISOString();
|
||||
if (!chunk) continue;
|
||||
parsed.push({
|
||||
ts,
|
||||
stream,
|
||||
chunk,
|
||||
dedupeKey: `log:${runId}:${ts}:${stream}:${chunk}`,
|
||||
});
|
||||
} catch {
|
||||
// Ignore malformed log rows.
|
||||
}
|
||||
}
|
||||
|
||||
return parsed;
|
||||
}
|
||||
|
||||
export function useLiveRunTranscripts({
|
||||
runs,
|
||||
companyId,
|
||||
maxChunksPerRun = 200,
|
||||
}: UseLiveRunTranscriptsOptions) {
|
||||
const [chunksByRun, setChunksByRun] = useState<Map<string, RunLogChunk[]>>(new Map());
|
||||
const seenChunkKeysRef = useRef(new Set<string>());
|
||||
const pendingLogRowsByRunRef = useRef(new Map<string, string>());
|
||||
const logOffsetByRunRef = useRef(new Map<string, number>());
|
||||
|
||||
const runById = useMemo(() => new Map(runs.map((run) => [run.id, run])), [runs]);
|
||||
const activeRunIds = useMemo(
|
||||
() => new Set(runs.filter((run) => !isTerminalStatus(run.status)).map((run) => run.id)),
|
||||
[runs],
|
||||
);
|
||||
const runIdsKey = useMemo(
|
||||
() => runs.map((run) => run.id).sort((a, b) => a.localeCompare(b)).join(","),
|
||||
[runs],
|
||||
);
|
||||
|
||||
const appendChunks = (runId: string, chunks: Array<RunLogChunk & { dedupeKey: string }>) => {
|
||||
if (chunks.length === 0) return;
|
||||
setChunksByRun((prev) => {
|
||||
const next = new Map(prev);
|
||||
const existing = [...(next.get(runId) ?? [])];
|
||||
let changed = false;
|
||||
|
||||
for (const chunk of chunks) {
|
||||
if (seenChunkKeysRef.current.has(chunk.dedupeKey)) continue;
|
||||
seenChunkKeysRef.current.add(chunk.dedupeKey);
|
||||
existing.push({ ts: chunk.ts, stream: chunk.stream, chunk: chunk.chunk });
|
||||
changed = true;
|
||||
}
|
||||
|
||||
if (!changed) return prev;
|
||||
if (seenChunkKeysRef.current.size > 12000) {
|
||||
seenChunkKeysRef.current.clear();
|
||||
}
|
||||
next.set(runId, existing.slice(-maxChunksPerRun));
|
||||
return next;
|
||||
});
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
const knownRunIds = new Set(runs.map((run) => run.id));
|
||||
setChunksByRun((prev) => {
|
||||
const next = new Map<string, RunLogChunk[]>();
|
||||
for (const [runId, chunks] of prev) {
|
||||
if (knownRunIds.has(runId)) {
|
||||
next.set(runId, chunks);
|
||||
}
|
||||
}
|
||||
return next.size === prev.size ? prev : next;
|
||||
});
|
||||
|
||||
for (const key of pendingLogRowsByRunRef.current.keys()) {
|
||||
const runId = key.replace(/:records$/, "");
|
||||
if (!knownRunIds.has(runId)) {
|
||||
pendingLogRowsByRunRef.current.delete(key);
|
||||
}
|
||||
}
|
||||
for (const runId of logOffsetByRunRef.current.keys()) {
|
||||
if (!knownRunIds.has(runId)) {
|
||||
logOffsetByRunRef.current.delete(runId);
|
||||
}
|
||||
}
|
||||
}, [runs]);
|
||||
|
||||
useEffect(() => {
|
||||
if (runs.length === 0) return;
|
||||
|
||||
let cancelled = false;
|
||||
|
||||
const readRunLog = async (run: LiveRunForIssue) => {
|
||||
const offset = logOffsetByRunRef.current.get(run.id) ?? 0;
|
||||
try {
|
||||
const result = await heartbeatsApi.log(run.id, offset, LOG_READ_LIMIT_BYTES);
|
||||
if (cancelled) return;
|
||||
|
||||
appendChunks(run.id, parsePersistedLogContent(run.id, result.content, pendingLogRowsByRunRef.current));
|
||||
|
||||
if (result.nextOffset !== undefined) {
|
||||
logOffsetByRunRef.current.set(run.id, result.nextOffset);
|
||||
return;
|
||||
}
|
||||
if (result.content.length > 0) {
|
||||
logOffsetByRunRef.current.set(run.id, offset + result.content.length);
|
||||
}
|
||||
} catch {
|
||||
// Ignore log read errors while output is initializing.
|
||||
}
|
||||
};
|
||||
|
||||
const readAll = async () => {
|
||||
await Promise.all(runs.map((run) => readRunLog(run)));
|
||||
};
|
||||
|
||||
void readAll();
|
||||
const interval = window.setInterval(() => {
|
||||
void readAll();
|
||||
}, LOG_POLL_INTERVAL_MS);
|
||||
|
||||
return () => {
|
||||
cancelled = true;
|
||||
window.clearInterval(interval);
|
||||
};
|
||||
}, [runIdsKey, runs]);
|
||||
|
||||
useEffect(() => {
|
||||
if (!companyId || activeRunIds.size === 0) return;
|
||||
|
||||
let closed = false;
|
||||
let reconnectTimer: number | null = null;
|
||||
let socket: WebSocket | null = null;
|
||||
|
||||
const scheduleReconnect = () => {
|
||||
if (closed) return;
|
||||
reconnectTimer = window.setTimeout(connect, 1500);
|
||||
};
|
||||
|
||||
const connect = () => {
|
||||
if (closed) return;
|
||||
const protocol = window.location.protocol === "https:" ? "wss" : "ws";
|
||||
const url = `${protocol}://${window.location.host}/api/companies/${encodeURIComponent(companyId)}/events/ws`;
|
||||
socket = new WebSocket(url);
|
||||
|
||||
socket.onmessage = (message) => {
|
||||
const raw = typeof message.data === "string" ? message.data : "";
|
||||
if (!raw) return;
|
||||
|
||||
let event: LiveEvent;
|
||||
try {
|
||||
event = JSON.parse(raw) as LiveEvent;
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.companyId !== companyId) return;
|
||||
const payload = event.payload ?? {};
|
||||
const runId = readString(payload["runId"]);
|
||||
if (!runId || !activeRunIds.has(runId)) return;
|
||||
if (!runById.has(runId)) return;
|
||||
|
||||
if (event.type === "heartbeat.run.log") {
|
||||
const chunk = readString(payload["chunk"]);
|
||||
if (!chunk) return;
|
||||
const ts = readString(payload["ts"]) ?? event.createdAt;
|
||||
const stream =
|
||||
readString(payload["stream"]) === "stderr"
|
||||
? "stderr"
|
||||
: readString(payload["stream"]) === "system"
|
||||
? "system"
|
||||
: "stdout";
|
||||
appendChunks(runId, [{
|
||||
ts,
|
||||
stream,
|
||||
chunk,
|
||||
dedupeKey: `log:${runId}:${ts}:${stream}:${chunk}`,
|
||||
}]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "heartbeat.run.event") {
|
||||
const seq = typeof payload["seq"] === "number" ? payload["seq"] : null;
|
||||
const eventType = readString(payload["eventType"]) ?? "event";
|
||||
const messageText = readString(payload["message"]) ?? eventType;
|
||||
appendChunks(runId, [{
|
||||
ts: event.createdAt,
|
||||
stream: eventType === "error" ? "stderr" : "system",
|
||||
chunk: messageText,
|
||||
dedupeKey: `socket:event:${runId}:${seq ?? `${eventType}:${messageText}:${event.createdAt}`}`,
|
||||
}]);
|
||||
return;
|
||||
}
|
||||
|
||||
if (event.type === "heartbeat.run.status") {
|
||||
const status = readString(payload["status"]) ?? "updated";
|
||||
appendChunks(runId, [{
|
||||
ts: event.createdAt,
|
||||
stream: isTerminalStatus(status) && status !== "succeeded" ? "stderr" : "system",
|
||||
chunk: `run ${status}`,
|
||||
dedupeKey: `socket:status:${runId}:${status}:${readString(payload["finishedAt"]) ?? ""}`,
|
||||
}]);
|
||||
}
|
||||
};
|
||||
|
||||
socket.onerror = () => {
|
||||
socket?.close();
|
||||
};
|
||||
|
||||
socket.onclose = () => {
|
||||
scheduleReconnect();
|
||||
};
|
||||
};
|
||||
|
||||
connect();
|
||||
|
||||
return () => {
|
||||
closed = true;
|
||||
if (reconnectTimer !== null) window.clearTimeout(reconnectTimer);
|
||||
if (socket) {
|
||||
socket.onmessage = null;
|
||||
socket.onerror = null;
|
||||
socket.onclose = null;
|
||||
socket.close(1000, "live_run_transcripts_unmount");
|
||||
}
|
||||
};
|
||||
}, [activeRunIds, companyId, runById]);
|
||||
|
||||
const transcriptByRun = useMemo(() => {
|
||||
const next = new Map<string, TranscriptEntry[]>();
|
||||
for (const run of runs) {
|
||||
const adapter = getUIAdapter(run.adapterType);
|
||||
next.set(run.id, buildTranscript(chunksByRun.get(run.id) ?? [], adapter.parseStdoutLine));
|
||||
}
|
||||
return next;
|
||||
}, [chunksByRun, runs]);
|
||||
|
||||
return {
|
||||
transcriptByRun,
|
||||
hasOutputForRun(runId: string) {
|
||||
return (chunksByRun.get(runId)?.length ?? 0) > 0;
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user