Phase 3: optional agent-core TraceEvent/CostUsage emission (additive, flag-gated)

src/hyrule_engineering_loop/agent_core_trace.py

@@ -0,0 +1,54 @@
+"""Optional, flag-gated emission of agent-core TraceEvent records (Phase 3).
+
+Best-effort and additive: a no-op unless ``HYRULE_ENGINEERING_AGENT_CORE_TRACE`` is
+truthy AND the optional ``agent-core`` package is importable. ``agent-core`` is NOT a
+declared dependency of this repo; ``agent_core`` is imported dynamically via ``importlib``
+so ``mypy --strict src`` never depends on it and CI without it simply emits nothing.
+Any failure is swallowed so emission can never affect the loop.
+
+Records are appended as JSON lines to ``HYRULE_ENGINEERING_AGENT_CORE_TRACE_PATH``
+(default ``reports/agent-core-trace.jsonl``). Higher fidelity than the knowledge loop:
+the engineering-loop state carries token/USD cost in ``backend_results[].cost``.
+"""
+
+from __future__ import annotations
+
+import importlib
+import json
+import os
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+FLAG_ENV = "HYRULE_ENGINEERING_AGENT_CORE_TRACE"
+PATH_ENV = "HYRULE_ENGINEERING_AGENT_CORE_TRACE_PATH"
+_DEFAULT_PATH = "reports/agent-core-trace.jsonl"
+
+
+def enabled() -> bool:
+    return os.environ.get(FLAG_ENV, "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _sink_path() -> Path:
+    return Path(os.environ.get(PATH_ENV) or _DEFAULT_PATH)
+
+
+def emit_loop_trace(state: Mapping[str, Any]) -> int:
+    """Emit one agent-core TraceEvent per loop-trace item; return count (0 if disabled)."""
+    if not enabled():
+        return 0
+    try:
+        adapter = importlib.import_module("agent_core.adapters.engineering_loop")
+        run_id = state.get("change_id")
+        events = adapter.trace_events_from_loop_trace(dict(state), run_id=run_id)
+        path = _sink_path()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        count = 0
+        with path.open("a", encoding="utf-8") as handle:
+            for event in events:
+                record: dict[str, Any] = event.model_dump(mode="json")
+                handle.write(json.dumps(record, sort_keys=True) + "\n")
+                count += 1
+        return count
+    except Exception:  # best-effort: emission must never break the loop
+        return 0

src/hyrule_engineering_loop/trace.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 from typing import Any, cast
 
+from hyrule_engineering_loop.agent_core_trace import emit_loop_trace
 from hyrule_engineering_loop.state import GraphState
 
 TRACE_FILENAME = "loop_trace.json"
@@ -344,6 +345,7 @@ def format_loop_trace_summary(trace: dict[str, Any]) -> str:
 
 def write_loop_trace(state: GraphState) -> str | None:
     """Write ``loop_trace.json`` beside the NOC handoff when configured."""
+    emit_loop_trace(state)
     output_dir = _resolve_trace_dir(state)
     if output_dir is None:
         return None

tests/test_agent_core_trace.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+pytest.importorskip("agent_core")
+
+from hyrule_engineering_loop import agent_core_trace
+
+
+def _state() -> dict[str, object]:
+    return {
+        "change_id": "chg-test-1",
+        "llm_outputs": [
+            {
+                "role": "security_auditor",
+                "approved": False,
+                "model_selection": {"provider": "anthropic", "model": "claude-sonnet-4-6"},
+            }
+        ],
+        "gate_results": [{"command": ["ruff"], "status": "failed", "returncode": 2}],
+        "backend_results": [
+            {
+                "backend": "mock",
+                "status": "completed",
+                "cost": {"input_tokens": 100, "output_tokens": 20, "usd": 0.0},
+            }
+        ],
+    }
+
+
+def test_disabled_by_default(monkeypatch, tmp_path):
+    monkeypatch.delenv("HYRULE_ENGINEERING_AGENT_CORE_TRACE", raising=False)
+    monkeypatch.setenv("HYRULE_ENGINEERING_AGENT_CORE_TRACE_PATH", str(tmp_path / "t.jsonl"))
+    assert agent_core_trace.emit_loop_trace(_state()) == 0
+    assert not (tmp_path / "t.jsonl").exists()
+
+
+def test_emits_when_enabled(monkeypatch, tmp_path):
+    sink = tmp_path / "t.jsonl"
+    monkeypatch.setenv("HYRULE_ENGINEERING_AGENT_CORE_TRACE", "1")
+    monkeypatch.setenv("HYRULE_ENGINEERING_AGENT_CORE_TRACE_PATH", str(sink))
+    count = agent_core_trace.emit_loop_trace(_state())
+    assert count == 3
+    lines = sink.read_text(encoding="utf-8").strip().splitlines()
+    assert len(lines) == 3
+    kinds = {json.loads(line)["event_type"] for line in lines}
+    assert kinds == {"model_call", "tool_call", "backend_execution"}
+    backend = next(
+        json.loads(line) for line in lines if json.loads(line)["event_type"] == "backend_execution"
+    )
+    assert backend["cost"]["input_tokens"] == 100
+    assert backend["run_id"] == "chg-test-1"