[Fellowship Challenge] Expiring memory (ttl.v1) for the memory layer

src/nandatown/layers/__init__.py

@@ -89,6 +89,7 @@ def plugins() -> dict[str, list[dict[str, Any]]]:
     data_facts,
     identity,
     memory,
+    memory_ttl,
     negotiation,
     payments,
     privacy,

src/nandatown/layers/memory_ttl.py

@@ -0,0 +1,61 @@
+"""Memory layer: expiring memory, declared keys stop being served after a limit.
+
+The limit is set per agent in the scenario: memory_ttl: {key: ticks}.
+Events carry keys and times only, never values.
+"""
+
+from __future__ import annotations
+
+from . import register
+from .memory import KeyValueMemory
+
+
+@register("memory", "ttl.v1")
+class ExpiringMemory(KeyValueMemory):
+    """Per-agent memory whose declared keys stop being returned after a limit."""
+
+    enforce = True
+
+    def __init__(self, engine):
+        super().__init__(engine)
+        self.expires_at: dict[tuple[str, str], float] = {}
+
+    def _limit(self, agent, key):
+        for a in self.engine.spec.agents:
+            if a.name == agent:
+                return a.config.get("memory_ttl", {}).get(key)
+        return None
+
+    def remember(self, agent, key, value):
+        self.stores.setdefault(agent, {})[key] = value
+        detail = {"key": key}
+        limit = self._limit(agent, key)
+        if limit is None:
+            self.expires_at.pop((agent, key), None)
+        else:
+            deadline = self.engine.now + limit
+            self.expires_at[(agent, key)] = deadline
+            detail["expires_at"] = deadline
+        self.engine.emit(agent, "memory_written", agent, detail)
+
+    def recall(self, agent, key):
+        deadline = self.expires_at.get((agent, key))
+        if deadline is None:
+            return super().recall(agent, key)
+        stale = self.engine.now >= deadline
+        if stale and self.enforce:
+            self.stores.get(agent, {}).pop(key, None)
+            del self.expires_at[(agent, key)]
+            self.engine.emit(agent, "memory_expired", agent,
+                             {"key": key, "expires_at": deadline})
+            return None
+        self.engine.emit(agent, "memory_recalled", agent,
+                         {"key": key, "expires_at": deadline, "stale": stale})
+        return self.stores.get(agent, {}).get(key)
+
+
+@register("memory", "ttl_off.v1")
+class ExpiringMemoryNotEnforced(ExpiringMemory):
+    """Negative control: records the limit but still returns expired entries."""
+
+    enforce = False

src/nandatown/report.py

@@ -42,6 +42,7 @@
     "settlement": "money moved only through recorded escrow",
     "reputation": "receipts drove the public score",
     "memory_reuse": "a remembered counterparty replaced a fresh lookup",
+    "memory_expiry": "an expired memory entry was never served; the buyer rediscovered",
     "announced": "the task was announced with its award rule",
     "bidding": "on-time bids counted, late bids rejected",
     "award": "the award followed the declared rule",

src/nandatown/sim/scenarios/marketplace_expiring_memory.yaml

@@ -0,0 +1,21 @@
+name: marketplace_expiring_memory
+description: Marketplace where the buyer's remembered seller expires between rounds, so round two must rediscover through the index instead of reusing stale memory.
+seed: 42
+layers:
+  memory: ttl.v1
+agents:
+  - name: seller-a
+    role: seller
+    config: {sku: widget, ask_cents: 2100, floor_cents: 1800, stock: 10, balance_cents: 0}
+  - name: seller-b
+    role: seller
+    config: {sku: widget, ask_cents: 1995, floor_cents: 1700, stock: 10, balance_cents: 0}
+  - name: buyer-1
+    role: buyer
+    config: {sku: widget, quantity: 2, cap_cents: 1900, balance_cents: 10000, rounds: 2, budget_cents: 10000, memory_ttl: {preferred_seller: 0.5}}
+faults:
+  - action: duplicate
+    kind: delivery
+    nth: 1
+redact_fields: [budget_cents]
+max_time: 60

src/nandatown/sim/validators.py

@@ -1508,3 +1508,34 @@ def evaluate_scenario(spec, run_id: str,
                           evaluator_version=LAB_EVALUATOR_VERSION,
                           stages=stages, verdict=stage_verdict(stages),
                           evaluated_at=time.time())
+
+
+@validator("marketplace_expiring_memory")
+def marketplace_expiring_memory(spec, trace: Trace) -> list[StageResult]:
+    """Marketplace with expiring memory: an expired entry is never used."""
+    kept = [s for s in marketplace(spec, trace)
+            if s.name not in ("discovery", "memory_reuse")]
+    registered = trace.ids("card_registered")
+    quote_requests = trace.find("message_sent", kind="quote_request")
+    discovery = _check(
+        "discovery", len(registered) >= 3 and len(quote_requests) >= 3,
+        registered + [q.event_id for q in quote_requests],
+        "expected both sellers plus buyer registered and at least three"
+        " quote requests")
+
+    stale_reads = [e for e in trace.find("memory_recalled")
+                   if e.detail.get("stale") is True]
+    expired = trace.ids("memory_expired")
+    if stale_reads:
+        expiry = _failed(
+            "memory_expiry", [e.event_id for e in stale_reads],
+            "the buyer recalled a memory entry after its expiry time")
+    elif not expired:
+        expiry = _missing(
+            "memory_expiry",
+            "no memory entry expired, so the rule was never exercised")
+    else:
+        expiry = _passed(
+            "memory_expiry", expired,
+            "an expired entry was dropped and never recalled")
+    return [discovery] + kept + [expiry]

tests/test_memory_ttl.py

@@ -0,0 +1,54 @@
+"""Expiring memory (memory: ttl.v1): an expired entry is never served."""
+
+from nandatown.sim.runner import build_engine
+from nandatown.sim.scenario import load_bundled
+from nandatown.sim.validators import evaluate_scenario
+
+SCENARIO = "marketplace_expiring_memory"
+
+
+def _run(memory_plugin=None):
+    spec = load_bundled(SCENARIO)
+    if memory_plugin:
+        spec.layers["memory"] = memory_plugin
+    engine = build_engine(spec)
+    engine.run()
+    return engine, evaluate_scenario(spec, engine.run_id, engine.events)
+
+
+def _stage(result, name):
+    return next(s for s in result.stages if s.name == name)
+
+
+def test_entry_served_before_expiry_and_dropped_after():
+    engine = build_engine(load_bundled(SCENARIO))
+    mem = engine.layers["memory"]
+    engine.now = 1.0
+    mem.remember("buyer-1", "preferred_seller", "seller-b")
+    engine.now = 1.4
+    assert mem.recall("buyer-1", "preferred_seller") == "seller-b"
+    engine.now = 1.5
+    assert mem.recall("buyer-1", "preferred_seller") is None
+    assert any(e.kind == "memory_expired" for e in engine.events)
+
+
+def test_key_without_a_declared_limit_never_expires():
+    engine = build_engine(load_bundled(SCENARIO))
+    mem = engine.layers["memory"]
+    engine.now = 1.0
+    mem.remember("buyer-1", "other_key", "x")
+    engine.now = 999.0
+    assert mem.recall("buyer-1", "other_key") == "x"
+
+
+def test_marketplace_rediscovers_after_expiry():
+    engine, result = _run()
+    assert _stage(result, "memory_expiry").status == "passed"
+    assert any(e.kind == "memory_expired" for e in engine.events)
+
+
+def test_negative_control_serves_the_stale_entry_and_fails():
+    engine, result = _run("ttl_off.v1")
+    assert _stage(result, "memory_expiry").status == "failed"
+    assert any(e.kind == "memory_recalled" and e.detail.get("stale")
+               for e in engine.events)

tests/test_sim.py

@@ -10,11 +10,13 @@
     load_bundled,
 )
 
-ALL_SCENARIOS = ["marketplace", "auction", "voting", "consensus",
-                 "supply_chain", "capability_spoofing", "capped_influence",
+ALL_SCENARIOS = ["marketplace", "marketplace_expiring_memory", "auction",
+                 "voting", "consensus", "supply_chain",
+                 "capability_spoofing", "capped_influence",
                  "registry_eviction"]
 FAILING_SCENARIOS = ["capability_spoofing_weak_auth",
-                     "capped_influence_uncapped_control", "registry_eviction_unowned"]
+                     "capped_influence_uncapped_control",
+                     "registry_eviction_unowned"]
 
 
 def trace_of(spec):