{"version":3,"file":"channel-effect.cjs","names":["channelProjection"],"sources":["../../../src/stream/projections/channel-effect.ts"],"sourcesContent":["/**\n * Side-effect counterpart to {@link channelProjection}.\n *\n * Where `channelProjection` retains a bounded *buffer* of raw events\n * for rendering, this helper delivers each event to a callback exactly\n * once — the idiomatic shape for analytics, logging, and other side\n * effects that should not live in render. Framework bindings wrap it as\n * `useChannelEffect` (React/Svelte/Vue) / `injectChannelEffect`\n * (Angular).\n *\n * # Why this is built on the registry\n *\n * It acquires the very same ref-counted {@link channelProjection} entry\n * that the selector hooks use, so:\n *\n *  - N consumers of the same `channels`/`namespace`/options share a\n *    single server subscription (the registry dedupes by `spec.key`).\n *  - The subscription survives thread swaps for free (the registry\n *    rebinds every live entry on `controller.hydrate(...)`).\n *\n * Each consumer then diffs the shared store independently, so multiple\n * `useChannelEffect` callers with *different* callbacks each receive\n * their own delivery.\n *\n * # Delivery semantics\n *\n *  - Only events that arrive while the effect is attached are\n *    delivered. Events already buffered when the effect attaches (e.g.\n *    a sibling `useChannel` populated the buffer first) are skipped —\n *    analytics must not double-count history on a late mount.\n *  - `channelProjection` appends exactly one event per store\n *    notification, so the cursor below tracks the last delivered event\n *    by identity and forwards everything after it. On a thread rebind\n *    the store resets to `[]`; the old cursor is no longer found, so\n *    the next batch of events for the new thread is delivered from the\n *    start.\n *  - `replay` is forwarded to the projection. It defaults to `false`\n *    here (live-only) — the opposite of `useChannel`'s buffer default —\n *    because firing analytics for replayed history is rarely what a\n *    side-effect consumer wants.\n */\nimport type { Channel, Event } from \"@langchain/protocol\";\nimport type { ChannelRegistry } from \"../channel-registry.js\";\nimport { channelProjection, type ChannelProjectionOptions } from \"./channel.js\";\n\n/**\n * Options for {@link acquireChannelEffect}. Extends\n * {@link ChannelProjectionOptions} (`bufferSize`, `replay`) with the\n * per-event callback and an optional error sink.\n */\nexport interface ChannelEffectOptions extends ChannelProjectionOptions {\n  /** Invoked once for every event observed while attached. */\n  onEvent: (event: Event) => void;\n  /**\n   * Invoked when {@link onEvent} throws. When omitted, a throwing\n   * `onEvent` is swallowed so one bad delivery cannot wedge the shared\n   * store's notification loop or other consumers.\n   */\n  onError?: (error: unknown) => void;\n}\n\n/**\n * Acquire a side-effect subscription over one or more channels.\n *\n * @param registry - The stream's {@link ChannelRegistry}.\n * @param channels - Channels to observe (e.g. `[\"lifecycle\", \"tools\"]`).\n * @param namespace - Resolved namespace (`[]` for the root).\n * @param options - Callbacks plus projection (`bufferSize`/`replay`)\n *   options.\n * @returns A `dispose()` function that detaches the listener and\n *   releases the ref-counted projection. Idempotent.\n */\nexport function acquireChannelEffect(\n  registry: ChannelRegistry,\n  channels: readonly Channel[],\n  namespace: readonly string[],\n  options: ChannelEffectOptions\n): () => void {\n  const { onEvent, onError, ...projectionOptions } = options;\n  const acquired = registry.acquire<Event[]>(\n    channelProjection(channels, namespace, {\n      // Live-only by default for side effects; callers opt into replay\n      // explicitly when they want to re-process history.\n      replay: projectionOptions.replay ?? false,\n      bufferSize: projectionOptions.bufferSize,\n    })\n  );\n  const { store } = acquired;\n\n  // Cursor: the last event already delivered to `onEvent`. Seeded with\n  // the current tail so events buffered before this consumer attached\n  // are skipped (no double-counting on a late mount).\n  const initial = store.getSnapshot();\n  let lastDelivered: Event | undefined =\n    initial.length > 0 ? initial[initial.length - 1] : undefined;\n\n  const deliver = (): void => {\n    const snapshot = store.getSnapshot();\n    let start = 0;\n    if (lastDelivered !== undefined) {\n      const index = snapshot.lastIndexOf(lastDelivered);\n      // `index === -1` means the cursor fell out of the buffer (thread\n      // rebind reset the store, or the bounded buffer evicted it).\n      // Deliver from the start of the current snapshot.\n      start = index === -1 ? 0 : index + 1;\n    }\n    for (let i = start; i < snapshot.length; i += 1) {\n      const event = snapshot[i] as Event;\n      try {\n        onEvent(event);\n      } catch (error) {\n        if (onError) onError(error);\n      }\n    }\n    if (snapshot.length > 0) {\n      lastDelivered = snapshot[snapshot.length - 1];\n    }\n  };\n\n  const unsubscribe = store.subscribe(deliver);\n\n  let released = false;\n  return () => {\n    if (released) return;\n    released = true;\n    unsubscribe();\n    acquired.release();\n  };\n}\n"],"mappings":";;;;;;;;;;;;;AAwEA,SAAgB,qBACd,UACA,UACA,WACA,SACY;CACZ,MAAM,EAAE,SAAS,SAAS,GAAG,sBAAsB;CACnD,MAAM,WAAW,SAAS,QACxBA,gBAAAA,kBAAkB,UAAU,WAAW;EAGrC,QAAQ,kBAAkB,UAAU;EACpC,YAAY,kBAAkB;EAC/B,CAAC,CACH;CACD,MAAM,EAAE,UAAU;CAKlB,MAAM,UAAU,MAAM,aAAa;CACnC,IAAI,gBACF,QAAQ,SAAS,IAAI,QAAQ,QAAQ,SAAS,KAAK,KAAA;CAErD,MAAM,gBAAsB;EAC1B,MAAM,WAAW,MAAM,aAAa;EACpC,IAAI,QAAQ;AACZ,MAAI,kBAAkB,KAAA,GAAW;GAC/B,MAAM,QAAQ,SAAS,YAAY,cAAc;AAIjD,WAAQ,UAAU,KAAK,IAAI,QAAQ;;AAErC,OAAK,IAAI,IAAI,OAAO,IAAI,SAAS,QAAQ,KAAK,GAAG;GAC/C,MAAM,QAAQ,SAAS;AACvB,OAAI;AACF,YAAQ,MAAM;YACP,OAAO;AACd,QAAI,QAAS,SAAQ,MAAM;;;AAG/B,MAAI,SAAS,SAAS,EACpB,iBAAgB,SAAS,SAAS,SAAS;;CAI/C,MAAM,cAAc,MAAM,UAAU,QAAQ;CAE5C,IAAI,WAAW;AACf,cAAa;AACX,MAAI,SAAU;AACd,aAAW;AACX,eAAa;AACb,WAAS,SAAS"}