Keep resumed threads in their own provider sessions

apps/server/src/services/threads/queued-messages.ts

@@ -1,11 +1,12 @@
 import {
-  claimQueuedThreadMessageGroup,
   claimNextQueuedThreadMessageGroup,
+  claimQueuedThreadMessageGroup,
   createQueuedThreadMessageInTransaction,
   deleteClaimedQueuedThreadMessageBatchInTransaction,
-  getQueuedThreadMessage,
   getEnvironment,
   getHost,
+  getQueuedThreadMessage,
+  getStoredProviderSession,
   getThread,
   isOrdinaryTurnEndQueuedMessage,
   isThreadQueueAutoSendPaused,
@@ -188,16 +189,17 @@ export interface CreateQueuedMessageForThreadArgs {
 function admitQueuedMessage(
   db: DbQueryConnection,
   thread: Thread,
-): { providerThreadId: string | null } {
+): { hasProviderSession: boolean } {
   ensureThreadQueueIsWritable(thread);
-  const providerThreadId = getLastProviderThreadId({ db }, thread.id);
+  const hasProviderSession =
+    getStoredProviderSession(db, thread.id).kind !== "none";
   if (thread.environmentId === null) {
-    if (providerThreadId !== null) {
+    if (hasProviderSession) {
       throwThreadEnvironmentUnavailable(
         threadEnvironmentUnavailableDetails("never_attached", null),
       );
     }
-    return { providerThreadId };
+    return { hasProviderSession };
   }
   const environment = getEnvironment(db, thread.environmentId);
   const goneDetails = environment
@@ -206,7 +208,7 @@ function admitQueuedMessage(
   if (goneDetails) {
     throwThreadEnvironmentUnavailable(goneDetails);
   }
-  return { providerThreadId };
+  return { hasProviderSession };
 }
 
 export async function createQueuedMessageForThread(
@@ -228,14 +230,14 @@ export async function createQueuedMessageForThread(
     senderThreadId: payload.senderThreadId,
     targetThread: thread,
   });
-  const { currentThread, providerThreadId, queuedMessage } =
+  const { currentThread, hasProviderSession, queuedMessage } =
     deps.db.transaction(
       (tx) => {
         const currentThread = getThread(tx, thread.id);
         if (!currentThread) {
           throw new ApiError(404, "thread_not_found", "Thread not found");
         }
-        const { providerThreadId } = admitQueuedMessage(tx, currentThread);
+        const { hasProviderSession } = admitQueuedMessage(tx, currentThread);
         const queuedMessage = createQueuedThreadMessageInTransaction(tx, {
           threadId: thread.id,
           content: payload.input,
@@ -261,7 +263,7 @@ export async function createQueuedMessageForThread(
           payload: { kind: "inline" },
           systemNotice: null,
         });
-        return { currentThread, providerThreadId, queuedMessage };
+        return { currentThread, hasProviderSession, queuedMessage };
       },
       { behavior: "immediate" },
     );
@@ -273,7 +275,7 @@ export async function createQueuedMessageForThread(
       providerId: thread.providerId,
     });
   }
-  if (currentThread.status === "idle" && providerThreadId !== null) {
+  if (currentThread.status === "idle" && hasProviderSession) {
     requestQueuedMessageDispatch(deps, {
       kind: "thread-ready",
       threadId: thread.id,

apps/server/src/services/threads/thread-commands.ts

@@ -24,7 +24,10 @@ import {
   LIVE_DAEMON_COMMAND_TIMEOUT_MS,
   startLiveHostCommand,
 } from "../hosts/live-command.js";
-import { getLastProviderThreadId } from "./thread-events.js";
+import {
+  getLastProviderThreadId,
+  requireDispatchableProviderThreadId,
+} from "./thread-events.js";
 import type { ThreadForkDescriptor } from "./thread-startup-store.js";
 import {
   resolveThreadRuntimeCommandConfig,
@@ -356,7 +359,8 @@ export async function prepareTurnSubmitCommandPayload(
 ): Promise<PreparedTurnSubmitCommandPayload> {
   await deps.providerRegistry.whenRegistrationsSettled();
   const providerThreadId = requireProviderThreadId(
-    args.providerThreadId ?? getLastProviderThreadId(deps, args.thread.id),
+    args.providerThreadId ??
+      requireDispatchableProviderThreadId(deps, args.thread.id),
     args.thread.id,
   );
   const runtimeContext = await resolveThreadRuntimeCommandConfig(deps, {

apps/server/src/services/threads/thread-edit-message.ts

@@ -8,6 +8,8 @@ import {
   getThread,
   hasQueuedThreadMessages,
   hasRootStoredTurnStarted,
+  classifyStoredProviderThreadClaim,
+  wouldRemoveSharedProviderSessionClaim,
   listActiveBackgroundTaskCountsByThreadIds,
   type DbQueryConnection,
 } from "@bb/db";
@@ -274,6 +276,23 @@ function resolveEditableTurnCandidate(
   ) {
     conflict("This earlier turn has no provider history");
   }
+  const precedingSessionClaim =
+    precedingCompletion?.providerThreadId == null
+      ? null
+      : classifyStoredProviderThreadClaim(db, {
+          providerThreadId: precedingCompletion.providerThreadId,
+          threadId: thread.id,
+        });
+  if (precedingSessionClaim === "foreign") {
+    conflict(
+      "This earlier turn is recorded under another thread's provider session",
+    );
+  }
+  if (precedingSessionClaim === "ambiguous") {
+    conflict(
+      "This earlier turn is recorded under a provider session another thread announced at the same moment",
+    );
+  }
   const precedingProviderCheckpoint =
     precedingTurnId === null
       ? null
@@ -285,10 +304,22 @@ function resolveEditableTurnCandidate(
   if (precedingTurnId !== null && precedingProviderCheckpoint === null) {
     conflict("This earlier provider turn has no editable history checkpoint");
   }
+  const oldMaxSequence = getHighWaterMarks(db, [thread.id])[thread.id] ?? 0;
+  if (
+    wouldRemoveSharedProviderSessionClaim(db, {
+      cutoffSequence: requestRow.sequence,
+      oldMaxSequence,
+      threadId: thread.id,
+    })
+  ) {
+    conflict(
+      "Editing this message would erase provider session ownership shared with another thread. Clear context (/clear or bb thread clear) for a new session; history is kept.",
+    );
+  }
   return {
     leadingAgentOnlyInput: getLeadingAgentOnlyInput(request.input),
     currentTurnId: accepted.turnId,
-    oldMaxSequence: getHighWaterMarks(db, [thread.id])[thread.id] ?? 0,
+    oldMaxSequence,
     precedingProviderCheckpoint,
     requestSequence: requestRow.sequence,
     sourceProviderThreadId:

apps/server/src/services/threads/thread-events.ts

@@ -4,6 +4,7 @@ import {
   getActiveStoredTurnId,
   getLastStoredProviderThreadId,
   getLastStoredTurnRequestEvent,
+  getStoredProviderSession,
   getStoredTurnRequestEventForTurn,
   getThread,
   listStoredTurnStartedKeys,
@@ -879,6 +880,38 @@ export function getLastProviderThreadId(
   return getLastStoredProviderThreadId(deps.db, threadId);
 }
 
+export function requireDispatchableProviderThreadId(
+  deps: ThreadEventReadDeps,
+  threadId: string,
+): string | null {
+  const session = getStoredProviderSession(deps.db, threadId);
+  switch (session.kind) {
+    case "none":
+      return null;
+    case "owned":
+      return session.providerThreadId;
+    case "invalid":
+    case "ambiguous":
+    case "foreign":
+      throw new ApiError(
+        409,
+        "provider_session_unavailable",
+        session.kind === "invalid"
+          ? "This thread has a stored identity without a valid provider session, so bb will not replace it silently. Clear context (/clear or bb thread clear) for a new session; history is kept."
+          : session.kind === "ambiguous"
+            ? "Another thread claimed this thread's provider session in the same millisecond, so bb will not guess whose it is. Clear context (/clear or bb thread clear) for a new session; history is kept."
+            : "This thread's only provider session belongs to another thread, so bb will not resume it. Clear context (/clear or bb thread clear) for a new session; history is kept.",
+        {
+          details: {
+            reason: session.kind,
+            providerThreadId: session.providerThreadId,
+            claimantThreadIds: session.claimantThreadIds,
+          },
+        },
+      );
+  }
+}
+
 export function getLastExecutionOptions(
   deps: Pick<AppDeps, "db">,
   threadId: string,

apps/server/src/services/threads/thread-fork-history.ts

@@ -2,6 +2,8 @@ import {
   copyStoredThreadEventsInTransaction,
   findLastCompletedRootStoredTurn,
   findLastRootStoredTurnStarted,
+  classifyStoredProviderThreadClaim,
+  getStoredProviderSession,
   listStoredEventRows,
   listStoredTurnCompletedRowsByTurnIds,
   type StoredEventRow,
@@ -10,10 +12,7 @@ import type { Thread, ThreadEvent, ThreadEventType } from "@bb/domain";
 import type { AppDeps } from "../../types.js";
 import { ApiError } from "../../errors.js";
 import { parseStoredEvent } from "./thread-data.js";
-import {
-  getLastProviderThreadId,
-  parseStoredTurnRequestEvent,
-} from "./thread-events.js";
+import { parseStoredTurnRequestEvent } from "./thread-events.js";
 import { resolveTurnProviderCheckpointId } from "./thread-edit-message.js";
 import type { ThreadForkDescriptor } from "./thread-startup-store.js";
 
@@ -50,6 +49,20 @@ function readTurnCompletion(
   return { event, sequence: row.sequence };
 }
 
+function classifyCompletionSession(
+  deps: Pick<AppDeps, "db">,
+  args: { completion: StoredTurnCompletion; sourceThreadId: string },
+): "owned" | "foreign" | "ambiguous" {
+  if (args.completion.event.providerThreadId === null) {
+    return "foreign";
+  }
+  const claim = classifyStoredProviderThreadClaim(deps.db, {
+    providerThreadId: args.completion.event.providerThreadId,
+    threadId: args.sourceThreadId,
+  });
+  return claim === "unannounced" ? "owned" : claim;
+}
+
 function resolveCheckpointForkDescriptor(args: {
   completion: StoredTurnCompletion;
   providerId: string;
@@ -99,6 +112,20 @@ function resolveAnchoredForkPoint(
       `Cannot fork at sequence ${args.sourceSeqEnd}: the turn containing it has no provider session`,
     );
   }
+  const completionSession = classifyCompletionSession(deps, {
+    completion,
+    sourceThreadId: args.sourceThread.id,
+  });
+  if (completionSession === "foreign") {
+    forkPointUnavailable(
+      `Cannot fork at sequence ${args.sourceSeqEnd}: the turn containing it is recorded under another thread's provider session`,
+    );
+  }
+  if (completionSession === "ambiguous") {
+    forkPointUnavailable(
+      `Cannot fork at sequence ${args.sourceSeqEnd}: the turn containing it is recorded under a provider session another thread announced at the same moment`,
+    );
+  }
   const latestRootTurn = findLastRootStoredTurnStarted(deps.db, {
     threadId: args.sourceThread.id,
   });
@@ -146,13 +173,26 @@ export function resolveThreadForkPoint(
       sourceThread: args.sourceThread,
     });
   }
-  const sourceProviderThreadId = getLastProviderThreadId(
-    deps,
-    args.sourceThread.id,
-  );
-  if (sourceProviderThreadId === null) {
+  const sourceSession = getStoredProviderSession(deps.db, args.sourceThread.id);
+  if (sourceSession.kind === "none") {
     return null;
   }
+  if (sourceSession.kind === "invalid") {
+    forkPointUnavailable(
+      "Cannot fork: the source thread has a stored identity without a valid provider session",
+    );
+  }
+  if (sourceSession.kind === "ambiguous") {
+    forkPointUnavailable(
+      "Cannot fork: another thread announced the source thread's provider session at the same moment, so bb cannot tell whose it is",
+    );
+  }
+  if (sourceSession.kind === "foreign") {
+    forkPointUnavailable(
+      "Cannot fork: the source thread's only provider session belongs to another thread",
+    );
+  }
+  const sourceProviderThreadId = sourceSession.providerThreadId;
   const lastCompletedTurn = findLastCompletedRootStoredTurn(deps.db, {
     threadId: args.sourceThread.id,
   });
@@ -178,7 +218,11 @@ export function resolveThreadForkPoint(
     turnId: lastCompletedTurn.turnId,
   });
   const descriptor =
-    completion === null
+    completion === null ||
+    classifyCompletionSession(deps, {
+      completion,
+      sourceThreadId: args.sourceThread.id,
+    }) !== "owned"
       ? null
       : resolveCheckpointForkDescriptor({
           completion,

apps/server/src/services/threads/thread-lifecycle.ts

@@ -61,13 +61,14 @@ import {
 } from "../../internal/command-result-side-effects.js";
 import {
   appendSystemErrorEventInTransaction,
-  buildSystemErrorEventData,
   appendThreadEventInTransaction,
   appendThreadEventsInTransaction,
   appendThreadInterruptedEventInTransaction,
   appendThreadProvisioningEventInTransaction,
+  buildSystemErrorEventData,
   getActiveTurnId,
   getLastProviderThreadId,
+  requireDispatchableProviderThreadId,
 } from "./thread-events.js";
 import {
   applyLoggedThreadLifecycleEvent,
@@ -776,6 +777,14 @@ function recordEmptyThreadStartProviderSessionInTransaction(
   ) {
     return;
   }
+  appendThreadEventInTransaction(args.deps.db, {
+    threadId: args.thread.id,
+    environmentId: args.command.environmentId,
+    providerThreadId: args.report.result.providerThreadId,
+    type: "thread/identity",
+    scope: threadScope(),
+    data: { providerThreadId: args.report.result.providerThreadId },
+  });
   appendThreadEventInTransaction(args.deps.db, {
     threadId: args.thread.id,
     environmentId: args.command.environmentId,
@@ -968,7 +977,10 @@ export async function prepareReadyThreadTurnCommand(
   await ensureHostSessionReadyForWork(deps, {
     hostId: args.environment.hostId,
   });
-  const providerThreadId = getLastProviderThreadId(deps, args.thread.id);
+  const providerThreadId = requireDispatchableProviderThreadId(
+    deps,
+    args.thread.id,
+  );
   if (providerThreadId) {
     const preparedCommand = await prepareTurnSubmitCommandPayload(deps, {
       environment: args.environment,

apps/server/test/helpers/seed.ts

@@ -323,6 +323,30 @@ export function seedTurnStarted(
   });
 }
 
+export function seedThreadIdentity(
+  deps: Pick<AppDeps, "db" | "hub">,
+  args: {
+    createdAt?: number;
+    environmentId?: string | null;
+    providerThreadId: string;
+    sequence?: number;
+    threadId: string;
+  },
+): void {
+  seedEvent(deps, {
+    threadId: args.threadId,
+    environmentId: args.environmentId ?? null,
+    providerThreadId: args.providerThreadId,
+    createdAt: args.createdAt,
+    sequence:
+      args.sequence ??
+      getLatestThreadSequence(deps.db, { threadId: args.threadId }) + 1,
+    type: "thread/identity",
+    scope: threadScope(),
+    data: {},
+  });
+}
+
 export function seedThreadRuntimeState(
   deps: Pick<AppDeps, "db" | "hub">,
   args: {