# proteus/game/runtime/multiagent_director.py """Deterministic scripted directors that author multi-agent handover memories. These do NOT run the live engine (which is single-focal). They own the multi-agent truth on an open field and emit a ``MemoryCheckpoint`` whose ``memory_turns`` carry per-sprite ``AgentFrame``s, resources, and events. The chosen agent (``a0``) is authored to embody a persona: it flees optimally (``risk_averse``, scenario ①) or beelines to the resource (``greedy``, scenario ②). Distractors wander. Geometry is open-field Manhattan. """ from __future__ import annotations import random from dataclasses import dataclass from proteus.game.runtime.memory import AgentFrame, MemoryCheckpoint, MemoryTurn from proteus.game.scenarios import _seat_layout as _seat _GRID = (64, 64) _DELTAS = {"up": (0, -1), "down": (0, 1), "left": (-1, 0), "right": (1, 0), "stay": (0, 0)} _MOVES = ("up", "down", "left", "right") # Identity map: a move name doubles as its facing label (kept explicit for the renderer). _FACING_FROM = {"right": "right", "left": "left", "up": "up", "down": "down"} _STAMP = "0000-00-00T00-00-00" @dataclass class _Agent: id: str x: int y: int size: int alive: bool = True is_chosen: bool = False def _in_bounds(x: int, y: int, size: int, grid=_GRID) -> bool: return 0 <= x and 0 <= y and x + size <= grid[0] and y + size <= grid[1] def _center(a: _Agent) -> tuple[int, int]: return (a.x + a.size // 2, a.y + a.size // 2) def _manhattan(a: tuple[int, int], b: tuple[int, int]) -> int: return abs(a[0] - b[0]) + abs(a[1] - b[1]) def _overlap(a: _Agent, b: _Agent) -> bool: return (a.x < b.x + b.size and b.x < a.x + a.size and a.y < b.y + b.size and b.y < a.y + a.size) def _legal_moves(a: _Agent, grid=_GRID) -> list[str]: out = [] for m in _MOVES: dx, dy = _DELTAS[m] if _in_bounds(a.x + dx, a.y + dy, a.size, grid): out.append(m) return out def _wander(a: _Agent, rng: random.Random) -> str: legal = _legal_moves(a) return rng.choice(legal) if legal else "stay" @dataclass class _RoamState: """Persistent per-agent heading for exploratory roaming.""" heading: str ttl: int # steps remaining before the heading is re-rolled def _roam(a: _Agent, rng: random.Random, state: _RoamState, run_min: int = 6, run_max: int = 12) -> str: """Exploratory roam: follow a persistent, seeded per-agent heading for a run of steps, then re-roll it. Headings persist (unlike :func:`_wander`'s fresh uniform draw each turn), so each agent travels far in a distinct direction and the pack covers a much wider area. Fully deterministic given *rng*. The agent stays NON-predator-aware (heading ignores the predator). Re-rolls when the run expires OR the current heading is blocked (boundary or another agent), choosing a fresh legal-ish heading and resetting the run. """ legal = _legal_moves(a) if not legal: return "stay" # Re-roll on expiry or when the committed heading is no longer in-bounds. if state.ttl <= 0 or state.heading not in legal: state.heading = rng.choice(legal) state.ttl = rng.randint(run_min, run_max) state.ttl -= 1 return state.heading def _flee(a: _Agent, threat: tuple[int, int]) -> str: """Pick the legal move that MAXIMISES Manhattan distance from *threat*.""" best, best_d = "stay", -1 for m in _legal_moves(a): dx, dy = _DELTAS[m] cc = (a.x + dx + a.size // 2, a.y + dy + a.size // 2) d = _manhattan(cc, threat) if d > best_d: best, best_d = m, d return best def _step_toward(a: _Agent, target: tuple[int, int]) -> str: """One greedy footprint-safe step reducing Manhattan distance to *target*.""" best, best_d = "stay", _manhattan(_center(a), target) for m in _legal_moves(a): dx, dy = _DELTAS[m] cc = (a.x + dx + a.size // 2, a.y + dy + a.size // 2) d = _manhattan(cc, target) if d < best_d: best, best_d = m, d return best def _apply(a: _Agent, action: str) -> None: dx, dy = _DELTAS[action] if _in_bounds(a.x + dx, a.y + dy, a.size): a.x, a.y = a.x + dx, a.y + dy def _apply_safe(a: _Agent, action: str, agents: list[_Agent]) -> None: """Commit a move only if in-bounds AND it does not overlap another alive agent's footprint. Agents are solid bodies; the predator is NOT in *agents* so it is never blocked (overlap with a distractor is the kill mechanic).""" dx, dy = _DELTAS[action] nx, ny = a.x + dx, a.y + dy if not _in_bounds(nx, ny, a.size): return for o in agents: if o is a or not o.alive: continue if (nx < o.x + o.size and o.x < nx + a.size and ny < o.y + o.size and o.y < ny + a.size): return # blocked by another agent -> stay this turn a.x, a.y = nx, ny def _frame(agents: list[_Agent], pred: _Agent, facing: str, resources, events, action: str, idx: int) -> MemoryTurn: frames = [AgentFrame(id=a.id, kind="agent", pos=(a.x, a.y), size=a.size, alive=a.alive, is_chosen=a.is_chosen) for a in agents] if pred is not None: frames.append(AgentFrame(id="predator", kind="predator", pos=(pred.x, pred.y), size=pred.size, facing=facing)) alive_n = sum(1 for a in agents if a.alive) return MemoryTurn( turn_idx=idx, frame_ascii=f"{alive_n} agents alive; predator at {(pred.x, pred.y) if pred else None}", action=action, focal_pos=(0, 0), predator_pos=(0, 0), # unused sentinels: memory_frames uses the agents branch agents=frames, resources=list(resources), events=list(events), ) def author_predator_chase( *, seed: int, agent_starts: list[tuple[int, int]], predator_start: tuple[int, int], agent_size: int = 2, predator_size: int = 3, free_turns: int = 24, predator_grace: int = 24, # predator roams (does not pursue) for this many turns, delaying the first kill panic_turns: int = 14, # after EACH kill all survivors flee for this many turns (the visible scatter), then resume roaming max_turns: int = 400, # grace + wider roaming + panic bursts lengthen the episode; seed=7 catches all 3 distractors by genuine contact within this budget persona_id: str = "risk_averse", ) -> MemoryCheckpoint: """Author scenario ①: pack flees a predator; one survivor (a0) remains. Pre-kill design (two levers): * **Longer preroll.** The predator holds back during a *grace* period (``predator_grace`` turns): instead of pursuing it merely roams, so the first genuine catch lands much later. ``free_turns`` (a0's pre-flee free-roam window) is widened to match. * **Wider, more varied roaming.** During the pre-catalyst phase every agent (distractors, and a0 in its free window) uses :func:`_roam` — a persistent seeded per-agent heading — instead of a uniform random walk, so the pack travels far in distinct directions and covers a wider area. Distractors stay NON-predator-aware (they never flee), preserving the "caught by genuine contact" contract. Post-kill SCATTER (panic burst). Once a distractor is eaten the whole pack scatters: for ``panic_turns`` turns AFTER each kill every surviving distractor flees the predator (maximising Manhattan distance), exactly like the chosen a0. This reproduces "when one agent is eaten, the whole pack moves away from the predator". The burst is finite: because the predator and the agents move at EQUAL speed, perfectly-fleeing distractors run to opposite field corners and deadlock — they can never be cornered. So after the burst expires the surviving distractors resume the (non-predator-aware) roam, which lets the predator close in and genuinely catch the next one; that next catch re-arms the burst, scattering the remaining survivors again. a0 (``is_chosen``, never killed) flees permanently once the catalyst fires. """ rng = random.Random(seed) agents = [_Agent(id=f"a{i}", x=p[0], y=p[1], size=agent_size, is_chosen=(i == 0)) for i, p in enumerate(agent_starts)] pred = _Agent(id="predator", x=predator_start[0], y=predator_start[1], size=predator_size) facing = "left" turns: list[MemoryTurn] = [] catalyst_done = False panic_ttl = 0 # >0 => survivors are mid-scatter (flee); re-armed on each kill # Per-agent + predator roam headings (seeded; ttl=0 forces a first roll). roam: dict[str, _RoamState] = {a.id: _RoamState(heading="stay", ttl=0) for a in agents} roam["predator"] = _RoamState(heading="stay", ttl=0) for t in range(1, max_turns + 1): events: list[str] = [] chosen = next(a for a in agents if a.is_chosen) # Record the PRE-move frame; the stored action is the chosen agent's. # a0 flees once the catalyst fires or its free window ends; otherwise it # roams widely (exploratory heading) rather than fleeing. chosen_action = (_flee(chosen, _center(pred)) if (t > free_turns or catalyst_done) else _roam(chosen, rng, roam[chosen.id])) turns.append(_frame(agents, pred, facing, [], events, chosen_action, t)) # --- predator-first resolution --- # During the grace window the predator roams (no pursuit), so distractors # are not yet hunted; afterwards it chases the nearest distractor. targets = [a for a in agents if a.alive and not a.is_chosen] pursuing = (t > predator_grace or catalyst_done) if pursuing and targets: nearest = min(targets, key=lambda a: _manhattan(_center(pred), _center(a))) move = _step_toward(pred, _center(nearest)) else: move = _roam(pred, rng, roam["predator"]) if move != "stay": facing = _FACING_FROM[move] _apply(pred, move) # chosen + distractors move (agent-agent collision avoidance; predator exempt) _apply_safe(chosen, chosen_action, agents) for a in agents: if a.alive and not a.is_chosen: # While a panic burst is active every surviving distractor FLEES # the predator (the scatter); otherwise it uses the wider, varied # roam (non-predator-aware) so the predator can close in for the # next genuine catch. Pre-catalyst there is never a burst, so this # is the original wide roam. a_action = (_flee(a, _center(pred)) if panic_ttl > 0 else _roam(a, rng, roam[a.id])) _apply_safe(a, a_action, agents) if panic_ttl > 0: panic_ttl -= 1 # kills (distractors only; chosen is the authored survivor) for a in agents: if a.alive and not a.is_chosen and _overlap(pred, a): a.alive = False catalyst_done = True panic_ttl = panic_turns # (re-)arm the scatter burst on every kill events.append(f"{a.id} eaten") # patch events onto the frame we just appended turns[-1].events = list(events) if sum(1 for a in agents if a.alive and not a.is_chosen) == 0: break # Safeguard: force-resolve any stragglers so exactly the chosen survives. for a in agents: if a.alive and not a.is_chosen: a.alive = False if turns: # Append a final settled frame (predator removed for clarity). turns.append(_frame(agents, pred, facing, [], ["only a0 survives"], "stay", len(turns) + 1)) return MemoryCheckpoint( model=f"director:{persona_id}", scenario="predator_chase", motive_category="survival", difficulty="easy", seed=seed, created_at=_STAMP, memory_turns=turns, outcome="survived", transparent_prompt="Pack-flee handover memory.", persona_weight_id=persona_id, chosen_agent_id="a0", ) def _covers(a: _Agent, cell: tuple[int, int]) -> bool: return (a.x <= cell[0] < a.x + a.size and a.y <= cell[1] < a.y + a.size) def author_resource_race( *, seed: int, agent_starts: list[tuple[int, int]], resource: tuple[int, int], agent_size: int = 2, max_turns: int = 120, persona_id: str = "greedy", ) -> MemoryCheckpoint: """Author scenario ②: a0 beelines to the lone resource; others wander.""" rng = random.Random(seed) agents = [_Agent(id=f"a{i}", x=p[0], y=p[1], size=agent_size, is_chosen=(i == 0)) for i, p in enumerate(agent_starts)] turns: list[MemoryTurn] = [] for t in range(1, max_turns + 1): chosen = next(a for a in agents if a.is_chosen) if _covers(chosen, resource): # Collected -> the resource leaves the field on this final frame. turns.append(_frame(agents, None, "right", [], ["a0 got resource"], "stay", t)) break action = _step_toward(chosen, resource) turns.append(_frame(agents, None, "right", [resource], [], action, t)) _apply_safe(chosen, action, agents) for a in agents: if not a.is_chosen: _apply_safe(a, _wander(a, rng), agents) return MemoryCheckpoint( model=f"director:{persona_id}", scenario="resource_race", motive_category="resource", difficulty="easy", seed=seed, created_at=_STAMP, memory_turns=turns, outcome="survived", transparent_prompt="Resource-race handover memory.", persona_weight_id=persona_id, chosen_agent_id="a0", ) def _seat_step(a: _Agent, target_anchor: tuple[int, int], fields: dict[tuple[int, int], dict]) -> str: """One table-aware step moving agent `a` toward `target_anchor` (BFS field).""" field = fields.get(target_anchor) if field is None: field = _seat.bfs_field(target_anchor, a.size, _seat.wall_cells()) fields[target_anchor] = field return _seat.step_toward_field((a.x, a.y), field) def author_take_a_seat( *, seed: int, agent_starts: list[tuple[int, int]], agent_size: int = 2, max_turns: int = 400, persona_id: str = "far_seat", ) -> MemoryCheckpoint: """Author the take_a_seat memory: norms grab the nearest seat, a0 circles to the far seat. a0 (chosen) follows a clockwise ring of corridor waypoints AROUND the table (the visible "circling"), then routes to the slot farthest from its start and seats there. a1..a4 each beeline (BFS, table-aware) to their nearest still- free seat — excluding a0's reserved far seat — and stay once seated. Fully deterministic given `agent_starts` (the geometry itself is seed-independent; `seed` is recorded on the checkpoint only). """ fields: dict[tuple[int, int], dict] = {} agents = [_Agent(id=f"a{i}", x=p[0], y=p[1], size=agent_size, is_chosen=(i == 0)) for i, p in enumerate(agent_starts)] chosen = agents[0] # a0 reserves the seat farthest from ITS start; norms may never take it. far = _seat.farthest_slot((chosen.x, chosen.y)) reserved = {far.anchor} # a0 waypoint queue: nearest ring corner -> full clockwise lap -> far seat. x0, y0, x1, y1 = _seat.TABLE ring = [(x0 - 7, y0 - 7), (x1 - 1 + 7, y0 - 7), (x1 - 1 + 7, y1 - 1 + 7), (x0 - 7, y1 - 1 + 7)] # TL, TR, BR, BL (anchors) start_i = min(range(4), key=lambda i: abs(ring[i][0] - chosen.x) + abs(ring[i][1] - chosen.y)) lap = [ring[(start_i + k) % 4] for k in range(5)] # full loop back to the start corner a0_waypoints = [*lap, far.entry, far.anchor] a0_wp = 0 # Per-norm assigned seat, chosen greedily by nearest free anchor (deterministic). claimed: set[tuple[int, int]] = set(reserved) norm_target: dict[str, tuple[int, int]] = {} for a in agents[1:]: free = [s for s in _seat.SLOTS if s.anchor not in claimed] pick = min(free, key=lambda s: abs(s.anchor[0] - a.x) + abs(s.anchor[1] - a.y)) norm_target[a.id] = pick.anchor claimed.add(pick.anchor) turns: list[MemoryTurn] = [] seats = _seat.seat_cells() def all_seated() -> bool: if (chosen.x, chosen.y) != far.anchor: return False return all((a.x, a.y) == norm_target[a.id] for a in agents[1:]) seated: set[str] = set() for t in range(1, max_turns + 1): events: list[str] = [] # advance a0's waypoint cursor when the current target is reached if a0_wp < len(a0_waypoints) - 1 and (chosen.x, chosen.y) == a0_waypoints[a0_wp]: a0_wp += 1 a0_action = _seat_step(chosen, a0_waypoints[a0_wp], fields) turns.append(_frame(agents, None, "right", seats, events, a0_action, t)) # commit moves (agent-agent collision avoidance; a0 then norms) _apply_safe(chosen, a0_action, agents) for a in agents[1:]: if (a.x, a.y) == norm_target[a.id]: continue # seated; stay put _apply_safe(a, _seat_step(a, norm_target[a.id], fields), agents) # seating events for narration (emit once, on arrival) for a in agents: tgt = far.anchor if a.is_chosen else norm_target[a.id] if a.id not in seated and (a.x, a.y) == tgt: events.append(f"{a.id} seated") seated.add(a.id) turns[-1].events = list(events) if all_seated(): break final_events = ["all seated"] if all_seated() else ["incomplete: max_turns reached"] turns.append(_frame(agents, None, "right", seats, final_events, "stay", len(turns) + 1)) return MemoryCheckpoint( model=f"director:{persona_id}", scenario="take_a_seat", motive_category="seating", difficulty="easy", seed=seed, created_at=_STAMP, memory_turns=turns, outcome="survived", transparent_prompt="Take-a-seat handover memory.", persona_weight_id=persona_id, chosen_agent_id="a0", wall_rects=_seat.wall_rects(), ) # --------------------------------------------------------------------------- # # errand_runner director (visual-only city errand; single-agent persona demo) # --------------------------------------------------------------------------- # def _errand_memory_caption(chosen: _Agent, ped: _Agent, ped_rescued: bool, wallet_present: bool, tick: int) -> str: """One-line text-path summary of the persona's state this memory tick. The web replay reconstructs a colour grid from ``agents``/``cells``, but the LLM only ever sees ``frame_ascii`` (via ``render_memory_block``). A bare ``"errand tN"`` placeholder gave the model zero spatial grounding, so the handover "watch how you handled the crosswalk/wallet/pedestrian" promise was empty. This packs the courier's position plus the salient decision context (where the persona's behaviour actually diverges) into a single compact line — full per-turn ASCII maps would blow the prompt (routes run ~80-120 turns). Memory runs on ``MEMORY_LAYOUT`` (a *similar*, not identical, city to the live ``GAME_LAYOUT``), so the coordinates convey trajectory shape while the flags carry the transferable behavioural lesson. """ from proteus.game.scenarios import errand_world as w lay = w.MEMORY_LAYOUT fp = set(w.footprint((chosen.x, chosen.y), chosen.size)) flags: list[str] = [] if fp & set(w.crosswalk_cells(lay)): flags.append("crosswalk-" + ("RED" if w.is_red(tick) else "GREEN")) if fp & w.constr_cells(lay): flags.append("construction") if fp & w.grass_cells(lay): flags.append("grass-shortcut") if not ped_rescued and _manhattan(_center(chosen), _center(ped)) <= w.PED_HELP_RADIUS: flags.append("by-fallen-pedestrian") suffix = (" " + " ".join(flags)) if flags else "" return f"t{tick} you@({chosen.x},{chosen.y}){suffix}" def _errand_frame_single(chosen: _Agent, ped: _Agent, ped_rescued: bool, wallet_present: bool, tick: int, action: str, events: list[str], idx: int) -> MemoryTurn: from proteus.game.scenarios import errand_world as w frames = [AgentFrame(id=chosen.id, kind="agent", pos=(chosen.x, chosen.y), size=chosen.size, alive=True, is_chosen=True)] frames.append(AgentFrame( id="pedestrian", kind=("npc_active" if ped_rescued else "npc_down"), pos=(ped.x, ped.y), size=ped.size)) return MemoryTurn( turn_idx=idx, frame_ascii=_errand_memory_caption(chosen, ped, ped_rescued, wallet_present, tick), action=action, focal_pos=(0, 0), predator_pos=(0, 0), agents=frames, cells=w.overlay_cells(w.MEMORY_LAYOUT, tick, wallet_present=wallet_present), resources=[], events=list(events)) def _errand_route_action(a: _Agent, tick: int, wallet_present: bool, ped: _Agent, ped_rescued: bool, policy: dict, home_field: dict, wallet_field: dict, walls: frozenset[tuple[int, int]], lay: "w.WorldLayout") -> str: """The chosen agent's persona-congruent BFS-routed action this tick.""" from proteus.game.scenarios import errand_world as w if (policy["pedestrian"] == "help" and not ped_rescued and _manhattan(_center(a), _center(ped)) <= w.PED_HELP_RADIUS): return "interact" if (policy["wallet"] == "grab" and wallet_present and _manhattan(_center(a), lay.wallet) <= w.GRAB_RADIUS): return w.step_toward_field((a.x, a.y), wallet_field) home_step = w.step_toward_field((a.x, a.y), home_field) cw = set(w.crosswalk_cells(lay)) nxt = (a.x + _DELTAS[home_step][0], a.y + _DELTAS[home_step][1]) if policy["light"] == "wait" and w.is_red(tick) and (set(w.footprint(nxt, a.size)) & cw): return "stay" return home_step def author_errand_runner(*, seed: int, persona: str | None = None, max_turns: int = 300) -> MemoryCheckpoint: """Author a single-agent errand memory on MEMORY_LAYOUT for one persona. The lone courier BFS-routes from start to the brown house (the goal), resolving crosswalk / construction / grass / wallet / pedestrian per *persona* (default: the seed-chosen persona, matching the live scenario). The routing field mirrors the live scenario's per-persona ``_policy_field``: grass-'avoid' personas (civic) treat the lawn as impassable and route the longer road-only way around it; grass-'cut' personas (warm_outlaw, opportunist) keep the plain field and BFS naturally takes the grass shortcut. construction-'detour' personas likewise treat the construction block as impassable so BFS takes the genuine bypass instead of oscillating at its edge; 'pass' personas route straight through.""" from proteus.game.scenarios import errand_world as w lay = w.MEMORY_LAYOUT r = random.Random(seed) _true_index, seed_persona = w.session_choices(r) persona_id = persona or seed_persona policy = w.PERSONAS[persona_id] walls = w.wall_cells(lay) home_anchor = w.home_target_anchor(lay) extra: set[tuple[int, int]] = set() if policy["grass"] == "avoid": extra |= set(w.grass_cells(lay)) if policy["construction"] == "detour": extra |= w.constr_cells(lay) home_field = w.bfs_field(home_anchor, w.AGENT, walls | frozenset(extra), lay.grid) wallet_field = w.bfs_field((lay.wallet[0], lay.wallet[1]), w.AGENT, walls, lay.grid) chosen = _Agent(id="a0", x=lay.start[0], y=lay.start[1], size=w.AGENT, is_chosen=True) ped = _Agent(id="pedestrian", x=lay.pedestrian[0], y=lay.pedestrian[1], size=w.AGENT) ped_rescued, wallet_present = False, True turns: list[MemoryTurn] = [] for t in range(1, max_turns + 1): action = _errand_route_action(chosen, t, wallet_present, ped, ped_rescued, policy, home_field, wallet_field, walls, lay) turns.append(_errand_frame_single(chosen, ped, ped_rescued, wallet_present, t, action, [], t)) if action == "interact": if not ped_rescued and _manhattan(_center(chosen), _center(ped)) <= w.PED_HELP_RADIUS: ped_rescued = True elif action != "stay": dx, dy = _DELTAS[action] cand = (chosen.x + dx, chosen.y + dy) if w.footprint_free(cand, chosen.size, walls, lay.grid): chosen.x, chosen.y = cand if wallet_present and (lay.wallet in set(w.footprint((chosen.x, chosen.y), chosen.size))): wallet_present = False # Tag the move that landed on the wallet so the text-path memory shows # the pickup (the frame was recorded pre-move, so the footprint flag # could never catch it — back-annotate the just-appended turn instead). turns[-1].frame_ascii += " grabbed-wallet" if ped_rescued: # rescued pedestrian drifts near origin order = ["up", "down", "left", "right"] order = order[t % 4:] + order[:t % 4] for m in order: cand = (ped.x + _DELTAS[m][0], ped.y + _DELTAS[m][1]) if (w.footprint_free(cand, ped.size, walls, lay.grid) and _manhattan(cand, lay.pedestrian) <= w.PED_WANDER_RADIUS): ped.x, ped.y = cand break if w.in_home((chosen.x, chosen.y), lay): turns.append(_errand_frame_single(chosen, ped, ped_rescued, wallet_present, t + 1, "stay", ["arrived home"], t + 1)) break return MemoryCheckpoint( model=f"director:{persona_id}", scenario="errand_runner", motive_category="errand", difficulty="easy", seed=seed, created_at=_STAMP, memory_turns=turns, outcome="survived", transparent_prompt="Errand-home handover memory (single-agent persona demo).", persona_weight_id=persona_id, chosen_agent_id="a0", wall_rects=w.wall_rects(lay)) def author_errand_runner_variants(*, seed: int) -> dict[str, MemoryCheckpoint]: """All three persona memories on the same MEMORY_LAYOUT, keyed by persona id.""" from proteus.game.scenarios import errand_world as w return {pid: author_errand_runner(seed=seed, persona=pid) for pid in w.PERSONAS}