共享执行事实源和消费隔离

策略的具体实现如下。

export function createEventScope(options: CreateDiagnosticsOptions = {}): DiagnosticContext {
  const { output } = options;
  const write = output?.write.bind(output);
  const accepts = output?.accepts?.bind(output);
  const runId = options.runId || invocationId();
  const spanId = options.spanId ?? 'run';
  // Shared by every child scope so `producerSequence` orders facts across the whole run.
  let sequence = 0;
  const observer = (source: import('../structs/execution/events.js').ExecutionObserver) => ({
    accepts: source.accepts?.bind(source),
    event: source.event.bind(source),
  });
  type Listener = ReturnType<typeof observer>;
  const initial = options.observe ? [observer(options.observe)] : [];
  const context = (
    span: string,
    parent?: string,
    fields: DiagnosticSpanFields = {},
    listeners: readonly Listener[] = initial,
  ): DiagnosticContext => {
    const identity = Object.freeze({
      ...fields,
      runId,
      spanId: span,
      ...(parent ? { parentSpanId: parent } : {}),
    });
    // A missing accepts hook means the sink takes everything except an explicit refusal.
    const logEnabled = (event: CoreEventName, content: 'metadata' | 'content'): boolean => {
      if (!write) return false;
      try {
        return accepts?.({ event, content }) !== false;
      } catch {
        return false;
      }
    };
    // Listeners default to metadata-only; content requires an explicit opt-in.
    const listenerEnabled = (
      listener: Listener,
      event: CoreEventName,
      content: 'metadata' | 'content',
    ): boolean => {
      try {
        return listener.accepts
          ? listener.accepts({ ...identity, event, content }) === true
          : content === 'metadata';
      } catch {
        return false;
      }
    };
    const fact = <K extends CoreEventName>(
      event: K,
      payload: () => CoreEventPayloads[K],
      debug?: DebugStepsOutput,
    ): CoreEvent => {
      const value = payload();
      const reported = captureSubFlowSnapshot(
        scope.enabled(event, 'content') ? debug?.recordedSubSteps : undefined,
        identity,
      );
      return {
        ...identity,
        ...reported,
        ...(typeof debug?.usedModel === 'boolean' ? { usedModel: debug.usedModel } : {}),
        version: 2,
        eventId:
          event === 'workflow.event'
            ? (value as CoreEventPayloads['workflow.event']).event.eventId
            : event === 'loop.event'
              ? (value as CoreEventPayloads['loop.event']).eventId
              : invocationId(),
        type: event,
        producerSequence: ++sequence,
        occurredAtMs: Date.now(),
        payload: value,
      } as CoreEvent;
    };
    const publish = (record: CoreEvent): void => {
      const event = record.type,
        kind = coreEventCatalog[event];
      if (logEnabled(event, kind)) {
        try {
          const pending = write?.(eventSnapshot(record, logEnabled(event, 'content')));
          if (pending) void Promise.resolve(pending).catch(() => {});
        } catch {}
      }
      for (const listener of listeners) {
        if (!listenerEnabled(listener, event, kind)) continue;
        try {
          const copy = eventSnapshot(record, listenerEnabled(listener, event, 'content'));
          if (copy.type === 'workflow.event') Object.freeze(copy.payload.event);
          const pending = listener.event(Object.freeze(copy));
          if (pending) void Promise.resolve(pending).catch(() => {});
        } catch {}
      }
    };
    const scope: DiagnosticContext = {
      identity,
      /** Whether any consumer would accept this event, optionally requiring content. */
      enabled(event, content) {
        if (!Object.hasOwn(coreEventCatalog, event)) return false;
        const kind = content === 'content' ? 'content' : coreEventCatalog[event];
        return (
          logEnabled(event, kind) ||
          listeners.some((listener) => listenerEnabled(listener, event, kind))
        );
      },
      /**
       * Build and publish a fact without waiting for anyone.
       *
       * Observation failures are swallowed so reporting cannot change execution.
       */
      emit<K extends CoreEventName>(
        event: K,
        payload: () => CoreEventPayloads[K],
        debug?: DebugStepsOutput,
      ) {
        if (!this.enabled(event)) return;
        try {
          publish(fact(event, payload, debug));
        } catch {
          /* Observation must not change execution. */
        }
      },
      /**
       * Publish a fact and then hand a copy of it to a delivery sink.
       *
       * Publication comes first so observers retain causal order even if the sink stalls.
       */
      async deliver(event, payload, sink) {
        const record = fact(event, payload);
        publish(record);
        await sink(eventSnapshot(record) as never);
      },
      /**
       * Derive a child span.
       *
       * A child that starts its own invocation records the parent invocation, which keeps
       * a request-scoped trace navigable without modelling the whole call tree.
       */
      child: (id, childFields = {}) =>
        context(
          `${span}/${id}`,
          span,
          {
            ...fields,
            ...(childFields.invocationId &&
            childFields.invocationId !== fields.invocationId &&
            fields.invocationId
              ? { parentInvocationId: fields.invocationId }
              : {}),
            ...childFields,
          },
          listeners,
        ),
      /** Add an observer to an existing scope while keeping its identity and sequence. */
      withObserver: (source) => context(span, parent, fields, [...listeners, observer(source)]),
    };
    return scope;
  };
  return context(spanId);
}

export function executionEvents(
  options?: import('../structs/common.js').ExecutionOptions,
): DiagnosticContext | undefined {
  const { diagnostics, observe } = options ?? {};
  return observe ? (diagnostics ?? createEventScope()).withObserver(observe) : diagnostics;
}