Mirror_Repos/anomalyco_opencode
2
0
Fork 0
Code Issues Pull Requests Actions 66 Packages Projects Releases Wiki Activity

effectify Bus service: migrate to Effect PubSub + InstanceState

Browse Source 
- Replace manual subscription Map with PubSub.unbounded per instance
- Per-type PubSubs + wildcard PubSub, cleaned up via addFinalizer
- InstanceDisposed published before PubSub shutdown so subscribers see it
- Replace makeRunPromise with makeRuntime (single runtime with runPromise + runFork)
- Update all 19 services to use makeRuntime destructuring
- Legacy facade preserved: publish/subscribe/subscribeAll same signatures
- subscribe/subscribeAll fork stream consumer fibers, return interrupt function
- Extract Format.file() for explicit formatting, remove event-driven subscription
- Inline Format.file() calls in write/edit/apply_patch tools
- Drop Bus.once (zero callers)
This commit is contained in:
Kit Langton
2026-03-21 21:41:29 -04:00
parent a724b01652
commit 050336f160
21 changed files with 42 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/opencode/src/account/index.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import { Clock, Duration, Effect, Layer, Option, Schema, SchemaGetter, ServiceMap } from "effect"
import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { withTransientReadRetry } from "@/util/effect-http-client"
import { AccountRepo, type AccountRow } from "./repo"
import {
@@ -379,7 +379,7 @@ export namespace Account {
export const defaultLayer = layer.pipe(Layer.provide(AccountRepo.layer), Layer.provide(FetchHttpClient.layer))
export const runPromise = makeRunPromise(Service, defaultLayer)
export const { runPromise } = makeRuntime(Service, defaultLayer)
export async function active(): Promise<Info | undefined> {
return Option.getOrUndefined(await runPromise((service) => service.active()))
packages/opencode/src/auth/index.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import path from "path"
import { Effect, Layer, Record, Result, Schema, ServiceMap } from "effect"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { zod } from "@/util/effect-zod"
import { Global } from "../global"
import { Filesystem } from "../util/filesystem"
@@ -95,7 +95,7 @@ export namespace Auth {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function get(providerID: string) {
return runPromise((service) => service.get(providerID))
packages/opencode/src/command/index.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { SessionID, MessageID } from "@/session/schema"
import { Effect, Layer, ServiceMap } from "effect"
import z from "zod"
@@ -173,7 +173,7 @@ export namespace Command {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function get(name: string) {
return runPromise((svc) => svc.get(name))
packages/opencode/src/effect/run-service.ts
-4
View File
@@ -14,7 +14,3 @@ export function makeRuntime<I, S, E>(service: ServiceMap.Service<I, S>, layer: L
}
}
/** @deprecated Use `makeRuntime` instead */
export function makeRunPromise<I, S, E>(service: ServiceMap.Service<I, S>, layer: Layer.Layer<I, E>) {
return makeRuntime(service, layer).runPromise
}
packages/opencode/src/file/index.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { git } from "@/util/git"
import { Effect, Fiber, Layer, Scope, ServiceMap } from "effect"
import { formatPatch, structuredPatch } from "diff"
@@ -688,7 +688,7 @@ export namespace File {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export function init() {
return runPromise((svc) => svc.init())
packages/opencode/src/file/time.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { DateTime, Effect, Layer, Semaphore, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { Flag } from "@/flag/flag"
import type { SessionID } from "@/session/schema"
import { Filesystem } from "../util/filesystem"
@@ -108,7 +108,7 @@ export namespace FileTime {
}),
).pipe(Layer.orDie)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export function read(sessionID: SessionID, file: string) {
return runPromise((s) => s.read(sessionID, file))
packages/opencode/src/file/watcher.ts
+2 -2
View File
@@ -8,7 +8,7 @@ import z from "zod"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { Flag } from "@/flag/flag"
import { Instance } from "@/project/instance"
import { git } from "@/util/git"
@@ -159,7 +159,7 @@ export namespace FileWatcher {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export function init() {
return runPromise((svc) => svc.init())
packages/opencode/src/format/index.ts
+2 -2
View File
@@ -1,6 +1,6 @@
import { Effect, Layer, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import path from "path"
import { mergeDeep } from "remeda"
import z from "zod"
@@ -167,7 +167,7 @@ export namespace Format {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function init() {
return runPromise((s) => s.init())
packages/opencode/src/installation/index.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import { NodeChildProcessSpawner, NodeFileSystem, NodePath } from "@effect/platform-node"
import { Effect, Layer, Schema, ServiceMap, Stream } from "effect"
import { FetchHttpClient, HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { withTransientReadRetry } from "@/util/effect-http-client"
import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import path from "path"
@@ -330,7 +330,7 @@ export namespace Installation {
Layer.provide(NodePath.layer),
)
const runPromise = makeRunPromise(Service, defaultLayer)
const { runPromise } = makeRuntime(Service, defaultLayer)
export async function info(): Promise<Info> {
return runPromise((svc) => svc.info())
packages/opencode/src/permission/index.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { Config } from "@/config/config"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { ProjectID } from "@/project/schema"
import { Instance } from "@/project/instance"
import { MessageID, SessionID } from "@/session/schema"
@@ -306,7 +306,7 @@ export namespace Permission {
return result
}
export const runPromise = makeRunPromise(Service, layer)
export const { runPromise } = makeRuntime(Service, layer)
export async function ask(input: z.infer<typeof AskInput>) {
return runPromise((s) => s.ask(input))
packages/opencode/src/plugin/index.ts
+2 -2
View File
@@ -13,7 +13,7 @@ import { CopilotAuthPlugin } from "./copilot"
import { gitlabAuthPlugin as GitlabAuthPlugin } from "opencode-gitlab-auth"
import { Effect, Layer, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
export namespace Plugin {
const log = Log.create({ service: "plugin" })
@@ -186,7 +186,7 @@ export namespace Plugin {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function trigger<
Name extends TriggerName,
packages/opencode/src/project/vcs.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import { Effect, Layer, ServiceMap } from "effect"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { FileWatcher } from "@/file/watcher"
import { Log } from "@/util/log"
import { git } from "@/util/git"
@@ -99,7 +99,7 @@ export namespace Vcs {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export function init() {
return runPromise((svc) => svc.init())
packages/opencode/src/provider/auth.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import type { AuthOuathResult, Hooks } from "@opencode-ai/plugin"
import { NamedError } from "@opencode-ai/util/error"
import { Auth } from "@/auth"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { Plugin } from "../plugin"
import { ProviderID } from "./schema"
import { Array as Arr, Effect, Layer, Record, Result, ServiceMap } from "effect"
@@ -231,7 +231,7 @@ export namespace ProviderAuth {
export const defaultLayer = layer.pipe(Layer.provide(Auth.layer))
const runPromise = makeRunPromise(Service, defaultLayer)
const { runPromise } = makeRuntime(Service, defaultLayer)
export async function methods() {
return runPromise((svc) => svc.methods())
packages/opencode/src/pty/index.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import { BusEvent } from "@/bus/bus-event"
import { Bus } from "@/bus"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { Instance } from "@/project/instance"
import { type IPty } from "bun-pty"
import z from "zod"
@@ -361,7 +361,7 @@ export namespace Pty {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function list() {
return runPromise((svc) => svc.list())
packages/opencode/src/question/index.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import { Deferred, Effect, Layer, Schema, ServiceMap } from "effect"
import { Bus } from "@/bus"
import { BusEvent } from "@/bus/bus-event"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { SessionID, MessageID } from "@/session/schema"
import { Log } from "@/util/log"
import z from "zod"
@@ -197,7 +197,7 @@ export namespace Question {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function ask(input: {
sessionID: SessionID
packages/opencode/src/session/status.ts
+2 -2
View File
@@ -1,7 +1,7 @@
import { BusEvent } from "@/bus/bus-event"
import { Bus } from "@/bus"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { SessionID } from "./schema"
import { Effect, Layer, ServiceMap } from "effect"
import z from "zod"
@@ -83,7 +83,7 @@ export namespace SessionStatus {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function get(sessionID: SessionID) {
return runPromise((svc) => svc.get(sessionID))
packages/opencode/src/skill/index.ts
+2 -2
View File
@@ -7,7 +7,7 @@ import { NamedError } from "@opencode-ai/util/error"
import type { Agent } from "@/agent/agent"
import { Bus } from "@/bus"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { Flag } from "@/flag/flag"
import { Global } from "@/global"
import { Permission } from "@/permission"
@@ -242,7 +242,7 @@ export namespace Skill {
return ["## Available Skills", ...list.map((skill) => `- **${skill.name}**: ${skill.description}`)].join("\n")
}
const runPromise = makeRunPromise(Service, defaultLayer)
const { runPromise } = makeRuntime(Service, defaultLayer)
export async function get(name: string) {
return runPromise((skill) => skill.get(name))
packages/opencode/src/snapshot/index.ts
+2 -2
View File
@@ -4,7 +4,7 @@ import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process"
import path from "path"
import z from "zod"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem"
import { Config } from "../config/config"
import { Global } from "../global"
@@ -360,7 +360,7 @@ export namespace Snapshot {
Layer.provide(NodePath.layer),
)
const runPromise = makeRunPromise(Service, defaultLayer)
const { runPromise } = makeRuntime(Service, defaultLayer)
export async function init() {
return runPromise((svc) => svc.init())
packages/opencode/src/tool/registry.ts
+2 -2
View File
@@ -31,7 +31,7 @@ import { Glob } from "../util/glob"
import { pathToFileURL } from "url"
import { Effect, Layer, ServiceMap } from "effect"
import { InstanceState } from "@/effect/instance-state"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
export namespace ToolRegistry {
const log = Log.create({ service: "tool.registry" })
@@ -198,7 +198,7 @@ export namespace ToolRegistry {
}),
)
const runPromise = makeRunPromise(Service, layer)
const { runPromise } = makeRuntime(Service, layer)
export async function register(tool: Tool.Info) {
return runPromise((svc) => svc.register(tool))
packages/opencode/src/tool/truncate.ts
+2 -2
View File
@@ -2,7 +2,7 @@ import { NodePath } from "@effect/platform-node"
import { Cause, Duration, Effect, Layer, Schedule, ServiceMap } from "effect"
import path from "path"
import type { Agent } from "../agent/agent"
import { makeRunPromise } from "@/effect/run-service"
import { makeRuntime } from "@/effect/run-service"
import { AppFileSystem } from "@/filesystem"
import { evaluate } from "@/permission/evaluate"
import { Identifier } from "../id/id"
@@ -136,7 +136,7 @@ export namespace Truncate {
export const defaultLayer = layer.pipe(Layer.provide(AppFileSystem.defaultLayer), Layer.provide(NodePath.layer))
const runPromise = makeRunPromise(Service, defaultLayer)
const { runPromise } = makeRuntime(Service, defaultLayer)
export async function output(text: string, options: Options = {}, agent?: Agent.Info): Promise<Result> {
return runPromise((s) => s.output(text, options, agent))
packages/opencode/test/effect/run-service.test.ts
+4 -4
View File
@@ -1,10 +1,10 @@
import { expect, test } from "bun:test"
import { Effect, Layer, ServiceMap } from "effect"
import { makeRunPromise } from "../../src/effect/run-service"
import { makeRuntime } from "../../src/effect/run-service"
class Shared extends ServiceMap.Service<Shared, { readonly id: number }>()("@test/Shared") {}
test("makeRunPromise shares dependent layers through the shared memo map", async () => {
test("makeRuntime shares dependent layers through the shared memo map", async () => {
let n = 0
const shared = Layer.effect(
@@ -37,8 +37,8 @@ test("makeRunPromise shares dependent layers through the shared memo map", async
}),
).pipe(Layer.provide(shared))
const runOne = makeRunPromise(One, one)
const runTwo = makeRunPromise(Two, two)
const { runPromise: runOne } = makeRuntime(One, one)
const { runPromise: runTwo } = makeRuntime(Two, two)
expect(await runOne((svc) => svc.get())).toBe(1)
expect(await runTwo((svc) => svc.get())).toBe(1)