next-tjwater-frontend/src/features/workbench/hooks/use-workbench-communication.test.tsx

import { act, cleanup, renderHook, waitFor } from "@testing-library/react";
import { afterEach, expect, it, vi } from "vitest";
import { SWRConfig } from "swr";
const harness = vi.hoisted(() => ({
  api: null as any,
  recovery: 0,
  flow: vi.fn(),
  task: vi.fn(),
  check: vi.fn()
}));
vi.mock("@/features/agent", async (original) => ({
  ...(await original<any>()),
  createAgentApiClient: () => harness.api
}));
vi.mock("@/features/agent/hooks/use-agent-connection", () => ({
  useAgentConnection: () => ({
    connectionStatus: "online",
    recoveryVersion: harness.recovery,
    recheck: harness.check
  })
}));
vi.mock("@/features/workbench/api/realtime-flow-client", () => ({
  fetchRealtimePipeFlows: harness.flow
}));
vi.mock("@/features/workbench/api/operational-task-client", async (original) => ({
  ...(await original<any>()),
  fetchOperationalTask: harness.task,
  fetchOperationalTaskNotifications: async () => []
}));
import { useWorkbenchRuntimeData } from "@/features/workbench/hooks/use-workbench-runtime-data";
import { useWorkbenchAgent } from "@/features/workbench/hooks/use-workbench-agent";
import { useOperationalTasks } from "@/features/workbench/hooks/use-operational-tasks";
import { WorkbenchSessionProvider } from "@/features/workbench/hooks/workbench-session-provider";
import { useAgentMessageInbox } from "@/features/agent/hooks/use-agent-message-inbox";
const access: any = {
  project: { project_id: "p", gs_workspace: "tjwater_next", status: "active" },
  getAccessToken: async () => "test"
};
const sync: any = {
  snapshot: null,
  events: [],
  status: "live",
  error: null,
  detailResetGeneration: 0,
  acknowledgeEventsThrough: vi.fn(),
  refresh: vi.fn(),
  subscribePresentationEvents: () => () => {}
};
afterEach(() => {
  cleanup();
  vi.useRealTimers();
  vi.restoreAllMocks();
  harness.flow.mockReset();
  harness.task.mockReset();
  harness.recovery = 0;
});
function wrapperFactory() {
  const cache = new Map();
  return ({ children }: any) => (
    <WorkbenchSessionProvider owner={access}>
      <SWRConfig
        value={{ provider: () => cache, revalidateOnFocus: false, shouldRetryOnError: false }}
      >
        {children}
      </SWRConfig>
    </WorkbenchSessionProvider>
  );
}
it("keeps one flow request and one polling timer when visibility changes during a request", async () => {
  vi.useFakeTimers();
  let visibility = "visible";
  vi.spyOn(document, "visibilityState", "get").mockImplementation(() => visibility as any);
  const done: Array<() => void> = [];
  harness.flow.mockImplementation(
    () => new Promise((resolve) => done.push(() => resolve({ records: [], observedAt: null })))
  );
  renderHook(() => useWorkbenchRuntimeData(access, sync));
  expect(harness.flow).toHaveBeenCalledTimes(1);
  await act(async () => {
    visibility = "hidden";
    document.dispatchEvent(new Event("visibilitychange"));
    visibility = "visible";
    document.dispatchEvent(new Event("visibilitychange"));
  });
  expect(harness.flow).toHaveBeenCalledTimes(1);
  await act(async () => {
    done[0]();
  });
  await act(async () => {
    await vi.advanceTimersByTimeAsync(15000);
  });
  expect(harness.flow).toHaveBeenCalledTimes(2);
  await act(async () => {
    done[1]();
    await vi.advanceTimersByTimeAsync(15000);
  });
  expect(harness.flow).toHaveBeenCalledTimes(3);
});
it("queues new task events without cancelling or repeating the current detail request", async () => {
  const resolve: Array<(value: unknown) => void> = [];
  harness.task.mockImplementation(() => new Promise((done) => resolve.push(done)));
  const event = (seq: number, id: string) => ({
    event_seq: seq,
    topic: "operational.progress.changed",
    aggregate_type: "operational_task",
    aggregate_id: id,
    metadata: { task_id: id }
  });
  const first = event(1, "task-a");
  const { result, rerender, unmount } = renderHook(
    ({ events }) => useOperationalTasks(access, { ...sync, events }),
    { initialProps: { events: [first] } }
  );
  rerender({ events: [first, event(2, "task-b")] });
  expect(harness.task.mock.calls.map((call) => call[0])).toEqual(["task-a"]);
  expect(harness.task.mock.calls[0][2].aborted).toBe(false);
  await act(async () => {
    resolve[0]({
      task_id: "task-a",
      title: "A",
      trigger: { kind: "manual" },
      source: { kind: "operator" },
      task_type: "event_investigation",
      evidence: [],
      updated_at: "2026-11-09T00:00:00Z"
    });
  });
  expect(harness.task.mock.calls.map((call) => call[0])).toEqual(["task-a", "task-b"]);
  await act(async () => {
    resolve[1]({
      task_id: "task-b",
      title: "B",
      trigger: { kind: "manual" },
      source: { kind: "operator" },
      task_type: "event_investigation",
      evidence: [],
      updated_at: "2026-11-09T00:00:00Z"
    });
  });
  expect(result.current.tasks.map((task) => task.task_id).sort()).toEqual(["task-a", "task-b"]);
  expect(sync.acknowledgeEventsThrough).toHaveBeenLastCalledWith(2);
  unmount();
});
it("refreshes once per health recovery and uses normal loads on session changes", async () => {
  const api: any = {
    getModels: vi.fn(async () => ({ models: [], defaultModel: "" })),
    getUiRegistry: vi.fn(async () => ({})),
    getFrontendActionRegistry: vi.fn(async () => ({})),
    listSessionPage: vi.fn(async () => ({ sessions: [], hasMore: false, nextCursor: null })),
    listTopics: vi.fn(async () => []),
    listInteractionRequests: vi.fn(async () => ({ items: [] })),
    listResponseInputs: vi.fn(async () => []),
    listAgentMessages: vi.fn(async () => ({ items: [], next_cursor: null })),
    streamAgentMessages: vi.fn(
      (_id: any, options: any) =>
        new Promise<void>((resolve) =>
          options.signal.addEventListener("abort", () => resolve(), { once: true })
        )
    ),
    loadSession: vi.fn(async (id: string) => ({
      id,
      sessionId: id,
      title: id,
      purpose: "ordinary_chat",
      messages: [],
      createdAt: 1,
      updatedAt: 1,
      runStatus: "completed",
      isStreaming: false,
      isTitleManuallyEdited: false
    }))
  };
  harness.api = api;
  const { result, rerender } = renderHook(
    () =>
      useWorkbenchAgent({
        getAccessToken: access,
        onUiEnvelope: () => {},
        onFrontendAction: async () => ({})
      }),
    { wrapper: wrapperFactory() }
  );
  await waitFor(() => expect(api.getModels).toHaveBeenCalledTimes(1));
  await act(async () => {
    await result.current.loadHistorySession("a");
  });
  await waitFor(() =>
    expect(api.listInteractionRequests.mock.calls.filter((c: any) => c[0] === "a").length).toBe(1)
  );
  harness.recovery = 1;
  rerender();
  await waitFor(() => expect(api.getModels).toHaveBeenCalledTimes(2));
  const before = api.getModels.mock.calls.length;
  await act(async () => {
    await result.current.loadHistorySession("b");
  });
  await waitFor(() => expect(api.getModels.mock.calls.length).toBe(before));
  await waitFor(() =>
    expect(api.listInteractionRequests.mock.calls.filter((c: any) => c[0] === "b").length).toBe(1)
  );
  const pending = api.listResponseInputs.mock.calls.filter((c: any) => c[0] === "b").length;
  expect(pending).toBe(1);
});
it("loads the inbox once when the stream is ready and refreshes on later notifications", async () => {
  const list = vi.fn(async () => ({ items: [], next_cursor: null }));
  const stream = vi.fn((_id: any, options: any) => {
    options.onMessage();
    return new Promise<void>((resolve) =>
      options.signal.addEventListener("abort", () => resolve(), { once: true })
    );
  });
  const api = { listAgentMessages: list, streamAgentMessages: stream } as any;
  renderHook(() => useAgentMessageInbox(api, "s"), { wrapper: wrapperFactory() });
  await waitFor(() => expect(list).toHaveBeenCalledTimes(1));
  await act(async () => {
    stream.mock.calls[0][1].onMessage();
  });
  await waitFor(() => expect(list).toHaveBeenCalledTimes(2));
});

it("keeps newer task changes and does not revive a task deleted during a read", async () => {
  const complete: Array<(value: unknown) => void> = [];
  harness.task.mockImplementation(() => new Promise((resolve) => complete.push(resolve)));
  const event = (seq: number, deleted = false) => ({
    event_seq: seq,
    topic: "operational.task.changed",
    aggregate_type: "operational_task",
    aggregate_id: "a",
    metadata: { task_id: "a", ...(deleted ? { change: "deleted" } : {}) }
  });
  const a = event(1);
  const { result, rerender } = renderHook(
    ({ events }) => useOperationalTasks(access, { ...sync, events }),
    {
      initialProps: { events: [a] }
    }
  );
  rerender({ events: [a, event(2)] });
  const task = {
    task_id: "a",
    title: "A",
    trigger: { kind: "manual" },
    source: { kind: "operator" },
    task_type: "event_investigation",
    evidence: [],
    updated_at: "2026-11-09T00:00:00Z"
  };
  await act(async () => {
    complete[0](task);
  });
  expect(harness.task).toHaveBeenCalledTimes(2);
  rerender({ events: [a, event(2), event(3, true)] });
  expect(result.current.tasks).toEqual([]);
  await act(async () => {
    complete[1](task);
  });
  await waitFor(() => expect(sync.acknowledgeEventsThrough).toHaveBeenLastCalledWith(3));
  expect(result.current.tasks).toEqual([]);
  expect(harness.task).toHaveBeenCalledTimes(2);
});

it("loads the initial inbox even when its stream cannot connect", async () => {
  const list = vi.fn(async () => ({ items: [], next_cursor: null }));
  const api = {
    listAgentMessages: list,
    streamAgentMessages: vi.fn(async () => {
      throw new Error("offline");
    })
  } as any;
  const { result } = renderHook(() => useAgentMessageInbox(api, "s"), {
    wrapper: wrapperFactory()
  });
  await waitFor(() => expect(result.current.loading).toBe(false));
  expect(list).toHaveBeenCalledTimes(1);
});

it("loads the inbox when its stream remains pending without a first frame", async () => {
  vi.useFakeTimers();
  const list = vi.fn(async () => ({ items: [], next_cursor: null }));
  const api = {
    listAgentMessages: list,
    streamAgentMessages: vi.fn(
      (_id: string, options: { signal: AbortSignal }) =>
        new Promise<void>((resolve) => {
          options.signal.addEventListener("abort", () => resolve(), { once: true });
        })
    )
  } as any;
  const { result } = renderHook(() => useAgentMessageInbox(api, "s"), {
    wrapper: wrapperFactory()
  });
  expect(list).not.toHaveBeenCalled();
  await act(async () => {
    await vi.advanceTimersByTimeAsync(1500);
  });
  expect(list).toHaveBeenCalledTimes(1);
  expect(result.current.loading).toBe(false);
});