Distillation

Date: Sep 16, 2026

• 🟡 (10:11) Code shown exports SessionRunCoordinator from "./run-coordinator" and defines Coordinator<Key, E> with active, run(key), wake(key), and interrupt(key); executions are serialized per key while different keys may run concurrently.
• 🟡 (10:11) run-coordinator tracks each active key with Entry<E> fields done: Deferred.Deferred<void, E>, optional owner?: Fiber.Fiber<void, never>, pendingWake: boolean, and stopping: boolean; new entries initialize both booleans to false.
• 🟡 (10:11) Coordinator start(key, entry, force, successor = false) launches options.drain(key, force) through FiberSet.makeRuntime; initial owners await a ready deferred, while successors first use Effect.yieldNow. Effect.onExit synchronously calls settle(key, entry, exit).
• 🟡 (10:11) Coordinator settlement coalesces follow-up work: after a successful owner exit, if !entry.stopping && entry.pendingWake, it clears pendingWake and starts one successor with force=false; wake(key) sets pendingWake=true when an entry is active.
• 🟡 (10:11) Durable event decoding uses Durable.get(event.type) and throws InvalidDurableEventError with message Unknown durable event type ${event.type} if the definition is missing or not durable. It reconstructs durable metadata as { aggregateID: event.aggregateID, seq: event.seq, version: definition.durable.version } and decodes data with Schema.decodeUnknownSync(definition.data)(event.data).
• 🟡 (10:11) Event Interface exposes publish, subscribe, all, durable, deprecated listen, project, replay, replayAll, remove, and claim. replay/replayAll options include publish?: boolean, ownerID?: string, and strictOwner?: boolean; replayAll returns Effect.Effect<string | undefined>.
• 🟡 (10:11) Strict replay ownership rejects a row when input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID; sequence validation can report Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}.
• 🟡 (10:11) Event notification iterates all listeners, optionally wrapping each with observe(event, listener) when isolateListeners is enabled, then publishes to the event-type-specific PubSub channel and finally pubsub.all.
• 🟡 (10:11) claim(aggregateID, ownerID) updates EventSequenceTable.owner_id where EventSequenceTable.aggregate_id equals aggregateID, executes with .run(), and converts database failure to a defect via Effect.orDie.
• 🟡 (10:11) Durable streaming subscribes to aggregate wake notifications before reading history, starts from input.after ?? -1, updates its sequence cursor from events.at(-1)?.durable?.seq, and concatenates historical events with live wake-triggered readAfter results.