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

fix(plugin): reload late SDK plugins (#35576)

Browse source
This commit is contained in:
Kit Langton 2026-07-06 13:36:19 -04:00 • committed by GitHub
commit 2830176972
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 1187 additions and 922 deletions
Download patch file Download diff file Expand all files Collapse all files

27
packages/core/src/event.ts
View file

@ -111,13 +111,26 @@ export type LogItem = Payload | EventLog.Synced
export const isSynced = (item: LogItem): item is EventLog.Synced => item.type === "log.synced"
export type SubscribePayload<D extends readonly Definition[]> = D[number] extends infer Item
? Item extends Definition
? Payload<Item>
: never
: never
export interface Subscribe {
<D extends Definition>(definition: D): Stream.Stream<Payload<D>>
<const D extends readonly [Definition, ...Definition[]]>(definitions: D): Stream.Stream<SubscribePayload<D>>
}
const isDefinition = (input: Definition | readonly Definition[]): input is Definition => !Array.isArray(input)
export interface Interface {
readonly publish: <D extends Definition>(
definition: D,
data: Data<D>,
options?: PublishOptions,
) => Effect.Effect<Payload<D>>
readonly subscribe: <D extends Definition>(definition: D) => Stream.Stream<Payload<D>>
readonly subscribe: Subscribe
/**
* Volatile live channel: every event published from now on, nothing before,
* nothing across a disconnect. The only channel that carries non-durable
@ -562,11 +575,21 @@ export const layerWith = (options?: LayerOptions) =>
),
)
const subscribe = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
const subscribeOne = <D extends Definition>(definition: D): Stream.Stream<Payload<D>> =>
local(Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub))))).pipe(
Stream.map((event) => event as Payload<D>),
)
function subscribe<D extends Definition>(definition: D): Stream.Stream<Payload<D>>
function subscribe<const D extends readonly [Definition, ...Definition[]]>(
definitions: D,
): Stream.Stream<SubscribePayload<D>>
function subscribe(input: Definition | readonly Definition[]): Stream.Stream<Payload> {
if (isDefinition(input)) return subscribeOne(input)
const types = new Set(input.map((definition) => definition.type))
return streamLive().pipe(Stream.filter((event) => types.has(event.type)))
}
const streamLive = (): Stream.Stream<Payload> => local(Stream.fromPubSub(pubsub.live))
const readAfter = (