fix(agent): recover turns through unified provider control

.changeset/steady-agent-recovery.md

@@ -0,0 +1,5 @@
+---
+"akeru-bot": patch
+---
+
+Recover interrupted chats after server restarts and route supported providers through the unified Akeru agent controller.

apps/mobile/src/features/threads/ThreadDetailScreen.tsx

@@ -32,6 +32,8 @@ import {
   AppState,
   Keyboard,
   Platform,
+  Pressable,
+  Text,
   useWindowDimensions,
   View,
   type GestureResponderEvent,
@@ -116,6 +118,9 @@ export interface ThreadDetailScreenProps {
   readonly onNativePasteImages: (uris: ReadonlyArray<string>) => Promise<void>;
   readonly onRemoveDraftImage: (imageId: string) => void;
   readonly onStopThread: () => void;
+  readonly onResumeThread: () => void;
+  readonly canResumeThread: boolean;
+  readonly resumingThread: boolean;
   readonly onSendMessage: () => Promise<MessageId | null>;
   readonly onReconnectEnvironment: () => void;
   readonly onUpdateThreadModelSelection: (modelSelection: ModelSelection) => void;
@@ -695,6 +700,25 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread
               </Animated.View>
             ) : null}
             <View className="w-full self-center" style={{ maxWidth: contentMaxWidth }}>
+              {props.canResumeThread ? (
+                <View className="mx-4 mb-3 flex-row items-center gap-3 rounded-2xl border border-destructive/30 bg-destructive/10 px-4 py-3">
+                  <Text className="min-w-0 flex-1 text-sm text-foreground">
+                    {props.selectedThread.session?.lastError ??
+                      "The request stopped before it could finish."}
+                  </Text>
+                  <Pressable
+                    accessibilityRole="button"
+                    accessibilityLabel="Resume interrupted request"
+                    className="rounded-xl bg-primary px-3 py-2 disabled:opacity-50"
+                    disabled={props.resumingThread}
+                    onPress={props.onResumeThread}
+                  >
+                    <Text className="font-semibold text-primary-foreground">
+                      {props.resumingThread ? "Resuming…" : "Resume"}
+                    </Text>
+                  </Pressable>
+                </View>
+              ) : null}
               {props.activePendingApproval || props.activePendingUserInput ? (
                 <Animated.View
                   className="shrink-0 gap-3 px-4 pb-3"

apps/mobile/src/features/threads/ThreadRouteScreen.tsx

@@ -214,6 +214,8 @@ function ThreadRouteContent(
   const gitActions = useSelectedThreadGitActions();
   const requests = useSelectedThreadRequests();
   const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, "thread interrupt");
+  const resumeThreadTurn = useAtomCommand(threadEnvironment.resumeTurn, "thread resume");
+  const [resumingThread, setResumingThread] = useState(false);
   const navigation = useNavigation();
   const params = props.route.params;
   const environmentIdRaw = firstRouteParam(params.environmentId);
@@ -496,6 +498,14 @@ function ThreadRouteContent(
       },
     });
   }, [interruptThreadTurn, selectedThread]);
+  const handleResumeThread = useCallback(() => {
+    if (!selectedThread || resumingThread) return;
+    setResumingThread(true);
+    void resumeThreadTurn({
+      environmentId: selectedThread.environmentId,
+      input: { threadId: selectedThread.id },
+    }).finally(() => setResumingThread(false));
+  }, [resumeThreadTurn, resumingThread, selectedThread]);
 
   const handleOpenTerminal = useCallback(
     (nextTerminalId?: string | null) => {
@@ -796,6 +806,15 @@ function ThreadRouteContent(
           onRemoveDraftImage={composer.onRemoveDraftImage}
           serverConfig={serverConfig}
           onStopThread={handleStopThread}
+          onResumeThread={handleResumeThread}
+          canResumeThread={
+            selectedThread.session?.status === "error" &&
+            (selectedThread.latestTurn?.state === "error" ||
+              selectedThread.latestTurn?.state === "interrupted" ||
+              (selectedThread.latestTurn === null &&
+                selectedThreadDetail?.messages.at(-1)?.role === "user"))
+          }
+          resumingThread={resumingThread}
           onSendMessage={composer.onSendMessage}
           onReconnectEnvironment={handleReconnectEnvironment}
           onUpdateThreadModelSelection={composer.onUpdateModelSelection}

apps/server/integration/TransferBudgetReport.integration.ts

@@ -41,6 +41,7 @@ const TRANSFER_BUDGET = {
 export const TRANSFER_BUDGETS: Readonly<Record<string, ProviderTransferBudget>> = {
   codex: TRANSFER_BUDGET,
   claudeAgent: TRANSFER_BUDGET,
+  opencode: TRANSFER_BUDGET,
 };
 
 function totalWireBytes(run: TransferBudgetRun): number {

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -1669,6 +1669,19 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
         });
         return;
       }
+      if (event.type === "thread.turn-resume-requested") {
+        const existing = yield* projectionThreadSessionRepository.getByThreadId({
+          threadId: event.payload.threadId,
+        });
+        if (Option.isNone(existing)) return;
+        yield* projectionThreadSessionRepository.upsert({
+          ...existing.value,
+          status: "starting",
+          lastError: null,
+          updatedAt: event.payload.createdAt,
+        });
+        return;
+      }
       if (event.type !== "thread.session-set") {
         return;
       }

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts

@@ -72,6 +72,7 @@ import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionTh
 import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
 import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";
 import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";
+import { ProjectionPendingTurnStart } from "../../persistence/Services/ProjectionTurns.ts";
 import {
   decodeThreadDetailPageCursor,
   encodeThreadDetailPageCursor,
@@ -938,6 +939,27 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
       `,
   });
 
+  const listPendingTurnStartRows = SqlSchema.findAll({
+    Request: Schema.Void,
+    Result: ProjectionPendingTurnStart,
+    execute: () =>
+      sql`
+        SELECT
+          thread_id AS "threadId",
+          pending_message_id AS "messageId",
+          responding_bot_id AS "respondingBotId",
+          source_proposed_plan_thread_id AS "sourceProposedPlanThreadId",
+          source_proposed_plan_id AS "sourceProposedPlanId",
+          requested_at AS "requestedAt"
+        FROM projection_turns
+        WHERE turn_id IS NULL
+          AND state = 'pending'
+          AND pending_message_id IS NOT NULL
+          AND checkpoint_turn_count IS NULL
+        ORDER BY requested_at ASC, thread_id ASC
+      `,
+  });
+
   const listActiveLatestTurnRows = SqlSchema.findAll({
     Request: Schema.Void,
     Result: ProjectionLatestTurnDbRowSchema,
@@ -3287,6 +3309,19 @@ pending_approval_requests AS (
     }));
   });
 
+  const listPendingTurnStarts = Effect.fn("ProjectionSnapshotQuery.listPendingTurnStarts")(
+    function* () {
+      return yield* listPendingTurnStartRows(undefined).pipe(
+        Effect.mapError(
+          toPersistenceSqlOrDecodeError(
+            "ProjectionSnapshotQuery.listPendingTurnStarts:query",
+            "ProjectionSnapshotQuery.listPendingTurnStarts:decodeRows",
+          ),
+        ),
+      );
+    },
+  );
+
   // Contiguous turn range bounding a windowed detail read; undefined loads the
   // full thread. Resolved from a window request inside the snapshot
   // transaction (see getThreadDetailSnapshot).
@@ -3714,6 +3749,7 @@ pending_approval_requests AS (
     getThreadShellById,
     getThreadRuntimeContext,
     getTurnStartMessage,
+    listPendingTurnStarts,
     getThreadDetailById,
     getThreadDetailSnapshot,
   } satisfies ProjectionSnapshotQueryShape;