feat(core): add location-scoped operation hooks

.changeset/v2-operation-hooks.md

@@ -0,0 +1,5 @@
+---
+"opencode": minor
+---
+
+Add Location-scoped V2 operation hooks with waterfall, serial, and parallel dispatch, deterministic plugin ordering, scoped disposal, and real operation payloads. Migrate agent, command, compaction, permission, text, tool, and turn lifecycle hooks off the EventV2 placeholder dispatch.

packages/core/src/config/plugin/external.ts

@@ -86,6 +86,6 @@ export const Plugin = define({
           })
         }).pipe(Effect.ignoreCause)
       }
-    }).pipe(Effect.forkScoped({ startImmediately: true }))
+    })
   }),
 })

packages/core/src/event.ts

@@ -145,36 +145,6 @@ export interface Interface {
   ) => Effect.Effect<string | undefined>
   readonly remove: (aggregateID: string) => Effect.Effect<void>
   readonly claim: (aggregateID: string, ownerID: string) => Effect.Effect<void>
-  /**
-   * Waterfall dispatch: listeners run sequentially in registration order. Each
-   * listener receives the event and a `next()` callback that delegates to the
-   * downstream listener. Skipping `next()` short-circuits and the listener's
-   * return value replaces whatever `next` would have produced. Mirrors the
-   * dsh `ctx.waterfall` semantics and is the right shape for `tool.execute.before`,
-   * `command.execute.before`, and `experimental.chat.messages.transform`.
-   */
-  readonly waterfall: <D extends Definition, R>(
-    definition: D,
-    listeners: ReadonlyArray<(event: Payload<D>, next: () => Effect.Effect<R>) => Effect.Effect<R>>,
-  ) => Effect.Effect<R>
-  /**
-   * Serial dispatch: listeners run sequentially in registration order, each
-   * awaits the previous, and the dispatch collects every listener's value. Use
-   * when every listener must run (decision aggregation, telemetry stages).
-   */
-  readonly serial: <D extends Definition, R>(
-    definition: D,
-    listeners: ReadonlyArray<(event: Payload<D>) => Effect.Effect<R>>,
-  ) => Effect.Effect<ReadonlyArray<R>>
-  /**
-   * Parallel dispatch: listeners run concurrently with `Effect.forEach`'s
-   * concurrency=unbounded. Returns every listener's value, never short-circuits.
-   * Use for observation-only subscribers that must not block each other.
-   */
-  readonly parallel: <D extends Definition, R>(
-    definition: D,
-    listeners: ReadonlyArray<(event: Payload<D>) => Effect.Effect<R>>,
-  ) => Effect.Effect<ReadonlyArray<R>>
 }
 
 export class Service extends Context.Service<Service, Interface>()("@opencode/Event") {}
@@ -649,55 +619,6 @@ export const layerWith = (options?: LayerOptions) =>
           projectors.set(definition.type, list)
         })
 
-      /**
-       * `waterfall`/`serial`/`parallel` don't take a published event; the
-       * listeners synthesise the shape they need. We pass a placeholder with
-       * the right `type` so `Payload<D>` typechecks; listeners that read fields
-       * are expected to gate themselves on the dispatch they subscribed to.
-       */
-      const placeholderEvent = <D extends Definition>(definition: D): Payload<D> =>
-        ({ id: ID.create(), type: definition.type, data: undefined }) as Payload<D>
-
-      const waterfall = <D extends Definition, R>(
-        _definition: D,
-        listeners: ReadonlyArray<(event: Payload<D>, next: () => Effect.Effect<R>) => Effect.Effect<R>>,
-      ): Effect.Effect<R> => {
-        if (listeners.length === 0) {
-          return Effect.die(new Error(`waterfall called with zero listeners`))
-        }
-        const head = listeners[0]!
-        const tail = listeners.slice(1)
-        return Effect.suspend(() => {
-          const event = placeholderEvent<D>(_definition)
-          return head(event, () =>
-            tail.length === 0
-              ? Effect.succeed(event as unknown as R)
-              : waterfall(_definition, tail),
-          )
-        })
-      }
-
-      const serial = <D extends Definition, R>(
-        definition: D,
-        listeners: ReadonlyArray<(event: Payload<D>) => Effect.Effect<R>>,
-      ): Effect.Effect<ReadonlyArray<R>> =>
-        Effect.suspend(() => {
-          const event = placeholderEvent<D>(definition)
-          return Effect.forEach(listeners, (listener) => listener(event))
-        })
-
-      const parallel = <D extends Definition, R>(
-        definition: D,
-        listeners: ReadonlyArray<(event: Payload<D>) => Effect.Effect<R>>,
-      ): Effect.Effect<ReadonlyArray<R>> =>
-        Effect.suspend(() => {
-          const event = placeholderEvent<D>(definition)
-          return Effect.forEach(listeners, (listener) => listener(event), {
-            concurrency: "unbounded",
-            discard: false,
-          })
-        })
-
       return Service.of({
         publish,
         subscribe,
@@ -706,9 +627,6 @@ export const layerWith = (options?: LayerOptions) =>
         listen,
         project,
         replay,
-        waterfall,
-        serial,
-        parallel,
         replayAll,
         remove,
         claim,

packages/core/src/location-services.ts

@@ -16,6 +16,7 @@ import { Integration } from "./integration"
 import { RuntimeInvariant } from "./invariant"
 import { Location } from "./location"
 import { LocationMutation } from "./location-mutation"
+import { OperationHook } from "./operation-hook"
 import { LocationServiceMap } from "./location-service-map"
 import { PermissionV2 } from "./permission"
 import { PluginV2 } from "./plugin"
@@ -72,6 +73,7 @@ export const locationServices = LayerNode.group([
   SystemContextRegistry.node,
   SystemContextBuiltIns.node,
   LocationMutation.node,
+  OperationHook.node,
   FileMutation.node,
   PermissionV2.node,
   ToolOutputStore.node,

packages/core/src/operation-hook.ts

@@ -0,0 +1,173 @@
+export * as OperationHook from "./operation-hook"
+
+import {
+  Definition,
+  Operation,
+  type Data,
+  type Hooks,
+  type Observer,
+  type ParallelDefinition,
+  type Payload,
+  type SerialDefinition,
+  type WaterfallDefinition,
+  type WaterfallHandler,
+} from "@opencode-ai/plugin/v2/effect/operation-hook"
+import { Context, Effect, Layer, Scope } from "effect"
+import { makeLocationNode } from "./effect/app-node"
+import { EventV2 } from "./event"
+import { Location } from "./location"
+
+export { Definition, Operation }
+
+export interface Interface {
+  readonly register: Hooks
+  readonly waterfall: <D extends WaterfallDefinition>(definition: D, data: Data<D>) => Effect.Effect<Data<D>>
+  readonly serial: <D extends SerialDefinition>(definition: D, data: Data<D>) => Effect.Effect<void>
+  readonly parallel: <D extends ParallelDefinition>(definition: D, data: Data<D>) => Effect.Effect<void>
+}
+
+export class Service extends Context.Service<Service, Interface>()("@opencode/v2/OperationHook") {}
+
+type InternalPayload = {
+  readonly id: string
+  readonly type: string
+  readonly location: Location.Ref
+  readonly data: unknown
+}
+
+type WaterfallEntry = {
+  readonly run: (event: InternalPayload, next: (data?: unknown) => Effect.Effect<unknown>) => Effect.Effect<unknown>
+}
+
+type ObserverEntry = {
+  readonly run: (event: InternalPayload) => Effect.Effect<void>
+}
+
+export const locationLayer = Layer.effect(
+  Service,
+  Effect.gen(function* () {
+    const location = yield* Location.Service
+    const waterfallEntries = new Map<string, WaterfallEntry[]>()
+    const serialEntries = new Map<string, ObserverEntry[]>()
+    const parallelEntries = new Map<string, ObserverEntry[]>()
+
+    const add = <Entry>(entries: Map<string, Entry[]>, type: string, entry: Entry) =>
+      Effect.gen(function* () {
+        const scope = yield* Scope.Scope
+        let active = true
+        const dispose = Effect.uninterruptible(
+          Effect.sync(() => {
+            if (!active) return
+            active = false
+            entries.set(type, (entries.get(type) ?? []).filter((item) => item !== entry))
+            if (entries.get(type)?.length === 0) entries.delete(type)
+          }),
+        )
+
+        yield* Effect.uninterruptible(
+          Effect.sync(() => entries.set(type, [...(entries.get(type) ?? []), entry])).pipe(
+            Effect.andThen(Scope.addFinalizer(scope, dispose)),
+          ),
+        )
+        return { dispose }
+      })
+
+    const waterfall = <D extends WaterfallDefinition>(definition: D, data: Data<D>) =>
+      Effect.gen(function* () {
+        requireMode(definition, "waterfall")
+        const entries = [...(waterfallEntries.get(definition.type) ?? [])]
+        const base = {
+          id: EventV2.ID.create(),
+          type: definition.type,
+          location,
+        }
+        const run = (index: number, current: unknown): Effect.Effect<unknown> => {
+          const entry = entries[index]
+          if (!entry) return Effect.succeed(current)
+          return entry.run({ ...base, data: current }, (next = current) => run(index + 1, next))
+        }
+        return (yield* run(0, data)) as Data<D>
+      })
+
+    const observe = (event: InternalPayload, entry: ObserverEntry) =>
+      entry.run(event).pipe(
+        Effect.catchCause((cause) =>
+          Effect.logError("Operation hook failed", { eventID: event.id, eventType: event.type, cause }),
+        ),
+      )
+
+    const dispatchObservers = <D extends SerialDefinition | ParallelDefinition>(
+      mode: "serial" | "parallel",
+      entries: Map<string, ObserverEntry[]>,
+      definition: D,
+      data: Data<D>,
+    ) =>
+      Effect.gen(function* () {
+        requireMode(definition, mode)
+        const event = {
+          id: EventV2.ID.create(),
+          type: definition.type,
+          location,
+          data,
+        }
+        yield* Effect.forEach(entries.get(definition.type) ?? [], (entry) => observe(event, entry), {
+          concurrency: mode === "parallel" ? "unbounded" : 1,
+          discard: true,
+        })
+      })
+
+    const registerWaterfall = <D extends WaterfallDefinition>(definition: D, callback: WaterfallHandler<D>) => {
+      requireMode(definition, "waterfall")
+      const entry: WaterfallEntry = {
+        run(event, next) {
+          if (event.type !== definition.type) return Effect.die(new Error(`Operation hook type mismatch`))
+          const payload = { ...event, type: definition.type, data: event.data as Data<D> } satisfies Payload<D>
+          return Effect.suspend(() => {
+            const result = callback(payload, (data) => next(data).pipe(Effect.map((value) => value as Data<D>)))
+            return Effect.isEffect(result) ? result : Effect.succeed(result)
+          })
+        },
+      }
+      return add(waterfallEntries, definition.type, entry)
+    }
+
+    const registerObserver = <D extends SerialDefinition | ParallelDefinition>(
+      mode: "serial" | "parallel",
+      entries: Map<string, ObserverEntry[]>,
+      definition: D,
+      callback: Observer<D>,
+    ) => {
+      requireMode(definition, mode)
+      const entry: ObserverEntry = {
+        run(event) {
+          if (event.type !== definition.type) return Effect.die(new Error(`Operation hook type mismatch`))
+          const payload = { ...event, type: definition.type, data: event.data as Data<D> } satisfies Payload<D>
+          return Effect.suspend(() => {
+            const result = callback(payload)
+            return Effect.isEffect(result) ? result : Effect.void
+          })
+        },
+      }
+      return add(entries, definition.type, entry)
+    }
+
+    return Service.of({
+      register: {
+        waterfall: registerWaterfall,
+        serial: (definition, callback) => registerObserver("serial", serialEntries, definition, callback),
+        parallel: (definition, callback) => registerObserver("parallel", parallelEntries, definition, callback),
+      },
+      waterfall,
+      serial: (definition, data) => dispatchObservers("serial", serialEntries, definition, data),
+      parallel: (definition, data) => dispatchObservers("parallel", parallelEntries, definition, data),
+    })
+  }),
+)
+
+function requireMode(definition: Definition, mode: Definition["mode"]) {
+  if (definition.mode !== mode) {
+    throw new Error(`Operation hook ${definition.type} uses ${definition.mode}, not ${mode}`)
+  }
+}
+
+export const node = makeLocationNode({ service: Service, layer: locationLayer, deps: [Location.node] })

packages/core/src/plugin.ts

@@ -11,6 +11,7 @@ import { Catalog } from "./catalog"
 import { CommandV2 } from "./command"
 import { EventV2 } from "./event"
 import { Integration } from "./integration"
+import { OperationHook } from "./operation-hook"
 import { KeyedMutex } from "./effect/keyed-mutex"
 import { PluginHost } from "./plugin/host"
 import { Reference } from "./reference"
@@ -180,6 +181,7 @@ export const locationLayer = layer.pipe(
   Layer.provideMerge(Catalog.locationLayer),
   Layer.provideMerge(CommandV2.locationLayer),
   Layer.provideMerge(Integration.locationLayer),
+  Layer.provideMerge(OperationHook.locationLayer),
   Layer.provideMerge(Reference.locationLayer),
 )
 
@@ -195,6 +197,7 @@ export const node = makeLocationNode({
     Catalog.node,
     CommandV2.node,
     Integration.node,
+    OperationHook.node,
     Reference.node,
     SkillV2.node,
   ],

packages/core/src/plugin/host.ts

@@ -10,6 +10,7 @@ import { CommandV2 } from "../command"
 import { Credential } from "../credential"
 import { Integration } from "../integration"
 import { ModelV2 } from "../model"
+import { OperationHook } from "../operation-hook"
 import { PluginV2 } from "../plugin"
 import { ProviderV2 } from "../provider"
 import { Reference } from "../reference"
@@ -25,6 +26,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: PluginV2.Int
   const catalog = yield* Catalog.Service
   const commands = yield* CommandV2.Service
   const integration = yield* Integration.Service
+  const operationHooks = yield* OperationHook.Service
   const reference = yield* Reference.Service
   const skill = yield* SkillV2.Service
 
@@ -254,6 +256,7 @@ export const make = Effect.fn("PluginHost.make")(function* (plugin: PluginV2.Int
           }),
         ),
     },
+    hook: operationHooks.register,
     plugin: {
       add: (input) => plugin.add(PluginV2.ID.make(input.id), input.effect),
       remove: (id) => plugin.remove(PluginV2.ID.make(id)),

packages/core/src/plugin/promise.ts

@@ -70,6 +70,29 @@ export function fromPromise(plugin: Plugin) {
               resolve: (connection) => Effect.runPromiseWith(context)(host.integration.connection.resolve(connection)),
             },
           },
+          hook: {
+            waterfall: (definition, callback) =>
+              register(
+                host.hook.waterfall(definition, (event, next) =>
+                  Effect.gen(function* () {
+                    const dispatchContext = yield* Effect.context<never>()
+                    return yield* Effect.promise((signal) =>
+                      Promise.resolve(
+                        callback(event, (data) => Effect.runPromiseWith(dispatchContext)(next(data), { signal })),
+                      ),
+                    )
+                  }),
+                ),
+              ),
+            serial: (definition, callback) =>
+              register(
+                host.hook.serial(definition, (event) => Effect.promise((_signal) => Promise.resolve(callback(event)))),
+              ),
+            parallel: (definition, callback) =>
+              register(
+                host.hook.parallel(definition, (event) => Effect.promise((_signal) => Promise.resolve(callback(event)))),
+              ),
+          },
           plugin: {
             add: (input) => {
               const child = fromPromise(input)