mirror of
https://github.com/anomalyco/opencode.git
synced 2026-07-24 11:46:17 +00:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d5ed132cd8 |
@@ -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,
|
||||
],
|
||||
})
|
||||
|
||||
|
||||
@@ -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",
|
||||
() =>
|
||||
|
||||
@@ -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)
|
||||
}),
|
||||
)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user