Spaces:
Running on Zero
Running on Zero
| 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 | |