mirror of
https://github.com/anomalyco/opencode.git
synced 2026-09-04 03:34:34 +00:00
6e954f75ee
Separate ID-bound Session policy from host routing. Bind Inbox and Location preparation dependencies at construction, preserve admission and execution semantics, and cover the extracted ownership contracts directly.
716 lines
29 KiB
TypeScript
716 lines
29 KiB
TypeScript
import { describe, expect } from "bun:test"
|
|
import { Deferred, Effect, Fiber, Layer, Schema, Stream } from "effect"
|
|
import { LanguageModel } from "@opencode-ai/ai"
|
|
import { OpenAIChat } from "@opencode-ai/ai/protocols"
|
|
import { TestLLM } from "@opencode-ai/ai/testing"
|
|
import path from "path"
|
|
import { Money } from "@opencode-ai/schema/money"
|
|
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
|
|
import { LayerNodePlatform } from "@opencode-ai/core/effect/app-node-platform"
|
|
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
|
|
import { Global } from "@opencode-ai/util/global"
|
|
import { makeGlobalNode, makeLocationNode } from "@opencode-ai/util/effect/app-node"
|
|
import { Database } from "@opencode-ai/core/database/database"
|
|
import { Bus } from "@opencode-ai/core/bus"
|
|
import { Config } from "@opencode-ai/core/config"
|
|
import { Location } from "@opencode-ai/core/location"
|
|
import { Model } from "@opencode-ai/core/model"
|
|
import { Provider } from "@opencode-ai/core/provider"
|
|
import { AbsolutePath } from "@opencode-ai/core/schema"
|
|
import { Agent } from "@opencode-ai/core/agent"
|
|
import { Job } from "@opencode-ai/core/job"
|
|
import { KV } from "@opencode-ai/core/kv"
|
|
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
|
|
import { Session } from "@opencode-ai/core/session"
|
|
import { SessionEvent } from "@opencode-ai/core/session/event"
|
|
import { SessionExecution } from "@opencode-ai/core/session/execution"
|
|
import { SessionRestart } from "@opencode-ai/core/session/execution/restart"
|
|
import { SessionInbox } from "@opencode-ai/core/session/inbox"
|
|
import { SessionMessage } from "@opencode-ai/core/session/message"
|
|
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
|
|
import { SessionStore } from "@opencode-ai/core/session/store"
|
|
import { PluginRuntime } from "@opencode-ai/core/plugin/runtime"
|
|
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
|
|
import { Permission } from "@opencode-ai/core/permission"
|
|
import { SubagentTool } from "@opencode-ai/core/tool/plugin/subagent"
|
|
import { Tool } from "@opencode-ai/core/tool"
|
|
import { tmpdir } from "./fixture/tmpdir"
|
|
import { tempGlobalLayer } from "./fixture/global"
|
|
import { testEffect } from "./lib/effect"
|
|
import { executeTool, registerToolPlugin, toolIdentity } from "./lib/tool"
|
|
|
|
const childText = "child final response"
|
|
const completedOutput = (sessionID: Session.ID) =>
|
|
`<subagent sessionID="${sessionID}" state="completed">\n${childText}\n</subagent>`
|
|
const childModel = Model.Ref.make({ id: Model.ID.make("child"), providerID: Provider.ID.make("test") })
|
|
const parentModel = Model.Ref.make({ id: Model.ID.make("parent"), providerID: Provider.ID.make("test") })
|
|
const tokens = { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }
|
|
|
|
const outputSessionID = (value: unknown) =>
|
|
Schema.decodeUnknownSync(Schema.Struct({ sessionID: Session.ID }))(value).sessionID
|
|
|
|
const executionNode = makeGlobalNode({
|
|
service: SessionExecution.Service,
|
|
layer: Layer.effect(
|
|
SessionExecution.Service,
|
|
Effect.gen(function* () {
|
|
const bus = yield* Bus.Service
|
|
const store = yield* SessionStore.Service
|
|
const completed = new Set<Session.ID>()
|
|
const complete = Effect.fn("SubagentTest.complete")(function* (sessionID: Session.ID) {
|
|
if (completed.has(sessionID)) return
|
|
if ((yield* store.get(sessionID))?.title?.includes("fail")) {
|
|
yield* new SessionRunnerModel.ModelNotSelectedError({ sessionID })
|
|
return
|
|
}
|
|
completed.add(sessionID)
|
|
const assistantMessageID = SessionMessage.ID.create()
|
|
yield* bus.publish(SessionEvent.Step.Started, {
|
|
sessionID,
|
|
assistantMessageID,
|
|
agent: Agent.ID.make("reviewer"),
|
|
model: childModel,
|
|
})
|
|
yield* bus.publish(SessionEvent.Text.Started, {
|
|
sessionID,
|
|
assistantMessageID,
|
|
ordinal: 0,
|
|
})
|
|
yield* bus.publish(SessionEvent.Text.Ended, {
|
|
sessionID,
|
|
assistantMessageID,
|
|
ordinal: 0,
|
|
text: childText,
|
|
})
|
|
yield* bus.publish(SessionEvent.Step.Ended, {
|
|
sessionID,
|
|
assistantMessageID,
|
|
finish: "stop",
|
|
cost: Money.USD.zero,
|
|
tokens,
|
|
})
|
|
})
|
|
return SessionExecution.Service.of({
|
|
active: Effect.succeed(new Set()),
|
|
isActive: () => Effect.succeed(false),
|
|
resume: complete,
|
|
wake: () => Effect.void,
|
|
interrupt: () => Effect.succeed(false),
|
|
awaitIdle: (sessionID) => complete(sessionID).pipe(Effect.exit, Effect.asVoid),
|
|
})
|
|
}),
|
|
),
|
|
deps: [Bus.node, SessionStore.node],
|
|
})
|
|
|
|
const subagentPluginSupervisor = makeLocationNode({
|
|
service: PluginSupervisor.Service,
|
|
layer: Layer.effect(
|
|
PluginSupervisor.Service,
|
|
registerToolPlugin(SubagentTool.Plugin).pipe(Effect.as(PluginSupervisor.Service.of({ flush: Effect.void }))),
|
|
),
|
|
deps: [Agent.node, Config.node, Permission.node, PluginRuntime.node, Tool.node],
|
|
})
|
|
|
|
const nodes = LayerNode.group([
|
|
Database.node,
|
|
Bus.node,
|
|
Job.node,
|
|
Session.node,
|
|
SessionExecution.node,
|
|
PluginRuntime.providerNode,
|
|
LocationServiceMap.node,
|
|
])
|
|
const replacements = [
|
|
[SessionExecution.node, executionNode],
|
|
[Global.node, tempGlobalLayer],
|
|
] satisfies LayerNode.Replacements
|
|
const productionIt = testEffect(AppNodeBuilder.build(nodes, replacements))
|
|
const it = testEffect(AppNodeBuilder.build(nodes, [...replacements, [PluginSupervisor.node, subagentPluginSupervisor]]))
|
|
const completionIt = testEffect(
|
|
AppNodeBuilder.build(LayerNode.group([nodes, SessionRestart.node, KV.node]), [
|
|
[Global.node, tempGlobalLayer],
|
|
[PluginSupervisor.node, subagentPluginSupervisor],
|
|
[LayerNodePlatform.llmClient, TestLLM.testLayer({ fallback: TestLLM.text(childText, "completion") })],
|
|
[
|
|
SessionRunnerModel.node,
|
|
Layer.succeed(SessionRunnerModel.Service, {
|
|
resolve: () =>
|
|
Effect.succeed(
|
|
SessionRunnerModel.resolved(
|
|
LanguageModel.make({ id: "child", provider: "test", route: OpenAIChat.route }),
|
|
{
|
|
capabilities: { tools: true, input: ["text"], output: ["text"] },
|
|
cost: [],
|
|
limit: { context: 200_000, output: 32_000 },
|
|
},
|
|
),
|
|
),
|
|
}),
|
|
],
|
|
]),
|
|
)
|
|
|
|
const withSubagent = (location: Location.Ref) =>
|
|
Effect.gen(function* () {
|
|
const locations = yield* LocationServiceMap.Service
|
|
yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(Effect.provide(locations.get(location)))
|
|
yield* Agent.Service.use((agents) =>
|
|
agents.transform((draft) => {
|
|
// The caller identity used by executeTool; subagent permission asserts against it.
|
|
draft.update(toolIdentity.agent, (agent) => {
|
|
agent.mode = "primary"
|
|
agent.permissions.push({ action: "*", resource: "*", effect: "allow" })
|
|
})
|
|
draft.update(Agent.ID.make("reviewer"), (agent) => {
|
|
agent.mode = "subagent"
|
|
agent.model = childModel
|
|
})
|
|
draft.update(Agent.ID.make("fallback"), (agent) => {
|
|
agent.mode = "subagent"
|
|
})
|
|
draft.update(Agent.ID.make("primary"), (agent) => {
|
|
agent.mode = "primary"
|
|
})
|
|
}),
|
|
).pipe(Effect.provide(locations.get(location)))
|
|
})
|
|
|
|
describe("SubagentTool", () => {
|
|
completionIt.live("admits one durable completion across live delivery and restart replay", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({
|
|
location: Location.Ref.make({ directory: AbsolutePath.make(dir.path) }),
|
|
model: parentModel,
|
|
title: "Completion recipient",
|
|
})
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
const jobs = yield* Job.Service
|
|
const bus = yield* Bus.Service
|
|
const admitted = yield* Deferred.make<Job.Background>()
|
|
const notifications: SessionMessage.ID[] = []
|
|
yield* bus.project(SessionEvent.InboxEnqueued, (event) =>
|
|
Effect.gen(function* () {
|
|
if (event.data.sessionID !== parent.id || event.data.item.type !== "synthetic") return
|
|
notifications.push(event.data.inboxID)
|
|
const marker = (yield* jobs.pendingBackground).find((job) => job.notificationID === event.data.inboxID)
|
|
// The marker must survive until admission commits, not merely until delivery starts.
|
|
expect(marker?.status).toBe("completed")
|
|
if (marker) yield* Deferred.succeed(admitted, marker)
|
|
}),
|
|
)
|
|
|
|
const result = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-completion-replay",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "background review", prompt: "review", background: true },
|
|
},
|
|
})
|
|
const marker = yield* Deferred.await(admitted)
|
|
yield* jobs.pendingBackground.pipe(Effect.repeat({ until: (pending) => pending.length === 0 }))
|
|
yield* sessions.wait(parent.id)
|
|
const messages = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")
|
|
expect(messages).toEqual([
|
|
expect.objectContaining({
|
|
id: marker.notificationID,
|
|
description: "background review",
|
|
text: `<subagent sessionID="${outputSessionID(result.metadata)}" state="completed" description="background review">\n${childText}\n</subagent>`,
|
|
metadata: {
|
|
source: "subagent",
|
|
childID: outputSessionID(result.metadata),
|
|
agent: "reviewer",
|
|
state: "completed",
|
|
},
|
|
}),
|
|
])
|
|
|
|
// Reproduce a crash after admission but before acknowledgment using the real persisted marker.
|
|
const kv = yield* KV.Service
|
|
yield* kv.set(`job.background/${marker.notificationID}`, marker)
|
|
const restart = yield* SessionRestart.Service
|
|
yield* restart.resumeSuspendedSessions
|
|
yield* sessions.wait(parent.id)
|
|
expect(notifications).toEqual([marker.notificationID])
|
|
expect((yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")).toEqual(
|
|
messages,
|
|
)
|
|
expect(yield* jobs.pendingBackground).toEqual([])
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
productionIt.live("registers globally while resolving agents from the caller location", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const session = yield* Session.Service
|
|
const parent = yield* session.create({ location })
|
|
yield* withSubagent(parent.location)
|
|
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
expect((yield* registry.snapshot()).definitions.map((tool) => tool.name)).toContain(SubagentTool.name)
|
|
expect(
|
|
yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-primary",
|
|
name: SubagentTool.name,
|
|
input: { agent: "primary", description: "primary", prompt: "should fail" },
|
|
},
|
|
}),
|
|
).toEqual({
|
|
status: "error",
|
|
error: { type: "tool.execution", message: "Agent primary cannot run as a subagent" },
|
|
})
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("prevents subagents from launching subagents by default", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const root = yield* sessions.create({ location })
|
|
const parent = yield* sessions.create({ parentID: root.id, title: "parent" })
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
|
|
expect(
|
|
yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-nested-subagent",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "nested", prompt: "should fail" },
|
|
},
|
|
}),
|
|
).toEqual({
|
|
status: "error",
|
|
error: {
|
|
type: "tool.execution",
|
|
message: expect.stringContaining("Subagent depth limit reached (1)"),
|
|
},
|
|
})
|
|
expect((yield* sessions.list({ parentID: parent.id })).data).toHaveLength(0)
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("allows nested subagents up to the configured depth", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
yield* Effect.promise(() =>
|
|
Bun.write(path.join(dir.path, "opencode.json"), JSON.stringify({ experimental: { subagent_depth: 2 } })),
|
|
)
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const root = yield* sessions.create({ location })
|
|
const parent = yield* sessions.create({ parentID: root.id, title: "parent", model: parentModel })
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
|
|
const settled = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-configured-nested-subagent",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "nested", prompt: "should run" },
|
|
},
|
|
})
|
|
|
|
expect(settled).toMatchObject({
|
|
status: "completed",
|
|
metadata: { status: "completed" },
|
|
content: [{ type: "text", text: expect.stringContaining(childText) }],
|
|
})
|
|
expect(settled.metadata).toEqual({
|
|
sessionID: outputSessionID(settled.metadata),
|
|
status: "completed",
|
|
})
|
|
expect((yield* sessions.get(outputSessionID(settled.metadata))).parentID).toBe(parent.id)
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("runs a foreground child session and returns the final assistant text", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({ location, model: parentModel })
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
const progress: Tool.Metadata[] = []
|
|
|
|
const settled = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
progress: (update) => Effect.sync(() => progress.push(update)),
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-subagent",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "review", prompt: "review this" },
|
|
},
|
|
})
|
|
|
|
expect(settled).toMatchObject({
|
|
status: "completed",
|
|
metadata: { status: "completed" },
|
|
content: [{ type: "text", text: expect.stringContaining(childText) }],
|
|
})
|
|
const child = yield* sessions.get(outputSessionID(settled.metadata))
|
|
expect(settled.content).toEqual([{ type: "text", text: completedOutput(child.id) }])
|
|
expect(settled.metadata).toEqual({ sessionID: child.id, status: "completed" })
|
|
expect(progress[0]).toEqual({ sessionID: child.id, status: "running" })
|
|
expect(child).toMatchObject({
|
|
parentID: parent.id,
|
|
location: parent.location,
|
|
agent: "reviewer",
|
|
model: childModel,
|
|
})
|
|
expect((yield* sessions.inbox(child.id)).find((message) => message.type === "user")?.payload.text).toBe(
|
|
"You are a subagent spawned by another session.\nreview this",
|
|
)
|
|
|
|
const fallback = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-subagent-fallback",
|
|
name: SubagentTool.name,
|
|
input: { agent: "fallback", description: "fallback", prompt: "fallback" },
|
|
},
|
|
})
|
|
const fallbackChild = yield* sessions.get(outputSessionID(fallback.metadata))
|
|
expect(fallbackChild).toMatchObject({ parentID: parent.id, model: parentModel })
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("continues an existing child session", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({ location, model: parentModel })
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
|
|
const first = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-subagent-first",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "review", prompt: "review this" },
|
|
},
|
|
})
|
|
const childID = outputSessionID(first.metadata)
|
|
const second = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-subagent-second",
|
|
name: SubagentTool.name,
|
|
input: {
|
|
agent: "reviewer",
|
|
description: "follow up",
|
|
prompt: "continue this",
|
|
sessionID: childID,
|
|
},
|
|
},
|
|
})
|
|
|
|
expect(outputSessionID(second.metadata)).toBe(childID)
|
|
expect((yield* sessions.list({ parentID: parent.id })).data).toHaveLength(1)
|
|
expect((yield* sessions.get(childID)).title).toBe("review")
|
|
expect(
|
|
(yield* sessions.inbox(childID)).flatMap((message) =>
|
|
message.type === "user" ? [message.payload.text] : [],
|
|
),
|
|
).toEqual(["You are a subagent spawned by another session.\nreview this", "continue this"])
|
|
expect(second.content).toEqual([{ type: "text", text: completedOutput(childID) }])
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("steers a running child session in the background", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({ location, model: parentModel })
|
|
const child = yield* sessions.create({
|
|
parentID: parent.id,
|
|
title: "review",
|
|
agent: Agent.ID.make("reviewer"),
|
|
model: childModel,
|
|
})
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
const jobs = yield* Job.Service
|
|
yield* jobs.start({ id: child.id, type: SubagentTool.name, run: Effect.never })
|
|
|
|
const result = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-running-subagent",
|
|
name: SubagentTool.name,
|
|
input: {
|
|
agent: "reviewer",
|
|
description: "follow up",
|
|
prompt: "continue while running",
|
|
sessionID: child.id,
|
|
background: true,
|
|
},
|
|
},
|
|
})
|
|
|
|
expect(result).toMatchObject({
|
|
status: "completed",
|
|
metadata: { sessionID: child.id, status: "running" },
|
|
})
|
|
expect((yield* sessions.inbox(child.id)).find((message) => message.type === "user")?.payload.text).toBe(
|
|
"continue while running",
|
|
)
|
|
expect((yield* jobs.get(child.id))?.status).toBe("running")
|
|
yield* jobs.cancel(child.id)
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("rejects unrelated children and switches agents on continuation", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({ location, model: parentModel })
|
|
const otherParent = yield* sessions.create({ location, model: parentModel })
|
|
const unrelated = yield* sessions.create({
|
|
parentID: otherParent.id,
|
|
title: "other review",
|
|
agent: Agent.ID.make("reviewer"),
|
|
})
|
|
const switched = yield* sessions.create({
|
|
parentID: parent.id,
|
|
title: "fallback review",
|
|
agent: Agent.ID.make("fallback"),
|
|
model: parentModel,
|
|
})
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
const call = (sessionID: Session.ID, id: string, agent = "reviewer") =>
|
|
executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call" as const,
|
|
id,
|
|
name: SubagentTool.name,
|
|
input: { agent, description: "follow up", prompt: "continue", sessionID },
|
|
},
|
|
})
|
|
|
|
const missing = Session.ID.create()
|
|
expect(yield* call(missing, "call-missing-child")).toEqual({
|
|
status: "error",
|
|
error: {
|
|
type: "tool.execution",
|
|
message: `Subagent session not found: ${missing}`,
|
|
},
|
|
})
|
|
expect(yield* call(unrelated.id, "call-unrelated-child")).toEqual({
|
|
status: "error",
|
|
error: {
|
|
type: "tool.execution",
|
|
message: `Session ${unrelated.id} is not a child of the current session`,
|
|
},
|
|
})
|
|
expect(yield* call(switched.id, "call-switched-child")).toMatchObject({
|
|
status: "completed",
|
|
metadata: { sessionID: switched.id, status: "completed" },
|
|
})
|
|
expect(yield* sessions.get(switched.id)).toMatchObject({
|
|
agent: "reviewer",
|
|
model: childModel,
|
|
})
|
|
// Switching to an agent without a configured model keeps the child's current model.
|
|
expect(yield* call(switched.id, "call-modelless-switch", "fallback")).toMatchObject({
|
|
status: "completed",
|
|
metadata: { sessionID: switched.id, status: "completed" },
|
|
})
|
|
expect(yield* sessions.get(switched.id)).toMatchObject({
|
|
agent: "fallback",
|
|
model: childModel,
|
|
})
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("returns child runner failures as tool errors", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({ location })
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
|
|
expect(
|
|
yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-subagent-failure",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "fail review", prompt: "please fail" },
|
|
},
|
|
}),
|
|
).toEqual({
|
|
status: "error",
|
|
error: {
|
|
type: "tool.execution",
|
|
message: expect.stringContaining("No model is available for session"),
|
|
},
|
|
})
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
|
|
it.live("notifies once when background work completes", () =>
|
|
Effect.acquireRelease(
|
|
Effect.promise(() => tmpdir()),
|
|
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
|
|
).pipe(
|
|
Effect.flatMap((dir) =>
|
|
Effect.gen(function* () {
|
|
const location = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
|
const sessions = yield* Session.Service
|
|
const parent = yield* sessions.create({ location })
|
|
yield* withSubagent(parent.location)
|
|
const locations = yield* LocationServiceMap.Service
|
|
const registry = yield* Tool.Service.pipe(Effect.provide(locations.get(parent.location)))
|
|
const bus = yield* Bus.Service
|
|
const admitted = yield* bus.subscribe(SessionEvent.InboxEnqueued).pipe(
|
|
Stream.filter((event) => event.data.sessionID === parent.id && event.data.item.type === "synthetic"),
|
|
Stream.take(1),
|
|
Stream.runCollect,
|
|
Effect.forkScoped({ startImmediately: true }),
|
|
)
|
|
|
|
const settled = yield* executeTool(registry, {
|
|
sessionID: parent.id,
|
|
...toolIdentity,
|
|
call: {
|
|
type: "tool-call",
|
|
id: "call-background-subagent",
|
|
name: SubagentTool.name,
|
|
input: { agent: "reviewer", description: "background review", prompt: "review", background: true },
|
|
},
|
|
})
|
|
const childID = outputSessionID(settled.metadata)
|
|
expect(settled.metadata).toMatchObject({
|
|
status: "running",
|
|
})
|
|
expect(settled.metadata).toEqual({ sessionID: childID, status: "running" })
|
|
expect(settled.content).toEqual([{ type: "text", text: expect.stringContaining(`sessionID: ${childID}`) }])
|
|
|
|
const admission = Array.from(yield* Fiber.join(admitted))[0]
|
|
expect(admission?.data.item.type).toBe("synthetic")
|
|
if (admission?.data.item.type !== "synthetic") return yield* Effect.die("Expected synthetic inbox item")
|
|
expect(admission?.data.item.payload.text).toContain(`<subagent sessionID="${childID}" state="completed"`)
|
|
expect(admission?.data.item.payload).toMatchObject({
|
|
description: "background review",
|
|
metadata: {
|
|
source: "subagent",
|
|
childID,
|
|
agent: "reviewer",
|
|
state: "completed",
|
|
},
|
|
})
|
|
const database = yield* Database.Service
|
|
yield* SessionInbox.promote(database.db, bus, parent.id, "steer")
|
|
const synthetic = (yield* sessions.context(parent.id)).filter((message) => message.type === "synthetic")
|
|
expect(synthetic).toHaveLength(1)
|
|
expect(synthetic[0]?.text).toContain(`<subagent sessionID="${childID}" state="completed"`)
|
|
expect(synthetic[0]?.text).toContain(childText)
|
|
}),
|
|
),
|
|
),
|
|
)
|
|
})
|