fix(FN-3187): stabilize agent run log streaming

2026-05-01 18:37:28 -07:00
parent 2f9364460a
commit a63e819ad0
6 changed files with 350 additions and 34 deletions
--- a/packages/core/src/tests/agent-store.test.ts
+++ b/packages/core/src/tests/agent-store.test.ts
@@ -2040,6 +2040,27 @@ describe("AgentStore", () => {
      const completed = await store.getCompletedHeartbeatRuns(agent.id);
      expect(completed.some((r) => r.id === structuredRun.id)).toBe(true);
    });
+
+    it("appendRunLog emits run:log and persists the entry", async () => {
+      const agent = await store.createAgent({ name: "RunLogger", role: "executor" });
+      const run = await store.startHeartbeatRun(agent.id);
+      const onRunLog = vi.fn();
+      store.on("run:log", onRunLog);
+
+      const entry = {
+        timestamp: "2026-01-01T00:00:00.000Z",
+        taskId: "agent-run",
+        text: "streamed output",
+        type: "text" as const,
+      };
+
+      await store.appendRunLog(agent.id, run.id, entry);
+
+      expect(onRunLog).toHaveBeenCalledWith(agent.id, run.id, expect.objectContaining(entry));
+      await expect(store.getRunLogs(agent.id, run.id)).resolves.toEqual([
+        expect.objectContaining(entry),
+      ]);
+    });
  });

  // ── blocked state persistence ─────────────────────────────────────
--- a/packages/core/src/agent-store.ts
+++ b/packages/core/src/agent-store.ts
@@ -88,6 +88,8 @@ export interface AgentStoreEvents {
  "agent:assigned": (agent: Agent, taskId: string) => void;
  /** Emitted when a rating is added */
  "rating:added": (rating: AgentRating) => void;
+  /** Emitted when a log entry is appended to a run's JSONL log. */
+  "run:log": (agentId: string, runId: string, entry: AgentLogEntry) => void;
 }

 /** Options for AgentStore constructor */
@@ -1838,6 +1840,7 @@ export class AgentStore extends EventEmitter {
    };
    const line = JSON.stringify(safeEntry) + "\n";
    await appendFile(this.runLogPath(agentId, runId), line, "utf-8");
+    this.emit("run:log", agentId, runId, safeEntry);
  }

  /**
--- a/packages/dashboard/app/components/AgentDetailView.tsx
+++ b/packages/dashboard/app/components/AgentDetailView.tsx
@@ -126,6 +126,7 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
  const [activeTab, setActiveTab] = useState<TabId>("dashboard");
  const [isStreaming, setIsStreaming] = useState(false);
  const [isTransitioning, setIsTransitioning] = useState(false);
+  const [latestRun, setLatestRun] = useState<AgentHeartbeatRun | null>(null);
  const logContainerRef = useRef<HTMLDivElement>(null);
  const agentDetailModalRef = useRef<HTMLDivElement>(null);
  const overlayMouseDownRef = useRef(false);
@@ -134,6 +135,7 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
  const addToastRef = useRef(addToast);
  const agentRef = useRef<AgentDetail | null>(null);
  const hasConfigChangesRef = useRef(false);
+  const loadedLatestRunLogsRef = useRef<string | null>(null);

  // Track the context version to detect stale events after project/agent switches.
  // Incremented whenever agentId or projectId changes, invalidating any in-flight SSE handlers.
@@ -168,29 +170,42 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
    const currentAgentId = agentId;
    const currentProjectId = projectId;

-    // Agent logs are tied to tasks, not agents directly.
-    // If the agent has a current task, we could show those logs.
-    // For now, we'll show heartbeat runs as the "activity" for the agent.
-    // If the agent is working on a task, we could show task logs.
-    if (agent?.taskId) {
-      try {
+    const isStale = () =>
+      contextVersionRef.current !== contextVersionAtCapture ||
+      agentId !== currentAgentId ||
+      projectId !== currentProjectId;
+
+    try {
+      if (agent?.taskId) {
+        setLatestRun(null);
+        loadedLatestRunLogsRef.current = null;
        const result = await fetchAgentLogsWithMeta(agent.taskId, currentProjectId, { limit: 100 });
-        // Reject stale response: check context version and current IDs
-        if (contextVersionRef.current !== contextVersionAtCapture ||
-            agentId !== currentAgentId ||
-            projectId !== currentProjectId) {
-          return;
-        }
+        if (isStale()) return;
        setLogs(result.entries);
-      } catch (err) {
-        // Reject stale error: check context version and current IDs
-        if (contextVersionRef.current !== contextVersionAtCapture ||
-            agentId !== currentAgentId ||
-            projectId !== currentProjectId) {
-          return;
-        }
-        console.error("Failed to load task logs:", err);
+        return;
      }
+
+      // Fallback: show the latest run's logs so the Logs tab is populated even
+      // when no task is currently assigned.
+      const runs = await fetchAgentRuns(currentAgentId, 1, currentProjectId);
+      if (isStale()) return;
+      const latest = runs[0] ?? null;
+      setLatestRun(latest);
+      if (!latest) {
+        loadedLatestRunLogsRef.current = null;
+        setLogs([]);
+        return;
+      }
+      if (loadedLatestRunLogsRef.current === latest.id) {
+        return;
+      }
+      const entries = await fetchAgentRunLogs(currentAgentId, latest.id, currentProjectId);
+      if (isStale()) return;
+      setLogs([...entries].reverse());
+      loadedLatestRunLogsRef.current = latest.id;
+    } catch (err) {
+      if (isStale()) return;
+      console.error("Failed to load agent logs:", err);
    }
  }, [agent?.taskId, agentId, projectId]);

@@ -215,10 +230,62 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
  }, [loadAgent]);

  useEffect(() => {
-    if (agent?.taskId) {
+    if (agent && activeTab === "logs") {
      void loadLogs();
    }
-  }, [agent?.taskId, loadLogs]);
+  }, [agent, activeTab, loadLogs]);
+
+  useEffect(() => {
+    if (activeTab !== "logs") {
+      loadedLatestRunLogsRef.current = null;
+    }
+  }, [activeTab]);
+
+  // When falling back to latest-run logs (no taskId) and that run is active,
+  // subscribe to the run-scoped SSE stream so the Logs tab tails updates.
+  useEffect(() => {
+    if (activeTab !== "logs" || agent?.taskId) return;
+    if (!latestRun || latestRun.status !== "active") return;
+
+    const contextVersionAtStart = contextVersionRef.current;
+    const currentAgentId = agentId;
+    const currentRunId = latestRun.id;
+    const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
+
+    const unsubscribe = subscribeSse(
+      `/api/agents/${encodeURIComponent(currentAgentId)}/runs/${encodeURIComponent(currentRunId)}/logs/stream${query}`,
+      {
+        events: {
+          "agent:log": (e) => {
+            if (contextVersionRef.current !== contextVersionAtStart) return;
+            try {
+              const entry: AgentLogEntry = JSON.parse(e.data);
+              setLogs(prev => [entry, ...prev]);
+            } catch {
+              // ignore malformed events
+            }
+          },
+        },
+        onOpen: () => {
+          if (contextVersionRef.current === contextVersionAtStart) {
+            setIsStreaming(true);
+          }
+        },
+        onError: () => {
+          if (contextVersionRef.current === contextVersionAtStart) {
+            setIsStreaming(false);
+          }
+        },
+      },
+    );
+
+    return () => {
+      unsubscribe();
+      if (contextVersionRef.current === contextVersionAtStart) {
+        setIsStreaming(false);
+      }
+    };
+  }, [activeTab, agent?.taskId, agentId, projectId, latestRun]);

  // Detect context changes (agentId or projectId) and invalidate stale handlers
  useEffect(() => {
@@ -230,6 +297,8 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
      // Clear stale logs and streaming state immediately
      setLogs([]);
      setIsStreaming(false);
+      setLatestRun(null);
+      loadedLatestRunLogsRef.current = null;
      hasConfigChangesRef.current = false;
    }
  }, [agentId, projectId]);
@@ -530,11 +599,12 @@ export function AgentDetailView({ agentId, projectId, onClose, addToast, onChild
          )}
          
          {activeTab === "logs" && (
-            <LogsTab 
-              logs={logs} 
+            <LogsTab
+              logs={logs}
              isStreaming={isStreaming}
              containerRef={logContainerRef}
-              hasTask={!!agent.taskId}
+              hasTask={!!agent.taskId || logs.length > 0 || latestRun !== null}
+              fallbackLabel={!agent.taskId && latestRun ? `Latest run · ${latestRun.id.slice(0, 8)}` : null}
            />
          )}
          
@@ -945,25 +1015,27 @@ function DashboardTab({

 // ── Logs Tab ──────────────────────────────────────────────────────────────

-function LogsTab({ 
-  logs, 
+function LogsTab({
+  logs,
  isStreaming,
  containerRef,
-  hasTask 
-}: { 
-  logs: AgentLogEntry[]; 
+  hasTask,
+  fallbackLabel,
+}: {
+  logs: AgentLogEntry[];
  isStreaming: boolean;
  containerRef: React.RefObject<HTMLDivElement | null>;
  hasTask: boolean;
+  fallbackLabel?: string | null;
 }) {
  if (!hasTask) {
    return (
      <div className="logs-tab">
        <div className="logs-empty">
          <FileText size={48} opacity={0.3} />
-          <p>No task assigned</p>
+          <p>No activity yet</p>
          <p className="text-muted">
-            Agent logs are available when the agent is assigned to a task
+            Agent logs will appear here from the current task or most recent run
          </p>
        </div>
      </div>
@@ -974,6 +1046,9 @@ function LogsTab({
    <div className="logs-tab">
      <div className="logs-header">
        <span className="logs-count">{logs.length} entries</span>
+        {fallbackLabel && (
+          <span className="text-muted" style={{ fontSize: "12px" }}>{fallbackLabel}</span>
+        )}
        {isStreaming && (
          <span className="streaming-indicator">
            <span className="streaming-dot" />
@@ -1105,6 +1180,9 @@ function RunsTab({

  // Poll for active runs
  const hasActiveRun = runs.some(r => r.status === "active");
+  const selectedRunStatus = selectedRunId
+    ? runs.find((run) => run.id === selectedRunId)?.status
+    : undefined;
  useEffect(() => {
    if (!hasActiveRun) return;
    const interval = setInterval(() => {
@@ -1113,6 +1191,31 @@ function RunsTab({
    return () => clearInterval(interval);
  }, [hasActiveRun, loadRuns]);

+  // While a selected run is still active, subscribe to its log stream so the
+  // expanded view tails updates without a refresh.  Mirrors the per-task log
+  // SSE pattern in useAgentLogs.
+  useEffect(() => {
+    if (!selectedRunId) return;
+    if (selectedRunStatus !== "active") return;
+
+    const query = projectId ? `?projectId=${encodeURIComponent(projectId)}` : "";
+    return subscribeSse(
+      `/api/agents/${encodeURIComponent(agentId)}/runs/${encodeURIComponent(selectedRunId)}/logs/stream${query}`,
+      {
+        events: {
+          "agent:log": (e) => {
+            try {
+              const entry: AgentLogEntry = JSON.parse(e.data);
+              setRunLogs(prev => [...prev, entry]);
+            } catch {
+              // ignore malformed events
+            }
+          },
+        },
+      },
+    );
+  }, [selectedRunId, selectedRunStatus, agentId, projectId]);
+
  // Load run detail when a run is selected
  const handleRunClick = useCallback(async (runId: string) => {
    if (selectedRunId === runId) {
--- a/packages/dashboard/app/components/tests/AgentDetailView.test.tsx
+++ b/packages/dashboard/app/components/tests/AgentDetailView.test.tsx
@@ -4,7 +4,7 @@ import { render, screen, waitFor, fireEvent, act } from "@testing-library/react"
 import userEvent from "@testing-library/user-event";
 import "@testing-library/jest-dom";
 import { AgentDetailView } from "../AgentDetailView";
-import type { AgentCapability, AgentDetail } from "../../api";
+import type { AgentCapability, AgentDetail, AgentHeartbeatRun } from "../../api";
 import type { AgentLogEntry } from "@fusion/core";
 import { DEFAULT_HEARTBEAT_INTERVAL_MS } from "../../utils/heartbeatIntervals";

@@ -93,6 +93,10 @@ vi.mock("../SkillMultiselect", () => ({
  ),
 }));

+vi.mock("../../sse-bus", () => ({
+  subscribeSse: vi.fn(() => () => {}),
+}));
+
 const mockConfirm = vi.fn();

 vi.mock("../../hooks/useConfirm", () => ({
@@ -100,6 +104,7 @@ vi.mock("../../hooks/useConfirm", () => ({
 }));

 import { fetchAgent, fetchAgents, updateAgent, updateAgentState, deleteAgent, fetchAgentChildren, fetchAgentRunLogs, fetchAgentRuns, fetchAgentRunDetail, fetchAgentTasks, fetchChainOfCommand, fetchAgentBudgetStatus, resetAgentBudget, updateAgentInstructions, updateAgentSoul, updateAgentMemory, fetchWorkspaceFileContent, saveWorkspaceFileContent, fetchDiscoveredSkills, fetchModels, fetchPluginRuntimes, fetchAgentLogsWithMeta, upgradeAgentHeartbeatProcedure } from "../../api";
+import { subscribeSse } from "../../sse-bus";

 const mockFetchAgent = vi.mocked(fetchAgent);
 const mockFetchAgents = vi.mocked(fetchAgents);
@@ -124,6 +129,7 @@ const mockFetchModels = vi.mocked(fetchModels);
 const mockFetchPluginRuntimes = vi.mocked(fetchPluginRuntimes);
 const mockFetchAgentLogsWithMeta = vi.mocked(fetchAgentLogsWithMeta);
 const mockUpgradeAgentHeartbeatProcedure = vi.mocked(upgradeAgentHeartbeatProcedure);
+const mockSubscribeSse = vi.mocked(subscribeSse);

 const MOCK_SKILLS = [
  { id: "skill-1", name: "Skill One", path: "/path/skill-1", relativePath: "skills/skill-1", enabled: true, metadata: { source: "*", scope: "user" as const, origin: "top-level" as const } },
@@ -166,6 +172,8 @@ describe("AgentDetailView", () => {
    vi.clearAllMocks();
    mockConfirm.mockReset();
    mockConfirm.mockResolvedValue(true);
+    mockSubscribeSse.mockReset();
+    mockSubscribeSse.mockReturnValue(vi.fn());
    const mockAgent = createMockAgent();
    mockFetchAgent.mockResolvedValue(mockAgent);
    mockFetchAgents.mockResolvedValue([
@@ -181,6 +189,7 @@ describe("AgentDetailView", () => {
      ...(mockAgent.activeRun ? [mockAgent.activeRun] : []),
      ...mockAgent.completedRuns,
    ]);
+    mockFetchAgentRunLogs.mockResolvedValue([]);
    mockFetchAgentRunDetail.mockResolvedValue(mockAgent.completedRuns[0]);
    mockFetchAgentChildren.mockResolvedValue([]);
    mockFetchAgentTasks.mockResolvedValue([]);
@@ -1055,6 +1064,55 @@ describe("AgentDetailView", () => {
    });
  });

+  describe("Logs tab", () => {
+    it("loads latest run logs lazily for agents without a current task", async () => {
+      const latestRun = {
+        id: "run-1001",
+        agentId: "agent-001",
+        startedAt: "2024-01-01T00:00:00.000Z",
+        endedAt: null,
+        status: "active",
+      } as AgentHeartbeatRun;
+      mockFetchAgent.mockResolvedValue(createMockAgent({
+        taskId: undefined,
+        activeRun: latestRun,
+        completedRuns: [],
+      }));
+      mockFetchAgentRuns.mockResolvedValue([latestRun]);
+      mockFetchAgentRunLogs.mockResolvedValue([
+        { timestamp: "2024-01-01T00:01:00.000Z", taskId: "agent-run", text: "First entry", type: "text" },
+        { timestamp: "2024-01-01T00:02:00.000Z", taskId: "agent-run", text: "Second entry", type: "text" },
+      ]);
+
+      render(
+        <AgentDetailView
+          agentId="agent-001"
+          onClose={vi.fn()}
+          addToast={vi.fn()}
+        />
+      );
+
+      await waitFor(() => {
+        expect(screen.getByText("Dashboard")).toBeInTheDocument();
+      });
+
+      expect(mockFetchAgentRuns).not.toHaveBeenCalled();
+      expect(mockFetchAgentRunLogs).not.toHaveBeenCalled();
+
+      fireEvent.click(screen.getByText("Logs"));
+
+      await waitFor(() => {
+        expect(mockFetchAgentRuns).toHaveBeenCalledWith("agent-001", 1, undefined);
+        expect(mockFetchAgentRunLogs).toHaveBeenCalledWith("agent-001", "run-1001", undefined);
+      });
+
+      expect(screen.getByText("Latest run · run-1001")).toBeInTheDocument();
+      expect(
+        Array.from(document.querySelectorAll(".log-text")).map((node) => node.textContent?.trim()),
+      ).toEqual(["Second entry", "First entry"]);
+    });
+  });
+
  describe("Tasks tab", () => {
    it("renders tasks returned by fetchAgentTasks", async () => {
      const user = userEvent.setup();
@@ -2625,6 +2683,84 @@ describe("AgentDetailView", () => {
      });
    });

+    it("keeps the active run log stream subscribed across run-list polling", async () => {
+      const intervalCallbacks: Array<() => void> = [];
+      const setIntervalSpy = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: TimerHandler) => {
+        if (typeof callback === "function") {
+          intervalCallbacks.push(callback as () => void);
+        }
+        return 1 as ReturnType<typeof setInterval>;
+      }) as typeof setInterval);
+      const clearIntervalSpy = vi.spyOn(globalThis, "clearInterval").mockImplementation(((id?: ReturnType<typeof setInterval>) => {
+        void id;
+      }) as typeof clearInterval);
+
+      try {
+        const activeRun = {
+          id: "run-live-1",
+          agentId: "agent-001",
+          startedAt: "2024-01-01T00:00:00.000Z",
+          endedAt: null,
+          status: "active",
+        } as AgentHeartbeatRun;
+        mockFetchAgent.mockResolvedValue(createMockAgent({
+          activeRun,
+          completedRuns: [],
+        }));
+        mockFetchAgentRuns.mockResolvedValue([activeRun]);
+        mockFetchAgentRunLogs.mockResolvedValue([]);
+        mockFetchAgentRunDetail.mockResolvedValue(activeRun);
+
+        render(
+          <AgentDetailView
+            agentId="agent-001"
+            onClose={vi.fn()}
+            addToast={vi.fn()}
+          />
+        );
+
+        await waitFor(() => {
+          expect(screen.getByText("Runs")).toBeInTheDocument();
+        });
+        fireEvent.click(screen.getByText("Runs"));
+
+        await waitFor(() => {
+          expect(screen.getByText("Live Run")).toBeInTheDocument();
+        });
+
+        const activeRunButton = screen.getAllByRole("button").find(
+          (btn) => btn.getAttribute("aria-label")?.includes("run-live")
+            && btn.getAttribute("aria-label")?.includes("active"),
+        );
+        expect(activeRunButton).toBeTruthy();
+        fireEvent.click(activeRunButton!);
+
+        await waitFor(() => {
+          expect(mockFetchAgentRunLogs).toHaveBeenCalledWith("agent-001", "run-live-1", undefined);
+        });
+
+        const streamUrl = "/api/agents/agent-001/runs/run-live-1/logs/stream";
+        expect(
+          mockSubscribeSse.mock.calls.filter(([url]) => url === streamUrl),
+        ).toHaveLength(1);
+
+        await act(async () => {
+          intervalCallbacks.forEach((callback) => callback());
+          await Promise.resolve();
+        });
+
+        await waitFor(() => {
+          expect(mockFetchAgentRuns.mock.calls.length).toBeGreaterThanOrEqual(2);
+        });
+        expect(
+          mockSubscribeSse.mock.calls.filter(([url]) => url === streamUrl),
+        ).toHaveLength(1);
+      } finally {
+        setIntervalSpy.mockRestore();
+        clearIntervalSpy.mockRestore();
+      }
+    });
+
    it("fetches and displays logs when clicking a completed run", async () => {
      const mockLogs: AgentLogEntry[] = [
        { timestamp: "2024-01-01T00:01:00.000Z", taskId: "FN-001", text: "Starting task execution", type: "text" },
--- a/packages/dashboard/src/server.ts
+++ b/packages/dashboard/src/server.ts
@@ -5,7 +5,7 @@ import { existsSync, readFileSync } from "node:fs";
 import { fileURLToPath } from "node:url";
 import { createSecureServer as createHttp2SecureServer, type Http2SecureServer } from "node:http2";
 import type { Server as HttpServer } from "node:http";
-import type { Task, TaskStore, MergeResult, AutomationStore, RoutineStore, CentralCore, MessageStore } from "@fusion/core";
+import type { Task, TaskStore, MergeResult, AutomationStore, RoutineStore, CentralCore, MessageStore, AgentLogEntry } from "@fusion/core";
 import { AgentStore, ChatStore } from "@fusion/core";
 import type { AuthStorageLike, ModelRegistryLike } from "./routes.js";
 import { createApiRoutes } from "./routes.js";
@@ -717,6 +717,54 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT
    });
  });

+  // Per-run SSE endpoint for live agent log streaming.
+  // Mirrors the per-task endpoint above but subscribes to AgentStore's
+  // "run:log" event (emitted from AgentStore.appendRunLog) and filters by
+  // agentId + runId.  We need the engine's AgentStore instance specifically,
+  // since that's the EventEmitter the heartbeat runtime writes to — a fresh
+  // store created here would never receive events.
+  app.get("/api/agents/:id/runs/:runId/logs/stream", async (req, res) => {
+    const agentId = req.params.id;
+    const runId = req.params.runId;
+    const projectId = typeof req.query.projectId === "string" ? req.query.projectId : undefined;
+
+    res.setHeader("Content-Type", "text/event-stream");
+    res.setHeader("Cache-Control", "no-cache");
+    res.setHeader("Connection", "keep-alive");
+    res.setHeader("X-Accel-Buffering", "no");
+    res.flushHeaders();
+
+    res.write(": connected\n\n");
+
+    const engineManager = options?.engineManager;
+    const engine = engineManager && projectId ? engineManager.getEngine(projectId) : options?.engine;
+    const agentStore = engine?.getAgentStore();
+
+    if (!agentStore) {
+      // No live engine — there is no event source to subscribe to. Close
+      // gracefully so the client falls back to its initial fetch.
+      res.write(`event: error\ndata: ${JSON.stringify({ message: "No active engine for project" })}\n\n`);
+      res.end();
+      return;
+    }
+
+    const onRunLog = (eventAgentId: string, eventRunId: string, entry: AgentLogEntry) => {
+      if (eventAgentId !== agentId || eventRunId !== runId) return;
+      res.write(`event: agent:log\ndata: ${JSON.stringify(entry)}\n\n`);
+    };
+
+    agentStore.on("run:log", onRunLog);
+
+    const heartbeat = setInterval(() => {
+      res.write(": heartbeat\n\n");
+    }, 30_000);
+
+    req.on("close", () => {
+      clearInterval(heartbeat);
+      agentStore.off("run:log", onRunLog);
+    });
+  });
+
  // Legacy Terminal SSE endpoint (deprecated, use WebSocket instead)
  app.get("/api/terminal/sessions/:id/stream", rateLimit(RATE_LIMITS.sse), (req, res) => {
    const sessionId = Array.isArray(req.params.id) ? req.params.id[0] : req.params.id;