Correctness invariants

These seven rules are treated as blocking defects when violated, not as design preferences. Each one is stated here with what it buys you and where it is tested, so you can check the claim rather than take it.

1. Atomic commit

State mutations and emitted outputs commit atomically with the Beam bundle. Every effect an agent produces — memory writes, cache inserts, intents, traces, outputs — is staged in the activation context and applied only on success. A failed or timed-out activation mutates nothing.

You can watch this from outside. In the four-outputs example one key raises after writing memory; that write never lands, and the activation's staged traces are discarded with it, leaving a single ERROR trace where a successful activation emits a full span set:

async def route_by_event(ctx: ActivationContext) -> Complete:
    """Take a different path per event so one pipeline exercises each output."""
    if ctx.event == b"BROKEN":
        # Routed to `.errors` as `activation_error`. Nothing this activation
        # staged — including this memory write — reaches durable state.
        ctx.memory.set("scratch", b"never-persisted")
        raise RuntimeError("downstream schema changed")

    if ctx.event == b"NOTIFY":
        ctx.act("slack.post", '{"channel": "#ops"}', ttl_ms=INTENT_TTL_MS)
        return Complete(output=b"notified")

    response = await ctx.call_model(
        LlmRequest(
            model_id="fake-1",
            messages=[ctx.event.decode()],
            tools_schema=None,
            sampling_params=None,
        )
    )
    return Complete(output=response.response)

Tested by tests/core/test_dofn_commit.py.

2. Deterministic intent IDs

intent_id = uuid5(NAMESPACE, f"{entity_key.hex()}|{seq}|{step_index}")

A pure function of the activation's position. Never a clock, never a counter, never randomness. A replayed bundle that walks the same path produces byte-identical intents, and the effector deduplicates on intent_id.

This is the entire effectively-once argument. Not an at-least-once delivery guarantee plus hope — a deterministic identity that makes duplicate suppression a lookup.

The id is computable ahead of time, which is how the re-injection example can name the intent it is about to receive a result for:

def _tool_result(key: bytes, intent_id: str, payload: bytes, t_ms: int):
    """One effector result, shaped as it arrives from the results topic."""
    env = AgentEnvelope(entity_key=key, event_time_ms=t_ms)
    env.tool_result.intent_id = intent_id
    env.tool_result.entity_key = key
    env.tool_result.payload = payload
    env.tool_result.status = ToolResult.OK
    return TimestampedValue(env, t_ms / 1000)

Tested by tests/semantics/test_retry_determinism.py and tests/semantics/test_effectively_once_e2e.py.

3. Replay cache

Every model call is keyed by sha256(model_id, canonical_json(messages), tools_schema, sampling_params, key, seq) and cached in keyed state — LRU, 64 entries maximum, 6-hour TTL, 100 KiB blob cap.

The property that matters: a bundle retry incurs zero additional provider calls on the cached path. Without it, a retried bundle would re-bill every model call in the activation and — worse — could take a different path if the model answered differently the second time.

Tested by tests/model/test_replay_cache_hits.py and the retry-determinism gate, which asserts zero extra FakeLLM calls under forced bundle retries.

4. Per-key serialization

Beam stateful DoFns process one element at a time per key. Memory is race-free by construction — not by locking, and not by careful ordering in the agent code. Cross-key parallelism comes from the runner.

The consequence for you: never introduce cross-key shared mutable state. The only sanctioned exceptions are documented worker-local singletons (circuit breakers, the vLLM sidecar via beam.utils.shared.Shared).

5. Side effects only via intents

Calling a side_effect=True tool directly raises. ctx.act(...) is the only effect path. External writes never execute inside the pipeline.

try:
    freeze_account(customer_id="cust-9")
except SideEffectToolError as exc:
    refusal = str(exc)
else:
    raise AssertionError("a side-effect tool must refuse a direct call")

There is one documented exception: idempotent upserts to the long-term MemoryStore, keyed by (key, seq).

Tested by tests/tools/test_side_effect_guard.py.

6. Timeouts fail closed at both layers

A pending approval that never arrives has to end, and it has to end without letting a late answer cause an effect afterwards. Two independent guards:

• Layer 1, in-pipeline. The HITL_TIMER fires and the policy's route runs — Deny, Drop, or Escalate.
• Layer 2, in the effector. An intent past its expires_at_ms is refused rather than executed. A non-positive expiry reads as expired, never as unbounded: the safe reading of "no expiry recorded" is "do not execute".

Late results that find no live continuation are dropped to .errors as orphaned_result.

def deny_on_timeout(fallback: FallbackContext) -> Route:
    """Route an unanswered approval to a deterministic denial.

    Pure and synchronous. It reads only the `FallbackContext` it is handed —
    which carries the suspended `seq`, the persisted snapshot, the elapsed
    deadline, the timer's fire time, and the intent ids nothing answered — so a
    retried timer bundle reaches the same decision.
    """
    return Deny(output=b"denied:no-approval:" + str(fallback.seq).encode())

The routing function must be pure, synchronous, and picklable. That is a correctness requirement: a timer callback re-executes when its bundle is retried, and a fallback that read a clock or called the model would make the retry diverge from the original. Every value the policy could need is carried on the FallbackContext.

Tested by tests/semantics/test_hitl_fail_closed.py.

7. State is protobuf, never pickle

All keyed state is protobuf with deterministic encoding. Pickle would make pipeline --update compatibility impossible to reason about and would tie the wire format to a Python version.

The upgrade rule follows from it: additive proto changes only. A breaking change requires a state_schema_version bump, lazy migration, and a golden-blob compatibility test.

Tested by tests/core/test_schema_compat.py and tests/core/test_coders.py.

How these are gated

The invariants are not checked by review alone. The repository runs a semantics test tier specifically for them — retry determinism under a chaos wrapper that forces bundle retries, effectively-once end to end against real Kafka/Redis/Flink with SIGKILLed effector workers and a killed TaskManager, and state compatibility against golden blobs. That tier gates every release and is never skipped or marked flaky.

See testing and CI for how the tiers are split and what runs where.

Watch them hold

Each invariant above is illustrated with a region of a real program. Those programs run offline on the DirectRunner, so the quickest way to check a claim is to run the one that demonstrates it:

• The four outputs — invariant 1, with one key that raises so you can watch the discard.
• Intents and resume — invariant 2, naming an intent id before the pipeline runs.
• The fast path — invariants 3 and 4 on the simplest possible activation.
• Human in the loop — invariant 6, fail-closed at both layers.