Spaces:
Running on Zero
Running on Zero
File size: 4,906 Bytes
32a601f f6566bb 32a601f f6566bb 32a601f ba6dd5f a2ca0e0 ba6dd5f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | from __future__ import annotations
from src.core.events import Event
from src.core.memory import EpisodicMemory
def _event(kind: str, actor: str = "x", turn: int = 1) -> Event:
return Event(run_id="r", turn=turn, kind=kind, actor=actor, payload={"text": f"{actor}:{kind}"}) # type: ignore[arg-type]
class TestEpisodicMemory:
def test_own_events_visible(self):
mem = EpisodicMemory("seedkeeper")
events = (_event("agent.spoke", actor="seedkeeper"),)
visible = mem.visible(events)
assert len(visible) == 1
def test_world_observed_visible_to_all(self):
mem = EpisodicMemory("pocket-actor")
events = (_event("world.observed", actor="scene-whisperer"),)
visible = mem.visible(events)
assert len(visible) == 1
def test_other_agent_spoke_is_visible(self):
# Peers' SPOKEN lines are the shared table — recallable across the whole run, not
# just this round's blackboard tail (ADR-0023 follow-up). A late-firing judge
# depends on this to read the discussion it rules on.
mem = EpisodicMemory("pocket-actor")
events = (_event("agent.spoke", actor="scene-whisperer"),)
assert len(mem.visible(events)) == 1
def test_other_agent_thought_not_visible(self):
# A private thought (the mind-reader content) rides only its own event — peers
# never read another mind's thinking, even though spoken lines are now shared.
mem = EpisodicMemory("pocket-actor")
events = (_event("agent.thought", actor="scene-whisperer"),)
assert len(mem.visible(events)) == 0
def test_oracle_spoke_visible_to_all(self):
# The custom public-speech kind (oracle-grove) is shared like agent.spoke.
mem = EpisodicMemory("scene-whisperer")
events = (_event("oracle.spoke", actor="fortune-teller"),)
assert len(mem.visible(events)) == 1
def test_user_injected_visible_to_all(self):
mem = EpisodicMemory("echo")
events = (_event("user.injected", actor="visitor"),)
visible = mem.visible(events)
assert len(visible) == 1
def test_capped_at_max_recent(self):
mem = EpisodicMemory("x", max_recent=3)
events = tuple(_event("world.observed", turn=i) for i in range(10))
visible = mem.visible(events)
assert len(visible) == 3
def test_returns_most_recent(self):
mem = EpisodicMemory("x", max_recent=2)
events = tuple(_event("world.observed", turn=i) for i in range(5))
visible = mem.visible(events)
assert visible[0].turn == 3
assert visible[1].turn == 4
def test_format_for_prompt_returns_string(self):
mem = EpisodicMemory("x")
events = (_event("world.observed", actor="narrator"),)
result = mem.format_for_prompt(events)
assert isinstance(result, str)
def test_format_empty_returns_placeholder(self):
mem = EpisodicMemory("x")
result = mem.format_for_prompt(())
assert "no prior" in result.lower() or result
def test_run_started_renders_goal_not_raw_payload(self):
# run.started carries {seed, goal} and is globally visible; the old formatter
# dumped str(payload) — leaking the raw seed into every prompt. Now it renders
# the shared goal only, never the seed dict.
events = (
Event(
run_id="r",
turn=0,
kind="run.started",
actor="conductor",
payload={"seed": "s3cr3t-seed", "goal": "catch the spy"}, # type: ignore[arg-type]
),
Event(run_id="r", turn=1, kind="agent.spoke", actor="x", payload={"text": "hello"}), # type: ignore[arg-type]
)
out = EpisodicMemory("x").format_for_prompt(events)
assert "s3cr3t-seed" not in out
assert "catch the spy" in out
assert "hello" in out
class _RaisingIndex:
"""A memory index whose backend is down — every call throws."""
def index(self, events):
raise RuntimeError("backend down")
def search(self, query, k, run_id=None):
raise RuntimeError("backend down")
class TestSalienceIndexResilience:
def test_index_failure_degrades_to_keyword(self):
# ADR-0018: the index is a derived, rebuildable lens, never load-bearing.
# A flaky backend must degrade to keyword relevance, not crash the turn —
# this is what kept the salience-using spy agents silent on the live path.
from src.core.memory import SalienceMemory
events = (Event(run_id="r", turn=1, kind="agent.spoke", actor="a", payload={"text": "warm fuel"}),) # type: ignore[arg-type]
mem = SalienceMemory("a", top_k=3, index=_RaisingIndex())
out = mem.format_for_prompt(events, current_turn=1, query="warm") # must not raise
assert "warm fuel" in out
|