trace: ship engineering events through sink_from_env

pyproject.toml

@@ -36,4 +36,4 @@ python_version = "3.12"
 strict = true
 
 [tool.uv.sources]
-agent-core = { git = "https://github.com/AS215932/agent-core", tag = "v0.1.0" }
+agent-core = { git = "https://github.com/AS215932/agent-core", tag = "v0.4.0" }

src/hyrule_engineering_loop/agent_core_trace.py

@@ -1,36 +1,49 @@
-"""Optional, flag-gated emission of agent-core TraceEvent records (Phase 3).
+"""Optional, flag-gated emission of agent-core TraceEvent records.
 
 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``.
+truthy and ``agent-core`` is importable. Delivery uses ``agent_core.tracing.sink_from_env``
+so operators can configure a JSONL path, an HTTP collector URL, or both.
+
+The historical JSONL fallback is preserved: when tracing is enabled without an explicit
+``*_PATH`` or ``*_COLLECTOR_URL``, events are appended to
+``reports/agent-core-trace.jsonl``. Any failure is swallowed so emission can never affect
+the loop, and the returned count reflects events delivered to at least one sink.
 """
 
 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"
+COLLECTOR_URL_ENV = f"{FLAG_ENV}_COLLECTOR_URL"
 _DEFAULT_PATH = "reports/agent-core-trace.jsonl"
+_TRUTHY = {"1", "true", "yes", "on"}
 
 
 def enabled() -> bool:
-    return os.environ.get(FLAG_ENV, "").strip().lower() in {"1", "true", "yes", "on"}
+    return os.environ.get(FLAG_ENV, "").strip().lower() in _TRUTHY
+
 
+def _sink_from_env() -> Any:
+    sink_mod = importlib.import_module("agent_core.tracing.sink")
+    path_configured = bool(os.environ.get(PATH_ENV, "").strip())
+    collector_configured = bool(os.environ.get(COLLECTOR_URL_ENV, "").strip())
+    if path_configured or collector_configured:
+        return sink_mod.sink_from_env(FLAG_ENV)
 
-def _sink_path() -> Path:
-    return Path(os.environ.get(PATH_ENV) or _DEFAULT_PATH)
+    original_path = os.environ.get(PATH_ENV)
+    os.environ[PATH_ENV] = _DEFAULT_PATH
+    try:
+        return sink_mod.sink_from_env(FLAG_ENV)
+    finally:
+        if original_path is None:
+            os.environ.pop(PATH_ENV, None)
+        else:
+            os.environ[PATH_ENV] = original_path
 
 
 def emit_loop_trace(state: Mapping[str, Any]) -> int:
@@ -39,15 +52,12 @@ def emit_loop_trace(state: Mapping[str, Any]) -> int:
         return 0
     try:
         adapter = importlib.import_module("agent_core.adapters.engineering_loop")
+        sink = _sink_from_env()
         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")
+        for event in events:
+            if sink.emit(event):
                 count += 1
         return count
     except Exception:  # best-effort: emission must never break the loop

tests/test_agent_core_trace.py

@@ -1,6 +1,11 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Iterator
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from threading import Thread
+from typing import Any
 
 import pytest
 
@@ -9,6 +14,32 @@
 from hyrule_engineering_loop import agent_core_trace
 
 
+@contextmanager
+def _collector() -> Iterator[tuple[str, list[dict[str, Any]]]]:
+    received: list[dict[str, Any]] = []
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self) -> None:
+            length = int(self.headers.get("content-length", "0"))
+            received.append(json.loads(self.rfile.read(length)))
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(b'{"status":"stored"}')
+
+        def log_message(self, _format: str, *args: object) -> None:
+            return
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}/v1/trace", received
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=2)
+
+
 def _state() -> dict[str, object]:
     return {
         "change_id": "chg-test-1",
@@ -52,3 +83,20 @@ def test_emits_when_enabled(monkeypatch, tmp_path):
     )
     assert backend["cost"]["input_tokens"] == 100
     assert backend["run_id"] == "chg-test-1"
+
+
+def test_emits_to_collector_and_file_when_collector_url_is_set(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))
+    with _collector() as (url, received):
+        monkeypatch.setenv("HYRULE_ENGINEERING_AGENT_CORE_TRACE_COLLECTOR_URL", url)
+        count = agent_core_trace.emit_loop_trace(_state())
+
+    assert count == 3
+    assert len(sink.read_text(encoding="utf-8").strip().splitlines()) == 3
+    assert [event["event_type"] for event in received] == [
+        "model_call",
+        "tool_call",
+        "backend_execution",
+    ]