openAgent

1export * as PluginV2 from "./plugin"
2 
3import { makeLocationNode } from "./effect/app-node"
4import { Context, Deferred, Effect, Exit, Layer, Scope } from "effect"
5import type { Plugin as PluginRuntime } from "@opencode-ai/plugin/v2/effect"
6import { Plugin } from "@opencode-ai/schema/plugin"
7import { AgentV2 } from "./agent"
8import { AISDK } from "./aisdk"
9import { Catalog } from "./catalog"
10import { CommandV2 } from "./command"
11import { EventV2 } from "./event"
12import { Integration } from "./integration"
13import { KeyedMutex } from "./effect/keyed-mutex"
14import { PluginHost } from "./plugin/host"
15import { Reference } from "./reference"
16import { SkillV2 } from "./skill"
17import { State } from "./state"
18 
19export const ID = Plugin.ID
20export type ID = typeof ID.Type
21export const Event = Plugin.Event
22 
23export interface Interface {
24  readonly add: (id: ID, effect: PluginRuntime["effect"]) => Effect.Effect<void>
25  readonly remove: (id: ID) => Effect.Effect<void>
26  readonly wait: (id: ID) => Effect.Effect<void>
27}
28 
29export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Plugin") {}
30 
31const layer = Layer.effect(
32  Service,
33  Effect.gen(function* () {
34    const events = yield* EventV2.Service
35    const locks = KeyedMutex.makeUnsafe<ID>()
36    const scope = yield* Scope.make()
37    const active = new Map<ID, Scope.Closeable>()
38    const loading = new Set<ID>()
39    const waiters = new Map<ID, Set<Deferred.Deferred<void>>>()
40    const failures = new Map<ID, Exit.Exit<void, never>>()
41    let host: Parameters<PluginRuntime["effect"]>[0]
42 
43    const add = Effect.fn("Plugin.add")(function* (id: ID, effect: PluginRuntime["effect"]) {
44      if (loading.has(id)) return yield* Effect.die(`Plugin load cycle detected for ${id}`)
45 
46      yield* locks.withLock(id)(
47        Effect.sync(() => {
48          loading.add(id)
49          failures.delete(id)
50        }).pipe(
51          Effect.andThen(
52            State.batch(
53              Effect.gen(function* () {
54                const existing = active.get(id)
55                active.delete(id)
56                if (existing) yield* Scope.close(existing, Exit.void).pipe(Effect.ignore)
57 
58                const child = yield* Scope.fork(scope)
59                yield* effect(host).pipe(
60                  Scope.provide(child),
61                  Effect.withSpan("Plugin.load", { attributes: { "plugin.id": id } }),
62                  Effect.onExit((exit) => (Exit.isFailure(exit) ? Scope.close(child, exit) : Effect.void)),
63                )
64                yield* events.publish(Event.Added, { id })
65                active.set(id, child)
66                yield* Effect.forEach(waiters.get(id) ?? [], (waiter) => Deferred.succeed(waiter, undefined), {
67                  discard: true,
68                })
69                waiters.delete(id)
70              }),
71            ),
72          ),
73          Effect.onExit((exit) => {
74            if (Exit.isSuccess(exit)) return Effect.void
75            failures.set(id, exit)
76            return Effect.forEach(waiters.get(id) ?? [], (waiter) => Deferred.done(waiter, exit), {
77              discard: true,
78            }).pipe(Effect.ensuring(Effect.sync(() => waiters.delete(id))))
79          }),
80          Effect.ensuring(Effect.sync(() => loading.delete(id))),
81        ),
82      )
83    })
84 
85    const remove = Effect.fn("Plugin.remove")(function* (id: ID) {
86      if (loading.has(id)) return yield* Effect.die(`Cannot remove plugin ${id} while it is loading`)
87 
88      yield* locks.withLock(id)(
89        State.batch(
90          Effect.gen(function* () {
91            const current = active.get(id)
92            active.delete(id)
93            failures.delete(id)
94            if (current) yield* Scope.close(current, Exit.void).pipe(Effect.ignore)
95          }),
96        ),
97      )
98    })
99 
100    const wait = Effect.fn("Plugin.wait")(function* (id: ID) {
101      const waiter = yield* Deferred.make<void>()
102      const pending = yield* locks.withLock(id)(
103        Effect.sync(() => {
104          if (active.has(id)) return false
105          const failure = failures.get(id)
106          if (failure) return failure
107          const current = waiters.get(id) ?? new Set()
108          current.add(waiter)
109          waiters.set(id, current)
110          return true
111        }),
112      )
113      if (!pending) return
114      if (typeof pending !== "boolean") return yield* pending
115      yield* Deferred.await(waiter).pipe(
116        Effect.ensuring(
117          locks.withLock(id)(
118            Effect.sync(() => {
119              const current = waiters.get(id)
120              current?.delete(waiter)
121              if (current?.size === 0) waiters.delete(id)
122            }),
123          ),
124        ),
125      )
126    })
127 
128    yield* Effect.addFinalizer((exit) =>
129      Effect.gen(function* () {
130        active.clear()
131        yield* State.batch(Scope.close(scope, exit))
132      }),
133    )
134 
135    const service = Service.of({
136      add,
137      remove,
138      wait,
139    })
140    host = yield* PluginHost.make(service)
141    return service
142  }),
143)
144 
145export const locationLayer = layer.pipe(
146  Layer.provideMerge(AgentV2.locationLayer),
147  Layer.provideMerge(AISDK.locationLayer),
148  Layer.provideMerge(Catalog.locationLayer),
149  Layer.provideMerge(CommandV2.locationLayer),
150  Layer.provideMerge(Integration.locationLayer),
151  Layer.provideMerge(Reference.locationLayer),
152)
153 
154export const node = makeLocationNode({
155  service: Service,
156  layer,
157  deps: [
158    EventV2.node,
159    AgentV2.node,
160    AISDK.node,
161    Catalog.node,
162    CommandV2.node,
163    Integration.node,
164    Reference.node,
165    SkillV2.node,
166  ],
167})
168