import { selectSessionMessages, selectSessionMessageWindow, type AgentActivityMessagePage } from "@tutti-os/agent-activity-core"; import { describe, expect, it, vi } from "vitest"; import { createTestAgentSessionEngine, type TestEngineCommandHandler } from "./shared/testing/createTestAgentSessionEngine"; import { createAgentConversationMessageController, type AgentConversationMessagePageInput } from "./agentConversationMessageController"; describe("createAgentConversationMessageController", () => { it("routes initial and latest hydration through Engine reconcile commands", async () => { const execute = vi.fn(async () => ({ affectedSessionIds: [], appliedMessages: [], session: null, status: "applied" })); const engine = createTestAgentSessionEngine("workspace-1", { execute }); const controller = createController({ engine }); controller.setActiveSession("session-1"); controller.requestInitial("session-1"); controller.requestLatest("session-1"); await vi.waitFor(() => { expect( execute.mock.calls.filter( ([command]) => command.type === "session/reconcile" ) ).toHaveLength(2); }); const reconcileCommands = execute.mock.calls .map(([command]) => command) .filter((command) => command.type === "session/reconcile"); expect(reconcileCommands).toEqual([ expect.objectContaining({ agentSessionId: "session-1", scope: "state_and_messages" }), expect.objectContaining({ agentSessionId: "session-1", scope: "messages" }) ]); controller.dispose(); engine.dispose(); }); it("treats repeated initial hydration as idempotent while reconcile is pending", async () => { type CommandResult = Awaited< ReturnType >; let resolveExecute: (result: CommandResult) => void = () => {}; const execute = vi.fn( () => new Promise((resolve) => { resolveExecute = resolve; }) ); const engine = createTestAgentSessionEngine("workspace-1", { execute }); const controller = createController({ engine }); controller.requestInitial("session-1"); controller.requestInitial("session-1"); await vi.waitFor(() => { expect( execute.mock.calls.filter( ([command]) => command.type === "session/reconcile" ) ).toHaveLength(1); }); resolveExecute({ affectedSessionIds: [], appliedMessages: [], session: null, status: "applied" }); controller.dispose(); engine.dispose(); }); it("loads older history only from the authoritative Engine window", async () => { const engine = createTestAgentSessionEngine("workspace-1"); const listSessionMessages = vi.fn(async () => messagePage([message(1)], false) ); const controller = createController({ engine, listSessionMessages }); controller.setActiveSession("session-1"); engine.dispatch({ messages: [message(446)], type: "message/snapshotReceived", workspaceId: "workspace-1" }); await controller.loadOlder(); expect(listSessionMessages).not.toHaveBeenCalled(); engine.dispatch({ messages: [], sessionMessageWindows: [ { agentSessionId: "session-1", hasOlderMessages: true, oldestLoadedVersion: 446 } ], type: "message/snapshotReceived", workspaceId: "workspace-1" }); await controller.loadOlder(); expect(listSessionMessages).toHaveBeenCalledWith({ agentSessionId: "session-1", beforeVersion: 446, limit: 100, order: "desc", signal: expect.any(AbortSignal), workspaceId: "workspace-1" }); expect( selectSessionMessages(engine.getSnapshot(), "session-1").map( (item) => item.version ) ).toEqual([1, 446]); expect( selectSessionMessageWindow(engine.getSnapshot(), "session-1") ).toEqual({ hasOlderMessages: false, oldestLoadedVersion: 1 }); controller.dispose(); engine.dispose(); }); it("deduplicates an in-flight cursor and permits retry after failure", async () => { const engine = createTestAgentSessionEngine("workspace-1"); let rejectFirst: ((error: unknown) => void) | null = null; const listSessionMessages = vi .fn() .mockImplementationOnce( () => new Promise((_resolve, reject) => { rejectFirst = reject; }) ) .mockResolvedValueOnce(messagePage([message(1)], false)); const controller = createController({ engine, listSessionMessages }); controller.setActiveSession("session-1"); setOlderWindow(engine, 5); const first = controller.loadOlder(); const duplicate = controller.loadOlder(); expect(listSessionMessages).toHaveBeenCalledTimes(1); rejectFirst!(new Error("temporary failure")); await first; await duplicate; expect(controller.getSnapshot().olderPagePhase).toBe("error"); await controller.loadOlder(); expect(listSessionMessages).toHaveBeenCalledTimes(2); expect(controller.getSnapshot().olderPagePhase).toBe("idle"); controller.dispose(); engine.dispose(); }); it("fences an older page after the active Session changes", async () => { const engine = createTestAgentSessionEngine("workspace-1"); let resolvePage: ((page: AgentActivityMessagePage) => void) | null = null; let requestSignal!: AbortSignal; const listSessionMessages = vi.fn( (input: AgentConversationMessagePageInput) => new Promise((resolve) => { requestSignal = input.signal; resolvePage = resolve; }) ); const controller = createController({ engine, listSessionMessages }); controller.setActiveSession("session-1"); setOlderWindow(engine, 5); const request = controller.loadOlder(); controller.setActiveSession("session-2"); expect(requestSignal.aborted).toBe(true); resolvePage!(messagePage([message(1)], false)); await request; expect(selectSessionMessages(engine.getSnapshot(), "session-1")).toEqual( [] ); expect(controller.getSnapshot()).toEqual({ agentSessionId: "session-2", error: null, olderPagePhase: "idle" }); controller.dispose(); engine.dispose(); }); it("retains synchronization only for the active Session", () => { const engine = createTestAgentSessionEngine("workspace-1"); const releases: string[] = []; const ensureSessionSynchronized = vi.fn( ({ agentSessionId }: { agentSessionId: string }) => () => { releases.push(agentSessionId); } ); const controller = createController({ engine, ensureSessionSynchronized }); controller.setActiveSession("session-1"); controller.setActiveSession("session-1"); controller.setActiveSession("session-2"); controller.setActiveSession(null); controller.setActiveSession("session-3"); controller.dispose(); expect(ensureSessionSynchronized.mock.calls).toEqual([ [ expect.objectContaining({ agentSessionId: "session-1", workspaceId: "workspace-1" }) ], [ expect.objectContaining({ agentSessionId: "session-2", workspaceId: "workspace-1" }) ], [ expect.objectContaining({ agentSessionId: "session-3", workspaceId: "workspace-1" }) ] ]); expect(releases).toEqual(["session-1", "session-2", "session-3"]); engine.dispose(); }); it("rejects a stale page when the host focus changes before synchronization", async () => { const engine = createTestAgentSessionEngine("workspace-1"); let focusedSessionId = "session-1"; let resolvePage: ((page: AgentActivityMessagePage) => void) | null = null; const controller = createController({ engine, isAvailable: (agentSessionId) => !agentSessionId || agentSessionId === focusedSessionId, listSessionMessages: () => new Promise((resolve) => { resolvePage = resolve; }) }); controller.setActiveSession("session-1"); setOlderWindow(engine, 5); const request = controller.loadOlder(); focusedSessionId = "session-2"; resolvePage!(messagePage([message(1)], false)); await request; expect(selectSessionMessages(engine.getSnapshot(), "session-1")).toEqual( [] ); controller.dispose(); engine.dispose(); }); }); function createController(input: { engine: ReturnType; ensureSessionSynchronized?: Parameters< typeof createAgentConversationMessageController >[0]["ensureSessionSynchronized"]; isAvailable?: (agentSessionId?: string | null) => boolean; listSessionMessages?: ( input: AgentConversationMessagePageInput ) => Promise; }) { return createAgentConversationMessageController({ engine: input.engine, ensureSessionSynchronized: input.ensureSessionSynchronized, isAvailable: input.isAvailable ?? (() => true), listSessionMessages: input.listSessionMessages ?? (async () => ({ hasMore: false, latestVersion: 0, messages: [] })), workspaceId: "workspace-1" }); } function setOlderWindow( engine: ReturnType, oldestLoadedVersion: number ): void { engine.dispatch({ messages: [], sessionMessageWindows: [ { agentSessionId: "session-1", hasOlderMessages: true, oldestLoadedVersion } ], type: "message/snapshotReceived", workspaceId: "workspace-1" }); } function message(version: number) { return { agentSessionId: "session-1", kind: "text", messageId: `message-${version}`, occurredAtUnixMs: version, payload: {}, role: "assistant", sequence: version, turnId: "turn-1", version }; } function messagePage( messages: ReturnType[], hasMore: boolean ): AgentActivityMessagePage { return { hasMore, latestVersion: messages.at(-1)?.version ?? 0, messages }; }