DictStore — in-memory (default)

DictStore is the in-memory ExecutionStore: a handful of plain Python dicts, lists and sets living in the process. It is the default backend for embedded, non-durable runs — tests, scenarios, notebooks, single-process hosts. Its defining property is that load returns the same Execution object that was handed to save/commit: there is no serialization round-trip, no JSON blob, no copy. A caller that holds a reference to an Execution sees every mutation the engine makes, and load hands back that identical object — the identity contract the in-place test harness relies on. Nothing survives the process exiting.

Data model

DictStore.__init__ allocates exactly these structures (all live on self):

self._by_id    : dict[str, Execution]            # execution_id -> the live Execution object
self._outbox   : list[OutboxEntry]               # FIFO of deferred events awaiting delivery
self._processed: set[tuple[str, str]]            # {(execution_id, event_id)} already-handled marks
self._timers   : dict[tuple[str, str], float]    # (execution_id, path) -> fire_at (epoch seconds)
self._spawns   : list[SpawnEntry]                # FIFO of pending child-Execution creations
self._trace    : dict[str, list[dict]]           # execution_id -> ordered trace steps (capped ring)
self._trace_idx: dict[str, int]                  # execution_id -> next monotonic step index
self.trace_max : int = DEFAULT_TRACE_MAX         # ring size (200); 0/None disables trimming
self._seq      : int = 0                          # monotonic counter for outbox seq numbers
self._spawn_seq: int = 0                          # monotonic counter for spawn seq numbers

What each one holds and why:

  • _by_id — the entire authoritative state of the store. Maps an Execution id to the live Execution instance (see execution.py: id, definition_id, status, outcome, active_path, history, context, version, the orthogonal parent_id/child_id/ children, etc.). Because the value is the object itself, no deserialization is needed and identity is preserved across save/load.

  • _outbox — the transactional outbox: a FIFO list of OutboxEntry(seq, target_id, event). commit appends an entry for every emitted event so delivery happens after the Execution is durably saved (here: after the dict is mutated), never inline. A relay drains it via pending_outbox/ack_outbox. target_id is the Execution to deliver to (or None for an untargeted event); seq is a monotonic ack token from _seq.

  • _processed — the dedupe set for at-least-once delivery. Holds (execution_id, event_id) pairs. If a redelivered event’s id is already in the set, the worker skips it. Populated by commit’s processed_event_id, queried by is_processed.

  • _timers — durable timers, keyed by (execution_id, path) mapping to the absolute fire_at (epoch seconds). The composite key means re-entry replaces: re-arming the timer for the same state path overwrites the prior fire_at rather than stacking timers. A TimerOp(action="schedule") upserts; action="cancel" removes.

  • _spawns — the spawn outbox: a FIFO of SpawnEntry(seq, parent_id, child_id, root_path, context), the orthogonal-fork sibling of _outbox. When a parent enters an AND-state, commit records the parent’s advance + its join expectations (the children dict on the Execution) and the intents to create each region child, all in one shot. A relay later creates the children idempotently via pending_spawns/ack_spawn. seq comes from _spawn_seq.

  • _trace — the opt-in execution timeline: execution_id -> list[dict], one dict per recorded step (event/transition/actions/context_out plus a stamped index). Capped to the last trace_max steps per id (a ring). This is a side-channel for monitoring; load never reads it and is unaffected by it.

  • _trace_idx — execution_id -> next monotonic step index. Survives ring trimming, so the stamped index keeps climbing (0, 1, 2, …) even after old steps are dropped — readers see a stable, gap-free-from-the-store’s-view ordering of recent steps.

  • trace_max — the ring size, defaulting to DEFAULT_TRACE_MAX = 200 (from _base.py). A truthy value trims; 0/None disables trimming (unbounded).

  • _seq / _spawn_seq — independent monotonic counters incremented before each append, so the first outbox/spawn entry gets seq == 1. They give every entry a stable ack token.

Operations

load(execution_id) -> Optional[Execution]

return self._by_id.get(execution_id)

Returns the live object (or None). No copy, no deserialization — the caller gets the exact instance under that id. This is what makes the store’s identity contract hold.

save(exe) -> None — the CAS check

prev = self._by_id.get(exe.id)
if prev is not None and prev is not exe and prev.version != exe.version:
    raise StoreConflict(exe.id, expected=exe.version, found=prev.version)
exe.version += 1
self._by_id[exe.id] = exe

This is the optimistic-concurrency (CAS) gate, the single-writer backstop. The condition prev is not None and prev is not exe and prev.version != exe.version bites only when a different object is already stored under the same id at a different version — i.e. a genuine concurrent writer that loaded the row, advanced it, and committed while this caller held a stale copy. It raises StoreConflict(execution_id, expected, found) so the caller can reload and retry (or drop the stale work).

Three deliberate exemptions:

  • Same object (prev is exe): the common embedded case where the engine mutates the very object the store holds. prev is not exe is false, so the check is skipped and the write always wins — there is no other writer.

  • First write (prev is None): nothing to conflict with.

  • Different object, same version: not a conflict — it’s a legitimate replacement at the expected version.

On success exe.version is bumped in place to the committed value and the dict entry is set.

commit(exe, emits, processed_event_id=None, timers=(), spawns=(), trace=None)

The atomic event-boundary write. Because everything is in-process and synchronous, “atomic” here means it all runs without interleaving. Step by step:

  1. CAS first — self.save(exe). This raises StoreConflict before any emit is enqueued, so a losing writer never pollutes the outbox.

  2. Outbox append — for each (target_id, event) in emits: bump _seq, append OutboxEntry(self._seq, target_id, event). Deferred delivery, post-save.

  3. Dedupe — if processed_event_id is not None, add (exe.id, processed_event_id) to _processed, marking the just-handled event so a redelivery is skipped.

  4. Timer ops — for each TimerOp in timers: schedule sets self._timers[(exe.id, op.path)] = op.fire_at (upsert/replace); anything else (cancel) does self._timers.pop((exe.id, op.path), None).

  5. Spawn append — for each (child_id, root_path, context) in spawns: bump _spawn_seq, append SpawnEntry(self._spawn_seq, exe.id, child_id, root_path, dict(context)). Note dict(context) — the context is copied here so the spawn intent is snapshotted independently of later parent mutations (one of the few copies the store makes).

  6. Trace record — if trace is not None, self._record_trace(exe.id, trace).

Why one call: the parent’s advance, its join expectations, the events it emitted, the dedupe mark, the timers it armed/cancelled, and the children it wants forked must all land together or not at all. The CAS guards the whole boundary.

is_processed(execution_id, event_id) -> bool

return (execution_id, event_id) in self._processed

The dedupe query. True iff that exact pair was recorded by a prior commit.

pending_outbox() -> list[OutboxEntry] / ack_outbox(seq)

pending_outbox: return list(self._outbox)            # a copy, oldest first
ack_outbox:     self._outbox = [e for e ... e.seq != seq]   # drop the acked entry

The relay reads pending entries, delivers them, then acks each by seq — ack_outbox rebuilds the list without that entry. Returning a copy means iterating the relay’s snapshot is safe even if commit appends concurrently (single loop, but still defensive).

pending_spawns() -> list[SpawnEntry] / ack_spawn(seq)

pending_spawns: return list(self._spawns)            # a copy, oldest first
ack_spawn:      self._spawns = [s for s ... s.seq != seq]   # drop the acked spawn

The orthogonal-fork twin of the outbox pair. The relay drains pending spawns, creates each child Execution idempotently (skip if it already exists), then acks by seq.

due_timers(now) -> list[tuple[str, str, float]]

return [(eid, path, fa) for (eid, path), fa in self._timers.items() if fa <= now]

Returns every armed timer whose fire_at <= now, as (execution_id, path, fire_at). The sweep calls this on the idle path and delivers a Timeout event for each.

delete_timer(execution_id, path, fire_at)

if self._timers.get((execution_id, path)) == fire_at:
    del self._timers[(execution_id, path)]

Removes the timer only if it still holds the exact fire_at that fired. This guard is the point: if the model re-scheduled the same (execution_id, path) to a new time between the sweep reading it and deleting it, the stored fire_at no longer matches, the delete is a no-op, and the freshly re-armed timer survives a stale sweep.

Trace: _record_trace / append_trace / read_trace

_record_trace(execution_id, entry):
    idx = self._trace_idx.get(execution_id, 0)
    self._trace_idx[execution_id] = idx + 1
    steps = self._trace.setdefault(execution_id, [])
    steps.append({**entry, "index": entry.get("index", idx)})
    if self.trace_max and len(steps) > self.trace_max:   # ring: keep only the last N
        del steps[: len(steps) - self.trace_max]

append_trace(execution_id, entry): self._record_trace(execution_id, entry)   # demo/test seam
read_trace(execution_id):          return list(self._trace.get(execution_id, []))  # a copy

_record_trace is the internal recorder commit calls; append_trace is the public demo/test seam that delegates to it; read_trace returns a copy of the per-id step list (or []).

The trace ring

_record_trace assigns each step a monotonic per-id index and trims the list to a ring:

  1. Read the next index for this id from _trace_idx (default 0).

  2. Advance _trace_idx[execution_id] by 1 — so the counter keeps climbing regardless of trimming.

  3. Append {**entry, "index": entry.get("index", idx)}: the caller’s entry, stamped with an index if it didn’t already carry one (an explicit index in the entry is honored).

  4. If trace_max is truthy and the list now exceeds it, del steps[: len(steps) - trace_max] drops the oldest overflow, keeping exactly the last trace_max steps.

The result: _trace[id] is a bounded window of the most recent steps, but the stamped index values reflect the true monotonic position in the full timeline (older indices are simply no longer present once trimmed). Default window: 200 steps.

Async twin

harel/engine/aio_store/dict.py defines AsyncDictStore, a faithful mirror of DictStore:

  • Identical structures — the same _by_id, _outbox, _processed, _timers, _spawns, _trace, _trace_idx, trace_max, _seq, _spawn_seq, with the same shapes and the same DEFAULT_TRACE_MAX.

  • Identical logic — the same CAS condition, the same commit ordering (CAS first, then outbox/dedupe/timers/spawns/trace), the same delete_timer guard, the same ring trimming.

  • async def surface — load, save, commit, is_processed, pending_outbox, ack_outbox, pending_spawns, ack_spawn, due_timers, delete_timer, close, and the trace seams read_trace/append_trace are coroutines, to satisfy the AsyncExecutionStore contract.

  • No lock — and none is needed. A single event loop schedules cooperatively, and none of these methods await internally, so each method body runs atomically between suspension points. There is no point at which another task can interleave mid-method, so the dict mutations are race-free without explicit locking. (_record_trace stays a plain def, called synchronously from inside commit/append_trace.)

Why it’s not durable

All state lives in the process’s _by_id/_outbox/_timers/_spawns/_processed/_trace dicts. When the process exits, it’s gone — there is no file, no socket, no JSON blob, no fsync. close() is a no-op precisely because there is nothing to release. That is the trade: zero serialization cost and live-object identity in exchange for no persistence and no cross-process sharing.

For durability and multi-process operation, use a durable backend (SQLite for one machine; Redis / Postgres / rqlite for distributed) — they serialize the Execution to a JSON blob with a broken-out version column and implement the same ExecutionStore Protocol, so swapping is a one-line change. See the stores hub and durability.