From 93089c0bc4fb5e6b1a0267ed54cb9081a8d8df9a Mon Sep 17 00:00:00 2001 From: Neonforge <48338160+Neonforge98@users.noreply.github.com> Date: Tue, 25 Aug 2026 08:43:24 +0800 Subject: [PATCH] feat(conversation): persist local continuations --- .../ChatPanel/ConversationStreamProvider.tsx | 4 +- .../hooks/useImportedSessionSubmitOverride.ts | 141 +++- .../activeConversationRunnersAtom.test.ts | 35 +- .../activeConversationRunnersAtom.ts | 50 +- .../conversationContinuation.test.ts | 57 ++ .../conversationContinuation.ts | 109 +++ .../conversationExecutionStore.test.ts | 151 ++++ .../conversationExecutionStore.ts | 387 ++++++++-- .../conversationOwnerPublisher.ts | 128 +--- .../conversationRunnerScope.tsx | 2 +- .../conversationRunnerSessions.ts | 85 ++- .../conversationTimeline.ts | 8 +- .../conversationTurnEvents.ts | 71 ++ .../conversationTurnRunner.test.ts | 383 ++++++++++ .../conversationTurnRunner.ts | 696 ++++++++++++++---- .../org2CloudConversationEventsClient.test.ts | 113 +++ .../org2CloudConversationEventsClient.ts | 66 ++ 17 files changed, 2138 insertions(+), 348 deletions(-) create mode 100644 src/features/Org2Cloud/SessionConversation/conversationContinuation.test.ts create mode 100644 src/features/Org2Cloud/SessionConversation/conversationContinuation.ts create mode 100644 src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts create mode 100644 src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts create mode 100644 src/features/Org2Cloud/org2CloudConversationEventsClient.test.ts diff --git a/src/engines/ChatPanel/ConversationStreamProvider.tsx b/src/engines/ChatPanel/ConversationStreamProvider.tsx index 2f29567d05..646399ef7e 100644 --- a/src/engines/ChatPanel/ConversationStreamProvider.tsx +++ b/src/engines/ChatPanel/ConversationStreamProvider.tsx @@ -7,6 +7,7 @@ import { useSessionCommentsContext } from "@src/features/Org2Cloud/SessionCommen import { activeConversationRunnersAtom, collectLandedTurnIds, + overlayableRunnerEvents, selectActiveRunners, } from "@src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom"; import { @@ -277,8 +278,7 @@ export function ConversationStreamProvider({ for (const runner of activeRunners) { const live = runnerEventsById.get(runner.runnerSessionId); if (!live?.length) continue; - for (const event of live) { - if (event.source === "user") continue; + for (const event of overlayableRunnerEvents(live, runner.turnIntentId)) { synthetic.push({ ...event, id: `runlive-${event.id}`, diff --git a/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts b/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts index 32965fd0b3..18a7a973db 100644 --- a/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts +++ b/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts @@ -12,7 +12,11 @@ import { type ConversationFamilyMember, resolveConversationFamily, } from "@src/features/Org2Cloud/SessionConversation/continuationEvents"; -import { cloudConversationExecutorScopeKey } from "@src/features/Org2Cloud/SessionConversation/conversationExecutionStore"; +import { + cloudConversationExecutorScopeKey, + cloudConversationSetupMemoryKey, + loadStoredOwnerPlaneCursor, +} from "@src/features/Org2Cloud/SessionConversation/conversationExecutionStore"; import { publishOwnerTurn } from "@src/features/Org2Cloud/SessionConversation/conversationOwnerPublisher"; import { bumpConversationPlaneSignal, @@ -20,11 +24,14 @@ import { conversationPlaneKey, conversationPlaneSignalAtom, } from "@src/features/Org2Cloud/SessionConversation/conversationPlaneAtom"; -import { buildConversationPlaneStreamEvents } from "@src/features/Org2Cloud/SessionConversation/conversationPlaneEvents"; -import { mergePlaneIntoTranscript } from "@src/features/Org2Cloud/SessionConversation/conversationTimeline"; import { - buildRunnerPrompt, - renderConversationContext, + conversationEventKey, + mergePlaneIntoTranscript, +} from "@src/features/Org2Cloud/SessionConversation/conversationTimeline"; +import { + CONVERSATION_CONTEXT_MAX_ENTRIES, + buildResumePrompt, + renderPlaneDeltaContext, runConversationTurn, } from "@src/features/Org2Cloud/SessionConversation/conversationTurnRunner"; import { @@ -37,6 +44,7 @@ import { org2CloudAuthIdentityKey, } from "@src/features/Org2Cloud/org2CloudAuthAtom"; import { ensureFreshSession } from "@src/features/Org2Cloud/org2CloudClient"; +import { listConversationEventsFrom } from "@src/features/Org2Cloud/org2CloudConversationEventsClient"; import { org2CloudRemoteSessionsAtom } from "@src/features/Org2Cloud/org2CloudRemoteSessionsAtom"; import { findImportedSession } from "@src/features/TeamCollaboration/engine/collabImportIdentity"; import { getSessionForkedFrom } from "@src/features/TeamCollaboration/forkSession"; @@ -157,7 +165,7 @@ export function useImportedSessionSubmitOverride({ // CONVERSATION PLANE (0024): once the backend supports the multi-writer // turn plane, implicit sends stop forking entirely — a member's turn runs - // in an invisible one-shot local session and publishes to the plane; the + // in an invisible persistent local continuation and publishes to the plane; the // owner's sends keep their own session but inject the plane delta as // context. The fork/tip paths below remain ONLY as the pre-0024 fallback. const setAuth = useSetAtom(org2CloudAuthAtom); @@ -246,8 +254,8 @@ export function useImportedSessionSubmitOverride({ async (input: SubmitOverrideInput): Promise => { const planeReady = planeInfo?.entry.state === "ready"; // (a) Member send on a plane-capable backend: publish the message to - // the conversation immediately, run the turn in an invisible one-shot - // local session, stream the agent tail back to the plane. No fork. + // the conversation immediately, run the turn in an invisible local + // continuation, stream the agent tail back to the plane. No fork. if (planeReady && planeInfo && !viewerOwnsRoot) { if (forkSubmitInFlightRef.current) { restorePendingDraft(input, sessionId); @@ -269,17 +277,6 @@ export function useImportedSessionSubmitOverride({ planeInfo.rootId, auth.supabaseUrl ); - const rootEvents = rootLocal - ? await eventStoreProxy - .getPersistedEvents(rootLocal.session_id) - .catch(() => [] as SessionEvent[]) - : []; - const timeline = mergePlaneIntoTranscript( - rootEvents, - planeInfo.entry.events, - sessionId, - auth.userId - ); // The root row's repo scope keys the setup memory AND resolves the // runner's local checkout — without it the dialog reappears and a // workspace-requiring agent cannot launch at all. @@ -288,6 +285,11 @@ export function useImportedSessionSubmitOverride({ (candidate) => candidate.sourceSessionId === planeInfo.rootId ) : undefined; + const authIdentity = org2CloudAuthIdentityKey(auth); + const executorScope = cloudConversationExecutorScopeKey( + authIdentity, + planeInfo.orgId + ); let publishResolve!: () => void; const userPublished = new Promise((resolve) => { publishResolve = resolve; @@ -319,14 +321,69 @@ export function useImportedSessionSubmitOverride({ displayText: input.displayText, agentContent: input.agentContent, imageDataUrls: input.imageDataUrls, - timeline, + loadInitialContext: async (excludeTurnIntentId) => { + const window = await listConversationEventsFrom( + await getAccessToken(), + { + orgId: planeInfo.orgId, + rootSessionId: planeInfo.rootId, + afterSeq: 0, + retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, + } + ); + const rows = window.events.filter( + (row) => row.turnId !== excludeTurnIntentId + ); + const rootEvents = rootLocal + ? await eventStoreProxy + .getPersistedEvents(rootLocal.session_id) + .catch(() => [] as SessionEvent[]) + : []; + const timeline = mergePlaneIntoTranscript( + rootEvents, + rows, + sessionId, + auth.userId + ); + const authorByEventKey = new Map( + rows.map((row) => [ + conversationEventKey(row.event), + row.authorDisplayName ?? row.authorUserId, + ]) + ); + const senders = new Map(); + for (const event of timeline) { + const sender = authorByEventKey.get( + conversationEventKey(event) + ); + if (sender) senders.set(event.id, sender); + } + return { + timeline, + senders, + readThroughPlaneSeq: window.lastSeq, + }; + }, + loadPlaneDelta: (afterSeq) => + getAccessToken().then((accessToken) => + listConversationEventsFrom(accessToken, { + orgId: planeInfo.orgId, + rootSessionId: planeInfo.rootId, + afterSeq, + retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, + }) + ), sourceScopeKey: rootRow?.repoScopeKey, sourceModel: currentSession?.model ?? rootRow?.model, - executionScopeKey: cloudConversationExecutorScopeKey( - org2CloudAuthIdentityKey(auth), - planeInfo.orgId + assignedAgentDefinitionId: rootRow?.agentDefinitionId, + setupMemoryKey: cloudConversationSetupMemoryKey( + authIdentity, + planeInfo.orgId, + planeInfo.rootId, + rootRow?.agentDefinitionId ), - onRunnerReady: (runnerSessionId, turnId) => { + executionScopeKey: executorScope, + onRunnerReady: (runnerSessionId, turnId, turnIntentId) => { // Plumbing session: never sync it to the cloud as a session. setAccessSettings((current) => withCloudSessionMode( @@ -344,7 +401,10 @@ export function useImportedSessionSubmitOverride({ const list = current[planeInfo.rootId] ?? []; return { ...current, - [planeInfo.rootId]: [...list, { runnerSessionId, turnId }], + [planeInfo.rootId]: [ + ...list, + { runnerSessionId, turnId, turnIntentId }, + ], }; }); }, @@ -387,15 +447,34 @@ export function useImportedSessionSubmitOverride({ const freshAuth = await ensureFreshSession(auth); if (!freshAuth) return false; commitRefreshedAuth(setAuth, auth, freshAuth); - const othersRows = planeInfo.entry.events.filter( + const executorScope = cloudConversationExecutorScopeKey( + org2CloudAuthIdentityKey(auth), + planeInfo.orgId + ); + const ownerCursor = + loadStoredOwnerPlaneCursor(executorScope, planeInfo.rootId) + ?.readThroughPlaneSeq ?? 0; + let delta; + try { + delta = await listConversationEventsFrom(freshAuth.accessToken, { + orgId: planeInfo.orgId, + rootSessionId: planeInfo.rootId, + afterSeq: ownerCursor, + retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, + }); + } catch (error) { + logger.error("owner conversation delta load failed", error); + restorePendingDraft(input, sessionId); + Message.error(t("collaboration.forkImported.sendFailed")); + return true; + } + const othersRows = delta.events.filter( (row) => row.authorUserId !== auth.userId ); const agentContent = othersRows.length > 0 - ? buildRunnerPrompt( - renderConversationContext( - buildConversationPlaneStreamEvents(othersRows, sessionId) - ), + ? buildResumePrompt( + renderPlaneDeltaContext(othersRows), input.agentContent ?? input.displayText ) : input.agentContent; @@ -424,6 +503,8 @@ export function useImportedSessionSubmitOverride({ sessionId, turnIntentId, displayText: input.displayText, + executorScope, + readThroughPlaneSeq: delta.lastSeq, onPushed: () => bumpConversationPlaneSignal(setPlaneSignal, planeInfo.orgId), }).catch((error: unknown) => { diff --git a/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.test.ts b/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.test.ts index 216304f9dd..f33b0fc832 100644 --- a/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.test.ts +++ b/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it } from "vitest"; +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; + import { collectLandedTurnIds, + overlayableRunnerEvents, selectActiveRunners, } from "./activeConversationRunnersAtom"; @@ -31,8 +34,8 @@ describe("collectLandedTurnIds", () => { describe("selectActiveRunners", () => { const runners = [ - { runnerSessionId: "r1", turnId: "t1" }, - { runnerSessionId: "r2", turnId: "t2" }, + { runnerSessionId: "r1", turnId: "t1", turnIntentId: "i1" }, + { runnerSessionId: "r2", turnId: "t2", turnIntentId: "i2" }, ]; it("keeps a runner while only its user row is on the plane", () => { @@ -49,3 +52,31 @@ describe("selectActiveRunners", () => { expect(selectActiveRunners(runners, landed)).toEqual([runners[1]]); }); }); + +describe("overlayableRunnerEvents", () => { + const event = (overrides: Partial) => + ({ + id: "event", + source: "assistant", + result: {}, + ...overrides, + }) as SessionEvent; + const user = (id: string, turnIntentId: string) => + event({ + id, + source: "user", + result: { turnIntentId }, + }); + + it("overlays only the current turn from a reused runner", () => { + const events = [ + user("old-user", "old-intent"), + event({ id: "old-tail" }), + user("current-user", "current-intent"), + event({ id: "current-tail" }), + ]; + expect( + overlayableRunnerEvents(events, "current-intent").map((item) => item.id) + ).toEqual(["current-tail"]); + }); +}); diff --git a/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.ts b/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.ts index 50cd3eb86c..9e208bfff5 100644 --- a/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.ts +++ b/src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom.ts @@ -1,7 +1,7 @@ /** * Live overlay registry for in-flight member turns. * - * A member's send runs the turn in an invisible one-shot local runner and + * A member's send runs the turn in an invisible persistent local runner and * only publishes the agent tail to the plane at terminal — so without this, * even the SENDER stares at their own message with no thinking, no tools, * no "Agent worked for Ns" until the whole turn lands at once. @@ -21,16 +21,50 @@ import { atom } from "jotai"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import { sliceTurnTailByIntent } from "./conversationTurnEvents"; + export interface ActiveConversationRunner { runnerSessionId: string; /** The turnId the tail is pushed under — the plane-landed drop signal. */ turnId: string; + /** Exact runtime turn overlaid from a reusable local session. */ + turnIntentId: string; +} + +type RunnerRegistry = Record; + +function dedupeByTurnId( + runners: readonly ActiveConversationRunner[] +): ActiveConversationRunner[] { + const seen = new Set(); + const kept: ActiveConversationRunner[] = []; + for (let index = runners.length - 1; index >= 0; index -= 1) { + const runner = runners[index]; + if (seen.has(runner.turnId)) continue; + seen.add(runner.turnId); + kept.push(runner); + } + return kept.reverse(); } /** plane rootSessionId → this device's in-flight member runners. */ -export const activeConversationRunnersAtom = atom< - Record ->({}); +const runnerRegistryStateAtom = atom({}); +export const activeConversationRunnersAtom = atom( + (get) => get(runnerRegistryStateAtom), + ( + get, + set, + update: RunnerRegistry | ((current: RunnerRegistry) => RunnerRegistry) + ) => { + const current = get(runnerRegistryStateAtom); + const proposed = typeof update === "function" ? update(current) : update; + const next: RunnerRegistry = {}; + for (const [rootSessionId, runners] of Object.entries(proposed)) { + next[rootSessionId] = dedupeByTurnId(runners); + } + set(runnerRegistryStateAtom, next); + } +); activeConversationRunnersAtom.debugLabel = "activeConversationRunnersAtom"; /** Plane turnIds whose agent tail has landed (a non-user row is present). */ @@ -44,6 +78,14 @@ export function collectLandedTurnIds( return landed; } +/** Exact current-turn tail from a reusable runner transcript. */ +export function overlayableRunnerEvents( + events: readonly SessionEvent[], + turnIntentId: string +): SessionEvent[] { + return sliceTurnTailByIntent(events, turnIntentId) ?? []; +} + /** Runners still worth overlaying: their turn has no agent tail on the plane yet. */ export function selectActiveRunners( runners: readonly ActiveConversationRunner[], diff --git a/src/features/Org2Cloud/SessionConversation/conversationContinuation.test.ts b/src/features/Org2Cloud/SessionConversation/conversationContinuation.test.ts new file mode 100644 index 0000000000..7da6157277 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/conversationContinuation.test.ts @@ -0,0 +1,57 @@ +import { describe, expect, it } from "vitest"; + +import type { ConversationContinuationRecord } from "./conversationContinuation"; +import { decideContinuation } from "./conversationContinuation"; + +const established: ConversationContinuationRecord = { + continuationSessionId: "runner-1", + readThroughPlaneSeq: 12, + established: true, + agentDefinitionId: "agent-a", + updatedAt: "2026-08-25T00:00:00Z", +}; + +describe("decideContinuation", () => { + it("starts fresh without a persisted episode", () => { + expect( + decideContinuation({ record: null, turnIntentId: "intent-1" }) + ).toEqual({ kind: "fresh" }); + }); + + it("resumes an established matching episode", () => { + expect( + decideContinuation({ + record: established, + turnIntentId: "intent-2", + assignedAgentDefinitionId: "agent-a", + }) + ).toEqual({ kind: "resume", record: established }); + }); + + it("rolls when the assigned agent changes", () => { + expect( + decideContinuation({ + record: established, + turnIntentId: "intent-2", + assignedAgentDefinitionId: "agent-b", + }) + ).toEqual({ kind: "fresh", rollReason: "assigned_agent_changed" }); + }); + + it("retries only the exact unestablished bootstrap intent", () => { + const prepared: ConversationContinuationRecord = { + ...established, + established: false, + bootstrapTurnIntentId: "intent-bootstrap", + }; + expect( + decideContinuation({ + record: prepared, + turnIntentId: "intent-bootstrap", + }) + ).toEqual({ kind: "bootstrap", record: prepared }); + expect( + decideContinuation({ record: prepared, turnIntentId: "intent-next" }) + ).toEqual({ kind: "fresh", rollReason: "bootstrap_intent_changed" }); + }); +}); diff --git a/src/features/Org2Cloud/SessionConversation/conversationContinuation.ts b/src/features/Org2Cloud/SessionConversation/conversationContinuation.ts new file mode 100644 index 0000000000..482b9ea396 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/conversationContinuation.ts @@ -0,0 +1,109 @@ +/** Persistent local execution episode for a shared conversation. */ +import { + type ConversationContinuationRecord, + advanceStoredContinuationReadThrough, + clearStoredContinuation, + loadStoredContinuation, + markStoredContinuationEstablished, + prepareStoredContinuation, + saveStoredContinuation, +} from "./conversationExecutionStore"; + +export type { ConversationContinuationRecord }; + +export function loadContinuation( + executorScope: string, + rootSessionId: string, + backing?: Storage | null +): ConversationContinuationRecord | null { + return loadStoredContinuation(executorScope, rootSessionId, backing); +} + +export function saveContinuation( + executorScope: string, + rootSessionId: string, + record: Omit, + backing?: Storage | null +): void { + saveStoredContinuation(executorScope, rootSessionId, record, backing); +} + +export function prepareContinuation( + executorScope: string, + rootSessionId: string, + record: Omit, + preparedAt: string, + backing?: Storage | null +): void { + prepareStoredContinuation( + executorScope, + rootSessionId, + record, + preparedAt, + backing + ); +} + +export function clearContinuation( + executorScope: string, + rootSessionId: string, + backing?: Storage | null +): void { + clearStoredContinuation(executorScope, rootSessionId, backing); +} + +export function advanceContinuationReadThrough( + executorScope: string, + rootSessionId: string, + planeSeq: number, + backing?: Storage | null +): void { + advanceStoredContinuationReadThrough( + executorScope, + rootSessionId, + planeSeq, + backing + ); +} + +export function markContinuationEstablished( + executorScope: string, + rootSessionId: string, + continuationSessionId: string, + bootstrapTurnIntentId: string, + backing?: Storage | null +): boolean { + return markStoredContinuationEstablished( + executorScope, + rootSessionId, + continuationSessionId, + bootstrapTurnIntentId, + backing + ); +} + +export type ContinuationDecision = + | { kind: "resume"; record: ConversationContinuationRecord } + | { kind: "bootstrap"; record: ConversationContinuationRecord } + | { kind: "fresh"; rollReason?: string }; + +export function decideContinuation(input: { + record: ConversationContinuationRecord | null; + turnIntentId: string; + assignedAgentDefinitionId?: string; +}): ContinuationDecision { + const { record } = input; + if (!record) return { kind: "fresh" }; + if ( + input.assignedAgentDefinitionId && + input.assignedAgentDefinitionId !== record.agentDefinitionId + ) { + return { kind: "fresh", rollReason: "assigned_agent_changed" }; + } + if (!record.established) { + return record.bootstrapTurnIntentId === input.turnIntentId + ? { kind: "bootstrap", record } + : { kind: "fresh", rollReason: "bootstrap_intent_changed" }; + } + return { kind: "resume", record }; +} diff --git a/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.test.ts b/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.test.ts index 248f21d271..39ee990df5 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.test.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.test.ts @@ -2,13 +2,21 @@ import { describe, expect, it } from "vitest"; import { __CONVERSATION_EXECUTION_STORE_INTERNALS, + advanceStoredContinuationReadThrough, + advanceStoredOwnerPlaneCursor, cloudConversationExecutorScopeKey, + cloudConversationSetupMemoryKey, collectStoredRunnerSessionIds, conversationExecutionKey, forgetStoredRunner, + loadStoredContinuation, + loadStoredOwnerPlaneCursor, loadStoredRunnerRegistryEntry, + markStoredContinuationEstablished, markStoredRunnerTerminal, + prepareStoredContinuation, registerStoredRunner, + saveStoredContinuation, } from "./conversationExecutionStore"; function fakeStorage(): Storage { @@ -63,6 +71,133 @@ describe("conversation execution store", () => { ); }); + it("shares setup memory across surfaces per account, root, and agent", () => { + const first = cloudConversationSetupMemoryKey( + "cloud|user", + "org", + "root", + "agent-a" + ); + expect( + cloudConversationSetupMemoryKey("cloud|user", "org", "root", "agent-a") + ).toBe(first); + expect( + cloudConversationSetupMemoryKey("cloud|user", "org", "root", "agent-b") + ).not.toBe(first); + }); + + it("preserves runner, continuation, and owner cursor in one entry", () => { + const backing = fakeStorage(); + const key = conversationExecutionKey("scope", "root"); + prepareStoredContinuation( + "scope", + "root", + { + continuationSessionId: "runner-1", + readThroughPlaneSeq: 12, + established: true, + agentDefinitionId: "agent-a", + }, + "2026-08-25T00:00:00Z", + backing + ); + advanceStoredOwnerPlaneCursor("scope", "root", 9, backing); + markStoredRunnerTerminal(key, "runner-1", backing); + + expect(loadStoredRunnerRegistryEntry(key, backing)).toMatchObject({ + runnerSessionIds: ["runner-1"], + terminalRunnerSessionIds: ["runner-1"], + }); + expect(loadStoredContinuation("scope", "root", backing)).toMatchObject({ + continuationSessionId: "runner-1", + readThroughPlaneSeq: 12, + }); + expect( + loadStoredOwnerPlaneCursor("scope", "root", backing)?.readThroughPlaneSeq + ).toBe(9); + expect(backing.length).toBe(1); + }); + + it("advances both plane cursors monotonically", () => { + const backing = fakeStorage(); + saveStoredContinuation( + "scope", + "root", + { + continuationSessionId: "runner-1", + readThroughPlaneSeq: 12, + established: true, + agentDefinitionId: "agent-a", + }, + backing + ); + advanceStoredContinuationReadThrough("scope", "root", 40, backing); + advanceStoredContinuationReadThrough("scope", "root", 30, backing); + advanceStoredOwnerPlaneCursor("scope", "root", 18, backing); + advanceStoredOwnerPlaneCursor("scope", "root", 11, backing); + + expect( + loadStoredContinuation("scope", "root", backing)?.readThroughPlaneSeq + ).toBe(40); + expect( + loadStoredOwnerPlaneCursor("scope", "root", backing)?.readThroughPlaneSeq + ).toBe(18); + expect(() => + advanceStoredOwnerPlaneCursor("scope", "root", -1, backing) + ).toThrow("invalid conversation plane seq"); + }); + + it("establishes only the exact bootstrap runner and intent", () => { + const backing = fakeStorage(); + saveStoredContinuation( + "scope", + "root", + { + continuationSessionId: "runner-1", + readThroughPlaneSeq: 0, + established: false, + bootstrapTurnIntentId: "intent-1", + agentDefinitionId: "agent-a", + }, + backing + ); + + expect( + markStoredContinuationEstablished( + "scope", + "root", + "runner-2", + "intent-1", + backing + ) + ).toBe(false); + expect( + markStoredContinuationEstablished( + "scope", + "root", + "runner-1", + "intent-2", + backing + ) + ).toBe(false); + expect( + markStoredContinuationEstablished( + "scope", + "root", + "runner-1", + "intent-1", + backing + ) + ).toBe(true); + expect(loadStoredContinuation("scope", "root", backing)).toMatchObject({ + established: true, + readThroughPlaneSeq: 0, + }); + expect( + loadStoredContinuation("scope", "root", backing)?.bootstrapTurnIntentId + ).toBeUndefined(); + }); + it("persists and sanitizes runner lifecycle per conversation", () => { const backing = fakeStorage(); const key = conversationExecutionKey("scope", "root"); @@ -172,6 +307,18 @@ describe("conversation execution store", () => { backing ); registerStoredRunner(second, "keep-other", "2026-08-25T00:00:02Z", backing); + saveStoredContinuation( + "scope", + "first", + { + continuationSessionId: "remove-me", + readThroughPlaneSeq: 4, + established: true, + agentDefinitionId: "agent-a", + }, + backing + ); + advanceStoredOwnerPlaneCursor("scope", "first", 3, backing); backing.setItem( __CONVERSATION_EXECUTION_STORE_INTERNALS.LEGACY_RUNNERS_KEY, JSON.stringify({ @@ -192,5 +339,9 @@ describe("conversation execution store", () => { runnerSessionIds: ["keep-current"], terminalRunnerSessionIds: [], }); + expect(loadStoredContinuation("scope", "first", backing)).toBeNull(); + expect( + loadStoredOwnerPlaneCursor("scope", "first", backing)?.readThroughPlaneSeq + ).toBe(3); }); }); diff --git a/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.ts b/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.ts index 9b4a5b1893..c1a701e25b 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationExecutionStore.ts @@ -19,9 +19,31 @@ export interface ConversationRunnerRegistryEntry { updatedAt: string; } +export interface ConversationContinuationRecord { + continuationSessionId: string; + /** Highest plane seq this continuation has successfully consumed. */ + readThroughPlaneSeq: number; + /** False only between preparing a blank runner and adapter acceptance. */ + established: boolean; + bootstrapTurnIntentId?: string; + agentDefinitionId: string; + accountId?: string; + model?: string; + workspaceRepoPath?: string | null; + updatedAt: string; +} + +export interface ConversationPlaneReadCursor { + readThroughPlaneSeq: number; + updatedAt: string; +} + interface ConversationExecutionEnvelope { version: typeof STORE_VERSION; runners?: ConversationRunnerRegistryEntry; + continuation?: ConversationContinuationRecord; + /** Independent cursor for execution directly in the owner's root session. */ + ownerPlaneCursor?: ConversationPlaneReadCursor; } function defaultStorage(): Storage | null { @@ -54,6 +76,19 @@ function uniqueStrings(value: unknown): string[] { ]; } +function validPlaneSeq(value: unknown): value is number { + return Number.isSafeInteger(value) && (value as number) >= 0; +} + +function optionalString( + source: Record, + key: string +): string | undefined { + return typeof source[key] === "string" && source[key].length > 0 + ? source[key] + : undefined; +} + function sanitizeRunners( value: unknown ): ConversationRunnerRegistryEntry | null { @@ -73,6 +108,62 @@ function sanitizeRunners( }; } +function sanitizeContinuation( + value: unknown +): ConversationContinuationRecord | null { + if (!isObject(value)) return null; + const continuationSessionId = optionalString(value, "continuationSessionId"); + const agentDefinitionId = optionalString(value, "agentDefinitionId"); + if ( + !continuationSessionId || + !agentDefinitionId || + !validPlaneSeq(value.readThroughPlaneSeq) || + typeof value.established !== "boolean" + ) { + return null; + } + const bootstrapTurnIntentId = optionalString(value, "bootstrapTurnIntentId"); + if ( + (!value.established && !bootstrapTurnIntentId) || + (value.established && bootstrapTurnIntentId) + ) { + return null; + } + const record: ConversationContinuationRecord = { + continuationSessionId, + readThroughPlaneSeq: value.readThroughPlaneSeq, + established: value.established, + agentDefinitionId, + updatedAt: optionalString(value, "updatedAt") ?? UNKNOWN_UPDATED_AT, + }; + if (bootstrapTurnIntentId) { + record.bootstrapTurnIntentId = bootstrapTurnIntentId; + } + const accountId = optionalString(value, "accountId"); + if (accountId) record.accountId = accountId; + const model = optionalString(value, "model"); + if (model) record.model = model; + if ( + typeof value.workspaceRepoPath === "string" || + value.workspaceRepoPath === null + ) { + record.workspaceRepoPath = value.workspaceRepoPath; + } + return record; +} + +function sanitizePlaneCursor( + value: unknown +): ConversationPlaneReadCursor | null { + if (!isObject(value) || !validPlaneSeq(value.readThroughPlaneSeq)) { + return null; + } + return { + readThroughPlaneSeq: value.readThroughPlaneSeq, + updatedAt: optionalString(value, "updatedAt") ?? UNKNOWN_UPDATED_AT, + }; +} + function normalizeExecutionKey(key: string): string | null { try { const parsed = JSON.parse(key) as unknown; @@ -108,33 +199,54 @@ function readEnvelope( const parsed = readJson(backing, entryStorageKey(executionKey)); if (!isObject(parsed) || parsed.version !== STORE_VERSION) return null; const runners = sanitizeRunners(parsed.runners); + const continuation = sanitizeContinuation(parsed.continuation); + const ownerPlaneCursor = sanitizePlaneCursor(parsed.ownerPlaneCursor); return { version: STORE_VERSION, ...(runners ? { runners } : {}), + ...(continuation ? { continuation } : {}), + ...(ownerPlaneCursor ? { ownerPlaneCursor } : {}), }; } -function writeRunners( +function hasExecutionData(envelope: ConversationExecutionEnvelope): boolean { + return Boolean( + envelope.runners || envelope.continuation || envelope.ownerPlaneCursor + ); +} + +function writeEnvelope( backing: Storage | null, executionKey: string, - runners: ConversationRunnerRegistryEntry | null + envelope: ConversationExecutionEnvelope ): void { if (!backing) return; try { - if (!runners) { + if (!hasExecutionData(envelope)) { backing.removeItem(entryStorageKey(executionKey)); return; } - const envelope: ConversationExecutionEnvelope = { - version: STORE_VERSION, - runners, - }; backing.setItem(entryStorageKey(executionKey), JSON.stringify(envelope)); } catch { - // Best-effort: losing this registry only makes a plumbing row visible. + // Best-effort: a failed local persistence write rolls through recovery. } } +function mutateEnvelope( + backing: Storage | null, + key: string, + mutate: (envelope: ConversationExecutionEnvelope) => boolean +): void { + const normalized = normalizeExecutionKey(key); + if (!normalized) + throw new Error(`invalid conversation execution key: ${key}`); + const envelope = readEnvelope(backing, normalized) ?? { + version: STORE_VERSION, + }; + if (!mutate(envelope)) return; + writeEnvelope(backing, normalized, envelope); +} + function allStorageKeys(backing: Storage | null): string[] { if (!backing) return []; const keys: string[] = []; @@ -169,6 +281,181 @@ export function cloudConversationExecutorScopeKey( ]); } +/** Execution setup is shared by every surface targeting this agent/root. */ +export function cloudConversationSetupMemoryKey( + authIdentity: string, + cloudOrgId: string, + rootSessionId: string, + assignedAgentDefinitionId?: string +): string { + return JSON.stringify([ + "cloud-conversation-setup", + authIdentity, + cloudOrgId, + rootSessionId, + assignedAgentDefinitionId ?? "unassigned", + ]); +} + +function keyFor(executorScope: string, rootSessionId: string): string { + return conversationExecutionKey(executorScope, rootSessionId); +} + +export function loadStoredContinuation( + executorScope: string, + rootSessionId: string, + backing: Storage | null = defaultStorage() +): ConversationContinuationRecord | null { + const key = keyFor(executorScope, rootSessionId); + return readEnvelope(backing, key)?.continuation ?? null; +} + +function assertValidContinuationRecord( + record: Omit +): void { + if ( + !validPlaneSeq(record.readThroughPlaneSeq) || + !record.continuationSessionId || + !record.agentDefinitionId || + (!record.established && !record.bootstrapTurnIntentId) || + (record.established && record.bootstrapTurnIntentId) + ) { + throw new Error("invalid conversation continuation record"); + } +} + +export function saveStoredContinuation( + executorScope: string, + rootSessionId: string, + record: Omit, + backing: Storage | null = defaultStorage() +): void { + assertValidContinuationRecord(record); + mutateEnvelope(backing, keyFor(executorScope, rootSessionId), (envelope) => { + envelope.continuation = { + ...record, + updatedAt: new Date().toISOString(), + }; + return true; + }); +} + +/** Atomically hide a newly created runner and install its continuation. */ +export function prepareStoredContinuation( + executorScope: string, + rootSessionId: string, + record: Omit, + preparedAt: string, + backing: Storage | null = defaultStorage() +): void { + assertValidContinuationRecord(record); + const key = keyFor(executorScope, rootSessionId); + mutateEnvelope(backing, key, (envelope) => { + const current = envelope.runners; + envelope.runners = { + runnerSessionIds: [ + ...new Set([ + ...(current?.runnerSessionIds ?? []), + record.continuationSessionId, + ]), + ], + terminalRunnerSessionIds: current?.terminalRunnerSessionIds ?? [], + updatedAt: preparedAt, + }; + envelope.continuation = { ...record, updatedAt: preparedAt }; + return true; + }); +} + +export function clearStoredContinuation( + executorScope: string, + rootSessionId: string, + backing: Storage | null = defaultStorage() +): void { + mutateEnvelope(backing, keyFor(executorScope, rootSessionId), (envelope) => { + if (!envelope.continuation) return false; + delete envelope.continuation; + return true; + }); +} + +/** Fence bootstrap completion to the exact local runner and logical intent. */ +export function markStoredContinuationEstablished( + executorScope: string, + rootSessionId: string, + continuationSessionId: string, + bootstrapTurnIntentId: string, + backing: Storage | null = defaultStorage() +): boolean { + let established = false; + mutateEnvelope(backing, keyFor(executorScope, rootSessionId), (envelope) => { + const continuation = envelope.continuation; + if ( + !continuation || + continuation.continuationSessionId !== continuationSessionId || + continuation.established || + continuation.bootstrapTurnIntentId !== bootstrapTurnIntentId + ) { + return false; + } + continuation.established = true; + delete continuation.bootstrapTurnIntentId; + continuation.updatedAt = new Date().toISOString(); + established = true; + return true; + }); + return established; +} + +export function advanceStoredContinuationReadThrough( + executorScope: string, + rootSessionId: string, + planeSeq: number, + backing: Storage | null = defaultStorage() +): void { + if (!validPlaneSeq(planeSeq)) { + throw new Error(`invalid conversation plane seq: ${planeSeq}`); + } + mutateEnvelope(backing, keyFor(executorScope, rootSessionId), (envelope) => { + const continuation = envelope.continuation; + if (!continuation || continuation.readThroughPlaneSeq >= planeSeq) { + return false; + } + continuation.readThroughPlaneSeq = planeSeq; + continuation.updatedAt = new Date().toISOString(); + return true; + }); +} + +export function loadStoredOwnerPlaneCursor( + executorScope: string, + rootSessionId: string, + backing: Storage | null = defaultStorage() +): ConversationPlaneReadCursor | null { + const key = keyFor(executorScope, rootSessionId); + return readEnvelope(backing, key)?.ownerPlaneCursor ?? null; +} + +export function advanceStoredOwnerPlaneCursor( + executorScope: string, + rootSessionId: string, + planeSeq: number, + backing: Storage | null = defaultStorage() +): void { + if (!validPlaneSeq(planeSeq)) { + throw new Error(`invalid conversation plane seq: ${planeSeq}`); + } + mutateEnvelope(backing, keyFor(executorScope, rootSessionId), (envelope) => { + const current = envelope.ownerPlaneCursor?.readThroughPlaneSeq ?? 0; + if (current >= planeSeq) return false; + envelope.ownerPlaneCursor = { + readThroughPlaneSeq: planeSeq, + updatedAt: new Date().toISOString(), + }; + return true; + }); +} + export function loadStoredRunnerRegistryEntry( key: string, backing: Storage | null = defaultStorage() @@ -185,16 +472,16 @@ export function registerStoredRunner( updatedAt: string, backing: Storage | null = defaultStorage() ): void { - const normalized = normalizeExecutionKey(key); - if (!normalized) - throw new Error(`invalid conversation execution key: ${key}`); - const current = readEnvelope(backing, normalized)?.runners; - writeRunners(backing, normalized, { - runnerSessionIds: [ - ...new Set([...(current?.runnerSessionIds ?? []), runnerSessionId]), - ], - terminalRunnerSessionIds: current?.terminalRunnerSessionIds ?? [], - updatedAt, + mutateEnvelope(backing, key, (envelope) => { + const current = envelope.runners; + envelope.runners = { + runnerSessionIds: [ + ...new Set([...(current?.runnerSessionIds ?? []), runnerSessionId]), + ], + terminalRunnerSessionIds: current?.terminalRunnerSessionIds ?? [], + updatedAt, + }; + return true; }); } @@ -205,14 +492,17 @@ export function markStoredRunnerTerminal( ): void { const normalized = normalizeExecutionKey(key); if (!normalized) return; - const current = readEnvelope(backing, normalized)?.runners; - if (!current?.runnerSessionIds.includes(runnerSessionId)) return; - writeRunners(backing, normalized, { - ...current, - terminalRunnerSessionIds: [ - ...new Set([...current.terminalRunnerSessionIds, runnerSessionId]), - ], - updatedAt: new Date().toISOString(), + mutateEnvelope(backing, normalized, (envelope) => { + const current = envelope.runners; + if (!current?.runnerSessionIds.includes(runnerSessionId)) return false; + envelope.runners = { + ...current, + terminalRunnerSessionIds: [ + ...new Set([...current.terminalRunnerSessionIds, runnerSessionId]), + ], + updatedAt: new Date().toISOString(), + }; + return true; }); } @@ -244,24 +534,35 @@ export function forgetStoredRunner( for (const storageKey of allStorageKeys(backing)) { const executionKey = executionKeyFromStorageKey(storageKey); if (!executionKey) continue; - const current = readEnvelope(backing, executionKey)?.runners; - if (!current?.runnerSessionIds.includes(runnerSessionId)) continue; - const runnerSessionIds = current.runnerSessionIds.filter( - (sessionId) => sessionId !== runnerSessionId - ); - writeRunners( - backing, - executionKey, - runnerSessionIds.length > 0 - ? { - ...current, - runnerSessionIds, - terminalRunnerSessionIds: current.terminalRunnerSessionIds.filter( - (sessionId) => sessionId !== runnerSessionId - ), - } - : null + const envelope = readEnvelope(backing, executionKey); + const current = envelope?.runners; + if (!envelope) continue; + const removesRunner = Boolean( + current?.runnerSessionIds.includes(runnerSessionId) ); + const removesContinuation = + envelope.continuation?.continuationSessionId === runnerSessionId; + if (!removesRunner && !removesContinuation) continue; + if (current && removesRunner) { + const runnerSessionIds = current.runnerSessionIds.filter( + (sessionId) => sessionId !== runnerSessionId + ); + if (runnerSessionIds.length === 0) { + delete envelope.runners; + } else { + envelope.runners = { + ...current, + runnerSessionIds, + terminalRunnerSessionIds: current.terminalRunnerSessionIds.filter( + (sessionId) => sessionId !== runnerSessionId + ), + }; + } + } + if (removesContinuation) { + delete envelope.continuation; + } + writeEnvelope(backing, executionKey, envelope); } if (!backing) return; diff --git a/src/features/Org2Cloud/SessionConversation/conversationOwnerPublisher.ts b/src/features/Org2Cloud/SessionConversation/conversationOwnerPublisher.ts index 65496c9fe4..016de33646 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationOwnerPublisher.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationOwnerPublisher.ts @@ -1,7 +1,7 @@ /** * Owner publisher — the owner's half of "every turn is on the plane". * - * A member's turn reaches the plane through its one-shot runner; the + * A member's turn reaches the plane through its local continuation; the * owner's turn runs in the owner's own session and used to reach other * clients only through the session replay (slow, and ordered by sender * clock against the plane). This publishes the owner's turn to the plane @@ -14,40 +14,29 @@ * lets every client fold the plane rows onto their local twins instead of * rendering a second copy. */ -import { - getLastTurnTerminal, - getTurnPhase, - turnLifecycleSignalAtom, -} from "@src/engines/SessionCore/control/turnLifecycle"; import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; import { extractChatEvents } from "@src/engines/SessionCore/core/store/useSessionEvents"; import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import { waitForTurnIntentOutcome } from "@src/engines/SessionCore/services/TurnDispatchService"; import { createLogger } from "@src/hooks/logger"; -import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; import { boundConversationEventForPush, pushConversationEventsChunked, } from "../org2CloudConversationEventsClient"; -import { conversationEventKey } from "./conversationTimeline"; +import { advanceStoredOwnerPlaneCursor } from "./conversationExecutionStore"; +import { + buildConversationPlaneUserEvent, + findUserEventByIntent, + sliceTurnTailByIntent, + turnIntentIdOf, +} from "./conversationTurnEvents"; const log = createLogger("ConversationOwnerPublisher"); const TURN_DEADLINE_MS = 15 * 60_000; -export function turnIntentIdOf(event: SessionEvent): string | null { - if (event.source !== "user") return null; - const intent = (event.result as { turnIntentId?: unknown } | undefined) - ?.turnIntentId; - return typeof intent === "string" && intent.length > 0 ? intent : null; -} - -export function findUserEventByIntent( - events: readonly SessionEvent[], - turnIntentId: string -): SessionEvent | null { - return events.find((event) => turnIntentIdOf(event) === turnIntentId) ?? null; -} +export { findUserEventByIntent }; /** * The clean user row for the plane: the user's visible words only (the @@ -59,27 +48,15 @@ export function buildOwnerUserRow( displayText: string ): SessionEvent { const turnIntentId = turnIntentIdOf(userEvent); - return { + if (!turnIntentId) { + throw new Error("owner user row is missing its turn intent"); + } + return buildConversationPlaneUserEvent({ id: userEvent.id, - chunk_id: userEvent.id, - sessionId: "conversation", createdAt: userEvent.createdAt, - functionName: "user_message", - uiCanonical: "user_message", - actionType: "raw", - args: {}, - result: { - type: "user", - message: { content: displayText, role: "user" }, - ...(turnIntentId ? { turnIntentId } : {}), - }, - source: "user", displayText, - displayStatus: "completed", - displayVariant: "message", - activityStatus: "agent", - payloadRefs: [], - } as SessionEvent; + turnIntentId, + }); } /** @@ -87,26 +64,7 @@ export function buildOwnerUserRow( * row, up to the next turn's user row. `null` when the user row is not in * the transcript (the dispatch failed and removed it). */ -export function sliceOwnerTurnTail( - events: readonly SessionEvent[], - turnIntentId: string -): SessionEvent[] | null { - const start = events.findIndex( - (event) => turnIntentIdOf(event) === turnIntentId - ); - if (start < 0) return null; - const turnKey = conversationEventKey(events[start]); - const tail: SessionEvent[] = []; - for (let index = start + 1; index < events.length; index += 1) { - const event = events[index]; - if (event.source === "user") { - if (conversationEventKey(event) !== turnKey) break; - continue; - } - tail.push(event); - } - return tail; -} +export const sliceOwnerTurnTail = sliceTurnTailByIntent; function waitForUserEvent( sessionId: string, @@ -142,38 +100,6 @@ function waitForUserEvent( }); } -function waitForTurnEnd( - sessionId: string, - userEventMs: number, - deadlineMs: number -): Promise { - const store = getInstrumentedStore(); - const isDone = (): boolean => - getTurnPhase(sessionId) === "idle" && - (getLastTurnTerminal(sessionId)?.at ?? 0) >= userEventMs; - if (isDone()) return Promise.resolve(); - return new Promise((resolve, reject) => { - const remainingMs = deadlineMs - Date.now(); - if (remainingMs <= 0) { - reject(new Error("owner turn timed out")); - return; - } - let unsubscribe: (() => void) | null = null; - const timer = setTimeout(() => { - unsubscribe?.(); - reject(new Error("owner turn timed out")); - }, remainingMs); - const check = (): void => { - if (!isDone()) return; - clearTimeout(timer); - unsubscribe?.(); - resolve(); - }; - unsubscribe = store.sub(turnLifecycleSignalAtom, check); - check(); - }); -} - export interface PublishOwnerTurnParams { /** Resolved before every push — a long turn outlives a captured token. */ getAccessToken: () => Promise; @@ -184,6 +110,9 @@ export interface PublishOwnerTurnParams { /** The intent id the dispatch was minted with; keys the local user row. */ turnIntentId: string; displayText: string; + executorScope: string; + /** Plane head fetched and injected immediately before this dispatch. */ + readThroughPlaneSeq: number; /** Fires after each successful push (signal-bump hook). */ onPushed?: () => void; } @@ -197,7 +126,7 @@ export async function publishOwnerTurn( params: PublishOwnerTurnParams ): Promise { const deadlineMs = Date.now() + TURN_DEADLINE_MS; - const turnId = crypto.randomUUID(); + const turnId = params.turnIntentId; const userEvent = await waitForUserEvent( params.sessionId, params.turnIntentId, @@ -215,16 +144,21 @@ export async function publishOwnerTurn( }); params.onPushed?.(); - const userEventMs = new Date(userEvent.createdAt).getTime(); - await waitForTurnEnd( - params.sessionId, - Number.isFinite(userEventMs) ? userEventMs : 0, + const outcome = await waitForTurnIntentOutcome( + params.turnIntentId, deadlineMs ); + if (outcome.status === "completed") { + advanceStoredOwnerPlaneCursor( + params.executorScope, + params.rootSessionId, + params.readThroughPlaneSeq + ); + } const persisted = await eventStoreProxy .getPersistedEvents(params.sessionId) .catch(() => [] as SessionEvent[]); - const tail = sliceOwnerTurnTail(persisted, params.turnIntentId) ?? []; + const tail = sliceTurnTailByIntent(persisted, params.turnIntentId) ?? []; if (tail.length > 0) { await pushConversationEventsChunked(await params.getAccessToken(), { orgId: params.orgId, diff --git a/src/features/Org2Cloud/SessionConversation/conversationRunnerScope.tsx b/src/features/Org2Cloud/SessionConversation/conversationRunnerScope.tsx index a68844a0b6..739aaa1379 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationRunnerScope.tsx +++ b/src/features/Org2Cloud/SessionConversation/conversationRunnerScope.tsx @@ -1,7 +1,7 @@ /** * The live runner scope for a mounted conversation surface. * - * A member's turn runs in an invisible one-shot local runner, so the mounted + * A member's turn runs in an invisible persistent local runner, so the mounted * imported session stays idle — its planning indicator and streaming-delta * footer never light up, and a long turn looks frozen (no "Thinking…", no * activity) until the tail lands. The conversation stream publishes the diff --git a/src/features/Org2Cloud/SessionConversation/conversationRunnerSessions.ts b/src/features/Org2Cloud/SessionConversation/conversationRunnerSessions.ts index c662cd2d31..4db84af3a9 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationRunnerSessions.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationRunnerSessions.ts @@ -1,11 +1,25 @@ -/** Local registry facade for invisible shared-conversation runners. */ +/** Local lifecycle facade for invisible shared-conversation runners. */ +import { deleteSession } from "@src/api/tauri/agent"; +import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; +import { clearCliTurnLifecycleSession } from "@src/hooks/cliSession/cliTurnLifecycleCoordinator"; +import { createLogger } from "@src/hooks/logger"; +import { removeSession, sessionsAtom } from "@src/store/session"; +import { persistSessions } from "@src/store/session/sessionAtom/persistence"; +import { isTerminalStatus } from "@src/types/session/session"; +import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; +import { invokeTauri } from "@src/util/platform/tauri/init"; +import { isCliSession } from "@src/util/session/sessionDispatch"; + import { collectStoredRunnerSessionIds, conversationExecutionKey, + forgetStoredRunner, + loadStoredRunnerRegistryEntry, markStoredRunnerTerminal, - registerStoredRunner, } from "./conversationExecutionStore"; +const log = createLogger("ConversationRunnerSessions"); + export function conversationRunnerKey( executorScope: string, rootSessionId: string @@ -18,17 +32,68 @@ export function collectConversationRunnerSessionIds(): Set { return collectStoredRunnerSessionIds(); } -export function registerConversationRunner( - key: string, - runnerSessionId: string, - updatedAt: string -): void { - registerStoredRunner(key, runnerSessionId, updatedAt); -} - export function markConversationRunnerTerminal( key: string, runnerSessionId: string ): void { markStoredRunnerTerminal(key, runnerSessionId); } + +/** Delete a hidden runner through the same category-specific paths as UI. */ +export async function cleanupConversationRunnerSession( + runnerSessionId: string +): Promise { + let deletedSessionIds = [runnerSessionId]; + if (isCliSession(runnerSessionId)) { + await invokeTauri("cli_agent_delete", { sessionId: runnerSessionId }); + clearCliTurnLifecycleSession(runnerSessionId); + } else { + const receipt = await deleteSession(runnerSessionId); + if (receipt.deletedSessionIds.length > 0) { + deletedSessionIds = receipt.deletedSessionIds; + } + } + await Promise.all( + deletedSessionIds.map((sessionId) => + eventStoreProxy.evictSession(sessionId).catch(() => undefined) + ) + ); + for (const sessionId of deletedSessionIds) { + removeSession(sessionId); + forgetStoredRunner(sessionId); + } + const store = getInstrumentedStore(); + persistSessions(store.get(sessionsAtom)); +} + +export async function cleanupConversationRunnerBestEffort( + runnerSessionId: string +): Promise { + try { + await cleanupConversationRunnerSession(runnerSessionId); + } catch (error) { + log.warn(`hidden runner cleanup failed for ${runnerSessionId}`, error); + } +} + +/** Sweep only sessions proven terminal; retain the active continuation. */ +export async function cleanupRetiredConversationRunners( + key: string, + keepSessionId: string +): Promise { + const entry = loadStoredRunnerRegistryEntry(key); + if (!entry) return; + const terminalIds = new Set(entry.terminalRunnerSessionIds); + for (const session of getInstrumentedStore().get(sessionsAtom)) { + if ( + entry.runnerSessionIds.includes(session.session_id) && + isTerminalStatus(session.status) + ) { + terminalIds.add(session.session_id); + } + } + terminalIds.delete(keepSessionId); + for (const sessionId of terminalIds) { + await cleanupConversationRunnerBestEffort(sessionId); + } +} diff --git a/src/features/Org2Cloud/SessionConversation/conversationTimeline.ts b/src/features/Org2Cloud/SessionConversation/conversationTimeline.ts index bf4df73798..90b63acd5e 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationTimeline.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationTimeline.ts @@ -18,6 +18,7 @@ import { sourceEventIdOf, } from "./continuationEvents"; import { buildConversationPlaneStreamEvents } from "./conversationPlaneEvents"; +import { turnIntentIdOf } from "./conversationTurnEvents"; /** * Plane identity of an event. User rows match on the turn-intent id so the @@ -27,11 +28,8 @@ import { buildConversationPlaneStreamEvents } from "./conversationPlaneEvents"; */ export function conversationEventKey(event: SessionEvent): string { if (event.source === "user") { - const intent = (event.result as { turnIntentId?: unknown } | undefined) - ?.turnIntentId; - if (typeof intent === "string" && intent.length > 0) { - return `intent:${intent}`; - } + const intent = turnIntentIdOf(event); + if (intent) return `intent:${intent}`; } return `event:${sourceEventIdOf(event)}`; } diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts new file mode 100644 index 0000000000..4a46b95882 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnEvents.ts @@ -0,0 +1,71 @@ +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; + +/** Canonical clean user row published to the shared conversation plane. */ +export function buildConversationPlaneUserEvent(input: { + id: string; + createdAt: string; + displayText: string; + turnIntentId: string; +}): SessionEvent { + return { + id: input.id, + chunk_id: input.id, + sessionId: "conversation", + createdAt: input.createdAt, + functionName: "user_message", + uiCanonical: "user_message", + actionType: "raw", + args: {}, + result: { + type: "user", + message: { content: input.displayText, role: "user" }, + turnIntentId: input.turnIntentId, + }, + source: "user", + displayText: input.displayText, + displayStatus: "completed", + displayVariant: "message", + activityStatus: "agent", + payloadRefs: [], + } as SessionEvent; +} + +export function turnIntentIdOf(event: SessionEvent): string | null { + if (event.source !== "user") return null; + const intent = (event.result as { turnIntentId?: unknown } | undefined) + ?.turnIntentId; + return typeof intent === "string" && intent.length > 0 ? intent : null; +} + +export function findUserEventByIntent( + events: readonly SessionEvent[], + turnIntentId: string +): SessionEvent | null { + return events.find((event) => turnIntentIdOf(event) === turnIntentId) ?? null; +} + +/** + * Non-user events belonging to one exact runtime turn. Duplicate frontend + * and backend user rows for the same intent are skipped; the next distinct + * user intent closes the slice. + */ +export function sliceTurnTailByIntent( + events: readonly SessionEvent[], + turnIntentId: string +): SessionEvent[] | null { + const start = events.findIndex( + (event) => turnIntentIdOf(event) === turnIntentId + ); + if (start < 0) return null; + + const tail: SessionEvent[] = []; + for (let index = start + 1; index < events.length; index += 1) { + const event = events[index]; + if (event.source === "user") { + if (turnIntentIdOf(event) !== turnIntentId) break; + continue; + } + tail.push(event); + } + return tail; +} diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts new file mode 100644 index 0000000000..615d751f04 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.test.ts @@ -0,0 +1,383 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +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 type { CloudConversationEvent } from "../org2CloudConversationEventsClient"; +import { loadContinuation, saveContinuation } from "./conversationContinuation"; +import { + CONVERSATION_TURN_LOCK_UNAVAILABLE, + type RunConversationTurnParams, + runConversationTurn, + withConversationTurnLock, +} from "./conversationTurnRunner"; + +const { state } = vi.hoisted(() => ({ + state: { + generation: 0, + persistedBatches: [] as SessionEvent[][], + pushes: [] as Array<{ kind: "user" | "tail"; turnId: string }>, + sent: [] as Record[], + rejectNextSend: false, + cleaned: [] as string[], + terminalStatus: "completed" as "completed" | "failed" | "cancelled", + }, +})); + +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/services/SessionService", () => ({ + SessionService: { + create: vi.fn(async () => ({ sessionId: "fresh-runner" })), + }, +})); +vi.mock("@src/engines/SessionCore/services/TurnDispatchService", () => ({ + reserveTurnDispatch: vi.fn( + (input: { sessionId: string; turnIntentId: string }) => ({ + ...input, + generation: ++state.generation, + optimisticSource: "dispatch", + }) + ), + sendReservedTurn: vi.fn(async (input: Record) => { + state.sent.push(input); + if (state.rejectNextSend) { + state.rejectNextSend = false; + throw new Error("session cannot accept turns"); + } + return { ...(input.dispatch as object), accepted: true }; + }), + waitForTurnOutcome: vi.fn(async (dispatch: Record) => ({ + ...dispatch, + status: state.terminalStatus, + at: 1, + })), +})); +vi.mock("@src/engines/SessionCore/sync/adapters/shared/eventFactories", () => ({ + mintTurnIntentId: () => "intent-1", +})); +vi.mock("@src/features/TeamCollaboration/forkSession", () => ({ + requestForkSessionSetup: vi.fn(async () => ({ + workspaceRepoPath: "/repo", + execution: { + agentDefinitionId: "agent-a", + accountId: "account-a", + model: "model-a", + }, + })), +})); +vi.mock("@src/features/TeamCollaboration/forkSetupMemory", () => ({ + loadForkSetupMemory: vi.fn(() => null), + saveForkSetupMemory: vi.fn(), + clearForkSetupMemory: vi.fn(), +})); +vi.mock("@src/hooks/logger", () => ({ + createLogger: () => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }), +})); +vi.mock("@src/i18n", () => ({ default: { t: (key: string) => key } })); +vi.mock("./conversationRunnerSessions", () => ({ + conversationRunnerKey: (scope: string, root: string) => + JSON.stringify([scope, root]), + registerConversationRunner: vi.fn(), + markConversationRunnerTerminal: vi.fn(), + cleanupRetiredConversationRunners: vi.fn(async () => undefined), + cleanupConversationRunnerBestEffort: vi.fn(async (sessionId: string) => { + state.cleaned.push(sessionId); + }), +})); +vi.mock("../org2CloudConversationEventsClient", () => ({ + boundConversationEventForPush: (event: SessionEvent) => event, + pushConversationEvents: vi.fn( + async (_token: string, input: { turnId: string }) => { + state.pushes.push({ kind: "user", turnId: input.turnId }); + return { firstSeq: 1, lastSeq: 1 }; + } + ), + pushConversationEventsChunked: vi.fn( + async (_token: string, input: { turnId: string }) => { + state.pushes.push({ kind: "tail", turnId: input.turnId }); + return { firstSeq: 2, lastSeq: 2 }; + } + ), +})); + +function fakeStorage(): Storage { + const values = new Map(); + return { + get length() { + return values.size; + }, + clear: () => values.clear(), + getItem: (key) => values.get(key) ?? null, + key: (index) => [...values.keys()][index] ?? null, + removeItem: (key) => { + values.delete(key); + }, + setItem: (key, value) => { + values.set(key, value); + }, + }; +} + +function event( + id: string, + source: "user" | "assistant", + text: string, + turnIntentId?: string +): SessionEvent { + return { + id, + chunk_id: id, + sessionId: "runner", + createdAt: "2026-08-25T00:00:00.000Z", + source, + displayText: text, + result: turnIntentId ? { turnIntentId } : {}, + } as SessionEvent; +} + +function row( + seq: number, + turnId: string, + authorDisplayName: string, + text: string +): CloudConversationEvent { + return { + id: `plane-${seq}`, + rootSessionId: "root", + authorUserId: authorDisplayName.toLowerCase(), + authorDisplayName, + turnId, + seq, + event: event(`plane-event-${seq}`, "user", text, turnId), + createdAt: "2026-08-25T00:00:00.000Z", + }; +} + +function params( + overrides: Partial = {} +): RunConversationTurnParams { + return { + getAccessToken: async () => "token", + orgId: "org", + rootSessionId: "root", + conversationTitle: "Conversation", + displayText: "new request", + executionScopeKey: "scope", + loadInitialContext: async () => ({ + timeline: [event("history", "assistant", "root answer")], + readThroughPlaneSeq: 12, + }), + loadPlaneDelta: async (afterSeq) => ({ events: [], lastSeq: afterSeq }), + ...overrides, + }; +} + +let storageBackup: PropertyDescriptor | undefined; +let locksBackup: PropertyDescriptor | undefined; + +beforeEach(() => { + state.generation = 0; + state.persistedBatches = []; + state.pushes = []; + state.sent = []; + state.rejectNextSend = false; + state.cleaned = []; + state.terminalStatus = "completed"; + storageBackup = Object.getOwnPropertyDescriptor(globalThis, "localStorage"); + Object.defineProperty(globalThis, "localStorage", { + configurable: true, + value: fakeStorage(), + }); + locksBackup = Object.getOwnPropertyDescriptor(navigator, "locks"); + Object.defineProperty(navigator, "locks", { + configurable: true, + value: { + request: ( + _name: string, + _options: LockOptions, + callback: () => Promise + ) => callback(), + } as unknown as LockManager, + }); + vi.clearAllMocks(); +}); + +afterEach(() => { + if (storageBackup) { + Object.defineProperty(globalThis, "localStorage", storageBackup); + } else { + Reflect.deleteProperty(globalThis, "localStorage"); + } + if (locksBackup) { + Object.defineProperty(navigator, "locks", locksBackup); + } else { + Reflect.deleteProperty(navigator, "locks"); + } +}); + +describe("conversation turn continuation", () => { + it("fails closed in a browser without a cross-window lock", async () => { + Reflect.deleteProperty(navigator, "locks"); + const run = vi.fn(async () => "never"); + await expect(withConversationTurnLock("root", run)).rejects.toThrow( + CONVERSATION_TURN_LOCK_UNAVAILABLE + ); + expect(run).not.toHaveBeenCalled(); + }); + + it("creates one idle runner and persists the verified initial cursor", async () => { + state.persistedBatches = [ + [ + event("user-1", "user", "new request", "intent-1"), + event("agent-1", "assistant", "answer"), + ], + ]; + const onRunnerReady = vi.fn(); + + const result = await runConversationTurn(params({ onRunnerReady })); + + expect(result).toMatchObject({ + runnerSessionId: "fresh-runner", + turnIntentId: "intent-1", + terminalStatus: "completed", + }); + expect(SessionService.create).toHaveBeenCalledWith( + expect.objectContaining({ task: "" }) + ); + expect(sendReservedTurn).toHaveBeenCalledTimes(1); + expect(state.sent[0].content).toContain("Assistant: root answer"); + expect(onRunnerReady).toHaveBeenCalledWith( + "fresh-runner", + "intent-1", + "intent-1" + ); + expect(loadContinuation("scope", "root")).toMatchObject({ + continuationSessionId: "fresh-runner", + established: true, + readThroughPlaneSeq: 12, + }); + expect(state.pushes).toEqual([ + { kind: "user", turnId: "intent-1" }, + { kind: "tail", turnId: "intent-1" }, + ]); + }); + + it("reuses the runner, injects only new foreign rows, and advances the cursor", async () => { + saveContinuation("scope", "root", { + continuationSessionId: "runner-live", + readThroughPlaneSeq: 55, + established: true, + agentDefinitionId: "agent-a", + }); + state.persistedBatches = [ + [ + event("old-user", "user", "prior", "own-prior"), + event("old-agent", "assistant", "prior answer"), + ], + [ + event("old-user", "user", "prior", "own-prior"), + event("old-agent", "assistant", "prior answer"), + event("user-2", "user", "new request", "intent-2"), + event("agent-2", "assistant", "fresh answer"), + ], + ]; + const loadInitialContext = vi.fn(); + const loadPlaneDelta = vi.fn(async () => ({ + events: [ + row(56, "own-prior", "Vince", "duplicate local turn"), + row(57, "alice-turn", "Alice", "note from Alice"), + ], + lastSeq: 57, + })); + + const result = await runConversationTurn( + params({ + turnIntentId: "intent-2", + loadInitialContext, + loadPlaneDelta, + }) + ); + + expect(result.runnerSessionId).toBe("runner-live"); + expect(SessionService.create).not.toHaveBeenCalled(); + expect(loadInitialContext).not.toHaveBeenCalled(); + 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); + }); + + it("rolls a rejected resume to fresh without publishing the user twice", async () => { + saveContinuation("scope", "root", { + continuationSessionId: "runner-dead", + readThroughPlaneSeq: 10, + established: true, + agentDefinitionId: "agent-a", + }); + state.rejectNextSend = true; + state.persistedBatches = [ + [], + [ + event("user-1", "user", "new request", "intent-1"), + event("agent-1", "assistant", "recovered answer"), + ], + ]; + + const result = await runConversationTurn(params()); + + expect(result.runnerSessionId).toBe("fresh-runner"); + expect(sendReservedTurn).toHaveBeenCalledTimes(2); + expect(SessionService.create).toHaveBeenCalledTimes(1); + expect(state.pushes.filter((push) => push.kind === "user")).toHaveLength(1); + expect(state.cleaned).toContain("runner-dead"); + }); + + it("deletes an unestablished runner before accepting a different intent", async () => { + saveContinuation("scope", "root", { + continuationSessionId: "runner-pending", + readThroughPlaneSeq: 0, + established: false, + bootstrapTurnIntentId: "intent-old", + agentDefinitionId: "agent-a", + }); + state.persistedBatches = [ + [ + event("user-1", "user", "new request", "intent-new"), + event("agent-1", "assistant", "answer"), + ], + ]; + + await runConversationTurn(params({ turnIntentId: "intent-new" })); + + expect(state.cleaned).toContain("runner-pending"); + expect(SessionService.create).toHaveBeenCalledTimes(1); + expect(loadContinuation("scope", "root")).toMatchObject({ + continuationSessionId: "fresh-runner", + established: true, + }); + }); + + it("clears and cleans a failed execution episode", async () => { + state.terminalStatus = "failed"; + state.persistedBatches = [ + [ + event("user-1", "user", "new request", "intent-1"), + event("agent-1", "assistant", "partial answer"), + ], + ]; + + const result = await runConversationTurn(params()); + + expect(result.terminalStatus).toBe("failed"); + expect(loadContinuation("scope", "root")).toBeNull(); + expect(state.cleaned).toContain("fresh-runner"); + }); +}); diff --git a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts index de83b057ec..d03fbadf46 100644 --- a/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts +++ b/src/features/Org2Cloud/SessionConversation/conversationTurnRunner.ts @@ -3,31 +3,30 @@ * plane (design: docs/conversation-events-plane-design-2026-08-21.md). * * When a member chats in a conversation they do not own, the turn executes - * in a LOCAL, invisible one-shot runner session on their machine + * in a LOCAL, invisible continuation session on their machine * (sender-runs / sender-pays) and the resulting events are pushed — * author-stamped — to the shared plane. No fork, no transcript copy, no new * sidebar entity. * - * ONE-SHOT per turn: `SessionService.create` is the only dispatch primitive - * proven headless (Routine/work-item background runs ride it), so every - * turn gets a fresh runner with the full bounded conversation context - * injected (the external-history handoff pattern) — never a dispatch into - * an unmounted surface. Runner sessions are plumbing: the caller forces - * their cloud sync OFF, and `collectConversationRunnerSessionIds` hides - * them from My Sessions. + * The first turn prepares an idle local runner and dispatches through the + * canonical turn boundary. Later turns reuse that same session and inject + * only the plane delta after its monotonic read cursor. A failed episode or + * assigned-agent change rolls to a fresh runner. * * Push order is Slack-shaped: the user's message row goes out FIRST (every * client sees it instantly), the agent tail follows under the same turnId * when the local run completes. */ import Message from "@src/components/Message"; -import { - getLastTurnTerminal, - turnLifecycleSignalAtom, -} from "@src/engines/SessionCore/control/turnLifecycle"; +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 { + reserveTurnDispatch, + sendReservedTurn, + waitForTurnOutcome, +} from "@src/engines/SessionCore/services/TurnDispatchService"; import { mintTurnIntentId } from "@src/engines/SessionCore/sync/adapters/shared/eventFactories"; import { requestForkSessionSetup } from "@src/features/TeamCollaboration/forkSession"; import { @@ -37,50 +36,82 @@ import { } from "@src/features/TeamCollaboration/forkSetupMemory"; import { createLogger } from "@src/hooks/logger"; import i18n from "@src/i18n"; -import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; import { + type CloudConversationEvent, + type ConversationEventWindow, boundConversationEventForPush, pushConversationEvents, pushConversationEventsChunked, } from "../org2CloudConversationEventsClient"; import { + advanceContinuationReadThrough, + clearContinuation, + decideContinuation, + loadContinuation, + markContinuationEstablished, + prepareContinuation, +} from "./conversationContinuation"; +import { + cleanupConversationRunnerBestEffort, + cleanupRetiredConversationRunners, conversationRunnerKey, markConversationRunnerTerminal, - registerConversationRunner, } from "./conversationRunnerSessions"; +import { + buildConversationPlaneUserEvent, + sliceTurnTailByIntent, + turnIntentIdOf, +} from "./conversationTurnEvents"; const log = createLogger("ConversationTurnRunner"); const TURN_DEADLINE_MS = 15 * 60_000; -const CONTEXT_MAX_ENTRIES = 60; +export const CONVERSATION_CONTEXT_MAX_ENTRIES = 60; +export const CONVERSATION_TURN_LOCK_UNAVAILABLE = + "ORG2_CONVERSATION_TURN_LOCK_UNAVAILABLE"; const CONTEXT_MAX_ENTRY_CHARS = 600; const CONTEXT_MAX_TOTAL_CHARS = 18_000; -/** Conversation timeline rendered as a bounded read-only context block. */ -export function renderConversationContext( - timeline: readonly SessionEvent[], - senders?: ReadonlyMap -): string { - const tail = timeline.slice(-CONTEXT_MAX_ENTRIES); +interface SharedContextEntry { + speaker: string; + text?: string; +} + +function renderSharedContext(entries: readonly SharedContextEntry[]): string { const lines: string[] = []; let total = 0; - for (const event of tail) { - const text = event.displayText?.trim(); + for (const entry of entries + .slice(-CONVERSATION_CONTEXT_MAX_ENTRIES) + .reverse()) { + const text = entry.text?.trim(); if (!text) continue; - const speaker = - event.source === "user" - ? (senders?.get(event.id) ?? "User") - : "Assistant"; - let line = `${speaker}: ${text.replace(/\s+/g, " ")}`; + let line = `${entry.speaker}: ${text.replace(/\s+/g, " ")}`; if (line.length > CONTEXT_MAX_ENTRY_CHARS) { - line = `${line.slice(0, CONTEXT_MAX_ENTRY_CHARS)}…`; + line = `${line.slice(0, CONTEXT_MAX_ENTRY_CHARS - 1)}…`; } - if (total + line.length > CONTEXT_MAX_TOTAL_CHARS) break; - total += line.length; + const addedLength = line.length + (lines.length > 0 ? 1 : 0); + if (total + addedLength > CONTEXT_MAX_TOTAL_CHARS) break; + total += addedLength; lines.push(line); } - return lines.join("\n"); + return lines.reverse().join("\n"); +} + +/** Conversation timeline rendered as a bounded read-only context block. */ +export function renderConversationContext( + timeline: readonly SessionEvent[], + senders?: ReadonlyMap +): string { + return renderSharedContext( + timeline.map((event) => ({ + speaker: + event.source === "user" + ? (senders?.get(event.id) ?? "User") + : "Assistant", + text: event.displayText, + })) + ); } export function buildRunnerPrompt( @@ -102,33 +133,38 @@ export function buildRunnerPrompt( ].join("\n"); } -async function waitForFirstTurnTerminal( - sessionId: string, - deadlineMs: number -): Promise { - const store = getInstrumentedStore(); - const isComplete = (): boolean => getLastTurnTerminal(sessionId) !== null; - if (isComplete()) return; - await new Promise((resolve, reject) => { - const remainingMs = deadlineMs - Date.now(); - if (remainingMs <= 0) { - reject(new Error("conversation turn timed out")); - return; - } - let unsubscribe: (() => void) | null = null; - const timer = setTimeout(() => { - unsubscribe?.(); - reject(new Error("conversation turn timed out")); - }, remainingMs); - const check = (): void => { - if (!isComplete()) return; - clearTimeout(timer); - unsubscribe?.(); - resolve(); - }; - unsubscribe = store.sub(turnLifecycleSignalAtom, check); - check(); - }); +export function renderPlaneDeltaContext( + rows: readonly CloudConversationEvent[] +): string { + return renderSharedContext( + rows.map((row) => ({ + speaker: + row.authorDisplayName ?? + (row.event.source === "user" ? "User" : "Assistant"), + text: row.event.displayText, + })) + ); +} + +export function buildResumePrompt(deltaBlock: string, request: string): string { + if (!deltaBlock) return request; + return [ + "New activity in the SHARED team conversation since your last turn —", + "read-only context from the other participants' machines:", + "", + "=== Shared conversation update ===", + deltaBlock, + "=== End of update ===", + "", + "Continue the conversation by handling this request:", + request, + ].join("\n"); +} + +export interface ConversationInitialContext { + timeline: readonly SessionEvent[]; + senders?: ReadonlyMap; + readThroughPlaneSeq: number; } /** @@ -136,31 +172,6 @@ async function waitForFirstTurnTerminal( * runner's own persisted user event carries the injected context prefix, * which must never leak into the shared conversation. */ -function buildPushedUserEvent( - sessionId: string, - displayText: string, - createdAt: string -): SessionEvent { - const id = `convturn-user-${mintTurnIntentId()}`; - return { - id, - chunk_id: id, - sessionId, - createdAt, - functionName: "user_message", - uiCanonical: "user_message", - actionType: "raw", - args: {}, - result: { type: "user", message: { content: displayText, role: "user" } }, - source: "user", - displayText, - displayStatus: "completed", - displayVariant: "message", - activityStatus: "agent", - payloadRefs: [], - } as SessionEvent; -} - export interface RunConversationTurnParams { /** * Resolved before EVERY push. A turn can outlive the access token that @@ -174,18 +185,28 @@ export interface RunConversationTurnParams { displayText: string; agentContent?: string; imageDataUrls?: string[]; - /** Merged conversation timeline for the read-only context prefix. */ - timeline: readonly SessionEvent[]; + /** Loaded only when a fresh execution episode needs a full context seed. */ + loadInitialContext: ( + excludeTurnIntentId: string + ) => Promise; + /** Plane rows after the continuation's exclusive read cursor. */ + loadPlaneDelta: (afterSeq: number) => Promise; sourceScopeKey?: string; sourceModel?: string; + assignedAgentDefinitionId?: string; + setupMemoryKey?: string; /** Account-and-org-scoped local executor identity. */ - executionScopeKey?: string; + executionScopeKey: string; + /** Stable logical id for durable redelivery; minted for ordinary chat. */ + turnIntentId?: string; /** - * Called as soon as the one-shot runner session id is known, with the - * turnId the tail will be pushed under. The caller overlays the runner's - * LIVE events until the plane carries this turnId. + * Called when the reusable runner and exact runtime intent are known. */ - onRunnerReady?: (runnerSessionId: string, turnId: string) => void; + onRunnerReady?: ( + runnerSessionId: string, + turnId: string, + turnIntentId: string + ) => void | Promise; /** * Fires after push #1 (the user's message row) lands on the plane — the * composer unblocks here; the agent tail streams in later under the same @@ -199,54 +220,326 @@ export interface RunConversationTurnParams { export interface RunConversationTurnResult { runnerSessionId: string; pushedEventCount: number; + turnIntentId: string; + terminalStatus: TurnTerminalStatus; +} + +interface TurnPushIo { + getAccessToken: () => Promise; + orgId: string; + rootSessionId: string; + turnId: string; + turnIntentId: string; + onPushed?: () => void; +} + +async function dispatchRunnerTurn( + params: RunConversationTurnParams, + io: TurnPushIo, + input: { + runnerSessionId: string; + content: string; + accountId?: string; + model?: string; + } +): Promise> { + const dispatch = reserveTurnDispatch({ + sessionId: input.runnerSessionId, + turnIntentId: io.turnIntentId, + optimisticSource: "dispatch", + }); + await sendReservedTurn({ + dispatch, + content: input.content, + displayText: params.displayText, + model: input.model, + accountId: input.accountId, + imageDataUrls: params.imageDataUrls, + clientMessageId: `conversation-turn:${io.turnIntentId}`, + turnIntentSource: "user_submit", + directUserIntent: true, + }); + return dispatch; +} + +async function pushUserRow( + io: TurnPushIo, + displayText: string, + dispatchIso: string +): Promise { + await pushConversationEvents(await io.getAccessToken(), { + orgId: io.orgId, + rootSessionId: io.rootSessionId, + turnId: io.turnId, + events: [ + boundConversationEventForPush( + buildConversationPlaneUserEvent({ + id: `convturn-user-${io.turnIntentId}`, + displayText, + createdAt: dispatchIso, + turnIntentId: io.turnIntentId, + }) + ), + ], + }); + io.onPushed?.(); +} + +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` + ); + } + 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, + }); + io.onPushed?.(); + return tail.length; +} + +function collectLocalTurnIntentIds( + events: readonly SessionEvent[] +): Set { + const ids = new Set(); + for (const event of events) { + const turnIntentId = turnIntentIdOf(event); + if (turnIntentId) ids.add(turnIntentId); + } + return ids; +} + +function assertAssignedAgent( + selectedAgentDefinitionId: string, + params: RunConversationTurnParams +): void { + if ( + params.assignedAgentDefinitionId && + params.assignedAgentDefinitionId !== selectedAgentDefinitionId + ) { + throw new Error( + `conversation requires agent ${params.assignedAgentDefinitionId}; ` + + `selected ${selectedAgentDefinitionId}` + ); + } +} + +const localTurnQueues = new Map>(); + +async function withLocalTurnQueue( + key: string, + run: () => Promise +): Promise { + const previous = localTurnQueues.get(key) ?? Promise.resolve(); + const next = previous.catch(() => undefined).then(run); + localTurnQueues.set(key, next); + try { + return await next; + } finally { + if (localTurnQueues.get(key) === next) localTurnQueues.delete(key); + } +} + +/** Serialize one continuation across windows; Node tests use the local queue. */ +export async function withConversationTurnLock( + key: string, + run: () => Promise +): Promise { + const locks = typeof navigator !== "undefined" ? navigator.locks : undefined; + if (locks) { + return locks.request( + `orgii:conversation-turn:${key}`, + { mode: "exclusive" }, + run + ); + } + if (typeof window !== "undefined") { + throw new Error(CONVERSATION_TURN_LOCK_UNAVAILABLE); + } + return withLocalTurnQueue(key, run); } export async function runConversationTurn( params: RunConversationTurnParams ): Promise { const key = conversationRunnerKey( - params.executionScopeKey ?? params.orgId, + params.executionScopeKey, + params.rootSessionId + ); + return withConversationTurnLock(key, () => + runConversationTurnSerialized(params) + ); +} + +async function runConversationTurnSerialized( + params: RunConversationTurnParams +): Promise { + const key = conversationRunnerKey( + params.executionScopeKey, params.rootSessionId ); - const contextBlock = renderConversationContext(params.timeline); const request = params.agentContent ?? params.displayText; const deadlineMs = Date.now() + TURN_DEADLINE_MS; const dispatchIso = new Date().toISOString(); - const turnId = crypto.randomUUID(); - - // The execution setup must exist BEFORE the user's words go public — a - // cancelled setup dialog cancels the whole send. Per-repo-scope memory - // keeps this silent after the first confirmation (the forkTeammateSession - // idiom): dialog once, remember, reuse with a toast; a failed remembered - // launch clears the memory and re-prompts exactly once below. - const remembered = loadForkSetupMemory(params.sourceScopeKey); - let usedRememberedSetup = Boolean(remembered); - let setup = - remembered ?? - (await requestForkSessionSetup({ - sourceTitle: params.conversationTitle, - sourceScopeKey: params.sourceScopeKey, - sourceModel: params.sourceModel, - })); - if (!remembered) saveForkSetupMemory(params.sourceScopeKey, setup); - - await pushConversationEvents(await params.getAccessToken(), { + const turnIntentId = params.turnIntentId ?? mintTurnIntentId(); + const io: TurnPushIo = { + getAccessToken: params.getAccessToken, orgId: params.orgId, rootSessionId: params.rootSessionId, - turnId, - events: [ - boundConversationEventForPush( - buildPushedUserEvent("conversation", params.displayText, dispatchIso) - ), - ], + turnId: turnIntentId, + turnIntentId, + onPushed: params.onPushed, + }; + const record = loadContinuation( + params.executionScopeKey, + params.rootSessionId + ); + const decision = decideContinuation({ + record, + turnIntentId, + assignedAgentDefinitionId: params.assignedAgentDefinitionId, }); - params.onPushed?.(); - params.onUserMessagePublished?.(); + if (decision.kind === "fresh" && decision.rollReason) { + log.info(`rolling conversation continuation: ${decision.rollReason}`); + clearContinuation(params.executionScopeKey, params.rootSessionId); + if (record) { + await cleanupConversationRunnerBestEffort(record.continuationSessionId); + } + } + if (decision.kind === "resume") { + const persistedBefore = await eventStoreProxy + .getPersistedEvents(decision.record.continuationSessionId) + .catch(() => [] as SessionEvent[]); + const delta = await params.loadPlaneDelta( + decision.record.readThroughPlaneSeq + ); + const localIntentIds = collectLocalTurnIntentIds(persistedBefore); + const contextRows = delta.events.filter( + (row) => !localIntentIds.has(row.turnId) + ); + await params.onRunnerReady?.( + decision.record.continuationSessionId, + turnIntentId, + turnIntentId + ); + await pushUserRow(io, params.displayText, dispatchIso); + params.onUserMessagePublished?.(); + + let dispatch; + try { + dispatch = await dispatchRunnerTurn(params, io, { + runnerSessionId: decision.record.continuationSessionId, + content: buildResumePrompt( + renderPlaneDeltaContext(contextRows), + request + ), + model: decision.record.model, + accountId: decision.record.accountId, + }); + } catch (error) { + log.warn("continuation send rejected; rolling to a fresh runner", error); + clearContinuation(params.executionScopeKey, params.rootSessionId); + await cleanupConversationRunnerBestEffort( + decision.record.continuationSessionId + ); + return startFreshEpisode(params, io, { + request, + deadlineMs, + dispatchIso, + userRowAlreadyPushed: true, + }); + } + return settleEpisode(params, io, { + key, + runnerSessionId: decision.record.continuationSessionId, + deadlineMs, + readThroughPlaneSeq: delta.lastSeq, + dispatch, + }); + } + + const initialContext = await params.loadInitialContext(turnIntentId); + if (decision.kind === "bootstrap") { + return dispatchBootstrapEpisode( + params, + io, + { + request, + deadlineMs, + dispatchIso, + userRowAlreadyPushed: false, + }, + { + runnerSessionId: decision.record.continuationSessionId, + accountId: decision.record.accountId, + model: decision.record.model, + }, + initialContext + ); + } + return startFreshEpisode( + params, + io, + { + request, + deadlineMs, + dispatchIso, + userRowAlreadyPushed: false, + }, + initialContext + ); +} + +interface BootstrapTurn { + request: string; + deadlineMs: number; + dispatchIso: string; + userRowAlreadyPushed: boolean; +} + +interface BootstrapEpisode { + runnerSessionId: string; + accountId?: string; + model?: string; +} + +async function startFreshEpisode( + params: RunConversationTurnParams, + io: TurnPushIo, + turn: BootstrapTurn, + loadedContext?: ConversationInitialContext +): Promise { + const setupMemoryKey = params.setupMemoryKey ?? params.sourceScopeKey; + const requestSetup = () => + requestForkSessionSetup({ + sourceTitle: params.conversationTitle, + sourceScopeKey: params.sourceScopeKey, + sourceModel: params.sourceModel, + sourceAgentDefinitionId: params.assignedAgentDefinitionId, + }); + const remembered = loadForkSetupMemory(setupMemoryKey); + let usedRememberedSetup = Boolean(remembered); + let setup = remembered ?? (await requestSetup()); + if (!remembered) saveForkSetupMemory(setupMemoryKey, setup); + assertAssignedAgent(setup.execution.agentDefinitionId, params); + const initialContext = + loadedContext ?? (await params.loadInitialContext(io.turnIntentId)); const createRunner = () => SessionService.create({ - task: buildRunnerPrompt(contextBlock, request), - imageDataUrls: params.imageDataUrls, + task: "", name: params.conversationTitle, repoPath: setup.workspaceRepoPath ?? undefined, model: setup.execution.model, @@ -260,16 +553,11 @@ export async function runConversationTurn( created = await createRunner(); } catch (error) { if (!usedRememberedSetup) throw error; - // The remembered setup went stale (checkout moved, account or model - // removed). Drop it and fall back to the dialog once. log.warn("remembered runner setup failed; re-prompting", error); - clearForkSetupMemory(params.sourceScopeKey); - setup = await requestForkSessionSetup({ - sourceTitle: params.conversationTitle, - sourceScopeKey: params.sourceScopeKey, - sourceModel: params.sourceModel, - }); - saveForkSetupMemory(params.sourceScopeKey, setup); + clearForkSetupMemory(setupMemoryKey); + setup = await requestSetup(); + assertAssignedAgent(setup.execution.agentDefinitionId, params); + saveForkSetupMemory(setupMemoryKey, setup); usedRememberedSetup = false; created = await createRunner(); } @@ -280,33 +568,133 @@ export async function runConversationTurn( }) ); } + const runnerSessionId = created.sessionId; - registerConversationRunner(key, runnerSessionId, dispatchIso); - params.onRunnerReady?.(runnerSessionId, turnId); - await waitForFirstTurnTerminal(runnerSessionId, deadlineMs); - markConversationRunnerTerminal(key, runnerSessionId); + prepareContinuation( + params.executionScopeKey, + params.rootSessionId, + { + continuationSessionId: runnerSessionId, + readThroughPlaneSeq: 0, + established: false, + bootstrapTurnIntentId: io.turnIntentId, + agentDefinitionId: setup.execution.agentDefinitionId, + accountId: setup.execution.accountId, + model: setup.execution.model, + workspaceRepoPath: setup.workspaceRepoPath ?? null, + }, + turn.dispatchIso + ); + return dispatchBootstrapEpisode( + params, + io, + turn, + { + runnerSessionId, + accountId: setup.execution.accountId, + model: setup.execution.model, + }, + initialContext + ); +} - const persisted = await eventStoreProxy - .getPersistedEvents(runnerSessionId) - .catch(() => [] as SessionEvent[]); - // The runner's own user event carries the injected context prefix (never - // pushed — the clean user row already went out in push #1); the agent and - // tool tail is the shared payload. - const agentTail = persisted - .filter((event) => event.source !== "user") - .map(boundConversationEventForPush); - - if (agentTail.length > 0) { - await pushConversationEventsChunked(await params.getAccessToken(), { - orgId: params.orgId, - rootSessionId: params.rootSessionId, - turnId, - events: agentTail, +async function dispatchBootstrapEpisode( + params: RunConversationTurnParams, + io: TurnPushIo, + turn: BootstrapTurn, + episode: BootstrapEpisode, + initialContext: ConversationInitialContext +): Promise { + const key = conversationRunnerKey( + params.executionScopeKey, + params.rootSessionId + ); + await params.onRunnerReady?.( + episode.runnerSessionId, + io.turnId, + io.turnIntentId + ); + if (!turn.userRowAlreadyPushed) { + await pushUserRow(io, params.displayText, turn.dispatchIso); + params.onUserMessagePublished?.(); + } + let dispatch; + try { + dispatch = await dispatchRunnerTurn(params, io, { + runnerSessionId: episode.runnerSessionId, + content: buildRunnerPrompt( + renderConversationContext( + initialContext.timeline, + initialContext.senders + ), + turn.request + ), + model: episode.model, + accountId: episode.accountId, }); - params.onPushed?.(); + } catch (error) { + clearContinuation(params.executionScopeKey, params.rootSessionId); + await cleanupConversationRunnerBestEffort(episode.runnerSessionId); + throw error; + } + if ( + !markContinuationEstablished( + params.executionScopeKey, + params.rootSessionId, + episode.runnerSessionId, + io.turnIntentId + ) + ) { + log.warn("continuation acceptance could not be persisted"); + } + return settleEpisode(params, io, { + key, + runnerSessionId: episode.runnerSessionId, + deadlineMs: turn.deadlineMs, + readThroughPlaneSeq: initialContext.readThroughPlaneSeq, + dispatch, + }); +} + +async function settleEpisode( + params: RunConversationTurnParams, + io: TurnPushIo, + input: { + key: string; + runnerSessionId: string; + deadlineMs: number; + readThroughPlaneSeq: number; + 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); + } finally { + if (outcome.status === "completed") { + await cleanupRetiredConversationRunners(input.key, input.runnerSessionId); + } else { + clearContinuation(params.executionScopeKey, params.rootSessionId); + await cleanupConversationRunnerBestEffort(input.runnerSessionId); + } } log.info( - `pushed conversation turn ${turnId}: 1 + ${agentTail.length} event(s) to ${key}` + `published conversation turn ${io.turnId}: 1 + ${tailCount} event(s) ` + + `to ${input.key} (${outcome.status})` ); - return { runnerSessionId, pushedEventCount: 1 + agentTail.length }; + return { + runnerSessionId: input.runnerSessionId, + pushedEventCount: 1 + tailCount, + turnIntentId: io.turnIntentId, + terminalStatus: outcome.status, + }; } diff --git a/src/features/Org2Cloud/org2CloudConversationEventsClient.test.ts b/src/features/Org2Cloud/org2CloudConversationEventsClient.test.ts new file mode 100644 index 0000000000..e1c0b66c9e --- /dev/null +++ b/src/features/Org2Cloud/org2CloudConversationEventsClient.test.ts @@ -0,0 +1,113 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; + +import { + type CloudConversationEvent, + listConversationEventsFrom, + retainConversationEventTail, +} from "./org2CloudConversationEventsClient"; + +const fetchMock = vi.fn(); + +function jsonResponse(body: unknown): Response { + return new Response(JSON.stringify(body), { + status: 200, + headers: { "content-type": "application/json" }, + }); +} + +function row(seq: number): CloudConversationEvent { + return { + id: `row-${seq}`, + rootSessionId: "root", + authorUserId: "user", + turnId: `turn-${seq}`, + seq, + event: { id: `event-${seq}` } as SessionEvent, + createdAt: "2026-08-25T00:00:00Z", + }; +} + +function wireRow(seq: number) { + const value = row(seq); + return { ...value, event: value.event }; +} + +beforeEach(() => { + vi.stubGlobal("fetch", fetchMock); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + fetchMock.mockReset(); +}); + +describe("retainConversationEventTail", () => { + it("keeps prompt memory bounded across pages", () => { + const first = retainConversationEventTail( + [], + Array.from({ length: 50 }, (_, index) => row(index + 1)), + 60 + ); + const second = retainConversationEventTail( + first, + Array.from({ length: 50 }, (_, index) => row(index + 51)), + 60 + ); + + expect(second).toHaveLength(60); + expect(second[0].seq).toBe(41); + expect(second.at(-1)?.seq).toBe(100); + }); + + it("rejects an invalid retention bound", () => { + expect(() => retainConversationEventTail([], [], 0)).toThrow( + "retainLast must be a positive integer" + ); + }); +}); + +describe("listConversationEventsFrom", () => { + it("traverses every page while retaining only the newest prompt rows", async () => { + fetchMock + .mockResolvedValueOnce( + jsonResponse({ + events: [wireRow(1), wireRow(2)], + hasMore: true, + }) + ) + .mockResolvedValueOnce( + jsonResponse({ events: [wireRow(3)], hasMore: false }) + ); + + const result = await listConversationEventsFrom("jwt", { + orgId: "org", + rootSessionId: "root", + afterSeq: 0, + retainLast: 2, + }); + + expect(result.lastSeq).toBe(3); + expect(result.events.map((event) => event.seq)).toEqual([2, 3]); + expect(fetchMock).toHaveBeenCalledTimes(2); + const bodies = fetchMock.mock.calls.map(([, init]) => + JSON.parse(String((init as RequestInit).body)) + ) as Array>; + expect(bodies.map((body) => body.p_after_seq)).toEqual([0, 2]); + }); + + it("rejects a non-increasing server sequence", async () => { + fetchMock.mockResolvedValueOnce( + jsonResponse({ events: [wireRow(4)], hasMore: false }) + ); + + await expect( + listConversationEventsFrom("jwt", { + orgId: "org", + rootSessionId: "root", + afterSeq: 4, + }) + ).rejects.toThrow("event sequence"); + }); +}); diff --git a/src/features/Org2Cloud/org2CloudConversationEventsClient.ts b/src/features/Org2Cloud/org2CloudConversationEventsClient.ts index 2e377f075a..e756c17edd 100644 --- a/src/features/Org2Cloud/org2CloudConversationEventsClient.ts +++ b/src/features/Org2Cloud/org2CloudConversationEventsClient.ts @@ -172,6 +172,72 @@ export interface PushConversationEventsResult { lastSeq: number; } +export interface ConversationEventWindow { + events: CloudConversationEvent[]; + /** Highest server seq traversed; `0` means no row has ever been read. */ + lastSeq: number; +} + +export function retainConversationEventTail( + current: readonly CloudConversationEvent[], + incoming: readonly CloudConversationEvent[], + retainLast: number +): CloudConversationEvent[] { + if (!Number.isSafeInteger(retainLast) || retainLast <= 0) { + throw new Org2CloudConversationError( + "ORG2_VALIDATION: retainLast must be a positive integer" + ); + } + const combined = [...current, ...incoming]; + return combined.length > retainLast + ? combined.slice(combined.length - retainLast) + : combined; +} + +/** + * Traverse the plane from one exclusive cursor to its authoritative head. + * Callers may retain only the newest rows for prompt construction without + * losing the exact head cursor. Sequence zero is the sole unread sentinel. + */ +export async function listConversationEventsFrom( + accessToken: string, + params: { + orgId: string; + rootSessionId: string; + afterSeq: number; + retainLast?: number; + } +): Promise { + if (!Number.isSafeInteger(params.afterSeq) || params.afterSeq < 0) { + throw new Org2CloudConversationError( + "ORG2_VALIDATION: afterSeq must be a non-negative integer" + ); + } + let events: CloudConversationEvent[] = []; + let lastSeq = params.afterSeq; + for (;;) { + const page = await listConversationEvents(accessToken, { + orgId: params.orgId, + rootSessionId: params.rootSessionId, + afterSeq: lastSeq, + }); + for (const row of page.events) { + if (!Number.isSafeInteger(row.seq) || row.seq <= lastSeq) { + throw new Org2CloudConversationError( + "unparseable cloud conversation event sequence" + ); + } + lastSeq = row.seq; + } + events = + params.retainLast === undefined + ? [...events, ...page.events] + : retainConversationEventTail(events, page.events, params.retainLast); + if (!page.hasMore || page.events.length === 0) break; + } + return { events, lastSeq }; +} + export async function pushConversationEvents( accessToken: string, params: {