[Turn Steal 5/N]: feat: rebuild, offload, and steal a paused turn

.changeset/turn-offload-steal.md

@@ -0,0 +1,6 @@
+---
+'@truefoundry/trueforge': patch
+'@truefoundry/trueforge-core': patch
+---
+
+Rebuild a paused turn on its owner, offload it after a keepalive, and steal it only when the owner's heartbeat is gone.

.github/fern/openapi/openapi.json

@@ -10055,7 +10055,17 @@
                 }
               }
             },
-            "description": "Turn is terminal, or an event id already exists."
+            "description": "Turn is terminal, running and not accepting events, or an event id already exists."
+          },
+          "500": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/RequestErrorResponse"
+                }
+              }
+            },
+            "description": "The executor ran the request and failed while applying the events."
           }
         },
         "summary": "Create turn events",

docs/openapi.json

@@ -10055,7 +10055,17 @@
                 }
               }
             },
-            "description": "Turn is terminal, or an event id already exists."
+            "description": "Turn is terminal, running and not accepting events, or an event id already exists."
+          },
+          "500": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/RequestErrorResponse"
+                }
+              }
+            },
+            "description": "The executor ran the request and failed while applying the events."
           }
         },
         "summary": "Create turn events",

packages/assistant-ui-runtime/src/convertTurnMessages.ts

@@ -653,58 +653,52 @@ async function resolveSessionTip(options: {
   sessionId: string;
   itemsAsc: readonly SessionEventItem[];
 }): Promise<SessionTip> {
-  const { server, sessionId, itemsAsc } = options;
-  const open = findOpenTurnCreated(itemsAsc);
-  if (open != null) {
-    const continuationInput = open.event.input ?? [];
-    if (typeof server.getTurn === 'function') {
-      try {
-        const turn = await server.getTurn({
-          sessionId,
-          turnId: open.turnId,
-        });
-        if (turn.state.status === TURN_STATUS.RUNNING || turn.state.status === TURN_STATUS.PAUSED) {
-          return {
-            runningTurn:
-              open.update?.state.status === TURN_STATUS.PAUSED && turn.state.status === TURN_STATUS.RUNNING
-                ? turnFromCreatedEvent({
-                    sessionId,
-                    turnId: open.turnId,
-                    event: open.event,
-                    update: open.update,
-                  })
-                : turn,
-            ...(open.sandboxId != null ? { sandboxId: open.sandboxId } : {}),
-            continuationInput: turn.input ?? continuationInput,
-          };
-        }
-        // Tip finished between listEvents and getTurn: nothing to
-        // resume, but its answers still belong in the fold, and the turn
-        // itself must be committed so the painted user message stays.
-        return { finishedTurn: turn, continuationInput: turn.input ?? continuationInput };
-      } catch {
-        // getTurn failed; synthesize from the open turn.created below.
-      }
-    }
-    return {
-      runningTurn: turnFromCreatedEvent({
+  const open = findOpenTurnCreated(options.itemsAsc);
+  if (open == null) {
+    return { continuationInput: [] };
+  }
+
+  const { server, sessionId } = options;
+  const continuationInput = open.event.input ?? [];
+  if (typeof server.getTurn === 'function') {
+    try {
+      const turn = await server.getTurn({
         sessionId,
         turnId: open.turnId,
-        event: open.event,
-        ...(open.update != null ? { update: open.update } : {}),
-      }),
-      ...(open.sandboxId != null ? { sandboxId: open.sandboxId } : {}),
-      continuationInput,
-    };
+      });
+      if (turn.state.status === TURN_STATUS.RUNNING || turn.state.status === TURN_STATUS.PAUSED) {
+        return {
+          runningTurn:
+            open.update?.state.status === TURN_STATUS.PAUSED && turn.state.status === TURN_STATUS.RUNNING
+              ? turnFromCreatedEvent({
+                  sessionId,
+                  turnId: open.turnId,
+                  event: open.event,
+                  update: open.update,
+                })
+              : turn,
+          ...(open.sandboxId != null ? { sandboxId: open.sandboxId } : {}),
+          continuationInput: turn.input ?? continuationInput,
+        };
+      }
+      // Tip finished between listEvents and getTurn: nothing to
+      // resume, but its answers still belong in the fold, and the turn
+      // itself must be committed so the painted user message stays.
+      return { finishedTurn: turn, continuationInput: turn.input ?? continuationInput };
+    } catch {
+      // getTurn failed; synthesize from the open turn.created below.
+    }
   }
-
-  // Hosts that exclude the running turn from listEvents still surface it as
-  // the first row of listTurns when that API is tip-first.
-  const turnsPage = await server.listTurns({ sessionId, limit: 1 });
-  const tip = turnsPage.data[0];
-  return tip?.state.status === TURN_STATUS.RUNNING || tip?.state.status === TURN_STATUS.PAUSED
-    ? { runningTurn: tip, continuationInput: tip.input ?? [] }
-    : { continuationInput: [] };
+  return {
+    runningTurn: turnFromCreatedEvent({
+      sessionId,
+      turnId: open.turnId,
+      event: open.event,
+      ...(open.update != null ? { update: open.update } : {}),
+    }),
+    ...(open.sandboxId != null ? { sandboxId: open.sandboxId } : {}),
+    continuationInput,
+  };
 }
 
 /**

packages/assistant-ui-runtime/src/useTrueForgeAgentMessages.ts

@@ -999,8 +999,7 @@ export function useTrueForgeAgentMessages({
         if (!runStreamStarted) {
           // A supersede already aborted the live stream. A send that fails
           // before it installs a message must not cancel the painted open tip.
-          const hasIncompleteStream =
-            snapshotRef.current.activeStream != null && snapshotRef.current.activeStream.streamComplete !== true;
+          const hasIncompleteStream = isIncompleteActiveStream(snapshotRef.current);
           if ('userMessage' in options && (pendingUserWasSet || hasIncompleteStream)) {
             const abandoned = abandonInFlightClientTurn(commitActiveStream(snapshotRef.current));
             snapshotRef.current = abandoned;

packages/assistant-ui-runtime/test/convertTurnMessages.test.ts

@@ -1699,6 +1699,7 @@ describe('convertTurnMessages', () => {
           status: { type: 'incomplete', reason: 'cancelled' },
           metadata: { custom: { cancellationReason: 'abandoned' } },
           sequenceNumber: 1,
+          turnState: { status: 'cancelled', reason: 'abandoned', completedAt: createdAt },
         },
       ]);
     });