Resilient agent harness: a cheat sheet

# vehicles.py: pure domain, no I/O
from dataclasses import dataclass, replace
import math

@dataclass(frozen=True)
class Vehicle:
    id: str
    x: float
    y: float
    heading: float          # radians, anticlockwise from +x
    wiring: str             # "2a" | "2b" | "3a" | "3b"

@dataclass(frozen=True)
class WorldState:
    tick: int
    vehicles: tuple[Vehicle, ...]
    lights: tuple[tuple[float, float], ...]
    schema_version: int = 1  # version saved state like an API

SENSOR_ANGLE, SENSOR_DIST, AXLE = 0.5, 0.5, 1.0

def _intensity(px, py, lights):
    return sum(1 / (1 + (px - lx) ** 2 + (py - ly) ** 2) for lx, ly in lights)

def sense(v, lights):
    def at(offset):
        a = v.heading + offset
        return _intensity(v.x + SENSOR_DIST * math.cos(a),
                          v.y + SENSOR_DIST * math.sin(a), lights)
    return at(+SENSOR_ANGLE), at(-SENSOR_ANGLE)   # (left, right)

WIRING = {
    "2a": lambda l, r: (l, r),
    "2b": lambda l, r: (r, l),
    "3a": lambda l, r: (1 - l, 1 - r),
    "3b": lambda l, r: (1 - r, 1 - l),
}

def move(v, lights, dt=0.1):
    left_motor, right_motor = WIRING[v.wiring](*sense(v, lights))
    speed = (left_motor + right_motor) / 2
    heading = v.heading + (right_motor - left_motor) / AXLE * dt
    return replace(v, heading=heading,
                   x=v.x + speed * math.cos(heading) * dt,
                   y=v.y + speed * math.sin(heading) * dt)

def step(state: WorldState) -> WorldState:
    return replace(state, tick=state.tick + 1,
                   vehicles=tuple(move(v, state.lights) for v in state.vehicles))

# harness.py: the only place with state, time and side effects
import time
from dataclasses import dataclass

@dataclass(frozen=True)
class Budget:
    max_ticks: int = 10_000
    max_seconds: float = 60.0

    def exceeded(self, state, started):
        if state.tick >= self.max_ticks:
            return "tick_budget"
        if time.monotonic() - started > self.max_seconds:
            return "time_budget"
        return None

@dataclass(frozen=True)
class Outcome:
    status: str             # "done" | "parked"
    state: object
    reason: str | None = None

class Harness:
    def __init__(self, store, budget, step_fn, sinks=(), done=lambda s: False):
        self.store, self.budget, self.step = store, budget, step_fn
        self.sinks, self.done = sinks, done

    def run(self, run_id, initial):
        state = self.store.load(run_id) or initial     # resume, don't restart
        started = time.monotonic()
        while not self.done(state):
            if reason := self.budget.exceeded(state, started):
                self.store.save(run_id, state)
                return Outcome("parked", state, reason)  # hand to a human
            state = self.step(state)
            for sink in self.sinks:                      # side effects, keyed
                sink.emit(run_id, state)
            self.store.save(run_id, state)               # checkpoint every tick
        return Outcome("done", state)

class IdempotentSink:
    """Wraps anything that acts on the world (audio, motors, network)."""
    def __init__(self, act):
        self.act, self.seen = act, set()   # use a persistent store in real life

    def emit(self, run_id, state):
        for v in state.vehicles:
            key = f"{run_id}:{v.id}:{state.tick}"   # stable IDs only
            if key not in self.seen:
                self.act(key, v)
                self.seen.add(key)

import math
from vehicles import Vehicle, WorldState, move, step
from harness import Harness, Budget, IdempotentSink

LIGHT_AHEAD_LEFT = ((5.0, 2.0),)

def vehicle(wiring):
    return Vehicle("v1", 0.0, 0.0, 0.0, wiring)

# Behavioural "evals": the wiring does what Braitenberg says
def test_fear_turns_away_from_light():
    assert move(vehicle("2a"), LIGHT_AHEAD_LEFT).heading < 0

def test_aggression_turns_towards_light():
    assert move(vehicle("2b"), LIGHT_AHEAD_LEFT).heading > 0

def test_love_turns_towards_light():
    assert move(vehicle("3a"), LIGHT_AHEAD_LEFT).heading > 0

# Pattern 3: a crash then resume ends where an uninterrupted run ends
class MemoryStore:
    def __init__(self): self.data = {}
    def load(self, k): return self.data.get(k)
    def save(self, k, s): self.data[k] = s

def world():
    return WorldState(0, (vehicle("2b"),), LIGHT_AHEAD_LEFT)

def test_resume_matches_uninterrupted_run():
    clean = Harness(MemoryStore(), Budget(max_ticks=50), step).run("a", world())

    store = MemoryStore()
    Harness(store, Budget(max_ticks=20), step).run("b", world())   # "crash" at 20
    resumed = Harness(store, Budget(max_ticks=50), step).run("b", world())

    assert resumed.state == clean.state

# Pattern 2: replaying a tick doesn't repeat side effects
def test_sink_acts_once_per_vehicle_per_tick():
    calls = []
    sink = IdempotentSink(lambda key, v: calls.append(key))
    s = step(world())
    sink.emit("run", s)
    sink.emit("run", s)          # retry of the same tick
    assert calls == ["run:v1:1"]

# Pattern 5: the loop is bounded and parks with a reason
def test_budget_parks_the_run():
    out = Harness(MemoryStore(), Budget(max_ticks=5), step).run("c", world())
    assert (out.status, out.reason, out.state.tick) == ("parked", "tick_budget", 5)