diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index 426755ea44..7cf2d7f455 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -11,7 +11,14 @@ import { GrepTool } from "./grep" import { BatchTool } from "./batch" // altimate_change end import { ReadTool } from "./read" -import { TaskTool } from "./task" +// altimate_change start — upstream_fix: the task tool owns the background schema, description and env fallback. +import { + TaskTool, + backgroundSubagentsEnabled as backgroundSubagentsFromEnv, + backgroundTaskDescription, + taskBackgroundJsonSchema, +} from "./task" +// altimate_change end import { TodoWriteTool } from "./todo" import { WebFetchTool } from "./webfetch" import { WriteTool } from "./write" @@ -343,9 +350,7 @@ export namespace ToolRegistry { function backgroundSubagentsEnabled(flags?: ToolRuntimeFlags) { if (flags) return flags.experimentalBackgroundSubagents - const value = process.env["OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS"] - if (value !== undefined) return ["1", "true", "yes"].includes(value.toLowerCase()) - return Flag.OPENCODE_EXPERIMENTAL + return backgroundSubagentsFromEnv() } // altimate_change start — upstream_fix: allow Parallel-enabled websearch for non-opencode providers. @@ -361,18 +366,34 @@ export namespace ToolRegistry { // altimate_change end function applyRuntimeToolSchemaFlags(id: string, tool: T, flags?: ToolRuntimeFlags): T { - if (id !== "task" || backgroundSubagentsEnabled(flags)) return tool - const schema = tool.jsonSchema + if (id !== "task") return tool + // Execution reads the same decision that shapes the schema, so the two gates cannot disagree. + const enabled = backgroundSubagentsEnabled(flags) + const execute = tool.execute + const gated: T = { + ...tool, + execute: (args, ctx) => execute(args, { ...ctx, extra: { ...ctx.extra, backgroundSubagents: enabled } }), + ...(enabled ? { description: backgroundTaskDescription(tool.description) } : {}), + } + const schema = gated.jsonSchema if (!schema || typeof schema !== "object" || !schema.properties || typeof schema.properties !== "object") { - return tool + return gated } const properties = { ...(schema.properties as Record) } - if (!("background" in properties)) return tool + // The task tool's default contract omits `background`; the experiment flag publishes it. + if (enabled) { + if ("background" in properties || !taskBackgroundJsonSchema) return gated + return { + ...gated, + jsonSchema: { ...schema, properties: { ...properties, background: taskBackgroundJsonSchema } }, + } + } + if (!("background" in properties)) return gated delete properties.background return { - ...tool, + ...gated, jsonSchema: { ...schema, properties, @@ -608,8 +629,9 @@ export namespace ToolRegistry { .map(async (t) => { using _ = log.time(t.id) // altimate_change start — v1.17.9: Tool.Info.init() returns an Effect of DefWithoutID - // altimate_change start — upstream_fix: pass caller agent through deferred tool init. - const tool = await Effect.runPromise(t.init({ agent })) + // altimate_change start — upstream_fix: pass caller agent through deferred tool init; runtime + // flags shape the definition before plugins see it. + const tool = applyRuntimeToolSchemaFlags(t.id, await Effect.runPromise(t.init({ agent })), runtimeFlags) // altimate_change end // altimate_change end const output = { @@ -623,7 +645,7 @@ export namespace ToolRegistry { registrySource: t.registrySource, // altimate_change end // altimate_change start — upstream_fix: hide disabled runtime-gated tool schema fields. - ...applyRuntimeToolSchemaFlags(t.id, tool, runtimeFlags), + ...tool, // altimate_change end description: output.description, parameters: output.parameters, diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 5a62dbd733..8afd30159d 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -24,10 +24,13 @@ import { RunBudget } from "@/altimate/run-budget" // altimate_change end // re-brand core (ModelV2/ProviderV2) IDs to the provider/schema brands SessionPrompt expects import { ModelID, ProviderID } from "@/provider/schema" -import { Effect } from "effect" +import { ConfigProvider, Effect, Layer } from "effect" // altimate_change start — upstream_fix: recover committed child findings when cancellation interrupts a task. import { interruptedTaskOutput } from "../session/task-output" // altimate_change end +// altimate_change start — upstream_fix: background subagents follow the experiment flag at execution time. +import { RuntimeFlags } from "@/effect/runtime-flags" +// altimate_change end const log = Log.create({ service: "tool.task" }) /** @@ -54,7 +57,12 @@ const parameters = z.object({ .optional(), command: z.string().describe("The command that triggered this task").optional(), // altimate_change start — upstream_fix: keep accepting legacy/manual background calls internally. - background: z.boolean().optional(), + background: z + .boolean() + .describe( + "Run the subagent in the background and return immediately with a task_id. Call task again with that task_id (same subagent_type) to wait for and collect its result. A background task you have not collected is cancelled when your turn ends.", + ) + .optional(), // altimate_change end }) @@ -74,6 +82,159 @@ const taskJsonSchema = (() => { })() // altimate_change end +// altimate_change start — upstream_fix: background subagents run without blocking the parent turn. +/** The `background` property the registry adds to the task schema when the experiment is on. */ +export const taskBackgroundJsonSchema = (zodToJsonSchema(parameters).properties as Record).background + +/** + * Whether background subagents are on, read from the environment exactly as RuntimeFlags reads it. + * The registry passes its own RuntimeFlags decision to execution (`extra.backgroundSubagents`), so + * the schema gate and the execution gate always agree; this is the fallback when nothing is passed. + */ +export function backgroundSubagentsEnabled() { + return Effect.runSync( + Effect.gen(function* () { + return (yield* RuntimeFlags.Service).experimentalBackgroundSubagents + }).pipe( + // A fresh env provider: the default one is a snapshot taken on first use. + Effect.provide(RuntimeFlags.defaultLayer.pipe(Layer.provide(ConfigProvider.layer(ConfigProvider.fromEnv())))), + ), + ) +} + +const NOTIFY_SENTENCE = "For background tasks, you will be notified automatically when the result is ready." +const COLLECT_SENTENCE = + "A background task (background: true) sends no notification: collect its result by calling task again with its task_id. Collect every background task before you end your turn; ending your turn cancels any background task you have not collected. Calling task with a background task's task_id always returns that task's recorded result and never sends it new work." + +/** The task description with the background experiment on: no notification exists, the parent collects. */ +export function backgroundTaskDescription(description: string) { + return description.replace(NOTIFY_SENTENCE, COLLECT_SENTENCE) +} + +/** Running background children one parent session may have at once. */ +export const MAX_RUNNING_BACKGROUND_TASKS = 4 +/** Recorded background tasks kept per parent session; the oldest settled records go first. */ +const MAX_BACKGROUND_RECORDS = 100 + +type TaskResult = { + title: string + metadata: { + sessionId: SessionID + model: { modelID: string; providerID: string } + interrupted?: boolean + background?: boolean + } + output: string +} + +type BackgroundTask = { + parentSessionID: string + messageID: MessageID + model: { modelID: string; providerID: string } + result: Promise + settled: boolean + collected: boolean + joins: number +} + +/** + * Background children keyed by child session id, with their outcome once settled. A child is bound + * to the abort signal of the parent tool call that started it: `ctx.abort` is that request's + * `AbortSignal.any` over the loop generation's signal. The generation signal is aborted on cancel, + * on a run-budget hard deadline and in the loop's cleanup when the parent's turn ends, so a child + * still running then is cancelled. A soft run-budget interrupt aborts only the request it targets, + * so it reaches a child launched in that request and not one launched in an earlier step. + * + * The record outlives the child: every later call with its task_id returns the recorded result + * (or, for a cancelled child, its partial findings marked interrupted) and never re-prompts it. + */ +const backgroundTasks = new Map() + +function runningBackgroundTasks(parentSessionID: string) { + let count = 0 + for (const task of backgroundTasks.values()) if (task.parentSessionID === parentSessionID && !task.settled) count++ + return count +} + +function evictBackgroundRecords(parentSessionID: string) { + const settled = [...backgroundTasks].filter(([, task]) => task.parentSessionID === parentSessionID && task.settled) + const total = [...backgroundTasks.values()].filter((task) => task.parentSessionID === parentSessionID).length + for (const [id] of settled.slice(0, Math.max(0, total - MAX_BACKGROUND_RECORDS))) backgroundTasks.delete(id) +} + +const resumeHeader = (id: string) => `task_id: ${id} (for resuming to continue this task if needed)` +const recordedHeader = (id: string) => + `task_id: ${id} (background task: calling task with this task_id returns this same result)` + +/** Points a background task's result at its record, since a task_id call collects rather than resumes. */ +function recordedResult(id: string, result: TaskResult): TaskResult { + return { ...result, output: result.output.replace(resumeHeader(id), recordedHeader(id)) } +} + +/** The tool part of one call in a message, or undefined when the message or part is not stored. */ +function toolPart(sessionID: SessionID, messageID: MessageID, callID: string) { + try { + return MessageV2.get({ sessionID, messageID }).parts.find( + (item): item is MessageV2.ToolPart => item.type === "tool" && item.callID === callID, + ) + } catch { + return undefined + } +} + +const PART_RUNNING_WAIT_MS = 2_000 +const PART_RUNNING_POLL_MS = 10 + +/** + * Waits until the loop has marked this call's tool part running. The loop records `ctx.metadata` only + * on a running part, and a collect reaches its metadata call before the loop has finished that write. + */ +async function untilToolPartRunning(ctx: { + sessionID: SessionID + messageID: MessageID + callID?: string + abort: AbortSignal +}) { + if (!ctx.callID) return + const until = Date.now() + PART_RUNNING_WAIT_MS + while (Date.now() < until && !ctx.abort.aborted) { + const part = toolPart(ctx.sessionID, ctx.messageID, ctx.callID) + if (!part || part.state.status !== "pending") return + await new Promise((resolve) => setTimeout(resolve, PART_RUNNING_POLL_MS)) + } +} + +/** Appends the cancellation to the launch tool part, so the turn's record shows what was lost. */ +async function recordCancellation(input: { + sessionID: SessionID + messageID: MessageID + callID?: string + childID: string + output: string +}) { + if (!input.callID) return + try { + const part = toolPart(input.sessionID, input.messageID, input.callID) + if (!part || part.state.status !== "completed") return + await Session.updatePart({ + ...part, + state: { + ...part.state, + output: [ + part.state.output, + "", + `Background task ${input.childID} was cancelled at the end of the turn, before its result was collected. Its partial findings:`, + input.output, + ].join("\n"), + metadata: { ...part.state.metadata, interrupted: true }, + }, + }) + } catch (error) { + log.warn("failed to record background task cancellation", { sessionID: input.sessionID, error }) + } +} +// altimate_change end + export const TaskTool = Tool.define("task", async (ctx) => { const agents = await Agent.list().then((x) => x.filter((a) => a.mode !== "primary")) @@ -97,9 +258,50 @@ export const TaskTool = Tool.define("task", async (ctx) => { // altimate_change end async execute(params: z.infer, ctx) { const config = await Config.get() - if (params.background === true) { + // altimate_change start — upstream_fix: honour background only behind the experiment flag; a + // task_id naming a running background child of this session joins it instead of re-prompting. + const background = params.background === true + const enabled = + typeof ctx.extra?.backgroundSubagents === "boolean" + ? ctx.extra.backgroundSubagents + : background && backgroundSubagentsEnabled() + if (background && !enabled) { throw new Error("Background subagents require OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS=true") } + const recorded = params.task_id ? backgroundTasks.get(params.task_id) : undefined + if (params.task_id && recorded?.parentSessionID === ctx.sessionID) { + const taskID = params.task_id + if (background && !recorded.settled) { + return { + title: params.description, + metadata: { sessionId: SessionID.make(taskID), model: recorded.model, background: true }, + output: [ + `task_id: ${taskID}`, + `Task is still running in the background. Its result is collected by calling task again with task_id=${taskID} (without background); it is cancelled if your turn ends before you collect it.`, + ].join("\n"), + } + } + // Publish the child's identity first, so a cancel during the wait recovers its partial findings. + await untilToolPartRunning(ctx) + await ctx.metadata({ + title: params.description, + metadata: { sessionId: taskID, messageID: recorded.messageID, model: recorded.model }, + }) + recorded.joins++ + try { + const result = await recorded.result + recorded.collected = true + return result + } finally { + recorded.joins-- + } + } + if (background && runningBackgroundTasks(ctx.sessionID) >= MAX_RUNNING_BACKGROUND_TASKS) { + throw new Error( + `Too many background tasks running for this session (limit ${MAX_RUNNING_BACKGROUND_TASKS}). Collect a running task's result by calling task with its task_id before launching another.`, + ) + } + // altimate_change end // Skip permission check when user explicitly invoked via @ or command subtask if (!ctx.extra?.bypassAgentCheck) { @@ -214,67 +416,128 @@ export const TaskTool = Tool.define("task", async (ctx) => { }) // altimate_change end } - ctx.abort.addEventListener("abort", cancel) - using _ = defer(() => ctx.abort.removeEventListener("abort", cancel)) - const promptParts = promptOps - ? await Effect.runPromise(promptOps.resolvePromptParts(params.prompt)) - : await SessionPrompt.resolvePromptParts(params.prompt) - // altimate_change start — upstream_fix: cancellation during task preparation must not start new child work. - ctx.abort.throwIfAborted() - // altimate_change end + // altimate_change start — upstream_fix: the child run is a unit a background task can detach. + const run = async (): Promise => { + ctx.abort.addEventListener("abort", cancel) + using _ = defer(() => ctx.abort.removeEventListener("abort", cancel)) + const promptParts = promptOps + ? await Effect.runPromise(promptOps.resolvePromptParts(params.prompt)) + : await SessionPrompt.resolvePromptParts(params.prompt) + // altimate_change start — upstream_fix: cancellation during task preparation must not start new child work. + ctx.abort.throwIfAborted() + // altimate_change end - const promptInput: SessionPrompt.PromptInput = { - messageID, - sessionID: session.id, - model: { - // altimate_change start — re-brand to provider/schema ModelID/ProviderID - // (identity at runtime); agent.model carries the core ModelV2/ProviderV2 brands. - modelID: ModelID.make(model.modelID), - providerID: ProviderID.make(model.providerID), - // altimate_change end - }, - agent: agent.name, - parts: promptParts, + const promptInput: SessionPrompt.PromptInput = { + messageID, + sessionID: session.id, + model: { + // altimate_change start — re-brand to provider/schema ModelID/ProviderID + // (identity at runtime); agent.model carries the core ModelV2/ProviderV2 brands. + modelID: ModelID.make(model.modelID), + providerID: ProviderID.make(model.providerID), + // altimate_change end + }, + agent: agent.name, + parts: promptParts, + } + // altimate_change start — keep the inherited query cutoff scoped to this child prompt + using queryBudget = defer(RunBudget.registerQueryDeadline(session.id, RunBudget.queryDeadline(ctx.sessionID))) + // altimate_change end + // altimate_change start — upstream_fix: return partial child findings after cancellation, even if its last turn is empty. + let partialOutput: string | undefined + const result = await (promptOps + ? Effect.runPromise(promptOps.prompt(promptInput)) + : SessionPrompt.prompt(promptInput) + ).catch(async (error) => { + if (!ctx.abort.aborted) throw error + partialOutput = await interruptedTaskOutput(session.id, messageID) + if (!partialOutput) throw error + return undefined + }) + if (ctx.abort.aborted) partialOutput ??= await interruptedTaskOutput(session.id, messageID) + + const text = result?.parts.findLast((x) => x.type === "text")?.text ?? "" + + const output = partialOutput ?? [ + `task_id: ${session.id} (for resuming to continue this task if needed)`, + "", + "", + ...(ctx.abort.aborted ? ["Task interrupted. Findings may be incomplete."] : []), + text, + "", + ].join("\n") + // altimate_change end + + return { + title: params.description, + metadata: { + sessionId: session.id, + model, + // altimate_change start — upstream_fix: distinguish partial task findings from completed work. + ...(ctx.abort.aborted ? { interrupted: true } : {}), + // altimate_change end + }, + output, + } } - // altimate_change start — keep the inherited query cutoff scoped to this child prompt - using queryBudget = defer(RunBudget.registerQueryDeadline(session.id, RunBudget.queryDeadline(ctx.sessionID))) - // altimate_change end - // altimate_change start — upstream_fix: return partial child findings after cancellation, even if its last turn is empty. - let partialOutput: string | undefined - const result = await (promptOps - ? Effect.runPromise(promptOps.prompt(promptInput)) - : SessionPrompt.prompt(promptInput) - ).catch(async (error) => { - if (!ctx.abort.aborted) throw error - partialOutput = await interruptedTaskOutput(session.id, messageID) - if (!partialOutput) throw error - return undefined - }) - if (ctx.abort.aborted) partialOutput ??= await interruptedTaskOutput(session.id, messageID) - - const text = result?.parts.findLast((x) => x.type === "text")?.text ?? "" - - const output = partialOutput ?? [ - `task_id: ${session.id} (for resuming to continue this task if needed)`, - "", - "", - ...(ctx.abort.aborted ? ["Task interrupted. Findings may be incomplete."] : []), - text, - "", - ].join("\n") - // altimate_change end + if (!background) return await run() + // The abort listener and query budget registered inside run() stay in place until the child + // settles, so a background child remains bound to the parent's abort signal and run budget. + ctx.abort.throwIfAborted() + const result = run() + .catch((error): TaskResult => { + if (!ctx.abort.aborted) throw error + return { + title: params.description, + metadata: { sessionId: session.id, model, interrupted: true }, + output: [ + resumeHeader(session.id), + "", + "", + "Task cancelled before it returned any findings.", + "", + ].join("\n"), + } + }) + .then((settled) => recordedResult(session.id, settled)) + const record: BackgroundTask = { + parentSessionID: ctx.sessionID, + messageID, + model, + result, + settled: false, + collected: false, + joins: 0, + } + result.then( + async (settled) => { + record.settled = true + if (!settled.metadata.interrupted || record.collected || record.joins > 0) return + await recordCancellation({ + sessionID: ctx.sessionID, + messageID: ctx.messageID, + callID: ctx.callID, + childID: session.id, + output: settled.output, + }) + }, + (err) => { + record.settled = true + log.warn("background task failed", { sessionID: session.id, err }) + }, + ) + backgroundTasks.set(session.id, record) + evictBackgroundRecords(ctx.sessionID) return { title: params.description, - metadata: { - sessionId: session.id, - model, - // altimate_change start — upstream_fix: distinguish partial task findings from completed work. - ...(ctx.abort.aborted ? { interrupted: true } : {}), - // altimate_change end - }, - output, + metadata: { sessionId: session.id, model, background: true }, + output: [ + `task_id: ${session.id}`, + `Task is running in the background. Its result is collected by calling task again with task_id=${session.id}; it is cancelled if your turn ends before you collect it.`, + ].join("\n"), } + // altimate_change end }, } }) diff --git a/packages/opencode/test/session/background-task-loop.test.ts b/packages/opencode/test/session/background-task-loop.test.ts new file mode 100644 index 0000000000..6f8260cbf7 --- /dev/null +++ b/packages/opencode/test/session/background-task-loop.test.ts @@ -0,0 +1,423 @@ +/** + * Background subagents through the real prompt loop with a mock LLM: a child's result is never + * lost or silently replaced. The parent and its child share the mock server; requests are told + * apart by the parent's first user text, which a child never sees. + */ +import { afterAll, beforeAll, expect } from "bun:test" +import { Effect, Fiber, Layer } from "effect" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Session } from "@/session/session" +import { SessionPrompt } from "../../src/session/prompt" +import { SessionSummary } from "../../src/session/summary" +import { MessageID, type SessionID } from "../../src/session/schema" +import type { SessionV1 } from "@opencode-ai/core/v1/session" +import { Database } from "@opencode-ai/core/database/database" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { testEffect } from "../lib/effect" +import { TestLLMServer, raw, reply, type Item, type Reply } from "../lib/llm-server" +import { provideTmpdirServerLegacy } from "./legacy-instance" +import { LSP } from "@/lsp/lsp" +import { MCP } from "../../src/mcp" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { RuntimeFlags } from "@/effect/runtime-flags" + +const mcp = Layer.succeed( + MCP.Service, + MCP.Service.of({ + status: () => Effect.succeed({}), + clients: () => Effect.succeed({}), + tools: () => Effect.succeed({}), + prompts: () => Effect.succeed({}), + resources: () => Effect.succeed({}), + add: () => Effect.succeed({ status: { status: "disabled" as const } }), + connect: () => Effect.void, + disconnect: () => Effect.void, + remove: () => Effect.void, + getPrompt: () => Effect.succeed(undefined), + readResource: () => Effect.succeed(undefined), + startAuth: () => Effect.die("unexpected MCP auth"), + authenticate: () => Effect.die("unexpected MCP auth"), + finishAuth: () => Effect.die("unexpected MCP auth"), + removeAuth: () => Effect.void, + supportsOAuth: () => Effect.succeed(false), + hasStoredTokens: () => Effect.succeed(false), + getAuthStatus: () => Effect.succeed("not_authenticated" as const), + entry: () => Effect.succeed(undefined), + listMeta: () => Effect.succeed(undefined), + snapshot: () => Effect.succeed({ tools: {}, meta: undefined }), + }), +) + +const lsp = Layer.succeed( + LSP.Service, + LSP.Service.of({ + init: () => Effect.void, + status: () => Effect.succeed([]), + hasClients: () => Effect.succeed(false), + touchFile: () => Effect.void, + diagnostics: () => Effect.succeed({}), + hover: () => Effect.succeed(undefined), + definition: () => Effect.succeed([]), + references: () => Effect.succeed([]), + implementation: () => Effect.succeed([]), + documentSymbol: () => Effect.succeed([]), + workspaceSymbol: () => Effect.succeed([]), + prepareCallHierarchy: () => Effect.succeed([]), + incomingCalls: () => Effect.succeed([]), + outgoingCalls: () => Effect.succeed([]), + }), +) + +const root = LayerNode.group([ + SessionPrompt.node, + Session.node, + SessionProjector.node, + SessionSummary.node, + Database.node, + CrossSpawnSpawner.node, + LayerNode.make(TestLLMServer.layer, []), +]) +const it = testEffect( + LayerNode.buildLayer(root, { + replacements: [ + LayerNode.replace(MCP.node, mcp), + LayerNode.replace(LSP.node, lsp), + LayerNode.replace( + RuntimeFlags.node, + RuntimeFlags.layer({ experimentalEventSystem: true, experimentalBackgroundSubagents: true }), + ), + ], + }), +) + +// The imperative prompt path builds its tools without RuntimeFlags, from the environment. +const FLAG = "OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS" +let previousFlag: string | undefined +beforeAll(() => { + previousFlag = process.env[FLAG] + process.env[FLAG] = "true" +}) +afterAll(() => { + if (previousFlag === undefined) delete process.env[FLAG] + else process.env[FLAG] = previousFlag +}) + +const providerCfg = (url: string) => ({ + provider: { + "altimate-backend": { + name: "Altimate (test)", + id: "altimate-backend", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: {}, + }, + }, + options: { apiKey: "test-key", baseURL: url }, + }, + }, + model: "altimate-backend/test-model", +}) + +const PARENT = "PARENT-TURN" +const CHILD = "CHILDWORK" +type Hit = { body: Record } +const text = (hit: Hit) => JSON.stringify(hit.body) +const isParent = (hit: Hit) => text(hit).includes(PARENT) +const isChild = (hit: Hit) => !isParent(hit) && text(hit).includes(CHILD) +const taskIdIn = (hit: Hit) => /task_id: (ses_[A-Za-z0-9]+)/.exec(text(hit))?.[1] + +const launch = { description: "scan", prompt: CHILD, subagent_type: "general", background: true } +const join = (taskID: string, description = "collect") => ({ + description, + prompt: "collect the result", + subagent_type: "general", + task_id: taskID, +}) + +const chunk = (delta: Record, finish?: string) => ({ + id: "chatcmpl-test", + object: "chat.completion.chunk", + choices: [{ delta, ...(finish ? { finish_reason: finish } : {}) }], +}) + +/** One assistant step with several tool calls (`Reply.tool` puts every call at index 0). */ +function toolCalls(...calls: { name: string; input: unknown }[]): Item { + return raw({ + head: [chunk({ role: "assistant" })], + tail: [ + ...calls.flatMap((call, index) => [ + chunk({ + tool_calls: [{ index, id: `call_${index}`, type: "function", function: { name: call.name, arguments: "" } }], + }), + chunk({ tool_calls: [{ index, function: { arguments: JSON.stringify(call.input) } }] }), + ]), + chunk({}, "tool_calls"), + ], + }) +} + +/** A parent reply built when its request arrives, so it can name the task_id the launch returned. */ +function lazy(build: (taskID: string) => Reply | Item): [(hit: Hit) => boolean, Item] { + const slot = raw({}) + return [ + (hit) => { + if (!isParent(hit)) return false + const taskID = taskIdIn(hit) + if (!taskID) return false + const built = build(taskID) + Object.assign(slot, "item" in built ? built.item() : built) + return true + }, + slot, + ] +} + +function gate() { + let open!: () => void + const promise = new Promise((resolve) => (open = resolve)) + return { promise, open } +} + +const poll = (read: () => Effect.Effect, label: string, ms = 10_000) => + Effect.gen(function* () { + const until = Date.now() + ms + while (Date.now() < until) { + const value = yield* read() + if (value !== undefined) return value + yield* Effect.sleep("20 millis") + } + throw new Error(`timed out waiting for ${label}`) + }) + +const taskParts = (messages: SessionV1.WithParts[]) => + messages.flatMap((message) => + message.parts.filter((part): part is SessionV1.ToolPart => part.type === "tool" && part.tool === "task"), + ) + +/** The child's request has started streaming its partial text (persisted when the stream ends). */ +const childStreaming = Effect.gen(function* () { + const llm = yield* TestLLMServer + yield* poll(() => Effect.map(llm.hits, (hits) => (hits.some(isChild) ? true : undefined)), "child request") + yield* Effect.sleep("300 millis") +}) + +const startSession = Effect.gen(function* () { + const sessions = yield* Session.Service + return yield* sessions.create({ + title: "background tasks", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) +}) + +const userTurn = (sessionID: SessionID, value: string) => + Effect.gen(function* () { + const prompt = yield* SessionPrompt.Service + return yield* prompt.prompt({ + sessionID, + messageID: MessageID.ascending(), + agent: "build", + parts: [{ type: "text", text: `${PARENT} ${value}` }], + }) + }) + +it.live( + "ending the turn without collecting cancels the child visibly, and a later collect returns its partial findings", + () => + provideTmpdirServerLegacy( + Effect.fnUntraced(function* ({ llm }) { + const sessions = yield* Session.Service + const session = yield* startSession + const parentDone = gate() + yield* llm.pushMatch(isParent, reply().tool("task", launch)) + yield* llm.pushMatch(isChild, reply().text("PARTIALFINDING").hang()) + yield* llm.pushMatch(isParent, reply().wait(parentDone.promise).text("Started it. I will be notified.").stop()) + + const turn = yield* userTurn(session.id, "one").pipe(Effect.forkChild) + const launched = yield* poll( + () => + Effect.gen(function* () { + const part = taskParts(yield* sessions.messages({ sessionID: session.id }))[0] + return part?.state.status === "completed" ? part : undefined + }), + "launch", + ) + if (launched.state.status !== "completed") throw new Error("launch not completed") + const childID = launched.state.metadata?.sessionId as SessionID + expect(launched.state.output).toContain(`task_id=${childID}`) + expect(launched.state.output).toContain("cancelled if your turn ends before you collect it") + yield* childStreaming + parentDone.open() + yield* Fiber.join(turn) + + // The launch part now records the cancellation and the partial findings. + const annotated = yield* poll( + () => + Effect.gen(function* () { + const part = taskParts(yield* sessions.messages({ sessionID: session.id }))[0] + return part?.state.status === "completed" && + part.state.output.includes("was cancelled at the end of the turn") + ? part + : undefined + }), + "cancellation record on the launch part", + ) + if (annotated.state.status !== "completed") throw new Error("unreachable") + expect(annotated.state.output).toContain(`Background task ${childID} was cancelled at the end of the turn`) + expect(annotated.state.output).toContain("PARTIALFINDING") + expect(annotated.state.metadata?.interrupted).toBe(true) + + // Next turn: collecting returns the recorded partial findings and never re-prompts the child. + yield* llm.pushMatch(isParent, reply().tool("task", join(childID))) + yield* llm.pushMatch(isParent, reply().text("final").stop()) + yield* userTurn(session.id, "two") + + const parts = taskParts(yield* sessions.messages({ sessionID: session.id })) + const collected = parts[parts.length - 1] + if (collected.state.status !== "completed") throw new Error(`collect ended ${collected.state.status}`) + expect(collected.state.output).toContain("PARTIALFINDING") + expect(collected.state.output).toContain("Task interrupted") + expect(collected.state.metadata?.interrupted).toBe(true) + const hits = yield* llm.hits + expect(hits.filter((hit) => !isParent(hit) && text(hit).includes("collect the result"))).toHaveLength(0) + const child = yield* sessions.messages({ sessionID: childID }) + expect(child.filter((message) => message.info.role === "user")).toHaveLength(1) + }), + { git: true, config: providerCfg }, + ), + { timeout: 30_000 }, +) + +it.live( + "parallel and repeated collects return the same result and prompt the child once", + () => + provideTmpdirServerLegacy( + Effect.fnUntraced(function* ({ llm }) { + const sessions = yield* Session.Service + const session = yield* startSession + const childDone = gate() + yield* llm.pushMatch(isParent, reply().tool("task", launch)) + yield* llm.pushMatch(isChild, reply().wait(childDone.promise).text("CHILDRESULT").stop()) + yield* llm.pushMatch( + ...lazy((id) => + toolCalls({ name: "task", input: join(id, "join a") }, { name: "task", input: join(id, "join b") }), + ), + ) + yield* llm.pushMatch(...lazy((id) => reply().tool("task", join(id, "join c")))) + yield* llm.pushMatch(isParent, reply().text("final").stop()) + + const turn = yield* userTurn(session.id, "one").pipe(Effect.forkChild) + yield* poll( + () => + Effect.gen(function* () { + const parts = taskParts(yield* sessions.messages({ sessionID: session.id })) + return parts.filter((part) => part.state.status === "running").length === 2 ? true : undefined + }), + "two running joins", + ) + childDone.open() + yield* Fiber.join(turn) + + const parts = taskParts(yield* sessions.messages({ sessionID: session.id })) + const joins = parts.filter((part) => part.state.status === "completed" && part.state.input.task_id) + expect(joins).toHaveLength(3) + const outputs = joins.map((part) => (part.state.status === "completed" ? part.state.output : "")) + for (const output of outputs) expect(output).toContain("\nCHILDRESULT\n") + expect(new Set(outputs).size).toBe(1) + const hits = yield* llm.hits + expect(hits.filter(isChild)).toHaveLength(1) + }), + { git: true, config: providerCfg }, + ), + { timeout: 30_000 }, +) + +it.live( + "an errored child gives the same outcome on every collect and is not re-run", + () => + provideTmpdirServerLegacy( + Effect.fnUntraced(function* ({ llm }) { + const sessions = yield* Session.Service + const session = yield* startSession + yield* llm.pushMatch(isParent, reply().tool("task", launch)) + yield* llm.pushMatch(isChild, { type: "http-error", status: 400, body: { error: { message: "CHILDBROKE" } } }) + yield* llm.pushMatch(...lazy((id) => reply().tool("task", join(id, "join a")))) + yield* llm.pushMatch(...lazy((id) => reply().tool("task", join(id, "join b")))) + yield* llm.pushMatch(isParent, reply().text("final").stop()) + + yield* userTurn(session.id, "one") + + const parts = taskParts(yield* sessions.messages({ sessionID: session.id })) + const joins = parts.filter((part) => part.state.input.task_id) + expect(joins).toHaveLength(2) + const outcome = (part: SessionV1.ToolPart) => + part.state.status === "completed" + ? { status: part.state.status, output: part.state.output } + : part.state.status === "error" + ? { status: part.state.status, output: part.state.error } + : { status: part.state.status, output: "" } + expect(outcome(joins[1])).toEqual(outcome(joins[0])) + const hits = yield* llm.hits + expect(hits.filter(isChild)).toHaveLength(1) + expect(hits.filter((hit) => !isParent(hit) && text(hit).includes("collect the result"))).toHaveLength(0) + }), + { git: true, config: providerCfg }, + ), + { timeout: 30_000 }, +) + +it.live( + "cancelling the parent while it waits on a collect keeps the child's partial findings", + () => + provideTmpdirServerLegacy( + Effect.fnUntraced(function* ({ llm }) { + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const session = yield* startSession + yield* llm.pushMatch(isParent, reply().tool("task", launch)) + yield* llm.pushMatch(isChild, reply().text("PARTIALFINDING").hang()) + yield* llm.pushMatch(...lazy((id) => reply().tool("task", join(id)))) + + const turn = yield* userTurn(session.id, "one").pipe(Effect.forkChild) + const childID = yield* poll( + () => + Effect.gen(function* () { + const parts = taskParts(yield* sessions.messages({ sessionID: session.id })) + const running = parts.find((part) => part.state.status === "running" && part.state.input.task_id) + return running?.state.input.task_id as SessionID | undefined + }), + "running collect", + ) + yield* childStreaming + yield* prompt.cancel(session.id) + yield* Fiber.await(turn) + + const collected = yield* poll( + () => + Effect.gen(function* () { + const parts = taskParts(yield* sessions.messages({ sessionID: session.id })) + const part = parts.find((part) => part.state.input.task_id) + return part && part.state.status !== "running" && part.state.status !== "pending" ? part : undefined + }), + "settled collect", + ) + if (collected.state.status !== "completed") { + throw new Error(`collect ended ${collected.state.status}: ${JSON.stringify(collected.state)}`) + } + expect(collected.state.output).toContain("Task interrupted. These are partial findings") + expect(collected.state.output).toContain("PARTIALFINDING") + }), + { git: true, config: providerCfg }, + ), + { timeout: 30_000 }, +) diff --git a/packages/opencode/test/tool/registry.test.ts b/packages/opencode/test/tool/registry.test.ts index ec6dcfa607..c1e917b40a 100644 --- a/packages/opencode/test/tool/registry.test.ts +++ b/packages/opencode/test/tool/registry.test.ts @@ -2,7 +2,7 @@ import { afterEach, describe, expect } from "bun:test" import path from "path" import fs from "fs/promises" import { fileURLToPath, pathToFileURL } from "url" -import { Effect, Layer, Result, Schema } from "effect" +import { Cause, Effect, Exit, Layer, Result, Schema } from "effect" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { ToolRegistry } from "@/tool/registry" import { Tool } from "@/tool/tool" @@ -70,6 +70,16 @@ const withParallel = testEffect( }), ) // altimate_change end +// altimate_change start — upstream_fix: the background experiment publishes task.background. +const withBackground = testEffect( + LayerNode.buildLayer(root, { + replacements: [ + LayerNode.replace(Config.node, configLayer), + LayerNode.replace(RuntimeFlags.node, RuntimeFlags.layer({ experimentalBackgroundSubagents: true })), + ], + }), +) +// altimate_change end afterEach(async () => { await disposeAllInstances() @@ -102,6 +112,91 @@ describe("tool.registry", () => { }), ) + // altimate_change start — upstream_fix: the background experiment publishes task.background. + withBackground.instance("exposes task background parameter when experimental background subagents are enabled", () => + Effect.gen(function* () { + const registry = yield* ToolRegistry.Service + const agent = yield* Agent.Service + const build = yield* agent.get("build") + if (!build) throw new Error("build agent not found") + const task = (yield* registry.tools({ + providerID: ProviderID.opencode, + modelID: ModelID.make("test"), + agent: build, + })).find((tool) => tool.id === "task") + + const properties = task?.jsonSchema?.properties as Record | undefined + expect(properties?.background?.type).toBe("boolean") + expect(properties?.task_id).toBeDefined() + }), + ) + + const taskFrom = Effect.gen(function* () { + const registry = yield* ToolRegistry.Service + const build = yield* (yield* Agent.Service).get("build") + if (!build) throw new Error("build agent not found") + return (yield* registry.tools({ + providerID: ProviderID.opencode, + modelID: ModelID.make("test"), + agent: build, + })).find((tool) => tool.id === "task")! + }) + + it.instance("keeps the task description when experimental background subagents are off", () => + Effect.gen(function* () { + const task = yield* taskFrom + expect(task.description).toContain("For background tasks, you will be notified automatically") + }), + ) + + withBackground.instance("tells the model to collect background tasks before its turn ends", () => + Effect.gen(function* () { + const task = yield* taskFrom + expect(task.description).not.toContain("notified automatically") + expect(task.description).toContain("calling task again with its task_id") + expect(task.description).toContain("ending your turn cancels") + }), + ) + + withBackground.instance("execution follows the same flag that published the background field", () => + Effect.gen(function* () { + const previous = { + flag: process.env.OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS, + all: process.env.OPENCODE_EXPERIMENTAL, + } + delete process.env.OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS + delete process.env.OPENCODE_EXPERIMENTAL + try { + const task = yield* taskFrom + const controller = new AbortController() + controller.abort() + const exit = yield* task + .execute( + { description: "x", prompt: "x", subagent_type: "general", background: true }, + { + sessionID: SessionID.make("ses_missing"), + messageID: MessageID.ascending(), + agent: "build", + abort: controller.signal, + extra: { bypassAgentCheck: true }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + .pipe(Effect.exit) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + expect(String(Cause.squash(exit.cause))).not.toContain("require OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS") + } + } finally { + if (previous.flag !== undefined) process.env.OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS = previous.flag + if (previous.all !== undefined) process.env.OPENCODE_EXPERIMENTAL = previous.all + } + }), + ) + // altimate_change end + withParallel.instance("exposes websearch for non-opencode providers when Parallel is enabled", () => Effect.gen(function* () { const registry = yield* ToolRegistry.Service diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index e682febd21..ebecd8d77b 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -2,7 +2,10 @@ import { afterEach, describe, expect } from "bun:test" import { SessionV1 } from "@opencode-ai/core/v1/session" import type { PermissionV1 } from "@opencode-ai/core/v1/permission" import { Database } from "@opencode-ai/core/database/database" -import { Deferred, Effect, Exit, Fiber, Layer } from "effect" +// altimate_change start — upstream_fix: background subagent tests assert error text and budget binding. +import { Cause, ConfigProvider, Deferred, Effect, Exit, Fiber, Layer } from "effect" +import { RunBudget } from "@/altimate/run-budget" +// altimate_change end import { Agent } from "../../src/agent/agent" import { BackgroundJob } from "@/background/job" import { EventV2Bridge } from "@/event-v2-bridge" @@ -17,7 +20,12 @@ import { SessionRunState } from "@/session/run-state" import { SessionStatus } from "@/session/status" import { interruptedTaskOutput } from "@/session/task-output" -import { TaskTool, type TaskPromptOps } from "../../src/tool/task" +import { + MAX_RUNNING_BACKGROUND_TASKS, + TaskTool, + backgroundSubagentsEnabled, + type TaskPromptOps, +} from "../../src/tool/task" import { Truncate } from "@/tool/truncate" import { ToolRegistry } from "@/tool/registry" import { RuntimeFlags } from "@/effect/runtime-flags" @@ -59,6 +67,24 @@ const layer = (flags: Partial = {}) => const it = testEffect(layer()) const background = testEffect(layer({ experimentalBackgroundSubagents: true })) +// altimate_change start — upstream_fix: the env fallback the task tool reads at execution time. +const BACKGROUND_FLAG = "OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS" +const withBackgroundEnv = (effect: Effect.Effect) => + Effect.acquireUseRelease( + Effect.sync(() => { + const previous = process.env[BACKGROUND_FLAG] + process.env[BACKGROUND_FLAG] = "true" + return previous + }), + () => effect, + (previous) => + Effect.sync(() => { + if (previous === undefined) delete process.env[BACKGROUND_FLAG] + else process.env[BACKGROUND_FLAG] = previous + }), + ) +// altimate_change end + function defer() { let resolve!: (value: T | PromiseLike) => void const promise = new Promise((done) => { @@ -589,6 +615,13 @@ describe("tool.task", () => { .pipe(Effect.exit) expect(Exit.isFailure(exit)).toBe(true) + // altimate_change start — upstream_fix: the disabled-experiment error text is unchanged. + if (Exit.isFailure(exit)) { + expect(Cause.pretty(exit.cause)).toContain( + "Background subagents require OPENCODE_EXPERIMENTAL_BACKGROUND_SUBAGENTS=true", + ) + } + // altimate_change end }), ) @@ -823,312 +856,274 @@ describe("tool.task", () => { }), ) - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("execute launches background tasks without waiting for completion", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() - - const result = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - { - sessionID: chat.id, - messageID: assistant.id, - agent: "build", - abort: new AbortController().signal, - extra: { - promptOps: { - ...stubOps(), - prompt: () => Effect.never, - } satisfies TaskPromptOps, - }, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - }, - ) - - const job = yield* jobs.get(result.metadata.sessionId) - expect(result.metadata.background).toBe(true) - expect(result.output).toContain(`state="running"`) - expect(job?.status).toBe("running") - }), - ) - - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("background task completion waits for running updates", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() - const first = defer() - const second = defer() - const updated = defer() - const injected = defer() - let prompts = 0 - const promptOps: TaskPromptOps = { - ...stubOps(), - prompt: (input) => { - if (input.sessionID === chat.id) { - injected.resolve(input) - return Effect.succeed(reply(input, "done")) - } - prompts++ - if (prompts === 1) return Effect.promise(() => first.promise).pipe(Effect.as(reply(input, "first done"))) - updated.resolve(input) - return Effect.promise(() => second.promise).pipe(Effect.as(reply(input, "second done"))) - }, - } - const context = { - sessionID: chat.id, - messageID: assistant.id, - agent: "build", - abort: new AbortController().signal, - extra: { promptOps }, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - } - - const started = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - context, - ) - const result = yield* def.execute( - { - description: "add investigation scope", - prompt: "also inspect cancellation", - subagent_type: "general", - task_id: started.metadata.sessionId, - }, - context, - ) - - expect(result.metadata.sessionId).toBe(started.metadata.sessionId) - expect(result.metadata.background).toBe(true) - expect(result.output).toContain("Background task updated") - first.resolve() - expect((yield* jobs.get(started.metadata.sessionId))?.status).toBe("running") - expect((yield* Effect.promise(() => updated.promise)).parts).toEqual([ - { type: "text", text: "also inspect cancellation" }, - ]) - - second.resolve() - 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") - }), - ) - - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("background tasks complete through the background job service", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() - - const result = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - { + // altimate_change start — upstream_fix: background subagents run without blocking the parent turn. + background.instance("background task returns immediately and a task_id call joins the child result", () => + withBackgroundEnv( + Effect.gen(function* () { + const { chat, assistant } = yield* seed() + const tool = yield* TaskTool + const def = yield* tool.init() + const started = defer() + const release = defer() + let prompts = 0 + let finished = false + const promptOps: TaskPromptOps = { + ...stubOps(), + prompt: (input) => + Effect.promise(async () => { + prompts++ + started.resolve() + await release.promise + finished = true + return reply(input, "background done") + }), + } + const context = { sessionID: chat.id, messageID: assistant.id, agent: "build", abort: new AbortController().signal, - extra: { promptOps: stubOps({ text: "background done" }) }, + extra: { promptOps }, messages: [], metadata: () => Effect.void, ask: () => Effect.void, - }, - ) + } + const deadline = Date.now() + 60_000 + const disposeDeadline = RunBudget.registerQueryDeadline(chat.id, deadline) - const waited = yield* jobs.wait({ id: result.metadata.sessionId, timeout: 1_000 }) - expect(waited.timedOut).toBe(false) - expect(waited.info?.status).toBe("completed") - expect(waited.info?.output).toBe("background done") - }), + const launched = yield* def.execute( + { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, + }, + context, + ) + const taskID = launched.metadata.sessionId + expect(launched.metadata.background).toBe(true) + expect(launched.output).toContain(`task_id: ${taskID}`) + expect(launched.output).toContain("running in the background") + yield* Effect.promise(() => started.promise).pipe(Effect.timeout("2 seconds")) + // The tool returned while the child's model call is still pending. + expect(finished).toBe(false) + // The child inherits the parent's run budget until it settles. + expect(RunBudget.queryDeadline(taskID)).toBe(deadline) + + let joinDone = false + const join = yield* def + .execute( + { + description: "collect result", + prompt: "ignored while running", + subagent_type: "general", + task_id: taskID, + }, + context, + ) + .pipe( + Effect.tap(() => Effect.sync(() => (joinDone = true))), + Effect.forkChild, + ) + yield* Effect.sleep("30 millis") + expect(joinDone).toBe(false) + + release.resolve() + const joined = yield* Fiber.join(join).pipe(Effect.timeout("2 seconds")) + disposeDeadline() + expect(joined.metadata.sessionId).toBe(taskID) + expect(joined.metadata.background).toBeUndefined() + expect(joined.output).toContain( + `task_id: ${taskID} (background task: calling task with this task_id returns this same result)`, + ) + expect(joined.output).toContain("\nbackground done\n") + expect(prompts).toBe(1) + expect(RunBudget.queryDeadline(taskID)).toBeUndefined() + }), + ), ) - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("background task completion does not wait for the parent async prompt", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() + background.instance("parent abort cancels a running background child", () => + withBackgroundEnv( + Effect.gen(function* () { + const { chat, assistant } = yield* seed() + const tool = yield* TaskTool + const def = yield* tool.init() + const started = defer() + const release = defer() + const cancelled = defer() + const controller = new AbortController() + const promptOps: TaskPromptOps = { + ...stubOps(), + cancel: (sessionID) => + Effect.sync(() => { + cancelled.resolve(sessionID) + release.resolve() + }), + prompt: (input) => + Effect.promise(async () => { + started.resolve() + await release.promise + return reply(input, "") + }), + } - const result = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - { - sessionID: chat.id, - messageID: assistant.id, - agent: "build", - abort: new AbortController().signal, - extra: { - promptOps: { - ...stubOps({ text: "background done" }), - prompt: (input) => - input.sessionID === chat.id ? Effect.never : Effect.succeed(reply(input, "background done")), - } satisfies TaskPromptOps, + const launched = yield* def.execute( + { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, }, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - }, - ) + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: controller.signal, + extra: { promptOps }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + expect(launched.metadata.background).toBe(true) + yield* Effect.promise(() => started.promise).pipe(Effect.timeout("2 seconds")) - const waited = yield* jobs.wait({ id: result.metadata.sessionId, timeout: 1_000 }) - expect(waited.timedOut).toBe(false) - expect(waited.info?.status).toBe("completed") - }), + // The parent loop aborts this signal on cancel, on a hard run deadline, and when its run ends. + controller.abort() + const id = yield* Effect.promise(() => cancelled.promise).pipe(Effect.timeout("2 seconds")) + expect(id).toBe(launched.metadata.sessionId) + }), + ), ) - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("removing the parent session cancels running background tasks", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const sessions = yield* Session.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() - - const result = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - { - sessionID: chat.id, - messageID: assistant.id, - agent: "build", - abort: new AbortController().signal, - extra: { - promptOps: { - ...stubOps(), - prompt: () => Effect.never, - } satisfies TaskPromptOps, - }, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - }, - ) + const backgroundContext = ( + chat: { id: SessionID }, + assistant: { id: MessageID }, + promptOps: TaskPromptOps, + abort: AbortSignal = new AbortController().signal, + ) => ({ + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort, + extra: { promptOps }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }) + const launchInput = { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, + } + const blockingOps = (started: { resolve: () => void }, release: Promise): TaskPromptOps => ({ + ...stubOps(), + prompt: (input) => + Effect.promise(async () => { + started.resolve() + await release + return reply(input, "background done") + }), + }) - yield* sessions.remove(chat.id) - const waited = yield* jobs.wait({ id: result.metadata.sessionId, timeout: 1_000 }) - expect(waited.timedOut).toBe(false) - expect(waited.info?.status).toBe("cancelled") - }), + background.instance("a launch on an already-aborted signal fails instead of reporting a running task", () => + withBackgroundEnv( + Effect.gen(function* () { + const { chat, assistant } = yield* seed() + const def = yield* (yield* TaskTool).init() + let prompts = 0 + const controller = new AbortController() + controller.abort() + const exit = yield* def + .execute( + launchInput, + backgroundContext(chat, assistant, stubOps({ onPrompt: () => prompts++ }), controller.signal), + ) + .pipe(Effect.exit) + expect(Exit.isFailure(exit)).toBe(true) + expect(prompts).toBe(0) + }), + ), ) - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("removing the child task session cancels its running background task", () => - Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const sessions = yield* Session.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() - - const result = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - { - sessionID: chat.id, - messageID: assistant.id, - agent: "build", - abort: new AbortController().signal, - extra: { - promptOps: { - ...stubOps(), - prompt: () => Effect.never, - } satisfies TaskPromptOps, - }, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - }, - ) + background.instance("background: true with the task_id of a running child returns its status at once", () => + withBackgroundEnv( + Effect.gen(function* () { + const { chat, assistant } = yield* seed() + const def = yield* (yield* TaskTool).init() + const started = defer() + const release = defer() + const context = backgroundContext(chat, assistant, blockingOps(started, release.promise)) + const launched = yield* def.execute(launchInput, context) + const taskID = launched.metadata.sessionId + yield* Effect.promise(() => started.promise).pipe(Effect.timeout("2 seconds")) + + const status = yield* def.execute({ ...launchInput, task_id: taskID }, context).pipe(Effect.timeout("1 second")) + expect(status.metadata.background).toBe(true) + expect(status.output).toContain("still running") + expect(status.output).toContain(`task_id=${taskID}`) + release.resolve() + }), + ), + ) - yield* sessions.remove(result.metadata.sessionId) - const waited = yield* jobs.wait({ id: result.metadata.sessionId, timeout: 1_000 }) - expect(waited.timedOut).toBe(false) - expect(waited.info?.status).toBe("cancelled") - }), + background.instance("a parent cannot run more background children than the cap", () => + withBackgroundEnv( + Effect.gen(function* () { + const { chat, assistant } = yield* seed() + const def = yield* (yield* TaskTool).init() + const release = defer() + const context = backgroundContext(chat, assistant, blockingOps({ resolve: () => {} }, release.promise)) + for (let index = 0; index < MAX_RUNNING_BACKGROUND_TASKS; index++) yield* def.execute(launchInput, context) + const exit = yield* def.execute(launchInput, context).pipe(Effect.exit) + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) expect(String(Cause.squash(exit.cause))).toContain("background tasks running") + release.resolve() + }), + ), ) - // BUG: blocked by the storage-layer DB split-brain (see note above). - background.instance.skip("cancelling the parent run cancels running background tasks", () => + it.instance("the execution flag check parses the variable exactly as RuntimeFlags does", () => Effect.gen(function* () { - const jobs = yield* BackgroundJob.Service - const runState = yield* SessionRunState.Service - const { chat, assistant } = yield* seed() - const tool = yield* TaskTool - const def = yield* tool.init() - - const result = yield* def.execute( - { - description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - background: true, - } as any, - { - sessionID: chat.id, - messageID: assistant.id, - agent: "build", - abort: new AbortController().signal, - extra: { - promptOps: { - ...stubOps(), - prompt: () => Effect.never, - } satisfies TaskPromptOps, - }, - messages: [], - metadata: () => Effect.void, - ask: () => Effect.void, - }, - ) - - yield* runState.cancel(chat.id) - const waited = yield* jobs.wait({ id: result.metadata.sessionId, timeout: 1_000 }) - expect(waited.timedOut).toBe(false) - expect(waited.info?.status).toBe("cancelled") + const previous = { flag: process.env[BACKGROUND_FLAG], experimental: process.env["OPENCODE_EXPERIMENTAL"] } + delete process.env["OPENCODE_EXPERIMENTAL"] + try { + // RuntimeFlags (Effect `Config.boolean`) is what publishes the `background` field. Run it on its + // own runtime: the test runtime has already memoized the flags it built at startup. + const runtimeFlag = () => + Effect.gen(function* () { + return (yield* RuntimeFlags.Service).experimentalBackgroundSubagents + }).pipe( + Effect.provide( + RuntimeFlags.defaultLayer.pipe(Layer.provide(ConfigProvider.layer(ConfigProvider.fromEnv()))), + ), + ) + const cases = { + on: true, + y: true, + "1": true, + yes: true, + true: true, + TRUE: false, + false: false, + "0": false, + off: false, + } + for (const [value, expected] of Object.entries(cases)) { + process.env[BACKGROUND_FLAG] = value + expect({ value, runtime: Effect.runSync(runtimeFlag()) }).toEqual({ value, runtime: expected }) + expect({ value, enabled: backgroundSubagentsEnabled() }).toEqual({ value, enabled: expected }) + } + } finally { + if (previous.flag === undefined) delete process.env[BACKGROUND_FLAG] + else process.env[BACKGROUND_FLAG] = previous.flag + if (previous.experimental !== undefined) process.env["OPENCODE_EXPERIMENTAL"] = previous.experimental + } }), ) + // altimate_change end it.instance("cancelling a child run cancels its own pre-runner task job", () => Effect.gen(function* () {