diff --git a/packages/core/src/session/error.ts b/packages/core/src/session/error.ts index 13eaae975f0a..da70f3c6bdbc 100644 --- a/packages/core/src/session/error.ts +++ b/packages/core/src/session/error.ts @@ -74,15 +74,6 @@ export class StepFailedError extends Schema.TaggedError()("Sess } } -export class UserInterruptedError extends Schema.TaggedError()( - "Session.UserInterruptedError", - {}, -) { - override get message() { - return "Session interrupted by user" - } -} - export class PromptConflictError extends Schema.TaggedError()("Session.PromptConflictError", { sessionID: SessionSchema.ID, messageID: SessionMessage.ID, diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 9ff89f5113f1..3af258f02fa3 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -12,7 +12,6 @@ import { SessionRunner } from "./runner/index.js" import { SessionSchema } from "./schema.js" import { SessionStore } from "./store.js" import { toSessionError } from "./to-session-error.js" -import { UserInterruptedError } from "./error.js" import { SessionInbox } from "./inbox.js" export interface Interface { @@ -43,9 +42,7 @@ type InterruptReason = "user" | "shutdown" export function terminal(exit: Exit.Exit, reason?: InterruptReason) { if (Exit.isSuccess(exit)) return { type: "succeeded" as const } if (Cause.hasInterrupts(exit.cause)) return { type: "interrupted" as const, reason: reason ?? "shutdown" } - const failure = Cause.squash(exit.cause) - if (failure instanceof UserInterruptedError) return { type: "interrupted" as const, reason: "user" as const } - return { type: "failed" as const, error: toSessionError(failure) } + return { type: "failed" as const, error: toSessionError(Cause.squash(exit.cause)) } } /** Process-local execution: drains run in this process using the selected instance. */ diff --git a/packages/core/src/session/execution/restart.ts b/packages/core/src/session/execution/restart.ts index 8f263e88213a..2369b55657ea 100644 --- a/packages/core/src/session/execution/restart.ts +++ b/packages/core/src/session/execution/restart.ts @@ -11,6 +11,7 @@ import { SessionSchema } from "../schema.js" import { SessionStore } from "../store.js" import { ShellResult } from "../../shell/result.js" import { SubagentCompletion } from "../subagent-completion.js" +import { SubagentJob } from "../subagent-job.js" const CONTINUE_AFTER_SERVER_RESTART = "The server restarted while you were working. Continue from where you left off without repeating completed work." @@ -163,28 +164,9 @@ export const layer = (options?: Options) => return } - yield* jobs.start({ + yield* SubagentJob.start(sessions, jobs, recovery, { id: background.id, - type: "subagent", - title: recovery.description, notificationID: background.notificationID, - recovery, - run: execution.resume(recovery.childSessionID).pipe( - Effect.andThen(store.context(recovery.childSessionID)), - Effect.map((messages) => { - const assistant = messages.findLast( - (message) => - message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, - ) - if (assistant?.type !== "assistant") return "Subagent completed without a text response." - return ( - assistant.content - .filter((part) => part.type === "text") - .map((part) => part.text) - .join("") || "Subagent completed without a text response." - ) - }), - ), }) yield* jobs.background(background.id) yield* jobs.wait({ id: background.id }).pipe( diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 38cb3afdef4e..c9db41925287 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -4,7 +4,7 @@ import type { AIError } from "@opencode-ai/ai" import { Context, Data, Effect } from "effect" import { SessionSchema } from "../schema.js" import type { Promotable } from "../inbox.js" -import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js" +import type { AgentNotFoundError, MessageDecodeError, StepFailedError } from "../error.js" import { SessionRunnerModel } from "./model.js" import type { Instructions } from "../../instructions/index.js" @@ -14,7 +14,6 @@ export type RunError = | MessageDecodeError | AgentNotFoundError | StepFailedError - | UserInterruptedError | Instructions.InitializationBlocked export type Continuation = { readonly step: number } diff --git a/packages/core/src/session/subagent-completion.ts b/packages/core/src/session/subagent-completion.ts index 9ccf24363f40..f69349c9ccf2 100644 --- a/packages/core/src/session/subagent-completion.ts +++ b/packages/core/src/session/subagent-completion.ts @@ -3,6 +3,7 @@ export * as SubagentCompletion from "./subagent-completion.js" import { Effect } from "effect" import type { Job } from "../job.js" import type { Session } from "../session.js" +import { SubagentJob } from "./subagent-job.js" export const deliver = Effect.fnUntraced(function* ( sessions: Pick, @@ -16,7 +17,7 @@ export const deliver = Effect.fnUntraced(function* ( const recovery = input.recovery const text = input.status === "completed" - ? (input.output ?? "Subagent completed without a text response.") + ? (input.output ?? SubagentJob.noText) : input.status === "error" ? (input.error ?? "Subagent failed") : "Subagent cancelled" diff --git a/packages/core/src/session/subagent-job.ts b/packages/core/src/session/subagent-job.ts new file mode 100644 index 000000000000..e69d98bf67bd --- /dev/null +++ b/packages/core/src/session/subagent-job.ts @@ -0,0 +1,39 @@ +export * as SubagentJob from "./subagent-job.js" + +import { Effect } from "effect" +import type { Job } from "../job.js" +import type { Session } from "../session.js" + +export const noText = "Subagent completed without a text response." + +/** The same child execution and result selection for live jobs and restart recovery. */ +export const start = ( + sessions: Pick, + jobs: Pick, + recovery: Extract, + options?: Pick, +) => + jobs.start({ + id: options?.id ?? recovery.childSessionID, + type: "subagent", + title: recovery.description, + metadata: {}, + notificationID: options?.notificationID, + recovery, + run: Effect.gen(function* () { + // A failed resume remains a job error, not a successful no-text result. + yield* sessions.resume(recovery.childSessionID) + const messages = yield* sessions.messages({ sessionID: recovery.childSessionID, order: "desc", limit: 20 }) + const assistant = messages.find( + (message) => + message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, + ) + if (assistant?.type !== "assistant") return noText + return ( + assistant.content + .filter((part) => part.type === "text") + .map((part) => part.text) + .join("") || noText + ) + }), + }) diff --git a/packages/core/src/session/to-session-error.ts b/packages/core/src/session/to-session-error.ts index d731a959c9a2..aed9ee43b2d0 100644 --- a/packages/core/src/session/to-session-error.ts +++ b/packages/core/src/session/to-session-error.ts @@ -3,7 +3,7 @@ import { Tool } from "@opencode-ai/schema/tool" import { SessionError } from "@opencode-ai/schema/session-error" import { Permission } from "../permission.js" import { Integration } from "../integration.js" -import { AgentNotFoundError, StepFailedError, UserInterruptedError } from "./error.js" +import { AgentNotFoundError, StepFailedError } from "./error.js" import { SessionRunnerModel } from "./runner/model.js" export function toSessionError(cause: unknown): SessionError.Error { @@ -45,7 +45,6 @@ export function toSessionError(cause: unknown): SessionError.Error { } if (cause instanceof StepFailedError) return cause.error if (cause instanceof AgentNotFoundError) return { type: "unknown", message: cause.message } - if (cause instanceof UserInterruptedError) return { type: "aborted", message: cause.message } if ( cause instanceof SessionRunnerModel.ModelNotSelectedError || cause instanceof SessionRunnerModel.ModelUnavailableError || diff --git a/packages/core/src/tool/plugin/subagent.ts b/packages/core/src/tool/plugin/subagent.ts index 58d245931d67..6d1d9f1693cd 100644 --- a/packages/core/src/tool/plugin/subagent.ts +++ b/packages/core/src/tool/plugin/subagent.ts @@ -10,10 +10,10 @@ import { Permission } from "../../permission.js" import { Session } from "../../session.js" import { SessionSchema } from "../../session/schema.js" import { SubagentCompletion } from "../../session/subagent-completion.js" +import { SubagentJob } from "../../session/subagent-job.js" export const name = "subagent" -const NO_TEXT = "Subagent completed without a text response." const backgroundResult = (sessionID: SessionSchema.ID) => ({ sessionID, status: "running" as const, @@ -65,22 +65,6 @@ export const Plugin = { // continuation job is observable even while a settled generation's observer is finalizing. const notifications = new Set() - // Concatenate the child's final completed assistant text. Distinguishes "completed with no - // text" (generic string) from "failed" (the run effect fails, surfaced as a job error). - const latestAssistantText = Effect.fn("SubagentTool.latestAssistantText")(function* (sessionID: SessionSchema.ID) { - const messages = yield* sessions.messages({ sessionID, order: "desc", limit: 20 }) - const assistant = messages.find( - (message) => - message.type === "assistant" && message.time.completed !== undefined && message.error === undefined, - ) - if (assistant === undefined || assistant.type !== "assistant") return NO_TEXT - const text = assistant.content - .filter((part): part is Extract => part.type === "text") - .map((part) => part.text) - .join("") - return text.length > 0 ? text : NO_TEXT - }) - const notifyWhenDone = Effect.fn("SubagentTool.notifyWhenDone")(function* ( recovery: Extract, startedAt: number, @@ -225,14 +209,7 @@ export const Plugin = { agent: agent.name, description: input.description, } - const info = yield* jobs.start({ - id: child.id, - type: name, - title: input.description, - metadata: {}, - recovery, - run: sessions.resume(child.id).pipe(Effect.andThen(latestAssistantText(child.id))), - }) + const info = yield* SubagentJob.start(sessions, jobs, recovery) if (background) { yield* jobs.background(info.id) @@ -258,7 +235,11 @@ export const Plugin = { }) if (result?.info.status === "cancelled") return yield* new ToolFailure({ message: `Subagent cancelled (sessionID: ${child.id})` }) - return { sessionID: child.id, status: "completed" as const, output: result?.info.output ?? NO_TEXT } + return { + sessionID: child.id, + status: "completed" as const, + output: result?.info.output ?? SubagentJob.noText, + } }).pipe( Effect.map((output) => ({ output, diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index 3d34e97f2c08..a72c939e9d66 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -1,5 +1,8 @@ import { describe, expect, test } from "bun:test" import { AIError, TransportError } from "@opencode-ai/ai" +import { Agent } from "@opencode-ai/schema/agent" +import { Model } from "@opencode-ai/schema/model" +import { Provider } from "@opencode-ai/schema/provider" import { Database } from "@opencode-ai/core/database/database" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/util/effect/layer-node" @@ -15,12 +18,12 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { SessionExecution } from "@opencode-ai/core/session/execution" import { SessionRestart } from "@opencode-ai/core/session/execution/restart" -import { UserInterruptedError } from "@opencode-ai/core/session/error" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionInbox } from "@opencode-ai/core/session/inbox" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionRunner } from "@opencode-ai/core/session/runner/index" -import { SessionInboxTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionInboxTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SubagentJob } from "@opencode-ai/core/session/subagent-job" import { SessionStore } from "@opencode-ai/core/session/store" import { Cause, Context, Deferred, Effect, Exit, Fiber, Layer, LayerMap, Scope } from "effect" import { eq } from "drizzle-orm" @@ -50,10 +53,6 @@ describe("SessionExecution lifecycle", () => { const interrupted = Effect.runSyncExit(Effect.interrupt) expect(SessionExecution.terminal(interrupted)).toEqual({ type: "interrupted", reason: "shutdown" }) expect(SessionExecution.terminal(interrupted, "user")).toEqual({ type: "interrupted", reason: "user" }) - expect(SessionExecution.terminal(Exit.fail(new UserInterruptedError()))).toEqual({ - type: "interrupted", - reason: "user", - }) }) it.effect("the sweep only lists claimed top-level Sessions", () => @@ -378,6 +377,99 @@ describe("SessionExecution lifecycle", () => { ) }) +describe("Subagent job results", () => { + for (const recovered of [false, true]) { + for (const result of ["text", "empty", "stale"] as const) { + it.effect(`${recovered ? "recovered" : "live"} jobs use the same ${result} result selection`, () => + Effect.gen(function* () { + const database = yield* Database.Service + const jobs = yield* Job.Service + const store = yield* SessionStore.Service + const parent = Session.ID.make("ses_result_parent") + const child = Session.ID.make("ses_result_child") + yield* seedSessions(database, [parent]) + yield* seedSessions(database, [child], { parent_id: parent }) + yield* store.claim(child) + const data = { + agent: Agent.ID.make("explore"), + model: Model.Ref.make({ id: Model.ID.make("test"), providerID: Provider.ID.make("test") }), + time: { created: 1, completed: 2 }, + } + yield* database.db + .insert(SessionMessageTable) + .values([ + ...[ + { ...data, content: [{ type: "text" as const, text: "Older answer" }] }, + { + ...data, + content: [ + { type: "reasoning" as const, text: "Not part of the result" }, + ...(result === "empty" + ? [] + : [ + { type: "text" as const, text: "Final " }, + { type: "text" as const, text: "answer" }, + ]), + ], + }, + { + ...data, + content: [{ type: "text" as const, text: "Failed answer" }], + error: { type: "unknown", message: "Failed" }, + }, + { ...data, time: { created: 1 }, content: [{ type: "text" as const, text: "Unfinished answer" }] }, + ].map((data, seq) => ({ + id: SessionMessage.ID.create(), + session_id: child, + type: "assistant" as const, + seq: seq + 1, + data, + })), + ...Array.from({ length: result === "stale" ? 20 : 0 }, (_, seq) => ({ + id: SessionMessage.ID.create(), + session_id: child, + type: "synthetic" as const, + seq: seq + 5, + data: { text: "Later activity", time: { created: 3 } }, + })), + ]) + .run() + .pipe(Effect.orDie) + + const recovery = { + kind: "subagent" as const, + parentSessionID: parent, + childSessionID: child, + agent: "explore", + description: "Select the child result", + } + const scope = yield* Scope.make() + yield* Effect.addFinalizer(() => Scope.close(scope, Exit.void)) + const restarted = yield* Job.make.pipe(Effect.provideService(Scope.Scope, scope)) + const context = yield* buildExecution(scope, () => Effect.void, undefined, restarted) + const sessions = Context.get(context, Session.Service) + const id = recovered ? "recovered-child-job" : child + if (recovered) { + yield* jobs.start({ id, type: "subagent", recovery, run: Effect.never }) + const marker = yield* jobs.background(id) + yield* Context.get(context, SessionRestart.Service).resumeSuspendedSessions + expect((yield* restarted.get(id))?.notificationID).toBe(marker?.notificationID) + } + if (!recovered) yield* SubagentJob.start(sessions, restarted, recovery) + + expect((yield* restarted.wait({ id })).info).toMatchObject({ + id, + type: "subagent", + title: recovery.description, + status: "completed", + output: result === "text" ? "Final answer" : "Subagent completed without a text response.", + }) + }), + ) + } + } +}) + describe("SessionRestart background recovery", () => { it.effect("wakes idle shell owners and delivers recovered notices exactly once", () => Effect.gen(function* () { @@ -1342,6 +1434,7 @@ function buildExecution( const execution = yield* SessionExecution.Service return Session.Service.of({ ...sessions, + resume: execution.resume, synthetic: (input) => sessions .synthetic({ ...input, resume: false })