diff --git a/src/hyrule_engineering_loop/agent_core_trace.py b/src/hyrule_engineering_loop/agent_core_trace.py new file mode 100644 index 0000000..39fb9f7 --- /dev/null +++ b/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 diff --git a/src/hyrule_engineering_loop/trace.py b/src/hyrule_engineering_loop/trace.py index 2f8932b..7f17d80 100644 --- a/src/hyrule_engineering_loop/trace.py +++ b/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 diff --git a/tests/test_agent_core_trace.py b/tests/test_agent_core_trace.py new file mode 100644 index 0000000..9aa9229 --- /dev/null +++ b/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"