From 88f691807daac7eb1d80fb875d12fadd1a0e4261 Mon Sep 17 00:00:00 2001 From: Adil Shaikh Date: Thu, 30 Apr 2026 01:07:02 +0530 Subject: [PATCH 1/3] fix(core): preserve reaction tracker across status oscillation (#1531) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(core): preserve reaction tracker across status oscillation (#1409) The reaction tracker retry budget was resetting on every status exit, allowing infinite CI failure and merge conflict messages instead of escalating to a human after the configured retry limit. * fix(core): hoist PERSISTENT_REACTION_KEYS to module scope, document notify priority change Address review feedback: - Move PERSISTENT_REACTION_KEYS to module level (avoids per-call Set allocation) - Document that executeReaction's notify path defaults to "info" priority (was "warning" in old direct-dispatch code); users with action:"notify" should set priority explicitly in their config * fix(core): address review feedback on reaction tracker oscillation 1. Remove dead "merge-conflicts" from PERSISTENT_REACTION_KEYS — statusToEventType never emits "merge.conflicts" so the transition handler can never reach it. Merge-conflict tracker lifecycle is managed in maybeDispatchMergeConflicts. 2. Fix escalation poisoning dedup flag — only set lastMergeConflictDispatched when the result is non-escalated. Escalation hands off to the human; the dedup flag must not suppress future agent dispatches. 3. Add incident boundary to persistent trackers — clear tracker after escalation (executeReaction) and when merge-conflicts resolve. Prevents permanent budget exhaustion in long-lived sessions. 4. Preserve "warning" priority for merge-conflict notify action — default priority to "warning" on enrichedConfig, matching old direct-dispatch behavior. Fixes review comments on #1531. * fix(core): bound ci-failed escalation — silence after escalate, reset on stable CI pass After escalation, the ci-failed tracker is now marked escalated=true instead of being deleted. This prevents the infinite re-escalation loop introduced by deleting the tracker (every 3rd oscillation cycle got a fresh budget, causing retries:2 to produce unbounded agent messages + human pages). Resolution: once escalated, the tracker short-circuits all further dispatches until CI has been passing for CI_PASSING_STABLE_THRESHOLD (2) consecutive polls. This ensures "stable passing" isn't confused with brief pending→passing flicker. Fixes the residual unbounded-dispatch bug flagged in the illegalcall review on #1531. * fix(core): count only 'passing' toward ci stable window — exclude 'pending' 'pending' (emitted while a CI run is in progress) was incorrectly counting toward the 2-poll stable-passing threshold, wiping the escalated tracker between failures in the exact production scenario the fix was meant to bound. With real GitHub CI, every transition out of 'failing' goes through 'pending' while the new check-run starts (~60 polls for a 5-min CI run). Two consecutive pending polls (10s) would clear the tracker before the run completed, giving each failure cycle a fresh budget — identical to the original #1409 symptom. Fix: require ciStatus === "passing" (not !== "failing") before incrementing stableCount. Pending/none reset the stability window the same as failing. Adds two regression tests: - pending CI does not count toward ci-failed tracker resolution - only passing CI resets ci-failed tracker — pending mid-run does not interfere --- .../src/__tests__/lifecycle-manager.test.ts | 552 ++++++++++++++++++ packages/core/src/lifecycle-manager.ts | 100 +++- 2 files changed, 632 insertions(+), 20 deletions(-) diff --git a/packages/core/src/__tests__/lifecycle-manager.test.ts b/packages/core/src/__tests__/lifecycle-manager.test.ts index fc3c35b63..a47582f23 100644 --- a/packages/core/src/__tests__/lifecycle-manager.test.ts +++ b/packages/core/src/__tests__/lifecycle-manager.test.ts @@ -2680,6 +2680,558 @@ describe("reactions", () => { ); expect(alertsNotifier.notify).not.toHaveBeenCalled(); }); + + it("CI failure tracker survives status oscillation and escalates after retries", async () => { + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing. Fix it.", + retries: 2, + escalateAfter: 2, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Oscillation 1: pr_open → ci_failed (attempt 1 — send to agent) + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("ci_failed"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI starts passing → ci_failed → pr_open (tracker survives) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("pr_open"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + + // Oscillation 2: pr_open → ci_failed (attempt 2 — send to agent) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("ci_failed"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI passes again + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + + // Oscillation 3: pr_open → ci_failed (attempt 3 > retries:2 — escalate) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + vi.mocked(notifier.notify).mockClear(); + await lm.check("app-1"); + + // Should NOT send to agent — should escalate to human + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + + // After escalation, tracker is marked escalated — needs 2 stable passing polls to clear + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); // stableCount = 1 + await lm.check("app-1"); // stableCount = 2 → clearReactionTracker + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(notifier.notify).mockClear(); + + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + + // Fresh budget — sends to agent (attempt 1 again), not escalate + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("merge conflict tracker resets on resolve — recurrence gets fresh budget", async () => { + const notifier = createMockNotifier(); + + config.reactions = { + "merge-conflicts": { + auto: true, + action: "send-to-agent", + message: "Resolve merge conflicts.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ hasConflicts: true }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // First conflict — dispatched to agent (attempt 1) + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // Conflicts resolve — tracker clears (incident boundary) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ hasConflicts: false }), + ); + await lm.check("app-1"); + const metadata = readMetadataRaw(env.sessionsDir, "app-1"); + expect(metadata?.["lastMergeConflictDispatched"]).toBeFalsy(); + + // Conflicts recur — fresh tracker (attempt 1 again, not 2) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ hasConflicts: true }), + ); + vi.mocked(notifier.notify).mockClear(); + await lm.check("app-1"); + + // Fresh budget — sends to agent (attempt 1), not escalate + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("non-persistent reaction keys still clear on status exit", async () => { + config.reactions = { + "changes-requested": { + auto: true, + action: "send-to-agent", + message: "Address review comments.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ reviewDecision: "changes_requested" }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + const registry = createMockRegistry({ + runtime: plugins.runtime, + agent: plugins.agent, + scm: mockSCM, + }); + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Transition to changes_requested (attempt 1 — send to agent) + await lm.check("app-1"); + expect(lm.getStates().get("app-1")).toBe("changes_requested"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // Transition away — tracker clears (non-persistent key) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing", reviewDecision: "none" }), + ); + await lm.check("app-1"); + + // Transition back — fresh tracker (attempt 1 again, NOT 2) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ reviewDecision: "changes_requested" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + + // Transition away and back again — still attempt 1, not escalating + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing", reviewDecision: "none" }), + ); + await lm.check("app-1"); + vi.mocked(mockSessionManager.send).mockClear(); + + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ reviewDecision: "changes_requested" }), + ); + await lm.check("app-1"); + // With retries:1, attempt 2 would escalate. But tracker was cleared, + // so this is attempt 1 again — still sends to agent. + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + }); + + it("CI escalation silences further dispatches — clears only after stable CI pass", async () => { + // retries:1 → attempt 1 sends, attempt 2 escalates. + // After escalation: tracker.escalated=true silences subsequent oscillations. + // Tracker clears only after 2 consecutive passing polls; then next failure gets fresh budget. + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing. Fix it.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Oscillation 1: pr_open → ci_failed (attempt 1 — send to agent) + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI passes briefly (ci_failed → pr_open, stableCount = 1) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + + // Oscillation 2: pr_open → ci_failed (attempt 2 > retries:1 — escalate, tracker.escalated = true) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + vi.mocked(notifier.notify).mockClear(); + + // CI passes once (stableCount = 1 — not enough to clear yet) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); + + // Oscillation 3: pr_open → ci_failed — escalated tracker short-circuits, NO dispatch + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(notifier.notify).mockClear(); + + // CI passes twice stably (stableCount → 1 → 2 → tracker cleared) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "passing" }), + ); + await lm.check("app-1"); // stableCount = 1 + await lm.check("app-1"); // stableCount = 2 → clearReactionTracker + + // Oscillation 4: pr_open → ci_failed — fresh budget: attempt 1, sends (not escalate) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation( + mockBatchEnrichment({ ciStatus: "failing" }), + ); + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("single passing poll does not reset escalated ci-failed tracker", async () => { + // Regression: one passing poll must NOT clear the tracker. Requires 2 consecutive passing polls. + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ enrichSessionsPRBatch: batchMock }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Reach escalated state: attempt 1 → send, attempt 2 → escalate + await lm.check("app-1"); // pr_open → ci_failed: attempt 1, send + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); // ci_failed → pr_open + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); // pr_open → ci_failed: attempt 2 → escalate + expect(notifier.notify).toHaveBeenCalledWith(expect.objectContaining({ type: "reaction.escalated" })); + vi.mocked(notifier.notify).mockClear(); + + // ONE passing poll (stableCount = 1, not enough) + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); + + // Next CI failure: tracker still escalated → short-circuit + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); + expect(notifier.notify).not.toHaveBeenCalledWith(expect.objectContaining({ type: "reaction.escalated" })); + }); + + it("pending CI does not count toward ci-failed tracker resolution", async () => { + // Regression: real CI goes failing → pending (new run started) → failing. + // "pending" must NOT count as resolution — only "passing" does. + // Without this, 2 pending polls between failures wipe the tracker and we're back at #1409. + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing.", + retries: 2, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ enrichSessionsPRBatch: batchMock }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // CI failing: pr_open → ci_failed, attempt 1 — send + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI goes pending (agent pushed a fix, new run started): ci_failed → pr_open + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "pending" })); + await lm.check("app-1"); // stableCount must NOT increment + await lm.check("app-1"); // two pending polls — must NOT clear tracker + + // CI fails again (run completed failing): pr_open → ci_failed, attempt 2 — send + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + // If pending had wrongly cleared the tracker, this would be attempt 1 (fresh), not attempt 2. + // Attempt 2 ≤ retries:2 → sends to agent (not escalates) + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + vi.mocked(mockSessionManager.send).mockClear(); + + // CI goes pending again, then failing — attempt 3 > retries:2 → escalate + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "pending" })); + await lm.check("app-1"); // pending: no clear + await lm.check("app-1"); // pending: no clear + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + expect(mockSessionManager.send).not.toHaveBeenCalled(); // escalated, not sent to agent + }); + + it("only passing CI resets ci-failed tracker — pending mid-run does not interfere", async () => { + // Complementary to previous: failing → pending(many) → passing(2) → failing SHOULD clear. + // Pending during CI run doesn't block resolution; only the final passing state matters. + const notifier = createMockNotifier(); + + config.reactions = { + "ci-failed": { + auto: true, + action: "send-to-agent", + message: "CI is failing.", + retries: 1, + escalateAfter: 1, + }, + }; + + const batchMock = mockBatchEnrichment({ ciStatus: "failing" }); + const mockSCM = createMockSCM({ enrichSessionsPRBatch: batchMock }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + vi.mocked(mockSessionManager.send).mockResolvedValue(undefined); + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + // Reach escalated state: attempt 1 → send, attempt 2 → escalate + await lm.check("app-1"); // attempt 1, send + vi.mocked(mockSessionManager.send).mockClear(); + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); // ci_failed → pr_open + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); // attempt 2 → escalate + vi.mocked(notifier.notify).mockClear(); + + // CI goes pending (new run) — stableCount stays 0, does NOT progress toward resolution + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "pending" })); + await lm.check("app-1"); + await lm.check("app-1"); + await lm.check("app-1"); // many pending polls — stableCount never reaches threshold + + // CI finally passes (2 stable polls) → tracker cleared + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "passing" })); + await lm.check("app-1"); // stableCount = 1 + await lm.check("app-1"); // stableCount = 2 → clearReactionTracker + + // Next CI failure gets fresh budget: attempt 1, send + vi.mocked(mockSCM.enrichSessionsPRBatch!).mockImplementation(mockBatchEnrichment({ ciStatus: "failing" })); + await lm.check("app-1"); + expect(mockSessionManager.send).toHaveBeenCalledTimes(1); + expect(notifier.notify).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "reaction.escalated" }), + ); + }); + + it("merge-conflict notify action preserves warning priority", async () => { + const notifier = createMockNotifier(); + + config.reactions = { + "merge-conflicts": { + auto: true, + action: "notify", + }, + }; + config.notificationRouting = { + urgent: ["desktop"], + action: ["desktop"], + warning: ["desktop"], + info: [], + }; + + const batchMock = mockBatchEnrichment({ hasConflicts: true }); + const mockSCM = createMockSCM({ + enrichSessionsPRBatch: batchMock, + }); + + const registry: PluginRegistry = { + ...mockRegistry, + get: vi.fn().mockImplementation((slot: string, name: string) => { + if (slot === "runtime") return plugins.runtime; + if (slot === "agent") return plugins.agent; + if (slot === "scm") return mockSCM; + if (slot === "notifier" && name === "desktop") return notifier; + return null; + }), + }; + + const lm = setupCheck("app-1", { + session: makeSession({ status: "pr_open", pr: makePR() }), + registry, + }); + + await lm.check("app-1"); + + // With info routing empty and warning routing to desktop, + // notify should fire at "warning" priority (not "info") + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ + type: "reaction.triggered", + priority: "warning", + }), + ); + }); }); describe("pollAll terminal status accounting", () => { diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index 68b9e4178..344eb5052 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -92,6 +92,19 @@ function parseDuration(str: string): number { } } +/** Reaction keys for conditions that can oscillate (e.g. CI failing→pending→failing). + * Their trackers survive status exit so the escalation budget accumulates + * across oscillations instead of resetting to zero each time. + * Note: "merge-conflicts" is NOT here — statusToEventType never emits + * "merge.conflicts", so the transition handler at line ~1892 can't reach it. + * Merge-conflict tracker lifecycle is managed in maybeDispatchMergeConflicts. */ +const PERSISTENT_REACTION_KEYS = new Set(["ci-failed"]); + +/** Number of consecutive CI-passing polls required before the ci-failed tracker + * (including its escalated flag) is cleared, allowing a fresh budget for the + * next real CI failure incident. */ +const CI_PASSING_STABLE_THRESHOLD = 2; + type WorkspaceBranchProbe = | { kind: "branch"; branch: string } | { kind: "detached" } @@ -385,6 +398,9 @@ export interface LifecycleManagerDeps { interface ReactionTracker { attempts: number; firstTriggered: Date; + /** True after this reaction has escalated. Short-circuits further dispatches + * until the underlying condition resolves and the tracker is explicitly cleared. */ + escalated?: boolean; } /** Create a LifecycleManager instance. */ @@ -1169,6 +1185,11 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan reactionTrackers.set(trackerKey, tracker); } + // Already escalated — wait for the condition to resolve before resuming. + if (tracker.escalated) { + return { reactionType: reactionKey, success: true, action: "escalated", escalated: true }; + } + // Increment attempts before checking escalation tracker.attempts++; @@ -1201,6 +1222,11 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan data: { reactionKey, attempts: tracker.attempts }, }); await notifyHuman(event, reactionConfig.priority ?? "urgent"); + + // Mark as escalated — silences further dispatches until the underlying + // condition resolves and clearReactionTracker() is called explicitly. + tracker.escalated = true; + return { reactionType: reactionKey, success: true, @@ -1656,35 +1682,44 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan (reactionConfig.auto !== false || reactionConfig.action === "notify") ) { try { - if (reactionConfig.action === "send-to-agent") { + // Build enriched config with dynamic base branch message. + // Preserve "warning" priority from old direct-dispatch code unless + // the user explicitly set a different priority in their config. + const enrichedConfig = { + ...reactionConfig, + priority: reactionConfig.priority ?? ("warning" as const), + }; + if (reactionConfig.action === "send-to-agent" && !reactionConfig.message) { const baseBranch = session.pr.baseBranch ?? "the default branch"; const behindNote = cachedData.isBehind ? ` is behind ${baseBranch} and` : ""; - const message = - reactionConfig.message ?? - `Your PR branch${behindNote} has merge conflicts with ${baseBranch}. Rebase your branch on ${baseBranch}, resolve the conflicts, and push. You should not need to call gh for merge status unless you need additional context — this information is current.`; - await sessionManager.send(session.id, message); - } else { - const event = createEvent("merge.conflicts", { - sessionId: session.id, - projectId: session.projectId, - message: `${session.id}: PR has merge conflicts`, - }); - await notifyHuman(event, reactionConfig.priority ?? "warning"); + enrichedConfig.message = `Your PR branch${behindNote} has merge conflicts with ${baseBranch}. Rebase your branch on ${baseBranch}, resolve the conflicts, and push. You should not need to call gh for merge status unless you need additional context — this information is current.`; } - updateSessionMetadata(session, { - lastMergeConflictDispatched: "true", - }); + const result = await executeReaction( + session.id, + session.projectId, + conflictReactionKey, + enrichedConfig, + ); + // Only set dedup flag for non-escalated success — escalation hands off + // to the human, so we must NOT suppress future agent dispatches if the + // condition recurs after the tracker resets. + if (result.success && result.action !== "escalated") { + updateSessionMetadata(session, { + lastMergeConflictDispatched: "true", + }); + } } catch { - // Send failed — will retry on next poll cycle + // Dispatch failed — will retry on next poll cycle } } } else if (lastDispatched === "true") { - // Conflicts resolved — clear so we can re-dispatch if they recur - clearReactionTracker(session.id, conflictReactionKey); + // Conflicts resolved — clear dedup flag and reaction tracker so future + // conflicts start a fresh incident with a fresh escalation budget. updateSessionMetadata(session, { lastMergeConflictDispatched: "", }); + clearReactionTracker(session.id, conflictReactionKey); } } @@ -1849,6 +1884,28 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan updateSessionMetadata(session, metadataUpdates); } + // CI resolution tracking — reset the ci-failed tracker (including its escalated + // flag) once CI has been passing for CI_PASSING_STABLE_THRESHOLD consecutive polls. + // This lets the next real CI failure start with a fresh budget. + if (session.pr) { + const prKey = `${session.pr.owner}/${session.pr.repo}#${session.pr.number}`; + const cachedData = prEnrichmentCache.get(prKey); + if (cachedData) { + if (cachedData.ciStatus === "passing") { + const stableCount = Number(session.metadata["ciPassingStableCount"] ?? "0") + 1; + if (stableCount >= CI_PASSING_STABLE_THRESHOLD) { + clearReactionTracker(session.id, "ci-failed"); + updateSessionMetadata(session, { ciPassingStableCount: "" }); + } else { + updateSessionMetadata(session, { ciPassingStableCount: String(stableCount) }); + } + } else if (session.metadata["ciPassingStableCount"]) { + // pending or failing resets the stability window — only "passing" counts as resolution + updateSessionMetadata(session, { ciPassingStableCount: "" }); + } + } + } + if (newStatus !== oldStatus) { const correlationId = createCorrelationId("lifecycle-transition"); // State transition detected @@ -1879,11 +1936,14 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan allCompleteEmitted = false; } - // Clear reaction trackers for the old status so retries reset on state changes + // Clear reaction trackers for the old status so retries reset on state changes. + // Persistent keys (ci-failed) are excluded — their trackers survive oscillation + // so the escalation budget accumulates across cycles. On escalation, the tracker + // is cleared in executeReaction so future incidents get a fresh budget. const oldEventType = statusToEventType(undefined, oldStatus); if (oldEventType) { const oldReactionKey = eventToReactionKey(oldEventType); - if (oldReactionKey) { + if (oldReactionKey && !PERSISTENT_REACTION_KEYS.has(oldReactionKey)) { clearReactionTracker(session.id, oldReactionKey); } } From 68756105fbd3517b4683dd4fc3649e6a4147efe2 Mon Sep 17 00:00:00 2001 From: fastestdevalive Date: Wed, 29 Apr 2026 13:05:04 -0700 Subject: [PATCH 2/3] refactor(web): replace SSE with WebSocket polling for session updates (#1259) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(web): remove SSE entirely — browser uses WebSocket only - Delete GET /api/events route (no consumers remain) - Refactor SessionBroadcaster: replaces SSE stream fetch with a plain setInterval polling GET /api/sessions/patches every 3s, eliminating the last server-side SSE consumer - Remove EventSource from useSessionEvents; hook is now WebSocket-only via mux.sessions; rename SSEAttentionMap → AttentionMap and sseAttentionLevels → attentionLevels throughout - Replace useSSESessionActivity with useMuxSessionActivity — thin selector over useMux().sessions, no network call - Delete SSESnapshotEvent and SSEActivityEvent types from lib/types.ts - Delete Dashboard.renderCadence.test.tsx (SSE-specific test) - Update ARCHITECTURE.md to reflect the simplified two-protocol design (HTTP + WebSocket only; no SSE anywhere in the system) Co-Authored-By: Claude Sonnet 4.6 * refactor(web): address PR review comments - useMuxSessionActivity: switch to useMuxOptional (consistent with page.tsx), add useMemo for referential stability - useSessionEvents: validate patch.status against VALID_SESSION_STATUSES before casting; type all three VALID_* sets with satisfies for exhaustiveness - mux-websocket: remove leading underscores from private fields (intervalId, polling) — private modifier already conveys intent - Dashboard.renderCadence test: port from SSE/EventSource to MuxProvider mock; covers same-membership-snapshot-only-rerenders-changed-card invariant - Remove .feature-plans/pending/remove-browser-sse.md (duplicated in PR body) Co-Authored-By: Claude Sonnet 4.6 * fix(web): guard broadcast against stale fetch after disconnect If disconnect() runs while fetchSnapshot() is in flight, the .then callback would still fire broadcast() into an empty (or re-populated) subscriber set. Guard with intervalId !== null so stale resolutions after the last subscriber leaves are silently dropped. Co-Authored-By: Claude Sonnet 4.6 * fix(web): remove SSE-specific tests from emptyState suite The upstream added two tests for live load-error banners driven by SSE onmessage events. Since this PR removes SSE entirely, those tests can't pass and the SSE mock setup is no longer needed. Co-Authored-By: Claude Sonnet 4.6 * fix(web): restore liveSessionsResolved to prevent premature banner dismiss mux?.status === "connected" fires on WebSocket handshake before any session data arrives. In the SSR-failure scenario (dashboardLoadError set), this was dismissing the error banner as soon as the WS opened, leaving users with a silent empty dashboard. Restore liveSessionsResolved: set it only from the first successful HTTP /api/sessions refresh or mux snapshot (same semantics as main). The reset action from the initialSessions effect intentionally does not set it (liveResolved flag absent = SSR-only reset, not live data). Co-Authored-By: Claude Sonnet 4.6 * fix(web): port live load-error banner from SSE to WS transport SessionBroadcaster now emits { ch: "sessions", type: "error" } on fetch failure instead of silently returning null. MuxProvider surfaces the error as lastError on the context. useSessionEvents restores loadError reducer state, synced from muxLastError, cleared on successful snapshot or HTTP refresh. Dashboard renders the live error banner via loadError ?? ssrLoadError. Two emptyState tests ported to drive errors through MuxProvider mock. Co-Authored-By: Claude Sonnet 4.6 --------- Co-authored-by: Claude Sonnet 4.6 Co-authored-by: Gaurav Bhola --- docs/ARCHITECTURE.md | 204 ++++++++++++++ .../server/__tests__/mux-websocket.test.ts | 168 +++++------ packages/web/server/mux-websocket.ts | 179 ++++++------ packages/web/src/__tests__/api-routes.test.ts | 30 -- packages/web/src/app/api/events/route.ts | 266 ------------------ .../web/src/app/sessions/[id]/page.test.tsx | 4 + packages/web/src/app/sessions/[id]/page.tsx | 6 +- packages/web/src/components/Dashboard.tsx | 43 +-- .../web/src/components/DynamicFavicon.tsx | 24 +- .../web/src/components/PullRequestsPage.tsx | 12 +- .../__tests__/Dashboard.emptyState.test.tsx | 100 ++++--- .../Dashboard.renderCadence.test.tsx | 125 +++++--- .../__tests__/DynamicFavicon.test.tsx | 40 +-- .../__tests__/useSSESessionActivity.test.ts | 207 -------------- .../web/src/hooks/useMuxSessionActivity.ts | 16 ++ .../web/src/hooks/useSSESessionActivity.ts | 54 ---- packages/web/src/hooks/useSessionEvents.ts | Bin 16628 -> 12994 bytes packages/web/src/lib/mux-protocol.ts | 1 + packages/web/src/lib/types.ts | 24 -- packages/web/src/providers/MuxProvider.tsx | 15 +- 20 files changed, 590 insertions(+), 928 deletions(-) create mode 100644 docs/ARCHITECTURE.md delete mode 100644 packages/web/src/app/api/events/route.ts delete mode 100644 packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts create mode 100644 packages/web/src/hooks/useMuxSessionActivity.ts delete mode 100644 packages/web/src/hooks/useSSESessionActivity.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md new file mode 100644 index 000000000..04eb6fa6a --- /dev/null +++ b/docs/ARCHITECTURE.md @@ -0,0 +1,204 @@ +# Agent Orchestrator — Technical Architecture + +This document explains how the various parts of the Agent Orchestrator communicate with each other: where HTTP is used, where WebSocket is used, and what each carries. + +--- + +## System Overview + +```mermaid +graph TB + subgraph Browser["Browser / Dashboard"] + UI["React Dashboard\n(Next.js App Router)"] + XTerm["xterm.js\nTerminal UI"] + end + + subgraph NextJS["Next.js Server — :3000 (single process)"] + subgraph HTTPAPI["① HTTP REST /api/* — request / response"] + Sessions["GET /api/sessions\nGET /api/sessions/:id\nPOST /api/sessions/:id/message\nPOST /api/sessions/:id/restore\nPOST /api/sessions/:id/kill\nPOST /api/spawn\nGET /api/projects GET /api/agents\nGET /api/issues POST /api/prs/:id/merge\nPOST /api/webhooks/**"] + Patches["GET /api/sessions/patches\n(lightweight: id, status, activity,\nattentionLevel, lastActivityAt)"] + end + SessionMgr["Session Manager\n(reads flat files in\n~/.agent-orchestrator/)"] + end + + subgraph MuxServer["② WebSocket Server — :14801 (separate Node process)"] + MuxWS["ws://host:14801/mux\nMultiplexed — two sub-channels\nover one connection"] + TermMgr["TerminalManager\n(node-pty → tmux PTY)"] + Broadcaster["SessionBroadcaster\n(setInterval every 3s →\nGET /api/sessions/patches)"] + end + + subgraph Agents["AI Agents (one tmux window each)"] + ClaudeCode["Claude Code"] + Codex["Codex"] + Aider["Aider"] + OpenCode["OpenCode"] + end + + subgraph External["External Services"] + GitHub["GitHub API"] + Linear["Linear API"] + end + + %% ① HTTP — user actions & data fetching + UI -- "① HTTP GET/POST\n(on demand: load sessions,\nsend message, spawn, merge PR…)" --> Sessions + + %% ② WebSocket terminal sub-channel + XTerm -- "② WS sub-channel 'terminal'\nkeystrokes → {ch:terminal, type:data}\noutput ← {ch:terminal, type:data}" --> MuxWS + MuxWS --> TermMgr + TermMgr -- "PTY read/write" --> ClaudeCode + TermMgr -- "PTY read/write" --> Codex + TermMgr -- "PTY read/write" --> Aider + TermMgr -- "PTY read/write" --> OpenCode + + %% ② WebSocket sessions sub-channel + Broadcaster -- "HTTP GET /api/sessions/patches\nevery 3s" --> Patches + Patches -- "reads" --> SessionMgr + Broadcaster -- "② WS sub-channel 'sessions'\n{ch:sessions, type:snapshot,\n sessions:[{id,status,activity,\n attentionLevel,lastActivityAt}]}" --> MuxWS + MuxWS -- "session patches\n→ useSessionEvents()\n→ useMuxSessionActivity()" --> UI + + %% Mux auto-recovery calls back to Next.js + TermMgr -- "① HTTP POST /api/sessions/:id/restore\n(auto-recovery when tmux dies)" --> Sessions + + %% External + Sessions -- "REST calls" --> GitHub + Sessions --> Linear + GitHub -- "POST /api/webhooks/**" --> Sessions +``` + +--- + +## Communication Channels + +### 1. HTTP / REST — `/api/*` on port 3000 + +Used for all request-response interactions. The browser calls these on demand; the CLI and the WebSocket server also use them. + +| Endpoint | Method | Purpose | +|----------|--------|---------| +| `/api/sessions` | GET | List all sessions (with PR / issue metadata) | +| `/api/sessions/light` | GET | Lightweight session list (minimal fields) | +| `/api/sessions/patches` | GET | Ultra-light patches (id, status, activity, attentionLevel) — polled by the WS server every 3s | +| `/api/sessions/:id` | GET | Full session detail | +| `/api/sessions/:id/message` | POST | Send a message/command to a live agent | +| `/api/sessions/:id/restore` | POST | Respawn a terminated session | +| `/api/sessions/:id/kill` | POST | Terminate a running session | +| `/api/sessions/:id/files` | GET | Browse workspace files | +| `/api/sessions/:id/diff/**` | GET | File diff view | +| `/api/sessions/:id/sub-sessions` | GET / POST | List / create sub-sessions (forked agents) | +| `/api/spawn` | POST | Spawn a new agent session | +| `/api/projects` | GET | List configured projects | +| `/api/agents` | GET | List registered agent plugins | +| `/api/issues` | GET | Fetch backlog issues | +| `/api/backlog` | GET | Backlog summary | +| `/api/prs/:id/merge` | POST | Merge a PR | +| `/api/observability` | GET | Health and metrics summary | +| `/api/verify` | POST | Verify environment setup | +| `/api/setup-labels` | POST | Set up GitHub labels | +| `/api/webhooks/**` | POST | Inbound webhooks from GitHub / GitLab | + +--- + +### 2. WebSocket (Multiplexed) — `ws://localhost:14801/mux` + +A **bidirectional multiplexed channel** on a separate Node.js process. A single WebSocket connection carries two independent sub-channels: + +- **`terminal` channel** — raw PTY I/O for xterm.js +- **`sessions` channel** — real-time session status patches (fed by `SessionBroadcaster` polling `/api/sessions/patches` every 3s) + +```mermaid +sequenceDiagram + participant XTerm as xterm.js + participant MuxClient as MuxProvider (browser) + participant MuxWS as WS Server :14801/mux + participant PTY as node-pty (tmux) + participant Next as Next.js :3000 + + MuxClient->>MuxWS: connect ws://localhost:14801/mux + + Note over MuxClient,MuxWS: Open a terminal + MuxClient->>MuxWS: {ch:"terminal", id:"sess-1", type:"open"} + MuxWS->>PTY: attach tmux PTY + MuxWS-->>MuxClient: {ch:"terminal", id:"sess-1", type:"opened"} + + Note over MuxClient,MuxWS: Terminal I/O + XTerm->>MuxClient: user keystrokes + MuxClient->>MuxWS: {ch:"terminal", id:"sess-1", type:"data", data:"ls\r"} + MuxWS->>PTY: write to PTY + PTY-->>MuxWS: output bytes + MuxWS-->>MuxClient: {ch:"terminal", id:"sess-1", type:"data", data:"file1 file2\r\n"} + MuxClient-->>XTerm: render output + + Note over MuxWS,Next: Session patches (every 3s) + MuxWS->>Next: GET /api/sessions/patches + Next-->>MuxWS: [{id, status, activity, attentionLevel, lastActivityAt}] + MuxWS-->>MuxClient: {ch:"sessions", type:"snapshot", sessions:[...]} + MuxClient-->>MuxClient: useSessionEvents() + useMuxSessionActivity() update React state + + Note over MuxWS,Next: Auto-recovery (session dead) + MuxWS->>Next: POST /api/sessions/sess-1/restore + Next-->>MuxWS: 200 OK + MuxWS->>PTY: reattach to new tmux session +``` + +**Message types:** + +| Direction | Channel | Type | Payload | +|-----------|---------|------|---------| +| Client→Server | `terminal` | `open` | `{ id }` | +| Client→Server | `terminal` | `data` | `{ id, data: string }` | +| Client→Server | `terminal` | `resize` | `{ id, cols, rows }` | +| Client→Server | `terminal` | `close` | `{ id }` | +| Client→Server | `subscribe` | — | `{ topics: ["sessions"] }` | +| Client→Server | `system` | `ping` | — | +| Server→Client | `terminal` | `opened` | `{ id }` | +| Server→Client | `terminal` | `data` | `{ id, data: string }` | +| Server→Client | `terminal` | `exited` | `{ id, code }` | +| Server→Client | `terminal` | `error` | `{ id, message }` | +| Server→Client | `sessions` | `snapshot` | `{ sessions: SessionPatch[] }` | +| Server→Client | `system` | `pong` | — | + +--- + +## Process Map + +```mermaid +graph LR + subgraph Host + CLI["ao CLI\n(packages/cli)"] + Next["Next.js\npackages/web — :3000"] + MuxSrv["Terminal WS Server\npackages/web/server — :14801"] + end + + subgraph Storage["Flat-file Storage"] + Sessions2["~/.agent-orchestrator/\n{hash}-{project}/\n sessions/{id} ← key-value\n worktrees/{id}/\n archive/{id}_{ts}/"] + end + + CLI -- "pnpm ao start\nspawns both servers" --> Next + CLI -- "spawns" --> MuxSrv + Next -- "reads / writes" --> Sessions2 + MuxSrv -- "GET /api/sessions/patches (every 3s)" --> Next + MuxSrv -- "POST /api/sessions/:id/restore (recovery)" --> Next +``` + +The CLI (`ao start`) forks two long-running processes: +- **Next.js** on `:3000` — serves the dashboard and all REST routes +- **Terminal WS server** on `:14801` — handles multiplexed WebSocket + PTY management + session patch polling + +Both processes share no in-memory state; coordination happens through flat files in `~/.agent-orchestrator/` and HTTP calls from the WS server to Next.js. + +--- + +## Data Flow Summary + +| Scenario | Protocol | Path | +|----------|----------|------| +| Load dashboard | HTTP GET | Browser → `:3000/` (SSR page) | +| List sessions | HTTP GET | Browser → `:3000/api/sessions` | +| Spawn new agent | HTTP POST | Browser → `:3000/api/spawn` | +| Send message to agent | HTTP POST | Browser → `:3000/api/sessions/:id/message` | +| Real-time session status | WebSocket | Browser ← `:14801/mux` `sessions` sub-channel (pushed every 3s) | +| Terminal output / input | WebSocket | Browser ↔ `:14801/mux` `terminal` sub-channel (bidirectional) | +| WS server fetches patches | HTTP GET | `:14801` → `:3000/api/sessions/patches` (every 3s) | +| WS server restores session | HTTP POST | `:14801` → `:3000/api/sessions/:id/restore` | +| GitHub notifies of CI / PR | HTTP POST | GitHub → `:3000/api/webhooks/github` | +| CLI queries sessions | HTTP GET | `ao` CLI → `:3000/api/sessions` | diff --git a/packages/web/server/__tests__/mux-websocket.test.ts b/packages/web/server/__tests__/mux-websocket.test.ts index 2fe0e83ef..d882df667 100644 --- a/packages/web/server/__tests__/mux-websocket.test.ts +++ b/packages/web/server/__tests__/mux-websocket.test.ts @@ -15,6 +15,7 @@ describe("SessionBroadcaster", () => { }); afterEach(() => { + vi.clearAllTimers(); vi.useRealTimers(); }); @@ -29,13 +30,10 @@ describe("SessionBroadcaster", () => { describe("subscribe", () => { it("sends an immediate snapshot to a new subscriber", async () => { const patches = [makePatch("s1")]; - // Mock the snapshot fetch mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: patches }), }); - // Mock the SSE connect (hangs forever) - mockFetch.mockReturnValueOnce(new Promise(() => {})); const callback = vi.fn(); broadcaster.subscribe(callback); @@ -50,94 +48,88 @@ describe("SessionBroadcaster", () => { expect(callback).toHaveBeenCalledWith(patches); }); - it("starts SSE connection on first subscriber", async () => { + it("starts polling interval on first subscriber", async () => { mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }), }); - // SSE connect - mockFetch.mockReturnValueOnce(new Promise(() => {})); broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); + // Snapshot fetch is called once on subscribe + expect(mockFetch).toHaveBeenCalledTimes(1); + + // After 3 seconds, polling interval should trigger a second fetch + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); + await vi.advanceTimersByTimeAsync(3000); + expect(mockFetch).toHaveBeenCalledTimes(2); - expect(mockFetch).toHaveBeenCalledWith( - "http://localhost:3000/api/events", - expect.objectContaining({ - headers: { Accept: "text/event-stream" }, - }), - ); }); - it("does not start a second SSE connection for additional subscribers", async () => { + it("does not start a second polling interval for additional subscribers", async () => { mockFetch.mockResolvedValue({ ok: true, json: async () => ({ sessions: [] }), }); - // SSE connect (first subscriber triggers it) - mockFetch.mockReturnValueOnce(new Promise(() => {})); broadcaster.subscribe(vi.fn()); broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); - // 1 snapshot for sub1 + 1 SSE connect + 1 snapshot for sub2 = 3 - // (SSE connect is only called once) - const sseConnects = mockFetch.mock.calls.filter( - (call) => call[0] === "http://localhost:3000/api/events", - ); - expect(sseConnects).toHaveLength(1); + // 1 snapshot for sub1 + 1 snapshot for sub2 = 2 + expect(mockFetch).toHaveBeenCalledTimes(2); + + // After 3 seconds, only one polling fetch happens + await vi.advanceTimersByTimeAsync(3000); + expect(mockFetch).toHaveBeenCalledTimes(3); }); - it("returns an unsubscribe function that disconnects when last subscriber leaves", async () => { + it("returns an unsubscribe function that stops polling when last subscriber leaves", async () => { mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }), }); - mockFetch.mockReturnValueOnce(new Promise(() => {})); const unsub = broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); + // Unsubscribe triggers disconnect unsub(); - // After unsubscribe, the abort should be triggered (disconnect) - // Further subscribe should trigger a new connection + + // Reset and advance past polling interval + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); + await vi.advanceTimersByTimeAsync(3000); + + // Should not have called fetch again after unsubscribe + expect(mockFetch).toHaveBeenCalledTimes(1); }); }); describe("broadcast", () => { - it("delivers patches to all subscribers", async () => { + it("delivers patches to all subscribers on each poll", async () => { const patches = [makePatch("s1"), makePatch("s2")]; - // Create a readable stream that sends one SSE event - const encoder = new TextEncoder(); - const sseData = `data: ${JSON.stringify({ type: "snapshot", sessions: patches })}\n\n`; - let readerDone = false; - const mockReader = { - read: vi.fn().mockImplementation(async () => { - if (!readerDone) { - readerDone = true; - return { done: false, value: encoder.encode(sseData) }; - } - return { done: true, value: undefined }; - }), - }; - - // Snapshot fetch + // Initial snapshot for first subscriber mockFetch.mockResolvedValueOnce({ ok: true, - json: async () => ({ sessions: [] }), + json: async () => ({ sessions: patches }), }); - // SSE connect returns a stream + // Initial snapshot for second subscriber mockFetch.mockResolvedValueOnce({ ok: true, - body: { getReader: () => mockReader }, + json: async () => ({ sessions: patches }), }); - // Second subscriber snapshot + // Polling fetch after 3s mockFetch.mockResolvedValueOnce({ ok: true, - json: async () => ({ sessions: [] }), + json: async () => ({ sessions: patches }), }); const cb1 = vi.fn(); @@ -147,30 +139,29 @@ describe("SessionBroadcaster", () => { await vi.advanceTimersByTimeAsync(10); - // Both callbacks should have received the broadcast + // Both callbacks should have received initial snapshot expect(cb1).toHaveBeenCalledWith(patches); expect(cb2).toHaveBeenCalledWith(patches); + + // Advance past poll interval (3s) and add buffer for promise resolution + await vi.advanceTimersByTimeAsync(3010); + + // Should be called again from polling + expect(cb1).toHaveBeenCalledTimes(2); + expect(cb2).toHaveBeenCalledTimes(2); }); it("isolates subscriber errors — one throw does not skip others", async () => { const patches = [makePatch("s1")]; - const encoder = new TextEncoder(); - const sseData = `data: ${JSON.stringify({ type: "snapshot", sessions: patches })}\n\n`; - let readerDone = false; - const mockReader = { - read: vi.fn().mockImplementation(async () => { - if (!readerDone) { - readerDone = true; - return { done: false, value: encoder.encode(sseData) }; - } - return { done: true, value: undefined }; - }), - }; - // Return null for snapshots so the initial callback doesn't fire (and throw) - mockFetch.mockResolvedValueOnce({ ok: false, status: 500 }); - mockFetch.mockResolvedValueOnce({ ok: true, body: { getReader: () => mockReader } }); - mockFetch.mockResolvedValueOnce({ ok: false, status: 500 }); + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: patches }), + }); + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: patches }), + }); const throwingCb = vi.fn().mockImplementation(() => { throw new Error("ws.send failed"); @@ -181,6 +172,7 @@ describe("SessionBroadcaster", () => { await vi.advanceTimersByTimeAsync(10); + // goodCb should have received patches despite throwingCb error expect(goodCb).toHaveBeenCalledWith(patches); }); }); @@ -188,7 +180,6 @@ describe("SessionBroadcaster", () => { describe("fetchSnapshot", () => { it("returns null on fetch failure", async () => { mockFetch.mockRejectedValueOnce(new Error("network error")); - mockFetch.mockReturnValueOnce(new Promise(() => {})); const callback = vi.fn(); broadcaster.subscribe(callback); @@ -200,7 +191,6 @@ describe("SessionBroadcaster", () => { it("returns null on non-OK response", async () => { mockFetch.mockResolvedValueOnce({ ok: false, status: 500 }); - mockFetch.mockReturnValueOnce(new Promise(() => {})); const callback = vi.fn(); broadcaster.subscribe(callback); @@ -210,48 +200,28 @@ describe("SessionBroadcaster", () => { }); }); - describe("SSE reconnection", () => { - it("reconnects after SSE connection drops if subscribers remain", async () => { - mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }) }); - // SSE connect fails - mockFetch.mockResolvedValueOnce({ ok: false, status: 503 }); - // Reconnect SSE after backoff - mockFetch.mockReturnValueOnce(new Promise(() => {})); - - broadcaster.subscribe(vi.fn()); - await vi.advanceTimersByTimeAsync(0); - - // Advance past the 5s reconnect backoff - await vi.advanceTimersByTimeAsync(6000); - - const sseConnects = mockFetch.mock.calls.filter( - (call) => call[0] === "http://localhost:3000/api/events", - ); - expect(sseConnects.length).toBeGreaterThanOrEqual(2); - }); - }); - describe("disconnect", () => { - it("clears reconnect timer on disconnect", async () => { - mockFetch.mockResolvedValueOnce({ ok: true, json: async () => ({ sessions: [] }) }); - // SSE connect fails to trigger reconnect path - mockFetch.mockResolvedValueOnce({ ok: false, status: 503 }); + it("stops polling when last subscriber unsubscribes", async () => { + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); const unsub = broadcaster.subscribe(vi.fn()); await vi.advanceTimersByTimeAsync(0); - // Unsubscribe triggers disconnect, which should clear reconnect timer + // Unsubscribe triggers disconnect unsub(); - // Advance past reconnect backoff — should NOT trigger a new connect - mockFetch.mockReturnValueOnce(new Promise(() => {})); - await vi.advanceTimersByTimeAsync(6000); + // Advance past polling interval + mockFetch.mockResolvedValueOnce({ + ok: true, + json: async () => ({ sessions: [] }), + }); + await vi.advanceTimersByTimeAsync(3000); - const sseConnects = mockFetch.mock.calls.filter( - (call) => call[0] === "http://localhost:3000/api/events", - ); - // Only 1 connect attempt, no reconnect after disconnect - expect(sseConnects).toHaveLength(1); + // Should only have 1 fetch (initial snapshot) + expect(mockFetch).toHaveBeenCalledTimes(1); }); }); }); diff --git a/packages/web/server/mux-websocket.ts b/packages/web/server/mux-websocket.ts index 22bf2b374..dfec567b3 100644 --- a/packages/web/server/mux-websocket.ts +++ b/packages/web/server/mux-websocket.ts @@ -2,9 +2,8 @@ * Multiplexed WebSocket server for terminal multiplexing. * Manages multiple terminal connections over a single persistent WebSocket. * - * Session updates are delivered via a single shared SSE connection from this - * process to Next.js /api/events, then broadcast to all subscribed clients. - * This replaces per-client HTTP polling and makes session updates event-driven. + * Session updates are delivered via polling of Next.js /api/sessions/patches + * every 3s, then broadcast to all subscribed clients via WebSocket. */ import { WebSocketServer, WebSocket } from "ws"; @@ -32,6 +31,7 @@ type ServerMessage = | { ch: "terminal"; id: string; type: "opened" } | { ch: "terminal"; id: string; type: "error"; message: string } | { ch: "sessions"; type: "snapshot"; sessions: SessionPatch[] } + | { ch: "sessions"; type: "error"; error: string } | { ch: "system"; type: "pong" } | { ch: "system"; type: "error"; message: string }; @@ -54,14 +54,15 @@ interface SessionPatch { } /** - * Manages a single shared SSE connection to Next.js /api/events. - * Broadcasts session patches to all subscribed callbacks. - * Lazily connects on first subscriber, disconnects when the last one leaves. + * Manages polling of session patches from Next.js /api/sessions/patches. + * Broadcasts to all subscribed callbacks. + * Lazily starts polling on first subscriber, stops when the last one leaves. */ export class SessionBroadcaster { private subscribers = new Set<(sessions: SessionPatch[]) => void>(); - private abortController: AbortController | null = null; - private reconnectTimer: ReturnType | null = null; + private errorSubscribers = new Set<(error: string) => void>(); + private intervalId: ReturnType | null = null; + private polling = false; private readonly baseUrl: string; constructor(nextPort: string) { @@ -69,27 +70,50 @@ export class SessionBroadcaster { } /** - * Subscribe to session patches. Returns an unsubscribe function. - * Sends an immediate snapshot to the new subscriber, then live SSE pushes. + * Subscribe to session patches and errors. Returns an unsubscribe function. + * Sends an immediate snapshot to the new subscriber, then polling updates. */ - subscribe(callback: (sessions: SessionPatch[]) => void): () => void { + subscribe(callback: (sessions: SessionPatch[]) => void, onError?: (error: string) => void): () => void { const wasEmpty = this.subscribers.size === 0; this.subscribers.add(callback); + if (onError) this.errorSubscribers.add(onError); // Immediately send a one-off snapshot to just this new subscriber - void this.fetchSnapshot().then((sessions) => { - if (sessions && this.subscribers.has(callback)) { - callback(sessions); + void this.fetchSnapshot().then((result) => { + if (result.sessions && this.subscribers.has(callback)) { + try { + callback(result.sessions); + } catch { + // Isolate subscriber errors so one bad subscriber doesn't break others + } + } else if (result.error && onError && this.errorSubscribers.has(onError)) { + try { + onError(result.error); + } catch { + // Isolate subscriber errors + } } }); - // Start the shared SSE connection if this is the first subscriber + // Start polling if this is the first subscriber if (wasEmpty) { - void this.connect(); + this.intervalId = setInterval(() => { + if (this.polling) return; + this.polling = true; + void this.fetchSnapshot() + .then((result) => { + if (result.sessions && this.intervalId !== null) this.broadcast(result.sessions); + else if (result.error && this.intervalId !== null) this.broadcastError(result.error); + }) + .finally(() => { + this.polling = false; + }); + }, 3000); } return () => { this.subscribers.delete(callback); + if (onError) this.errorSubscribers.delete(onError); if (this.subscribers.size === 0) { this.disconnect(); } @@ -106,101 +130,45 @@ export class SessionBroadcaster { } } - /** One-shot HTTP fetch of the current session list for immediate delivery. */ - private async fetchSnapshot(): Promise { - try { - const controller = new AbortController(); - const timeoutId = setTimeout(() => controller.abort(), 4000); + private broadcastError(error: string): void { + for (const callback of this.errorSubscribers) { try { - const res = await fetch(`${this.baseUrl}/api/sessions/patches`, { - signal: controller.signal, - }); - clearTimeout(timeoutId); - if (!res.ok) return null; - const data = (await res.json()) as { sessions?: SessionPatch[] }; - return data.sessions ?? null; - } catch { - clearTimeout(timeoutId); - return null; + callback(error); + } catch (err) { + console.error("[MuxServer] Session error subscriber threw:", err); } - } catch { - return null; } } - /** Open a persistent SSE connection and stream events to all subscribers. */ - private async connect(): Promise { - if (this.abortController) return; - + /** One-shot HTTP fetch of the current session list. */ + private async fetchSnapshot(): Promise<{ sessions: SessionPatch[] | null; error: string | null }> { const controller = new AbortController(); - this.abortController = controller; - const { signal } = controller; - + const timeoutId = setTimeout(() => controller.abort(), 4000); try { - const res = await fetch(`${this.baseUrl}/api/events`, { - signal, - headers: { Accept: "text/event-stream" }, + const res = await fetch(`${this.baseUrl}/api/sessions/patches`, { + signal: controller.signal, }); - - if (!res.ok || !res.body) { - throw new Error(`SSE connect failed: ${res.status}`); - } - - const reader = res.body.getReader(); - const decoder = new TextDecoder(); - let buffer = ""; - - while (!signal.aborted) { - const { done, value } = await reader.read(); - if (done) break; - - buffer += decoder.decode(value, { stream: true }); - const lines = buffer.split("\n"); - buffer = lines.pop() ?? ""; - - for (const line of lines) { - if (!line.startsWith("data: ")) continue; - try { - const event = JSON.parse(line.slice(6)) as { - type: string; - sessions?: SessionPatch[]; - }; - if (event.type === "snapshot" && event.sessions) { - this.broadcast(event.sessions); - } - } catch { - // ignore malformed events - } - } + clearTimeout(timeoutId); + if (!res.ok) { + const msg = `Session fetch failed: HTTP ${res.status}`; + console.warn(`[SessionBroadcaster] ${msg}`); + return { sessions: null, error: msg }; } + const data = (await res.json()) as { sessions?: SessionPatch[] }; + return { sessions: data.sessions ?? null, error: null }; } catch (err) { - if (signal.aborted) return; // intentional disconnect, not an error - console.warn("[MuxServer] SSE connection lost:", err instanceof Error ? err.message : err); - } finally { - // Only clear our own controller — a concurrent connect() may have already - // set a new one (e.g. disconnect() → subscribe() → connect() in the same tick). - if (this.abortController === controller) { - this.abortController = null; - } - } - - // Reconnect with backoff if there are still subscribers - if (this.subscribers.size > 0) { - console.log("[MuxServer] SSE reconnecting in 5s"); - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = null; - if (this.subscribers.size > 0) void this.connect(); - }, 5000); + clearTimeout(timeoutId); + const msg = err instanceof Error ? err.message : String(err); + console.warn("[SessionBroadcaster] fetchSnapshot error:", msg); + return { sessions: null, error: msg }; } } private disconnect(): void { - if (this.reconnectTimer) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = null; + if (this.intervalId !== null) { + clearInterval(this.intervalId); + this.intervalId = null; } - this.abortController?.abort(); - this.abortController = null; } } @@ -228,6 +196,7 @@ interface ManagedTerminal { } const RING_BUFFER_MAX = 50 * 1024; // 50KB max per terminal +const WS_BUFFER_HIGH_WATERMARK = 64 * 1024; // 64KB const MAX_REATTACH_ATTEMPTS = 3; /** @@ -570,12 +539,22 @@ export function createMuxWebSocket(tmuxPath?: string): WebSocketServer | null { } } else if (msg.ch === "subscribe") { if (msg.topics.includes("sessions") && !sessionUnsubscribe) { - sessionUnsubscribe = broadcaster.subscribe((sessions) => { - if (ws.readyState === WebSocket.OPEN) { + sessionUnsubscribe = broadcaster.subscribe( + (sessions) => { + if (ws.readyState !== WebSocket.OPEN) return; + if (ws.bufferedAmount > WS_BUFFER_HIGH_WATERMARK) { + console.warn("[MuxServer] Skipping session snapshot — socket backpressured"); + return; + } const snapMsg: ServerMessage = { ch: "sessions", type: "snapshot", sessions }; ws.send(JSON.stringify(snapMsg)); - } - }); + }, + (error) => { + if (ws.readyState !== WebSocket.OPEN) return; + const errMsg: ServerMessage = { ch: "sessions", type: "error", error }; + ws.send(JSON.stringify(errMsg)); + }, + ); } } } catch (err) { diff --git a/packages/web/src/__tests__/api-routes.test.ts b/packages/web/src/__tests__/api-routes.test.ts index 6cfecdc9d..65b7e06d2 100644 --- a/packages/web/src/__tests__/api-routes.test.ts +++ b/packages/web/src/__tests__/api-routes.test.ts @@ -221,7 +221,6 @@ import { POST as killPOST } from "@/app/api/sessions/[id]/kill/route"; import { POST as restorePOST } from "@/app/api/sessions/[id]/restore/route"; import { POST as remapPOST } from "@/app/api/sessions/[id]/remap/route"; import { POST as mergePOST } from "@/app/api/prs/[id]/merge/route"; -import { GET as eventsGET } from "@/app/api/events/route"; import { GET as observabilityGET } from "@/app/api/observability/route"; import { GET as runtimeTerminalGET } from "@/app/api/runtime/terminal/route"; import { GET as verifyGET, POST as verifyPOST } from "@/app/api/verify/route"; @@ -1286,35 +1285,6 @@ describe("API Routes", () => { }); }); - // ── GET /api/events (SSE) ────────────────────────────────────────── - - describe("GET /api/events", () => { - it("returns SSE content type", async () => { - const req = makeRequest("/api/events", { method: "GET" }); - const res = await eventsGET(req); - expect(res.headers.get("Content-Type")).toBe("text/event-stream"); - expect(res.headers.get("Cache-Control")).toBe("no-cache"); - }); - - it("streams initial snapshot event", async () => { - const req = makeRequest("/api/events", { method: "GET" }); - const res = await eventsGET(req); - const reader = res.body!.getReader(); - const { value } = await reader.read(); - reader.cancel(); - const text = new TextDecoder().decode(value); - expect(text).toContain("data: "); - const jsonStr = text.replace("data: ", "").trim(); - const event = JSON.parse(jsonStr); - expect(event.type).toBe("snapshot"); - expect(event.correlationId).toBeTruthy(); - expect(Array.isArray(event.sessions)).toBe(true); - expect(event.sessions.length).toBeGreaterThan(0); - expect(event.sessions[0]).toHaveProperty("id"); - expect(event.sessions[0]).toHaveProperty("attentionLevel"); - }); - }); - describe("GET /api/observability", () => { it("returns observability summary with correlation header", async () => { const req = makeRequest("/api/observability", { method: "GET" }); diff --git a/packages/web/src/app/api/events/route.ts b/packages/web/src/app/api/events/route.ts deleted file mode 100644 index d89014601..000000000 --- a/packages/web/src/app/api/events/route.ts +++ /dev/null @@ -1,266 +0,0 @@ -import { getServices } from "@/lib/services"; -import { sessionToDashboard } from "@/lib/serialize"; -import { getAttentionLevel } from "@/lib/types"; -import { filterWorkerSessions } from "@/lib/project-utils"; -import { - createCorrelationId, - createProjectObserver, - type ProjectObserver, -} from "@aoagents/ao-core"; - -export const dynamic = "force-dynamic"; - -const SESSION_EVENTS_POLL_INTERVAL_MS = 5000; - -/** - * GET /api/events — SSE stream for real-time lifecycle events - * - * Sends session state updates to connected clients. - * Polls SessionManager.list() on an interval (no SSE push from core yet). - */ -export async function GET(request: Request): Promise { - const encoder = new TextEncoder(); - const correlationId = createCorrelationId("sse"); - const { searchParams } = new URL(request.url); - const projectFilter = searchParams.get("project"); - type ServicesConfig = Awaited>["config"]; - let heartbeat: ReturnType | undefined; - let updates: ReturnType | undefined; - let observerProjectId: string | undefined; - let observer: ProjectObserver | null = null; - let streamClosed = false; - - const ensureObserver = (config: ServicesConfig): ProjectObserver | null => { - if (!observerProjectId) { - const requestedProjectId = - projectFilter && projectFilter !== "all" && config.projects[projectFilter] - ? projectFilter - : undefined; - observerProjectId = requestedProjectId ?? Object.keys(config.projects)[0]; - } - if (!observerProjectId) return null; - if (!observer) { - observer = createProjectObserver(config, "web-events"); - } - return observer; - }; - - const stopStream = () => { - if (streamClosed) return; - streamClosed = true; - if (heartbeat) clearInterval(heartbeat); - if (updates) clearInterval(updates); - }; - - const encodeEvent = (payload: unknown) => encoder.encode(`data: ${JSON.stringify(payload)}\n\n`); - - const stream = new ReadableStream({ - start(controller) { - const safeEnqueue = (payload: unknown): boolean => { - if (streamClosed) return false; - try { - controller.enqueue(encodeEvent(payload)); - return true; - } catch { - stopStream(); - return false; - } - }; - - void (async () => { - try { - const { config } = await getServices(); - const projectObserver = ensureObserver(config); - if (projectObserver && observerProjectId) { - projectObserver.recordOperation({ - metric: "sse_connect", - operation: "sse.connect", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { path: "/api/events" }, - level: "info", - }); - projectObserver.setHealth({ - surface: "sse.events", - status: "ok", - projectId: observerProjectId, - correlationId, - details: { projectId: observerProjectId, connection: "open" }, - }); - } - } catch { - void 0; - } - - try { - const { config, sessionManager } = await getServices(); - const requestedProjectId = - projectFilter && projectFilter !== "all" && config.projects[projectFilter] - ? projectFilter - : undefined; - const sessions = await sessionManager.list(requestedProjectId); - const workerSessions = filterWorkerSessions(sessions, projectFilter, config.projects); - const dashboardSessions = workerSessions.map(sessionToDashboard); - const projectObserver = ensureObserver(config); - - const attentionZones = config.dashboard?.attentionZones ?? "simple"; - const initialEvent = { - type: "snapshot", - correlationId, - emittedAt: new Date().toISOString(), - sessions: dashboardSessions.map((s) => ({ - id: s.id, - status: s.status, - activity: s.activity, - attentionLevel: getAttentionLevel(s, attentionZones), - lastActivityAt: s.lastActivityAt, - })), - }; - safeEnqueue(initialEvent); - if (projectObserver && observerProjectId) { - projectObserver.recordOperation({ - metric: "sse_snapshot", - operation: "sse.snapshot", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { sessionCount: dashboardSessions.length, initial: true }, - level: "info", - }); - } - } catch (error) { - safeEnqueue({ - type: "error", - correlationId, - emittedAt: new Date().toISOString(), - error: error instanceof Error ? error.message : "Failed to load live dashboard data", - }); - } - })(); - - // Send periodic heartbeat - heartbeat = setInterval(() => { - if (streamClosed) return; - try { - controller.enqueue(encoder.encode(`: heartbeat\n\n`)); - } catch { - stopStream(); - } - }, 15000); - - // Poll for session state changes frequently enough that new workers - // appear in the dashboard/sidebar quickly after the orchestrator spawns them. - updates = setInterval(() => { - void (async () => { - let dashboardSessions; - try { - const { config, sessionManager } = await getServices(); - const requestedProjectId = - projectFilter && projectFilter !== "all" && config.projects[projectFilter] - ? projectFilter - : undefined; - const sessions = await sessionManager.list(requestedProjectId); - const workerSessions = filterWorkerSessions(sessions, projectFilter, config.projects); - dashboardSessions = workerSessions.map(sessionToDashboard); - const projectObserver = ensureObserver(config); - - if (projectObserver && observerProjectId) { - projectObserver.setHealth({ - surface: "sse.events", - status: "ok", - projectId: observerProjectId, - correlationId, - details: { - projectId: observerProjectId, - sessionCount: dashboardSessions.length, - lastEventAt: new Date().toISOString(), - }, - }); - } - - try { - const attentionZones = config.dashboard?.attentionZones ?? "simple"; - const event = { - type: "snapshot", - correlationId, - emittedAt: new Date().toISOString(), - sessions: dashboardSessions.map((s) => ({ - id: s.id, - status: s.status, - activity: s.activity, - attentionLevel: getAttentionLevel(s, attentionZones), - lastActivityAt: s.lastActivityAt, - })), - }; - safeEnqueue(event); - if (projectObserver && observerProjectId) { - projectObserver.recordOperation({ - metric: "sse_snapshot", - operation: "sse.snapshot", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { sessionCount: dashboardSessions.length, initial: false }, - level: "info", - }); - } - } catch (error) { - safeEnqueue({ - type: "error", - correlationId, - emittedAt: new Date().toISOString(), - error: error instanceof Error ? error.message : "Live dashboard update failed", - }); - } - } catch (error) { - safeEnqueue({ - type: "error", - correlationId, - emittedAt: new Date().toISOString(), - error: error instanceof Error ? error.message : "Live dashboard update failed", - }); - } - })(); - }, SESSION_EVENTS_POLL_INTERVAL_MS); - }, - cancel() { - stopStream(); - void (async () => { - try { - const { config } = await getServices(); - const projectObserver = ensureObserver(config); - if (!projectObserver || !observerProjectId) return; - projectObserver.recordOperation({ - metric: "sse_disconnect", - operation: "sse.disconnect", - outcome: "success", - correlationId, - projectId: observerProjectId, - data: { path: "/api/events" }, - level: "info", - }); - projectObserver.setHealth({ - surface: "sse.events", - status: "warn", - projectId: observerProjectId, - correlationId, - reason: "SSE connection closed", - details: { projectId: observerProjectId, connection: "closed" }, - }); - } catch { - void 0; - } - })(); - }, - }); - - return new Response(stream, { - headers: { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - Connection: "keep-alive", - "X-Accel-Buffering": "no", - }, - }); -} diff --git a/packages/web/src/app/sessions/[id]/page.test.tsx b/packages/web/src/app/sessions/[id]/page.test.tsx index eb8b0622c..6a295dcb8 100644 --- a/packages/web/src/app/sessions/[id]/page.test.tsx +++ b/packages/web/src/app/sessions/[id]/page.test.tsx @@ -31,6 +31,10 @@ vi.mock("@/components/SessionDetail", () => ({ }, })); +vi.mock("@/hooks/useMuxSessionActivity", () => ({ + useMuxSessionActivity: () => null, +})); + function makeWorkerSession(): DashboardSession { return { id: "worker-1", diff --git a/packages/web/src/app/sessions/[id]/page.tsx b/packages/web/src/app/sessions/[id]/page.tsx index 307cd7638..fb32d5a67 100644 --- a/packages/web/src/app/sessions/[id]/page.tsx +++ b/packages/web/src/app/sessions/[id]/page.tsx @@ -11,7 +11,7 @@ import { type DashboardSession, type ActivityState, getAttentionLevel } from "@/ import { activityIcon } from "@/lib/activity-icons"; import type { ProjectInfo } from "@/lib/project-name"; import { getSessionTitle } from "@/lib/format"; -import { useSSESessionActivity } from "@/hooks/useSSESessionActivity"; +import { useMuxSessionActivity } from "@/hooks/useMuxSessionActivity"; import { useMuxOptional } from "@/providers/MuxProvider"; import type { SessionPatch } from "@/lib/mux-protocol"; import { projectSessionPath } from "@/lib/routes"; @@ -439,8 +439,8 @@ export default function SessionPage() { } }, []); - // Subscribe to SSE for real-time activity updates (title emoji) - const sseActivity = useSSESessionActivity(id, sessionProjectId ?? expectedProjectId ?? undefined); + // Get real-time activity updates from WebSocket (title emoji) + const sseActivity = useMuxSessionActivity(id); // Update document title based on session data + SSE activity override useEffect(() => { diff --git a/packages/web/src/components/Dashboard.tsx b/packages/web/src/components/Dashboard.tsx index 6ac961737..a98c75b31 100644 --- a/packages/web/src/components/Dashboard.tsx +++ b/packages/web/src/components/Dashboard.tsx @@ -158,23 +158,28 @@ function DashboardInner({ } return levels; }, [initialSessions, attentionZones]); - const { sessions, connectionStatus, sseAttentionLevels, liveSessionsResolved, loadError } = - useSessionEvents({ - initialSessions, - // No project filter — sidebar needs all sessions across all projects. - // Kanban filtering is applied client-side via projectSessions below. - muxSessions: mux?.status === "connected" ? mux.sessions : undefined, - initialAttentionLevels, - attentionZones, - }); + const { sessions, attentionLevels, liveSessionsResolved, loadError } = useSessionEvents({ + initialSessions, + // No project filter — sidebar needs all sessions across all projects. + // Kanban filtering is applied client-side via projectSessions below. + muxSessions: mux?.status === "connected" ? mux.sessions : undefined, + muxLastError: mux?.lastError, + initialAttentionLevels, + attentionZones, + }); const projectSessions = useMemo(() => { if (!projectId) return sessions; return sessions.filter((s) => s.projectId === projectId); }, [sessions, projectId]); + const connectionStatus: "connected" | "reconnecting" | "disconnected" = + mux?.status === "disconnected" ? "disconnected" + : mux?.status === "connected" ? "connected" + : "reconnecting"; const recoveredFromLoadError = Boolean(dashboardLoadError) && liveSessionsResolved; - const visibleDashboardLoadError = - loadError ?? (recoveredFromLoadError ? undefined : dashboardLoadError); + const ssrLoadError = recoveredFromLoadError ? undefined : dashboardLoadError; + // Live WS error takes precedence; fall back to SSR load error when live data hasn't resolved it. + const visibleLoadError = loadError ?? ssrLoadError; const searchParams = useSearchParams(); const router = useRouter(); const routerRef = useRef(router); @@ -229,12 +234,12 @@ function DashboardInner({ setActiveOrchestrators((current) => mergeOrchestrators(current, orchestratorLinks)); }, [orchestratorLinks]); - // Update document title with live attention counts from SSE + // Update document title with live attention counts useEffect(() => { - const needsAttention = countNeedingAttention(sseAttentionLevels); + const needsAttention = countNeedingAttention(attentionLevels); const label = projectName ?? "ao"; document.title = needsAttention > 0 ? `${label} (${needsAttention} need attention)` : label; - }, [sseAttentionLevels, projectName]); + }, [attentionLevels, projectName]); useEffect(() => { setMobileMenuOpen(false); @@ -442,9 +447,9 @@ function DashboardInner({ }; const hasAnySessions = kanbanLevels.some((level) => grouped[level].length > 0); - const showEmptyState = !allProjectsView && !hasAnySessions && !visibleDashboardLoadError; + const showEmptyState = !allProjectsView && !hasAnySessions && !visibleLoadError; - const loadErrorBanner = visibleDashboardLoadError ? ( + const loadErrorBanner = visibleLoadError ? (
- {visibleDashboardLoadError} + {visibleLoadError} Confirm agent-orchestrator.yaml exists and is @@ -612,8 +617,8 @@ function DashboardInner({
setMobileMenuOpen(false)} /> )} -
- +
+

Dashboard

diff --git a/packages/web/src/components/DynamicFavicon.tsx b/packages/web/src/components/DynamicFavicon.tsx index 36603ad49..0219c20b7 100644 --- a/packages/web/src/components/DynamicFavicon.tsx +++ b/packages/web/src/components/DynamicFavicon.tsx @@ -1,15 +1,15 @@ "use client"; import { useEffect } from "react"; -import type { SSEAttentionMap } from "@/hooks/useSessionEvents"; +import type { AttentionMap } from "@/hooks/useSessionEvents"; /** - * Determine overall health from SSE attention levels. + * Determine overall health from attention levels. * - "green" — all sessions working/done/pending, nothing needs attention * - "yellow" — some sessions need review or response * - "red" — critical: sessions stuck, errored, or needing immediate action */ -function computeHealthFromLevels(levels: SSEAttentionMap): "green" | "yellow" | "red" { +function computeHealthFromLevels(levels: AttentionMap): "green" | "yellow" | "red" { const entries = Object.values(levels); if (entries.length === 0) return "green"; @@ -46,7 +46,7 @@ function generateFaviconSvg(initial: string, color: string): string { } /** Count sessions that need human attention (respond, review, action, merge). */ -export function countNeedingAttention(levels: SSEAttentionMap): number { +export function countNeedingAttention(levels: AttentionMap): number { let count = 0; for (const level of Object.values(levels)) { if ( @@ -62,23 +62,23 @@ export function countNeedingAttention(levels: SSEAttentionMap): number { } interface DynamicFaviconProps { - /** Server-computed attention levels from SSE snapshots. */ - sseAttentionLevels: SSEAttentionMap; + /** Server-computed attention levels from session snapshots. */ + attentionLevels: AttentionMap; projectName?: string; } /** * Client component that dynamically updates the browser favicon - * based on system health (session attention levels from SSE). + * based on system health (session attention levels). * - * Uses server-computed attention levels from SSE snapshots for real-time - * updates, rather than recomputing from the full sessions array. + * Uses server-computed attention levels for real-time updates, + * rather than recomputing from the full sessions array. */ -export function DynamicFavicon({ sseAttentionLevels, projectName = "A" }: DynamicFaviconProps) { +export function DynamicFavicon({ attentionLevels, projectName = "A" }: DynamicFaviconProps) { const initial = projectName.charAt(0).toUpperCase(); useEffect(() => { - const health = computeHealthFromLevels(sseAttentionLevels); + const health = computeHealthFromLevels(attentionLevels); const color = HEALTH_COLORS[health]; const href = generateFaviconSvg(initial, color); @@ -91,7 +91,7 @@ export function DynamicFavicon({ sseAttentionLevels, projectName = "A" }: Dynami } link.type = "image/svg+xml"; link.href = href; - }, [sseAttentionLevels, initial]); + }, [attentionLevels, initial]); return null; } diff --git a/packages/web/src/components/PullRequestsPage.tsx b/packages/web/src/components/PullRequestsPage.tsx index be0079b7d..17b72d84a 100644 --- a/packages/web/src/components/PullRequestsPage.tsx +++ b/packages/web/src/components/PullRequestsPage.tsx @@ -52,10 +52,10 @@ export function PullRequestsPage({ }: PullRequestsPageProps) { const orchestratorLinks = orchestrators ?? EMPTY_ORCHESTRATORS; const mux = useMuxOptional(); - // Seed initial attention levels using the same mode the server SSE will - // use when it sends snapshots (read from `config.dashboard.attentionZones` - // upstream). This prevents the sseAttentionLevels map from oscillating - // between detailed (seed/refresh) and simple (server snapshot) values. + // Seed initial attention levels using the same mode used during refresh + // (read from `config.dashboard.attentionZones` upstream). This prevents + // the attentionLevels map from oscillating between detailed (seed/refresh) + // and simple (server snapshot) values. const initialAttentionLevels = useMemo(() => { const levels: Record> = {}; for (const s of initialSessions) { @@ -63,7 +63,7 @@ export function PullRequestsPage({ } return levels; }, [initialSessions, attentionZones]); - const { sessions, sseAttentionLevels } = useSessionEvents({ + const { sessions, attentionLevels } = useSessionEvents({ initialSessions, project: projectId, muxSessions: mux?.status === "connected" ? mux.sessions : undefined, @@ -127,7 +127,7 @@ export function PullRequestsPage({

setMobileMenuOpen(false)} /> )}
- + {isMobile ? (
diff --git a/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx b/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx index bd72334d3..18aa4b742 100644 --- a/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx +++ b/packages/web/src/components/__tests__/Dashboard.emptyState.test.tsx @@ -1,13 +1,6 @@ -import { fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { act, fireEvent, render, screen, waitFor } from "@testing-library/react"; +import { useState } from "react"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import { Dashboard } from "../Dashboard"; - -const eventSourceConstructorMock = vi.hoisted(() => vi.fn()); -let lastEventSourceMock: { - onmessage: ((event: MessageEvent) => void) | null; - onerror: ((event?: Event) => void) | null; - close: ReturnType; -}; vi.mock("next/navigation", () => ({ useRouter: () => ({ push: vi.fn(), replace: vi.fn(), refresh: vi.fn() }), @@ -15,21 +8,29 @@ vi.mock("next/navigation", () => ({ useSearchParams: () => new URLSearchParams(), })); +let currentMuxLastError: string | null = null; + +vi.mock("@/providers/MuxProvider", () => ({ + useMuxOptional: () => ({ + subscribeTerminal: () => () => {}, + writeTerminal: () => {}, + openTerminal: () => {}, + closeTerminal: () => {}, + resizeTerminal: () => {}, + // "connecting" so muxSessions stays undefined — prevents snapshot dispatch from + // immediately flipping liveSessionsResolved and masking the SSR load error. + status: "connecting" as const, + sessions: [], + lastError: currentMuxLastError, + }), + MuxProvider: ({ children }: { children: React.ReactNode }) => <>{children}, +})); + +import { Dashboard } from "../Dashboard"; + beforeEach(() => { - const eventSourceMock = { - onmessage: null, - onerror: null, - close: vi.fn(), - }; - lastEventSourceMock = eventSourceMock; - eventSourceConstructorMock.mockReset(); - eventSourceConstructorMock.mockImplementation(() => eventSourceMock as unknown as EventSource); - global.EventSource = Object.assign(eventSourceConstructorMock, { - CONNECTING: 0, - OPEN: 1, - CLOSED: 2, - }) as unknown as typeof EventSource; global.fetch = vi.fn(); + currentMuxLastError = null; }); describe("Dashboard empty state", () => { @@ -113,36 +114,51 @@ describe("Dashboard empty state", () => { expect(screen.queryByText(/Ready to orchestrate/i)).not.toBeInTheDocument(); expect(screen.getByRole("alert")).toHaveTextContent("Orchestrator failed to load"); expect(screen.getByRole("alert")).toHaveTextContent("No agent-orchestrator.yaml found"); - expect(eventSourceConstructorMock).toHaveBeenCalledTimes(1); }); - it("shows a live load error banner when the SSE stream reports a backend failure", async () => { - render(); + it("shows live load-error banner when WS transport reports a fetch failure", async () => { + let forceUpdate: () => void = () => {}; + function Wrapper() { + const [, tick] = useState(0); + forceUpdate = () => tick((n) => n + 1); + return ; + } - await waitFor(() => { - lastEventSourceMock.onmessage?.({ - data: JSON.stringify({ type: "error", error: "session list timed out" }), - } as MessageEvent); - expect(screen.getByRole("alert")).toHaveTextContent("session list timed out"); + render(); + expect(screen.queryByRole("alert")).not.toBeInTheDocument(); + + await act(async () => { + currentMuxLastError = "Session fetch failed: HTTP 503"; + forceUpdate(); }); + + expect(screen.getByRole("alert")).toHaveTextContent("Orchestrator failed to load"); + expect(screen.getByRole("alert")).toHaveTextContent("Session fetch failed: HTTP 503"); }); - it("clears a live load error banner after a healthy snapshot with unchanged sessions", async () => { - render(); + it("clears live load-error banner when the next WS snapshot succeeds", async () => { + let forceUpdate: () => void = () => {}; + function Wrapper() { + const [, tick] = useState(0); + forceUpdate = () => tick((n) => n + 1); + return ; + } - await waitFor(() => { - lastEventSourceMock.onmessage?.({ - data: JSON.stringify({ type: "error", error: "session list timed out" }), - } as MessageEvent); - expect(screen.getByRole("alert")).toHaveTextContent("session list timed out"); + render(); + + await act(async () => { + currentMuxLastError = "Session fetch failed: HTTP 503"; + forceUpdate(); }); - await waitFor(() => { - lastEventSourceMock.onmessage?.({ - data: JSON.stringify({ type: "snapshot", sessions: [] }), - } as MessageEvent); - expect(screen.queryByRole("alert")).not.toBeInTheDocument(); + expect(screen.getByRole("alert")).toBeInTheDocument(); + + await act(async () => { + currentMuxLastError = null; + forceUpdate(); }); + + expect(screen.queryByRole("alert")).not.toBeInTheDocument(); }); it("shows empty state when only done sessions exist", () => { diff --git a/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx b/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx index 7ee103c91..a00ef4a5d 100644 --- a/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx +++ b/packages/web/src/components/__tests__/Dashboard.renderCadence.test.tsx @@ -1,5 +1,5 @@ -import { act, render, waitFor } from "@testing-library/react"; -import { memo } from "react"; +import { act, render } from "@testing-library/react"; +import { memo, useState } from "react"; import { beforeEach, describe, expect, it, vi } from "vitest"; const renderCounts = new Map(); @@ -17,69 +17,108 @@ vi.mock("@/components/SessionCard", () => ({ }), })); +import type { SessionPatch } from "@/lib/mux-protocol"; + +// Module-level sessions updated by tests, read by the mock on every render. +let currentMuxSessions: SessionPatch[] = []; + +vi.mock("@/providers/MuxProvider", () => ({ + useMuxOptional: () => ({ + subscribeTerminal: () => () => {}, + writeTerminal: () => {}, + openTerminal: () => {}, + closeTerminal: () => {}, + resizeTerminal: () => {}, + status: "connected" as const, + sessions: currentMuxSessions, + }), + MuxProvider: ({ children }: { children: React.ReactNode }) => <>{children}, +})); + import { Dashboard } from "../Dashboard"; import { makeSession } from "../../__tests__/helpers"; -describe("Dashboard render cadence", () => { - let eventSourceMock: { - onmessage: ((event: MessageEvent) => void) | null; - onerror: (() => void) | null; - close: () => void; - }; +// Wrapper that exposes a forceUpdate so tests can trigger re-renders after +// updating currentMuxSessions, causing Dashboard to re-call useMuxOptional(). +let forceUpdate: () => void = () => {}; +function ControlledDashboard({ + initialSessions, +}: { + initialSessions: ReturnType[]; +}) { + const [, tick] = useState(0); + forceUpdate = () => tick((n) => n + 1); + return ; +} + +describe("Dashboard render cadence", () => { beforeEach(() => { renderCounts.clear(); - eventSourceMock = { - onmessage: null, - onerror: null, - close: vi.fn(), - }; - const eventSourceConstructor = vi.fn(() => eventSourceMock as unknown as EventSource); - global.EventSource = Object.assign(eventSourceConstructor, { - CONNECTING: 0, - OPEN: 1, - CLOSED: 2, - }) as unknown as typeof EventSource; + currentMuxSessions = []; global.fetch = vi.fn(); }); it("rerenders only the changed session card for same-membership snapshots", async () => { + const ts = new Date().toISOString(); const initialSessions = [ - makeSession({ id: "session-1", summary: "First session" }), - makeSession({ id: "session-2", summary: "Second session" }), + makeSession({ id: "session-1", status: "working", activity: "active", lastActivityAt: ts }), + makeSession({ id: "session-2", status: "working", activity: "active", lastActivityAt: ts }), ]; - render(); + render(); expect(renderCounts.get("session-1")).toBe(1); expect(renderCounts.get("session-2")).toBe(1); - await waitFor(() => expect(eventSourceMock.onmessage).not.toBeNull()); - + // Push a snapshot where only session-1 changes await act(async () => { - eventSourceMock.onmessage!({ - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { - id: "session-1", - status: "working", - activity: "idle", - lastActivityAt: new Date().toISOString(), - }, - { - id: "session-2", - status: initialSessions[1].status, - activity: initialSessions[1].activity, - lastActivityAt: initialSessions[1].lastActivityAt, - }, - ], - }), - } as MessageEvent); + currentMuxSessions = [ + { + id: "session-1", + status: "working", + activity: "idle", + attentionLevel: "working", + lastActivityAt: new Date(Date.now() + 1000).toISOString(), + }, + { + id: "session-2", + status: "working", + activity: "active", + attentionLevel: "working", + lastActivityAt: ts, + }, + ]; + forceUpdate(); }); expect(renderCounts.get("session-1")).toBe(2); expect(renderCounts.get("session-2")).toBe(1); expect(fetch).not.toHaveBeenCalled(); }); + + it("does not rerender any card when snapshot data is identical", async () => { + const ts = new Date().toISOString(); + const initialSessions = [ + makeSession({ id: "session-1", status: "working", activity: "active", lastActivityAt: ts }), + ]; + + render(); + const countAfterInit = renderCounts.get("session-1") ?? 0; + + await act(async () => { + currentMuxSessions = [ + { + id: "session-1", + status: "working", + activity: "active", + attentionLevel: "working", + lastActivityAt: ts, + }, + ]; + forceUpdate(); + }); + + expect(renderCounts.get("session-1")).toBe(countAfterInit); + }); }); diff --git a/packages/web/src/components/__tests__/DynamicFavicon.test.tsx b/packages/web/src/components/__tests__/DynamicFavicon.test.tsx index 0c48802ca..2ddf5e05a 100644 --- a/packages/web/src/components/__tests__/DynamicFavicon.test.tsx +++ b/packages/web/src/components/__tests__/DynamicFavicon.test.tsx @@ -1,7 +1,7 @@ import { describe, it, expect, beforeEach } from "vitest"; import { render } from "@testing-library/react"; import { countNeedingAttention, DynamicFavicon } from "../DynamicFavicon"; -import type { SSEAttentionMap } from "@/hooks/useSessionEvents"; +import type { AttentionMap } from "@/hooks/useSessionEvents"; describe("countNeedingAttention", () => { it("returns 0 for empty map", () => { @@ -9,7 +9,7 @@ describe("countNeedingAttention", () => { }); it("returns 0 when all sessions are working/pending/done", () => { - const levels: SSEAttentionMap = { + const levels: AttentionMap = { "s-1": "working", "s-2": "pending", "s-3": "done", @@ -18,7 +18,7 @@ describe("countNeedingAttention", () => { }); it("counts respond, review, action, and merge sessions", () => { - const levels: SSEAttentionMap = { + const levels: AttentionMap = { "s-1": "respond", "s-2": "review", "s-3": "merge", @@ -30,7 +30,7 @@ describe("countNeedingAttention", () => { }); it("counts a single attention-needing session", () => { - const levels: SSEAttentionMap = { + const levels: AttentionMap = { "s-1": "respond", }; expect(countNeedingAttention(levels)).toBe(1); @@ -44,8 +44,8 @@ describe("DynamicFavicon", () => { }); it("creates a green favicon when no sessions need attention", () => { - const levels: SSEAttentionMap = { "s-1": "working", "s-2": "done" }; - render(); + const levels: AttentionMap = { "s-1": "working", "s-2": "done" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link).not.toBeNull(); @@ -53,16 +53,16 @@ describe("DynamicFavicon", () => { }); it("creates a yellow favicon when sessions need review", () => { - const levels: SSEAttentionMap = { "s-1": "review", "s-2": "working" }; - render(); + const levels: AttentionMap = { "s-1": "review", "s-2": "working" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23eab308"); // yellow }); it("creates a red favicon when sessions need response", () => { - const levels: SSEAttentionMap = { "s-1": "respond" }; - render(); + const levels: AttentionMap = { "s-1": "respond" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23ef4444"); // red @@ -72,32 +72,32 @@ describe("DynamicFavicon", () => { // "action" collapses respond + review, so it contains routine review work // (ci_failed, changes_requested). Escalating to red would cry wolf on // every typical PR. - const levels: SSEAttentionMap = { "s-1": "action", "s-2": "working" }; - render(); + const levels: AttentionMap = { "s-1": "action", "s-2": "working" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23eab308"); // yellow }); it("still escalates to red when detailed 'respond' is present alongside 'action'", () => { - const levels: SSEAttentionMap = { "s-1": "action", "s-2": "respond" }; - render(); + const levels: AttentionMap = { "s-1": "action", "s-2": "respond" }; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23ef4444"); // red }); it("uses first letter of projectName as initial", () => { - const levels: SSEAttentionMap = {}; - render(); + const levels: AttentionMap = {}; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("M"); // initial letter }); it("defaults to A when no projectName given", () => { - const levels: SSEAttentionMap = {}; - render(); + const levels: AttentionMap = {}; + render(); const link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("A"); @@ -105,13 +105,13 @@ describe("DynamicFavicon", () => { it("updates favicon when attention levels change", () => { const { rerender } = render( - , + , ); let link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%2322c55e"); // green - rerender(); + rerender(); link = document.querySelector('link[rel="icon"]'); expect(link!.href).toContain("%23ef4444"); // red diff --git a/packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts b/packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts deleted file mode 100644 index 3f2102990..000000000 --- a/packages/web/src/hooks/__tests__/useSSESessionActivity.test.ts +++ /dev/null @@ -1,207 +0,0 @@ -import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; -import { renderHook, act } from "@testing-library/react"; -import { useSSESessionActivity } from "../useSSESessionActivity"; - -describe("useSSESessionActivity", () => { - let eventSourceMock: { - onmessage: ((event: MessageEvent) => void) | null; - onopen: (() => void) | null; - onerror: (() => void) | null; - readyState: number; - close: ReturnType; - }; - - beforeEach(() => { - const eventSourceConstructor = vi.fn(() => { - const instance = { - onmessage: null as ((event: MessageEvent) => void) | null, - onopen: null as (() => void) | null, - onerror: null as (() => void) | null, - readyState: 0, - close: vi.fn(), - }; - eventSourceMock = instance; - return instance as unknown as EventSource; - }); - global.EventSource = Object.assign(eventSourceConstructor, { - CONNECTING: 0, - OPEN: 1, - CLOSED: 2, - }) as unknown as typeof EventSource; - }); - - afterEach(() => { - vi.restoreAllMocks(); - }); - - it("returns null initially", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - expect(result.current).toBeNull(); - }); - - it("updates activity when SSE snapshot contains the target session", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - { id: "session-2", status: "working", activity: "idle", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current).toEqual({ activity: "active" }); - }); - - it("updates when activity changes", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current?.activity).toBe("active"); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "idle", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current?.activity).toBe("idle"); - }); - - it("does not update when activity stays the same (referential stability)", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - const first = result.current; - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current).toBe(first); - }); - - it("ignores snapshots that do not contain the session", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-other", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current).toBeNull(); - }); - - it("passes project parameter to EventSource URL", () => { - renderHook(() => useSSESessionActivity("session-1", "my-project")); - expect(global.EventSource).toHaveBeenCalledWith("/api/events?project=my-project"); - }); - - it("uses default URL when no project is specified", () => { - renderHook(() => useSSESessionActivity("session-1")); - expect(global.EventSource).toHaveBeenCalledWith("/api/events"); - }); - - it("closes EventSource on unmount", () => { - const { unmount } = renderHook(() => useSSESessionActivity("session-1")); - unmount(); - expect(eventSourceMock.close).toHaveBeenCalled(); - }); - - it("ignores non-snapshot events", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: JSON.stringify({ type: "heartbeat" }), - } as MessageEvent); - }); - - expect(result.current).toBeNull(); - }); - - it("handles malformed JSON gracefully", () => { - const { result } = renderHook(() => useSSESessionActivity("session-1")); - - act(() => { - eventSourceMock.onmessage!.call(eventSourceMock, { - data: "not-json", - } as MessageEvent); - }); - - expect(result.current).toBeNull(); - }); - - it("resets activity and reconnects when sessionId changes", () => { - let currentSessionId = "session-1"; - const { result, rerender } = renderHook(() => - useSSESessionActivity(currentSessionId), - ); - - const firstInstance = eventSourceMock; - - act(() => { - firstInstance.onmessage!.call(firstInstance, { - data: JSON.stringify({ - type: "snapshot", - sessions: [ - { id: "session-1", status: "working", activity: "active", attentionLevel: "working", lastActivityAt: new Date().toISOString() }, - ], - }), - } as MessageEvent); - }); - - expect(result.current?.activity).toBe("active"); - - currentSessionId = "session-2"; - rerender(); - - // State should reset to null immediately on sessionId change - expect(result.current).toBeNull(); - - // First EventSource should have been closed - expect(firstInstance.close).toHaveBeenCalled(); - }); -}); diff --git a/packages/web/src/hooks/useMuxSessionActivity.ts b/packages/web/src/hooks/useMuxSessionActivity.ts new file mode 100644 index 000000000..34d39804c --- /dev/null +++ b/packages/web/src/hooks/useMuxSessionActivity.ts @@ -0,0 +1,16 @@ +"use client"; + +import { useMemo } from "react"; +import type { ActivityState } from "@/lib/types"; +import { useMuxOptional } from "@/providers/MuxProvider"; + +export function useMuxSessionActivity( + sessionId: string, +): { activity: ActivityState | null } | null { + const mux = useMuxOptional(); + const patch = mux?.sessions.find((s) => s.id === sessionId); + return useMemo( + () => (patch ? { activity: (patch.activity as ActivityState) ?? null } : null), + [patch], + ); +} diff --git a/packages/web/src/hooks/useSSESessionActivity.ts b/packages/web/src/hooks/useSSESessionActivity.ts deleted file mode 100644 index b528b4318..000000000 --- a/packages/web/src/hooks/useSSESessionActivity.ts +++ /dev/null @@ -1,54 +0,0 @@ -"use client"; - -import { useEffect, useState } from "react"; -import type { ActivityState, SSESnapshotEvent } from "@/lib/types"; - -interface SessionActivity { - activity: ActivityState | null; -} - -/** - * Lightweight SSE subscriber that tracks a single session's activity state. - * - * Used by the session detail page to update document.title emoji in real-time - * without waiting for the full session HTTP poll cycle. - */ -export function useSSESessionActivity( - sessionId: string, - project?: string, -): SessionActivity | null { - const [activity, setActivity] = useState(null); - - useEffect(() => { - setActivity(null); - const url = project ? `/api/events?project=${encodeURIComponent(project)}` : "/api/events"; - const es = new EventSource(url); - let disposed = false; - - es.onmessage = (event: MessageEvent) => { - if (disposed) return; - try { - const data = JSON.parse(event.data as string) as { type: string }; - if (data.type !== "snapshot") return; - - const snapshot = data as SSESnapshotEvent; - const match = snapshot.sessions.find((s) => s.id === sessionId); - if (!match) return; - - setActivity((prev) => { - if (prev && prev.activity === match.activity) return prev; - return { activity: match.activity }; - }); - } catch { - return; - } - }; - - return () => { - disposed = true; - es.close(); - }; - }, [sessionId, project]); - - return activity; -} diff --git a/packages/web/src/hooks/useSessionEvents.ts b/packages/web/src/hooks/useSessionEvents.ts index d2df890d90220efe943773234177d73901bf7877..afe4aed7cd16b756dc14113917a7612cc103fe26 100644 GIT binary patch delta 2332 zcmb7FO>Y}j6vgDD&XA@}h2y3l>a9gwJ8^9KK`AzA;+7O5g|@1lAQ~cfGJdW{9={o8 z-o&+9%M`I=)9SKdK|%ckAY_S6kw7e2K!wDrLWoVJqAt2%(IxlI*okcl5?MHR-n=*W z+;i_a@2{JIAK0zSNl~{LT!$QU9Npv>RiTxk0VUh4LS!Lj=o_i3du`UTO<`)LarK^NZ*6OBc=;mKRqS zmvb-&oYg3&cs(lwd#1&QF;KL%`dGk<=-vj^4a`i55DedB|Hgh zQ{uDf<8}J47Y+GXQ*}W=3Ocu35lJ*78-}S>n1s!)>4JhSZ2y4PlcN_|ZZVgW#7ST) z^w33JyJbG{5}ISFHQuJBxy=g`j1qWkS$4rB?g*2n7fPybc=&84X)SD%s+CoKl{p2Q zz3(zdVyAXkGy|Jn=ygcl@XGdL`Cc=a%3X^h&REfDc8G{L!qqCY%xv4Vz4H~_@DsPG z^E#J^2xeDxPDAw{RQ1-JQIyp5G|iE{$?RFptP=IC2rA7KUKu=yHwP#2&LAI>o3gk* z2xtwR7?-cNN6twD-Ww|IQ~+QaY*wOiE91g&{LDK!pc&yr9yoH=6HqJbq>hqKTm0=C ztVA_|%Io0xGJv8gR8Vc3hQz^dhT}~DnSPm`!K1*}mMF8WBf?fWjkql!Mma4{f+Z$} zC|l&(ES`k(u3=Dq#-YqqLzL@Dm`)K0!&HlllG!wYw5iK*132x?C=ErC+Qx^&b0_;s zo=?M;?%r0d3>bv&5_3$08cZ)YOjC9f-`p`3>hguPb!bq^8*m+;oC-~3{CfHlV>oyP zYa^i@HWRMR0dI}$-{luR@AC+0{AuJ+&|hn0&gUwmU_?W6{0%{qTewB>Zf*-oCB8du|O6jb9Etht2Vi4uD?l406UBQsJ<)Lf7Xq9rHv=ulrOAP}CjirCqwq zU?}upwEGlj6MNiaX}&waydcq0L89VQYz zoVDJYc`$&L)O1hlT3@9u44h4+V1-vrmy20>&qMnXN_&9~(qxNlgWhWlgid!A5LS!s eJGuaXEUw}t3G5`EI2ghoPkxBSV^#cZA^bNnu^T-A delta 4794 zcma)AO>7(25r!g1wrIz;WJR**MjEeenN*~u95+ouSCt8oF`UYhEZRNTb*xG{_l_S zybj}q41VJQl&@B27v|?Hvo|W`yH}PsC89>jV6Ox zb%*V_)xZcsjvKVpy$?W3E);~)Y0`iqr*#3Gi--hfIy?+ zCL+o>7?>L#7!J#esY7}3@uBkB-CkiBczNRLHKQrM*n2M9L4ey@7PK<|uI zmS6x|=+6V#1WHd4aCfkiBkdBofoa-}Xy>_x;lDa3`i4$)Y2y9fW%2FIjmZS@B80BR zYPQQPi5!9lpzC_hFbQRyf)k~~&mNE;8i$9}hi@MKoaoC89I$JU6BC(JLy_}&genMq zS31WoWyE-YR#=(7Z1g*9QDy-_=M1#Ovu>@J5uf%I#Z>?E^Xj+uXp8a`#MU>-MT(zI zU@~QGq$n1!zp};u7PjlBw)6cE+|MSvr}XsempRIfHsSt55R~#V}&Y#pGM6=5(Qw4G3=}$yy@b#O!%R1eS5Qy21x>ser zcTvfLTF8zag_#-1ZEg{mNH_8C!RNNiLvM8*Np~}G@Z=S7`MK=&-%oz2OO@b4_Vcn4 zJc>?cpd`m!k@qXJ&X z1Psf>TX(?^(Fy4$UZiSIju7PCgmggBgZvJg77~!sf~mk7#x*Gl45qX6rK*}KhtoS6 z$Sm=f;Zxhc&ptdPem2?vWFqic4Yu-1`jM~rQZ{8yYG)owU>}|FW3f9a-ah*e(R+5J zg#paPYOAG{8z5>aMPR;P=v)b-tp&;Ho zcX4nhD-zOZq}cxG+;aE!-|`=K?}YiJ3{$)}S=vXbxBoS{&@CPf4~if^_%c-sWxx8W z-6TE&wp4;Yo(<&0VXWaOLVT0Esp%%!$ipTsiuWgn@@@0ML?ZU-x>50o4+_IAdFw2a z7iGY1pcs&Z5HABPS|Ir{a z{NloE6@;eqR5|tRu*Xp(Q=9UAcp-l=y&#X_?YbHIC_h1}jMLl14@X8@G_eBBc=DZb z-E-p#M3*I6MYyKsTnR>76ah9S;(>ap*inNF1m%_L!o1!ze9n{=?Ebh;?NB~0zlM)g=CNCjWoQjx->TQAg3{>@^%05|L5z ziS&{W5&MXY8kiMd$Tz}JeC4U0+mCTF?)v0s0#qWVkuV`#F*?ZVwr9@x7L<#@9o0IL z7H9nF*>jO4JrN<&jm`_j2l}9J^&#<+KKumwJg>i?bslU#MiGgh>Mx3SayfCN@Qhf^ zmBp39BEFvy?~Y|6{vZZ8prFE}f^0|WF+B5{*l#=QLRQYM8ptISCWlTVi)HQQp+z6* z;@ln&UHge@2k#XzIvtWZP40_>ijcLhqdQnMa$>^y0u_s>nNkgXc5b0sDW|i)e~8o< zDtq;Xbor(!5I<(cY>`?^mRfsRYX4^-MV2UDtY!88MW>j^09~8~(;=n(YC9k{Mn;cQ zjih&w@XXR;TQXs64BDZjF(nPANRP<(#&X!*E{i{%*Rb0it?ZKf9lq|LY@mou2m44aFOFRpE=KKYVrOL+(pg|; -} - -/** SSE activity update event from /api/events */ -export interface SSEActivityEvent { - type: "session.activity"; - sessionId: string; - activity: ActivityState | null; - status: SessionStatus; - attentionLevel: AttentionLevel; - timestamp: string; -} - /** * Returns true when this PR's enrichment data couldn't be fetched due to * API rate limiting. When true, CI status / review decision / mergeability diff --git a/packages/web/src/providers/MuxProvider.tsx b/packages/web/src/providers/MuxProvider.tsx index 66bb9ed57..43c77f867 100644 --- a/packages/web/src/providers/MuxProvider.tsx +++ b/packages/web/src/providers/MuxProvider.tsx @@ -11,6 +11,8 @@ interface MuxContextValue { resizeTerminal: (id: string, cols: number, rows: number) => void; status: "connecting" | "connected" | "reconnecting" | "disconnected"; sessions: SessionPatch[]; + /** Last session-fetch error from the server, null when healthy. */ + lastError: string | null; } const MuxContext = React.createContext(undefined); @@ -76,6 +78,7 @@ export function MuxProvider({ children }: { children: ReactNode }) { "connecting", ); const [sessions, setSessions] = useState([]); + const [lastError, setLastError] = useState(null); const reconnectAttempt = useRef(0); const reconnectTimer = useRef | null>(null); const runtimeConfigRef = useRef<{ directTerminalPort?: string; proxyWsPath?: string }>({}); @@ -155,8 +158,13 @@ export function MuxProvider({ children }: { children: ReactNode }) { } else if (msg.type === "error") { console.error(`[MuxProvider] Terminal error for ${msg.id}:`, msg.message); } - } else if (msg.ch === "sessions" && msg.type === "snapshot") { - setSessions(msg.sessions); + } else if (msg.ch === "sessions") { + if (msg.type === "snapshot") { + setSessions(msg.sessions); + setLastError(null); + } else if (msg.type === "error") { + setLastError(msg.error); + } } } catch (err) { console.error("[MuxProvider] Error processing message:", err); @@ -319,8 +327,9 @@ export function MuxProvider({ children }: { children: ReactNode }) { resizeTerminal, status, sessions, + lastError, }), - [subscribeTerminal, writeTerminal, openTerminal, closeTerminal, resizeTerminal, status, sessions], + [subscribeTerminal, writeTerminal, openTerminal, closeTerminal, resizeTerminal, status, sessions, lastError], ); return {children}; From 818f11f9879334f1d55cbb8a14eb3678f552896e Mon Sep 17 00:00:00 2001 From: Priyanshu Choudhary <57816400+Priyanchew@users.noreply.github.com> Date: Thu, 30 Apr 2026 20:34:30 +0530 Subject: [PATCH 3/3] fix(cli): register project command (#1576) --- packages/cli/__tests__/commands/project.test.ts | 4 ++-- packages/cli/__tests__/program.test.ts | 4 ++++ packages/cli/src/commands/project.ts | 2 +- packages/cli/src/program.ts | 2 ++ 4 files changed, 9 insertions(+), 3 deletions(-) diff --git a/packages/cli/__tests__/commands/project.test.ts b/packages/cli/__tests__/commands/project.test.ts index ebf8c2291..264b64f03 100644 --- a/packages/cli/__tests__/commands/project.test.ts +++ b/packages/cli/__tests__/commands/project.test.ts @@ -41,7 +41,7 @@ vi.mock("../../src/lib/prompts.js", () => ({ promptConfirm: vi.fn(async () => true), })); -import { registerProject_cmd } from "../../src/commands/project.js"; +import { registerProjectCommand } from "../../src/commands/project.js"; let program: Command; let logSpy: ReturnType; @@ -52,7 +52,7 @@ beforeEach(() => { vi.clearAllMocks(); program = new Command(); program.exitOverride(); - registerProject_cmd(program); + registerProjectCommand(program); logSpy = vi.spyOn(console, "log").mockImplementation(() => {}); errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); _exitSpy = vi.spyOn(process, "exit").mockImplementation(((code?: number) => { diff --git a/packages/cli/__tests__/program.test.ts b/packages/cli/__tests__/program.test.ts index 656ac51be..e854a2b03 100644 --- a/packages/cli/__tests__/program.test.ts +++ b/packages/cli/__tests__/program.test.ts @@ -6,4 +6,8 @@ describe("createProgram", () => { it("uses the CLI package version", () => { expect(createProgram().version()).toBe(packageJson.version); }); + + it("registers the project command", () => { + expect(createProgram().commands.some((command) => command.name() === "project")).toBe(true); + }); }); diff --git a/packages/cli/src/commands/project.ts b/packages/cli/src/commands/project.ts index b052fa5ac..5dfc35c21 100644 --- a/packages/cli/src/commands/project.ts +++ b/packages/cli/src/commands/project.ts @@ -29,7 +29,7 @@ function assertPortfolioEnabled(): void { process.exit(1); } -export function registerProject_cmd(program: Command): void { +export function registerProjectCommand(program: Command): void { const project = program.command("project").description("Manage portfolio projects"); // ao project ls diff --git a/packages/cli/src/program.ts b/packages/cli/src/program.ts index 0680a8971..f5110afb9 100644 --- a/packages/cli/src/program.ts +++ b/packages/cli/src/program.ts @@ -14,6 +14,7 @@ import { registerDoctor } from "./commands/doctor.js"; import { registerUpdate } from "./commands/update.js"; import { registerSetup } from "./commands/setup.js"; import { registerPlugin } from "./commands/plugin.js"; +import { registerProjectCommand } from "./commands/project.js"; import { registerMigrateStorage } from "./commands/migrate-storage.js"; import { registerCompletion } from "./commands/completion.js"; import { getConfigInstruction } from "./lib/config-instruction.js"; @@ -45,6 +46,7 @@ export function createProgram(): Command { registerUpdate(program); registerSetup(program); registerPlugin(program); + registerProjectCommand(program); registerMigrateStorage(program); registerCompletion(program);