Compare commits

...
Author SHA1 Message Date
Dax d5ed132cd8 fix(session): notify parent when subagents finish 2026-07-02 23:53:46 +00:00
4 changed files with 207 additions and 62 deletions
+106 -1
View File
@@ -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 [
`<task id="${input.sessionID}" state="${input.state}">`,
`<summary>${input.summary}</summary>`,
`<${tag}>`,
input.text,
`</${tag}>`,
"</task>",
].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<SessionV1.Assistant["error"]>) {
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<void>
readonly prompt: (input: PromptInput) => Effect.Effect<SessionV1.WithParts, Image.Error>
@@ -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,
],
})
+5 -50
View File
@@ -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()
}
@@ -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(`<task id="${child.id}" state="completed">`) &&
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(`<task id="${child.id}"`),
)
expect(parentSynthetic).toHaveLength(0)
}),
)
withMcpInstructions.instance(
"loop includes MCP instructions in model system context",
() =>
+16 -11
View File
@@ -495,14 +495,15 @@ describe("tool.task", () => {
const def = yield* tool.init()
const ready = yield* Deferred.make<void>()
const done = yield* Deferred.make<void>()
const injected = yield* Deferred.make<SessionPrompt.PromptInput>()
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<void>()
const second = defer<void>()
const updated = defer<SessionPrompt.PromptInput>()
const injected = defer<SessionPrompt.PromptInput>()
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)
}),
)