Compare commits

...

13 Commits

Author SHA1 Message Date
Shoubhit Dash 45360e5e0b fix(task): handle running background resumes 2026-05-04 20:42:32 +05:30
Shoubhit Dash 227b8c668b feat(task): gate background tasks experimentally 2026-05-02 14:06:57 +05:30
Shoubhit Dash a84edc224f fix(tui): show background task progress 2026-05-01 19:05:16 +05:30
Shoubhit Dash 2f919b8bc7 refactor(task): use background jobs 2026-05-01 19:05:06 +05:30
Shoubhit Dash 085fac7c2c feat(background): add job service 2026-05-01 19:04:56 +05:30
Shoubhit Dash b26fe0d357 Merge branch 'dev' into nxl/background-subagents 2026-05-01 15:29:32 +05:30
Shoubhit Dash 80aeb78b38 Merge branch 'dev' into nxl/background-subagents 2026-04-24 20:32:04 +05:30
Shoubhit Dash 601fe03a3a refactor(task): simplify effect wrappers 2026-04-24 20:30:53 +05:30
Shoubhit Dash 3f4b9d9ef4 test(task): use branded session id in schema test 2026-04-24 20:21:02 +05:30
Shoubhit Dash 1357bb984f style: fix background task formatting 2026-04-24 20:20:04 +05:30
Shoubhit Dash ecde8ab363 test(task): update parameter schema snapshot 2026-04-24 20:20:04 +05:30
Shoubhit Dash 7970130720 fix(ui): label background task cards 2026-04-24 20:08:32 +05:30
Shoubhit Dash 971c837ad4 feat(task): add background subagent support 2026-04-24 20:08:24 +05:30
18 changed files with 1302 additions and 70 deletions
+173
View File
@@ -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<string, unknown>
}
type Active = {
info: Info
done: Deferred.Deferred<Info>
fiber?: Fiber.Fiber<void, unknown>
}
type State = {
jobs: Map<string, Active>
scope: Scope.Scope
}
export type StartInput = {
id?: string
type: string
title?: string
metadata?: Record<string, unknown>
run: Effect.Effect<string, unknown>
}
export type WaitInput = {
id: string
timeout?: number
}
export type WaitResult = {
info?: Info
timedOut: boolean
}
export interface Interface {
readonly list: () => Effect.Effect<Info[]>
readonly get: (id: string) => Effect.Effect<Info | undefined>
readonly start: (input: StartInput) => Effect.Effect<Info>
readonly wait: (input: WaitInput) => Effect.Effect<WaitResult>
readonly cancel: (id: string) => Effect.Effect<Info | undefined>
}
export class Service extends Context.Service<Service, Interface>()("@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<State>(
Effect.fn("BackgroundJob.state")(function* () {
return {
jobs: new Map(),
scope: yield* Scope.Scope,
}
}),
)
const finish = Effect.fn("BackgroundJob.finish")(function* (
job: Active,
status: Exclude<Status, "running">,
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<Info>(),
}
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"
@@ -1960,12 +1960,15 @@ function Task(props: ToolProps<typeof TaskTool>) {
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<typeof TaskTool>) {
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,7 +2002,8 @@ function Task(props: ToolProps<typeof TaskTool>) {
const content = createMemo(() => {
if (!props.input.description) return ""
let content = [`${Locale.titlecase(props.input.subagent_type ?? "General")} Task — ${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) {
// content[0] += ` · ${tools().length} toolcalls`
@@ -2001,7 +2014,7 @@ function Task(props: ToolProps<typeof TaskTool>) {
} 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())}`)
}
@@ -2016,8 +2029,9 @@ function Task(props: ToolProps<typeof TaskTool>) {
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 })
}
}}
>
@@ -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,
+1
View File
@@ -2,6 +2,7 @@ import z from "zod"
import { randomBytes } from "crypto"
const prefixes = {
job: "job",
event: "evt",
session: "ses",
message: "msg",
+1
View File
@@ -117,6 +117,7 @@ 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),
} satisfies TaskPromptOps
})
+10
View File
@@ -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"
@@ -46,6 +47,8 @@ import { Bus } from "../bus"
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" })
@@ -78,11 +81,13 @@ export const layer: Layer.Layer<
| Agent.Service
| Skill.Service
| Session.Service
| SessionStatus.Service
| Provider.Service
| LSP.Service
| Instruction.Service
| AppFileSystem.Service
| Bus.Service
| BackgroundJob.Service
| HttpClient.HttpClient
| ChildProcessSpawner
| Ripgrep.Service
@@ -113,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<State>(
@@ -193,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),
@@ -215,6 +222,7 @@ export const layer: Layer.Layer<
tool.edit,
tool.write,
tool.task,
...(Flag.OPENCODE_EXPERIMENTAL ? [tool.taskstatus] : []),
tool.fetch,
tool.todo,
tool.search,
@@ -331,11 +339,13 @@ 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),
Layer.provide(AppFileSystem.defaultLayer),
Layer.provide(Bus.layer),
Layer.provide(BackgroundJob.defaultLayer),
Layer.provide(FetchHttpClient.layer),
Layer.provide(Format.defaultLayer),
Layer.provide(CrossSpawnSpawner.defaultLayer),
+181 -43
View File
@@ -1,17 +1,23 @@
import * as Tool from "./tool"
import DESCRIPTION from "./task.txt"
import { Bus } from "@/bus"
import { Session } from "@/session/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 { TuiEvent } from "@/cli/cmd/tui/event"
import { Cause, Effect, Option, Schema, Scope, Stream } from "effect"
import { Config } from "@/config/config"
import { Effect, Schema } from "effect"
import { BackgroundJob } from "@/background/job"
import { Flag } from "@opencode-ai/core/flag/flag"
export interface TaskPromptOps {
cancel(sessionID: SessionID): void
resolvePromptParts(template: string): Effect.Effect<SessionPrompt.PromptInput["parts"]>
prompt(input: SessionPrompt.PromptInput): Effect.Effect<MessageV2.WithParts>
loop(input: SessionPrompt.LoopInput): Effect.Effect<MessageV2.WithParts>
}
const id = "task"
@@ -20,24 +26,67 @@ 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)`,
"",
"<task_result>",
text,
"</task_result>",
].join("\n")
}
function backgroundOutput(sessionID: SessionID) {
return [
`task_id: ${sessionID} (for polling this task with task_status)`,
"state: running",
"",
"<task_result>",
"Background task started. Continue your current work and call task_status when you need the result.",
"</task_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, `</${tag}>`].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 jobs = yield* BackgroundJob.Service
const scope = yield* Scope.Scope
const run = Effect.fn("TaskTool.execute")(function* (
params: Schema.Schema.Type<typeof Parameters>,
ctx: Tool.Context,
) {
const run = Effect.fn(
"TaskTool.execute",
)(function* (params: Schema.Schema.Type<typeof Parameters>, ctx: Tool.Context) {
const cfg = yield* config.get()
if (!ctx.extra?.bypassAgentCheck) {
@@ -62,7 +111,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 parent = yield* sessions.get(ctx.sessionID)
const nextSession =
@@ -107,19 +156,134 @@ 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
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,
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 resumeParent: (input: {
userID: MessageID
state: "completed" | "error"
attempts?: number
}) => Effect.Effect<void> = 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
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* resumeParent({ userID: message.info.id, state }).pipe(Effect.ignore, Effect.forkIn(scope))
})
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).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)),
)
},
}),
),
})
return {
title: params.description,
metadata,
output: backgroundOutput(nextSession.id),
}
}
function cancel() {
ops.cancel(nextSession.id)
@@ -131,36 +295,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)`,
"",
"<task_result>",
result.parts.findLast((item) => item.type === "text")?.text ?? "",
"</task_result>",
].join("\n"),
metadata,
output: output(nextSession.id, text),
}
}),
() =>
@@ -168,13 +307,12 @@ export const TaskTool = Tool.define(
ctx.abort.removeEventListener("abort", cancel)
}),
)
})
}, Effect.orDie)
return {
description: DESCRIPTION,
parameters: Parameters,
execute: (params: Schema.Schema.Type<typeof Parameters>, ctx: Tool.Context) =>
run(params, ctx).pipe(Effect.orDie),
execute: run,
}
}),
)
+7 -5
View File
@@ -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):
+197
View File
@@ -0,0 +1,197 @@
import * as Tool from "./tool"
import DESCRIPTION from "./task_status.txt"
import { Session } from "@/session/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"
import { BackgroundJob } from "@/background/job"
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}`, "", "<task_result>", input.text, "</task_result>"].join(
"\n",
)
}
function errorText(error: NonNullable<MessageV2.Assistant["error"]>) {
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
}
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<InspectResult> = 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)
return yield* waitForTerminal(taskID, timeout - sleep)
})
const run = Effect.fn(
"TaskStatusTool.execute",
)(function* (params: Schema.Schema.Type<typeof Parameters>, _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)
: { 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,
}),
}
}, Effect.orDie)
return {
description: DESCRIPTION,
parameters: Parameters,
execute: run,
}
}),
)
@@ -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`)
- `<task_result>...</task_result>` containing final output, error summary, or current progress text
@@ -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<void>()
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<void>()
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")
}),
),
)
})
@@ -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),
@@ -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),
@@ -320,6 +320,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",
@@ -338,6 +342,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",
},
},
@@ -25,6 +25,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 = <S extends Schema.Decoder<unknown>>(schema: S, input: unknown): S["Type"] =>
Schema.decodeUnknownSync(schema)(input)
@@ -203,6 +204,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: SessionID.make("ses_test"),
command: "/cmd",
background: true,
})
expect(parsed.task_id).toBe(SessionID.make("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)
})
+337 -10
View File
@@ -1,21 +1,28 @@
import { afterEach, describe, expect } from "bun:test"
import { 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"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { Instance } from "../../src/project/instance"
import { Session } from "@/session/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 "@/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()
})
@@ -27,9 +34,12 @@ const ref = {
const it = testEffect(
Layer.mergeAll(
Agent.defaultLayer,
Bus.defaultLayer,
Config.defaultLayer,
CrossSpawnSpawner.defaultLayer,
Session.defaultLayer,
SessionStatus.defaultLayer,
BackgroundJob.defaultLayer,
Truncate.defaultLayer,
ToolRegistry.defaultLayer,
),
@@ -64,15 +74,62 @@ 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; wait?: Effect.Effect<void> },
): 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")
if (opts?.wait) yield* opts.wait
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",
),
),
}
}
@@ -195,7 +252,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 +286,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<string, any>) =>
def.execute(
@@ -282,14 +340,14 @@ 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",
task_id: SessionID.make("ses_missing"),
},
{
sessionID: chat.id,
@@ -322,7 +380,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 +442,273 @@ 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()
const tool = yield* TaskTool
const def = yield* tool.init()
const latch = yield* Deferred.make<void>()
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,
},
)
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((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")
}),
),
)
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()
const tool = yield* TaskTool
const def = yield* tool.init()
const loops: string[] = []
const resumed = yield* Deferred.make<void>()
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<void>()
yield* status.set(chat.id, { type: "busy" })
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",
),
).pipe(Effect.tap(() => Deferred.succeed(resumed, undefined)))
},
} satisfies TaskPromptOps,
},
messages: [],
metadata: () => Effect.void,
ask: () => Effect.void,
},
)
expect((yield* jobs.wait({ id: result.metadata.sessionId })).info?.status).toBe("completed")
expect(loops).toEqual([])
yield* status.set(chat.id, { type: "idle" })
yield* Deferred.await(resumed).pipe(Effect.timeout("1 second"))
expect(loops).toEqual([chat.id])
}),
),
)
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<void>()
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)
}),
),
)
})
@@ -0,0 +1,280 @@
import { afterEach, describe, expect } from "bun:test"
import { Effect, Layer, Scope } from "effect"
import { Agent } from "../../src/agent/agent"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import { Instance } from "../../src/project/instance"
import { Session } from "@/session/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 "@/tool/truncate"
import { ModelID, ProviderID } from "../../src/provider/schema"
import { BackgroundJob } from "@/background/job"
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,
BackgroundJob.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.")
}),
),
)
})
+5 -3
View File
@@ -1719,9 +1719,11 @@ 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")