[codex] Add loop decision memory export

exports/manifest.json

@@ -1 +1 @@
-{"claim_count":4601,"claim_extracted_at":"2026-07-01T02:48:05Z","concept_count":875,"context_pack_count":0,"edge_count":1899,"enrichment_run_count":2,"eval_case_count":53,"eval_result_count":53,"insight_decision_count":3,"insight_label_count":3,"insight_policy_version":"agent_core_insight_v1","learning_event_count":4,"learning_ledger_version":"learning_ledger_v1","learning_review_count":0,"observation_count":2,"policy_decision_count":0,"policy_version":"knowledge_policy_v1","quality_finding_count":42,"retrieval_version":"retrieval_v1","run_id":"local-20260701024805","source_ref_count":1020,"source_shas":{}}
+{"claim_count":4601,"claim_extracted_at":"2026-07-02T21:26:39Z","concept_count":875,"context_pack_count":0,"edge_count":1899,"enrichment_run_count":2,"eval_case_count":53,"eval_result_count":53,"insight_decision_count":3,"insight_label_count":3,"insight_policy_version":"agent_core_insight_v1","learning_event_count":4,"learning_ledger_version":"learning_ledger_v1","learning_review_count":0,"loop_decision_envelope_count":0,"loop_decision_memory_version":"agent_core_loop_decision_v1","observation_count":2,"policy_decision_count":0,"policy_version":"knowledge_policy_v1","quality_finding_count":42,"retrieval_version":"retrieval_v1","run_id":"local-20260702212639","source_ref_count":1020,"source_shas":{}}

schema/sqlite-schema.sql

@@ -241,6 +241,27 @@ CREATE TABLE IF NOT EXISTS insight_labels (
 CREATE INDEX IF NOT EXISTS idx_insight_labels_insight ON insight_labels(insight_id);
 CREATE INDEX IF NOT EXISTS idx_insight_labels_loop ON insight_labels(loop);
 
+CREATE TABLE IF NOT EXISTS loop_decision_envelopes (
+  envelope_id TEXT PRIMARY KEY,
+  loop TEXT NOT NULL,
+  created_at TEXT NOT NULL,
+  fingerprint TEXT NOT NULL,
+  decision TEXT NOT NULL,
+  insight_id TEXT,
+  case_id TEXT,
+  meta_case_id TEXT,
+  evidence_refs_json TEXT NOT NULL DEFAULT '[]',
+  proposed_action_json TEXT NOT NULL DEFAULT '{}',
+  human_outcome_json TEXT NOT NULL DEFAULT '{}',
+  governance_json TEXT NOT NULL DEFAULT '{}',
+  body_json TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_loop_decision_envelopes_fingerprint ON loop_decision_envelopes(fingerprint);
+CREATE INDEX IF NOT EXISTS idx_loop_decision_envelopes_loop ON loop_decision_envelopes(loop);
+CREATE INDEX IF NOT EXISTS idx_loop_decision_envelopes_decision ON loop_decision_envelopes(decision);
+CREATE INDEX IF NOT EXISTS idx_loop_decision_envelopes_case ON loop_decision_envelopes(case_id);
+
 CREATE TABLE IF NOT EXISTS enrichment_runs (
   run_id TEXT PRIMARY KEY,
   concept_id TEXT,

src/hyrule_knowledge/cli.py

@@ -35,6 +35,7 @@
     ledger_check,
     write_learning_ledger_reports,
 )
+from .loop_memory import query_loop_decision_envelopes, shadow_arbitrate_loop_decisions
 from .models import Concept, RepoSnapshot, SourceRef
 from .observe import collect_safe_health
 from .okf_writer import reset_generated, write_concepts, write_indexes
@@ -496,6 +497,29 @@ def cmd_claims(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_loop_decisions(args: argparse.Namespace) -> int:
+    config = load_config(Path(args.config))
+    rows = query_loop_decision_envelopes(
+        config.exports_dir / "knowledge.sqlite",
+        fingerprint=args.fingerprint,
+        loop=args.loop,
+        limit=args.limit,
+    )
+    print_json({"loop_decision_envelopes": rows})
+    return 0
+
+
+def cmd_loop_arbiter_shadow(args: argparse.Namespace) -> int:
+    config = load_config(Path(args.config))
+    result = shadow_arbitrate_loop_decisions(
+        config.exports_dir / "knowledge.sqlite",
+        fingerprint=args.fingerprint,
+        limit=args.limit,
+    )
+    print_json(result)
+    return 0
+
+
 def cmd_neighborhood(args: argparse.Namespace) -> int:
     try:
         edge_types = set(args.edge_type or []) or None
@@ -918,6 +942,17 @@ def build_parser() -> argparse.ArgumentParser:
     claims.add_argument("--limit", type=int, default=50)
     claims.set_defaults(func=cmd_claims)
 
+    loop_decisions = subparsers.add_parser("loop-decisions")
+    loop_decisions.add_argument("--fingerprint")
+    loop_decisions.add_argument("--loop", choices=["engineering", "knowledge", "noc", "soc"])
+    loop_decisions.add_argument("--limit", type=int, default=50)
+    loop_decisions.set_defaults(func=cmd_loop_decisions)
+
+    arbiter_shadow = subparsers.add_parser("loop-arbiter-shadow")
+    arbiter_shadow.add_argument("--fingerprint", required=True)
+    arbiter_shadow.add_argument("--limit", type=int, default=50)
+    arbiter_shadow.set_defaults(func=cmd_loop_arbiter_shadow)
+
     neighborhood = subparsers.add_parser("neighborhood")
     neighborhood.add_argument("concept_id")
     neighborhood.add_argument("--depth", type=int, default=1)

src/hyrule_knowledge/exporter.py

@@ -12,6 +12,7 @@
 from .claims import compile_claims
 from .insight_policy import load_insight_policy_fixtures
 from .learning_ledger import load_learning_events
+from .loop_memory import load_loop_decision_envelopes
 from .models import Edge
 from .okf_writer import edge_json
 from .validator import parse_frontmatter
@@ -169,6 +170,7 @@ def write_exports(
     learning_event_rows = load_learning_events()
     learning_review_rows = _learning_review_rows()
     insight_decision_rows, insight_label_rows = load_insight_policy_fixtures()
+    loop_decision_rows = load_loop_decision_envelopes()
 
     write_jsonl(exports_dir / "concepts.jsonl", concepts_public)
     write_jsonl(exports_dir / "sources.jsonl", source_rows)
@@ -185,6 +187,7 @@ def write_exports(
     write_jsonl(exports_dir / "learning-reviews.jsonl", learning_review_rows)
     write_jsonl(exports_dir / "insight-decisions.jsonl", insight_decision_rows)
     write_jsonl(exports_dir / "insight-labels.jsonl", insight_label_rows)
+    write_jsonl(exports_dir / "loop-decision-envelopes.jsonl", loop_decision_rows)
     manifest = {
         "concept_count": len(concepts),
         "edge_count": len(edge_rows),
@@ -201,8 +204,10 @@ def write_exports(
         "learning_review_count": len(learning_review_rows),
         "insight_decision_count": len(insight_decision_rows),
         "insight_label_count": len(insight_label_rows),
+        "loop_decision_envelope_count": len(loop_decision_rows),
         "learning_ledger_version": "learning_ledger_v1",
         "insight_policy_version": "agent_core_insight_v1",
+        "loop_decision_memory_version": "agent_core_loop_decision_v1",
         "retrieval_version": "retrieval_v1",
         "policy_version": "knowledge_policy_v1",
         "source_shas": source_shas or {},
@@ -228,6 +233,7 @@ def write_exports(
         learning_review_rows,
         insight_decision_rows,
         insight_label_rows,
+        loop_decision_rows,
         manifest,
     )
 
@@ -281,6 +287,7 @@ def write_sqlite(
     learning_review_rows: list[dict[str, Any]],
     insight_decision_rows: list[dict[str, Any]],
     insight_label_rows: list[dict[str, Any]],
+    loop_decision_rows: list[dict[str, Any]],
     manifest: dict[str, Any],
 ) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -575,6 +582,32 @@ def write_sqlite(
                     json.dumps(label, sort_keys=True),
                 ),
             )
+        for decision in loop_decision_rows:
+            conn.execute(
+                """
+                INSERT OR REPLACE INTO loop_decision_envelopes (
+                  envelope_id, loop, created_at, fingerprint, decision,
+                  insight_id, case_id, meta_case_id, evidence_refs_json,
+                  proposed_action_json, human_outcome_json, governance_json,
+                  body_json
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    decision.get("envelope_id"),
+                    decision.get("loop"),
+                    decision.get("created_at"),
+                    decision.get("fingerprint"),
+                    decision.get("decision"),
+                    decision.get("insight_id"),
+                    decision.get("case_id"),
+                    decision.get("meta_case_id"),
+                    json.dumps(decision.get("evidence_refs", []), sort_keys=True),
+                    json.dumps(decision.get("proposed_action", {}), sort_keys=True),
+                    json.dumps(decision.get("human_outcome", {}), sort_keys=True),
+                    json.dumps(decision.get("governance", {}), sort_keys=True),
+                    json.dumps(decision, sort_keys=True),
+                ),
+            )
         conn.execute(
             "INSERT INTO runs (run_id, started_at, completed_at, source_shas_json, concept_count, edge_count) VALUES (?, datetime('now'), datetime('now'), ?, ?, ?)",
             (

src/hyrule_knowledge/loop_memory.py

@@ -0,0 +1,137 @@
+"""Shared memory helpers for loop decision envelopes."""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any
+
+from agent_core.arbiter import arbitrate_cross_loop_event
+from agent_core.contracts import LoopDecisionEnvelope
+
+from .learning_ledger import _forbidden_payload_findings
+
+DEFAULT_LOOP_DECISION_PATHS = (Path("reports/loop-decision-envelopes.jsonl"),)
+
+
+class LoopDecisionMemoryError(ValueError):
+    """Raised when imported loop decision records are malformed or unsafe."""
+
+
+def load_loop_decision_envelopes(paths: list[Path] | None = None) -> list[dict[str, Any]]:
+    rows: list[dict[str, Any]] = []
+    for path in paths or list(DEFAULT_LOOP_DECISION_PATHS):
+        for row in _read_rows(path):
+            if not isinstance(row, dict):
+                raise LoopDecisionMemoryError(f"loop decision row must be an object: {path}")
+            if not row.get("created_at"):
+                raise LoopDecisionMemoryError(f"loop decision row must include created_at for deterministic export: {path}")
+            findings = _forbidden_payload_findings(row)
+            if findings:
+                raise LoopDecisionMemoryError(f"unsafe loop decision row in {path}: {'; '.join(findings)}")
+            rows.append(LoopDecisionEnvelope.model_validate(row).model_dump(mode="json"))
+    return sorted(rows, key=lambda row: (str(row.get("fingerprint") or ""), str(row.get("envelope_id") or "")))
+
+
+def query_loop_decision_envelopes(
+    db_path: Path,
+    *,
+    fingerprint: str | None = None,
+    loop: str | None = None,
+    limit: int = 50,
+) -> list[dict[str, Any]]:
+    clauses: list[str] = []
+    params: list[Any] = []
+    if fingerprint:
+        clauses.append("fingerprint = ?")
+        params.append(fingerprint)
+    if loop:
+        clauses.append("loop = ?")
+        params.append(loop)
+    where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
+    params.append(max(1, min(limit, 500)))
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    try:
+        rows = conn.execute(
+            f"""
+            SELECT envelope_id, loop, created_at, fingerprint, decision, insight_id,
+                   case_id, meta_case_id, evidence_refs_json, proposed_action_json,
+                   human_outcome_json, governance_json, body_json
+            FROM loop_decision_envelopes
+            {where}
+            ORDER BY created_at DESC, envelope_id ASC
+            LIMIT ?
+            """,
+            params,
+        ).fetchall()
+    finally:
+        conn.close()
+    return [_row_json(row) for row in rows]
+
+
+def shadow_arbitrate_loop_decisions(
+    db_path: Path,
+    *,
+    fingerprint: str,
+    limit: int = 50,
+) -> dict[str, Any]:
+    rows = query_loop_decision_envelopes(db_path, fingerprint=fingerprint, limit=limit)
+    candidates = [_arbiter_candidate(row) for row in rows]
+    evidence_refs = []
+    for row in rows:
+        evidence_refs.extend(row.get("evidence_refs", []))
+    decision = arbitrate_cross_loop_event(
+        event_fingerprint=fingerprint,
+        candidates=candidates,
+        evidence_refs=evidence_refs,
+    )
+    return {
+        "shadow_mode": True,
+        "suppression_applied": False,
+        "candidate_count": len(candidates),
+        "arbiter_decision": decision.model_dump(mode="json"),
+    }
+
+
+def _read_rows(path: Path) -> list[Any]:
+    if not path.exists():
+        return []
+    text = path.read_text(encoding="utf-8").strip()
+    if not text:
+        return []
+    if path.suffix == ".json":
+        loaded = json.loads(text)
+        if isinstance(loaded, list):
+            return loaded
+        if isinstance(loaded, dict) and isinstance(loaded.get("loop_decision_envelopes"), list):
+            return list(loaded["loop_decision_envelopes"])
+        if isinstance(loaded, dict):
+            return [loaded]
+        return []
+    return [json.loads(line) for line in text.splitlines() if line.strip()]
+
+
+def _row_json(row: sqlite3.Row) -> dict[str, Any]:
+    result = dict(row)
+    for key in ("evidence_refs_json", "proposed_action_json", "human_outcome_json", "governance_json", "body_json"):
+        output_key = key.removesuffix("_json")
+        result[output_key] = json.loads(str(result.pop(key) or "{}"))
+    return result
+
+
+def _arbiter_candidate(row: dict[str, Any]) -> dict[str, Any]:
+    body = row.get("body")
+    candidate = body if isinstance(body, dict) else {}
+    proposed = row.get("proposed_action")
+    return {
+        **candidate,
+        "loop": row.get("loop") or candidate.get("loop"),
+        "action_selected": row.get("decision") or candidate.get("decision"),
+        "case_id": row.get("case_id") or candidate.get("case_id"),
+        "meta_case_id": row.get("meta_case_id") or candidate.get("meta_case_id"),
+        "candidate_type": candidate.get("candidate_type") or (proposed.get("type") if isinstance(proposed, dict) else None),
+        "candidate_source": candidate.get("candidate_source") or candidate.get("policy_version"),
+        "proposed_action": proposed,
+    }

tests/test_exporter.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import sqlite3
 from pathlib import Path
 
 from hyrule_knowledge.exporter import exports_match, read_jsonl, write_exports
@@ -33,15 +34,59 @@ def _write_observation(root: Path) -> None:
     )
 
 
-def test_write_exports_uses_run_timestamp_for_observation_freshness(tmp_path: Path) -> None:
+def _write_schema(root: Path) -> None:
+    source = Path(__file__).resolve().parents[1] / "schema/sqlite-schema.sql"
+    schema = root / "schema/sqlite-schema.sql"
+    schema.parent.mkdir(parents=True)
+    schema.write_text(source.read_text(encoding="utf-8"), encoding="utf-8")
+
+
+def _write_loop_decision_report(root: Path) -> None:
+    report = root / "reports/loop-decision-envelopes.jsonl"
+    report.parent.mkdir(parents=True)
+    report.write_text(
+        json.dumps(
+            {
+                "schema_version": "0.1.0",
+                "envelope_id": "ldec_export_1",
+                "loop": "engineering",
+                "created_at": "2026-07-02T21:31:00Z",
+                "input_event": {"change_id": "chg_1"},
+                "decision": "draft",
+                "evidence_refs": [{"kind": "github_pr", "ref": "https://github.com/AS215932/knowledge/pull/1"}],
+                "proposed_action": {"type": "draft_pr"},
+                "governance": {"sensitivity_class": "internal", "approval_tier": "operator"},
+                "case_id": "case_1",
+                "fingerprint": "shared-fp",
+            },
+            sort_keys=True,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+
+def test_write_exports_uses_run_timestamp_for_observation_freshness(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.chdir(tmp_path)
+    _write_schema(tmp_path)
     root = tmp_path / "okf"
     exports = tmp_path / "exports"
     _write_observation(root)
+    _write_loop_decision_report(tmp_path)
 
     write_exports(root, exports, edges=[], run_id="local-20260103000000")
 
     manifest = json.loads((exports / "manifest.json").read_text(encoding="utf-8"))
     claims = read_jsonl(exports / "claims.jsonl")
+    loop_decisions = read_jsonl(exports / "loop-decision-envelopes.jsonl")
     assert manifest["claim_extracted_at"] == "2026-01-03T00:00:00Z"
+    assert manifest["loop_decision_envelope_count"] == 1
+    assert loop_decisions[0]["envelope_id"] == "ldec_export_1"
     assert {claim["freshness_status"] for claim in claims} == {"expired"}
+    conn = sqlite3.connect(exports / "knowledge.sqlite")
+    try:
+        stored = conn.execute("SELECT loop, fingerprint FROM loop_decision_envelopes").fetchone()
+    finally:
+        conn.close()
+    assert stored == ("engineering", "shared-fp")
     assert exports_match(root, exports)