Skip to content

Share the cloud conversation runtime adapter #945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Neonforge98 wants to merge 1 commit into from
Closed

Share the cloud conversation runtime adapter #945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
167 changes: 36 additions & 131 deletions src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -19,45 +24,29 @@ 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";
import type { ForkImportedErrorKind } from "@src/features/TeamCollaboration/useForkImportedSession";
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";
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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<string> => {
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) => {
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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<string, string>();
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,
Expand All @@ -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.
Expand Down Expand Up @@ -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
Expand All @@ -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);
Expand Down Expand Up @@ -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);
});
Expand Down Expand Up @@ -632,10 +541,6 @@ export function useImportedSessionSubmitOverride({
restorePendingDraft,
sessionId,
sessions,
setAccessSettings,
setActiveRunners,
setAuth,
setPlaneSignal,
submitIntoForkedSession,
t,
tipImportedCopy,
Expand Down
99 changes: 99 additions & 0 deletions src/features/Org2Cloud/SessionConversation/cloudConversationRuntime.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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");
});
});
Loading
Loading