diff --git a/src/app/root/E2EBootstrap.tsx b/src/app/root/E2EBootstrap.tsx index f279556b7..0d6a788fb 100644 --- a/src/app/root/E2EBootstrap.tsx +++ b/src/app/root/E2EBootstrap.tsx @@ -222,6 +222,7 @@ export const E2EBootstrap: FC = () => { inspectOrgtrackFileSessionHistory, inspectCliSessionStatus, inspectCliHistoryMutation, + inspectCliHistory, resetToNewSession, openSession, reloadSessionList, @@ -436,6 +437,7 @@ export const E2EBootstrap: FC = () => { inspectOrgtrackFileSessionHistory, inspectCliSessionStatus, inspectCliHistoryMutation, + inspectCliHistory, resetToNewSession, openSession, reloadSessionList, diff --git a/src/app/root/e2e/helpers/sessions.ts b/src/app/root/e2e/helpers/sessions.ts index 2fed1c21f..cad9a121a 100644 --- a/src/app/root/e2e/helpers/sessions.ts +++ b/src/app/root/e2e/helpers/sessions.ts @@ -271,6 +271,26 @@ export function createSessionHelpers(store: E2EStore) { } }; + const inspectCliHistory = async ( + sessionId: string + ): Promise> => { + try { + if (!sessionId) { + return { + ok: false, + error: "inspectCliHistory: `sessionId` is required", + }; + } + const events = await cliAdapter.loadHistory( + sessionId, + new AbortController().signal + ); + return { ok: true, events: events as unknown as Json[] }; + } catch (err) { + return asError(err); + } + }; + const resetToNewSession = async (): Promise<{ ok: true } | Result> => { try { store.set(clearSessionAtom); @@ -830,6 +850,7 @@ export function createSessionHelpers(store: E2EStore) { inspectOrgtrackFileSessionHistory, inspectCliSessionStatus, inspectCliHistoryMutation, + inspectCliHistory, resetToNewSession, openSession, reloadSessionList, diff --git a/src/app/root/e2e/types.ts b/src/app/root/e2e/types.ts index cca9c57a3..79de3acc5 100644 --- a/src/app/root/e2e/types.ts +++ b/src/app/root/e2e/types.ts @@ -485,6 +485,7 @@ export interface E2EHelpers { inspectCliHistoryMutation: ( sessionId: string ) => Promise>; + inspectCliHistory: (sessionId: string) => Promise>; resetToNewSession: () => Promise<{ ok: true } | Err>; openSession: (sessionId: string) => Promise>; debugSessionSecuritySnapshot: ( diff --git a/src/engines/SessionCore/services/TurnDispatchService.test.ts b/src/engines/SessionCore/services/TurnDispatchService.test.ts index e913c4ea4..13f704461 100644 --- a/src/engines/SessionCore/services/TurnDispatchService.test.ts +++ b/src/engines/SessionCore/services/TurnDispatchService.test.ts @@ -51,6 +51,7 @@ vi.mock("./SessionService", () => ({ })); const SESSION = "sdeagent-session-1"; +const CLI_SESSION = "cliagent-session-1"; describe("TurnDispatchService", () => { beforeEach(() => { @@ -70,6 +71,7 @@ describe("TurnDispatchService", () => { afterEach(() => { resetTurnDispatchMonitorsForTests(); clearRecentOptimisticTurn(SESSION); + clearRecentOptimisticTurn(CLI_SESSION); clearRecentOptimisticTurn("cursoride-session-1"); resetTurnLifecycleForTests(); }); @@ -469,6 +471,45 @@ describe("TurnDispatchService", () => { } }); + it("polls accepted CLI finality when no live status channel is available", async () => { + vi.useFakeTimers(); + try { + mocks.sendMessage.mockResolvedValueOnce({ + duplicate: false, + turnIntentStatus: "running", + effectiveTurnIntentId: "intent-cli-background", + }); + mocks.getTurnIntentStatus.mockResolvedValueOnce({ + status: "completed", + effectiveTurnIntentId: "intent-cli-background", + }); + const dispatch = reserveTurnDispatch({ + sessionId: CLI_SESSION, + turnIntentId: "intent-cli-background", + }); + + await sendReservedTurn({ + dispatch, + content: "background CLI turn", + turnIntentSource: "user_submit", + }); + const outcome = waitForTurnOutcome(dispatch, Date.now() + 1_000); + expect(vi.getTimerCount()).toBe(2); + + await vi.advanceTimersByTimeAsync(100); + + await expect(outcome).resolves.toMatchObject({ status: "completed" }); + expect(mocks.getTurnIntentStatus).toHaveBeenCalledWith( + CLI_SESSION, + "intent-cli-background" + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + resetTurnDispatchMonitorsForTests(); + vi.useRealTimers(); + } + }); + it("stops an ambiguous exact-X monitor on a live terminal", async () => { vi.useFakeTimers(); try { diff --git a/src/engines/SessionCore/services/TurnDispatchService.ts b/src/engines/SessionCore/services/TurnDispatchService.ts index 012d60bbf..7670348c7 100644 --- a/src/engines/SessionCore/services/TurnDispatchService.ts +++ b/src/engines/SessionCore/services/TurnDispatchService.ts @@ -32,7 +32,10 @@ import { setSessionRuntimeStatusAtom, } from "@src/store/session/cliSessionStatusAtom"; import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; -import { isCursorIdeSession } from "@src/util/session/sessionDispatch"; +import { + isCliSession, + isCursorIdeSession, +} from "@src/util/session/sessionDispatch"; import { SessionService } from "./SessionService"; import type { SessionSendMessageParams } from "./types"; @@ -541,6 +544,13 @@ export async function sendReservedTurn( confirmTurnRunning(dispatch.sessionId, { generation: dispatch.generation }); markSessionActive(dispatch.sessionId); + if (isCliSession(dispatch.sessionId)) { + // CLI live status normally arrives over the window-level WebSocket, but + // hidden/background runners and isolated app instances may not have that + // channel. Reuse the canonical exact-intent durable monitor so provider + // finality never depends on a mounted transcript or an active socket. + startEffectiveTurnStatusMonitor(dispatch, effectiveTurnIntentId); + } if (isCursorIdeSession(dispatch.sessionId)) { if (getTurnGeneration(dispatch.sessionId) !== dispatch.generation) { return { ...dispatch, accepted: true }; diff --git a/src/engines/SessionCore/sync/authoritativeSessionEvents.ts b/src/engines/SessionCore/sync/authoritativeSessionEvents.ts new file mode 100644 index 000000000..f065808c7 --- /dev/null +++ b/src/engines/SessionCore/sync/authoritativeSessionEvents.ts @@ -0,0 +1,56 @@ +/** + * Canonical full-history read for a managed local session. + * + * Most runtimes persist normalized events in EventStore. External CLI + * sessions can instead keep their transcript exclusively in the provider's + * native store, so an empty EventStore is not proof of an empty transcript. + * Keep that distinction here so cloud sync, background continuation, and + * future consumers cannot accidentally publish a hollow CLI session. + */ +import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import { isCliSession } from "@src/util/session/sessionDispatch"; + +import { loadCliHistory } from "./adapters/cli/cliHistory"; + +export interface AuthoritativeSessionEvents { + events: SessionEvent[]; + /** Stable EventStore revision when EventStore was the authoritative source. */ + localContentRevision?: number; + source: "event_store" | "cli_history"; +} + +export async function loadAuthoritativeSessionEvents( + sessionId: string, + signal: AbortSignal = new AbortController().signal +): Promise { + // CLI adapters own both of their durable transcript modes: legacy chunks + // and provider-native stores. EventStore can contain only an optimistic + // user row while the native transcript already contains the completed + // assistant tail, so "persisted is non-empty" is not an authority test. + if (isCliSession(sessionId)) { + return { + events: await loadCliHistory(sessionId, signal), + source: "cli_history", + }; + } + + const revisionBefore = + await eventStoreProxy.getPersistedEventRevision(sessionId); + const persisted = await eventStoreProxy.getPersistedEvents(sessionId); + const revisionAfter = + await eventStoreProxy.getPersistedEventRevision(sessionId); + const localContentRevision = + revisionBefore && + revisionAfter && + revisionBefore.revision === revisionAfter.revision && + revisionAfter.eventCount === persisted.length + ? revisionAfter.revision + : undefined; + + return { + events: persisted, + localContentRevision, + source: "event_store", + }; +} diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.test.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.test.ts new file mode 100644 index 000000000..4908bab69 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, it } from "vitest"; + +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; + +import { sliceAppendedTurnTail } from "./conversationTurnEvents"; + +function event( + id: string, + source: "user" | "assistant" | "system" +): SessionEvent { + return { + id, + chunk_id: id, + sessionId: "cliagent-session", + source, + } as SessionEvent; +} + +describe("sliceAppendedTurnTail", () => { + it("uses a stable native-transcript prefix as the turn boundary", () => { + const before = [event("user-old", "user"), event("reply-old", "assistant")]; + const after = [ + ...before, + event("user-new", "user"), + event("tool-new", "system"), + event("reply-new", "assistant"), + ]; + + expect( + sliceAppendedTurnTail(before, after)?.map((item) => item.id) + ).toEqual(["tool-new", "reply-new"]); + }); + + it("fails closed when the provider rewrites the previous prefix", () => { + const before = [event("user-old", "user"), event("reply-old", "assistant")]; + const after = [ + event("user-old", "user"), + event("reply-rewritten", "assistant"), + event("user-new", "user"), + event("reply-new", "assistant"), + ]; + + expect(sliceAppendedTurnTail(before, after)).toBeNull(); + }); + + it("requires an appended user boundary", () => { + expect( + sliceAppendedTurnTail([], [event("assistant-only", "assistant")]) + ).toBeNull(); + }); +}); diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts index 4a46b9588..e7dd00f3d 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts @@ -69,3 +69,43 @@ export function sliceTurnTailByIntent( } return tail; } + +/** + * Slice one newly appended native-transcript turn from two authoritative + * snapshots. External CLIs own their transcript schema and therefore cannot + * persist ORG2's internal turnIntentId. Their stable normalized event ids are + * the boundary instead: the old snapshot must remain an exact prefix, then + * the first appended user row anchors the new agent tail. + * + * `null` fails closed when history was rewritten or no user boundary exists; + * publishing from an ambiguous offset could duplicate an older agent turn. + */ +export function sliceAppendedTurnTail( + before: readonly SessionEvent[], + after: readonly SessionEvent[] +): SessionEvent[] | null { + if (after.length < before.length) return null; + for (let index = 0; index < before.length; index += 1) { + const previous = before[index]; + const current = after[index]; + if ( + previous.id !== current.id || + previous.chunk_id !== current.chunk_id || + previous.source !== current.source + ) { + return null; + } + } + + const appended = after.slice(before.length); + const userIndex = appended.findIndex((event) => event.source === "user"); + if (userIndex < 0) return null; + + const tail: SessionEvent[] = []; + for (let index = userIndex + 1; index < appended.length; index += 1) { + const event = appended[index]; + if (event.source === "user") break; + tail.push(event); + } + return tail; +} diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts index ada32a751..3e47c64b9 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts @@ -4,6 +4,10 @@ import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import { SessionService } from "@src/engines/SessionCore/services/SessionService"; import { sendReservedTurn } from "@src/engines/SessionCore/services/TurnDispatchService"; import { requestForkSessionSetup } from "@src/features/TeamCollaboration/forkSession"; +import { + clearForkSetupMemory, + loadForkSetupMemory, +} from "@src/features/TeamCollaboration/forkSetupMemory"; import type { CloudConversationEvent } from "../org2CloudConversationEventsClient"; import { loadContinuation, saveContinuation } from "./conversationContinuation"; @@ -19,6 +23,8 @@ const { state } = vi.hoisted(() => ({ generation: 0, persistedBatches: [] as SessionEvent[][], pushes: [] as Array<{ kind: "user" | "tail"; turnId: string }>, + pushLastSeqs: [] as number[], + tailEventIds: [] as string[][], sent: [] as Record[], rejectNextSend: false, cleaned: [] as string[], @@ -29,10 +35,11 @@ const { state } = vi.hoisted(() => ({ vi.mock("@src/components/Message", () => ({ default: { info: vi.fn(), error: vi.fn() }, })); -vi.mock("@src/engines/SessionCore/core/store/EventStoreProxy", () => ({ - eventStoreProxy: { - getPersistedEvents: vi.fn(async () => state.persistedBatches.shift() ?? []), - }, +vi.mock("@src/engines/SessionCore/sync/authoritativeSessionEvents", () => ({ + loadAuthoritativeSessionEvents: vi.fn(async () => ({ + events: state.persistedBatches.shift() ?? [], + source: "event_store", + })), })); vi.mock("@src/engines/SessionCore/services/SessionService", () => ({ SessionService: { @@ -98,13 +105,19 @@ vi.mock("../org2CloudConversationEventsClient", () => ({ pushConversationEvents: vi.fn( async (_token: string, input: { turnId: string }) => { state.pushes.push({ kind: "user", turnId: input.turnId }); - return { firstSeq: 1, lastSeq: 1 }; + const lastSeq = state.pushLastSeqs.shift() ?? 1; + return { firstSeq: lastSeq, lastSeq }; } ), pushConversationEventsChunked: vi.fn( - async (_token: string, input: { turnId: string }) => { + async ( + _token: string, + input: { turnId: string; events: SessionEvent[] } + ) => { state.pushes.push({ kind: "tail", turnId: input.turnId }); - return { firstSeq: 2, lastSeq: 2 }; + state.tailEventIds.push(input.events.map((event) => event.id)); + const lastSeq = state.pushLastSeqs.shift() ?? 2; + return { firstSeq: lastSeq, lastSeq }; } ), })); @@ -129,7 +142,7 @@ function fakeStorage(): Storage { function event( id: string, - source: "user" | "assistant", + source: "user" | "assistant" | "system", text: string, turnIntentId?: string ): SessionEvent { @@ -188,6 +201,8 @@ beforeEach(() => { state.generation = 0; state.persistedBatches = []; state.pushes = []; + state.pushLastSeqs = []; + state.tailEventIds = []; state.sent = []; state.rejectNextSend = false; state.cleaned = []; @@ -305,6 +320,7 @@ describe("conversation turn continuation", () => { ], lastSeq: 57, })); + state.pushLastSeqs = [58, 59]; const result = await runConversationTurn( params({ @@ -320,7 +336,7 @@ describe("conversation turn continuation", () => { expect(loadPlaneDelta).toHaveBeenCalledWith(55); expect(state.sent[0].content).toContain("Alice: note from Alice"); expect(state.sent[0].content).not.toContain("duplicate local turn"); - expect(loadContinuation("scope", "root")?.readThroughPlaneSeq).toBe(57); + expect(loadContinuation("scope", "root")?.readThroughPlaneSeq).toBe(59); }); it("rolls a rejected resume to fresh without publishing the user twice", async () => { @@ -469,11 +485,13 @@ describe("conversation turn continuation", () => { execution: { agentDefinitionId: "agent-a", cliAgentType: "codex", + accountId: "codex-account", }, }); state.persistedBatches = [ + [], [ - event("user-1", "user", "new request", "intent-1"), + event("native-user-1", "user", "new request"), event("agent-1", "assistant", "answer"), ], ]; @@ -491,14 +509,112 @@ describe("conversation turn continuation", () => { expect.objectContaining({ task: "", cliAgentType: "codex", + accountId: "codex-account", agentDefinitionId: "agent-a", }) ); expect(loadContinuation("scope", "root")).toMatchObject({ continuationSessionId: "fresh-runner", cliAgentType: "codex", + accountId: "codex-account", agentDefinitionId: "agent-a", }); + expect(state.tailEventIds).toEqual([["agent-1"]]); + }); + + it("resumes a native-transcript CLI by authoritative snapshot delta", async () => { + saveContinuation("scope", "root", { + continuationSessionId: "cliagent-runner-live", + readThroughPlaneSeq: 20, + established: true, + agentDefinitionId: "agent-a", + cliAgentType: "codex", + accountId: "codex-account", + }); + const previous = [ + event("native-user-1", "user", "first request"), + event("native-agent-1", "assistant", "first answer"), + ]; + state.persistedBatches = [ + previous, + [ + ...previous, + event("native-user-2", "user", "second request"), + event("native-tool-2", "system", "tool output"), + event("native-agent-2", "assistant", "second answer"), + ], + ]; + state.pushLastSeqs = [21, 24]; + + const result = await runConversationTurn( + params({ + turnIntentId: "intent-2", + loadPlaneDelta: async () => ({ events: [], lastSeq: 20 }), + }) + ); + + expect(result.runnerSessionId).toBe("cliagent-runner-live"); + expect(SessionService.create).not.toHaveBeenCalled(); + expect(state.tailEventIds).toEqual([["native-tool-2", "native-agent-2"]]); + expect(loadContinuation("scope", "root")?.readThroughPlaneSeq).toBe(24); + }); + + it("rereads a completed CLI transcript until its agent tail is visible", async () => { + vi.mocked(requestForkSessionSetup).mockResolvedValueOnce({ + workspaceRepoPath: "/repo", + execution: { + agentDefinitionId: "agent-a", + cliAgentType: "codex", + accountId: "codex-account", + }, + }); + const nativeUser = event("native-user-1", "user", "new request"); + state.persistedBatches = [ + [], + [nativeUser], + [nativeUser, event("native-agent-1", "assistant", "answer")], + ]; + + const result = await runConversationTurn(params()); + + expect(result.terminalStatus).toBe("completed"); + expect(state.tailEventIds).toEqual([["native-agent-1"]]); + }); + + it("discards a remembered CLI setup that predates explicit account binding", async () => { + vi.mocked(loadForkSetupMemory).mockReturnValueOnce({ + workspaceRepoPath: "/old-repo", + execution: { + agentDefinitionId: "agent-a", + cliAgentType: "codex", + }, + }); + vi.mocked(requestForkSessionSetup).mockResolvedValueOnce({ + workspaceRepoPath: "/repo", + execution: { + agentDefinitionId: "agent-a", + cliAgentType: "codex", + accountId: "codex-account", + }, + }); + state.persistedBatches = [ + [], + [ + event("native-user-1", "user", "new request"), + event("agent-1", "assistant", "answer"), + ], + ]; + + await runConversationTurn(params({ sourceScopeKey: "scope" })); + + expect(clearForkSetupMemory).toHaveBeenCalledWith("scope"); + expect(requestForkSessionSetup).toHaveBeenCalledTimes(1); + expect(SessionService.create).toHaveBeenCalledWith( + expect.objectContaining({ + cliAgentType: "codex", + accountId: "codex-account", + }) + ); }); it("clears and cleans a failed execution episode", async () => { diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts index 9929d08ca..f74ab41cd 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts @@ -19,7 +19,6 @@ */ import Message from "@src/components/Message"; import type { TurnTerminalStatus } from "@src/engines/SessionCore/control/turnLifecycle"; -import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import { SessionService } from "@src/engines/SessionCore/services/SessionService"; import { @@ -28,6 +27,7 @@ import { waitForTurnOutcome, } from "@src/engines/SessionCore/services/TurnDispatchService"; import { mintTurnIntentId } from "@src/engines/SessionCore/sync/adapters/shared/eventFactories"; +import { loadAuthoritativeSessionEvents } from "@src/engines/SessionCore/sync/authoritativeSessionEvents"; import { requestForkSessionSetup } from "@src/features/TeamCollaboration/forkSession"; import { clearForkSetupMemory, @@ -60,6 +60,7 @@ import { } from "./conversationRunnerSessions"; import { buildConversationPlaneUserEvent, + sliceAppendedTurnTail, sliceTurnTailByIntent, turnIntentIdOf, } from "./conversationTurnEvents"; @@ -67,6 +68,8 @@ import { const log = createLogger("ConversationTurnRunner"); const TURN_DEADLINE_MS = 15 * 60_000; +const CLI_TRANSCRIPT_SETTLE_TIMEOUT_MS = 5_000; +const CLI_TRANSCRIPT_SETTLE_POLL_MS = 100; export const CONVERSATION_CONTEXT_MAX_ENTRIES = 60; export const CONVERSATION_TURN_LOCK_UNAVAILABLE = "ORG2_CONVERSATION_TURN_LOCK_UNAVAILABLE"; @@ -282,8 +285,8 @@ async function pushUserRow( io: TurnPushIo, displayText: string, dispatchIso: string -): Promise { - await pushConversationEvents(await io.getAccessToken(), { +): Promise { + const pushed = await pushConversationEvents(await io.getAccessToken(), { orgId: io.orgId, rootSessionId: io.rootSessionId, turnId: io.turnId, @@ -299,31 +302,49 @@ async function pushUserRow( ], }); io.onPushed?.(); + return pushed.lastSeq; } async function pushAgentTail( io: TurnPushIo, - runnerSessionId: string -): Promise { - const persisted = await eventStoreProxy - .getPersistedEvents(runnerSessionId) - .catch(() => [] as SessionEvent[]); - const sliced = sliceTurnTailByIntent(persisted, io.turnIntentId); - if (sliced === null) { - throw new Error( - `conversation turn ${io.turnIntentId} is missing its user anchor` + runnerSessionId: string, + deadlineMs: number, + eventsBefore?: readonly SessionEvent[] +): Promise<{ count: number; lastSeq?: number }> { + const settleDeadline = eventsBefore + ? Math.min(deadlineMs, Date.now() + CLI_TRANSCRIPT_SETTLE_TIMEOUT_MS) + : Date.now(); + let sliced: SessionEvent[] | null = null; + for (;;) { + const { events } = await loadAuthoritativeSessionEvents(runnerSessionId); + sliced = + sliceTurnTailByIntent(events, io.turnIntentId) ?? + (eventsBefore ? sliceAppendedTurnTail(eventsBefore, events) : null); + if (sliced && (sliced.length > 0 || !eventsBefore)) break; + if (!eventsBefore || Date.now() >= settleDeadline) { + throw new Error( + eventsBefore + ? `conversation turn ${io.turnIntentId} native transcript did not expose an agent tail` + : `conversation turn ${io.turnIntentId} is missing its user anchor` + ); + } + await new Promise((resolve) => + setTimeout(resolve, CLI_TRANSCRIPT_SETTLE_POLL_MS) ); } const tail = sliced.map(boundConversationEventForPush); - if (tail.length === 0) return 0; - await pushConversationEventsChunked(await io.getAccessToken(), { - orgId: io.orgId, - rootSessionId: io.rootSessionId, - turnId: io.turnId, - events: tail, - }); + if (tail.length === 0) return { count: 0 }; + const pushed = await pushConversationEventsChunked( + await io.getAccessToken(), + { + orgId: io.orgId, + rootSessionId: io.rootSessionId, + turnId: io.turnId, + events: tail, + } + ); io.onPushed?.(); - return tail.length; + return { count: tail.length, lastSeq: pushed.lastSeq }; } function collectLocalTurnIntentIds( @@ -451,9 +472,11 @@ async function runConversationTurnSerialized( } } if (decision.kind === "resume") { - const persistedBefore = await eventStoreProxy - .getPersistedEvents(decision.record.continuationSessionId) - .catch(() => [] as SessionEvent[]); + const persistedBefore = ( + await loadAuthoritativeSessionEvents( + decision.record.continuationSessionId + ) + ).events; const delta = await params.loadPlaneDelta( decision.record.readThroughPlaneSeq ); @@ -466,7 +489,11 @@ async function runConversationTurnSerialized( turnIntentId, turnIntentId ); - await pushUserRow(io, params.displayText, dispatchIso); + const userRowLastSeq = await pushUserRow( + io, + params.displayText, + dispatchIso + ); params.onUserMessagePublished?.(); let dispatch; @@ -491,7 +518,7 @@ async function runConversationTurnSerialized( request, deadlineMs, dispatchIso, - userRowAlreadyPushed: true, + userRowLastSeq, }); } await params.onTurnAccepted?.( @@ -503,6 +530,8 @@ async function runConversationTurnSerialized( runnerSessionId: decision.record.continuationSessionId, deadlineMs, readThroughPlaneSeq: delta.lastSeq, + userRowLastSeq, + eventsBefore: decision.record.cliAgentType ? persistedBefore : undefined, dispatch, }); } @@ -516,10 +545,10 @@ async function runConversationTurnSerialized( request, deadlineMs, dispatchIso, - userRowAlreadyPushed: false, }, { runnerSessionId: decision.record.continuationSessionId, + cliAgentType: decision.record.cliAgentType, accountId: decision.record.accountId, model: decision.record.model, }, @@ -533,7 +562,6 @@ async function runConversationTurnSerialized( request, deadlineMs, dispatchIso, - userRowAlreadyPushed: false, }, initialContext ); @@ -543,11 +571,13 @@ interface BootstrapTurn { request: string; deadlineMs: number; dispatchIso: string; - userRowAlreadyPushed: boolean; + /** Present when a rejected resume already published the idempotent row. */ + userRowLastSeq?: number; } interface BootstrapEpisode { runnerSessionId: string; + cliAgentType?: string; accountId?: string; model?: string; } @@ -568,9 +598,22 @@ async function startFreshEpisode( allowCliRuntime: true, lockSourceAgent: Boolean(params.assignedAgentDefinitionId), }); - const remembered = loadForkSetupMemory(setupMemoryKey); + const rememberedCandidate = loadForkSetupMemory(setupMemoryKey); + const remembered = + rememberedCandidate?.execution.cliAgentType && + !rememberedCandidate.execution.accountId + ? null + : rememberedCandidate; + if (rememberedCandidate && !remembered) { + clearForkSetupMemory(setupMemoryKey); + } let usedRememberedSetup = Boolean(remembered); let setup = remembered ?? (await requestSetup()); + if (setup.execution.cliAgentType && !setup.execution.accountId) { + throw new Error( + "External CLI continuation requires an explicit local account" + ); + } if (!remembered) saveForkSetupMemory(setupMemoryKey, setup); assertAssignedAgent(setup.execution.agentDefinitionId, params); const initialContext = @@ -596,6 +639,11 @@ async function startFreshEpisode( log.warn("remembered runner setup failed; re-prompting", error); clearForkSetupMemory(setupMemoryKey); setup = await requestSetup(); + if (setup.execution.cliAgentType && !setup.execution.accountId) { + throw new Error( + "External CLI continuation requires an explicit local account" + ); + } assertAssignedAgent(setup.execution.agentDefinitionId, params); saveForkSetupMemory(setupMemoryKey, setup); usedRememberedSetup = false; @@ -635,6 +683,7 @@ async function startFreshEpisode( turn, { runnerSessionId, + cliAgentType: setup.execution.cliAgentType, accountId: setup.execution.accountId, model: setup.execution.model, }, @@ -658,10 +707,18 @@ async function dispatchBootstrapEpisode( io.turnId, io.turnIntentId ); - if (!turn.userRowAlreadyPushed) { - await pushUserRow(io, params.displayText, turn.dispatchIso); + let userRowLastSeq = turn.userRowLastSeq; + if (userRowLastSeq === undefined) { + userRowLastSeq = await pushUserRow( + io, + params.displayText, + turn.dispatchIso + ); params.onUserMessagePublished?.(); } + const eventsBefore = episode.cliAgentType + ? (await loadAuthoritativeSessionEvents(episode.runnerSessionId)).events + : undefined; let dispatch; try { dispatch = await dispatchRunnerTurn(params, io, { @@ -699,6 +756,8 @@ async function dispatchBootstrapEpisode( runnerSessionId: episode.runnerSessionId, deadlineMs: turn.deadlineMs, readThroughPlaneSeq: initialContext.readThroughPlaneSeq, + userRowLastSeq, + eventsBefore, dispatch, }); } @@ -711,21 +770,33 @@ async function settleEpisode( runnerSessionId: string; deadlineMs: number; readThroughPlaneSeq: number; + userRowLastSeq: number; + eventsBefore?: readonly SessionEvent[]; dispatch: ReturnType; } ): Promise { const outcome = await waitForTurnOutcome(input.dispatch, input.deadlineMs); markConversationRunnerTerminal(input.key, input.runnerSessionId); - if (outcome.status === "completed") { - advanceContinuationReadThrough( - params.executionScopeKey, - params.rootSessionId, - input.readThroughPlaneSeq - ); - } let tailCount = 0; try { - tailCount = await pushAgentTail(io, input.runnerSessionId); + const tail = await pushAgentTail( + io, + input.runnerSessionId, + input.deadlineMs, + input.eventsBefore + ); + tailCount = tail.count; + if (outcome.status === "completed") { + advanceContinuationReadThrough( + params.executionScopeKey, + params.rootSessionId, + Math.max( + input.readThroughPlaneSeq, + input.userRowLastSeq, + tail.lastSeq ?? 0 + ) + ); + } } finally { if (outcome.status === "completed") { await cleanupRetiredConversationRunners(input.key, input.runnerSessionId); diff --git a/src/features/Org2Cloud/org2CloudSessionSync.pushEvents.ts b/src/features/Org2Cloud/org2CloudSessionSync.pushEvents.ts index 5be249093..1b20a4c1e 100644 --- a/src/features/Org2Cloud/org2CloudSessionSync.pushEvents.ts +++ b/src/features/Org2Cloud/org2CloudSessionSync.pushEvents.ts @@ -9,16 +9,12 @@ * network-facing halves live further down. */ import { getImportedHistorySourceBySessionId } from "@src/api/tauri/externalHistory"; -import { rpc } from "@src/api/tauri/rpc"; -import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import { processChunksRust } from "@src/engines/SessionCore/ingestion/rustBridge"; +import { loadAuthoritativeSessionEvents } from "@src/engines/SessionCore/sync/authoritativeSessionEvents"; import { createLogger } from "@src/hooks/logger"; import type { ActivityChunk } from "@src/types/session/session"; -import { - isCliSession, - isImportedHistorySession, -} from "@src/util/session/sessionDispatch"; +import { isImportedHistorySession } from "@src/util/session/sessionDispatch"; import { sha256Hex, @@ -172,29 +168,7 @@ export class Org2CloudSessionSyncPushEvents extends Org2CloudSessionSyncState { } return { events }; } - const revisionBefore = - await eventStoreProxy.getPersistedEventRevision(sessionId); - const persisted = await eventStoreProxy.getPersistedEvents(sessionId); - const revisionAfter = - await eventStoreProxy.getPersistedEventRevision(sessionId); - const localContentRevision = - revisionBefore && - revisionAfter && - revisionBefore.revision === revisionAfter.revision && - revisionAfter.eventCount === persisted.length - ? revisionAfter.revision - : undefined; - if (persisted.length > 0 || !isCliSession(sessionId)) { - return { events: persisted, localContentRevision }; - } - // Live CLI sessions keep their transcript of record in the CLI's native - // store (account-profile aware) and never write the events cache, so a - // persisted read alone pushes a hollow session: metadata with no replay, - // and the pass then stamps the event plane clean. Load the full native - // transcript through the same command the session-resume path uses. - const chunks = (await rpc.cli.chunks({ sessionId })) as ActivityChunk[]; - if (!Array.isArray(chunks) || chunks.length === 0) return { events: [] }; - return { events: await processChunksRust(chunks, sessionId) }; + return loadAuthoritativeSessionEvents(sessionId); } /** Authoritative complete loader retained for first anchor and recovery. */ diff --git a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts index 28f41c86a..a8b74bb0e 100644 --- a/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts +++ b/src/features/Org2Cloud/org2CloudSyncEngine.sessions.test.ts @@ -93,8 +93,7 @@ describe("Org2CloudSyncEngine session publishing", () => { ); }); - it("publishes live CLI sessions from the native transcript when the events cache is empty", async () => { - eventStoreMock.getPersistedEvents.mockResolvedValueOnce([]); + it("publishes live CLI sessions from the native transcript", async () => { const chunks = [{ id: "cli-chunk" }] as never; const chunksSpy = vi.spyOn(rpc.cli, "chunks").mockResolvedValue(chunks); const converted = [makeEvent("cli-event")]; @@ -114,13 +113,17 @@ describe("Org2CloudSyncEngine session publishing", () => { chunks, "cliagent-123-native" ); + expect(eventStoreMock.getPersistedEvents).not.toHaveBeenCalledWith( + "cliagent-123-native" + ); chunksSpy.mockRestore(); }); - it("prefers the persisted event cache for CLI sessions when it is populated", async () => { - const persisted = [makeEvent("persisted-cli-event")]; - eventStoreMock.getPersistedEvents.mockResolvedValueOnce(persisted); - const chunksSpy = vi.spyOn(rpc.cli, "chunks"); + it("uses CLI-owned history even when EventStore has an optimistic row", async () => { + const chunks = [{ id: "authoritative-cli-chunk" }] as never; + const chunksSpy = vi.spyOn(rpc.cli, "chunks").mockResolvedValue(chunks); + const converted = [makeEvent("authoritative-cli-event")]; + processChunksRustMock.mockResolvedValueOnce(converted); const events = await ( engine as unknown as { @@ -128,8 +131,17 @@ describe("Org2CloudSyncEngine session publishing", () => { } ).loadPushEvents("cliagent-123-native"); - expect(events).toEqual(persisted); - expect(chunksSpy).not.toHaveBeenCalled(); + expect(events).toEqual(converted); + expect(chunksSpy).toHaveBeenCalledWith({ + sessionId: "cliagent-123-native", + }); + expect(processChunksRustMock).toHaveBeenCalledWith( + chunks, + "cliagent-123-native" + ); + expect(eventStoreMock.getPersistedEvents).not.toHaveBeenCalledWith( + "cliagent-123-native" + ); chunksSpy.mockRestore(); }); it("pushes only scope-matched own sessions (metadata + epoch-1 rewrite)", async () => { diff --git a/src/features/TeamCollaboration/components/ForkSessionSetupDialog/index.tsx b/src/features/TeamCollaboration/components/ForkSessionSetupDialog/index.tsx index 8b00fc226..50827c495 100644 --- a/src/features/TeamCollaboration/components/ForkSessionSetupDialog/index.tsx +++ b/src/features/TeamCollaboration/components/ForkSessionSetupDialog/index.tsx @@ -8,6 +8,7 @@ import Button from "@src/components/Button"; import Select from "@src/components/Select"; import type { SelectOption } from "@src/components/Select"; import { getCliTransportLabel } from "@src/config/cliAgents"; +import { getCliCompatibleAccountsForAgent } from "@src/hooks/models/useAgentCompatibility"; import { accountHasModel, accountModelIds, @@ -110,7 +111,7 @@ const ForkSessionSetupForm: React.FC = ({ getShareableScopeKeyVersion ); - const runnableAccounts = useMemo( + const runnableNativeAccounts = useMemo( () => accounts.filter( (account) => @@ -123,15 +124,6 @@ const ForkSessionSetupForm: React.FC = ({ ); const sourceModel = request.sourceModel; const sourceAgentDefinitionId = request.sourceAgentDefinitionId; - const preferredAccount = useMemo( - () => - (sourceModel - ? runnableAccounts.find((account) => - accountHasModel(account, sourceModel) - ) - : undefined) ?? runnableAccounts[0], - [sourceModel, runnableAccounts] - ); const preferredAgent = useMemo(() => { const sourceAgent = sourceAgentDefinitionId ? allAgents.find((agent) => agent.id === sourceAgentDefinitionId) @@ -158,25 +150,6 @@ const ForkSessionSetupForm: React.FC = ({ ) ?? null, [allAgents, chosenAgentDefinitionId, preferredAgent?.id] ); - const agentPreferredAccountId = selectedAgent?.selectedAccountId - ? runnableAccounts.find( - (account) => account.id === selectedAgent.selectedAccountId - )?.id - : undefined; - const accountId = - chosenAccountId || agentPreferredAccountId || preferredAccount?.id || ""; - const selectedAccount = runnableAccounts.find( - (account) => account.id === accountId - ); - const accountOptions = useMemo( - () => - runnableAccounts.map((account) => ({ - value: account.id, - label: `${account.name} · ${account.modelType}`, - triggerLabel: account.name, - })), - [runnableAccounts] - ); const agentOptions = useMemo( () => allAgents.map((agent) => ({ @@ -222,6 +195,48 @@ const ForkSessionSetupForm: React.FC = ({ ) ?? null ); }, [chosenRuntime, runnableCliAgents]); + const runnableCliAccounts = useMemo( + () => + selectedCliAgent + ? getCliCompatibleAccountsForAgent( + selectedCliAgent.agent, + selectedCliAgent.cliAgentType, + accounts + ).filter((account) => account.enabled && account.hasKey) + : [], + [accounts, selectedCliAgent] + ); + const executionAccounts = selectedCliAgent + ? runnableCliAccounts + : runnableNativeAccounts; + const preferredAccount = useMemo( + () => + (sourceModel + ? executionAccounts.find((account) => + accountHasModel(account, sourceModel) + ) + : undefined) ?? executionAccounts[0], + [executionAccounts, sourceModel] + ); + const agentPreferredAccountId = selectedAgent?.selectedAccountId + ? executionAccounts.find( + (account) => account.id === selectedAgent.selectedAccountId + )?.id + : undefined; + const accountId = + chosenAccountId || agentPreferredAccountId || preferredAccount?.id || ""; + const selectedAccount = executionAccounts.find( + (account) => account.id === accountId + ); + const accountOptions = useMemo( + () => + executionAccounts.map((account) => ({ + value: account.id, + label: `${account.name} · ${account.modelType}`, + triggerLabel: account.name, + })), + [executionAccounts] + ); const modelOptions = useMemo(() => { if (!selectedAccount) return []; return accountModelIds(selectedAccount) @@ -259,7 +274,7 @@ const ForkSessionSetupForm: React.FC = ({ Boolean(selectedAccount && accountId && model) && Boolean(selectedAccount && accountHasModel(selectedAccount, model)); const executionReady = selectedCliAgent - ? true + ? Boolean(selectedAccount && accountId) : chosenRuntime === "native" && nativeExecutionReady; const canContinue = Boolean(selectedAgent) && @@ -299,6 +314,8 @@ const ForkSessionSetupForm: React.FC = ({ ? { agentDefinitionId: selectedAgent.id, cliAgentType: selectedCliAgent.cliAgentType, + accountId, + model: model || undefined, } : { agentDefinitionId: selectedAgent.id, @@ -407,42 +424,44 @@ const ForkSessionSetupForm: React.FC = ({ { + setChosenAccountId(String(value)); + setChosenModel(""); + }} + style={{ width: "100%" }} + dataTestId="fork-setup-account" + /> + {!selectedCliAgent ? ( - <> - - +