diff --git a/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts b/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts index 18a7a973db..b6a352b802 100644 --- a/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts +++ b/src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts @@ -3,11 +3,16 @@ import { useCallback, useMemo, useRef } from "react"; import { useTranslation } from "react-i18next"; import Message from "@src/components/Message"; -import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; -import type { SessionEvent } from "@src/engines/SessionCore/core/types"; import { mintTurnIntentId } from "@src/engines/SessionCore/sync/adapters/shared/eventFactories"; import { waitForSessionChannelReady } from "@src/engines/SessionCore/sync/useSessionChannel"; -import { activeConversationRunnersAtom } from "@src/features/Org2Cloud/SessionConversation/activeConversationRunnersAtom"; +import { + getRequiredCloudAccessToken, + loadCloudConversationInitialContext, + loadCloudConversationPlaneDelta, + registerCloudConversationRunner, + settleCloudConversationRunner, + signalCloudConversationPlane, +} from "@src/features/Org2Cloud/SessionConversation/cloudConversationRuntime"; import { type ConversationFamilyMember, resolveConversationFamily, @@ -19,32 +24,18 @@ import { } from "@src/features/Org2Cloud/SessionConversation/conversationExecutionStore"; import { publishOwnerTurn } from "@src/features/Org2Cloud/SessionConversation/conversationOwnerPublisher"; import { - bumpConversationPlaneSignal, conversationPlaneAtom, conversationPlaneKey, - conversationPlaneSignalAtom, } from "@src/features/Org2Cloud/SessionConversation/conversationPlaneAtom"; import { - conversationEventKey, - mergePlaneIntoTranscript, -} from "@src/features/Org2Cloud/SessionConversation/conversationTimeline"; -import { - CONVERSATION_CONTEXT_MAX_ENTRIES, buildResumePrompt, renderPlaneDeltaContext, runConversationTurn, } from "@src/features/Org2Cloud/SessionConversation/conversationTurnRunner"; import { - org2CloudAccessSettingsAtom, - withCloudSessionMode, -} from "@src/features/Org2Cloud/org2CloudAccessSettings"; -import { - commitRefreshedAuth, org2CloudAuthAtom, 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"; @@ -52,12 +43,10 @@ import type { ForkImportedErrorKind } from "@src/features/TeamCollaboration/useF import { useForkImportedSession } from "@src/features/TeamCollaboration/useForkImportedSession"; import { createLogger } from "@src/hooks/logger"; import { useSessionView } from "@src/hooks/ui/tabs/useSessionView"; -import { COLLAB_SESSION_ACCESS_MODE } from "@src/store/collaboration/types"; import type { Session } from "@src/store/session"; import { sessionsAtom } from "@src/store/session"; import { restoreToInputAtom } from "@src/store/session/cliSessionStatusAtom"; import type { SessionContinuation } from "@src/store/session/sessionTabPlacementAtom"; -import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; import type { SubmitOverrideInput } from "./useInputArea/types"; import { useUserIntentSubmit } from "./useWorkspaceChat/useUserIntentSubmit"; @@ -168,11 +157,7 @@ export function useImportedSessionSubmitOverride({ // 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); const planeEntries = useAtomValue(conversationPlaneAtom); - const setPlaneSignal = useSetAtom(conversationPlaneSignalAtom); - const setAccessSettings = useSetAtom(org2CloudAccessSettingsAtom); - const setActiveRunners = useSetAtom(activeConversationRunnersAtom); const conversationRootId = useMemo(() => { if (lineage) return lineage.rootSessionId ?? lineage.sourceSessionId; if (currentSession?.importedFrom) { @@ -230,14 +215,7 @@ export function useImportedSessionSubmitOverride({ // A turn can outlive the access token valid at dispatch (a 10-minute // member turn did, live — its tail push failed with "JWT expired"), so // every plane push resolves a fresh token from the CURRENT auth state. - const getAccessToken = useCallback(async (): Promise => { - const current = getInstrumentedStore().get(org2CloudAuthAtom); - if (!current) throw new Error("cloud sign-in required"); - const fresh = await ensureFreshSession(current); - if (!fresh) throw new Error("cloud auth refresh failed"); - commitRefreshedAuth(setAuth, current, fresh); - return fresh.accessToken; - }, [setAuth]); + const getAccessToken = useCallback(() => getRequiredCloudAccessToken(), []); const restorePendingDraft = useCallback( (pending: SubmitOverrideInput, targetSessionId: string) => { @@ -264,9 +242,6 @@ export function useImportedSessionSubmitOverride({ forkSubmitInFlightRef.current = true; try { if (!auth) throw new Error("cloud sign-in required"); - const freshAuth = await ensureFreshSession(auth); - if (!freshAuth) throw new Error("cloud auth refresh failed"); - commitRefreshedAuth(setAuth, auth, freshAuth); const rootLocal = sessions.find( (candidate) => candidate.session_id === planeInfo.rootId @@ -299,18 +274,7 @@ export function useImportedSessionSubmitOverride({ const runnerSessionId = liveRunnerSessionId; if (!runnerSessionId) return; liveRunnerSessionId = null; - setActiveRunners((current) => { - const list = current[planeInfo.rootId]; - if (!list) return current; - const kept = list.filter( - (runner) => runner.runnerSessionId !== runnerSessionId - ); - if (kept.length === list.length) return current; - const next = { ...current }; - if (kept.length === 0) delete next[planeInfo.rootId]; - else next[planeInfo.rootId] = kept; - return next; - }); + settleCloudConversationRunner(planeInfo.rootId, runnerSessionId); }; const turnPromise = runConversationTurn({ getAccessToken, @@ -321,57 +285,19 @@ export function useImportedSessionSubmitOverride({ displayText: input.displayText, agentContent: input.agentContent, imageDataUrls: input.imageDataUrls, - 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, - }; - }, + loadInitialContext: (excludeTurnIntentId) => + loadCloudConversationInitialContext({ + orgId: planeInfo.orgId, + rootSessionId: planeInfo.rootId, + streamSessionId: sessionId, + excludeTurnIntentId, + }), loadPlaneDelta: (afterSeq) => - getAccessToken().then((accessToken) => - listConversationEventsFrom(accessToken, { - orgId: planeInfo.orgId, - rootSessionId: planeInfo.rootId, - afterSeq, - retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, - }) + loadCloudConversationPlaneDelta( + planeInfo.orgId, + planeInfo.rootId, + afterSeq, + getAccessToken ), sourceScopeKey: rootRow?.repoScopeKey, sourceModel: currentSession?.model ?? rootRow?.model, @@ -384,33 +310,20 @@ export function useImportedSessionSubmitOverride({ ), executionScopeKey: executorScope, onRunnerReady: (runnerSessionId, turnId, turnIntentId) => { - // Plumbing session: never sync it to the cloud as a session. - setAccessSettings((current) => - withCloudSessionMode( - current, - planeInfo.orgId, - runnerSessionId, - COLLAB_SESSION_ACCESS_MODE.OFF - ) - ); // Overlay the runner's LIVE events (thinking / tools / worked-for) // into the conversation until the plane carries this turn's // agent tail — or the turn settles without one. liveRunnerSessionId = runnerSessionId; - setActiveRunners((current) => { - const list = current[planeInfo.rootId] ?? []; - return { - ...current, - [planeInfo.rootId]: [ - ...list, - { runnerSessionId, turnId, turnIntentId }, - ], - }; + registerCloudConversationRunner({ + orgId: planeInfo.orgId, + rootSessionId: planeInfo.rootId, + runnerSessionId, + turnId, + turnIntentId, }); }, onUserMessagePublished: publishResolve, - onPushed: () => - bumpConversationPlaneSignal(setPlaneSignal, planeInfo.orgId), + onPushed: () => signalCloudConversationPlane(planeInfo.orgId), }); // The composer unblocks as soon as the user's words are on the // plane; the agent tail continues in the background. @@ -444,9 +357,6 @@ export function useImportedSessionSubmitOverride({ // Group-chat routing owns its own sends. if (await onFallbackSubmit(input)) return true; if (!auth) return false; - const freshAuth = await ensureFreshSession(auth); - if (!freshAuth) return false; - commitRefreshedAuth(setAuth, auth, freshAuth); const executorScope = cloudConversationExecutorScopeKey( org2CloudAuthIdentityKey(auth), planeInfo.orgId @@ -456,12 +366,12 @@ export function useImportedSessionSubmitOverride({ ?.readThroughPlaneSeq ?? 0; let delta; try { - delta = await listConversationEventsFrom(freshAuth.accessToken, { - orgId: planeInfo.orgId, - rootSessionId: planeInfo.rootId, - afterSeq: ownerCursor, - retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, - }); + delta = await loadCloudConversationPlaneDelta( + planeInfo.orgId, + planeInfo.rootId, + ownerCursor, + getAccessToken + ); } catch (error) { logger.error("owner conversation delta load failed", error); restorePendingDraft(input, sessionId); @@ -505,8 +415,7 @@ export function useImportedSessionSubmitOverride({ displayText: input.displayText, executorScope, readThroughPlaneSeq: delta.lastSeq, - onPushed: () => - bumpConversationPlaneSignal(setPlaneSignal, planeInfo.orgId), + onPushed: () => signalCloudConversationPlane(planeInfo.orgId), }).catch((error: unknown) => { logger.warn("owner turn publish failed", error); }); @@ -632,10 +541,6 @@ export function useImportedSessionSubmitOverride({ restorePendingDraft, sessionId, sessions, - setAccessSettings, - setActiveRunners, - setAuth, - setPlaneSignal, submitIntoForkedSession, t, tipImportedCopy, diff --git a/src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.test.ts b/src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.test.ts new file mode 100644 index 0000000000..3fe63fd243 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import type { Session } from "@src/store/session"; + +import type { CloudConversationEvent } from "../org2CloudConversationEventsClient"; +import { + type CloudConversationContextDeps, + loadCloudConversationInitialContext, +} from "./cloudConversationRuntime"; + +function event( + id: string, + source: "user" | "assistant", + text: string, + turnIntentId?: string +): SessionEvent { + return { + id, + chunk_id: id, + sessionId: "root", + createdAt: "2026-08-25T00:00:00.000Z", + source, + displayText: text, + args: {}, + result: turnIntentId ? { turnIntentId } : {}, + } as SessionEvent; +} + +function row( + seq: number, + turnId: string, + author: string, + inner: SessionEvent +): CloudConversationEvent { + return { + id: `row-${seq}`, + rootSessionId: "root", + authorUserId: author.toLowerCase(), + authorDisplayName: author, + turnId, + seq, + event: inner, + createdAt: inner.createdAt, + }; +} + +describe("cloud conversation runtime context", () => { + it("loads and attributes the initial plane once while excluding redelivery", async () => { + const priorUser = event("prior-user", "user", "please review", "prior"); + const excluded = event( + "redelivered-user", + "user", + "same request", + "current" + ); + const rootHistory = event("root-answer", "assistant", "earlier answer"); + const deps: CloudConversationContextDeps = { + getAccessToken: vi.fn(async () => "jwt"), + getAuth: () => ({ userId: "viewer", supabaseUrl: "https://cloud" }), + getSessions: () => [{ session_id: "root", name: "Root" } as Session], + loadPlane: vi.fn(async () => ({ + events: [ + row(8, "prior", "Alice", priorUser), + row(9, "current", "Viewer", excluded), + ], + lastSeq: 9, + })), + loadPersistedEvents: vi.fn(async () => [rootHistory]), + }; + + const context = await loadCloudConversationInitialContext( + { + orgId: "org", + rootSessionId: "root", + streamSessionId: "surface", + excludeTurnIntentId: "current", + }, + deps + ); + + expect(context.timeline.map((item) => item.displayText)).toEqual([ + "earlier answer", + "please review", + ]); + const planeUser = context.timeline.find( + (item) => item.displayText === "please review" + ); + expect(context.senders?.get(planeUser?.id ?? "")).toBe("Alice"); + expect(context.readThroughPlaneSeq).toBe(9); + expect(deps.loadPlane).toHaveBeenCalledWith("jwt", { + orgId: "org", + rootSessionId: "root", + afterSeq: 0, + retainLast: 60, + }); + expect(deps.loadPersistedEvents).toHaveBeenCalledWith("root"); + }); +}); diff --git a/src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.ts b/src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.ts new file mode 100644 index 0000000000..5c1ea0d290 --- /dev/null +++ b/src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.ts @@ -0,0 +1,196 @@ +/** Shared managed-cloud adapter for every local conversation executor. */ +import { eventStoreProxy } from "@src/engines/SessionCore/core/store/EventStoreProxy"; +import type { SessionEvent } from "@src/engines/SessionCore/core/types"; +import { findImportedSession } from "@src/features/TeamCollaboration/engine/collabImportIdentity"; +import { COLLAB_SESSION_ACCESS_MODE } from "@src/store/collaboration/types"; +import { type Session, sessionsAtom } from "@src/store/session"; +import { getInstrumentedStore } from "@src/util/core/state/instrumentedStore"; + +import { getFreshCloudAccessToken } from "../cloudShortId"; +import { + org2CloudAccessSettingsAtom, + withCloudSessionMode, +} from "../org2CloudAccessSettings"; +import { org2CloudAuthAtom } from "../org2CloudAuthAtom"; +import { + type ConversationEventWindow, + listConversationEventsFrom, +} from "../org2CloudConversationEventsClient"; +import { activeConversationRunnersAtom } from "./activeConversationRunnersAtom"; +import { + bumpConversationPlaneSignal, + conversationPlaneSignalAtom, +} from "./conversationPlaneAtom"; +import { + conversationEventKey, + mergePlaneIntoTranscript, +} from "./conversationTimeline"; +import { + CONVERSATION_CONTEXT_MAX_ENTRIES, + type ConversationInitialContext, +} from "./conversationTurnRunner"; + +export async function getRequiredCloudAccessToken(): Promise { + const token = await getFreshCloudAccessToken(); + if (!token) throw new Error("cloud sign-in required"); + return token; +} + +export interface CloudConversationContextDeps { + getAccessToken: () => Promise; + getAuth: () => { userId: string; supabaseUrl: string } | null; + getSessions: () => Session[]; + loadPlane: ( + accessToken: string, + params: { + orgId: string; + rootSessionId: string; + afterSeq: number; + retainLast: number; + } + ) => Promise; + loadPersistedEvents: (sessionId: string) => Promise; +} + +function createContextDeps(): CloudConversationContextDeps { + const store = getInstrumentedStore(); + return { + getAccessToken: getRequiredCloudAccessToken, + getAuth: () => store.get(org2CloudAuthAtom), + getSessions: () => store.get(sessionsAtom), + loadPlane: (accessToken, params) => + listConversationEventsFrom(accessToken, params), + loadPersistedEvents: (sessionId) => + eventStoreProxy + .getPersistedEvents(sessionId) + .catch(() => [] as SessionEvent[]), + }; +} + +export async function loadCloudConversationInitialContext( + params: { + orgId: string; + rootSessionId: string; + streamSessionId: string; + excludeTurnIntentId: string; + }, + deps: CloudConversationContextDeps = createContextDeps() +): Promise { + const auth = deps.getAuth(); + if (!auth) throw new Error("cloud sign-in required"); + const window = await deps.loadPlane(await deps.getAccessToken(), { + orgId: params.orgId, + rootSessionId: params.rootSessionId, + afterSeq: 0, + retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, + }); + const rows = window.events.filter( + (row) => row.turnId !== params.excludeTurnIntentId + ); + const sessions = deps.getSessions(); + const localRoot = + sessions.find( + (candidate) => candidate.session_id === params.rootSessionId + ) ?? + findImportedSession( + sessions, + params.orgId, + params.rootSessionId, + auth.supabaseUrl + ); + const rootEvents = localRoot + ? await deps.loadPersistedEvents(localRoot.session_id) + : []; + const timeline = mergePlaneIntoTranscript( + rootEvents, + rows, + params.streamSessionId, + 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, + }; +} + +export async function loadCloudConversationPlaneDelta( + orgId: string, + rootSessionId: string, + afterSeq: number, + getAccessToken: () => Promise = getRequiredCloudAccessToken +): Promise { + return listConversationEventsFrom(await getAccessToken(), { + orgId, + rootSessionId, + afterSeq, + retainLast: CONVERSATION_CONTEXT_MAX_ENTRIES, + }); +} + +export function registerCloudConversationRunner(input: { + orgId: string; + rootSessionId: string; + runnerSessionId: string; + turnId: string; + turnIntentId: string; +}): void { + const store = getInstrumentedStore(); + store.set(org2CloudAccessSettingsAtom, (current) => + withCloudSessionMode( + current, + input.orgId, + input.runnerSessionId, + COLLAB_SESSION_ACCESS_MODE.OFF + ) + ); + store.set(activeConversationRunnersAtom, (current) => ({ + ...current, + [input.rootSessionId]: [ + ...(current[input.rootSessionId] ?? []), + { + runnerSessionId: input.runnerSessionId, + turnId: input.turnId, + turnIntentId: input.turnIntentId, + }, + ], + })); +} + +export function settleCloudConversationRunner( + rootSessionId: string, + runnerSessionId: string +): void { + const store = getInstrumentedStore(); + store.set(activeConversationRunnersAtom, (current) => { + const list = current[rootSessionId]; + if (!list) return current; + const kept = list.filter( + (runner) => runner.runnerSessionId !== runnerSessionId + ); + if (kept.length === list.length) return current; + const next = { ...current }; + if (kept.length === 0) delete next[rootSessionId]; + else next[rootSessionId] = kept; + return next; + }); +} + +export function signalCloudConversationPlane(orgId: string): void { + const store = getInstrumentedStore(); + bumpConversationPlaneSignal( + (update) => store.set(conversationPlaneSignalAtom, update), + orgId + ); +}