feat(runner): Add onCommit callback support for Cell and Stream operations (#1879)

* Add onCommit callback support for Cell and Stream operations

This change adds the ability to register callbacks that are invoked when
Cell.set(), Cell.send(), or Stream.send() operations commit to storage,
regardless of whether the commit succeeds or fails.

## What Changed

### API Changes
- Added optional `onCommit` parameter to:
  - `Cell.set(value, onCommit?)`
  - `Cell.send(value, onCommit?)`
  - `Stream.send(event, onCommit?)`
  - `IScheduler.queueEvent(eventRef, event, retries?, onCommit?)`

- Added `IExtendedStorageTransaction.addCommitCallback(callback)` method

### Callback Behavior
- **For Cell operations**: Callback is invoked after the transaction commits,
  whether the commit succeeds or fails. Multiple set() calls on the same
  transaction can each register callbacks - all are invoked on commit.

- **For Stream events**: Callback is invoked when either:
  - The event handler commits successfully, OR
  - The event handler fails and all retry attempts are exhausted
  - Callbacks are NOT invoked during intermediate retry attempts

- **Error handling**: Each callback is wrapped in try/catch to prevent one
  failing callback from breaking others. Errors are logged but don't affect
  transaction outcome.

- **Inspecting results**: Callbacks receive the transaction as a parameter
  and can call `tx.status()` to determine if the commit succeeded or failed.

### Implementation Details
- `ExtendedStorageTransaction` maintains a Set of callbacks and invokes them
  in its commit().then() handler after the commit completes
- `Scheduler.execute()` calls the callback when commit succeeds OR when the
  final retry fails (retriesLeft === 0)
- All callbacks are invoked synchronously within the microtask queue after
  commit resolution

## Why This Matters

This feature enables:
- Tracking completion of writes for UI updates or synchronization
- Implementing retry logic or error handling at the call site
- Coordinating dependent operations across transaction boundaries
- Debugging and observability of storage operations

The callbacks provide a way to observe transaction outcomes without blocking
the synchronous flow of Cell/Stream operations, maintaining the existing
non-blocking API while adding observability.

## Testing
- Added 7 new tests covering success cases, failure cases, error handling,
  multiple callbacks, and retry behavior
- All existing tests continue to pass (backward compatible)
- Total: 79 tests passing

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* clarified in comment that commit promise isn't rejected on tx failure, but instead an error code is passed.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: seefeldb

packages/runner/src/cell.ts

@@ -159,8 +159,14 @@ import { ContextualFlowControl } from "./cfc.ts";
 declare module "@commontools/api" {
   interface Cell<T> {
     get(): Readonly<T>;
-    set(value: Cellify<T> | T): void;
-    send(value: Cellify<T> | T): void;
+    set(
+      value: Cellify<T> | T,
+      onCommit?: (tx: IExtendedStorageTransaction) => void,
+    ): void;
+    send(
+      value: Cellify<T> | T,
+      onCommit?: (tx: IExtendedStorageTransaction) => void,
+    ): void;
     update<V extends Cellify<Partial<T> | Partial<T>>>(
       values: V extends object ? V : never,
     ): void;
@@ -249,7 +255,7 @@ declare module "@commontools/api" {
   }
 
   interface Stream<T> {
-    send(event: T): void;
+    send(event: T, onCommit?: (tx: IExtendedStorageTransaction) => void): void;
     sink(callback: (event: Readonly<T>) => Cancel | undefined | void): Cancel;
     sync(): Promise<Stream<T>> | Stream<T>;
     getRaw(options?: IReadOptions): any;
@@ -350,11 +356,11 @@ class StreamCell<T> implements Stream<T> {
     return this.link.rootSchema;
   }
 
-  send(event: T): void {
+  send(event: T, onCommit?: (tx: IExtendedStorageTransaction) => void): void {
     event = convertCellsToLinks(event) as T;
 
     // Use runtime from doc if available
-    this.runtime.scheduler.queueEvent(this.link, event);
+    this.runtime.scheduler.queueEvent(this.link, event, undefined, onCommit);
 
     this.cleanup?.();
     const [cancel, addCancel] = useCancelGroup();
@@ -432,7 +438,10 @@ export class RegularCell<T> implements Cell<T> {
     return validateAndTransform(this.runtime, this.tx, this.link, this.synced);
   }
 
-  set(newValue: Cellify<T> | T): void {
+  set(
+    newValue: Cellify<T> | T,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
+  ): void {
     if (!this.tx) throw new Error("Transaction required for set");
 
     // No await for the sync, just kicking this off, so we have the data to
@@ -447,10 +456,18 @@ export class RegularCell<T> implements Cell<T> {
       newValue,
       getTopFrame()?.cause,
     );
+
+    // Register commit callback if provided
+    if (onCommit) {
+      this.tx.addCommitCallback(onCommit);
+    }
   }
 
-  send(newValue: Cellify<T> | T): void {
-    this.set(newValue);
+  send(
+    newValue: Cellify<T> | T,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
+  ): void {
+    this.set(newValue, onCommit);
   }
 
   update<V extends Cellify<Partial<T> | Partial<T>>>(

packages/runner/src/runtime.ts

@@ -209,7 +209,12 @@ export interface IScheduler {
   unsubscribe(action: Action): void;
   onConsole(fn: ConsoleHandler): void;
   onError(fn: ErrorHandler): void;
-  queueEvent(eventRef: NormalizedFullLink, event: any): void;
+  queueEvent(
+    eventRef: NormalizedFullLink,
+    event: any,
+    retries?: number,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
+  ): void;
   addEventHandler(handler: EventHandler, ref: NormalizedFullLink): Cancel;
   runningPromise: Promise<unknown> | undefined;
 }

packages/runner/src/scheduler.ts

@@ -82,7 +82,11 @@ const DEFAULT_RETRIES_FOR_EVENTS = 5;
 const MAX_RETRIES_FOR_REACTIVE = 10;
 
 export class Scheduler implements IScheduler {
-  private eventQueue: { action: Action; retriesLeft: number }[] = [];
+  private eventQueue: {
+    action: Action;
+    retriesLeft: number;
+    onCommit?: (tx: IExtendedStorageTransaction) => void;
+  }[] = [];
   private eventHandlers: [NormalizedFullLink, EventHandler][] = [];
 
   private pending = new Set<Action>();
@@ -313,13 +317,15 @@ export class Scheduler implements IScheduler {
     eventLink: NormalizedFullLink,
     event: any,
     retries: number = DEFAULT_RETRIES_FOR_EVENTS,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
   ): void {
     for (const [link, handler] of this.eventHandlers) {
       if (areNormalizedLinksSame(link, eventLink)) {
         this.queueExecution();
         this.eventQueue.push({
           action: (tx: IExtendedStorageTransaction) => handler(tx, event),
           retriesLeft: retries,
+          onCommit,
         });
       }
     }
@@ -480,7 +486,7 @@ export class Scheduler implements IScheduler {
     // the topological sort in the right way.
     const event = this.eventQueue.shift();
     if (event) {
-      const { action, retriesLeft } = event;
+      const { action, retriesLeft, onCommit } = event;
       this.runtime.telemetry.submit({
         type: "scheduler.invocation",
         handler: action,
@@ -496,10 +502,23 @@ export class Scheduler implements IScheduler {
             // enough, especially for a series of event that act on the same
             // conflicting data.
             if (error && retriesLeft > 0) {
-              this.eventQueue.unshift({ action, retriesLeft: retriesLeft - 1 });
+              this.eventQueue.unshift({
+                action,
+                retriesLeft: retriesLeft - 1,
+                onCommit,
+              });
               // Ensure the re-queued event gets processed even if the scheduler
               // finished this cycle before the commit completed.
               this.queueExecution();
+            } else if (onCommit) {
+              // Call commit callback when:
+              // - Commit succeeds (!error), OR
+              // - Commit fails but we're out of retries (retriesLeft === 0)
+              try {
+                onCommit(tx);
+              } catch (callbackError) {
+                logger.error("Error in event commit callback:", callbackError);
+              }
             }
           });
         }

packages/runner/src/storage/extended-storage-transaction.ts

@@ -42,6 +42,10 @@ const logResult = (
 };
 
 export class ExtendedStorageTransaction implements IExtendedStorageTransaction {
+  private commitCallbacks = new Set<
+    (tx: IExtendedStorageTransaction) => void
+  >();
+
   constructor(public tx: IStorageTransaction) {}
 
   get journal(): ITransactionJournal {
@@ -175,6 +179,38 @@ export class ExtendedStorageTransaction implements IExtendedStorageTransaction {
   }
 
   commit(): Promise<Result<Unit, CommitError>> {
-    return this.tx.commit();
+    const promise = this.tx.commit();
+
+    // Call commit callbacks after commit completes (success or failure) Note
+    // that promise always resolves, even if the commit fails, in which case it
+    // passes an error message as result. An exception here would be an internal
+    // error that should propagate.
+    promise.then((_result) => {
+      // Call all callbacks, wrapping each in try/catch to prevent one
+      // failing callback from breaking others
+      for (const callback of this.commitCallbacks) {
+        try {
+          callback(this);
+        } catch (error) {
+          logger.error("Error in commit callback:", error);
+        }
+      }
+    });
+
+    return promise;
+  }
+
+  /**
+   * Add a callback to be called when the transaction commit completes.
+   * The callback receives the transaction as a parameter and is called
+   * regardless of whether the commit succeeded or failed.
+   *
+   * Note: Callbacks are called synchronously after commit completes.
+   * If a callback throws, the error is logged but doesn't affect other callbacks.
+   *
+   * @param callback - Function to call after commit
+   */
+  addCommitCallback(callback: (tx: IExtendedStorageTransaction) => void): void {
+    this.commitCallbacks.add(callback);
   }
 }

packages/runner/src/storage/interface.ts

@@ -496,6 +496,18 @@ export interface IStorageTransaction {
 export interface IExtendedStorageTransaction extends IStorageTransaction {
   tx: IStorageTransaction;
 
+  /**
+   * Add a callback to be called when the transaction commit completes.
+   * The callback receives the transaction as a parameter and is called
+   * regardless of whether the commit succeeded or failed.
+   *
+   * Note: Callbacks are called synchronously after commit completes.
+   * If a callback throws, the error is logged but doesn't affect other callbacks.
+   *
+   * @param callback - Function to call after commit
+   */
+  addCommitCallback(callback: (tx: IExtendedStorageTransaction) => void): void;
+
   /**
    * Reads a value from a (local) memory address and throws on error, except for
    * `NotFoundError` which is returned as undefined.