? RunContext : never;
+export type GetRunInstance = T extends RunInstance ? P : never;
+
+export class Run extends LazyPromise {
+ constructor(
+ handler: () => Promise,
+ protected readonly runContext: RunContext,
+ ) {
+ super(handler);
+ }
+
+ readonly [Symbol.toStringTag] = "Promise";
+
+ observe(fn: (emitter: Emitter) => void) {
+ fn(this.runContext.emitter as any);
+ return this;
+ }
+
+ context(value: object) {
+ Object.assign(this.runContext.context, value);
+ Object.assign(this.runContext.emitter.context, value);
+ return this;
+ }
+
+ middleware(fn: (context: RunContext) => void) {
+ fn(this.runContext);
+ return this;
+ }
+}
+
+export class RunContext extends Serializable {
+ static #storage = new AsyncLocalStorage>();
+
+ protected readonly controller: AbortController;
+ public readonly runId: string;
+ public readonly groupId: string;
+ public readonly parentId?: string;
+ public readonly emitter;
+ public readonly context: object;
+
+ get signal() {
+ return this.controller.signal;
+ }
+
+ abort() {
+ this.controller.abort();
+ }
+
+ constructor(
+ protected readonly instance: RunInstance,
+ parent?: RunContext,
+ signal?: AbortSignal,
+ ) {
+ super();
+ this.runId = createRandomHash(5);
+ this.parentId = parent?.runId;
+ this.groupId = parent?.groupId ?? createRandomHash();
+ this.context = createNonOverridableObject(
+ omit((parent?.context ?? {}) as any, ["id", "parentId"]),
+ );
+
+ this.controller = new AbortController();
+ registerSignals(this.controller, [signal, parent?.signal]);
+
+ this.emitter = instance.emitter.child({
+ context: this.context,
+ trace: {
+ id: this.groupId,
+ runId: this.runId,
+ parentRunId: parent?.runId,
+ },
+ });
+ if (parent) {
+ this.emitter.pipe(parent.emitter);
+ }
+ }
+
+ destroy() {
+ this.emitter.destroy();
+ this.controller.abort(new FrameworkError("Context destroyed."));
+ }
+
+ static enter(
+ self: RunInstance,
+ fn: (context: RunContext) => Promise,
+ signal?: AbortSignal,
+ ) {
+ const parent = RunContext.#storage.getStore();
+ const runContext = new RunContext(self, parent, signal);
+
+ return new Run(async () => {
+ const emitter = runContext.emitter.child({
+ namespace: ["run"],
+ creator: runContext,
+ context: { internal: true },
+ });
+
+ try {
+ await emitter.emit("start", null);
+ const result = await Promise.race([
+ RunContext.#storage.run(runContext, fn, runContext),
+ new Promise((_, reject) =>
+ runContext.signal.addEventListener("abort", () => reject(runContext.signal.reason)),
+ ),
+ ]);
+ await emitter.emit("success", result);
+ return result;
+ } catch (_e) {
+ const e = FrameworkError.ensure(_e);
+ await emitter.emit("error", e);
+ throw e;
+ } finally {
+ await emitter.emit("finish", null);
+ runContext.destroy();
+ }
+ }, runContext);
+ }
+
+ static {
+ this.register();
+ }
+
+ createSnapshot() {
+ return {
+ controller: this.controller,
+ runId: this.runId,
+ groupId: this.groupId,
+ parentId: this.parentId,
+ emitter: this.emitter,
+ context: this.context,
+ };
+ }
+
+ loadSnapshot(snapshot: ReturnType) {
+ Object.assign(this, snapshot);
+ }
+}
diff --git a/src/emitter/__snapshots__/emitter.test.ts.snap b/src/emitter/__snapshots__/emitter.test.ts.snap
new file mode 100644
index 0000000..9d138dd
--- /dev/null
+++ b/src/emitter/__snapshots__/emitter.test.ts.snap
@@ -0,0 +1,36 @@
+// Vitest Snapshot v1, https://vitest.dev/guide/snapshot.html
+
+exports[`Emitter > Complex workflow > Emits events in-order 1`] = `
+[
+ [
+ "start",
+ {
+ "a": 1,
+ },
+ ],
+ [
+ "success",
+ "a",
+ ],
+ [
+ "success",
+ "b",
+ ],
+ [
+ "start",
+ {
+ "a": 2,
+ },
+ ],
+ [
+ "start",
+ {
+ "a": 3,
+ },
+ ],
+ [
+ "success",
+ "c",
+ ],
+]
+`;
diff --git a/src/emitter/emitter.test.ts b/src/emitter/emitter.test.ts
new file mode 100644
index 0000000..75016b0
--- /dev/null
+++ b/src/emitter/emitter.test.ts
@@ -0,0 +1,114 @@
+/**
+ * Copyright 2024 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { Emitter } from "@/emitter/emitter.js";
+import { describe, expect } from "vitest";
+import { EventMeta } from "@/emitter/types.js";
+
+describe("Emitter", () => {
+ it("Emits", async () => {
+ const emitter = new Emitter({
+ namespace: ["app"],
+ creator: this,
+ });
+ const fn = vitest.fn();
+ emitter.on("onStart", fn);
+ await emitter.emit("onStart", "1");
+ expect(fn).toBeCalledWith(
+ "1",
+ expect.objectContaining({
+ name: "onStart",
+ path: "app.onStart",
+ creator: this,
+ source: emitter,
+ createdAt: expect.any(Date),
+ context: expect.any(Object),
+ }),
+ );
+ });
+
+ describe("Complex workflow", () => {
+ const recorder: [keyof MyEvents, unknown][] = [];
+
+ interface MyEvents {
+ start: (data: { a: number }) => void;
+ success: (data: string) => void;
+ }
+
+ const emitter = new Emitter({
+ namespace: ["agent"],
+ });
+
+ it("Emits events in-order", async () => {
+ emitter.on("start", (data) => recorder.push(["start", data]));
+ emitter.on("success", (data) => recorder.push(["success", data]));
+ await emitter.emit("start", { a: 1 });
+ await emitter.emit("success", "a");
+ await emitter.emit("success", "b");
+ await emitter.emit("start", { a: 2 });
+ await emitter.emit("start", { a: 3 });
+ await emitter.emit("success", "c");
+
+ expect(recorder).toMatchSnapshot();
+ });
+
+ it("Resets", async () => {
+ recorder.length = 0;
+ emitter.reset();
+ await emitter.emit("success", "c");
+ expect(recorder).toHaveLength(0);
+ });
+ });
+
+ it("Accepts callbacks", async () => {
+ const emitter = new Emitter<{
+ start: (value: number) => void;
+ }>();
+ const recorder: string[] = [];
+ emitter.registerCallbacks({
+ start: (_, event) => recorder.push(event.path),
+ });
+ await emitter.emit("start", 10);
+ expect(recorder).toHaveLength(1);
+ });
+
+ it("Handles nesting", async () => {
+ const recorder: { source: string; meta: EventMeta }[] = [];
+
+ const root = Emitter.root;
+ root.match("*", (data, meta) => {
+ recorder.push({ source: "root.*", meta });
+ });
+ //root.on("agent.onStart", (_, meta) =>
+ // recorder.push({ name: meta.name, path: meta.path, source: "root" }),
+ //);
+ //root.on("agent.runner.onStart", (_, meta) => recorder.push({ source: "root.runner", meta }));
+
+ const agent = root.child({
+ namespace: ["agent"],
+ });
+ agent.on("onStart", (_, meta) => recorder.push({ source: "agent", meta }));
+ agent.on("*", (_, meta) => recorder.push({ source: "agent.*", meta }));
+ await agent.emit("onStart", null);
+
+ // agent.runner.onStart
+ const runner = agent.child({
+ namespace: ["runner"],
+ });
+ // agent.runner, agent.runner.onStart
+ await runner.emit("onStart", null);
+ });
+});
diff --git a/src/emitter/emitter.ts b/src/emitter/emitter.ts
new file mode 100644
index 0000000..1496ed7
--- /dev/null
+++ b/src/emitter/emitter.ts
@@ -0,0 +1,238 @@
+/**
+ * Copyright 2024 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import {
+ Callback,
+ CleanupFn,
+ EmitterOptions,
+ EventMeta,
+ EventTrace,
+ Matcher,
+ StringKey,
+} from "@/emitter/types.js";
+export type { EventMeta, EventTrace };
+import { Cache } from "@/cache/decoratorCache.js";
+import { createFullPath, isPath, assertValidName, assertValidNamespace } from "@/emitter/utils.js";
+import { EmitterError } from "@/emitter/errors.js";
+import { createRandomHash } from "@/internals/helpers/hash.js";
+import { shallowCopy, toBoundedFunction } from "@/serializer/utils.js";
+import { RequiredAll } from "@/internals/types.js";
+import { Serializable } from "@/internals/serializable.js";
+import { pick } from "remeda";
+
+export interface EmitterInput {
+ groupId?: string;
+ namespace?: string[];
+ creator?: object;
+ context?: E & object;
+ trace?: EventTrace;
+}
+export interface EmitterChildInput {
+ groupId?: string;
+ namespace?: string[];
+ creator?: object;
+ context?: E & object;
+ trace?: EventTrace;
+}
+
+interface Listener {
+ readonly match: (event: EventMeta) => boolean;
+ readonly raw: Matcher;
+ readonly callback: Callback;
+ readonly options?: EmitterOptions;
+}
+
+export class Emitter>> extends Serializable {
+ protected listeners = new Set();
+ protected readonly groupId?: string;
+ public readonly namespace: string[];
+ public readonly creator?: object;
+ public readonly context: object;
+ protected readonly trace?: EventTrace;
+
+ constructor(input: EmitterInput = {}) {
+ super();
+ this.groupId = input?.groupId;
+ this.namespace = input?.namespace ?? [];
+ this.creator = input.creator ?? Object.prototype;
+ this.context = input?.context ?? {};
+ this.trace = input.trace;
+ assertValidNamespace(this.namespace);
+ }
+
+ static {
+ this.register();
+ }
+
+ @Cache()
+ static get root() {
+ return new Emitter>>({ creator: Object.create(null) });
+ }
+
+ child(input: EmitterChildInput = {}): Emitter {
+ const child = new Emitter({
+ trace: input.trace ?? this.trace,
+ groupId: input?.groupId ?? this.groupId,
+ context: { ...this.context, ...input?.context },
+ creator: input?.creator ?? this.creator,
+ namespace: input?.namespace
+ ? [...this.namespace, ...input.namespace]
+ : this.namespace.slice(),
+ });
+
+ child.pipe(this);
+ return child;
+ }
+
+ pipe(target: Emitter): CleanupFn {
+ return this.on(
+ // @ts-expect-error
+ "*.*",
+ toBoundedFunction(
+ // @ts-expect-error
+ (...args) => target.invoke(...args),
+ [{ value: target, name: "target" }],
+ ),
+ {
+ isBlocking: true,
+ once: false,
+ persistent: true,
+ },
+ );
+ }
+
+ destroy() {
+ this.listeners.clear();
+ }
+
+ reset() {
+ for (const listener of this.listeners) {
+ if (!listener.options?.persistent) {
+ this.listeners.delete(listener);
+ }
+ }
+ }
+
+ registerCallbacks>>(
+ callbacks: Partial<
+ Record extends Callback ? Callback : never>
+ >,
+ options?: Partial>,
+ ): CleanupFn {
+ const listeners: CleanupFn[] = [];
+ Object.entries(callbacks).forEach(([key, value]) => {
+ if (value) {
+ // @ts-expect-error
+ listeners.push(this.on(key, value, options?.[key]));
+ }
+ });
+ return () => listeners.forEach((cleanup) => cleanup());
+ }
+
+ on>>(
+ event: K,
+ callback: NonNullable extends Callback ? T[K] : never,
+ options?: EmitterOptions,
+ ): CleanupFn {
+ return this.match(event as Matcher, callback as Callback, options);
+ }
+
+ match(matcher: Matcher, callback: Callback, options?: EmitterOptions): CleanupFn {
+ const listener: Listener = {
+ options,
+ callback,
+ raw: matcher,
+ match: (() => {
+ if (matcher === "*") {
+ return (event) => event.path === createFullPath(this.namespace, event.name);
+ } else if (matcher === "*.*") {
+ return () => true;
+ } else if (matcher instanceof RegExp) {
+ return (event) => matcher.test(event.path);
+ } else if (typeof matcher === "function") {
+ return matcher;
+ } else if (typeof matcher === "string") {
+ return isPath(matcher)
+ ? (event) => event.path === matcher
+ : (event) =>
+ event.name === matcher && event.path === createFullPath(this.namespace, event.name);
+ } else {
+ throw new EmitterError("Invalid matcher provided!");
+ }
+ })(),
+ };
+ this.listeners.add(listener);
+
+ return () => this.listeners.delete(listener);
+ }
+
+ async emit>>(
+ name: K,
+ value: NonNullable extends Callback
+ 
+
+