Audit: fix correctness, accounting and robustness bugs across the testbed

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "verdi-agents"
-version = "0.1.1"
+version = "0.2.0"
 description = "Auditable testbed for temporal value expression and institutional recovery in LLM agent societies"
 readme = "README.md"
 license = "MIT"

src/verdi/__init__.py

@@ -1,3 +1,3 @@
 """Verdi — auditable testbed for value expression and institutional recovery in LLM agent societies."""
 
-__version__ = "0.1.1"
+__version__ = "0.2.0"

src/verdi/agents/client.py

@@ -25,6 +25,8 @@
 RETRYABLE_STATUS = (429, 502, 503, 524)
 _BACKOFF_BASE = 0.5
 _BACKOFF_CAP = 10.0
+_CONNECT_TIMEOUT = 15.0
+_POOL_TIMEOUT = 30.0
 
 
 class ModelError(RuntimeError):
@@ -66,6 +68,11 @@ def __init__(
     ) -> None:
         self._config = config
         self._rng = rng or random.Random()
+        #: Successful completions so far. A liveness signal for the stall
+        #: watchdog: batched phases (reflection, consolidation) only write
+        #: model_calls.jsonl after the whole batch, so file lines alone can
+        #: stay flat for a long, healthy batch.
+        self.completed_calls = 0
         self._client = httpx.AsyncClient(
             base_url=config.base_url,
             headers={"Authorization": f"Bearer {config.api_key}"},
@@ -80,8 +87,8 @@ def __init__(
             # socket) always trips the read timeout and becomes retryable,
             # rather than blocking indefinitely.
             timeout=httpx.Timeout(
-                connect=15.0, read=config.timeout_seconds,
-                write=config.timeout_seconds, pool=30.0,
+                connect=_CONNECT_TIMEOUT, read=config.timeout_seconds,
+                write=config.timeout_seconds, pool=_POOL_TIMEOUT,
             ),
             # Bounded pool sized to the concurrency we actually run at. Keepalive
             # REUSE is allowed (capped) so sockets are recycled rather than opening
@@ -156,7 +163,9 @@ async def generate_structured(
                 continue
 
             if response.status_code == 200:
-                return self._parse(response, trace_id)
+                result = self._parse(response, trace_id)
+                self.completed_calls += 1
+                return result
 
             raise ModelError(
                 f"request {trace_id} failed with HTTP {response.status_code}"
@@ -169,15 +178,26 @@ async def generate_structured(
 
     @staticmethod
     def _parse(response: httpx.Response, trace_id: str) -> ModelResult:
-        body = json.loads(response.text)
         try:
+            body = json.loads(response.text)
             message = body["choices"][0]["message"]
             content = message["content"]
-        except (KeyError, IndexError, TypeError) as exc:
+        except (ValueError, KeyError, IndexError, TypeError) as exc:
+            # ValueError covers JSONDecodeError: a proxy/CDN can answer 200 with
+            # an HTML page. Surface it as the typed ModelError callers handle.
             raise ModelError(
                 f"request {trace_id} returned a structurally invalid completion body",
                 trace_id=trace_id,
             ) from exc
+        # Reasoning models may return ``"content": null`` (e.g. max_tokens spent
+        # on thinking). Normalise to "" so callers can rely on a str.
+        if content is None:
+            content = ""
+        elif not isinstance(content, str):
+            raise ModelError(
+                f"request {trace_id} returned non-string completion content",
+                trace_id=trace_id,
+            )
         return ModelResult(
             content=content,
             request_id=str(body.get("id", "")),

src/verdi/analysis/comparative_report.py

@@ -158,18 +158,19 @@ def _institution_rows(metrics: dict[str, Any]) -> list[str]:
         deltas = enactment.get("rule_deltas", {})
         delta_bits = []
         for key, added in (deltas.get("added") or {}).items():
-            delta_bits.append(f"+{key}={_esc(added)}")
+            delta_bits.append(f"+{_esc(key)}={_esc(added)}")
         for key, change in (deltas.get("changed") or {}).items():
-            delta_bits.append(f"~{key}:{_esc(change['from'])}→{_esc(change['to'])}")
+            delta_bits.append(f"~{_esc(key)}:{_esc(change['from'])}→{_esc(change['to'])}")
         for key in (deltas.get("removed") or {}):
-            delta_bits.append(f"-{key}")
+            delta_bits.append(f"-{_esc(key)}")
+        # One f-string, escaped once: a ``.format`` pass over already-inserted
+        # rule values would treat their braces (e.g. dict values) as fields.
         rows.append(
-            "<tr><td>Enacted amendment <code>{v}</code> @ tick {t}</td>"
+            f"<tr><td>Enacted amendment <code>{_esc(enactment['version'])}</code>"
+            f" @ tick {_esc(enactment['tick'])}</td>"
             f"<td>{len(delta_bits)} rule change(s)</td>"
-            f"<td>{_esc(' '.join(delta_bits))} "
-            f"<span class='ev'>constitution_history.jsonl</span></td></tr>".format(
-                v=_esc(enactment["version"]), t=enactment["tick"]
-            )
+            f"<td>{' '.join(delta_bits)} "
+            f"<span class='ev'>constitution_history.jsonl</span></td></tr>"
         )
     enacted_ids = mech.get("amendment_enacted_event_ids", [])
     rejected_ids = mech.get("amendment_rejected_event_ids", [])
@@ -207,6 +208,13 @@ def _recovery_row(summary: dict[str, Any], arm: str, seed: Any) -> str:
                 f"unavailable ({_esc(exc)})</p>")
     if rec["recovered"]:
         label = f"recovered at tick {rec['recovery_tick']}"
+    elif not rec["pre_shock_rate"]:
+        # No objective due in the preregistered baseline window (None), or a
+        # 0.0 baseline: the criterion is undefined -- not evidence of
+        # non-recovery (recovery_time still reports it censored).
+        label = ("not evaluable — no delivery objective is due in the "
+                 "preregistered baseline window" if rec["pre_shock_rate"] is None
+                 else "not evaluable — pre-shock baseline rate is 0")
     elif rec["censored"]:
         label = "CENSORED — never recovered by run end"
     else:

src/verdi/analysis/cross_run_compare.py

@@ -126,7 +126,10 @@ def compare_runs(run_dirs: list[Path]) -> dict[str, Any]:
                 round(sum(1 for c in consistent if not c) / len(consistent), 3)
                 if consistent else None
             ),
-            "ticks_audited": sorted({int(r.get("tick", -1)) for r in prov}),
+            "ticks_audited": sorted({
+                int(r["tick"]) for r in prov
+                if isinstance(r.get("tick"), (int, float)) and not isinstance(r.get("tick"), bool)
+            }),
         }
         runs.append({
             "run_dir": str(rd),

src/verdi/analysis/outcome_metrics.py

@@ -38,6 +38,7 @@
 import json
 import math
 import os
+import re
 import sys
 from pathlib import Path
 from typing import Any
@@ -141,6 +142,27 @@ def gini(values: list[float]) -> float | None:
 # 1. completion rate (objective task success)
 # --------------------------------------------------------------------------- #
 
+_TASK_TICK = re.compile(r"-t(\d+)-")
+
+
+def _declared_due_by_objective(run_dir: Path) -> dict[str, int]:
+    """objective_id -> declared due tick from ``scenario_resolved.yaml``
+    ({} when absent or malformed)."""
+    try:
+        import yaml
+
+        from verdi.world.economy import parse_task_objectives
+
+        resolved = yaml.safe_load(
+            (run_dir / "scenario_resolved.yaml").read_text(encoding="utf-8"))
+        if not isinstance(resolved, dict):
+            return {}
+        return {o["objective_id"]: int(o["due_tick"])
+                for o in parse_task_objectives(resolved.get("tasks"))}
+    except (OSError, ValueError, yaml.YAMLError):
+        return {}
+
+
 def completion_rate(run_dir: Path) -> dict[str, Any]:
     """Verified delivery-objective outcomes, straight off the event ledger.
 
@@ -158,25 +180,49 @@ def completion_rate(run_dir: Path) -> dict[str, Any]:
     Returns::
 
         {"objectives": [{objective_id, status, required, baseline, actual,
-                          delta, due_tick, evidence_event_ids}],
+                          delta, due_tick, ledger_tick, evidence_event_ids}],
          "overall": float | None,
          "completed": int, "failed": int, "evaluated": int}
     """
     run_dir = _require_run_dir(run_dir)
     events = _read_jsonl(run_dir, "events")
+    declared_due = _declared_due_by_objective(run_dir)
     objectives: list[dict[str, Any]] = []
+    # Lifecycle tick of each outcome: under an economy there is exactly one
+    # end-of-tick ``consumption`` event per lifecycle tick, written just
+    # before objectives are evaluated, so (consumption events so far) - 1 is
+    # exact even when a whole tick's actions were quarantined. Otherwise the
+    # tick is carried forward from task ids (soc-tNN-..).
+    has_consumption = any(e.get("kind") == "consumption" for e in events)
+    lifecycle = 0
+    consumed = 0
     for ev in events:
+        if has_consumption:
+            if ev.get("kind") == "consumption":
+                consumed += 1
+                lifecycle = consumed - 1
+        else:
+            m = _TASK_TICK.search(str((ev.get("payload") or {}).get("task_id") or ""))
+            if m:
+                lifecycle = int(m.group(1))
         if ev.get("kind") not in ("task_completed", "task_failed"):
             continue
         payload = ev.get("payload") or {}
+        oid = str(payload.get("objective_id", ""))
+        ledger_tick = int(ev.get("tick", -1))
         objectives.append({
-            "objective_id": str(payload.get("objective_id", "")),
+            "objective_id": oid,
             "status": "completed" if ev["kind"] == "task_completed" else "failed",
             "required": payload.get("required"),
             "baseline": payload.get("baseline"),
             "actual": payload.get("actual"),
             "delta": payload.get("delta"),
-            "due_tick": int(ev.get("tick", -1)),
+            # Lifecycle due tick: declared in the frozen scenario when present,
+            # else the lifecycle tick of the actions just before the outcome
+            # (objectives are evaluated after that tick's actions). The
+            # event's own ``tick`` is the ledger transition counter.
+            "due_tick": declared_due.get(oid, lifecycle),
+            "ledger_tick": ledger_tick,
             "evidence_event_ids": [int(ev["event_id"])],
         })
     objectives.sort(key=lambda o: (o["due_tick"], o["objective_id"]))
@@ -325,7 +371,16 @@ def exclusion_episodes(run_dir: Path) -> dict[str, Any]:
     run_dir = _require_run_dir(run_dir)
     sanctions = _read_jsonl(run_dir, "sanctions")
     events = _read_jsonl(run_dir, "events")
-    last_ledger_tick = max((int(e.get("tick", -1)) for e in events), default=-1)
+    # Sanction ticks are LIFECYCLE ticks; events.jsonl ``tick`` is the
+    # reducer's transition counter (often 5x larger). The run's last lifecycle
+    # tick comes from resources.jsonl (one row per lifecycle tick); the
+    # ledger tick is only a fallback for runs without it.
+    resource_ticks = [r.get("tick") for r in _read_jsonl(run_dir, "resources")]
+    resource_ticks = [int(t) for t in resource_ticks if isinstance(t, (int, float))]
+    # Open episodes count up to (not including) the last observed tick --
+    # the convention pinned by test_exclusion_repeat_episode_flag_synthetic.
+    run_end_tick = (max(resource_ticks) if resource_ticks
+                    else max((int(e.get("tick", -1)) for e in events), default=-1))
 
     agents: dict[str, dict[str, Any]] = {}
     totals = {"sanctions_imposed": 0, "appeals_filed": 0,
@@ -360,9 +415,6 @@ def _agent(agent_id: str) -> dict[str, Any]:
                 continue
             subject, episode = hit
             episode["restored_tick"] = int(rec.get("tick"))
-            entry = _agent(subject)
-            lo, hi = episode["effective_tick"], episode["restored_tick"]
-            entry["excluded_ticks"] += max(0, hi - lo)
         elif rtype == "appeal":
             # sanctions.jsonl appeal records: kind appeal_filed (phase "filed")
             # and adjudication events with kind appeal_upheld / appeal_dismissed
@@ -373,19 +425,24 @@ def _agent(agent_id: str) -> dict[str, Any]:
             elif kind == "appeal_upheld":
                 totals["appeals_upheld"] += 1
 
-    # Open episodes: excluded until the last ledger tick (inclusive count).
-    for subject, episode in open_episodes.values():
-        entry = _agent(subject)
-        hi = max(episode["effective_tick"], last_ledger_tick)
-        entry["excluded_ticks"] += max(0, hi - episode["effective_tick"])
     del open_episodes
 
     out_agents: dict[str, Any] = {}
     for aid in sorted(agents):
         entry = agents[aid]
         entry["episodes"].sort(key=lambda e: (e["effective_tick"], e["sanction_id"]))
-        closed = [e for e in entry["episodes"] if e["restored_tick"] is not None]
-        entry["repeat_episode"] = len(closed) > 1 or len(entry["episodes"]) > 1
+        # Overlapping sanctions are ONE exclusion: merge intervals before
+        # counting (open episodes run to the last lifecycle tick).
+        merged: list[list[int]] = []
+        for e in entry["episodes"]:
+            lo = e["effective_tick"]
+            hi = e["restored_tick"] if e["restored_tick"] is not None else max(lo, run_end_tick)
+            if merged and lo <= merged[-1][1]:
+                merged[-1][1] = max(merged[-1][1], hi)
+            else:
+                merged.append([lo, hi])
+        entry["excluded_ticks"] = sum(max(0, hi - lo) for lo, hi in merged)
+        entry["repeat_episode"] = len(merged) > 1
         out_agents[aid] = entry
     return {"agents": out_agents, "totals": totals}
 
@@ -457,19 +514,34 @@ def recovery_time(
     scenario = manifest.get("scenario") if isinstance(manifest.get("scenario"), dict) else {}
     for delivery in (scenario.get("tasks") or {}).get("deliveries", []) or []:
         if isinstance(delivery, dict) and delivery.get("due_tick") is not None:
-            due_ticks.append(int(delivery["due_tick"]))
+            try:
+                due_ticks.append(int(delivery["due_tick"]))
+            except (TypeError, ValueError):
+                due_ticks = []  # malformed: degrade to raw ledger ticks
+                break
     # The campaign writes the frozen scenario it actually ran next to the
     # artifacts; that is the authoritative source of declared due ticks.
     try:
         import yaml  # type: ignore[import-not-found]
 
         resolved = yaml.safe_load(
             (run_dir / "scenario_resolved.yaml").read_text(encoding="utf-8"))
-        for delivery in (resolved.get("tasks") or {}).get("deliveries", []) or []:
-            if isinstance(delivery, dict) and delivery.get("due_tick") is not None:
-                due_ticks.append(int(delivery["due_tick"]))
-    except (OSError, ValueError):
-        due_ticks = []
+    except (OSError, ValueError, yaml.YAMLError):
+        resolved = None
+    if isinstance(resolved, dict):
+        # Authoritative: replaces (never appends to) manifest-derived ticks.
+        try:
+            due_ticks = [
+                int(delivery["due_tick"])
+                for delivery in (resolved.get("tasks") or {}).get("deliveries", []) or []
+                if isinstance(delivery, dict) and delivery.get("due_tick") is not None
+            ]
+        except (AttributeError, TypeError, ValueError):
+            due_ticks = []  # malformed declaration: degrade to raw ledger ticks
+    # The campaign evaluates objectives in due-tick order (declaration order
+    # within a tick), so the declared list must be put in that order before
+    # zipping; a YAML listing like [12, 30, 46, 30] would otherwise mislabel.
+    due_ticks.sort()
 
     outcomes = [
         ev for ev in events if ev.get("kind") in ("task_completed", "task_failed")
@@ -509,7 +581,10 @@ def _rate(tick: int) -> float | None:
         if r is not None:
             rates[tick] = r
 
-    threshold = (target_fraction * pre_shock_rate) if pre_shock_rate is not None else None
+    # A zero baseline gives a zero threshold that any rate (even 0.0) meets,
+    # i.e. "recovery" with nothing recovered; treat it like an undefined
+    # baseline (criterion not evaluable -> censored).
+    threshold = (target_fraction * pre_shock_rate) if pre_shock_rate else None
     recovered = False
     recovery_tick: int | None = None
     if threshold is not None: