diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 6734a1f5ac..40899d0a21 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -49,6 +49,7 @@ import { SessionRunState } from "./run-state" import { RuntimeFlags } from "@/effect/runtime-flags" import { EventV2Bridge } from "@/event-v2-bridge" import { Database } from "@opencode-ai/core/database/database" +import { BackgroundJob } from "@/background/job" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" import { eq } from "drizzle-orm" @@ -99,6 +100,48 @@ function isOrphanedInterruptedTool(part: SessionV1.ToolPart) { return part.state.status === "error" && part.state.metadata?.interrupted === true } +function renderSubagentOutput(input: { + sessionID: SessionID + state: "completed" | "error" + summary: string + text: string +}) { + const tag = input.state === "error" ? "task_error" : "task_result" + return [ + ``, + `${input.summary}`, + `<${tag}>`, + input.text, + ``, + "", + ].join("\n") +} + +function subagentResultText(result: SessionV1.WithParts) { + const text = result.parts + .filter((part): part is SessionV1.TextPart => part.type === "text") + .map((part) => part.text) + .join("\n") + .trim() + if (text) return text + if (result.info.role === "assistant" && result.info.error) return assistantErrorMessage(result.info.error) + return "Subagent finished without a text response." +} + +function assistantErrorMessage(error: NonNullable) { + if (typeof error.data === "object" && error.data && "message" in error.data && typeof error.data.message === "string") { + return error.data.message + } + return error.name +} + +function subagentNotifiedMetadata(session: Session.Info, messageID: string) { + return { + ...session.metadata, + subagent_last_notified_message_id: messageID, + } +} + export interface Interface { readonly cancel: (sessionID: SessionID) => Effect.Effect readonly prompt: (input: PromptInput) => Effect.Effect @@ -140,6 +183,7 @@ const layer = Layer.effect( const events = yield* EventV2Bridge.Service const flags = yield* RuntimeFlags.Service const database = yield* Database.Service + const background = yield* BackgroundJob.Service const { db } = database const ops = Effect.fn("SessionPrompt.ops")(function* () { return { @@ -1070,6 +1114,64 @@ const layer = Layer.effect( return yield* loop({ sessionID: input.sessionID }) }) + const notifyParent = Effect.fn("SessionPrompt.notifyParent")(function* (input: { + session: Session.Info + result: SessionV1.WithParts + }) { + if (!input.session.parentID) return + const current = yield* sessions.get(input.session.id).pipe(Effect.catchCause(() => Effect.succeed(input.session))) + if (current.metadata?.subagent_last_notified_message_id === input.result.info.id) return + const job = yield* background.get(input.session.id) + if ( + job?.status === "running" && + job.metadata?.parentSessionId === current.parentID && + job.metadata?.background !== true + ) { + yield* sessions.setMetadata({ + sessionID: current.id, + metadata: subagentNotifiedMetadata(current, input.result.info.id), + }) + return + } + + if (!current.parentID) return + const parent = yield* sessions.get(current.parentID).pipe( + Effect.catchCause((cause) => + Effect.logWarning("failed to load parent session for subagent notification", { + sessionID: current.id, + parentID: current.parentID, + cause: Cause.pretty(cause), + }).pipe(Effect.as(undefined)), + ), + ) + if (!parent) return + yield* sessions.setMetadata({ + sessionID: current.id, + metadata: subagentNotifiedMetadata(current, input.result.info.id), + }) + + const state = input.result.info.role === "assistant" && input.result.info.error ? "error" : "completed" + yield* prompt({ + sessionID: current.parentID, + agent: parent.agent, + parts: [ + { + type: "text", + synthetic: true, + text: renderSubagentOutput({ + sessionID: current.id, + state, + summary: + state === "completed" + ? `Subagent completed: ${current.title}` + : `Subagent failed: ${current.title}`, + text: subagentResultText(input.result), + }), + }, + ], + }).pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true })) + }) + const lastAssistant = Effect.fnUntraced(function* (sessionID: SessionID) { const match = yield* sessions.findMessage(sessionID, (m) => m.info.role !== "user").pipe(Effect.orDie) if (Option.isSome(match)) return match.value @@ -1335,7 +1437,9 @@ const layer = Layer.effect( } yield* compaction.prune({ sessionID }).pipe(Effect.ignore, Effect.forkIn(scope)) - return yield* lastAssistant(sessionID) + const result = yield* lastAssistant(sessionID) + yield* notifyParent({ session, result }).pipe(Effect.ignore) + return result }, ) @@ -1624,6 +1728,7 @@ export const node = LayerNode.make({ EventV2Bridge.node, RuntimeFlags.node, Database.node, + BackgroundJob.node, ], }) diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index b0a866c90e..727b7b9191 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -10,7 +10,7 @@ import { Agent } from "../agent/agent" import { deriveSubagentSessionPermission } from "../agent/subagent-permissions" import type { SessionPrompt } from "../session/prompt" import { Config } from "@/config/config" -import { Effect, Exit, Schema, Scope } from "effect" +import { Effect, Exit, Schema } from "effect" import { EffectBridge } from "@/effect/bridge" import { RuntimeFlags } from "@/effect/runtime-flags" import { Database } from "@opencode-ai/core/database/database" @@ -85,7 +85,6 @@ export const TaskTool = Tool.define( const background = yield* BackgroundJob.Service const config = yield* Config.Service const sessions = yield* Session.Service - const scope = yield* Scope.Scope const flags = yield* RuntimeFlags.Service const database = yield* Database.Service @@ -199,46 +198,6 @@ export const TaskTool = Tool.define( return result.parts.findLast((item) => item.type === "text")?.text ?? "" }) - const inject = Effect.fn("TaskTool.injectBackgroundResult")(function* ( - state: "completed" | "error", - text: string, - ) { - const currentParent = yield* sessions.get(ctx.sessionID) - yield* ops - .prompt({ - sessionID: ctx.sessionID, - agent: currentParent.agent ?? ctx.agent, - variant, - parts: [ - { - type: "text", - synthetic: true, - text: renderOutput({ - sessionID: nextSession.id, - state, - summary: - state === "completed" - ? `Background task completed: ${params.description}` - : `Background task failed: ${params.description}`, - text, - }), - }, - ], - }) - .pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true })) - }) - - const notify = Effect.fn("TaskTool.notifyBackgroundResult")(function* (jobID: string) { - yield* background.wait({ id: jobID }).pipe( - Effect.flatMap((result) => { - if (result.info?.status === "completed") return inject("completed", result.info.output ?? "") - if (result.info?.status === "error") return inject("error", result.info.error ?? "") - return Effect.void - }), - Effect.forkIn(scope, { startImmediately: true }), - ) - }) - if (yield* background.extend({ id: nextSession.id, run: runTask() })) { return { title: params.description, @@ -261,13 +220,10 @@ export const TaskTool = Tool.define( type: id, title: params.description, metadata, - onPromote: Effect.all([ - ctx.metadata({ - title: params.description, - metadata: { ...metadata, background: true, jobId: nextSession.id }, - }), - notify(nextSession.id), - ]), + onPromote: ctx.metadata({ + title: params.description, + metadata: { ...metadata, background: true, jobId: nextSession.id }, + }), run: runTask().pipe(Effect.onInterrupt(() => ops.cancel(nextSession.id))), }) @@ -289,7 +245,6 @@ export const TaskTool = Tool.define( } if (runInBackground) { - yield* notify(info.id) return backgroundResult() } diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 491ad06aaf..7541e3ec20 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -514,6 +514,86 @@ it.instance("loop calls LLM and returns assistant message", () => }), ) +it.instance("child session completion notifies the parent with a synthetic result", () => + Effect.gen(function* () { + const { llm } = yield* useServerConfig(providerCfg) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const parent = yield* sessions.create({ title: "Parent", agent: "build" }) + const child = yield* sessions.create({ parentID: parent.id, title: "Inspect bug", agent: "general" }) + yield* llm.text("child done") + yield* prompt.prompt({ + sessionID: child.id, + agent: "general", + model: ref, + noReply: true, + parts: [{ type: "text", text: "look into the cache key path" }], + }) + + const result = yield* prompt.loop({ sessionID: child.id }) + const notification = yield* pollWithTimeout( + Effect.gen(function* () { + const msgs = yield* MessageV2.filterCompactedEffect(parent.id) + return msgs + .filter((msg) => msg.info.role === "user") + .flatMap((msg) => msg.parts) + .find( + (part): part is SessionV1.TextPart => + part.type === "text" && + part.synthetic === true && + part.text.includes(``) && + part.text.includes("child done"), + ) + }), + "parent never received child completion notification", + "5 seconds", + ) + + expect(notification.text).toContain("Subagent completed: Inspect bug") + expect((yield* sessions.get(child.id)).metadata?.subagent_last_notified_message_id).toBe(result.info.id) + }), +) + +it.instance("foreground task subagents return through the tool result without a parent synthetic prompt", () => + Effect.gen(function* () { + const { llm } = yield* useServerConfig(providerCfg) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const chat = yield* sessions.create({ title: "Parent", agent: "build" }) + yield* llm.tool("task", { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + }) + yield* llm.text("child done") + yield* llm.text("parent final") + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + model: ref, + noReply: true, + parts: [{ type: "text", text: "delegate" }], + }) + + const result = yield* prompt.loop({ sessionID: chat.id }) + expect(result.parts.some((part) => part.type === "text" && part.text === "parent final")).toBe(true) + + const child = (yield* sessions.children(chat.id))[0] + expect(child).toBeDefined() + if (!child) return + expect((yield* sessions.get(child.id)).metadata?.subagent_last_notified_message_id).toBeDefined() + + const parentSynthetic = (yield* MessageV2.filterCompactedEffect(chat.id)) + .filter((msg) => msg.info.role === "user") + .flatMap((msg) => msg.parts) + .filter( + (part): part is SessionV1.TextPart => + part.type === "text" && part.synthetic === true && part.text.includes(` diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index 6238a6a077..eedbeea1da 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -495,14 +495,15 @@ describe("tool.task", () => { const def = yield* tool.init() const ready = yield* Deferred.make() const done = yield* Deferred.make() - const injected = yield* Deferred.make() let runs = 0 + let parentPrompts = 0 const promptOps: TaskPromptOps = { cancel: () => Effect.void, resolvePromptParts: (template) => Effect.succeed([{ type: "text" as const, text: template }]), prompt: (input) => { if (input.sessionID === chat.id) { - return Deferred.succeed(injected, input).pipe(Effect.as(reply(input, "injected"))) + parentPrompts += 1 + return Effect.succeed(reply(input, "unexpected parent prompt")) } return Effect.gen(function* () { runs += 1 @@ -548,7 +549,7 @@ describe("tool.task", () => { yield* Deferred.succeed(done, undefined) expect((yield* jobs.wait({ id: result.metadata.sessionId })).info?.output).toBe("background done") - expect((yield* Deferred.await(injected)).parts[0]?.type).toBe("text") + expect(parentPrompts).toBe(0) expect(runs).toBe(1) }), ) @@ -600,13 +601,13 @@ describe("tool.task", () => { const first = defer() const second = defer() const updated = defer() - const injected = defer() let prompts = 0 + let parentPrompts = 0 const promptOps: TaskPromptOps = { ...stubOps(), prompt: (input) => { if (input.sessionID === chat.id) { - injected.resolve(input) + parentPrompts += 1 return Effect.succeed(reply(input, "done")) } prompts++ @@ -658,10 +659,7 @@ describe("tool.task", () => { const waited = yield* jobs.wait({ id: started.metadata.sessionId, timeout: 1_000 }) expect(waited.info?.status).toBe("completed") expect(waited.info?.output).toBe("second done") - const notification = yield* Effect.promise(() => injected.promise) - expect(notification.variant).toBe("xhigh") - expect(notification.parts[0]?.type).toBe("text") - if (notification.parts[0]?.type === "text") expect(notification.parts[0].text).toContain("second done") + expect(parentPrompts).toBe(0) }), ) @@ -698,12 +696,13 @@ describe("tool.task", () => { }), ) - background.instance("background task completion does not wait for the parent async prompt", () => + background.instance("background task completion does not prompt the parent directly", () => Effect.gen(function* () { const jobs = yield* BackgroundJob.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() + let parentPrompts = 0 const result = yield* def.execute( { @@ -721,7 +720,12 @@ describe("tool.task", () => { promptOps: { ...stubOps({ text: "background done" }), prompt: (input) => - input.sessionID === chat.id ? Effect.never : Effect.succeed(reply(input, "background done")), + input.sessionID === chat.id + ? Effect.sync(() => { + parentPrompts += 1 + return reply(input, "unexpected parent prompt") + }) + : Effect.succeed(reply(input, "background done")), } satisfies TaskPromptOps, }, messages: [], @@ -733,6 +737,7 @@ describe("tool.task", () => { const waited = yield* jobs.wait({ id: result.metadata.sessionId, timeout: 1_000 }) expect(waited.timedOut).toBe(false) expect(waited.info?.status).toBe("completed") + expect(parentPrompts).toBe(0) }), )