irregular6612 Claude Opus 4.8 (1M context) commited on
Commit
957367c
·
1 Parent(s): 6225269

feat(director): predator_chase memory — longer pre-kill preroll + wider, varied agent roaming

Browse files
proteus/game/runtime/multiagent_director.py CHANGED
@@ -64,6 +64,35 @@ def _wander(a: _Agent, rng: random.Random) -> str:
64
  return rng.choice(legal) if legal else "stay"
65
 
66
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
67
  def _flee(a: _Agent, threat: tuple[int, int]) -> str:
68
  """Pick the legal move that MAXIMISES Manhattan distance from *threat*."""
69
  best, best_d = "stay", -1
@@ -128,11 +157,26 @@ def _frame(agents: list[_Agent], pred: _Agent, facing: str,
128
 
129
  def author_predator_chase(
130
  *, seed: int, agent_starts: list[tuple[int, int]], predator_start: tuple[int, int],
131
- agent_size: int = 2, predator_size: int = 3, free_turns: int = 8,
132
- max_turns: int = 160, # >80 plan default: seed=7 distractors are all caught by genuine contact at ~turn 124; smaller budgets trip the safeguard
 
133
  persona_id: str = "risk_averse",
134
  ) -> MemoryCheckpoint:
135
- """Author scenario ①: pack flees a predator; one survivor (a0) remains."""
 
 
 
 
 
 
 
 
 
 
 
 
 
 
136
  rng = random.Random(seed)
137
  agents = [_Agent(id=f"a{i}", x=p[0], y=p[1], size=agent_size, is_chosen=(i == 0))
138
  for i, p in enumerate(agent_starts)]
@@ -140,28 +184,40 @@ def author_predator_chase(
140
  facing = "left"
141
  turns: list[MemoryTurn] = []
142
  catalyst_done = False
 
 
 
143
 
144
  for t in range(1, max_turns + 1):
145
  events: list[str] = []
146
  chosen = next(a for a in agents if a.is_chosen)
147
  # Record the PRE-move frame; the stored action is the chosen agent's.
 
 
148
  chosen_action = (_flee(chosen, _center(pred)) if (t > free_turns or catalyst_done)
149
- else _wander(chosen, rng))
150
  turns.append(_frame(agents, pred, facing, [], events, chosen_action, t))
151
 
152
  # --- predator-first resolution ---
 
 
153
  targets = [a for a in agents if a.alive and not a.is_chosen]
154
- if targets:
 
155
  nearest = min(targets, key=lambda a: _manhattan(_center(pred), _center(a)))
156
  move = _step_toward(pred, _center(nearest))
157
- if move != "stay":
158
- facing = _FACING_FROM[move]
159
- _apply(pred, move)
 
 
160
  # chosen + distractors move (agent-agent collision avoidance; predator exempt)
161
  _apply_safe(chosen, chosen_action, agents)
162
  for a in agents:
163
  if a.alive and not a.is_chosen:
164
- _apply_safe(a, _wander(a, rng), agents)
 
 
165
  # kills (distractors only; chosen is the authored survivor)
166
  for a in agents:
167
  if a.alive and not a.is_chosen and _overlap(pred, a):
 
64
  return rng.choice(legal) if legal else "stay"
65
 
66
 
67
+ @dataclass
68
+ class _RoamState:
69
+ """Persistent per-agent heading for exploratory roaming."""
70
+ heading: str
71
+ ttl: int # steps remaining before the heading is re-rolled
72
+
73
+
74
+ def _roam(a: _Agent, rng: random.Random, state: _RoamState,
75
+ run_min: int = 6, run_max: int = 12) -> str:
76
+ """Exploratory roam: follow a persistent, seeded per-agent heading for a
77
+ run of steps, then re-roll it. Headings persist (unlike :func:`_wander`'s
78
+ fresh uniform draw each turn), so each agent travels far in a distinct
79
+ direction and the pack covers a much wider area. Fully deterministic given
80
+ *rng*. The agent stays NON-predator-aware (heading ignores the predator).
81
+
82
+ Re-rolls when the run expires OR the current heading is blocked (boundary
83
+ or another agent), choosing a fresh legal-ish heading and resetting the run.
84
+ """
85
+ legal = _legal_moves(a)
86
+ if not legal:
87
+ return "stay"
88
+ # Re-roll on expiry or when the committed heading is no longer in-bounds.
89
+ if state.ttl <= 0 or state.heading not in legal:
90
+ state.heading = rng.choice(legal)
91
+ state.ttl = rng.randint(run_min, run_max)
92
+ state.ttl -= 1
93
+ return state.heading
94
+
95
+
96
  def _flee(a: _Agent, threat: tuple[int, int]) -> str:
97
  """Pick the legal move that MAXIMISES Manhattan distance from *threat*."""
98
  best, best_d = "stay", -1
 
157
 
158
  def author_predator_chase(
159
  *, seed: int, agent_starts: list[tuple[int, int]], predator_start: tuple[int, int],
160
+ agent_size: int = 2, predator_size: int = 3, free_turns: int = 24,
161
+ predator_grace: int = 24, # predator roams (does not pursue) for this many turns, delaying the first kill
162
+ max_turns: int = 260, # grace + wider roaming lengthen the episode; seed=7 catches all 3 distractors by genuine contact well within this budget
163
  persona_id: str = "risk_averse",
164
  ) -> MemoryCheckpoint:
165
+ """Author scenario ①: pack flees a predator; one survivor (a0) remains.
166
+
167
+ Pre-kill design (two levers):
168
+
169
+ * **Longer preroll.** The predator holds back during a *grace* period
170
+ (``predator_grace`` turns): instead of pursuing it merely roams, so the
171
+ first genuine catch lands much later. ``free_turns`` (a0's pre-flee
172
+ free-roam window) is widened to match.
173
+ * **Wider, more varied roaming.** During the pre-catalyst phase every agent
174
+ (distractors, and a0 in its free window) uses :func:`_roam` — a persistent
175
+ seeded per-agent heading — instead of a uniform random walk, so the pack
176
+ travels far in distinct directions and covers a wider area. Distractors
177
+ stay NON-predator-aware (they never flee), preserving the "caught by
178
+ genuine contact" contract.
179
+ """
180
  rng = random.Random(seed)
181
  agents = [_Agent(id=f"a{i}", x=p[0], y=p[1], size=agent_size, is_chosen=(i == 0))
182
  for i, p in enumerate(agent_starts)]
 
184
  facing = "left"
185
  turns: list[MemoryTurn] = []
186
  catalyst_done = False
187
+ # Per-agent + predator roam headings (seeded; ttl=0 forces a first roll).
188
+ roam: dict[str, _RoamState] = {a.id: _RoamState(heading="stay", ttl=0) for a in agents}
189
+ roam["predator"] = _RoamState(heading="stay", ttl=0)
190
 
191
  for t in range(1, max_turns + 1):
192
  events: list[str] = []
193
  chosen = next(a for a in agents if a.is_chosen)
194
  # Record the PRE-move frame; the stored action is the chosen agent's.
195
+ # a0 flees once the catalyst fires or its free window ends; otherwise it
196
+ # roams widely (exploratory heading) rather than fleeing.
197
  chosen_action = (_flee(chosen, _center(pred)) if (t > free_turns or catalyst_done)
198
+ else _roam(chosen, rng, roam[chosen.id]))
199
  turns.append(_frame(agents, pred, facing, [], events, chosen_action, t))
200
 
201
  # --- predator-first resolution ---
202
+ # During the grace window the predator roams (no pursuit), so distractors
203
+ # are not yet hunted; afterwards it chases the nearest distractor.
204
  targets = [a for a in agents if a.alive and not a.is_chosen]
205
+ pursuing = (t > predator_grace or catalyst_done)
206
+ if pursuing and targets:
207
  nearest = min(targets, key=lambda a: _manhattan(_center(pred), _center(a)))
208
  move = _step_toward(pred, _center(nearest))
209
+ else:
210
+ move = _roam(pred, rng, roam["predator"])
211
+ if move != "stay":
212
+ facing = _FACING_FROM[move]
213
+ _apply(pred, move)
214
  # chosen + distractors move (agent-agent collision avoidance; predator exempt)
215
  _apply_safe(chosen, chosen_action, agents)
216
  for a in agents:
217
  if a.alive and not a.is_chosen:
218
+ # Wider, varied pre-kill roam; distractors stay non-predator-aware.
219
+ a_action = _roam(a, rng, roam[a.id]) if not catalyst_done else _wander(a, rng)
220
+ _apply_safe(a, a_action, agents)
221
  # kills (distractors only; chosen is the authored survivor)
222
  for a in agents:
223
  if a.alive and not a.is_chosen and _overlap(pred, a):
tests/runtime/test_predator_chase_memory.py ADDED
@@ -0,0 +1,86 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ # tests/runtime/test_predator_chase_memory.py
2
+ """Redesigned predator_chase memory: longer pre-kill preroll + wider, more
3
+ varied pre-kill roaming.
4
+
5
+ Baseline (pre-redesign, seed 7):
6
+ - first "eaten" event at turn_idx 71
7
+ - max distractor Manhattan displacement from start during the pre-kill phase
8
+ was only 12 cells (uniform random walk drifts back near the start).
9
+
10
+ These tests pin measurable improvements:
11
+ - the first kill must land meaningfully later (preroll floor),
12
+ - distractors must travel noticeably farther from their starts before the
13
+ first kill (wider action radius).
14
+
15
+ Hard invariants (a0 survivor, 3 genuine eaten, determinism) are asserted in
16
+ test_director_predator_chase.py and reinforced here as guardrails.
17
+ """
18
+ from proteus.game.runtime.multiagent_director import author_predator_chase
19
+
20
+ _STARTS = [(10, 30), (14, 38), (18, 24), (12, 46)]
21
+ _PRED = (54, 31)
22
+
23
+ # Justified thresholds vs baseline (seed 7):
24
+ # first-eat baseline = 71 -> require >= 100 (the redesign lands it at 104,
25
+ # +33 turns / ~1.46x later; the floor sits a few turns below the achieved
26
+ # value because first-eat is non-monotonic in the grace/roam interaction).
27
+ # max pre-kill distractor displacement baseline = 12 -> require >= 20
28
+ # (the persistent-heading roam achieves 40, vs the random walk's 12).
29
+ _FIRST_EAT_FLOOR = 100
30
+ _SPREAD_FLOOR = 20
31
+
32
+
33
+ def _run():
34
+ return author_predator_chase(seed=7, agent_starts=_STARTS, predator_start=_PRED)
35
+
36
+
37
+ def _first_eat_turn(ck):
38
+ for t in ck.memory_turns:
39
+ for e in t.events:
40
+ if "eaten" in e:
41
+ return t.turn_idx
42
+ return None
43
+
44
+
45
+ def test_first_kill_happens_later():
46
+ """First genuine 'eaten' event must occur at a clearly later turn."""
47
+ first = _first_eat_turn(_run())
48
+ assert first is not None, "expected at least one 'eaten' event"
49
+ assert first >= _FIRST_EAT_FLOOR, (
50
+ f"first eat at turn {first}, want >= {_FIRST_EAT_FLOOR}"
51
+ )
52
+
53
+
54
+ def test_distractors_roam_wider_before_first_kill():
55
+ """Max distractor displacement from start during the pre-kill phase must
56
+ clearly exceed the uniform-random-walk baseline."""
57
+ ck = _run()
58
+ first = _first_eat_turn(ck)
59
+ assert first is not None
60
+ starts = {f"a{i}": p for i, p in enumerate(_STARTS) if i != 0}
61
+ max_disp = 0
62
+ for t in ck.memory_turns:
63
+ if t.turn_idx >= first:
64
+ break
65
+ for a in t.agents:
66
+ if a.id in starts:
67
+ s = starts[a.id]
68
+ d = abs(a.pos[0] - s[0]) + abs(a.pos[1] - s[1])
69
+ max_disp = max(max_disp, d)
70
+ assert max_disp >= _SPREAD_FLOOR, (
71
+ f"max pre-kill distractor displacement {max_disp}, want >= {_SPREAD_FLOOR}"
72
+ )
73
+
74
+
75
+ def test_guardrails_survivor_and_three_genuine_eaten():
76
+ """a0 is the sole survivor, alive every turn, and exactly 3 distractors are
77
+ eaten by GENUINE predator contact (not the silent end safeguard)."""
78
+ ck = _run()
79
+ for t in ck.memory_turns:
80
+ chosen = next(a for a in t.agents if a.id == "a0")
81
+ assert chosen.alive
82
+ eaten = [e for t in ck.memory_turns for e in t.events if "eaten" in e]
83
+ assert len(eaten) == 3, f"expected 3 genuine eaten events, got {eaten}"
84
+ last = ck.memory_turns[-1]
85
+ alive = [a for a in last.agents if a.kind == "agent" and a.alive]
86
+ assert len(alive) == 1 and alive[0].id == "a0"