ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore(observability): merge v2

Browse source
This commit is contained in:
starptech 2026-07-08 21:36:57 +02:00
commit 18da3deee7
16 changed files with 489 additions and 697 deletions
Download patch file Download diff file Expand all files Collapse all files

2
packages/core/test/config/plugin.test.ts
View file

@ -229,7 +229,7 @@ describe("PluginSupervisor config", () => {
const ready = Effect.fnUntraced(function* () {
const supervisor = yield* PluginSupervisor.Service
yield* supervisor.ready
yield* supervisor.flush
})
function withLocation<A, E, R>(

270
packages/core/test/location-layer.test.ts
View file

@ -4,7 +4,7 @@ import { describe, expect } from "bun:test"
import { Config } from "@opencode-ai/schema/config"
import { Plugin } from "@opencode-ai/schema/plugin"
import { Money } from "@opencode-ai/schema/money"
import { Context, DateTime, Effect, Equal, Hash, RcMap, Schema, Stream } from "effect"
import { Context, DateTime, Deferred, Effect, Equal, Fiber, Hash, RcMap, Schema, Stream } from "effect"
import { Plugin as EffectPlugin } from "@opencode-ai/plugin/v2/effect"
import { AgentV2 } from "@opencode-ai/core/agent"
import { Catalog } from "@opencode-ai/core/catalog"
@ -54,7 +54,7 @@ describe("LocationServiceMap", () => {
const ref = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
const read = Effect.gen(function* () {
const supervisor = yield* PluginSupervisor.Service
yield* supervisor.ready
yield* supervisor.flush
const agents = yield* AgentV2.Service
return yield* agents.get(id)
})
@ -67,6 +67,268 @@ describe("LocationServiceMap", () => {
),
)
itWithSdk.live("waits for explorer activation to complete", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
).pipe(
Effect.flatMap((dir) =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const sdk = yield* SdkPlugins.Service
yield* sdk.register(
EffectPlugin.define({
id: "blocked-initial-activation",
effect: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release))),
}),
)
const locations = yield* LocationServiceMap.Service
const context = yield* locations.contextEffect(Location.Ref.make({ directory: AbsolutePath.make(dir.path) }))
yield* Deferred.await(started)
const flushFiber = yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.forkChild,
)
expect(flushFiber.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(flushFiber)
yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.timeout("1 second"),
)
const explorer = yield* Effect.gen(function* () {
const agents = yield* AgentV2.Service
return yield* agents.resolve("explore")
}).pipe(Effect.provide(context))
expect(explorer).toBeDefined()
expect(explorer?.permissions.length).toBeGreaterThan(0)
}),
),
),
)
itWithSdk.live("reruns activation for SDK plugins registered during startup", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
).pipe(
Effect.flatMap((dir) =>
Effect.gen(function* () {
const firstStarted = yield* Deferred.make<void>()
const releaseFirst = yield* Deferred.make<void>()
const secondStarted = yield* Deferred.make<void>()
const releaseSecond = yield* Deferred.make<void>()
const sdk = yield* SdkPlugins.Service
yield* sdk.register(
EffectPlugin.define({
id: "fixed-target-first-plugin",
effect: () =>
Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseFirst))),
}),
)
const locations = yield* LocationServiceMap.Service
const context = yield* locations.contextEffect(Location.Ref.make({ directory: AbsolutePath.make(dir.path) }))
yield* Deferred.await(firstStarted)
const flushFiber = yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.forkChild({ startImmediately: true }),
)
yield* Effect.yieldNow
yield* sdk.register(
EffectPlugin.define({
id: "fixed-target-second-plugin",
effect: () =>
Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseSecond))),
}),
)
yield* Deferred.succeed(releaseFirst, undefined)
yield* Deferred.await(secondStarted)
expect(flushFiber.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(releaseSecond, undefined)
yield* Fiber.join(flushFiber)
}),
),
),
)
itWithSdk.live("reruns activation for Config updates during startup", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
).pipe(
Effect.flatMap((dir) =>
Effect.gen(function* () {
const activations = { count: 0 }
const file = path.join(dir.path, "opencode.json")
yield* Effect.promise(() => fs.writeFile(file, "{}"))
const firstStarted = yield* Deferred.make<void>()
const releaseFirst = yield* Deferred.make<void>()
const secondStarted = yield* Deferred.make<void>()
const releaseSecond = yield* Deferred.make<void>()
const sdk = yield* SdkPlugins.Service
yield* sdk.register(
EffectPlugin.define({
id: "blocked-config-reload",
effect: () =>
Effect.sync(() => ++activations.count).pipe(
Effect.flatMap((activation) =>
activation === 1
? Deferred.succeed(firstStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseFirst)))
: Deferred.succeed(secondStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseSecond))),
),
),
}),
)
const locations = yield* LocationServiceMap.Service
const context = yield* locations.contextEffect(Location.Ref.make({ directory: AbsolutePath.make(dir.path) }))
yield* Deferred.await(firstStarted)
const events = yield* EventV2.Service
const updated = yield* events.subscribe(Config.Event.Updated).pipe(
Stream.filter((event) => event.location?.directory === dir.path),
Stream.runHead,
Effect.forkChild({ startImmediately: true }),
)
yield* Effect.promise(() =>
fs.writeFile(
file,
JSON.stringify({ plugins: [path.join(import.meta.dir, "plugin/fixtures/config-effect-plugin.ts")] }),
),
)
yield* Fiber.join(updated)
const flushFiber = yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.forkChild,
)
yield* Deferred.succeed(releaseFirst, undefined)
yield* Deferred.await(secondStarted)
expect(flushFiber.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(releaseSecond, undefined)
yield* Fiber.join(flushFiber)
expect(activations.count).toBe(2)
}),
),
),
)
itWithSdk.live("keeps flush pending while startup updates continue", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
).pipe(
Effect.flatMap((dir) =>
Effect.gen(function* () {
const locations = yield* LocationServiceMap.Service
const context = yield* locations.contextEffect(Location.Ref.make({ directory: AbsolutePath.make(dir.path) }))
const flushFiber = yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.forkChild({ startImmediately: true }),
)
const events = yield* EventV2.Service
yield* Effect.forEach(
Array.from({ length: 5 }),
() => events.publish(SdkPlugins.Updated, {}).pipe(Effect.andThen(Effect.sleep("50 millis"))),
{ discard: true },
)
expect(flushFiber.pollUnsafe()).toBeUndefined()
yield* Fiber.join(flushFiber)
}),
),
),
)
itWithSdk.live("keeps flush open while later hot reload runs", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
).pipe(
Effect.flatMap((dir) =>
Effect.gen(function* () {
const locations = yield* LocationServiceMap.Service
const context = yield* locations.contextEffect(Location.Ref.make({ directory: AbsolutePath.make(dir.path) }))
yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(Effect.provide(context))
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const completed = yield* Deferred.make<void>()
const sdk = yield* SdkPlugins.Service
yield* sdk.register(
EffectPlugin.define({
id: "post-ready-plugin",
effect: () =>
Deferred.succeed(started, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.andThen(Deferred.succeed(completed, undefined)),
),
}),
)
yield* Deferred.await(started)
yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.timeout("1 second"),
)
yield* Deferred.succeed(release, undefined)
yield* Deferred.await(completed)
}),
),
),
)
itWithSdk.live("does not cancel activation when a flush waiter is interrupted", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
(dir) => Effect.promise(() => dir[Symbol.asyncDispose]()),
).pipe(
Effect.flatMap((dir) =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const completed = yield* Deferred.make<void>()
const sdk = yield* SdkPlugins.Service
yield* sdk.register(
EffectPlugin.define({
id: "interrupted-waiter-plugin",
effect: () =>
Deferred.succeed(started, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.andThen(Deferred.succeed(completed, undefined)),
),
}),
)
const locations = yield* LocationServiceMap.Service
const context = yield* locations.contextEffect(Location.Ref.make({ directory: AbsolutePath.make(dir.path) }))
yield* Deferred.await(started)
const flushFiber = yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.forkChild({ startImmediately: true }),
)
yield* Fiber.interrupt(flushFiber)
yield* Deferred.succeed(release, undefined)
yield* Deferred.await(completed)
yield* PluginSupervisor.Service.use((supervisor) => supervisor.flush).pipe(
Effect.provide(context),
Effect.timeout("500 millis"),
)
}),
),
),
)
it.live("applies ordered plugin config operations during boot", () =>
Effect.acquireRelease(
Effect.promise(() => tmpdir()),
@ -79,7 +341,7 @@ describe("LocationServiceMap", () => {
)
const plugins = yield* Effect.gen(function* () {
const plugins = yield* PluginV2.Service
yield* (yield* PluginSupervisor.Service).ready
yield* (yield* PluginSupervisor.Service).flush
return yield* plugins.list()
}).pipe(
Effect.scoped,
@ -106,7 +368,7 @@ describe("LocationServiceMap", () => {
yield* Effect.gen(function* () {
const registry = yield* PluginV2.Service
const supervisor = yield* PluginSupervisor.Service
yield* supervisor.ready
yield* supervisor.flush
expect((yield* registry.list()).map((plugin) => String(plugin.id))).toEqual(["opencode.agent"])
yield* Effect.promise(() => fs.writeFile(file, JSON.stringify({ plugins: ["-*", "opencode.command"] })))

10
packages/core/test/session-runner-recorded.test.ts
View file

@ -43,6 +43,7 @@ import { SkillGuidance } from "@opencode-ai/core/skill/guidance"
import { ReferenceGuidance } from "@opencode-ai/core/reference/guidance"
import { McpGuidance } from "@opencode-ai/core/mcp/guidance"
import { SessionTelemetry } from "@opencode-ai/core/observability/session"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
import { describe, expect } from "bun:test"
import { eq } from "drizzle-orm"
import { Effect, Layer, References, Tracer } from "effect"
@ -83,6 +84,7 @@ const skillGuidance = Layer.mock(SkillGuidance.Service, { load: () => Effect.suc
const referenceGuidance = Layer.mock(ReferenceGuidance.Service, { load: () => Effect.succeed(Instructions.empty) })
const mcpGuidance = Layer.mock(McpGuidance.Service, { load: () => Effect.succeed(Instructions.empty) })
const config = Layer.succeed(Config.Service, Config.Service.of({ entries: () => Effect.succeed([]) }))
const pluginSupervisor = Layer.succeed(PluginSupervisor.Service, PluginSupervisor.Service.of({ flush: Effect.void }))
const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
[Snapshot.node, Snapshot.noopLayer],
[LayerNodePlatform.llmClient, client],
@ -96,6 +98,7 @@ const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
[Config.node, config],
[PermissionV2.node, permission],
[ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig],
[PluginSupervisor.node, pluginSupervisor],
])
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
@ -162,6 +165,7 @@ const it = testEffect(
[ReferenceGuidance.node, referenceGuidance],
[Config.node, config],
[Snapshot.node, Snapshot.noopLayer],
[PluginSupervisor.node, pluginSupervisor],
[SessionExecution.node, execution],
],
),
@ -172,6 +176,12 @@ describe("SessionRunnerLLM recorded", () => {
it.effect("executes one recorded V2 prompt through the recorded HTTP transport", () =>
Effect.gen(function* () {
spans.length = 0
const agents = yield* AgentV2.Service
yield* agents.transform((draft) =>
draft.update(AgentV2.ID.make("build"), (agent) => {
agent.mode = "primary"
}),
)
const { db } = yield* Database.Service
yield* db
.insert(ProjectTable)

71
packages/core/test/session-runner.test.ts
View file

@ -75,6 +75,7 @@ import * as SessionRunnerLLM from "@opencode-ai/core/session/runner/llm"
import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { SessionRunnerSystemPrompt } from "@opencode-ai/core/session/runner/system-prompt"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
import { QuestionTool } from "@opencode-ai/core/tool/question"
import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
import { AgentV2 } from "@opencode-ai/core/agent"
@ -344,6 +345,13 @@ const config = Layer.succeed(
]),
}),
)
let pluginFlushHook = Effect.void
const pluginSupervisor = Layer.succeed(
PluginSupervisor.Service,
PluginSupervisor.Service.of({
flush: Effect.suspend(() => pluginFlushHook),
}),
)
const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
[Snapshot.node, Snapshot.noopLayer],
[LayerNodePlatform.llmClient, client],
@ -357,6 +365,7 @@ const runnerLayer = AppNodeBuilder.build(SessionRunnerLLM.node, [
[Config.node, config],
[McpGuidance.node, mcpGuidance],
[ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig],
[PluginSupervisor.node, pluginSupervisor],
])
const spans: Tracer.NativeSpan[] = []
const tracer = Tracer.make({
@ -429,6 +438,7 @@ const it = testEffect(
[SessionExecution.node, execution],
[Config.node, config],
[ToolOutputStore.node, ToolOutputStore.nodeWithoutConfig],
[PluginSupervisor.node, pluginSupervisor],
],
),
)
@ -466,6 +476,7 @@ const setup = Effect.gen(function* () {
systemUnavailable = false
systemLoadHook = Effect.void
modelResolveHook = Effect.void
pluginFlushHook = Effect.void
currentModel = model
skillBaselines.clear()
responses = undefined
@ -479,6 +490,12 @@ const setup = Effect.gen(function* () {
activeToolExecutions = 0
maxActiveToolExecutions = 0
spans.length = 0
const agents = yield* AgentV2.Service
yield* agents.transform((draft) =>
draft.update(AgentV2.ID.make("build"), (agent) => {
agent.mode = "primary"
}),
)
yield* db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
@ -1307,10 +1324,64 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("fails before the model request when the selected agent is unavailable", () =>
Effect.gen(function* () {
yield* setup
const { db } = yield* Database.Service
yield* db
.update(SessionTable)
.set({ agent: "explore" })
.where(eq(SessionTable.id, sessionID))
.run()
.pipe(Effect.orDie)
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Inspect files" }), resume: false })
requests.length = 0
response = []
const failure = yield* session.resume(sessionID).pipe(Effect.flip)
expect(failure).toMatchObject({
_tag: "Session.AgentNotFoundError",
sessionID,
agent: "explore",
})
expect(requests).toHaveLength(0)
}),
)
it.effect("waits for initial plugin readiness before constructing the model request", () =>
Effect.gen(function* () {
yield* setup
const release = yield* Deferred.make<void>()
pluginFlushHook = Deferred.await(release)
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: PromptInput.Prompt.make({ text: "Wait for plugins" }), resume: false })
requests.length = 0
response = []
const running = yield* session.resume(sessionID).pipe(Effect.forkChild({ startImmediately: true }))
yield* Effect.yieldNow
expect(requests).toHaveLength(0)
expect(running.pollUnsafe()).toBeUndefined()
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(running)
expect(requests).toHaveLength(1)
}),
)
it.effect("updates selected-agent skill guidance after an agent switch", () =>
Effect.gen(function* () {
const session = yield* setup
const events = yield* EventV2.Service
const agents = yield* AgentV2.Service
yield* agents.transform((draft) =>
draft.update(AgentV2.ID.make("reviewer"), (agent) => {
agent.mode = "primary"
}),
)
skillBaselines.set(AgentV2.ID.make("build"), "Build skills")
yield* admit(session, "First")

2
packages/core/test/tool-subagent.test.ts
View file

@ -25,6 +25,7 @@ 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 { SubagentTool } from "@opencode-ai/core/tool/subagent"
import { ToolRegistry } from "@opencode-ai/core/tool/registry"
import { ToolOutputStore } from "@opencode-ai/core/tool-output-store"
@ -125,6 +126,7 @@ const it = testEffect(layer)
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* AgentV2.Service.use((agents) =>
agents.transform((draft) => {
// The caller identity used by executeTool; subagent permission asserts against it.