From 971c837ad479c21950f7ec0b3dcfb2a5e7b24c40 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 24 Apr 2026 20:08:24 +0530 Subject: [PATCH 01/11] feat(task): add background subagent support --- packages/opencode/src/session/prompt.ts | 4 + packages/opencode/src/tool/registry.ts | 11 +- packages/opencode/src/tool/task.ts | 182 +++++++++--- packages/opencode/src/tool/task.txt | 12 +- packages/opencode/src/tool/task_status.ts | 145 +++++++++ packages/opencode/src/tool/task_status.txt | 13 + packages/opencode/test/tool/task.test.ts | 185 +++++++++++- .../opencode/test/tool/task_status.test.ts | 278 ++++++++++++++++++ 8 files changed, 775 insertions(+), 55 deletions(-) create mode 100644 packages/opencode/src/tool/task_status.ts create mode 100644 packages/opencode/src/tool/task_status.txt create mode 100644 packages/opencode/test/tool/task_status.test.ts diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 5f3530bcef..c96f655bd0 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -115,6 +115,10 @@ export const layer = Layer.effect( cancel: (sessionID: SessionID) => run.fork(cancel(sessionID)), resolvePromptParts: (template: string) => resolvePromptParts(template), prompt: (input: PromptInput) => prompt(input), + loop: (input: LoopInput) => loop(input), + fork: (effect: Effect.Effect) => { + run.fork(effect) + }, } satisfies TaskPromptOps }) diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index 539ad63202..eb0c75c7ac 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -7,6 +7,7 @@ import { GlobTool } from "./glob" import { GrepTool } from "./grep" import { ReadTool } from "./read" import { TaskTool } from "./task" +import { TaskStatusTool } from "./task_status" import { TodoWriteTool } from "./todo" import { WebFetchTool } from "./webfetch" import { WriteTool } from "./write" @@ -47,6 +48,7 @@ import { Bus } from "../bus" import { Agent } from "../agent/agent" import { Skill } from "../skill" import { Permission } from "@/permission" +import { SessionStatus } from "@/session/status" const log = Log.create({ service: "tool.registry" }) @@ -78,8 +80,9 @@ export const layer: Layer.Layer< | Todo.Service | Agent.Service | Skill.Service - | Session.Service - | Provider.Service + | Session.Service + | SessionStatus.Service + | Provider.Service | LSP.Service | Instruction.Service | AppFileSystem.Service @@ -115,6 +118,7 @@ export const layer: Layer.Layer< const greptool = yield* GrepTool const patchtool = yield* ApplyPatchTool const skilltool = yield* SkillTool + const taskstatus = yield* TaskStatusTool const agent = yield* Agent.Service const state = yield* InstanceState.make( @@ -195,6 +199,7 @@ export const layer: Layer.Layer< edit: Tool.init(edit), write: Tool.init(writetool), task: Tool.init(task), + taskstatus: Tool.init(taskstatus), fetch: Tool.init(webfetch), todo: Tool.init(todo), search: Tool.init(websearch), @@ -218,6 +223,7 @@ export const layer: Layer.Layer< tool.edit, tool.write, tool.task, + tool.taskstatus, tool.fetch, tool.todo, tool.search, @@ -335,6 +341,7 @@ export const defaultLayer = Layer.suspend(() => Layer.provide(Skill.defaultLayer), Layer.provide(Agent.defaultLayer), Layer.provide(Session.defaultLayer), + Layer.provide(SessionStatus.defaultLayer), Layer.provide(Provider.defaultLayer), Layer.provide(LSP.defaultLayer), Layer.provide(Instruction.defaultLayer), diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 5cb0dc6a83..2bc58cdbb4 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -1,17 +1,22 @@ import * as Tool from "./tool" import DESCRIPTION from "./task.txt" +import { Bus } from "../bus" import { Session } from "../session" import { SessionID, MessageID } from "../session/schema" import { MessageV2 } from "../session/message-v2" import { Agent } from "../agent/agent" import type { SessionPrompt } from "../session/prompt" +import { SessionStatus } from "../session/status" import { Config } from "../config" -import { Effect, Schema } from "effect" +import { TuiEvent } from "@/cli/cmd/tui/event" +import { Cause, Effect, Option, Schema } from "effect" export interface TaskPromptOps { cancel(sessionID: SessionID): void resolvePromptParts(template: string): Effect.Effect prompt(input: SessionPrompt.PromptInput): Effect.Effect + loop(input: SessionPrompt.LoopInput): Effect.Effect + fork(effect: Effect.Effect): void } const id = "task" @@ -20,19 +25,61 @@ export const Parameters = Schema.Struct({ description: Schema.String.annotate({ description: "A short (3-5 words) description of the task" }), prompt: Schema.String.annotate({ description: "The task for the agent to perform" }), subagent_type: Schema.String.annotate({ description: "The type of specialized agent to use for this task" }), - task_id: Schema.optional(Schema.String).annotate({ + task_id: Schema.optional(SessionID).annotate({ description: "This should only be set if you mean to resume a previous task (you can pass a prior task_id and the task will continue the same subagent session as before instead of creating a fresh one)", }), command: Schema.optional(Schema.String).annotate({ description: "The command that triggered this task" }), + background: Schema.optional(Schema.Boolean).annotate({ + description: "When true, launch the subagent in the background and return immediately", + }), }) +function output(sessionID: SessionID, text: string) { + return [ + `task_id: ${sessionID} (for resuming to continue this task if needed)`, + "", + "", + text, + "", + ].join("\n") +} + +function backgroundOutput(sessionID: SessionID) { + return [ + `task_id: ${sessionID} (for polling this task with task_status)`, + "state: running", + "", + "", + "Background task started. Continue your current work and call task_status when you need the result.", + "", + ].join("\n") +} + +function backgroundMessage(input: { sessionID: SessionID; description: string; state: "completed" | "error"; text: string }) { + const tag = input.state === "completed" ? "task_result" : "task_error" + const title = + input.state === "completed" + ? `Background task completed: ${input.description}` + : `Background task failed: ${input.description}` + return [title, `task_id: ${input.sessionID}`, `state: ${input.state}`, `<${tag}>`, input.text, ``].join( + "\n", + ) +} + +function errorText(error: unknown) { + if (error instanceof Error) return error.message + return String(error) +} + export const TaskTool = Tool.define( id, Effect.gen(function* () { const agent = yield* Agent.Service + const bus = yield* Bus.Service const config = yield* Config.Service const sessions = yield* Session.Service + const status = yield* SessionStatus.Service const run = Effect.fn("TaskTool.execute")(function* ( params: Schema.Schema.Type, @@ -62,7 +109,7 @@ export const TaskTool = Tool.define( const taskID = params.task_id const session = taskID - ? yield* sessions.get(SessionID.make(taskID)).pipe(Effect.catchCause(() => Effect.succeed(undefined))) + ? yield* sessions.get(taskID).pipe(Effect.catchCause(() => Effect.succeed(undefined))) : undefined const nextSession = session ?? @@ -103,19 +150,107 @@ export const TaskTool = Tool.define( modelID: msg.info.modelID, providerID: msg.info.providerID, } + const parentModel = { + modelID: msg.info.modelID, + providerID: msg.info.providerID, + } + const background = params.background === true + + const metadata = { + sessionId: nextSession.id, + model, + ...(background ? { background: true } : {}), + } yield* ctx.metadata({ title: params.description, - metadata: { - sessionId: nextSession.id, - model, - }, + metadata, }) const ops = ctx.extra?.promptOps as TaskPromptOps if (!ops) return yield* Effect.fail(new Error("TaskTool requires promptOps in ctx.extra")) - const messageID = MessageID.ascending() + const runTask = Effect.fn("TaskTool.runTask")(function* () { + const parts = yield* ops.resolvePromptParts(params.prompt) + const result = yield* ops.prompt({ + messageID: MessageID.ascending(), + sessionID: nextSession.id, + model: { + modelID: model.modelID, + providerID: model.providerID, + }, + agent: next.name, + tools: { + ...(canTodo ? {} : { todowrite: false }), + ...(canTask ? {} : { task: false }), + ...Object.fromEntries((cfg.experimental?.primary_tools ?? []).map((item) => [item, false])), + }, + parts, + }) + return result.parts.findLast((item) => item.type === "text")?.text ?? "" + }) + + const continueIfIdle = Effect.fn("TaskTool.continueIfIdle")(function* (input: { + userID: MessageID + state: "completed" | "error" + }) { + if ((yield* status.get(ctx.sessionID)).type !== "idle") return + const latest = yield* sessions.findMessage(ctx.sessionID, (item) => item.info.role === "user") + if (Option.isNone(latest)) return + if (latest.value.info.id !== input.userID) return + yield* bus.publish(TuiEvent.ToastShow, { + title: input.state === "completed" ? "Background task complete" : "Background task failed", + message: + input.state === "completed" + ? `Background task \"${params.description}\" finished. Resuming the main thread.` + : `Background task \"${params.description}\" failed. Resuming the main thread.`, + variant: input.state === "completed" ? "success" : "error", + duration: 5000, + }) + yield* ops.loop({ sessionID: ctx.sessionID }).pipe(Effect.ignore) + }) + + if (background) { + const inject = Effect.fn("TaskTool.injectBackgroundResult")(function* (state: "completed" | "error", text: string) { + const message = yield* ops.prompt({ + sessionID: ctx.sessionID, + noReply: true, + model: parentModel, + agent: ctx.agent, + parts: [ + { + type: "text", + synthetic: true, + text: backgroundMessage({ + sessionID: nextSession.id, + description: params.description, + state, + text, + }), + }, + ], + }) + yield* continueIfIdle({ userID: message.info.id, state }) + }) + + ops.fork( + runTask().pipe( + Effect.matchCauseEffect({ + onSuccess: (text) => inject("completed", text), + onFailure: (cause) => + inject("error", errorText(Cause.squash(cause))).pipe(Effect.catchCause(() => Effect.void)), + }), + Effect.catchCause(() => Effect.void), + Effect.asVoid, + ), + ) + + return { + title: params.description, + metadata, + output: backgroundOutput(nextSession.id), + } + } function cancel() { ops.cancel(nextSession.id) @@ -127,36 +262,11 @@ export const TaskTool = Tool.define( }), () => Effect.gen(function* () { - const parts = yield* ops.resolvePromptParts(params.prompt) - const result = yield* ops.prompt({ - messageID, - sessionID: nextSession.id, - model: { - modelID: model.modelID, - providerID: model.providerID, - }, - agent: next.name, - tools: { - ...(canTodo ? {} : { todowrite: false }), - ...(canTask ? {} : { task: false }), - ...Object.fromEntries((cfg.experimental?.primary_tools ?? []).map((item) => [item, false])), - }, - parts, - }) - + const text = yield* runTask() return { title: params.description, - metadata: { - sessionId: nextSession.id, - model, - }, - output: [ - `task_id: ${nextSession.id} (for resuming to continue this task if needed)`, - "", - "", - result.parts.findLast((item) => item.type === "text")?.text ?? "", - "", - ].join("\n"), + metadata, + output: output(nextSession.id, text), } }), () => diff --git a/packages/opencode/src/tool/task.txt b/packages/opencode/src/tool/task.txt index fba8470d1b..5d26066a8c 100644 --- a/packages/opencode/src/tool/task.txt +++ b/packages/opencode/src/tool/task.txt @@ -14,11 +14,13 @@ When NOT to use the Task tool: Usage notes: 1. Launch multiple agents concurrently whenever possible, to maximize performance; to do that, use a single message with multiple tool uses -2. When the agent is done, it will return a single message back to you. The result returned by the agent is not visible to the user. To show the user the result, you should send a text message back to the user with a concise summary of the result. The output includes a task_id you can reuse later to continue the same subagent session. -3. Each agent invocation starts with a fresh context unless you provide task_id to resume the same subagent session (which continues with its previous messages and tool outputs). When starting fresh, your prompt should contain a highly detailed task description for the agent to perform autonomously and you should specify exactly what information the agent should return back to you in its final and only message to you. -4. The agent's outputs should generally be trusted -5. Clearly tell the agent whether you expect it to write code or just to do research (search, file reads, web fetches, etc.), since it is not aware of the user's intent. Tell it how to verify its work if possible (e.g., relevant test commands). -6. If the agent description mentions that it should be used proactively, then you should try your best to use it without the user having to ask for it first. Use your judgement. +2. By default, task waits for completion and returns the result immediately, along with a task_id you can reuse later to continue the same subagent session. +3. Set background=true to launch asynchronously. In background mode, continue your current work without waiting. +4. For background runs, use task_status(task_id=..., wait=false) to poll, or wait=true to block until done (optionally with timeout_ms). +5. Each agent invocation starts with a fresh context unless you provide task_id to resume the same subagent session (which continues with its previous messages and tool outputs). When starting fresh, your prompt should contain a highly detailed task description for the agent to perform autonomously and you should specify exactly what information the agent should return back to you in its final and only message to you. +6. The agent's outputs should generally be trusted +7. Clearly tell the agent whether you expect it to write code or just to do research (search, file reads, web fetches, etc.), since it is not aware of the user's intent. Tell it how to verify its work if possible (e.g., relevant test commands). +8. If the agent description mentions that it should be used proactively, then you should try your best to use it without the user having to ask for it first. Use your judgement. Example usage (NOTE: The agents below are fictional examples for illustration only - use the actual agents listed above): diff --git a/packages/opencode/src/tool/task_status.ts b/packages/opencode/src/tool/task_status.ts new file mode 100644 index 0000000000..60d5218a38 --- /dev/null +++ b/packages/opencode/src/tool/task_status.ts @@ -0,0 +1,145 @@ +import * as Tool from "./tool" +import DESCRIPTION from "./task_status.txt" +import { Session } from "../session" +import { SessionID } from "../session/schema" +import { MessageV2 } from "../session/message-v2" +import { SessionStatus } from "../session/status" +import { PositiveInt } from "@/util/schema" +import { Effect, Option, Schema } from "effect" + +const DEFAULT_TIMEOUT = 60_000 +const POLL_MS = 300 + +const Parameters = Schema.Struct({ + task_id: SessionID.annotate({ description: "The task_id returned by the task tool" }), + wait: Schema.optional(Schema.Boolean).annotate({ description: "When true, wait until the task reaches a terminal state or timeout" }), + timeout_ms: Schema.optional(PositiveInt).annotate({ + description: "Maximum milliseconds to wait when wait=true (default: 60000)", + }), +}) + +type State = "running" | "completed" | "error" +type InspectResult = { state: State; text: string } + +function format(input: { taskID: SessionID; state: State; text: string }) { + return [`task_id: ${input.taskID}`, `state: ${input.state}`, "", "", input.text, ""].join( + "\n", + ) +} + +function errorText(error: NonNullable) { + const data = Reflect.get(error, "data") + const message = data && typeof data === "object" ? Reflect.get(data, "message") : undefined + if (typeof message === "string" && message) return message + return error.name +} + +export const TaskStatusTool = Tool.define( + "task_status", + Effect.gen(function* () { + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + + const inspect: (taskID: SessionID) => Effect.Effect = Effect.fn("TaskStatusTool.inspect")(function* ( + taskID: SessionID, + ) { + const current = yield* status.get(taskID) + if (current.type === "busy" || current.type === "retry") { + return { + state: "running" as const, + text: current.type === "retry" ? `Task is retrying: ${current.message}` : "Task is still running.", + } + } + + const latestAssistant = yield* sessions.findMessage(taskID, (item) => item.info.role === "assistant") + if (Option.isNone(latestAssistant)) { + return { + state: "running" as const, + text: "Task has started but has not produced output yet.", + } + } + if (latestAssistant.value.info.role !== "assistant") { + return { + state: "running" as const, + text: "Task has started but has not produced output yet.", + } + } + + const latestUser = yield* sessions.findMessage(taskID, (item) => item.info.role === "user") + if (Option.isSome(latestUser) && latestUser.value.info.role === "user" && latestUser.value.info.id > latestAssistant.value.info.id) { + return { + state: "running" as const, + text: "Task is starting.", + } + } + + const text = latestAssistant.value.parts.findLast((part) => part.type === "text")?.text ?? "" + if (latestAssistant.value.info.error) { + return { + state: "error" as const, + text: text || errorText(latestAssistant.value.info.error), + } + } + + const done = + !!latestAssistant.value.info.finish && !["tool-calls", "unknown"].includes(latestAssistant.value.info.finish) + if (done) { + return { + state: "completed" as const, + text, + } + } + + return { + state: "running" as const, + text: text || "Task is still running.", + } + }) + + const waitForTerminal: (taskID: SessionID, timeout: number) => Effect.Effect<{ result: InspectResult; timedOut: boolean }> = + Effect.fn("TaskStatusTool.waitForTerminal")(function* (taskID: SessionID, timeout: number) { + const result = yield* inspect(taskID) + if (result.state !== "running") return { result, timedOut: false } + if (timeout <= 0) return { result, timedOut: true } + const sleep = Math.min(POLL_MS, timeout) + yield* Effect.sleep(`${sleep} millis`) + return yield* waitForTerminal(taskID, timeout - sleep) + }) + + const run = Effect.fn("TaskStatusTool.execute")(function* ( + params: Schema.Schema.Type, + _ctx: Tool.Context, + ) { + yield* sessions.get(params.task_id) + + const waited = + params.wait === true + ? yield* waitForTerminal(params.task_id, params.timeout_ms ?? DEFAULT_TIMEOUT) + : { result: yield* inspect(params.task_id), timedOut: false } + + const outputText = waited.timedOut + ? `Timed out after ${params.timeout_ms ?? DEFAULT_TIMEOUT}ms while waiting for task completion.` + : waited.result.text + + return { + title: "Task status", + metadata: { + task_id: params.task_id, + state: waited.result.state, + timed_out: waited.timedOut, + }, + output: format({ + taskID: params.task_id, + state: waited.result.state, + text: outputText, + }), + } + }) + + return { + description: DESCRIPTION, + parameters: Parameters, + execute: (params: Schema.Schema.Type, ctx: Tool.Context) => run(params, ctx).pipe(Effect.orDie), + } + }), +) diff --git a/packages/opencode/src/tool/task_status.txt b/packages/opencode/src/tool/task_status.txt new file mode 100644 index 0000000000..3f8af0d609 --- /dev/null +++ b/packages/opencode/src/tool/task_status.txt @@ -0,0 +1,13 @@ +Poll the status of a subagent task launched with the task tool. + +Use this to check background tasks started with `task(background=true)`. + +Parameters: +- `task_id` (required): the task session id returned by the task tool +- `wait` (optional): when true, wait for completion +- `timeout_ms` (optional): max wait duration in milliseconds when `wait=true` + +Returns compact, parseable output: +- `task_id` +- `state` (`running`, `completed`, or `error`) +- `...` containing final output, error summary, or current progress text diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index b94dd52086..95ee49a0c2 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -1,13 +1,15 @@ import { afterEach, describe, expect } from "bun:test" import { Effect, Layer } from "effect" import { Agent } from "../../src/agent/agent" +import { Bus } from "../../src/bus" import { Config } from "../../src/config" import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner" import { Instance } from "../../src/project/instance" import { Session } from "../../src/session" import { MessageV2 } from "../../src/session/message-v2" import type { SessionPrompt } from "../../src/session/prompt" -import { MessageID, PartID } from "../../src/session/schema" +import { MessageID, PartID, SessionID } from "../../src/session/schema" +import { SessionStatus } from "../../src/session/status" import { ModelID, ProviderID } from "../../src/provider/schema" import { TaskTool, type TaskPromptOps } from "../../src/tool/task" import { Truncate } from "../../src/tool" @@ -27,9 +29,11 @@ const ref = { const it = testEffect( Layer.mergeAll( Agent.defaultLayer, + Bus.defaultLayer, Config.defaultLayer, CrossSpawnSpawner.defaultLayer, Session.defaultLayer, + SessionStatus.defaultLayer, Truncate.defaultLayer, ToolRegistry.defaultLayer, ), @@ -64,15 +68,59 @@ const seed = Effect.fn("TaskToolTest.seed")(function* (title = "Pinned") { return { chat, assistant } }) -function stubOps(opts?: { onPrompt?: (input: SessionPrompt.PromptInput) => void; text?: string }): TaskPromptOps { +function stubOps(session: Session.Interface, opts?: { onPrompt?: (input: SessionPrompt.PromptInput) => void; text?: string }): TaskPromptOps { return { cancel() {}, resolvePromptParts: (template) => Effect.succeed([{ type: "text" as const, text: template }]), prompt: (input) => - Effect.sync(() => { + Effect.gen(function* () { opts?.onPrompt?.(input) - return reply(input, opts?.text ?? "done") + const userID = input.messageID ?? MessageID.ascending() + const user: MessageV2.User = { + id: userID, + role: "user", + sessionID: input.sessionID, + agent: input.agent ?? "build", + model: input.model ?? ref, + tools: input.tools, + time: { created: Date.now() }, + } + yield* session.updateMessage(user) + + const parts = input.parts.map((part) => ({ + ...part, + id: part.id ?? PartID.ascending(), + messageID: user.id, + sessionID: input.sessionID, + })) + yield* Effect.forEach(parts, (part) => session.updatePart(part), { discard: true }) + + if (input.noReply) { + return { + info: user, + parts, + } + } + + const result = reply({ ...input, messageID: user.id }, opts?.text ?? "done") + yield* session.updateMessage(result.info) + yield* Effect.forEach(result.parts, (part) => session.updatePart(part), { discard: true }) + return result }), + loop: (input) => + Effect.sync(() => + reply( + { + sessionID: input.sessionID, + messageID: MessageID.ascending(), + agent: "build", + model: ref, + parts: [], + }, + opts?.text ?? "done", + ), + ), + fork() {}, } } @@ -195,7 +243,7 @@ describe("tool.task", () => { const tool = yield* TaskTool const def = yield* tool.init() let seen: SessionPrompt.PromptInput | undefined - const promptOps = stubOps({ text: "resumed", onPrompt: (input) => (seen = input) }) + const promptOps = stubOps(sessions, { text: "resumed", onPrompt: (input) => (seen = input) }) const result = yield* def.execute( { @@ -229,11 +277,12 @@ describe("tool.task", () => { it.live("execute asks by default and skips checks when bypassed", () => provideTmpdirInstance(() => Effect.gen(function* () { + const sessions = yield* Session.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() const calls: unknown[] = [] - const promptOps = stubOps() + const promptOps = stubOps(sessions) const exec = (extra?: Record) => def.execute( @@ -282,15 +331,15 @@ describe("tool.task", () => { const tool = yield* TaskTool const def = yield* tool.init() let seen: SessionPrompt.PromptInput | undefined - const promptOps = stubOps({ text: "created", onPrompt: (input) => (seen = input) }) + const promptOps = stubOps(sessions, { text: "created", onPrompt: (input) => (seen = input) }) const result = yield* def.execute( { description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - task_id: "ses_missing", - }, + prompt: "look into the cache key path", + subagent_type: "general", + task_id: SessionID.make("ses_missing"), + }, { sessionID: chat.id, messageID: assistant.id, @@ -322,7 +371,7 @@ describe("tool.task", () => { const tool = yield* TaskTool const def = yield* tool.init() let seen: SessionPrompt.PromptInput | undefined - const promptOps = stubOps({ onPrompt: (input) => (seen = input) }) + const promptOps = stubOps(sessions, { onPrompt: (input) => (seen = input) }) const result = yield* def.execute( { @@ -384,4 +433,116 @@ describe("tool.task", () => { }, ), ) + + it.live("execute launches background tasks without waiting for completion", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const { chat, assistant } = yield* seed() + const tool = yield* TaskTool + const def = yield* tool.init() + const forks: Effect.Effect[] = [] + + const result = yield* def.execute( + { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { + promptOps: { + ...stubOps(sessions), + fork(effect) { + forks.push(effect) + }, + } satisfies TaskPromptOps, + }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.metadata.sessionId).toBeDefined() + expect(result.metadata.background).toBe(true) + expect(result.output).toContain(`task_id: ${result.metadata.sessionId}`) + expect(result.output).toContain("state: running") + expect(forks).toHaveLength(1) + }), + ), + ) + + it.live("background tasks inject completion into the parent session and resume when idle", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const { chat, assistant } = yield* seed() + const tool = yield* TaskTool + const def = yield* tool.init() + const forks: Effect.Effect[] = [] + const loops: string[] = [] + + const result = yield* def.execute( + { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { + promptOps: { + ...stubOps(sessions, { text: "background done" }), + loop(input) { + loops.push(input.sessionID) + return Effect.sync(() => + reply( + { + sessionID: input.sessionID, + messageID: MessageID.ascending(), + agent: "build", + model: ref, + parts: [], + }, + "looped", + ), + ) + }, + fork(effect) { + forks.push(effect) + }, + } satisfies TaskPromptOps, + }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + yield* forks[0]! + + const parent = yield* sessions.findMessage(chat.id, (msg) => msg.info.role === "user") + expect(parent._tag).toBe("Some") + if (parent._tag !== "Some") return + expect(parent.value.parts.find((part) => part.type === "text")?.text).toContain("Background task completed") + expect(parent.value.parts.find((part) => part.type === "text")?.text).toContain("background done") + expect(loops).toEqual([chat.id]) + + const child = yield* sessions.findMessage(result.metadata.sessionId, (msg) => msg.info.role === "assistant") + expect(child._tag).toBe("Some") + if (child._tag !== "Some") return + expect(child.value.parts.find((part) => part.type === "text")?.text).toBe("background done") + }), + ), + ) }) diff --git a/packages/opencode/test/tool/task_status.test.ts b/packages/opencode/test/tool/task_status.test.ts new file mode 100644 index 0000000000..6081f6a923 --- /dev/null +++ b/packages/opencode/test/tool/task_status.test.ts @@ -0,0 +1,278 @@ +import { afterEach, describe, expect } from "bun:test" +import { Effect, Layer, Scope } from "effect" +import { Agent } from "../../src/agent/agent" +import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner" +import { Instance } from "../../src/project/instance" +import { Session } from "../../src/session" +import { MessageV2 } from "../../src/session/message-v2" +import { MessageID, PartID } from "../../src/session/schema" +import { SessionStatus } from "../../src/session/status" +import { TaskStatusTool } from "../../src/tool/task_status" +import { Truncate } from "../../src/tool" +import { ModelID, ProviderID } from "../../src/provider/schema" +import { provideTmpdirInstance } from "../fixture/fixture" +import { testEffect } from "../lib/effect" + +afterEach(async () => { + await Instance.disposeAll() +}) + +const ref = { + providerID: ProviderID.make("test"), + modelID: ModelID.make("test-model"), +} + +const it = testEffect( + Layer.mergeAll( + Agent.defaultLayer, + CrossSpawnSpawner.defaultLayer, + Session.defaultLayer, + SessionStatus.defaultLayer, + Truncate.defaultLayer, + ), +) + +const seedUser = Effect.fn("TaskStatusToolTest.seedUser")(function* (sessionID: Session.Info["id"]) { + const session = yield* Session.Service + return yield* session.updateMessage({ + id: MessageID.ascending(), + role: "user", + sessionID, + agent: "build", + model: ref, + time: { created: Date.now() }, + }) +}) + +const seedAssistant = Effect.fn("TaskStatusToolTest.seedAssistant")(function* (input: { + sessionID: Session.Info["id"] + text: string + error?: string +}) { + const session = yield* Session.Service + const user = yield* seedUser(input.sessionID) + const message = yield* session.updateMessage({ + id: MessageID.ascending(), + role: "assistant", + parentID: user.id, + sessionID: input.sessionID, + mode: "build", + agent: "build", + cost: 0, + path: { cwd: "/tmp", root: "/tmp" }, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: ref.modelID, + providerID: ref.providerID, + time: { created: Date.now(), completed: Date.now() }, + finish: "stop", + ...(input.error + ? { + error: new MessageV2.APIError({ + message: input.error, + isRetryable: false, + }).toObject(), + } + : {}), + }) + + yield* session.updatePart({ + id: PartID.ascending(), + messageID: message.id, + sessionID: input.sessionID, + type: "text", + text: input.text, + }) +}) + +describe("tool.task_status", () => { + it.live("returns running while session status is busy", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const tool = yield* TaskStatusTool + const def = yield* tool.init() + const chat = yield* sessions.create({}) + + yield* status.set(chat.id, { type: "busy" }) + const result = yield* def.execute( + { task_id: chat.id }, + { + sessionID: chat.id, + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("state: running") + }), + ), + ) + + it.live("returns completed with final task output", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const tool = yield* TaskStatusTool + const def = yield* tool.init() + const chat = yield* sessions.create({}) + + yield* seedAssistant({ sessionID: chat.id, text: "all done" }) + + const result = yield* def.execute( + { task_id: chat.id }, + { + sessionID: chat.id, + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("state: completed") + expect(result.output).toContain("all done") + }), + ), + ) + + it.live("wait=true blocks until terminal status", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const tool = yield* TaskStatusTool + const def = yield* tool.init() + const chat = yield* sessions.create({}) + const scope = yield* Scope.Scope + + yield* status.set(chat.id, { type: "busy" }) + yield* Effect.gen(function* () { + yield* Effect.sleep("150 millis") + yield* status.set(chat.id, { type: "idle" }) + yield* seedAssistant({ sessionID: chat.id, text: "finished later" }) + }).pipe(Effect.forkIn(scope)) + + const result = yield* def.execute( + { + task_id: chat.id, + wait: true, + timeout_ms: 4_000, + }, + { + sessionID: chat.id, + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("state: completed") + expect(result.output).toContain("finished later") + }), + ), + ) + + it.live("returns error when child run fails", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const tool = yield* TaskStatusTool + const def = yield* tool.init() + const chat = yield* sessions.create({}) + + yield* seedAssistant({ sessionID: chat.id, text: "", error: "child failed" }) + + const result = yield* def.execute( + { task_id: chat.id }, + { + sessionID: chat.id, + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("state: error") + expect(result.output).toContain("child failed") + expect(result.metadata.state).toBe("error") + }), + ), + ) + + it.live("wait=true times out with timed_out metadata", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const tool = yield* TaskStatusTool + const def = yield* tool.init() + const chat = yield* sessions.create({}) + + yield* status.set(chat.id, { type: "busy" }) + const result = yield* def.execute( + { + task_id: chat.id, + wait: true, + timeout_ms: 80, + }, + { + sessionID: chat.id, + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("Timed out after 80ms") + expect(result.metadata.timed_out).toBe(true) + expect(result.metadata.state).toBe("running") + }), + ), + ) + + it.live("returns running for resumed task with a newer user turn", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const sessions = yield* Session.Service + const tool = yield* TaskStatusTool + const def = yield* tool.init() + const chat = yield* sessions.create({}) + + yield* seedAssistant({ sessionID: chat.id, text: "old done" }) + yield* seedUser(chat.id) + + const result = yield* def.execute( + { task_id: chat.id }, + { + sessionID: chat.id, + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect(result.output).toContain("state: running") + expect(result.output).toContain("Task is starting.") + }), + ), + ) +}) From 7970130720a1acf4a405b5bf587cb880e3273713 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 24 Apr 2026 20:08:32 +0530 Subject: [PATCH 02/11] fix(ui): label background task cards --- packages/opencode/src/cli/cmd/tui/routes/session/index.tsx | 3 ++- packages/ui/src/components/message-part.tsx | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx index c04e58acec..b6cde459e9 100644 --- a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx +++ b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx @@ -2000,7 +2000,8 @@ function Task(props: ToolProps) { const content = createMemo(() => { if (!props.input.description) return "" - let content = [`${Locale.titlecase(props.input.subagent_type ?? "General")} Task — ${props.input.description}`] + const description = props.metadata.background === true ? `${props.input.description} (background)` : props.input.description + let content = [`${Locale.titlecase(props.input.subagent_type ?? "General")} Task — ${description}`] if (isRunning() && tools().length > 0) { // content[0] += ` · ${tools().length} toolcalls` diff --git a/packages/ui/src/components/message-part.tsx b/packages/ui/src/components/message-part.tsx index 9c0c90c000..3694b204fb 100644 --- a/packages/ui/src/components/message-part.tsx +++ b/packages/ui/src/components/message-part.tsx @@ -1751,9 +1751,10 @@ ToolRegistry.register({ const title = createMemo(() => agent().name ?? i18n.t("ui.tool.agent.default")) const tone = createMemo(() => agent().color) const subtitle = createMemo(() => { - const value = props.input.description - if (typeof value === "string" && value) return value - return childSessionId() + const value = typeof props.input.description === "string" && props.input.description ? props.input.description : childSessionId() + if (!value) return value + if (props.metadata.background === true) return `${value} (background)` + return value }) const running = createMemo(() => props.status === "pending" || props.status === "running") From ecde8ab3633b10919b5fa67938282d7e806b121d Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 24 Apr 2026 20:20:04 +0530 Subject: [PATCH 03/11] test(task): update parameter schema snapshot --- .../test/tool/__snapshots__/parameters.test.ts.snap | 5 +++++ packages/opencode/test/tool/parameters.test.ts | 13 +++++++++++++ 2 files changed, 18 insertions(+) diff --git a/packages/opencode/test/tool/__snapshots__/parameters.test.ts.snap b/packages/opencode/test/tool/__snapshots__/parameters.test.ts.snap index eb3fe6cce4..ffb1b9f55f 100644 --- a/packages/opencode/test/tool/__snapshots__/parameters.test.ts.snap +++ b/packages/opencode/test/tool/__snapshots__/parameters.test.ts.snap @@ -334,6 +334,10 @@ exports[`tool parameters JSON Schema (wire shape) task 1`] = ` { "$schema": "https://json-schema.org/draft/2020-12/schema", "properties": { + "background": { + "description": "When true, launch the subagent in the background and return immediately", + "type": "boolean", + }, "command": { "description": "The command that triggered this task", "type": "string", @@ -352,6 +356,7 @@ exports[`tool parameters JSON Schema (wire shape) task 1`] = ` }, "task_id": { "description": "This should only be set if you mean to resume a previous task (you can pass a prior task_id and the task will continue the same subagent session as before instead of creating a fresh one)", + "pattern": "^ses.*", "type": "string", }, }, diff --git a/packages/opencode/test/tool/parameters.test.ts b/packages/opencode/test/tool/parameters.test.ts index 8ea008a457..487e6faa17 100644 --- a/packages/opencode/test/tool/parameters.test.ts +++ b/packages/opencode/test/tool/parameters.test.ts @@ -220,6 +220,19 @@ describe("tool parameters", () => { const parsed = parse(Task, { description: "d", prompt: "p", subagent_type: "general" }) expect(parsed.subagent_type).toBe("general") }) + test("accepts optional task_id + command + background", () => { + const parsed = parse(Task, { + description: "d", + prompt: "p", + subagent_type: "general", + task_id: "ses_test", + command: "/cmd", + background: true, + }) + expect(parsed.task_id).toBe("ses_test") + expect(parsed.command).toBe("/cmd") + expect(parsed.background).toBe(true) + }) test("rejects missing prompt", () => { expect(accepts(Task, { description: "d", subagent_type: "general" })).toBe(false) }) From 1357bb984f8fe70222d468a6ddc03340336ee4c5 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 24 Apr 2026 20:20:04 +0530 Subject: [PATCH 04/11] style: fix background task formatting --- .../src/cli/cmd/tui/routes/session/index.tsx | 3 ++- packages/opencode/src/tool/registry.ts | 6 +++--- packages/opencode/src/tool/task_status.ts | 14 +++++++++++--- packages/opencode/test/tool/task.test.ts | 8 ++++---- packages/ui/src/components/message-part.tsx | 3 ++- 5 files changed, 22 insertions(+), 12 deletions(-) diff --git a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx index b6cde459e9..50e39c7c3a 100644 --- a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx +++ b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx @@ -2000,7 +2000,8 @@ function Task(props: ToolProps) { const content = createMemo(() => { if (!props.input.description) return "" - const description = props.metadata.background === true ? `${props.input.description} (background)` : props.input.description + const description = + props.metadata.background === true ? `${props.input.description} (background)` : props.input.description let content = [`${Locale.titlecase(props.input.subagent_type ?? "General")} Task — ${description}`] if (isRunning() && tools().length > 0) { diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index eb0c75c7ac..64c38ccb4f 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -80,9 +80,9 @@ export const layer: Layer.Layer< | Todo.Service | Agent.Service | Skill.Service - | Session.Service - | SessionStatus.Service - | Provider.Service + | Session.Service + | SessionStatus.Service + | Provider.Service | LSP.Service | Instruction.Service | AppFileSystem.Service diff --git a/packages/opencode/src/tool/task_status.ts b/packages/opencode/src/tool/task_status.ts index 60d5218a38..db7960b485 100644 --- a/packages/opencode/src/tool/task_status.ts +++ b/packages/opencode/src/tool/task_status.ts @@ -66,7 +66,11 @@ export const TaskStatusTool = Tool.define( } const latestUser = yield* sessions.findMessage(taskID, (item) => item.info.role === "user") - if (Option.isSome(latestUser) && latestUser.value.info.role === "user" && latestUser.value.info.id > latestAssistant.value.info.id) { + if ( + Option.isSome(latestUser) && + latestUser.value.info.role === "user" && + latestUser.value.info.id > latestAssistant.value.info.id + ) { return { state: "running" as const, text: "Task is starting.", @@ -96,8 +100,12 @@ export const TaskStatusTool = Tool.define( } }) - const waitForTerminal: (taskID: SessionID, timeout: number) => Effect.Effect<{ result: InspectResult; timedOut: boolean }> = - Effect.fn("TaskStatusTool.waitForTerminal")(function* (taskID: SessionID, timeout: number) { + const waitForTerminal: ( + taskID: SessionID, + timeout: number, + ) => Effect.Effect<{ result: InspectResult; timedOut: boolean }> = Effect.fn( + "TaskStatusTool.waitForTerminal", + )(function* (taskID: SessionID, timeout: number) { const result = yield* inspect(taskID) if (result.state !== "running") return { result, timedOut: false } if (timeout <= 0) return { result, timedOut: true } diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index 95ee49a0c2..a5724630ec 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -336,10 +336,10 @@ describe("tool.task", () => { const result = yield* def.execute( { description: "inspect bug", - prompt: "look into the cache key path", - subagent_type: "general", - task_id: SessionID.make("ses_missing"), - }, + prompt: "look into the cache key path", + subagent_type: "general", + task_id: SessionID.make("ses_missing"), + }, { sessionID: chat.id, messageID: assistant.id, diff --git a/packages/ui/src/components/message-part.tsx b/packages/ui/src/components/message-part.tsx index 3694b204fb..eb90b4d9e6 100644 --- a/packages/ui/src/components/message-part.tsx +++ b/packages/ui/src/components/message-part.tsx @@ -1751,7 +1751,8 @@ ToolRegistry.register({ const title = createMemo(() => agent().name ?? i18n.t("ui.tool.agent.default")) const tone = createMemo(() => agent().color) const subtitle = createMemo(() => { - const value = typeof props.input.description === "string" && props.input.description ? props.input.description : childSessionId() + const value = + typeof props.input.description === "string" && props.input.description ? props.input.description : childSessionId() if (!value) return value if (props.metadata.background === true) return `${value} (background)` return value From 3f4b9d9ef4bf8c8450015c9cc3a7aaa2437058cc Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 24 Apr 2026 20:21:02 +0530 Subject: [PATCH 05/11] test(task): use branded session id in schema test --- packages/opencode/test/tool/parameters.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/opencode/test/tool/parameters.test.ts b/packages/opencode/test/tool/parameters.test.ts index 487e6faa17..52d7f44ee3 100644 --- a/packages/opencode/test/tool/parameters.test.ts +++ b/packages/opencode/test/tool/parameters.test.ts @@ -26,6 +26,7 @@ import { Parameters as Todo } from "../../src/tool/todo" import { Parameters as WebFetch } from "../../src/tool/webfetch" import { Parameters as WebSearch } from "../../src/tool/websearch" import { Parameters as Write } from "../../src/tool/write" +import { SessionID } from "../../src/session/schema" const parse = >(schema: S, input: unknown): S["Type"] => Schema.decodeUnknownSync(schema)(input) @@ -225,11 +226,11 @@ describe("tool parameters", () => { description: "d", prompt: "p", subagent_type: "general", - task_id: "ses_test", + task_id: SessionID.make("ses_test"), command: "/cmd", background: true, }) - expect(parsed.task_id).toBe("ses_test") + expect(parsed.task_id).toBe(SessionID.make("ses_test")) expect(parsed.command).toBe("/cmd") expect(parsed.background).toBe(true) }) From 601fe03a3a571b8da3a9aabc52ee869624fe6e61 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 24 Apr 2026 20:30:53 +0530 Subject: [PATCH 06/11] refactor(task): simplify effect wrappers --- packages/opencode/src/tool/task.ts | 12 +++++------- packages/opencode/src/tool/task_status.ts | 13 ++++++------- 2 files changed, 11 insertions(+), 14 deletions(-) diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 2bc58cdbb4..61b3294be5 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -81,10 +81,9 @@ export const TaskTool = Tool.define( const sessions = yield* Session.Service const status = yield* SessionStatus.Service - const run = Effect.fn("TaskTool.execute")(function* ( - params: Schema.Schema.Type, - ctx: Tool.Context, - ) { + const run = Effect.fn( + "TaskTool.execute", + )(function* (params: Schema.Schema.Type, ctx: Tool.Context) { const cfg = yield* config.get() if (!ctx.extra?.bypassAgentCheck) { @@ -274,13 +273,12 @@ export const TaskTool = Tool.define( ctx.abort.removeEventListener("abort", cancel) }), ) - }) + }, Effect.orDie) return { description: DESCRIPTION, parameters: Parameters, - execute: (params: Schema.Schema.Type, ctx: Tool.Context) => - run(params, ctx).pipe(Effect.orDie), + execute: run, } }), ) diff --git a/packages/opencode/src/tool/task_status.ts b/packages/opencode/src/tool/task_status.ts index db7960b485..eba29576b3 100644 --- a/packages/opencode/src/tool/task_status.ts +++ b/packages/opencode/src/tool/task_status.ts @@ -110,14 +110,13 @@ export const TaskStatusTool = Tool.define( if (result.state !== "running") return { result, timedOut: false } if (timeout <= 0) return { result, timedOut: true } const sleep = Math.min(POLL_MS, timeout) - yield* Effect.sleep(`${sleep} millis`) + yield* Effect.sleep(sleep) return yield* waitForTerminal(taskID, timeout - sleep) }) - const run = Effect.fn("TaskStatusTool.execute")(function* ( - params: Schema.Schema.Type, - _ctx: Tool.Context, - ) { + const run = Effect.fn( + "TaskStatusTool.execute", + )(function* (params: Schema.Schema.Type, _ctx: Tool.Context) { yield* sessions.get(params.task_id) const waited = @@ -142,12 +141,12 @@ export const TaskStatusTool = Tool.define( text: outputText, }), } - }) + }, Effect.orDie) return { description: DESCRIPTION, parameters: Parameters, - execute: (params: Schema.Schema.Type, ctx: Tool.Context) => run(params, ctx).pipe(Effect.orDie), + execute: run, } }), ) From 085fac7c2c41ce773a4a2690e05d015782ca250d Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 1 May 2026 19:04:56 +0530 Subject: [PATCH 07/11] feat(background): add job service --- packages/opencode/src/background/job.ts | 173 ++++++++++++++++++ packages/opencode/src/id/id.ts | 1 + packages/opencode/test/background/job.test.ts | 49 +++++ 3 files changed, 223 insertions(+) create mode 100644 packages/opencode/src/background/job.ts create mode 100644 packages/opencode/test/background/job.test.ts diff --git a/packages/opencode/src/background/job.ts b/packages/opencode/src/background/job.ts new file mode 100644 index 0000000000..5603fb733a --- /dev/null +++ b/packages/opencode/src/background/job.ts @@ -0,0 +1,173 @@ +import { InstanceState } from "@/effect/instance-state" +import { Identifier } from "@/id/id" +import { Cause, Deferred, Effect, Fiber, Layer, Scope, Context } from "effect" + +export type Status = "running" | "completed" | "error" | "cancelled" + +export type Info = { + id: string + type: string + title?: string + status: Status + started_at: number + completed_at?: number + output?: string + error?: string + metadata?: Record +} + +type Active = { + info: Info + done: Deferred.Deferred + fiber?: Fiber.Fiber +} + +type State = { + jobs: Map + scope: Scope.Scope +} + +export type StartInput = { + id?: string + type: string + title?: string + metadata?: Record + run: Effect.Effect +} + +export type WaitInput = { + id: string + timeout?: number +} + +export type WaitResult = { + info?: Info + timedOut: boolean +} + +export interface Interface { + readonly list: () => Effect.Effect + readonly get: (id: string) => Effect.Effect + readonly start: (input: StartInput) => Effect.Effect + readonly wait: (input: WaitInput) => Effect.Effect + readonly cancel: (id: string) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/BackgroundJob") {} + +function snapshot(job: Active): Info { + return { + ...job.info, + ...(job.info.metadata ? { metadata: { ...job.info.metadata } } : {}), + } +} + +function errorText(error: unknown) { + if (error instanceof Error) return error.message + return String(error) +} + +export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const state = yield* InstanceState.make( + Effect.fn("BackgroundJob.state")(function* () { + return { + jobs: new Map(), + scope: yield* Scope.Scope, + } + }), + ) + + const finish = Effect.fn("BackgroundJob.finish")(function* ( + job: Active, + status: Exclude, + data?: { output?: string; error?: string }, + ) { + if (job.info.status !== "running") return snapshot(job) + job.info.status = status + job.info.completed_at = Date.now() + if (data?.output !== undefined) job.info.output = data.output + if (data?.error !== undefined) job.info.error = data.error + job.fiber = undefined + const info = snapshot(job) + yield* Deferred.succeed(job.done, info).pipe(Effect.ignore) + return info + }) + + const list: Interface["list"] = Effect.fn("BackgroundJob.list")(function* () { + const s = yield* InstanceState.get(state) + return Array.from(s.jobs.values()) + .map(snapshot) + .toSorted((a, b) => a.started_at - b.started_at) + }) + + const get: Interface["get"] = Effect.fn("BackgroundJob.get")(function* (id) { + const s = yield* InstanceState.get(state) + const job = s.jobs.get(id) + if (!job) return + return snapshot(job) + }) + + const start: Interface["start"] = Effect.fn("BackgroundJob.start")(function* (input) { + const s = yield* InstanceState.get(state) + const id = input.id ?? Identifier.ascending("job") + const existing = s.jobs.get(id) + if (existing?.info.status === "running") return snapshot(existing) + + const job: Active = { + info: { + id, + type: input.type, + title: input.title, + status: "running", + started_at: Date.now(), + metadata: input.metadata, + }, + done: yield* Deferred.make(), + } + s.jobs.set(id, job) + job.fiber = yield* input.run.pipe( + Effect.matchCauseEffect({ + onSuccess: (output) => finish(job, "completed", { output }), + onFailure: (cause) => + finish(job, Cause.hasInterruptsOnly(cause) ? "cancelled" : "error", { + error: errorText(Cause.squash(cause)), + }), + }), + Effect.asVoid, + Effect.forkIn(s.scope), + ) + return snapshot(job) + }) + + const wait: Interface["wait"] = Effect.fn("BackgroundJob.wait")(function* (input) { + const s = yield* InstanceState.get(state) + const job = s.jobs.get(input.id) + if (!job) return { timedOut: false } + if (job.info.status !== "running") return { info: snapshot(job), timedOut: false } + if (!input.timeout) return { info: yield* Deferred.await(job.done), timedOut: false } + return yield* Effect.raceAll([ + Deferred.await(job.done).pipe(Effect.map((info) => ({ info, timedOut: false }))), + Effect.sleep(input.timeout).pipe(Effect.as({ info: snapshot(job), timedOut: true })), + ]) + }) + + const cancel: Interface["cancel"] = Effect.fn("BackgroundJob.cancel")(function* (id) { + const s = yield* InstanceState.get(state) + const job = s.jobs.get(id) + if (!job) return + if (job.info.status !== "running") return snapshot(job) + const fiber = job.fiber + const info = yield* finish(job, "cancelled") + if (fiber) yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + return info + }) + + return Service.of({ list, get, start, wait, cancel }) + }), +) + +export const defaultLayer = layer + +export * as BackgroundJob from "./job" diff --git a/packages/opencode/src/id/id.ts b/packages/opencode/src/id/id.ts index 46c210fa5d..737c9a3446 100644 --- a/packages/opencode/src/id/id.ts +++ b/packages/opencode/src/id/id.ts @@ -2,6 +2,7 @@ import z from "zod" import { randomBytes } from "crypto" const prefixes = { + job: "job", event: "evt", session: "ses", message: "msg", diff --git a/packages/opencode/test/background/job.test.ts b/packages/opencode/test/background/job.test.ts new file mode 100644 index 0000000000..6601042295 --- /dev/null +++ b/packages/opencode/test/background/job.test.ts @@ -0,0 +1,49 @@ +import { describe, expect } from "bun:test" +import { Deferred, Effect, Layer } from "effect" +import { BackgroundJob } from "@/background/job" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { provideTmpdirInstance } from "../fixture/fixture" +import { testEffect } from "../lib/effect" + +const it = testEffect(Layer.mergeAll(BackgroundJob.defaultLayer, CrossSpawnSpawner.defaultLayer)) + +describe("background.job", () => { + it.live("tracks started jobs through completion", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const jobs = yield* BackgroundJob.Service + const latch = yield* Deferred.make() + const job = yield* jobs.start({ + type: "test", + title: "test job", + run: Deferred.await(latch).pipe(Effect.as("done")), + }) + + expect(job.status).toBe("running") + yield* Deferred.succeed(latch, undefined) + const done = yield* jobs.wait({ id: job.id }) + + expect(done.info?.status).toBe("completed") + expect(done.info?.output).toBe("done") + expect((yield* jobs.list()).map((item) => item.id)).toEqual([job.id]) + }), + ), + ) + + it.live("can cancel running jobs", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + const jobs = yield* BackgroundJob.Service + const latch = yield* Deferred.make() + const job = yield* jobs.start({ + type: "test", + run: Deferred.await(latch).pipe(Effect.as("done")), + }) + + const cancelled = yield* jobs.cancel(job.id) + + expect(cancelled?.status).toBe("cancelled") + }), + ), + ) +}) From 2f919b8bc73bc2bfd8424854d7740196a8340d37 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 1 May 2026 19:05:06 +0530 Subject: [PATCH 08/11] refactor(task): use background jobs --- packages/opencode/src/effect/app-runtime.ts | 2 + packages/opencode/src/session/prompt.ts | 3 -- packages/opencode/src/tool/registry.ts | 3 ++ packages/opencode/src/tool/task.ts | 30 +++++++++---- packages/opencode/src/tool/task_status.ts | 45 +++++++++++++++++++ packages/opencode/test/session/prompt.test.ts | 2 + .../test/session/snapshot-tool-race.test.ts | 2 + packages/opencode/test/tool/task.test.ts | 33 +++++++------- .../opencode/test/tool/task_status.test.ts | 2 + 9 files changed, 94 insertions(+), 28 deletions(-) diff --git a/packages/opencode/src/effect/app-runtime.ts b/packages/opencode/src/effect/app-runtime.ts index 06969ff9d1..a69b2e00ad 100644 --- a/packages/opencode/src/effect/app-runtime.ts +++ b/packages/opencode/src/effect/app-runtime.ts @@ -50,6 +50,7 @@ import { SessionShare } from "@/share/session" import { SyncEvent } from "@/sync" import { Npm } from "@opencode-ai/core/npm" import { memoMap } from "@opencode-ai/core/effect/memo-map" +import { BackgroundJob } from "@/background/job" export const AppLayer = Layer.mergeAll( Npm.defaultLayer, @@ -75,6 +76,7 @@ export const AppLayer = Layer.mergeAll( Todo.defaultLayer, Session.defaultLayer, SessionStatus.defaultLayer, + BackgroundJob.defaultLayer, SessionRunState.defaultLayer, SessionProcessor.defaultLayer, SessionCompaction.defaultLayer, diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 8fafc10761..e57d867e09 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -118,9 +118,6 @@ export const layer = Layer.effect( resolvePromptParts: (template: string) => resolvePromptParts(template), prompt: (input: PromptInput) => prompt(input), loop: (input: LoopInput) => loop(input), - fork: (effect: Effect.Effect) => { - run.fork(effect) - }, } satisfies TaskPromptOps }) diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index 148c9de4e6..fd57101d1b 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -48,6 +48,7 @@ import { Agent } from "../agent/agent" import { Skill } from "../skill" import { Permission } from "@/permission" import { SessionStatus } from "@/session/status" +import { BackgroundJob } from "@/background/job" const log = Log.create({ service: "tool.registry" }) @@ -86,6 +87,7 @@ export const layer: Layer.Layer< | Instruction.Service | AppFileSystem.Service | Bus.Service + | BackgroundJob.Service | HttpClient.HttpClient | ChildProcessSpawner | Ripgrep.Service @@ -343,6 +345,7 @@ export const defaultLayer = Layer.suspend(() => Layer.provide(Instruction.defaultLayer), Layer.provide(AppFileSystem.defaultLayer), Layer.provide(Bus.layer), + Layer.provide(BackgroundJob.defaultLayer), Layer.provide(FetchHttpClient.layer), Layer.provide(Format.defaultLayer), Layer.provide(CrossSpawnSpawner.defaultLayer), diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 061321b6b0..8bd01f88af 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -10,13 +10,13 @@ import { SessionStatus } from "@/session/status" import { TuiEvent } from "@/cli/cmd/tui/event" import { Cause, Effect, Option, Schema } from "effect" import { Config } from "@/config/config" +import { BackgroundJob } from "@/background/job" export interface TaskPromptOps { cancel(sessionID: SessionID): void resolvePromptParts(template: string): Effect.Effect prompt(input: SessionPrompt.PromptInput): Effect.Effect loop(input: SessionPrompt.LoopInput): Effect.Effect - fork(effect: Effect.Effect): void } const id = "task" @@ -80,6 +80,7 @@ export const TaskTool = Tool.define( const config = yield* Config.Service const sessions = yield* Session.Service const status = yield* SessionStatus.Service + const jobs = yield* BackgroundJob.Service const run = Effect.fn( "TaskTool.execute", @@ -236,17 +237,28 @@ export const TaskTool = Tool.define( yield* continueIfIdle({ userID: message.info.id, state }) }) - ops.fork( - runTask().pipe( + yield* jobs.start({ + id: nextSession.id, + type: id, + title: params.description, + metadata: { + parentSessionID: ctx.sessionID, + sessionID: nextSession.id, + subagent: next.name, + }, + run: runTask().pipe( Effect.matchCauseEffect({ - onSuccess: (text) => inject("completed", text), - onFailure: (cause) => - inject("error", errorText(Cause.squash(cause))).pipe(Effect.catchCause(() => Effect.void)), + onSuccess: (text) => inject("completed", text).pipe(Effect.as(text)), + onFailure: (cause) => { + const text = errorText(Cause.squash(cause)) + return inject("error", text).pipe( + Effect.catchCause(() => Effect.void), + Effect.andThen(Effect.failCause(cause)), + ) + }, }), - Effect.catchCause(() => Effect.void), - Effect.asVoid, ), - ) + }) return { title: params.description, diff --git a/packages/opencode/src/tool/task_status.ts b/packages/opencode/src/tool/task_status.ts index f49994d10c..90240d8220 100644 --- a/packages/opencode/src/tool/task_status.ts +++ b/packages/opencode/src/tool/task_status.ts @@ -6,6 +6,7 @@ import { MessageV2 } from "@/session/message-v2" import { SessionStatus } from "@/session/status" import { PositiveInt } from "@/util/schema" import { Effect, Option, Schema } from "effect" +import { BackgroundJob } from "@/background/job" const DEFAULT_TIMEOUT = 60_000 const POLL_MS = 300 @@ -34,11 +35,31 @@ function errorText(error: NonNullable) { return error.name } +function jobResult(job: BackgroundJob.Info): InspectResult { + if (job.status === "running") { + return { + state: "running", + text: "Task is still running.", + } + } + if (job.status === "completed") { + return { + state: "completed", + text: job.output ?? "", + } + } + return { + state: "error", + text: job.error ?? `Task ${job.status}.`, + } +} + export const TaskStatusTool = Tool.define( "task_status", Effect.gen(function* () { const sessions = yield* Session.Service const status = yield* SessionStatus.Service + const jobs = yield* BackgroundJob.Service const inspect: (taskID: SessionID) => Effect.Effect = Effect.fn("TaskStatusTool.inspect")(function* ( taskID: SessionID, @@ -119,6 +140,30 @@ export const TaskStatusTool = Tool.define( )(function* (params: Schema.Schema.Type, _ctx: Tool.Context) { yield* sessions.get(params.task_id) + const job = yield* jobs.get(params.task_id) + const waitedJob = + job && params.wait === true + ? yield* jobs.wait({ id: params.task_id, timeout: params.timeout_ms ?? DEFAULT_TIMEOUT }) + : { info: job, timedOut: false } + if (waitedJob.info) { + const result = jobResult(waitedJob.info) + return { + title: "Task status", + metadata: { + task_id: params.task_id, + state: result.state, + timed_out: waitedJob.timedOut, + }, + output: format({ + taskID: params.task_id, + state: result.state, + text: waitedJob.timedOut + ? `Timed out after ${params.timeout_ms ?? DEFAULT_TIMEOUT}ms while waiting for task completion.` + : result.text, + }), + } + } + const waited = params.wait === true ? yield* waitForTerminal(params.task_id, params.timeout_ms ?? DEFAULT_TIMEOUT) diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index 5330569401..221357dc84 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -41,6 +41,7 @@ import * as Log from "@opencode-ai/core/util/log" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { Ripgrep } from "../../src/file/ripgrep" import { Format } from "../../src/format" +import { BackgroundJob } from "@/background/job" import { provideTmpdirInstance, provideTmpdirServer } from "../fixture/fixture" import { testEffect } from "../lib/effect" import { reply, TestLLMServer } from "../lib/llm-server" @@ -177,6 +178,7 @@ function makeHttp() { Layer.provide(CrossSpawnSpawner.defaultLayer), Layer.provide(Ripgrep.defaultLayer), Layer.provide(Format.defaultLayer), + Layer.provide(BackgroundJob.defaultLayer), Layer.provideMerge(todo), Layer.provideMerge(question), Layer.provideMerge(deps), diff --git a/packages/opencode/test/session/snapshot-tool-race.test.ts b/packages/opencode/test/session/snapshot-tool-race.test.ts index ab5a3ab7ed..20c2925bae 100644 --- a/packages/opencode/test/session/snapshot-tool-race.test.ts +++ b/packages/opencode/test/session/snapshot-tool-race.test.ts @@ -55,6 +55,7 @@ import { AppFileSystem } from "@opencode-ai/core/filesystem" import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" import { Ripgrep } from "../../src/file/ripgrep" import { Format } from "../../src/format" +import { BackgroundJob } from "@/background/job" void Log.init({ print: false }) @@ -130,6 +131,7 @@ function makeHttp() { Layer.provide(CrossSpawnSpawner.defaultLayer), Layer.provide(Ripgrep.defaultLayer), Layer.provide(Format.defaultLayer), + Layer.provide(BackgroundJob.defaultLayer), Layer.provideMerge(todo), Layer.provideMerge(question), Layer.provideMerge(deps), diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index 7f80da1dd5..6267d8b506 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect } from "bun:test" -import { Effect, Layer } from "effect" +import { Deferred, Effect, Layer } from "effect" import { Agent } from "../../src/agent/agent" import { Bus } from "../../src/bus" import { Config } from "@/config/config" @@ -14,6 +14,7 @@ import { ModelID, ProviderID } from "../../src/provider/schema" import { TaskTool, type TaskPromptOps } from "../../src/tool/task" import { Truncate } from "@/tool/truncate" import { ToolRegistry } from "@/tool/registry" +import { BackgroundJob } from "@/background/job" import { provideTmpdirInstance } from "../fixture/fixture" import { testEffect } from "../lib/effect" @@ -34,6 +35,7 @@ const it = testEffect( CrossSpawnSpawner.defaultLayer, Session.defaultLayer, SessionStatus.defaultLayer, + BackgroundJob.defaultLayer, Truncate.defaultLayer, ToolRegistry.defaultLayer, ), @@ -68,13 +70,17 @@ const seed = Effect.fn("TaskToolTest.seed")(function* (title = "Pinned") { return { chat, assistant } }) -function stubOps(session: Session.Interface, opts?: { onPrompt?: (input: SessionPrompt.PromptInput) => void; text?: string }): TaskPromptOps { +function stubOps( + session: Session.Interface, + opts?: { onPrompt?: (input: SessionPrompt.PromptInput) => void; text?: string; wait?: Effect.Effect }, +): TaskPromptOps { return { cancel() {}, resolvePromptParts: (template) => Effect.succeed([{ type: "text" as const, text: template }]), prompt: (input) => Effect.gen(function* () { opts?.onPrompt?.(input) + if (opts?.wait) yield* opts.wait const userID = input.messageID ?? MessageID.ascending() const user: MessageV2.User = { id: userID, @@ -120,7 +126,6 @@ function stubOps(session: Session.Interface, opts?: { onPrompt?: (input: Session opts?.text ?? "done", ), ), - fork() {}, } } @@ -438,10 +443,11 @@ describe("tool.task", () => { provideTmpdirInstance(() => Effect.gen(function* () { const sessions = yield* Session.Service + const jobs = yield* BackgroundJob.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() - const forks: Effect.Effect[] = [] + const latch = yield* Deferred.make() const result = yield* def.execute( { @@ -456,12 +462,7 @@ describe("tool.task", () => { agent: "build", abort: new AbortController().signal, extra: { - promptOps: { - ...stubOps(sessions), - fork(effect) { - forks.push(effect) - }, - } satisfies TaskPromptOps, + promptOps: stubOps(sessions, { wait: Deferred.await(latch) }), }, messages: [], metadata: () => Effect.void, @@ -473,7 +474,10 @@ describe("tool.task", () => { expect(result.metadata.background).toBe(true) expect(result.output).toContain(`task_id: ${result.metadata.sessionId}`) expect(result.output).toContain("state: running") - expect(forks).toHaveLength(1) + expect((yield* jobs.get(result.metadata.sessionId))?.status).toBe("running") + + yield* Deferred.succeed(latch, undefined) + expect((yield* jobs.wait({ id: result.metadata.sessionId })).info?.status).toBe("completed") }), ), ) @@ -482,10 +486,10 @@ describe("tool.task", () => { provideTmpdirInstance(() => Effect.gen(function* () { const sessions = yield* Session.Service + const jobs = yield* BackgroundJob.Service const { chat, assistant } = yield* seed() const tool = yield* TaskTool const def = yield* tool.init() - const forks: Effect.Effect[] = [] const loops: string[] = [] const result = yield* def.execute( @@ -518,9 +522,6 @@ describe("tool.task", () => { ), ) }, - fork(effect) { - forks.push(effect) - }, } satisfies TaskPromptOps, }, messages: [], @@ -529,7 +530,7 @@ describe("tool.task", () => { }, ) - yield* forks[0]! + expect((yield* jobs.wait({ id: result.metadata.sessionId })).info?.status).toBe("completed") const parent = yield* sessions.findMessage(chat.id, (msg) => msg.info.role === "user") expect(parent._tag).toBe("Some") diff --git a/packages/opencode/test/tool/task_status.test.ts b/packages/opencode/test/tool/task_status.test.ts index 89bf49622d..cffe35cf18 100644 --- a/packages/opencode/test/tool/task_status.test.ts +++ b/packages/opencode/test/tool/task_status.test.ts @@ -10,6 +10,7 @@ import { SessionStatus } from "../../src/session/status" import { TaskStatusTool } from "../../src/tool/task_status" import { Truncate } from "@/tool/truncate" import { ModelID, ProviderID } from "../../src/provider/schema" +import { BackgroundJob } from "@/background/job" import { provideTmpdirInstance } from "../fixture/fixture" import { testEffect } from "../lib/effect" @@ -28,6 +29,7 @@ const it = testEffect( CrossSpawnSpawner.defaultLayer, Session.defaultLayer, SessionStatus.defaultLayer, + BackgroundJob.defaultLayer, Truncate.defaultLayer, ), ) From a84edc224ff1381b20272721d0f1e9397aadc6cd Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Fri, 1 May 2026 19:05:16 +0530 Subject: [PATCH 09/11] fix(tui): show background task progress --- .../src/cli/cmd/tui/routes/session/index.tsx | 32 +++++++++++++------ 1 file changed, 22 insertions(+), 10 deletions(-) diff --git a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx index 4878c5af80..c96fdf268e 100644 --- a/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx +++ b/packages/opencode/src/cli/cmd/tui/routes/session/index.tsx @@ -1960,12 +1960,15 @@ function Task(props: ToolProps) { const { navigate } = useRoute() const sync = useSync() - onMount(() => { - if (props.metadata.sessionId && !sync.data.message[props.metadata.sessionId]?.length) - void sync.session.sync(props.metadata.sessionId) + createEffect(() => { + const sessionID = props.metadata.sessionId + if (!sessionID) return + if (sync.data.message[sessionID]?.length) return + void sync.session.sync(sessionID) }) - const messages = createMemo(() => sync.data.message[props.metadata.sessionId ?? ""] ?? []) + const childSessionID = createMemo(() => props.metadata.sessionId) + const messages = createMemo(() => sync.data.message[childSessionID() ?? ""] ?? []) const tools = createMemo(() => { return messages().flatMap((msg) => @@ -1979,7 +1982,16 @@ function Task(props: ToolProps) { tools().findLast((x) => (x.state.status === "running" || x.state.status === "completed") && x.state.title), ) - const isRunning = createMemo(() => props.part.state.status === "running") + const isBackground = createMemo(() => props.metadata.background === true) + const isBackgroundRunning = createMemo(() => { + const sessionID = childSessionID() + if (!isBackground() || !sessionID) return false + const status = sync.data.session_status[sessionID]?.type + if (status === "busy" || status === "retry") return true + if (status === "idle") return false + return !messages().some((x) => x.role === "assistant" && x.time.completed) + }) + const isRunning = createMemo(() => props.part.state.status === "running" || isBackgroundRunning()) const duration = createMemo(() => { const first = messages().find((x) => x.role === "user")?.time.created @@ -1990,8 +2002,7 @@ function Task(props: ToolProps) { const content = createMemo(() => { if (!props.input.description) return "" - const description = - props.metadata.background === true ? `${props.input.description} (background)` : props.input.description + const description = isBackground() ? `${props.input.description} (background)` : props.input.description let content = [`${Locale.titlecase(props.input.subagent_type ?? "General")} Task — ${description}`] if (isRunning() && tools().length > 0) { @@ -2003,7 +2014,7 @@ function Task(props: ToolProps) { } else content.push(`↳ ${tools().length} toolcalls`) } - if (props.part.state.status === "completed") { + if (!isRunning() && props.part.state.status === "completed") { content.push(`└ ${tools().length} toolcalls · ${Locale.duration(duration())}`) } @@ -2018,8 +2029,9 @@ function Task(props: ToolProps) { pending="Delegating..." part={props.part} onClick={() => { - if (props.metadata.sessionId) { - navigate({ type: "session", sessionID: props.metadata.sessionId }) + const sessionID = childSessionID() + if (sessionID) { + navigate({ type: "session", sessionID }) } }} > From 227b8c668b44633739e1180971bb09c01e1e5323 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Sat, 2 May 2026 14:06:57 +0530 Subject: [PATCH 10/11] feat(task): gate background tasks experimentally --- packages/opencode/src/tool/registry.ts | 2 +- packages/opencode/src/tool/task.ts | 4 ++++ packages/opencode/test/tool/task.test.ts | 6 ++++++ 3 files changed, 11 insertions(+), 1 deletion(-) diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index fd57101d1b..691d6ef514 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -222,7 +222,7 @@ export const layer: Layer.Layer< tool.edit, tool.write, tool.task, - tool.taskstatus, + ...(Flag.OPENCODE_EXPERIMENTAL ? [tool.taskstatus] : []), tool.fetch, tool.todo, tool.search, diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 8bd01f88af..43c02107dd 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -11,6 +11,7 @@ import { TuiEvent } from "@/cli/cmd/tui/event" import { Cause, Effect, Option, Schema } from "effect" import { Config } from "@/config/config" import { BackgroundJob } from "@/background/job" +import { Flag } from "@opencode-ai/core/flag/flag" export interface TaskPromptOps { cancel(sessionID: SessionID): void @@ -159,6 +160,9 @@ export const TaskTool = Tool.define( providerID: msg.info.providerID, } const background = params.background === true + if (background && !Flag.OPENCODE_EXPERIMENTAL) { + return yield* Effect.fail(new Error("Background tasks require OPENCODE_EXPERIMENTAL=true")) + } const metadata = { sessionId: nextSession.id, diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index 6267d8b506..63ccc78bd1 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -15,10 +15,14 @@ import { TaskTool, type TaskPromptOps } from "../../src/tool/task" import { Truncate } from "@/tool/truncate" import { ToolRegistry } from "@/tool/registry" import { BackgroundJob } from "@/background/job" +import { Flag } from "@opencode-ai/core/flag/flag" import { provideTmpdirInstance } from "../fixture/fixture" import { testEffect } from "../lib/effect" +const originalExperimental = Flag.OPENCODE_EXPERIMENTAL + afterEach(async () => { + Flag.OPENCODE_EXPERIMENTAL = originalExperimental await Instance.disposeAll() }) @@ -442,6 +446,7 @@ describe("tool.task", () => { it.live("execute launches background tasks without waiting for completion", () => provideTmpdirInstance(() => Effect.gen(function* () { + Flag.OPENCODE_EXPERIMENTAL = true const sessions = yield* Session.Service const jobs = yield* BackgroundJob.Service const { chat, assistant } = yield* seed() @@ -485,6 +490,7 @@ describe("tool.task", () => { it.live("background tasks inject completion into the parent session and resume when idle", () => provideTmpdirInstance(() => Effect.gen(function* () { + Flag.OPENCODE_EXPERIMENTAL = true const sessions = yield* Session.Service const jobs = yield* BackgroundJob.Service const { chat, assistant } = yield* seed() From 45360e5e0bba73b40b4c982df778d60902cf4576 Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Mon, 4 May 2026 20:42:32 +0530 Subject: [PATCH 11/11] fix(task): handle running background resumes --- packages/opencode/src/tool/task.ts | 24 ++- packages/opencode/test/tool/task.test.ts | 183 +++++++++++++++++++++-- 2 files changed, 190 insertions(+), 17 deletions(-) diff --git a/packages/opencode/src/tool/task.ts b/packages/opencode/src/tool/task.ts index 43c02107dd..17eaa5e86e 100644 --- a/packages/opencode/src/tool/task.ts +++ b/packages/opencode/src/tool/task.ts @@ -8,7 +8,7 @@ import { Agent } from "../agent/agent" import type { SessionPrompt } from "../session/prompt" import { SessionStatus } from "@/session/status" import { TuiEvent } from "@/cli/cmd/tui/event" -import { Cause, Effect, Option, Schema } from "effect" +import { Cause, Effect, Option, Schema, Scope, Stream } from "effect" import { Config } from "@/config/config" import { BackgroundJob } from "@/background/job" import { Flag } from "@opencode-ai/core/flag/flag" @@ -82,6 +82,7 @@ export const TaskTool = Tool.define( const sessions = yield* Session.Service const status = yield* SessionStatus.Service const jobs = yield* BackgroundJob.Service + const scope = yield* Scope.Scope const run = Effect.fn( "TaskTool.execute", @@ -163,6 +164,9 @@ export const TaskTool = Tool.define( if (background && !Flag.OPENCODE_EXPERIMENTAL) { return yield* Effect.fail(new Error("Background tasks require OPENCODE_EXPERIMENTAL=true")) } + if ((yield* jobs.get(nextSession.id))?.status === "running") { + return yield* Effect.fail(new Error(`Task ${nextSession.id} is already running`)) + } const metadata = { sessionId: nextSession.id, @@ -198,11 +202,21 @@ export const TaskTool = Tool.define( return result.parts.findLast((item) => item.type === "text")?.text ?? "" }) - const continueIfIdle = Effect.fn("TaskTool.continueIfIdle")(function* (input: { + const resumeParent: (input: { userID: MessageID state: "completed" | "error" - }) { - if ((yield* status.get(ctx.sessionID)).type !== "idle") return + attempts?: number + }) => Effect.Effect = Effect.fn("TaskTool.resumeParent")(function* (input) { + if ((yield* status.get(ctx.sessionID)).type !== "idle") { + if ((input.attempts ?? 0) >= 60) return + yield* bus.subscribe(SessionStatus.Event.Idle).pipe( + Stream.filter((event) => event.properties.sessionID === ctx.sessionID), + Stream.take(1), + Stream.runDrain, + Effect.timeoutOption("1 second"), + ) + return yield* resumeParent({ ...input, attempts: (input.attempts ?? 0) + 1 }) + } const latest = yield* sessions.findMessage(ctx.sessionID, (item) => item.info.role === "user") if (Option.isNone(latest)) return if (latest.value.info.id !== input.userID) return @@ -238,7 +252,7 @@ export const TaskTool = Tool.define( }, ], }) - yield* continueIfIdle({ userID: message.info.id, state }) + yield* resumeParent({ userID: message.info.id, state }).pipe(Effect.ignore, Effect.forkIn(scope)) }) yield* jobs.start({ diff --git a/packages/opencode/test/tool/task.test.ts b/packages/opencode/test/tool/task.test.ts index 63ccc78bd1..5209ea2e6c 100644 --- a/packages/opencode/test/tool/task.test.ts +++ b/packages/opencode/test/tool/task.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect } from "bun:test" -import { Deferred, Effect, Layer } from "effect" +import { Cause, Deferred, Effect, Exit, Layer } from "effect" import { Agent } from "../../src/agent/agent" import { Bus } from "../../src/bus" import { Config } from "@/config/config" @@ -497,6 +497,82 @@ describe("tool.task", () => { const tool = yield* TaskTool const def = yield* tool.init() const loops: string[] = [] + const resumed = yield* Deferred.make() + + const result = yield* def.execute( + { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { + promptOps: { + ...stubOps(sessions, { text: "background done" }), + loop(input) { + loops.push(input.sessionID) + return Deferred.succeed(resumed, undefined).pipe( + Effect.andThen( + Effect.sync(() => + reply( + { + sessionID: input.sessionID, + messageID: MessageID.ascending(), + agent: "build", + model: ref, + parts: [], + }, + "looped", + ), + ), + ), + ) + }, + } satisfies TaskPromptOps, + }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + expect((yield* jobs.wait({ id: result.metadata.sessionId })).info?.status).toBe("completed") + yield* Deferred.await(resumed).pipe(Effect.timeout("1 second")) + + const parent = yield* sessions.findMessage(chat.id, (msg) => msg.info.role === "user") + expect(parent._tag).toBe("Some") + if (parent._tag !== "Some") return + expect(parent.value.parts.find((part) => part.type === "text")?.text).toContain("Background task completed") + expect(parent.value.parts.find((part) => part.type === "text")?.text).toContain("background done") + expect(loops).toEqual([chat.id]) + + const child = yield* sessions.findMessage(result.metadata.sessionId, (msg) => msg.info.role === "assistant") + expect(child._tag).toBe("Some") + if (child._tag !== "Some") return + expect(child.value.parts.find((part) => part.type === "text")?.text).toBe("background done") + }), + ), + ) + + it.live("background task resumes parent after it becomes idle", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + Flag.OPENCODE_EXPERIMENTAL = true + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const jobs = yield* BackgroundJob.Service + const { chat, assistant } = yield* seed() + const tool = yield* TaskTool + const def = yield* tool.init() + const loops: string[] = [] + const resumed = yield* Deferred.make() + + yield* status.set(chat.id, { type: "busy" }) const result = yield* def.execute( { @@ -526,7 +602,7 @@ describe("tool.task", () => { }, "looped", ), - ) + ).pipe(Effect.tap(() => Deferred.succeed(resumed, undefined))) }, } satisfies TaskPromptOps, }, @@ -537,18 +613,101 @@ describe("tool.task", () => { ) expect((yield* jobs.wait({ id: result.metadata.sessionId })).info?.status).toBe("completed") - - const parent = yield* sessions.findMessage(chat.id, (msg) => msg.info.role === "user") - expect(parent._tag).toBe("Some") - if (parent._tag !== "Some") return - expect(parent.value.parts.find((part) => part.type === "text")?.text).toContain("Background task completed") - expect(parent.value.parts.find((part) => part.type === "text")?.text).toContain("background done") + expect(loops).toEqual([]) + yield* status.set(chat.id, { type: "idle" }) + yield* Deferred.await(resumed).pipe(Effect.timeout("1 second")) expect(loops).toEqual([chat.id]) + }), + ), + ) - const child = yield* sessions.findMessage(result.metadata.sessionId, (msg) => msg.info.role === "assistant") - expect(child._tag).toBe("Some") - if (child._tag !== "Some") return - expect(child.value.parts.find((part) => part.type === "text")?.text).toBe("background done") + it.live("background resume fails while task is already running", () => + provideTmpdirInstance(() => + Effect.gen(function* () { + Flag.OPENCODE_EXPERIMENTAL = true + const sessions = yield* Session.Service + const { chat, assistant } = yield* seed() + const tool = yield* TaskTool + const def = yield* tool.init() + const latch = yield* Deferred.make() + + const result = yield* def.execute( + { + description: "inspect bug", + prompt: "look into the cache key path", + subagent_type: "general", + background: true, + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { + promptOps: stubOps(sessions, { wait: Deferred.await(latch) }), + }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + + const exit = yield* def + .execute( + { + description: "inspect bug again", + prompt: "second prompt", + subagent_type: "general", + task_id: result.metadata.sessionId, + background: true, + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { promptOps: stubOps(sessions) }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + .pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause) + expect(error instanceof Error ? error.message : String(error)).toContain("is already running") + } + + const foregroundExit = yield* def + .execute( + { + description: "inspect bug again", + prompt: "second prompt", + subagent_type: "general", + task_id: result.metadata.sessionId, + }, + { + sessionID: chat.id, + messageID: assistant.id, + agent: "build", + abort: new AbortController().signal, + extra: { promptOps: stubOps(sessions) }, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ) + .pipe(Effect.exit) + + expect(Exit.isFailure(foregroundExit)).toBe(true) + if (Exit.isFailure(foregroundExit)) { + const error = Cause.squash(foregroundExit.cause) + expect(error instanceof Error ? error.message : String(error)).toContain("is already running") + } + + yield* Deferred.succeed(latch, undefined) }), ), )