Compare commits

...

4 Commits

Author SHA1 Message Date
Dax Raad 668cd41ac5 chore: regenerate sdk 2026-05-13 22:23:28 -04:00
Dax Raad dae6598510 Merge remote-tracking branch 'origin/dev' into core-event-v2-legacy-bridge 2026-05-13 22:14:36 -04:00
Dax Raad 4f8b7bb217 refactor(core): expose event registry 2026-05-13 21:59:03 -04:00
Dax Raad 688ba45a66 feat(core): add effect event system 2026-05-13 21:27:55 -04:00
11 changed files with 619 additions and 129 deletions
+15 -6
View File
@@ -6,6 +6,7 @@ import { ModelV2 } from "./model"
import { PluginV2 } from "./plugin"
import { ProviderV2 } from "./provider"
import { Instance } from "./instance"
import { Event as EventV2 } from "./event"
type ProviderRecord = {
provider: ProviderV2.Info
@@ -24,6 +25,15 @@ export class ModelNotFoundError extends Schema.TaggedErrorClass<ModelNotFoundErr
modelID: ModelV2.ID,
}) {}
export const Event = {
ModelUpdated: EventV2.define({
type: "catalog.model.updated",
schema: {
model: ModelV2.Info,
},
}),
}
export interface Interface {
readonly provider: {
readonly get: (providerID: ProviderV2.ID) => Effect.Effect<ProviderV2.Info, ProviderNotFoundError>
@@ -61,6 +71,7 @@ export const layer = Layer.effect(
let records = HashMap.empty<ProviderV2.ID, ProviderRecord>()
let defaultModel: { providerID: ProviderV2.ID; modelID: ModelV2.ID } | undefined
const plugin = yield* PluginV2.Service
const events = yield* EventV2.Service
const resolve = (model: ModelV2.Info) => {
const provider = Option.getOrThrow(HashMap.get(records, model.providerID)).provider
@@ -157,14 +168,12 @@ export const layer = Layer.effect(
)
const updated = yield* plugin.trigger("model.update", {}, { model, cancel: false })
if (updated.cancel) return
const next = new ModelV2.Info({ ...updated.model, id: modelID, providerID })
records = HashMap.set(records, providerID, {
provider: record.provider,
models: HashMap.set(
record.models,
modelID,
new ModelV2.Info({ ...updated.model, id: modelID, providerID }),
),
models: HashMap.set(record.models, modelID, next),
})
yield* events.publish(Event.ModelUpdated, { model: resolve(next) })
return
}),
@@ -257,4 +266,4 @@ export const layer = Layer.effect(
const SMALL_MODEL_RE = /\b(nano|flash|lite|mini|haiku|small|fast)\b/
export const defaultLayer = layer.pipe(Layer.provide(PluginV2.defaultLayer))
export const defaultLayer = layer.pipe(Layer.provideMerge(EventV2.defaultLayer), Layer.provide(PluginV2.defaultLayer))
+148
View File
@@ -0,0 +1,148 @@
import { Context, Effect, Layer, Option, PubSub, Schema, Stream } from "effect"
import { Instance } from "./instance"
import { withStatics } from "./schema"
import { Identifier } from "./util/identifier"
export const ID = Schema.String.pipe(
Schema.brand("Event.ID"),
withStatics((schema) => ({ create: () => schema.make("evt_" + Identifier.ascending()) })),
)
export type ID = typeof ID.Type
export const InstanceRef = Schema.Struct({
directory: Schema.String,
workspaceID: Schema.optional(Schema.String),
}).annotate({ identifier: "Event.Instance" })
export type InstanceRef = Instance.Ref
export type Definition<Type extends string = string, Data = unknown> = Schema.Schema<{
readonly id: ID
readonly metadata?: Record<string, unknown>
readonly type: Type
readonly version?: number
readonly instance?: InstanceRef
readonly data: Data
}> & {
readonly type: Type
readonly version?: number
readonly schema: Schema.Schema<Data>
}
export type Payload<D extends Definition = Definition> = Schema.Schema.Type<D>
export type Data<D extends Definition> = Payload<D>["data"]
export const registry = new Map<string, Definition>()
export function define<const Type extends string, Fields extends Schema.Struct.Fields>(input: {
readonly type: Type
readonly version?: number
readonly schema: Fields
}): Definition<Type, Schema.Schema.Type<Schema.Struct<Fields>>> {
const Data = Schema.Struct(input.schema)
const Payload = Schema.Struct({
id: ID,
metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
type: Schema.Literal(input.type),
version: Schema.optional(Schema.Number),
instance: Schema.optional(InstanceRef),
data: Data,
}).annotate({ identifier: input.type })
const definition = Object.assign(Payload, {
type: input.type,
...(input.version === undefined ? {} : { version: input.version }),
schema: Data,
})
registry.set(input.type, definition)
return definition as Definition<Type, Schema.Schema.Type<Schema.Struct<Fields>>>
}
export function definitions() {
return registry.values().toArray()
}
export interface PublishOptions<D extends Definition> {
readonly id?: ID
readonly metadata?: Record<string, unknown>
readonly instance?: InstanceRef | false
}
export interface Interface {
readonly publish: <D extends Definition>(
definition: D,
data: Data<D>,
options?: PublishOptions<D>,
) => Effect.Effect<Payload<D>>
readonly publishEvent: <D extends Definition>(event: Payload<D>) => Effect.Effect<Payload<D>>
readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
readonly subscribeAll: () => Stream.Stream<Payload>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/Event") {}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const all = yield* PubSub.unbounded<Payload>()
const typed = new Map<string, PubSub.PubSub<Payload>>()
const getOrCreate = (definition: Definition) =>
Effect.gen(function* () {
const existing = typed.get(definition.type)
if (existing) return existing
const pubsub = yield* PubSub.unbounded<Payload>()
typed.set(definition.type, pubsub)
return pubsub
})
yield* Effect.addFinalizer(() =>
Effect.gen(function* () {
yield* PubSub.shutdown(all)
yield* Effect.forEach(typed.values(), PubSub.shutdown, { discard: true })
}),
)
function publishEvent<D extends Definition>(event: Payload<D>) {
return Effect.gen(function* () {
const pubsub = typed.get(event.type)
if (pubsub) yield* PubSub.publish(pubsub, event as Payload)
yield* PubSub.publish(all, event as Payload)
return event
})
}
function publish<D extends Definition>(
definition: D,
data: Data<D>,
options?: PublishOptions<D>,
) {
return Effect.gen(function* () {
const instance = options?.instance === false
? undefined
: (options?.instance ?? Option.getOrUndefined(yield* Effect.serviceOption(Instance.Service)))
const event = {
id: options?.id ?? ID.create(),
...(options?.metadata ? { metadata: options.metadata } : {}),
type: definition.type,
...(definition.version === undefined ? {} : { version: definition.version }),
...(instance ? { instance } : {}),
data,
} as Payload<D>
return yield* publishEvent(event)
})
}
const subscribe = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe(
Stream.map((event) => event as Payload<D>),
)
const subscribeAll = (): Stream.Stream<Payload> => Stream.fromPubSub(all)
return Service.of({ publish, publishEvent, subscribe, subscribeAll })
}),
)
export const defaultLayer = layer
export * as Event from "./event"
+34 -2
View File
@@ -1,6 +1,7 @@
import { describe, expect } from "bun:test"
import { DateTime, Effect, Layer, Option } from "effect"
import { DateTime, Effect, Fiber, Layer, Option, Stream } from "effect"
import { Catalog } from "@opencode-ai/core/catalog"
import { Event } from "@opencode-ai/core/event"
import { Instance } from "@opencode-ai/core/instance"
import { ModelV2 } from "@opencode-ai/core/model"
import { PluginV2 } from "@opencode-ai/core/plugin"
@@ -8,7 +9,13 @@ import { ProviderV2 } from "@opencode-ai/core/provider"
import { testEffect } from "./lib/effect"
const instanceLayer = Layer.succeed(Instance.Service, Instance.Service.of({ directory: "test" }))
const it = testEffect(Catalog.layer.pipe(Layer.provideMerge(PluginV2.defaultLayer), Layer.provide(instanceLayer)))
const it = testEffect(
Catalog.layer.pipe(
Layer.provideMerge(Event.defaultLayer),
Layer.provideMerge(PluginV2.defaultLayer),
Layer.provideMerge(instanceLayer),
),
)
describe("CatalogV2", () => {
it.effect("normalizes provider baseURL into endpoint url", () =>
@@ -69,6 +76,31 @@ describe("CatalogV2", () => {
}),
)
it.effect("publishes model updated events", () =>
Effect.gen(function* () {
const catalog = yield* Catalog.Service
const events = yield* Event.Service
const providerID = ProviderV2.ID.make("test")
const modelID = ModelV2.ID.make("model")
const fiber = yield* events
.subscribe(Catalog.Event.ModelUpdated)
.pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
yield* catalog.provider.update(providerID, () => {})
yield* catalog.model.update(providerID, modelID, (model) => {
model.name = "Updated Model"
})
const event = Array.from(yield* Fiber.join(fiber))[0]
expect(event?.type).toBe("catalog.model.updated")
expect(event?.data.model.providerID).toBe(providerID)
expect(event?.data.model.id).toBe(modelID)
expect(event?.data.model.name).toBe("Updated Model")
expect(event?.instance).toEqual({ directory: "test" })
}),
)
it.effect("resolves unknown model endpoint from provider endpoint", () =>
Effect.gen(function* () {
const catalog = yield* Catalog.Service
+101
View File
@@ -0,0 +1,101 @@
import { describe, expect } from "bun:test"
import { Effect, Fiber, Layer, Schema, Stream } from "effect"
import { Event } from "@opencode-ai/core/event"
import { Instance } from "@opencode-ai/core/instance"
import { testEffect } from "./lib/effect"
const instanceLayer = Layer.succeed(
Instance.Service,
Instance.Service.of({ directory: "project", workspaceID: "workspace" }),
)
const it = testEffect(Event.layer.pipe(Layer.provideMerge(instanceLayer)))
const itWithoutInstance = testEffect(Event.layer)
const Message = Event.define({
type: "test.message",
schema: {
text: Schema.String,
},
})
const GlobalMessage = Event.define({
type: "test.global",
schema: {
text: Schema.String,
},
})
const VersionedMessage = Event.define({
type: "test.versioned",
version: 2,
schema: {
text: Schema.String,
},
})
describe("Event", () => {
it.effect("publishes events with the current instance", () =>
Effect.gen(function* () {
const events = yield* Event.Service
const fiber = yield* events.subscribe(Message).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
const event = yield* events.publish(Message, { text: "hello" })
const received = Array.from(yield* Fiber.join(fiber))
expect(received).toEqual([event])
expect(event.type).toBe("test.message")
expect(event).not.toHaveProperty("version")
expect(event.data).toEqual({ text: "hello" })
expect(event.instance).toEqual({ directory: "project", workspaceID: "workspace" })
}),
)
itWithoutInstance.effect("omits instance when no instance is available", () =>
Effect.gen(function* () {
const events = yield* Event.Service
const event = yield* events.publish(GlobalMessage, { text: "hello" })
expect(event).not.toHaveProperty("instance")
expect(event.type).toBe("test.global")
}),
)
it.effect("can publish globally from an instance context", () =>
Effect.gen(function* () {
const events = yield* Event.Service
const event = yield* events.publish(GlobalMessage, { text: "hello" }, { instance: false })
expect(event).not.toHaveProperty("instance")
expect(event.type).toBe("test.global")
}),
)
it.effect("publishes definition version", () =>
Effect.gen(function* () {
const events = yield* Event.Service
const event = yield* events.publish(VersionedMessage, { text: "hello" })
expect(event.type).toBe("test.versioned")
expect(event.version).toBe(2)
}),
)
it.effect("stores definitions in the exported registry", () =>
Effect.sync(() => {
expect(Event.registry.get(Message.type)).toBe(Message)
}),
)
it.effect("publishes to typed and wildcard subscriptions", () =>
Effect.gen(function* () {
const events = yield* Event.Service
const typed = yield* events.subscribe(Message).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
const wildcard = yield* events.subscribeAll().pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
yield* Effect.yieldNow
const event = yield* events.publish(Message, { text: "hello" })
expect(Array.from(yield* Fiber.join(typed))).toEqual([event])
expect(Array.from(yield* Fiber.join(wildcard))).toEqual([event])
}),
)
})
+24 -10
View File
@@ -1,4 +1,5 @@
import { Schema } from "effect"
import { Event } from "@opencode-ai/core/event"
export type Definition<Type extends string = string, Properties extends Schema.Top = Schema.Top> = {
type: Type
@@ -17,16 +18,29 @@ export function define<Type extends string, Properties extends Schema.Top>(
}
export function effectPayloads() {
return registry
.entries()
.map(([type, def]) =>
Schema.Struct({
id: Schema.String,
type: Schema.Literal(type),
properties: def.properties,
}).annotate({ identifier: `Event.${type}` }),
)
.toArray()
return [
...registry
.entries()
.map(([type, def]) =>
Schema.Struct({
id: Schema.String,
type: Schema.Literal(type),
properties: def.properties,
}).annotate({ identifier: `Event.${type}` }),
)
.toArray(),
...Event.registry
.values()
.filter((definition) => definition.version === undefined)
.map((definition) =>
Schema.Struct({
id: Schema.String,
type: Schema.Literal(definition.type),
properties: definition.schema,
}).annotate({ identifier: `Event.${definition.type}` }),
)
.toArray(),
]
}
export * as BusEvent from "./bus-event"
@@ -56,6 +56,7 @@ import { Npm } from "@opencode-ai/core/npm"
import { memoMap } from "@opencode-ai/core/effect/memo-map"
import { DataMigration } from "@/data-migration"
import { BackgroundJob } from "@/background/job"
import { EventLegacy } from "@/event-legacy"
export const AppLayer = Layer.mergeAll(
Npm.defaultLayer,
@@ -109,6 +110,7 @@ export const AppLayer = Layer.mergeAll(
ShareNext.defaultLayer,
SessionShare.defaultLayer,
SyncEvent.defaultLayer,
EventLegacy.defaultLayer,
DataMigration.defaultLayer,
).pipe(Layer.provideMerge(InstanceLayer.layer), Layer.provideMerge(Observability.layer))
+56
View File
@@ -0,0 +1,56 @@
import { Bus as ProjectBus } from "@/bus"
import { GlobalBus } from "@/bus/global"
import { SyncEvent } from "@/sync"
import { Event } from "@opencode-ai/core/event"
import "@opencode-ai/core/catalog"
import { Effect, Layer, Stream } from "effect"
function emitNormal(event: Event.Payload) {
GlobalBus.emit("event", {
directory: event.instance?.directory,
workspace: event.instance?.workspaceID,
payload: {
id: event.id,
type: event.type,
properties: event.data,
},
})
}
export const layer = Layer.effectDiscard(
Effect.gen(function* () {
const events = yield* Event.Service
const bus = yield* ProjectBus.Service
yield* events.subscribeAll().pipe(Stream.runForEach(republish(bus)), Effect.forkScoped)
}),
)
export const defaultLayer = layer.pipe(Layer.provideMerge(Event.defaultLayer), Layer.provide(ProjectBus.defaultLayer))
const republish = (bus: ProjectBus.Interface) => (event: Event.Payload) => {
const definition = Event.registry.get(event.type)
if (!definition) return Effect.void
if (definition.version !== undefined) {
return Effect.sync(() => {
GlobalBus.emit("event", {
directory: event.instance?.directory,
workspace: event.instance?.workspaceID,
payload: {
type: "sync",
name: SyncEvent.versionedType(definition.type, definition.version!),
id: event.id,
seq: 0,
aggregateID: event.id,
data: event.data,
},
})
})
}
return bus.publish({ type: definition.type, properties: definition.schema }, event.data, { id: event.id }).pipe(
Effect.catch(() => Effect.sync(() => emitNormal(event))),
)
}
export * as EventLegacy from "./event-legacy"
@@ -1,6 +1,7 @@
import { Config } from "@/config/config"
import { BusEvent } from "@/bus/bus-event"
import { SyncEvent } from "@/sync"
import "@/event-legacy"
import "@/server/event"
import { Schema } from "effect"
import { HttpApi, HttpApiEndpoint, HttpApiError, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
@@ -44,6 +44,7 @@ import { SessionSummary } from "@/session/summary"
import { Todo } from "@/session/todo"
import { SessionShare } from "@/share/session"
import { ShareNext } from "@/share/share-next"
import { EventLegacy } from "@/event-legacy"
import { Skill } from "@/skill"
import { Snapshot } from "@/snapshot"
import { SyncEvent } from "@/sync"
@@ -216,6 +217,7 @@ export function createRoutes(corsOptions?: CorsOptions) {
ShareNext.defaultLayer,
Snapshot.defaultLayer,
SyncEvent.defaultLayer,
EventLegacy.defaultLayer,
Skill.defaultLayer,
Todo.defaultLayer,
ToolRegistry.defaultLayer,
+30 -13
View File
@@ -9,6 +9,7 @@ import type { WorkspaceID } from "@/control-plane/schema"
import { EventID } from "./schema"
import { Context, Effect, Layer, Schema as EffectSchema } from "effect"
import type { DeepMutable } from "@opencode-ai/core/schema"
import { Event as CoreEvent } from "@opencode-ai/core/event"
import { serviceUse } from "@/effect/service-use"
import { InstanceState } from "@/effect/instance-state"
import { RuntimeFlags } from "@/effect/runtime-flags"
@@ -355,19 +356,35 @@ function process<Def extends Definition>(
}
export function effectPayloads() {
return registry
.entries()
.map(([type, def]) =>
EffectSchema.Struct({
type: EffectSchema.Literal("sync"),
name: EffectSchema.Literal(type),
id: EffectSchema.String,
seq: EffectSchema.Finite,
aggregateID: EffectSchema.Literal(def.aggregate),
data: def.schema,
}).annotate({ identifier: `SyncEvent.${type}` }),
)
.toArray()
return [
...registry
.entries()
.map(([type, def]) =>
EffectSchema.Struct({
type: EffectSchema.Literal("sync"),
name: EffectSchema.Literal(type),
id: EffectSchema.String,
seq: EffectSchema.Finite,
aggregateID: EffectSchema.Literal(def.aggregate),
data: def.schema,
}).annotate({ identifier: `SyncEvent.${type}` }),
)
.toArray(),
...CoreEvent.registry
.values()
.filter((definition) => definition.version !== undefined)
.map((definition) =>
EffectSchema.Struct({
type: EffectSchema.Literal("sync"),
name: EffectSchema.Literal(versionedType(definition.type, definition.version!)),
id: EffectSchema.String,
seq: EffectSchema.Finite,
aggregateID: EffectSchema.String,
data: definition.schema,
}).annotate({ identifier: `SyncEvent.${definition.type}` }),
)
.toArray(),
]
}
export * as SyncEvent from "."
+206 -98
View File
@@ -77,6 +77,7 @@ export type Event =
| EventSessionNextCompactionEnded
| EventServerConnected
| EventGlobalDisposed
| EventCatalogModelUpdated
export type OAuth = {
type: "oauth"
@@ -861,6 +862,7 @@ export type GlobalEvent = {
| EventSessionNextCompactionEnded
| EventServerConnected
| EventGlobalDisposed
| EventCatalogModelUpdated
| SyncEventMessageUpdated
| SyncEventMessageRemoved
| SyncEventMessagePartUpdated
@@ -3143,6 +3145,112 @@ export type EventGlobalDisposed = {
}
}
export type ModelV2Info = {
id: string
apiID: string
providerID: string
family?: string
name: string
endpoint:
| {
type: "unknown"
}
| {
type: "openai/responses"
url: string
websocket?: boolean
}
| {
type: "openai/completions"
url: string
reasoning?:
| {
type: "reasoning_content"
}
| {
type: "reasoning_details"
}
}
| {
type: "anthropic/messages"
url: string
}
| {
type: "aisdk"
package: string
url?: string
}
capabilities: {
tools: boolean
input: Array<string>
output: Array<string>
}
options: {
headers: {
[key: string]: string
}
body: {
[key: string]: unknown
}
aisdk: {
provider: {
[key: string]: unknown
}
request: {
[key: string]: unknown
}
}
variant?: string
}
variants: Array<{
id: string
headers: {
[key: string]: string
}
body: {
[key: string]: unknown
}
aisdk: {
provider: {
[key: string]: unknown
}
request: {
[key: string]: unknown
}
}
}>
time: {
released: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN"
}
cost: Array<{
tier?: {
type: "context"
size: number
}
input: number
output: number
cache: {
read: number
write: number
}
}>
status: "alpha" | "beta" | "deprecated" | "active"
enabled: boolean
limit: {
context: number
input?: number
output: number
}
}
export type EventCatalogModelUpdated = {
id: string
type: "catalog.model.updated"
properties: {
model: ModelV2Info
}
}
export type SessionInfo = {
id: string
parentID?: string
@@ -3378,104 +3486,6 @@ export type SessionMessage =
| SessionMessageAssistant
| SessionMessageCompaction
export type ModelV2Info = {
id: string
apiID: string
providerID: string
family?: string
name: string
endpoint:
| {
type: "unknown"
}
| {
type: "openai/responses"
url: string
websocket?: boolean
}
| {
type: "openai/completions"
url: string
reasoning?:
| {
type: "reasoning_content"
}
| {
type: "reasoning_details"
}
}
| {
type: "anthropic/messages"
url: string
}
| {
type: "aisdk"
package: string
url?: string
}
capabilities: {
tools: boolean
input: Array<string>
output: Array<string>
}
options: {
headers: {
[key: string]: string
}
body: {
[key: string]: unknown
}
aisdk: {
provider: {
[key: string]: unknown
}
request: {
[key: string]: unknown
}
}
variant?: string
}
variants: Array<{
id: string
headers: {
[key: string]: string
}
body: {
[key: string]: unknown
}
aisdk: {
provider: {
[key: string]: unknown
}
request: {
[key: string]: unknown
}
}
}>
time: {
released: number | "NaN" | "Infinity" | "-Infinity" | "Infinity" | "-Infinity" | "NaN"
}
cost: Array<{
tier?: {
type: "context"
size: number
}
input: number
output: number
cache: {
read: number
write: number
}
}>
status: "alpha" | "beta" | "deprecated" | "active"
enabled: boolean
limit: {
context: number
input?: number
output: number
}
}
export type ProviderV2Info = {
id: string
name: string
@@ -3554,6 +3564,104 @@ export type EventTuiToastShow1 = {
}
}
export type ModelV2Info1 = {
id: string
apiID: string
providerID: string
family?: string
name: string
endpoint:
| {
type: "unknown"
}
| {
type: "openai/responses"
url: string
websocket?: boolean
}
| {
type: "openai/completions"
url: string
reasoning?:
| {
type: "reasoning_content"
}
| {
type: "reasoning_details"
}
}
| {
type: "anthropic/messages"
url: string
}
| {
type: "aisdk"
package: string
url?: string
}
capabilities: {
tools: boolean
input: Array<string>
output: Array<string>
}
options: {
headers: {
[key: string]: string
}
body: {
[key: string]: unknown
}
aisdk: {
provider: {
[key: string]: unknown
}
request: {
[key: string]: unknown
}
}
variant?: string
}
variants: Array<{
id: string
headers: {
[key: string]: string
}
body: {
[key: string]: unknown
}
aisdk: {
provider: {
[key: string]: unknown
}
request: {
[key: string]: unknown
}
}
}>
time: {
released: number | "NaN" | "Infinity" | "-Infinity"
}
cost: Array<{
tier?: {
type: "context"
size: number
}
input: number
output: number
cache: {
read: number
write: number
}
}>
status: "alpha" | "beta" | "deprecated" | "active"
enabled: boolean
limit: {
context: number
input?: number
output: number
}
}
export type BadRequestError = {
name: "BadRequest"
data: {