Persist local shared-conversation continuations

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}`,

src/engines/ChatPanel/hooks/useImportedSessionSubmitOverride.ts

@@ -12,19 +12,26 @@ 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,
   conversationPlaneAtom,
   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<boolean> => {
       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<void>((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<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,
+              };
+            },
+            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) => {

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<SessionEvent>) =>
+    ({
+      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"]);
+  });
+});

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<string, ActiveConversationRunner[]>;
+
+function dedupeByTurnId(
+  runners: readonly ActiveConversationRunner[]
+): ActiveConversationRunner[] {
+  const seen = new Set<string>();
+  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<string, ActiveConversationRunner[]>
->({});
+const runnerRegistryStateAtom = atom<RunnerRegistry>({});
+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[],