diff --git a/.changeset/steady-agent-recovery.md b/.changeset/steady-agent-recovery.md new file mode 100644 index 000000000000..3e46f512fe6d --- /dev/null +++ b/.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. diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 80c82e71a316..971e983ee6bc 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/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) => Promise; readonly onRemoveDraftImage: (imageId: string) => void; readonly onStopThread: () => void; + readonly onResumeThread: () => void; + readonly canResumeThread: boolean; + readonly resumingThread: boolean; readonly onSendMessage: () => Promise; readonly onReconnectEnvironment: () => void; readonly onUpdateThreadModelSelection: (modelSelection: ModelSelection) => void; @@ -695,6 +700,25 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread ) : null} + {props.canResumeThread ? ( + + + {props.selectedThread.session?.lastError ?? + "The request stopped before it could finish."} + + + + {props.resumingThread ? "Resuming…" : "Resume"} + + + + ) : null} {props.activePendingApproval || props.activePendingUserInput ? ( { + 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} diff --git a/apps/server/integration/TransferBudgetReport.integration.ts b/apps/server/integration/TransferBudgetReport.integration.ts index f773b5b8b844..2576ddedbd94 100644 --- a/apps/server/integration/TransferBudgetReport.integration.ts +++ b/apps/server/integration/TransferBudgetReport.integration.ts @@ -41,6 +41,7 @@ const TRANSFER_BUDGET = { export const TRANSFER_BUDGETS: Readonly> = { codex: TRANSFER_BUDGET, claudeAgent: TRANSFER_BUDGET, + opencode: TRANSFER_BUDGET, }; function totalWireBytes(run: TransferBudgetRun): number { diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 3a8ef58a95b8..3bf907a1c509 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/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; } diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 7a823f7458a3..c4a46ae28856 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/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; diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 21282429e7fc..3c12af15c634 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -189,12 +189,21 @@ describe("ProviderCommandReactor", () => { readonly requiresNewThreadForModelChange?: boolean; readonly titleRegenerationCompletionDispatchFailures?: number; readonly titleRegenerationBeforeStart?: "one" | "two"; + readonly turnStartBeforeReactor?: boolean; + readonly runningTurnBeforeReactor?: boolean; + readonly resumeBeforeReactor?: boolean; + readonly replayPersistedResumeOnSubscribe?: boolean; + readonly pendingRequestBeforeReactor?: "approval" | "user-input"; readonly interruptTurnEffect?: () => Effect.Effect; readonly interruptTurnRemovesSession?: boolean; readonly stopSessionEffect?: () => Effect.Effect; readonly startSessionEffect?: ( session: ProviderSession, ) => Effect.Effect; + readonly sendTurnEffect?: () => Effect.Effect< + { readonly threadId: ThreadId; readonly turnId: TurnId }, + ProviderAdapterRequestError + >; readonly botEngine?: { readonly provider: string; readonly model: string } | null; readonly botUsageCap?: { readonly unit: "tokens"; readonly limit: number } | null; readonly bindTurnFailure?: boolean; @@ -277,10 +286,12 @@ describe("ProviderCommandReactor", () => { ); }); const sendTurn = vi.fn((_: unknown) => - Effect.succeed({ - threadId: ThreadId.make("thread-1"), - turnId: asTurnId("turn-1"), - }), + input?.sendTurnEffect + ? input.sendTurnEffect() + : Effect.succeed({ + threadId: ThreadId.make("thread-1"), + turnId: asTurnId("turn-1"), + }), ); const interruptTurn = vi.fn((interruptInput: unknown) => (input?.interruptTurnEffect?.() ?? Effect.void).pipe( @@ -490,7 +501,33 @@ describe("ProviderCommandReactor", () => { get streamDomainEvents() { return engine.streamDomainEvents; }, - subscribeDomainEvents: engine.subscribeDomainEvents, + subscribeDomainEvents: + input?.replayPersistedResumeOnSubscribe === true + ? engine.subscribeDomainEvents.pipe( + Effect.flatMap((liveEvents) => + engine.latestSequence.pipe( + Effect.flatMap((throughSequence) => + Stream.runCollect( + engine.readThreadEvents({ + threadId: ThreadId.make("thread-1"), + fromSequenceExclusive: 0, + toSequenceInclusive: throughSequence, + limit: 500, + }), + ).pipe(Effect.orDie), + ), + Effect.map((events) => { + const resume = Array.from(events).findLast( + (event) => event.type === "thread.turn-resume-requested", + ); + return resume + ? Stream.concat(Stream.make(resume, resume), liveEvents) + : liveEvents; + }), + ), + ), + ) + : engine.subscribeDomainEvents, latestSequence: engine.latestSequence, } satisfies OrchestrationEngineService["Service"]; }), @@ -596,6 +633,111 @@ describe("ProviderCommandReactor", () => { createdAt: now, }), ); + await Effect.runPromise( + Effect.gen(function* () { + if ( + input?.turnStartBeforeReactor === true || + input?.runningTurnBeforeReactor === true || + input?.resumeBeforeReactor === true + ) { + yield* engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-turn-start-before-reactor"), + threadId: ThreadId.make("thread-1"), + message: { + messageId: asMessageId("user-message-before-reactor"), + role: "user", + text: "recover this persisted request", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: now, + }); + } + if (input?.runningTurnBeforeReactor === true || input?.resumeBeforeReactor === true) { + yield* engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-running-before-reactor"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + mcpServerIds: [], + activeTurnId: asTurnId("turn-before-reactor"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + if (input.pendingRequestBeforeReactor) { + const requestId = `${input.pendingRequestBeforeReactor}-before-restart`; + yield* engine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(`cmd-${requestId}`), + threadId: ThreadId.make("thread-1"), + activity: { + id: EventId.make(`activity-${requestId}`), + tone: input.pendingRequestBeforeReactor === "approval" ? "approval" : "info", + kind: + input.pendingRequestBeforeReactor === "approval" + ? "approval.requested" + : "user-input.requested", + summary: + input.pendingRequestBeforeReactor === "approval" + ? "Approval requested" + : "User input requested", + payload: + input.pendingRequestBeforeReactor === "approval" + ? { requestId, requestKind: "command" } + : { + requestId, + questions: [ + { + id: "choice", + header: "Choice", + question: "Continue?", + options: [{ label: "Yes", description: "Continue the work" }], + }, + ], + }, + turnId: asTurnId("turn-before-reactor"), + createdAt: now, + }, + createdAt: now, + }); + } + } + if (input?.resumeBeforeReactor === true) { + yield* engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-session-error-before-resume"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "error", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + mcpServerIds: [], + activeTurnId: null, + lastError: "Automatic recovery failed.", + updatedAt: now, + }, + createdAt: now, + }); + yield* engine.dispatch({ + type: "thread.turn.resume", + commandId: CommandId.make("cmd-resume-before-reactor"), + threadId: ThreadId.make("thread-1"), + createdAt: now, + }); + } + }), + ); if (input?.titleRegenerationBeforeStart === "two") { await Effect.runPromise( engine.dispatch({ @@ -729,6 +871,245 @@ describe("ProviderCommandReactor", () => { expect(thread?.session?.runtimeMode).toBe("approval-required"); }); + it("replays a persisted turn start that predates reactor startup exactly once", async () => { + const harness = await createHarness({ turnStartBeforeReactor: true }); + + await harness.drain(); + + expect(harness.startSession).toHaveBeenCalledTimes(1); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId: ThreadId.make("thread-1"), + input: "recover this persisted request", + }); + }); + + it("replays a persisted resume that predates reactor startup exactly once", async () => { + const harness = await createHarness({ resumeBeforeReactor: true }); + + await harness.drain(); + + expect(harness.startSession).toHaveBeenCalledTimes(1); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId: ThreadId.make("thread-1"), + input: expect.stringContaining("Resume the interrupted request"), + }); + }); + + it("deduplicates a persisted resume delivered by both startup recovery and the live stream", async () => { + const harness = await createHarness({ + resumeBeforeReactor: true, + replayPersistedResumeOnSubscribe: true, + }); + + await harness.drain(); + + expect(harness.startSession).toHaveBeenCalledTimes(1); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + }); + + it("continues a running turn after reactor startup without replaying the user prompt", async () => { + const harness = await createHarness({ runningTurnBeforeReactor: true }); + + await harness.drain(); + + expect(harness.startSession).toHaveBeenCalledTimes(1); + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId: ThreadId.make("thread-1"), + input: expect.stringContaining("server restarted"), + }); + expect(harness.sendTurn.mock.calls[0]?.[0]).not.toMatchObject({ + input: "recover this persisted request", + }); + }); + + it("marks an interrupted turn resumable when automatic recovery fails", async () => { + const harness = await createHarness({ + runningTurnBeforeReactor: true, + sendTurnEffect: () => + Effect.fail( + new ProviderAdapterRequestError({ + provider: ProviderDriverKind.make("codex"), + method: "thread.turn.start", + detail: "Provider was temporarily unavailable.", + }), + ), + }); + + await harness.drain(); + + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(thread?.session).toMatchObject({ + status: "error", + lastError: expect.stringContaining("Use Resume to continue"), + }); + expect(thread?.latestTurn?.state).toBe("error"); + }); + + it.each(["approval", "user-input"] as const)( + "expires a stale pending %s before automatic restart recovery", + async (requestKind) => { + const harness = await createHarness({ + runningTurnBeforeReactor: true, + pendingRequestBeforeReactor: requestKind, + }); + + await harness.drain(); + + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(thread?.activities).toContainEqual( + expect.objectContaining({ + kind: requestKind === "approval" ? "approval.resolved" : "user-input.resolved", + payload: expect.objectContaining({ + requestId: `${requestKind}-before-restart`, + outcome: "interrupted", + }), + }), + ); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + input: expect.stringContaining("server restarted"), + }); + }, + ); + + it("resumes an errored turn without adding or replaying a user message", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-turn-before-manual-resume"), + threadId: ThreadId.make("thread-1"), + message: { + messageId: asMessageId("user-message-before-manual-resume"), + role: "user", + text: "finish the migration", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: now, + }), + ); + await harness.drain(); + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-running-before-manual-resume"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "running", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + mcpServerIds: [], + activeTurnId: asTurnId("turn-before-manual-resume"), + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.make("cmd-error-before-manual-resume"), + threadId: ThreadId.make("thread-1"), + session: { + threadId: ThreadId.make("thread-1"), + status: "error", + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "approval-required", + mcpServerIds: [], + activeTurnId: null, + lastError: "Automatic recovery failed.", + updatedAt: now, + }, + createdAt: now, + }), + ); + harness.sendTurn.mockClear(); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.turn.resume", + commandId: CommandId.make("cmd-manual-resume"), + threadId: ThreadId.make("thread-1"), + createdAt: now, + }), + ); + await harness.drain(); + + expect(harness.sendTurn).toHaveBeenCalledTimes(1); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ + threadId: ThreadId.make("thread-1"), + input: expect.stringContaining("Resume the interrupted request"), + }); + expect(harness.sendTurn.mock.calls[0]?.[0]).not.toMatchObject({ + input: "finish the migration", + }); + const readModel = await harness.readModel(); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); + expect(thread?.messages.filter((message) => message.role === "user")).toHaveLength(1); + }); + + it("retries the original request when session startup failed before provider acceptance", async () => { + let attempts = 0; + const harness = await createHarness({ + startSessionEffect: (session) => { + attempts += 1; + return attempts === 1 + ? Effect.fail( + new ProviderAdapterRequestError({ + provider: ProviderDriverKind.make("codex"), + method: "startSession", + detail: "not connected", + cause: new Error("not connected"), + }), + ) + : Effect.succeed(session); + }, + }); + const now = "2026-01-01T00:00:00.000Z"; + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-pre-provider-failure"), + threadId: ThreadId.make("thread-1"), + message: { + messageId: asMessageId("message-pre-provider-failure"), + role: "user", + text: "send this only once", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: now, + }), + ); + await harness.drain(); + expect(harness.sendTurn).not.toHaveBeenCalled(); + + await harness.runEffect( + harness.engine.dispatch({ + type: "thread.turn.resume", + commandId: CommandId.make("cmd-resume-pre-provider-failure"), + threadId: ThreadId.make("thread-1"), + createdAt: now, + }), + ); + await harness.drain(); + + expect(harness.sendTurn).toHaveBeenCalledOnce(); + expect(harness.sendTurn.mock.calls[0]?.[0]).toMatchObject({ input: "send this only once" }); + }); + it.each([ ["codex", "gpt-5.6-sol"], ["claudeAgent", "claude-fable-5"], diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index b63bdeac3208..73dfa5c6d0cf 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -27,6 +27,7 @@ import * as Cache from "effect/Cache"; import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; import * as Duration from "effect/Duration"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; import * as Exit from "effect/Exit"; @@ -56,6 +57,7 @@ import { BotUsageLedger, } from "../../usage/BotUsageLedger.ts"; import { AgentController } from "../../provider/Services/AgentController.ts"; +import { ProviderSessionDirectory } from "../../provider/Services/ProviderSessionDirectory.ts"; import { ProviderRegistry } from "../../provider/Services/ProviderRegistry.ts"; import { ProjectionBotRepository } from "../../persistence/Services/ProjectionBots.ts"; import { ProjectionMcpServerRepository } from "../../persistence/Services/ProjectionMcpServers.ts"; @@ -101,6 +103,7 @@ type ProviderIntentEvent = Extract< | "thread.meta-updated" | "thread.runtime-mode-set" | "thread.turn-start-requested" + | "thread.turn-resume-requested" | "thread.turn-interrupt-requested" | "thread.approval-response-requested" | "thread.user-input-response-requested" @@ -132,15 +135,26 @@ function mapProviderSessionStatusToOrchestrationStatus( } } -const turnStartKeyForEvent = (event: ProviderIntentEvent): string => +const turnRequestKeyForEvent = (event: ProviderIntentEvent): string => event.commandId !== null ? `command:${event.commandId}` : `event:${event.eventId}`; -const HANDLED_TURN_START_KEY_MAX = 10_000; -const HANDLED_TURN_START_KEY_TTL = Duration.minutes(30); +const HANDLED_TURN_REQUEST_KEY_MAX = 10_000; +const HANDLED_TURN_REQUEST_KEY_TTL = Duration.minutes(30); const DEFAULT_RUNTIME_MODE: RuntimeMode = "full-access"; const MAX_REGENERATION_ATTACHMENTS = 4; const MAX_THREAD_TITLE_CONTEXT_CHARS = 8_000; const MAX_FIRST_USER_TITLE_CONTEXT_CHARS = 2_000; +const STARTUP_RECOVERY_INPUT = [ + "The Akeru server restarted while you were handling the current request.", + "Continue the existing request from the last durable conversation state.", + "Inspect the current workspace and tool state before acting, and do not repeat side effects that already completed.", + "If an approval or question was open, recreate it only if it is still needed.", +].join(" "); +const MANUAL_RECOVERY_INPUT = [ + "Resume the interrupted request from the last durable conversation state.", + "Inspect the current workspace and tool state before acting, and do not repeat side effects that already completed.", + "If an approval or question was open, recreate it only if it is still needed.", +].join(" "); const THREAD_TITLE_CONTEXT_TRUNCATION_MARKER = "[Earlier content truncated]\n\n"; const FIRST_USER_CONTEXT_TRUNCATION_MARKER = "\n[First user message truncated]"; @@ -353,6 +367,7 @@ const make = Effect.gen(function* () { const projectionBotRepository = yield* ProjectionBotRepository; const projectionMcpServerRepository = yield* ProjectionMcpServerRepository; const agentController = yield* AgentController; + const providerSessionDirectory = yield* Effect.serviceOption(ProviderSessionDirectory); const providerRegistry = yield* ProviderRegistry; const gitWorkflow = yield* GitWorkflowService; const fileSystem = yield* FileSystem.FileSystem; @@ -402,16 +417,16 @@ const make = Effect.gen(function* () { const serverCommandId = (tag: string) => crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); const serverEventId = () => crypto.randomUUIDv4.pipe(Effect.map(EventId.make)); - const handledTurnStartKeys = yield* Cache.make({ - capacity: HANDLED_TURN_START_KEY_MAX, - timeToLive: HANDLED_TURN_START_KEY_TTL, + const handledTurnRequestKeys = yield* Cache.make({ + capacity: HANDLED_TURN_REQUEST_KEY_MAX, + timeToLive: HANDLED_TURN_REQUEST_KEY_TTL, lookup: () => Effect.succeed(true), }); - const hasHandledTurnStartRecently = (key: string) => - Cache.getOption(handledTurnStartKeys, key).pipe( + const hasHandledTurnRequestRecently = (key: string) => + Cache.getOption(handledTurnRequestKeys, key).pipe( Effect.flatMap((cached) => - Cache.set(handledTurnStartKeys, key, true).pipe(Effect.as(Option.isSome(cached))), + Cache.set(handledTurnRequestKeys, key, true).pipe(Effect.as(Option.isSome(cached))), ), ); @@ -1126,7 +1141,26 @@ const make = Effect.gen(function* () { return { threadId: restartedSession.threadId, engine: desiredEngine }; } - const startedSession = yield* startProviderSession(undefined); + const persistedBinding = Option.isSome(providerSessionDirectory) + ? yield* providerSessionDirectory.value.getBinding(threadId).pipe( + Effect.catchCause((cause) => + Effect.logWarning("provider command reactor could not read resumable session binding", { + threadId, + cause: Cause.pretty(cause), + }).pipe(Effect.as(Option.none())), + ), + ) + : Option.none(); + const resumableBinding = Option.getOrUndefined(persistedBinding); + const resumeCursor = + resumableBinding?.provider === preferredProvider && + resumableBinding.providerInstanceId === desiredInstanceId && + resumableBinding.resumeCursor != null + ? resumableBinding.resumeCursor + : undefined; + const startedSession = yield* startProviderSession( + resumeCursor === undefined ? undefined : { resumeCursor }, + ); yield* bindSessionToThread(startedSession); threadBotWorkspaceKeys.set(threadId, botWorkspaceKey); threadMcpServers.set(threadId, mcpServers); @@ -1438,8 +1472,8 @@ const make = Effect.gen(function* () { const processTurnStartRequested = Effect.fn("processTurnStartRequested")(function* ( event: Extract, ) { - const key = turnStartKeyForEvent(event); - if (yield* hasHandledTurnStartRecently(key)) { + const key = turnRequestKeyForEvent(event); + if (yield* hasHandledTurnRequestRecently(key)) { return; } @@ -1730,6 +1764,73 @@ const make = Effect.gen(function* () { .pipe(Effect.catchCause(recoverInterruptFailure)); }); + const resumeInterruptedTurn = Effect.fn("resumeInterruptedTurn")(function* (input: { + readonly threadId: ThreadId; + readonly createdAt: string; + readonly messageText: string; + readonly attachments?: ReadonlyArray; + }) { + const thread = yield* resolveThreadShell(input.threadId); + if (!thread) return; + yield* buildSendTurnRequestForThread({ + threadId: thread.id, + messageText: input.messageText, + ...(input.attachments ? { attachments: input.attachments } : {}), + modelSelection: thread.modelSelection, + interactionMode: thread.interactionMode, + createdAt: input.createdAt, + }).pipe(Effect.flatMap(agentController.sendTurn)); + }); + + const processTurnResumeRequested = Effect.fn("processTurnResumeRequested")(function* ( + event: Extract, + ) { + const key = turnRequestKeyForEvent(event); + if (yield* hasHandledTurnRequestRecently(key)) { + return; + } + + const detail = Option.getOrUndefined( + yield* projectionSnapshotQuery.getThreadDetailById(event.payload.threadId), + ); + const failedBeforeProviderAccepted = detail?.latestTurn === null; + const originalRequest = failedBeforeProviderAccepted + ? detail.messages.findLast((message) => message.role === "user") + : undefined; + yield* resumeInterruptedTurn({ + threadId: event.payload.threadId, + createdAt: event.payload.createdAt, + messageText: originalRequest?.text ?? MANUAL_RECOVERY_INPUT, + ...(originalRequest?.attachments ? { attachments: originalRequest.attachments } : {}), + }).pipe( + Effect.tap(() => + Effect.logInfo("provider command reactor resumed turn after user request", { + threadId: event.payload.threadId, + }), + ), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : setThreadSessionErrorOnTurnStartFailure({ + threadId: event.payload.threadId, + detail: formatFailureDetail(cause), + createdAt: event.payload.createdAt, + }).pipe( + Effect.andThen( + appendProviderFailureActivity({ + threadId: event.payload.threadId, + kind: "provider.turn.start.failed", + summary: "Could not resume the request", + detail: formatFailureDetail(cause), + turnId: null, + createdAt: event.payload.createdAt, + }), + ), + ), + ), + ); + }); + const processApprovalResponseRequested = Effect.fn("processApprovalResponseRequested")(function* ( event: Extract, ) { @@ -1989,6 +2090,9 @@ const make = Effect.gen(function* () { case "thread.turn-start-requested": yield* processTurnStartRequested(event); return; + case "thread.turn-resume-requested": + yield* processTurnResumeRequested(event); + return; case "thread.turn-interrupt-requested": yield* processTurnInterruptRequested(event); return; @@ -2025,6 +2129,248 @@ const make = Effect.gen(function* () { const worker = yield* makeDrainableWorker(processDomainEventSafely); + const findPersistedTurnStart = Effect.fn("findPersistedTurnStart")(function* (input: { + readonly threadId: ThreadId; + readonly messageId: MessageId; + readonly throughSequence: number; + }) { + let cursor = 0; + let match: Extract | undefined; + while (cursor < input.throughSequence) { + const page = Array.from( + yield* Stream.runCollect( + orchestrationEngine.readThreadEvents({ + threadId: input.threadId, + fromSequenceExclusive: cursor, + toSequenceInclusive: input.throughSequence, + limit: 500, + }), + ), + ); + if (page.length === 0) break; + for (const event of page) { + if ( + event.type === "thread.turn-start-requested" && + event.payload.messageId === input.messageId + ) { + match = event; + } + } + const nextCursor = page.at(-1)?.sequence ?? cursor; + if (nextCursor <= cursor) break; + cursor = nextCursor; + } + return match; + }); + + const findPersistedTurnResume = Effect.fn("findPersistedTurnResume")(function* (input: { + readonly threadId: ThreadId; + readonly throughSequence: number; + }) { + let cursor = 0; + let match: Extract | undefined; + while (cursor < input.throughSequence) { + const page = Array.from( + yield* Stream.runCollect( + orchestrationEngine.readThreadEvents({ + threadId: input.threadId, + fromSequenceExclusive: cursor, + toSequenceInclusive: input.throughSequence, + limit: 500, + }), + ), + ); + if (page.length === 0) break; + for (const event of page) { + if (event.type === "thread.turn-resume-requested") match = event; + } + const nextCursor = page.at(-1)?.sequence ?? cursor; + if (nextCursor <= cursor) break; + cursor = nextCursor; + } + return match; + }); + + const settleStalePendingRequests = Effect.fn("settleStalePendingRequests")(function* (input: { + readonly threadId: ThreadId; + readonly activities: ReadonlyArray<{ + readonly kind: string; + readonly payload: unknown; + readonly turnId: TurnId | null; + }>; + readonly createdAt: string; + }) { + const pending = new Map< + string, + { readonly kind: "approval" | "user-input"; readonly turnId: TurnId | null } + >(); + for (const activity of input.activities) { + const payload = + typeof activity.payload === "object" && activity.payload !== null + ? (activity.payload as Record) + : null; + const requestId = typeof payload?.requestId === "string" ? payload.requestId : null; + if (!requestId) continue; + if (activity.kind === "approval.requested") { + pending.set(requestId, { kind: "approval", turnId: activity.turnId }); + } else if (activity.kind === "user-input.requested") { + pending.set(requestId, { kind: "user-input", turnId: activity.turnId }); + } else if (activity.kind === "approval.resolved" || activity.kind === "user-input.resolved") { + pending.delete(requestId); + } else if ( + activity.kind === "provider.approval.respond.failed" || + activity.kind === "provider.user-input.respond.failed" + ) { + const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : ""; + if (detail.includes("stale pending") || detail.includes("unknown pending")) { + pending.delete(requestId); + } + } + } + + yield* Effect.forEach( + pending, + ([requestId, request]) => + serverEventId().pipe( + Effect.flatMap((eventId) => + orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.make(`server:restart-request-expired:${eventId}`), + threadId: input.threadId, + activity: { + id: eventId, + tone: request.kind === "approval" ? "approval" : "info", + kind: request.kind === "approval" ? "approval.resolved" : "user-input.resolved", + summary: + request.kind === "approval" + ? "Approval expired after restart" + : "Question expired after restart", + payload: { + requestId, + outcome: "interrupted", + ...(request.kind === "user-input" ? { answers: {} } : {}), + }, + turnId: request.turnId, + createdAt: input.createdAt, + }, + createdAt: input.createdAt, + }), + ), + ), + { discard: true }, + ); + }); + + const recoverStartupProviderWork = Effect.fn("recoverStartupProviderWork")(function* () { + const throughSequence = yield* orchestrationEngine.latestSequence; + const initialReadModel = yield* projectionSnapshotQuery.getCommandReadModel(); + const pendingTurnStarts = projectionSnapshotQuery.listPendingTurnStarts + ? yield* projectionSnapshotQuery.listPendingTurnStarts() + : []; + const pendingThreadIds = new Set(pendingTurnStarts.map((pending) => String(pending.threadId))); + + for (const pending of pendingTurnStarts) { + const event = yield* findPersistedTurnStart({ + threadId: pending.threadId, + messageId: pending.messageId, + throughSequence, + }); + if (event) { + yield* worker.enqueue(event); + } else { + yield* Effect.logWarning("provider command reactor could not recover pending turn event", { + threadId: pending.threadId, + messageId: pending.messageId, + }); + } + } + + const pendingResumes = initialReadModel.threads.filter( + (thread) => + thread.deletedAt === null && + thread.archivedAt === null && + thread.session?.status === "starting" && + (thread.latestTurn === null || + thread.latestTurn.state === "error" || + thread.latestTurn.state === "interrupted"), + ); + for (const thread of pendingResumes) { + const event = yield* findPersistedTurnResume({ + threadId: thread.id, + throughSequence, + }); + if (event) { + pendingThreadIds.add(String(thread.id)); + yield* worker.enqueue(event); + } else { + yield* Effect.logWarning( + "provider command reactor could not recover pending resume event", + { + threadId: thread.id, + }, + ); + } + } + yield* worker.drain; + + const liveThreadIds = new Set( + (yield* agentController.listSessions()).map((session) => String(session.threadId)), + ); + const readModel = yield* projectionSnapshotQuery.getCommandReadModel(); + const interrupted = readModel.threads.filter( + (thread) => + thread.deletedAt === null && + thread.archivedAt === null && + !pendingThreadIds.has(String(thread.id)) && + !liveThreadIds.has(String(thread.id)) && + thread.latestTurn?.state === "running" && + thread.session !== null && + (thread.session.status === "starting" || + thread.session.status === "running" || + thread.session.activeTurnId !== null), + ); + + for (const thread of interrupted) { + const recoveredAt = DateTime.formatIso(yield* DateTime.now); + const threadDetail = yield* projectionSnapshotQuery + .getThreadDetailById(thread.id) + .pipe(Effect.map(Option.getOrUndefined)); + yield* settleStalePendingRequests({ + threadId: thread.id, + activities: threadDetail?.activities ?? [], + createdAt: recoveredAt, + }); + yield* resumeInterruptedTurn({ + threadId: thread.id, + messageText: STARTUP_RECOVERY_INPUT, + createdAt: recoveredAt, + }).pipe( + Effect.tap(() => + Effect.logInfo("provider command reactor resumed interrupted turn", { + threadId: thread.id, + previousTurnId: thread.latestTurn?.turnId, + }), + ), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : setThreadSessionErrorOnTurnStartFailure({ + threadId: thread.id, + detail: `Automatic recovery failed. Use Resume to continue. ${formatFailureDetail(cause)}`, + createdAt: recoveredAt, + }).pipe( + Effect.andThen( + Effect.logWarning("provider command reactor could not resume interrupted turn", { + threadId: thread.id, + cause: Cause.pretty(cause), + }), + ), + ), + ), + ); + } + }); + const start: ProviderCommandReactorShape["start"] = Effect.fn("start")(function* () { const interruptedTitleRegenerations = yield* findInterruptedThreadTitleRegenerations().pipe( Effect.catchCause((cause) => { @@ -2042,6 +2388,7 @@ const make = Effect.gen(function* () { (event.type === "thread.meta-updated" && event.payload.regenerateTitle === true) || event.type === "thread.runtime-mode-set" || event.type === "thread.turn-start-requested" || + event.type === "thread.turn-resume-requested" || event.type === "thread.turn-interrupt-requested" || event.type === "thread.approval-response-requested" || event.type === "thread.user-input-response-requested" || @@ -2056,9 +2403,18 @@ const make = Effect.gen(function* () { const domainEvents = yield* orchestrationEngine.subscribeDomainEvents; yield* forkParked(Stream.runForEach(domainEvents, processEvent)); - // The domain event stream is hot, so work pending before this reactor - // starts cannot be resumed. Correlated completions only clear the request - // captured here, leaving any newer request untouched. + yield* recoverStartupProviderWork().pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("provider command reactor startup recovery failed", { + cause: Cause.pretty(cause), + }), + ), + ); + + // Correlated completions only clear the request captured here, leaving any + // newer request untouched. const clearInterrupted = clearInterruptedThreadTitleRegenerations( interruptedTitleRegenerations, ).pipe( diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index b185b59e00e6..539393a5c352 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -33,6 +33,7 @@ import type * as Option from "effect/Option"; import type * as Effect from "effect/Effect"; import type { ProjectionRepositoryError } from "../../persistence/Errors.ts"; +import type { ProjectionPendingTurnStart } from "../../persistence/Services/ProjectionTurns.ts"; export interface ProjectionSnapshotCounts { readonly projectCount: number; @@ -230,6 +231,16 @@ export interface ProjectionSnapshotQueryShape { readonly messageId: MessageId; }) => Effect.Effect, ProjectionRepositoryError>; + /** + * Lists turn starts that were durably projected but have not yet been bound + * to a concrete provider turn. Startup recovery uses these rows to close the + * hot event-stream crash window. + */ + readonly listPendingTurnStarts?: () => Effect.Effect< + ReadonlyArray, + ProjectionRepositoryError + >; + /** * Read a single active thread detail snapshot by id. */ diff --git a/apps/server/src/orchestration/decider.resume.test.ts b/apps/server/src/orchestration/decider.resume.test.ts new file mode 100644 index 000000000000..d1998c67e464 --- /dev/null +++ b/apps/server/src/orchestration/decider.resume.test.ts @@ -0,0 +1,179 @@ +import { + CommandId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationReadModel, + type OrchestrationThread, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const NOW = "2026-09-15T12:00:00.000Z"; +const THREAD_ID = ThreadId.make("thread-resume"); + +function makeThread(status: "error" | "ready") { + return { + id: THREAD_ID, + projectId: ProjectId.make("project-1"), + botId: null, + groupId: null, + respondingBotId: null, + title: "Interrupted chat", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6-sol" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: { + turnId: TurnId.make("turn-1"), + state: status === "error" ? "error" : "completed", + requestedAt: NOW, + startedAt: NOW, + completedAt: status === "error" ? NOW : null, + assistantMessageId: null, + }, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: { + threadId: THREAD_ID, + status, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access", + activeTurnId: null, + lastError: status === "error" ? "Automatic recovery failed." : null, + updatedAt: NOW, + }, + } satisfies OrchestrationThread; +} + +function makeReadModel(thread: OrchestrationThread): OrchestrationReadModel { + return { ...createEmptyReadModel(NOW), threads: [thread] }; +} + +it.layer(NodeServices.layer)("turn resume decider", (it) => { + it.effect("emits an invisible resume request for an errored turn", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.resume", + commandId: CommandId.make("command-resume"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel: makeReadModel(makeThread("error")), + }); + + expect(event).toMatchObject({ + type: "thread.turn-resume-requested", + payload: { threadId: THREAD_ID, createdAt: NOW }, + }); + }), + ); + + it.effect("rejects resume when no interrupted turn exists", () => + Effect.gen(function* () { + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.resume", + commandId: CommandId.make("command-resume-completed"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel: makeReadModel(makeThread("ready")), + }).pipe(Effect.flip); + + expect(error._tag).toBe("OrchestrationCommandInvariantError"); + expect(error.message).toContain("does not have an interrupted request"); + }), + ); + + it.effect("reserves the interrupted turn before another resume can be accepted", () => + Effect.gen(function* () { + const readModel = makeReadModel(makeThread("error")); + const first = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.resume", + commandId: CommandId.make("command-resume-first"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel, + }); + if (!("type" in first) || first.type !== "thread.turn-resume-requested") { + return yield* Effect.die("Expected a resume request event"); + } + const reserved = yield* projectEvent(readModel, { + sequence: 1, + eventId: first.eventId, + aggregateKind: first.aggregateKind, + aggregateId: first.aggregateId, + occurredAt: first.occurredAt, + commandId: first.commandId, + causationEventId: first.causationEventId, + correlationId: first.correlationId, + metadata: first.metadata, + type: "thread.turn-resume-requested", + payload: { threadId: THREAD_ID, createdAt: NOW }, + }); + expect(reserved.threads[0]?.session?.status).toBe("starting"); + + const error = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.resume", + commandId: CommandId.make("command-resume-second"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel: reserved, + }).pipe(Effect.flip); + + expect(error.message).toContain("already active"); + }), + ); + + it.effect("allows retry when the provider rejected the request before creating a turn", () => + Effect.gen(function* () { + const thread = makeThread("error"); + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.resume", + commandId: CommandId.make("command-resume-pre-provider"), + threadId: THREAD_ID, + createdAt: NOW, + }, + readModel: makeReadModel({ + ...thread, + latestTurn: null, + messages: [ + { + id: "message-pre-provider" as never, + role: "user", + text: "Retry this exact request.", + turnId: null, + streaming: false, + createdAt: NOW, + updatedAt: NOW, + }, + ], + }), + }); + + expect(event).toMatchObject({ type: "thread.turn-resume-requested" }); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 2bc7c09782c5..72b258e9dedc 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -2766,6 +2766,50 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" ]; } + case "thread.turn.resume": { + const thread = yield* requireThread({ + readModel, + command, + threadId: command.threadId, + }); + const failedBeforeProviderAccepted = + thread.latestTurn === null && thread.session?.status === "error"; + if ( + !failedBeforeProviderAccepted && + (thread.latestTurn === null || + (thread.latestTurn.state !== "error" && thread.latestTurn.state !== "interrupted")) + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Chat '${command.threadId}' does not have an interrupted request to resume.`, + }); + } + if ( + thread.session !== null && + thread.session.status !== "error" && + thread.session.status !== "interrupted" && + thread.session.status !== "stopped" + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Chat '${command.threadId}' is already active.`, + }); + } + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.turn-resume-requested", + payload: { + threadId: command.threadId, + createdAt: command.createdAt, + }, + }; + } + case "thread.turn.interrupt": { yield* requireThread({ readModel, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index dc78ca83fb9f..cef3e363d715 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -26,6 +26,7 @@ import { RoutineRunningPayload, RoutineSkillAssignedPayload, RoutineSkillUnassignedPayload, + ThreadTurnResumeRequestedPayload, } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Schema from "effect/Schema"; @@ -1134,6 +1135,31 @@ export function projectEvent( })), ); + case "thread.turn-resume-requested": + return decodeForEvent( + ThreadTurnResumeRequestedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => { + const thread = findProjectedThread(nextBase.threads, payload.threadId); + if (!thread?.session) return nextBase; + return { + ...nextBase, + threads: updateThread(nextBase.threads, payload.threadId, { + session: { + ...thread.session, + status: "starting", + lastError: null, + updatedAt: payload.createdAt, + }, + updatedAt: payload.createdAt, + }), + }; + }), + ); + case "thread.session-set": return Effect.gen(function* () { const payload = yield* decodeForEvent( diff --git a/apps/server/src/provider/AkeruAgentInstructions.ts b/apps/server/src/provider/AkeruAgentInstructions.ts index f64be5c4a5e0..c017044ddb7e 100644 --- a/apps/server/src/provider/AkeruAgentInstructions.ts +++ b/apps/server/src/provider/AkeruAgentInstructions.ts @@ -18,32 +18,32 @@ export function createAkeruAgentInstructions(context: AkeruInstructionContext = const date = currentDate(context.now ?? DateTime.nowUnsafe()); return [ `You are ${name}, a sharp, curious general assistant. Today is ${date}.`, - "Treat each request on its own terms. Coding is one kind of work, not the default.", - "Write like a thoughtful human teammate. Match the user's tone and detail. Be warm without fawning and confident without hiding uncertainty.", - "Lead with the answer. Use plain, specific language, active voice, and natural rhythm. Use first person when it fits.", - "Have a point of view when the facts support one. Name real tradeoffs instead of flattening every answer into neutral pros and cons.", - "Keep structure proportional to the task. Prefer short prose. Use headings and lists only when they make the answer easier to use.", - "Silently reread before sending. Cut filler, puffery, canned praise, vague claims, repeated conclusions, and stock launch language. Rewrite generic sentences.", - "For copy, use concrete facts. Name every drawback the user gives you. Never invent or disguise a benefit.", - "Use periods and commas for asides. Never use em or en dashes, parenthetical asides, or spaced hyphens as dashes.", - "Start with substance instead of a greeting. End on the useful point instead of a generic offer to do more.", - "Use enabled plugin tools when they help with the request.", - "When preview_* tools are present, use them for browser work so the user can watch in the shared browser.", - "Call preview_status first. Call preview_open if no browser is attached, then use preview_navigate, preview_snapshot, and the focused preview interaction tools.", - "Prefer preview_* tools over browser_* tools when both are present.", - "Use workspace tools only when the task requires file or command work.", - "When asked about available tools, describe only tools present in the current turn.", - "Do not claim that a tool ran unless its result is present in this turn.", + "Treat each request on its own terms. Coding is not the default.", + "Write like a thoughtful human teammate. Match the user's tone, be direct, and name uncertainty.", + "Lead with the answer in plain, specific language. Keep structure proportional and have a point of view when facts support one.", + "Silently reread before sending. Cut filler, praise, vague claims, repetition, and stock phrases.", + "For copy, use concrete facts. Name every drawback the user gives you and never invent a benefit.", + "Never use em or en dashes. Start with substance and end on the useful point.", + "Use enabled plugin tools when helpful and workspace tools only for file or command work.", + "For browser work, call preview_status, then preview_open when needed. Prefer preview_* tools over browser_* tools.", + "Own the requested outcome. If a safe needed action is available, do it instead of only explaining it.", + "Carry multi-step work through implementation and proportionate verification while a safe next step remains.", + "Make reasonable, reversible assumptions. Ask only when the answer materially changes the result or requires new authority.", + "Investigate failures from evidence, preserve user work, retry safe failures, and exhaust useful alternatives before reporting a blocker.", + "Never report success before the outcome is complete. State what was verified and what remains unverified.", + "Describe only tools present now, and never claim a tool ran without a result.", "Use akeru_create_routine when the user asks for recurring or scheduled work.", - "Use akeru_list_routines before answering whether a routine exists or what state it is in, including enabled, disabled, paused, blocked, or failed.", - "When the user asks to delete routines, call akeru_list_routines, then call akeru_delete_routines with the selected IDs. The delete tool asks for confirmation, so do not ask for separate confirmation.", - "Routine output goes to the current chat unless the user names an enabled plugin such as Slack. Use the current device timezone unless the user names another timezone.", + "Use akeru_list_routines before reporting routine existence or state.", + "For deletion, list routines then call akeru_delete_routines. Its tool confirmation is sufficient.", + "Send routine output here unless the user names an enabled plugin. Use the device timezone by default.", ].join("\n"); } export const AKERU_BOT_TURN_INSTRUCTIONS = [ "Before you use a tool for a visible user request, first answer with one short plain-language sentence that acknowledges the request and says what you will do next.", - "During longer tool work, add one short plain-language status note after meaningful progress or a change in direction and before the next tool call. Do not narrate every tool call.", + "During longer tool work, add one short plain-language status note after meaningful progress, a discovered risk, a failed assumption, or a change in direction and before the next tool call. Do not narrate every tool call.", + "Keep working after a status note. A status update is not a final answer and must not replace the next useful action.", + "If a tool fails, inspect the failure and try a safe correction or alternate route before asking the user to intervene.", "Treat a hidden system reminder or automatic continuation as ongoing work, not a new user request. Skip the opening reply and continue with only useful status notes.", ].join("\n"); diff --git a/apps/server/src/provider/AkeruMastraHarness.test.ts b/apps/server/src/provider/AkeruMastraHarness.test.ts index 1cf8de00b86d..27b9d43d88bc 100644 --- a/apps/server/src/provider/AkeruMastraHarness.test.ts +++ b/apps/server/src/provider/AkeruMastraHarness.test.ts @@ -416,6 +416,58 @@ describe("AkeruMastraHarness", () => { expect(getCredential).toHaveBeenCalledWith("openai-codex"); }); + it("uses the selected instance key without reading the provider-wide credential", () => { + const authStorage = new AuthStorage("/tmp/akeru-unused-instance-auth.json"); + const getCredential = vi.fn(() => ({ + type: "api-key" as const, + access: "provider-wide-key", + })); + expect( + resolveAkeruMastraModel( + "xai/grok-code-fast-1", + authStorage, + undefined, + undefined, + undefined, + getCredential, + { + environment: { + XAI_API_KEY: "instance-key", + XAI_BASE_URL: "https://instance.example/v1", + }, + instanceEnvironment: { + XAI_API_KEY: "instance-key", + XAI_BASE_URL: "https://instance.example/v1", + }, + useSavedCredential: false, + }, + ), + ).toMatchObject({ modelId: "grok-code-fast-1", provider: "xai.chat" }); + expect(getCredential).not.toHaveBeenCalled(); + }); + + it("does not leak provider-wide credentials into an isolated instance", () => { + const authStorage = new AuthStorage("/tmp/akeru-unused-isolated-auth.json"); + const getCredential = vi.fn(() => ({ + type: "api-key" as const, + access: "provider-wide-key", + })); + assert.throws( + () => + resolveAkeruMastraModel( + "anthropic/claude-fable-5", + authStorage, + undefined, + undefined, + undefined, + getCredential, + { environment: {}, instanceEnvironment: {}, useSavedCredential: false }, + ), + "has no API key or auth token transport", + ); + expect(getCredential).not.toHaveBeenCalled(); + }); + it("keeps Kimi model names on the Kimi subscription transport", () => { const authStorage = new AuthStorage("/tmp/akeru-unused-auth.json"); assert.equal( @@ -456,6 +508,36 @@ describe("AkeruMastraHarness", () => { ); }); + it("uses an isolated OpenCode Go inline connection", () => { + const authStorage = new AuthStorage("/tmp/akeru-unused-inline-opencode-auth.json"); + const getCredential = vi.fn(() => ({ + type: "api-key" as const, + access: "provider-wide-key", + })); + const environment = { + OPENCODE_CONFIG_CONTENT: JSON.stringify({ + provider: { + "opencode-go": { + options: { apiKey: "inline-key", baseURL: "https://inline.example/v1" }, + }, + }, + }), + }; + assert.deepInclude( + resolveAkeruMastraModel( + "opencode-go/gpt-5.6-luna", + authStorage, + undefined, + undefined, + undefined, + getCredential, + { environment, instanceEnvironment: environment, useSavedCredential: false }, + ), + { provider: "opencode-go.responses", modelId: "gpt-5.6-luna" }, + ); + expect(getCredential).not.toHaveBeenCalled(); + }); + it("builds a compact, human prompt with the bot name and current date", () => { const instructions = createAkeruAgentInstructions({ name: " Research\nBot ", @@ -470,6 +552,9 @@ describe("AkeruMastraHarness", () => { assert.include(instructions, "Never use em or en dashes"); assert.include(instructions, "enabled plugin tools"); assert.include(instructions, "Prefer preview_* tools over browser_* tools"); + assert.include(instructions, "Own the requested outcome"); + assert.include(instructions, "Carry multi-step work through implementation"); + assert.include(instructions, "Never report success before"); assert.include(instructions, "akeru_list_routines"); assert.notInclude(instructions, "—"); assert.notInclude(instructions, "coding agent"); diff --git a/apps/server/src/provider/AkeruMastraHarness.ts b/apps/server/src/provider/AkeruMastraHarness.ts index dee6b9d0dc76..61e837664e80 100644 --- a/apps/server/src/provider/AkeruMastraHarness.ts +++ b/apps/server/src/provider/AkeruMastraHarness.ts @@ -2,7 +2,11 @@ import * as NodeURL from "node:url"; import { AuthStorage } from "@mastra/code-sdk/auth/storage"; +import { opencodeClaudeMaxProvider } from "@mastra/code-sdk/providers/claude-max"; import { openaiCodexProvider } from "@mastra/code-sdk/providers/openai-codex"; +import { xaiProvider } from "@mastra/code-sdk/providers/xai"; +import { createAnthropic } from "@ai-sdk/anthropic"; +import { createOpenAICompatible } from "@ai-sdk/openai-compatible"; import { isThinkingLevelSetting } from "@mastra/code-sdk/thinking"; import type { ToolsInput } from "@mastra/core/agent"; import { @@ -144,6 +148,7 @@ const routineDeleteResultSchema = z.object({ export type AkeruRoutineDeleteResult = z.infer; export interface AkeruMastraState { + readonly providerInstanceId?: string; readonly projectPath?: string; readonly yolo?: boolean; readonly botConversation?: boolean; @@ -191,6 +196,13 @@ export interface AkeruMastraHarnessOptions { readonly getKimiAccess?: () => Promise; readonly getOpenCodeGoApiKey?: () => Promise; readonly getSubscriptionApiKey?: SubscriptionAuthService["getApiKeyCredential"]; + readonly getModelConnection?: (providerInstanceId: string) => + | { + readonly environment: NodeJS.ProcessEnv; + readonly instanceEnvironment: NodeJS.ProcessEnv; + readonly useSavedCredential: boolean; + } + | undefined; readonly memoryDbPath: string; readonly startMemoryCall?: (input: { readonly threadId: string; @@ -319,6 +331,19 @@ function controllerModelOptions(requestContext: RequestContext): AkeruMastraStat : undefined; } +function controllerModelConnection( + requestContext: RequestContext, + getModelConnection: AkeruMastraHarnessOptions["getModelConnection"], +) { + const state = controllerContext(requestContext)?.state; + if (typeof state !== "object" || state === null || !("providerInstanceId" in state)) { + return undefined; + } + return typeof state.providerInstanceId === "string" + ? getModelConnection?.(state.providerInstanceId) + : undefined; +} + function controllerResourceId(requestContext: RequestContext): string | undefined { const value = controllerContext(requestContext)?.resourceId; return typeof value === "string" ? value : undefined; @@ -441,6 +466,8 @@ export async function createAkeruMastraMemory( const MASTRA_MODEL_PREFIX = { codex: "openai", + claudeAgent: "anthropic", + grok: "xai", kimi: "kimi-for-coding", opencodeGo: "opencode-go", } as const; @@ -453,6 +480,32 @@ export function mastraModelId(provider: ProviderDriverKind, model: string): stri return trimmed.startsWith(token) ? trimmed : `${token}${trimmed}`; } +export function openCodeGoInlineConnection(environment: NodeJS.ProcessEnv | undefined): { + readonly apiKey?: string; + readonly baseUrl?: string; +} { + const content = environment?.OPENCODE_CONFIG_CONTENT?.trim(); + if (!content) return {}; + try { + const parsed = JSON.parse(content) as { + readonly provider?: { + readonly "opencode-go"?: { + readonly options?: { readonly apiKey?: unknown; readonly baseURL?: unknown }; + }; + }; + }; + const options = parsed.provider?.["opencode-go"]?.options; + const apiKey = typeof options?.apiKey === "string" ? options.apiKey.trim() : ""; + const baseUrl = typeof options?.baseURL === "string" ? options.baseURL.trim() : ""; + return { + ...(apiKey ? { apiKey } : {}), + ...(baseUrl ? { baseUrl } : {}), + }; + } catch { + return {}; + } +} + export function resolveAkeruMastraModel( modelId: string, authStorage: AuthStorage, @@ -460,13 +513,35 @@ export function resolveAkeruMastraModel( getOpenCodeGoApiKey?: () => Promise, modelOptions?: AkeruMastraState["modelOptions"], getSubscriptionApiKey?: SubscriptionAuthService["getApiKeyCredential"], + connection?: { + readonly environment: NodeJS.ProcessEnv; + readonly instanceEnvironment: NodeJS.ProcessEnv; + readonly useSavedCredential: boolean; + }, ) { const trimmed = modelId.trim(); + const environment = connection?.useSavedCredential + ? connection.environment + : connection?.instanceEnvironment; + const useSavedCredential = connection?.useSavedCredential !== false; if (trimmed.startsWith("openai/")) { - if (getSubscriptionApiKey?.("openai-codex")) { - return akeruOpenAIProvider(trimmed.slice("openai/".length), () => - getSubscriptionApiKey("openai-codex"), - ); + const instanceApiKey = environment?.OPENAI_API_KEY?.trim(); + const getCredential = instanceApiKey + ? () => ({ + type: "api-key" as const, + access: instanceApiKey, + ...(environment?.OPENAI_BASE_URL?.trim() + ? { baseUrl: environment.OPENAI_BASE_URL.trim() } + : {}), + }) + : useSavedCredential + ? () => getSubscriptionApiKey?.("openai-codex") + : undefined; + if (getCredential?.()) { + return akeruOpenAIProvider(trimmed.slice("openai/".length), () => getCredential()); + } + if (!useSavedCredential) { + throw new Error("This Codex instance has no OPENAI_API_KEY transport for Akeru Mastra."); } const reasoningEffort = modelOptions?.reasoningEffort; return openaiCodexProvider(trimmed.slice("openai/".length), { @@ -474,16 +549,81 @@ export function resolveAkeruMastraModel( ...(isThinkingLevelSetting(reasoningEffort) ? { thinkingLevel: reasoningEffort } : {}), }); } + if (trimmed.startsWith("anthropic/")) { + const model = trimmed.slice("anthropic/".length); + const instanceApiKey = environment?.ANTHROPIC_API_KEY?.trim(); + const instanceAuthToken = + environment?.ANTHROPIC_AUTH_TOKEN?.trim() ?? environment?.CLAUDE_CODE_OAUTH_TOKEN?.trim(); + if (instanceApiKey || instanceAuthToken) { + return createAnthropic({ + ...(instanceApiKey ? { apiKey: instanceApiKey } : { authToken: instanceAuthToken! }), + ...(environment?.ANTHROPIC_BASE_URL?.trim() + ? { baseURL: environment.ANTHROPIC_BASE_URL.trim() } + : {}), + })(model); + } + const credential = useSavedCredential ? getSubscriptionApiKey?.("anthropic") : undefined; + if (credential) { + return createAnthropic({ + apiKey: credential.access, + ...(credential.baseUrl ? { baseURL: credential.baseUrl } : {}), + })(model); + } + if (!useSavedCredential) { + throw new Error( + "This Claude instance has no API key or auth token transport for Akeru Mastra.", + ); + } + return opencodeClaudeMaxProvider(model, { authStorage }); + } + if (trimmed.startsWith("xai/")) { + const model = trimmed.slice("xai/".length); + const instanceApiKey = environment?.XAI_API_KEY?.trim(); + const credential = instanceApiKey + ? { + access: instanceApiKey, + baseUrl: environment?.XAI_BASE_URL?.trim() || undefined, + } + : useSavedCredential + ? getSubscriptionApiKey?.("xai") + : undefined; + if (credential) { + return createOpenAICompatible({ + name: "xai", + apiKey: credential.access, + baseURL: credential.baseUrl ?? "https://api.x.ai/v1", + })(model); + } + if (!useSavedCredential) { + throw new Error("This Grok instance has no XAI_API_KEY transport for Akeru Mastra."); + } + return xaiProvider(model, { authStorage }); + } if (trimmed.startsWith("kimi-for-coding/")) { + if (!useSavedCredential) { + throw new Error( + "Custom Kimi instance credentials are not supported by the Akeru Mastra transport.", + ); + } if (!getKimiAccess) throw new Error("Kimi For Coding subscription access is unavailable."); return akeruKimiProvider(trimmed.slice("kimi-for-coding/".length), getKimiAccess); } if (trimmed.startsWith("opencode-go/")) { - if (!getOpenCodeGoApiKey) throw new Error("OpenCode Go subscription access is unavailable."); + const inlineConnection = openCodeGoInlineConnection(environment); + const instanceApiKey = environment?.OPENCODE_API_KEY?.trim() || inlineConnection.apiKey; + const resolveApiKey = instanceApiKey + ? async () => instanceApiKey + : useSavedCredential + ? getOpenCodeGoApiKey + : undefined; + if (!resolveApiKey) throw new Error("OpenCode Go subscription access is unavailable."); return akeruOpenCodeGoProvider( trimmed.slice("opencode-go/".length), - getOpenCodeGoApiKey, - () => getSubscriptionApiKey?.("opencode-go")?.baseUrl, + resolveApiKey, + () => + environment?.OPENCODE_BASE_URL?.trim() || + inlineConnection.baseUrl || + (useSavedCredential ? getSubscriptionApiKey?.("opencode-go")?.baseUrl : undefined), ); } throw new Error(`Mastra has no subscription transport for model '${modelId}'.`); @@ -879,6 +1019,7 @@ export async function createAkeruMastraHarness( options.getOpenCodeGoApiKey, controllerModelOptions(requestContext), options.getSubscriptionApiKey, + controllerModelConnection(requestContext, options.getModelConnection), ), tools: ({ requestContext }) => resolveAkeruTools(requestContext, options), memory: observationalMemory.memory, diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index 1c626aaff48e..23155f54776b 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -26,6 +26,7 @@ import { ChildProcessSpawner } from "effect/unstable/process"; import { makeClaudeTextGeneration } from "../../textGeneration/ClaudeTextGeneration.ts"; import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; import { ServerConfig } from "../../config.ts"; +import { instanceUsesSavedCredential } from "../../subscription-auth/runtime.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import { ProviderDriverError } from "../Errors.ts"; import { makeClaudeAdapter } from "../Layers/ClaudeAdapter.ts"; @@ -43,6 +44,7 @@ import { type ProviderInstance, } from "../ProviderDriver.ts"; import type { ServerProviderDraft } from "../providerSnapshot.ts"; +import { explicitProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; import { mergeSubscriptionInstanceEnvironment } from "../../subscription-auth/runtime.ts"; import { enrichProviderSnapshotWithVersionAdvisory, @@ -236,6 +238,15 @@ export const ClaudeDriver: ProviderDriver = { displayName, accentColor, enabled, + mastraConnection: { + environment: processEnv, + instanceEnvironment: explicitProviderInstanceEnvironment(environment), + useSavedCredential: instanceUsesSavedCredential("anthropic", { + driver: DRIVER_KIND, + environment, + config, + }), + }, snapshot, adapter, textGeneration, diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 80a848c7ce3b..314beaa6b237 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -34,6 +34,7 @@ import { makeCodexTextGeneration } from "../../textGeneration/CodexTextGeneratio import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; +import { instanceUsesSavedCredential } from "../../subscription-auth/runtime.ts"; import { ProviderDriverError } from "../Errors.ts"; import { makeCodexAdapter } from "../Layers/CodexAdapter.ts"; import { checkCodexProviderStatus, makePendingCodexProvider } from "../Layers/CodexProvider.ts"; @@ -42,7 +43,10 @@ import { makeManagedServerProvider } from "../makeManagedServerProvider.ts"; import * as ModelManifest from "../ModelManifest.ts"; import type { ProviderDriver, ProviderInstance } from "../ProviderDriver.ts"; import type { ServerProviderDraft } from "../providerSnapshot.ts"; -import { mergeProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; +import { + explicitProviderInstanceEnvironment, + mergeProviderInstanceEnvironment, +} from "../ProviderInstanceEnvironment.ts"; import { enrichProviderSnapshotWithVersionAdvisory, makePackageManagedProviderMaintenanceResolver, @@ -224,6 +228,15 @@ export const CodexDriver: ProviderDriver = { displayName, accentColor, enabled, + mastraConnection: { + environment: processEnv, + instanceEnvironment: explicitProviderInstanceEnvironment(environment), + useSavedCredential: instanceUsesSavedCredential("openai-codex", { + driver: DRIVER_KIND, + environment, + config, + }), + }, snapshot, adapter, textGeneration, diff --git a/apps/server/src/provider/Drivers/GrokDriver.ts b/apps/server/src/provider/Drivers/GrokDriver.ts index e5f05588fc02..2b0e17429f33 100644 --- a/apps/server/src/provider/Drivers/GrokDriver.ts +++ b/apps/server/src/provider/Drivers/GrokDriver.ts @@ -9,6 +9,7 @@ import { ChildProcessSpawner } from "effect/unstable/process"; import * as BackgroundPolicy from "../../background/BackgroundPolicy.ts"; import { ServerConfig } from "../../config.ts"; +import { instanceUsesSavedCredential } from "../../subscription-auth/runtime.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import { makeGrokTextGeneration } from "../../textGeneration/GrokTextGeneration.ts"; import { ProviderDriverError } from "../Errors.ts"; @@ -27,6 +28,7 @@ import { type ProviderInstance, } from "../ProviderDriver.ts"; import type { ServerProviderDraft } from "../providerSnapshot.ts"; +import { explicitProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; import { mergeSubscriptionInstanceEnvironment } from "../../subscription-auth/runtime.ts"; import { makeManualOnlyProviderMaintenanceCapabilities, @@ -176,6 +178,15 @@ export const GrokDriver: ProviderDriver = { displayName, accentColor, enabled, + mastraConnection: { + environment: processEnv, + instanceEnvironment: explicitProviderInstanceEnvironment(environment), + useSavedCredential: instanceUsesSavedCredential("xai", { + driver: DRIVER_KIND, + environment, + config, + }), + }, snapshot, snapshotForCwd, adapter, diff --git a/apps/server/src/provider/Drivers/KimiDriver.ts b/apps/server/src/provider/Drivers/KimiDriver.ts index 033b3e3e811b..e24a02300a78 100644 --- a/apps/server/src/provider/Drivers/KimiDriver.ts +++ b/apps/server/src/provider/Drivers/KimiDriver.ts @@ -12,7 +12,12 @@ import * as Stream from "effect/Stream"; import { ServerConfig } from "../../config.ts"; import { SubscriptionAuthService } from "../../subscription-auth/service.ts"; +import { + instanceUsesSavedCredential, + mergeSubscriptionInstanceEnvironment, +} from "../../subscription-auth/runtime.ts"; import type { ProviderDriver } from "../ProviderDriver.ts"; +import { explicitProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; import { defaultProviderContinuationIdentity } from "../ProviderDriver.ts"; import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts"; @@ -39,7 +44,7 @@ export const KimiDriver: ProviderDriver = { metadata: { displayName: "Kimi For Coding", supportsMultipleInstances: false }, configSchema: KimiSettings, defaultConfig: () => decodeSettings({}), - create: ({ instanceId, displayName, accentColor, enabled, config }) => + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => Effect.gen(function* () { const serverConfig = yield* ServerConfig; const auth = SubscriptionAuthService.forSecretsDir(serverConfig.secretsDir); @@ -48,6 +53,7 @@ export const KimiDriver: ProviderDriver = { PubSub.shutdown, ); const effectiveEnabled = enabled && config.enabled; + const processEnv = mergeSubscriptionInstanceEnvironment(environment); const continuationIdentity = defaultProviderContinuationIdentity({ driverKind: DRIVER_KIND, instanceId, @@ -86,6 +92,15 @@ export const KimiDriver: ProviderDriver = { displayName, accentColor, enabled: effectiveEnabled, + mastraConnection: { + environment: processEnv, + instanceEnvironment: explicitProviderInstanceEnvironment(environment), + useSavedCredential: instanceUsesSavedCredential("kimi-for-coding", { + driver: DRIVER_KIND, + environment, + config, + }), + }, adapter: undefined, textGeneration: undefined, snapshot: { diff --git a/apps/server/src/provider/Drivers/OpenCodeGoDriver.ts b/apps/server/src/provider/Drivers/OpenCodeGoDriver.ts index 558353d5af45..62cdbcf5c5a5 100644 --- a/apps/server/src/provider/Drivers/OpenCodeGoDriver.ts +++ b/apps/server/src/provider/Drivers/OpenCodeGoDriver.ts @@ -12,7 +12,12 @@ import * as Stream from "effect/Stream"; import { ServerConfig } from "../../config.ts"; import { SubscriptionAuthService } from "../../subscription-auth/service.ts"; +import { + instanceUsesSavedCredential, + mergeSubscriptionInstanceEnvironment, +} from "../../subscription-auth/runtime.ts"; import type { ProviderDriver } from "../ProviderDriver.ts"; +import { explicitProviderInstanceEnvironment } from "../ProviderInstanceEnvironment.ts"; import { defaultProviderContinuationIdentity } from "../ProviderDriver.ts"; import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts"; @@ -76,7 +81,7 @@ export const OpenCodeGoDriver: ProviderDriver decodeSettings({}), - create: ({ instanceId, displayName, accentColor, enabled, config }) => + create: ({ instanceId, displayName, accentColor, environment, enabled, config }) => Effect.gen(function* () { const serverConfig = yield* ServerConfig; const auth = SubscriptionAuthService.forSecretsDir(serverConfig.secretsDir); @@ -85,6 +90,7 @@ export const OpenCodeGoDriver: ProviderDriver { + it("uses the exact instance transport when deciding readiness", () => { + assert.isUndefined( + mastraConnectionIssue( + ProviderDriverKind.make("grok"), + { + environment: { XAI_API_KEY: "ambient-key" }, + instanceEnvironment: { XAI_API_KEY: "instance-key" }, + useSavedCredential: false, + }, + false, + ), + ); + assert.include( + mastraConnectionIssue( + ProviderDriverKind.make("grok"), + { + environment: { XAI_API_KEY: "ambient-key" }, + instanceEnvironment: {}, + useSavedCredential: false, + }, + true, + ) ?? "", + "XAI_API_KEY", + ); + }); + + it("requires the provider-wide connection only when the instance opted into it", () => { + const connection = { environment: {}, instanceEnvironment: {}, useSavedCredential: true }; + assert.isUndefined( + mastraConnectionIssue(ProviderDriverKind.make("claudeAgent"), connection, true), + ); + assert.include( + mastraConnectionIssue(ProviderDriverKind.make("claudeAgent"), connection, false) ?? "", + "Connect", + ); + }); + + it.each([ + [ProviderDriverKind.make("codex"), { OPENAI_API_KEY: "ambient-key" }], + [ProviderDriverKind.make("claudeAgent"), { ANTHROPIC_API_KEY: "ambient-key" }], + [ProviderDriverKind.make("grok"), { XAI_API_KEY: "ambient-key" }], + [ProviderDriverKind.make("opencodeGo"), { OPENCODE_API_KEY: "ambient-key" }], + ] as const)("accepts ambient credentials for %s without a saved connection", (provider, env) => { + assert.isUndefined( + mastraConnectionIssue( + provider, + { environment: env, instanceEnvironment: {}, useSavedCredential: true }, + false, + ), + ); + }); + + it("accepts an isolated OpenCode Go inline credential", () => { + const environment = { + OPENCODE_CONFIG_CONTENT: JSON.stringify({ + provider: { + "opencode-go": { + options: { apiKey: "inline-key", baseURL: "https://inline.example/v1" }, + }, + }, + }), + }; + assert.isUndefined( + mastraConnectionIssue( + ProviderDriverKind.make("opencodeGo"), + { environment, instanceEnvironment: environment, useSavedCredential: false }, + false, + ), + ); + }); +}); + function computerUseServer() { return { id: McpServerId.make("builtin-computer-use"), @@ -717,9 +793,7 @@ describe("AgentControllerLive", () => { }); expect(attachment).not.toHaveBeenCalled(); expect(manager.init).toHaveBeenCalledOnce(); - expect(bridge.startSession).toHaveBeenCalledTimes( - provider === "codex" || provider === "kimi" || provider === "opencodeGo" ? 0 : 1, - ); + expect(bridge.startSession).toHaveBeenCalledTimes(provider === "opencode" ? 1 : 0); yield* controller.stopSession({ threadId }); }), bridge.service, @@ -2026,11 +2100,14 @@ describe("AgentControllerLive", () => { requestId: ApprovalRequestId.make("send-tool-1"), decision: "decline", }); - yield* controller.respondToRequest({ - threadId: codexThreadId, - requestId: ApprovalRequestId.make("send-tool-1"), - decision: "accept", - }); + const duplicateResponseError = yield* controller + .respondToRequest({ + threadId: codexThreadId, + requestId: ApprovalRequestId.make("send-tool-1"), + decision: "accept", + }) + .pipe(Effect.flip); + expect(duplicateResponseError.message).toContain("no longer active"); expect(mastra.session.respondToToolApproval).toHaveBeenCalledTimes(2); expect(mastra.session.respondToToolApproval).toHaveBeenLastCalledWith({ toolCallId: "send-tool-1", @@ -2048,11 +2125,14 @@ describe("AgentControllerLive", () => { result: "cancelled", isError: true, } as AgentControllerEvent); - yield* controller.respondToRequest({ - threadId: codexThreadId, - requestId: ApprovalRequestId.make("shell-tool-stale"), - decision: "accept", - }); + const staleResponseError = yield* controller + .respondToRequest({ + threadId: codexThreadId, + requestId: ApprovalRequestId.make("shell-tool-stale"), + decision: "accept", + }) + .pipe(Effect.flip); + expect(staleResponseError.message).toContain("no longer active"); expect(mastra.session.respondToToolApproval).not.toHaveBeenCalledWith({ toolCallId: "shell-tool-stale", decision: "approve", @@ -3330,7 +3410,7 @@ describe("AgentControllerLive", () => { }).pipe(Effect.provide(layer), Effect.orDie); }); - it.effect("keeps Claude on the existing provider adapter", () => { + it.effect("runs Claude through the Akeru Mastra harness", () => { const bridge = makeBridge(); const mastra = makeMastraHarness(); return provideController( @@ -3355,11 +3435,53 @@ describe("AgentControllerLive", () => { input: "Use Claude.", }); - assert.equal(result.turnId, TurnId.make("legacy-turn")); - expect(bridge.startSession).toHaveBeenCalledOnce(); - expect(bridge.sendTurn).toHaveBeenCalledOnce(); - expect(mastra.createSession).not.toHaveBeenCalled(); - expect(mastra.sendMessage).not.toHaveBeenCalled(); + expect(String(result.turnId)).toMatch(/^mastra-turn-/); + expect(bridge.startSession).not.toHaveBeenCalled(); + expect(bridge.sendTurn).not.toHaveBeenCalled(); + expect(mastra.createSession).toHaveBeenCalledOnce(); + expect(mastra.session.model.switch).toHaveBeenCalledWith({ + modelId: "anthropic/claude-fable-5", + }); + expect(mastra.sendMessage).toHaveBeenCalledOnce(); + }), + bridge.service, + mastra.factory, + ); + }); + + it.effect("runs Grok through the Akeru Mastra harness", () => { + const bridge = makeBridge(); + const mastra = makeMastraHarness(); + return provideController( + Effect.gen(function* () { + const controller = yield* AgentController; + yield* controller.resolveEngine({ + threadId: grokThreadId, + engine: { provider: "grok", model: "grok-code-fast-1" }, + fallback: codexSelection, + mode: "default", + botConversation: true, + }); + yield* controller.startSession(grokThreadId, { + threadId: grokThreadId, + provider: ProviderDriverKind.make("grok"), + providerInstanceId: grokInstanceId, + cwd: process.cwd(), + runtimeMode: "approval-required", + }); + const result = yield* controller.sendTurn({ + threadId: grokThreadId, + input: "Use Grok.", + }); + + expect(String(result.turnId)).toMatch(/^mastra-turn-/); + expect(bridge.startSession).not.toHaveBeenCalled(); + expect(bridge.sendTurn).not.toHaveBeenCalled(); + expect(mastra.createSession).toHaveBeenCalledOnce(); + expect(mastra.session.model.switch).toHaveBeenCalledWith({ + modelId: "xai/grok-code-fast-1", + }); + expect(mastra.sendMessage).toHaveBeenCalledOnce(); }), bridge.service, mastra.factory, diff --git a/apps/server/src/provider/Layers/AgentController.ts b/apps/server/src/provider/Layers/AgentController.ts index c2cc2fc41dda..3544e3fe56c1 100644 --- a/apps/server/src/provider/Layers/AgentController.ts +++ b/apps/server/src/provider/Layers/AgentController.ts @@ -85,11 +85,13 @@ import { createAkeruMastraHarness, criticalAkeruAction, mastraModelId, + openCodeGoInlineConnection, type AkeruMastraHarness, type AkeruMastraHarnessOptions, type AkeruMastraSession, } from "../AkeruMastraHarness.ts"; import { AKERU_BOT_TURN_INSTRUCTIONS } from "../AkeruAgentInstructions.ts"; +import type { ProviderInstanceRoutingInfo } from "../Services/ProviderAdapterRegistry.ts"; import { createAkeruChannelRuntime, type AkeruChannelRuntime } from "../AkeruChannelRuntime.ts"; import { createAkeruBotStateRuntime, type AkeruBotStateRuntime } from "../AkeruBotStateRuntime.ts"; import { @@ -402,7 +404,13 @@ function approvalDetail(toolName: string, action: string | null, oneUse: boolean } function usesMastraCode(provider: ProviderDriverKind): boolean { - return provider === "codex" || provider === "kimi" || provider === "opencodeGo"; + return ( + provider === "codex" || + provider === "claudeAgent" || + provider === "grok" || + provider === "kimi" || + provider === "opencodeGo" + ); } function disabledProviderError( @@ -436,6 +444,65 @@ function subscriptionProviderForDriver( } } +export function mastraConnectionIssue( + provider: ProviderDriverKind, + connection: ProviderInstanceRoutingInfo["mastraConnection"], + savedCredentialConnected: boolean, +): string | undefined { + if (!connection) return undefined; + const env = connection.useSavedCredential + ? connection.environment + : connection.instanceEnvironment; + if (connection.useSavedCredential) { + const hasAmbientCredential = (() => { + switch (String(provider)) { + case "codex": + return Boolean(env.OPENAI_API_KEY?.trim()); + case "claudeAgent": + return Boolean( + env.ANTHROPIC_API_KEY?.trim() || + env.ANTHROPIC_AUTH_TOKEN?.trim() || + env.CLAUDE_CODE_OAUTH_TOKEN?.trim(), + ); + case "grok": + return Boolean(env.XAI_API_KEY?.trim()); + case "opencodeGo": + return Boolean(env.OPENCODE_API_KEY?.trim() || openCodeGoInlineConnection(env).apiKey); + default: + return false; + } + })(); + if (hasAmbientCredential) return undefined; + return savedCredentialConnected + ? undefined + : `Connect ${provider} in Settings before starting.`; + } + switch (String(provider)) { + case "codex": + return env.OPENAI_API_KEY?.trim() + ? undefined + : "This Codex instance needs OPENAI_API_KEY for the Akeru harness."; + case "claudeAgent": + return env.ANTHROPIC_API_KEY?.trim() || + env.ANTHROPIC_AUTH_TOKEN?.trim() || + env.CLAUDE_CODE_OAUTH_TOKEN?.trim() + ? undefined + : "This Claude instance needs an API key or auth token for the Akeru harness."; + case "grok": + return env.XAI_API_KEY?.trim() + ? undefined + : "This Grok instance needs XAI_API_KEY for the Akeru harness."; + case "kimi": + return "Custom Kimi credentials are not supported by the Akeru harness."; + case "opencodeGo": + return env.OPENCODE_API_KEY?.trim() || openCodeGoInlineConnection(env).apiKey + ? undefined + : "This OpenCode Go instance needs OPENCODE_API_KEY for the Akeru harness."; + default: + return `Provider '${provider}' has no Akeru Mastra transport.`; + } +} + export function recordProviderAccessHealth( subscriptionAuth: SubscriptionAuthService, event: ProviderRuntimeEvent, @@ -500,6 +567,10 @@ const make = (options?: AgentControllerLiveOptions) => const orchestrationEngine = yield* Effect.serviceOption(OrchestrationEngineService); const projectionSnapshotQuery = yield* Effect.serviceOption(ProjectionSnapshotQuery); const resolvedByThread = new Map(); + const modelConnections = new Map< + string, + NonNullable + >(); const sessions = new Map(); const memoryUsageByThread = new Map< string, @@ -721,6 +792,7 @@ const make = (options?: AgentControllerLiveOptions) => getOpenCodeGoApiKey: async () => subscriptionAuth.getApiKeyCredential("opencode-go")?.access, getSubscriptionApiKey: (provider) => subscriptionAuth.getApiKeyCredential(provider), + getModelConnection: (providerInstanceId) => modelConnections.get(providerInstanceId), memoryDbPath: NodePath.join(config.stateDir, "mastra-observational-memory.sqlite"), syncThreadToolApproval: async (threadId, toolName, protectedAction) => { const active = sessions.get(threadId); @@ -1591,6 +1663,20 @@ const make = (options?: AgentControllerLiveOptions) => modelSelection.instanceId, ); } + if (routing.mastraConnection) { + modelConnections.set(String(modelSelection.instanceId), routing.mastraConnection); + } else { + modelConnections.delete(String(modelSelection.instanceId)); + } + if (usesMastraCode(routing.driverKind)) { + const subscriptionProvider = subscriptionProviderForDriver(routing.driverKind); + const issue = mastraConnectionIssue( + routing.driverKind, + routing.mastraConnection, + subscriptionProvider ? subscriptionAuth.isConnected(subscriptionProvider) : false, + ); + if (issue) return yield* unavailable(new Error(issue)); + } const capabilities = usesMastraCode(routing.driverKind) ? { sessionModelSwitch: "in-session" as const } : yield* legacyProviderBridge @@ -1627,6 +1713,7 @@ const make = (options?: AgentControllerLiveOptions) => yield* runMastra("state.set", () => active.session.state.set({ ...activeState, + providerInstanceId: String(resolved.providerInstanceId), ...(nextModelOptions ? { modelOptions: nextModelOptions } : {}), }), ); @@ -1999,6 +2086,7 @@ const make = (options?: AgentControllerLiveOptions) => const modelOptions = mastraModelOptions(resolved); yield* runMastra("state.set", () => session.state.set({ + providerInstanceId: String(resolved.providerInstanceId), ...(input.cwd ? { projectPath: input.cwd } : {}), yolo: false, botConversation: resolved.botConversation, @@ -2258,7 +2346,12 @@ const make = (options?: AgentControllerLiveOptions) => } const toolRequest = active.approvalRequests.get(toolCallId); const pendingApproval = active.pendingApprovals.get(toolCallId); - if (!toolRequest || !pendingApproval) return; + if (!toolRequest || !pendingApproval) { + return yield* new AgentControllerRuntimeError({ + operation: "respondToRequest", + detail: `Stale pending approval request: ${input.requestId}. The request is no longer active.`, + }); + } const { name: toolName, input: toolInput } = toolRequest; const akeruTool = AKERU_TOOL_CATALOG.find((tool) => tool.id === toolName); const runtimeToolId = akeruTool?.id ?? (isMemoryToolId(toolName) ? toolName : undefined); diff --git a/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts b/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts index 6b32480d271b..07d4799f361d 100644 --- a/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts +++ b/apps/server/src/provider/Layers/ProviderAdapterRegistry.ts @@ -62,6 +62,7 @@ const makeProviderAdapterRegistry = Effect.fn("makeProviderAdapterRegistry")(fun accentColor: instance.accentColor, enabled: instance.enabled, continuationIdentity: instance.continuationIdentity, + ...(instance.mastraConnection ? { mastraConnection: instance.mastraConnection } : {}), }), ), ); diff --git a/apps/server/src/provider/ProviderDriver.ts b/apps/server/src/provider/ProviderDriver.ts index 866e418c3b85..a5ce1d684c2e 100644 --- a/apps/server/src/provider/ProviderDriver.ts +++ b/apps/server/src/provider/ProviderDriver.ts @@ -69,6 +69,15 @@ export interface ProviderInstance { readonly displayName: string | undefined; readonly accentColor?: string | undefined; readonly enabled: boolean; + /** + * Server-only connection inputs for Akeru's Mastra model transport. + * This value must never be projected into provider snapshots or wire contracts. + */ + readonly mastraConnection?: { + readonly environment: NodeJS.ProcessEnv; + readonly instanceEnvironment: NodeJS.ProcessEnv; + readonly useSavedCredential: boolean; + }; readonly snapshot: ServerProviderShape; /** * Optional workspace-scoped snapshot. Grok uses this to probe `grok inspect --json` diff --git a/apps/server/src/provider/ProviderInstanceEnvironment.ts b/apps/server/src/provider/ProviderInstanceEnvironment.ts index e469253604e6..3473577b1267 100644 --- a/apps/server/src/provider/ProviderInstanceEnvironment.ts +++ b/apps/server/src/provider/ProviderInstanceEnvironment.ts @@ -14,3 +14,9 @@ export function mergeProviderInstanceEnvironment( } return next; } + +export function explicitProviderInstanceEnvironment( + environment: ProviderInstanceEnvironment | undefined, +): NodeJS.ProcessEnv { + return Object.fromEntries(environment?.map((variable) => [variable.name, variable.value]) ?? []); +} diff --git a/apps/server/src/provider/Services/ProviderAdapterRegistry.ts b/apps/server/src/provider/Services/ProviderAdapterRegistry.ts index a642e94e9e8b..be1d38e6cfcd 100644 --- a/apps/server/src/provider/Services/ProviderAdapterRegistry.ts +++ b/apps/server/src/provider/Services/ProviderAdapterRegistry.ts @@ -38,6 +38,12 @@ export interface ProviderInstanceRoutingInfo { readonly accentColor?: string | undefined; readonly enabled: boolean; readonly continuationIdentity: ProviderContinuationIdentity; + /** Server-only. Never include this in a client projection or persisted event. */ + readonly mastraConnection?: { + readonly environment: NodeJS.ProcessEnv; + readonly instanceEnvironment: NodeJS.ProcessEnv; + readonly useSavedCredential: boolean; + }; } /** diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index b30f98a98e26..ac5cff2b2a53 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -8190,7 +8190,7 @@ it.live( "reports thread HTTP and WebSocket transfer budgets", () => Effect.gen(function* () { - const providers = [ProviderDriverKind.make("claudeAgent")] as const; + const providers = [ProviderDriverKind.make("opencode")] as const; const runs = yield* Effect.forEach( providers, diff --git a/apps/server/src/subscription-auth/runtime.ts b/apps/server/src/subscription-auth/runtime.ts index 2ab69324d350..b54adfa1e114 100644 --- a/apps/server/src/subscription-auth/runtime.ts +++ b/apps/server/src/subscription-auth/runtime.ts @@ -82,6 +82,7 @@ function hasExplicitEnvironmentKey(environment: SubscriptionEnvironment, key: st } const CONNECTION_ENV_KEYS: Partial>> = { + "openai-codex": ["OPENAI_API_KEY", "OPENAI_BASE_URL", "CODEX_HOME"], anthropic: [ "ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN", @@ -89,8 +90,9 @@ const CONNECTION_ENV_KEYS: Partial connectionKeys.includes(name))) return false; + if (provider === "opencode-go") { + const inlineConfig = instance.environment?.find( + ({ name }) => name === "OPENCODE_CONFIG_CONTENT", + )?.value; + if (inlineConfig) { + try { + const config = JSON.parse(inlineConfig) as { + readonly provider?: { + readonly "opencode-go"?: { readonly options?: Record }; + }; + }; + const options = config.provider?.["opencode-go"]?.options; + if (options && (Object.hasOwn(options, "apiKey") || Object.hasOwn(options, "baseURL"))) { + return false; + } + } catch { + return false; + } + } + } if (provider === "anthropic") { const config = instance.config; const homePath = @@ -109,6 +131,14 @@ export function instanceUsesSavedCredential( : undefined; if (typeof homePath === "string" && homePath.trim().length > 0) return false; } + if (provider === "openai-codex") { + const config = instance.config; + const homePath = + typeof config === "object" && config !== null && "homePath" in config + ? config.homePath + : undefined; + if (typeof homePath === "string" && homePath.trim().length > 0) return false; + } return true; } diff --git a/apps/web/src/components/chat/ThreadErrorBanner.test.tsx b/apps/web/src/components/chat/ThreadErrorBanner.test.tsx index 2137d44176fa..752aec0da43e 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.test.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.test.tsx @@ -95,4 +95,12 @@ describe("ThreadErrorBanner", () => { "A request failed in a bot chat.\n\nError details:\nProvider crashed", ); }); + + it("offers Resume for a recoverable failed request", () => { + const markup = renderToStaticMarkup( + {}} />, + ); + + expect(markup).toContain(">Resume<"); + }); }); diff --git a/apps/web/src/components/chat/ThreadErrorBanner.tsx b/apps/web/src/components/chat/ThreadErrorBanner.tsx index 3b4038ea8ff9..53b6091a360c 100644 --- a/apps/web/src/components/chat/ThreadErrorBanner.tsx +++ b/apps/web/src/components/chat/ThreadErrorBanner.tsx @@ -41,9 +41,13 @@ export function isThreadErrorBannerDismissedForSession(bannerKey: string | null) export const ThreadErrorBanner = memo(function ThreadErrorBanner({ error, onDismiss, + onResume, + resuming = false, }: { error: string | null; onDismiss?: () => void; + onResume?: () => void; + resuming?: boolean; }) { if (!error) return null; return ( @@ -59,6 +63,11 @@ export const ThreadErrorBanner = memo(function ThreadErrorBanner({ + {onResume ? ( + + ) : null}