From fe8e2f00951a861247aff2c74fedaa858ebab00b Mon Sep 17 00:00:00 2001 From: Jaideep <67646710+jdchawla29@users.noreply.github.com> Date: Thu, 25 Jun 2026 12:49:14 -0700 Subject: [PATCH 1/3] feat: capture filetracking artifacts --- hud/environment/file_tracker.py | 287 +++++++++++-- hud/environment/file_tracking.py | 75 ---- hud/environment/tests/test_file_tracker.py | 68 ++++ hud/environment/tests/test_file_tracking.py | 13 +- hud/environment/workspace.py | 3 +- hud/eval/file_tracking.py | 143 ++++--- hud/eval/tests/test_file_tracking_observer.py | 380 +++++++++++------- hud/settings.py | 4 +- hud/telemetry/filetracking.py | 76 ---- hud/telemetry/tests/test_filetracking.py | 60 --- 10 files changed, 659 insertions(+), 450 deletions(-) delete mode 100644 hud/environment/file_tracking.py delete mode 100644 hud/telemetry/filetracking.py delete mode 100644 hud/telemetry/tests/test_filetracking.py diff --git a/hud/environment/file_tracker.py b/hud/environment/file_tracker.py index 3a856e51e..e997e78eb 100644 --- a/hud/environment/file_tracker.py +++ b/hud/environment/file_tracker.py @@ -2,9 +2,9 @@ Snapshots a directory tree and produces unified diffs (patches) between snapshots, so a rollout can record exactly what changed on disk over time. The -tracker is pure computation over a local ``root`` — it does no networking and -holds no credentials; a serving layer (:mod:`hud.environment.file_tracking`) -exposes it as a capability and the client side decides what to record. +tracker is pure computation over a local ``root``; ``serve_file_tracking`` wraps +one tracker in the small ``filetracking/1`` framed-JSON adapter used by +workspaces. Ported from the orchestrator sidecar's ``/proc``-scanning tracker, with the Kubernetes coupling removed (it scans an injected ``root`` instead of @@ -15,16 +15,21 @@ from __future__ import annotations +import asyncio +import base64 import difflib import fnmatch import hashlib import logging +import mimetypes import os import time from dataclasses import dataclass, field from pathlib import Path, PurePosixPath from typing import Any +from .utils import error, read_frame, reply, send_frame + LOGGER = logging.getLogger("hud.environment.file_tracker") #: Noise paths excluded from tracking entirely (build output, caches, VCS internals). @@ -68,6 +73,36 @@ #: Skip files larger than this during scanning (default 10 MB). DEFAULT_MAX_FILE_SIZE: int = 10 * 1024 * 1024 +#: Artifacts worth preserving as downloads. Common source edits are already +#: represented by diffs; HTML is included because it is often the rendered result. +DEFAULT_CAPTURE_EXTENSIONS: tuple[str, ...] = ( + ".docx", + ".htm", + ".html", + ".jpeg", + ".jpg", + ".pdf", + ".png", + ".pptx", + ".webp", + ".xlsm", + ".xlsx", +) + +CAPTURE_CONTENT_TYPES: dict[str, str] = { + ".docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ".htm": "text/html", + ".html": "text/html", + ".jpeg": "image/jpeg", + ".jpg": "image/jpeg", + ".pdf": "application/pdf", + ".png": "image/png", + ".pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation", + ".webp": "image/webp", + ".xlsm": "application/vnd.ms-excel.sheet.macroEnabled.12", + ".xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", +} + @dataclass(frozen=True) class FileEntry: @@ -179,17 +214,26 @@ class FileTracker: # instead of a full unified diff, so ``difflib`` never allocates unbounded. _MAX_DIFF_FILE_BYTES: int = 1 * 1024 * 1024 + # Artifact capture is stricter than scan/diff limits because it emits base64 + # into telemetry. Keep the raw total comfortably below the default 25 MB span request limit + # (after base64 overhead). + _MAX_CAPTURE_FILES: int = 20 + _MAX_CAPTURE_FILE_BYTES: int = 5 * 1024 * 1024 + _MAX_CAPTURE_TOTAL_BYTES: int = 15 * 1024 * 1024 + def __init__( self, root: Path | str, *, exclude_patterns: tuple[str, ...] = DEFAULT_EXCLUDE_PATTERNS, + capture_extensions: tuple[str, ...] = DEFAULT_CAPTURE_EXTENSIONS, honor_gitignore: bool = True, max_file_size: int = DEFAULT_MAX_FILE_SIZE, secret_deny_patterns: tuple[str, ...] = SECRET_DENY_PATTERNS, ) -> None: self._root = Path(root).resolve() self._exclude_patterns = exclude_patterns + self._capture_extensions = tuple(ext.lower() for ext in capture_extensions) self._secret_deny_patterns = secret_deny_patterns self._honor_gitignore = honor_gitignore self._max_file_size = max_file_size @@ -222,14 +266,14 @@ def take_baseline(self) -> Snapshot: return snapshot def advance_baseline(self) -> None: - """Re-scan and update the previous snapshot WITHOUT producing a diff. + """Re-scan and update the tracking baseline WITHOUT producing a diff. Used after scenario setup (which writes many files that are not agent - edits) and after a truncated diff (a ``git checkout`` / ``npm install`` - burst) so the next snapshot starts clean. + edits), so subsequent diffs and final artifact capture start clean. """ prev_count = len(self._previous_snapshot.files) if self._previous_snapshot else 0 snapshot = self._scan() + self._baseline_snapshot = snapshot self._previous_snapshot = snapshot if len(snapshot.files) != prev_count: LOGGER.info( @@ -249,7 +293,32 @@ def take_snapshot(self) -> DiffResult: files_changed=0, ) + return self._diff_current(self._scan()) + + def flush_changes(self) -> dict[str, Any]: + """Return the final diff plus changed artifact payloads in one scan.""" + if self._previous_snapshot is None: + LOGGER.warning("No baseline snapshot; taking one now") + baseline = self.take_baseline() + return { + "diff": DiffResult( + patches=[], + snapshot_timestamp=baseline.timestamp, + scan_duration_ms=baseline.scan_duration_ms, + files_scanned=len(baseline.files), + files_changed=0, + ).to_dict(), + "capture": self._capture_artifacts(baseline), + } + current = self._scan() + capture = self._capture_artifacts(current) + return {"diff": self._diff_current(current).to_dict(), "capture": capture} + + def _diff_current(self, current: Snapshot) -> DiffResult: + if self._previous_snapshot is None: + raise RuntimeError("cannot diff without a baseline") + diff = self._diff(self._previous_snapshot, current) # Keep budget-skipped files pending: revert each to its previous state in # the new baseline (drop added-but-skipped files) so the next scan still @@ -271,18 +340,6 @@ def take_snapshot(self) -> DiffResult: ) return diff - def get_cumulative_diff(self) -> DiffResult: - """Diff from the baseline to the current state (a final summary).""" - if self._baseline_snapshot is None: - return DiffResult( - patches=[], - snapshot_timestamp=time.time(), - scan_duration_ms=0.0, - files_scanned=0, - files_changed=0, - ) - return self._diff(self._baseline_snapshot, self._scan()) - def current_manifest(self) -> list[dict[str, Any]]: """The latest file manifest: ``[{path, size, content_hash}, ...]``. @@ -297,6 +354,106 @@ def current_manifest(self) -> list[dict[str, Any]]: for e in sorted(snapshot.files.values(), key=lambda e: e.rel_path) ] + def _capture_artifacts(self, current: Snapshot) -> dict[str, Any]: + changed: list[tuple[str, FileEntry]] = [] + if self._baseline_snapshot is not None: + for path in sorted(current.files): + entry = current.files[path] + baseline_entry = self._baseline_snapshot.files.get(path) + if baseline_entry is None: + changed.append(("added", entry)) + elif baseline_entry.content_hash != entry.content_hash: + changed.append(("modified", entry)) + + captured: list[dict[str, Any]] = [] + files_eligible = 0 + files_skipped = 0 + total_bytes = 0 + truncated = False + + for status, entry in changed: + suffix = PurePosixPath(entry.rel_path).suffix.lower() + if entry.redacted or suffix not in self._capture_extensions: + continue + files_eligible += 1 + + if ( + len(captured) >= self._MAX_CAPTURE_FILES + or entry.size > self._MAX_CAPTURE_FILE_BYTES + or total_bytes + entry.size > self._MAX_CAPTURE_TOTAL_BYTES + ): + files_skipped += 1 + truncated = True + continue + + file = self._capture_file(status, entry) + if file is None: + files_skipped += 1 + continue + + total_bytes += int(file["size"]) + captured.append(file) + + payload = { + "snapshot_timestamp": current.timestamp, + "scan_duration_ms": round(current.scan_duration_ms, 2), + "files_scanned": len(current.files), + "files_changed": len(changed), + "files_eligible": files_eligible, + "files_captured": len(captured), + "files_skipped": files_skipped, + "files": captured, + } + if truncated: + payload["truncated"] = True + return payload + + def _capture_file(self, status: str, entry: FileEntry) -> dict[str, Any] | None: + raw_path = self._root / entry.rel_path + try: + if raw_path.is_symlink(): + return None + except OSError: + return None + + path = raw_path.resolve() + try: + path.relative_to(self._root) + except ValueError: + return None + + try: + if not path.is_file(): + return None + stat = path.stat() + if stat.st_size != entry.size or stat.st_size > self._MAX_CAPTURE_FILE_BYTES: + return None + with path.open("rb") as f: + data = f.read() + except (PermissionError, OSError): + return None + + if len(data) != entry.size: + return None + + suffix = PurePosixPath(entry.rel_path).suffix.lower() + content_type = CAPTURE_CONTENT_TYPES.get( + suffix, + mimetypes.guess_type(entry.rel_path)[0] or "application/octet-stream", + ) + return { + "path": entry.rel_path, + "status": status, + "size": entry.size, + "content_hash": hashlib.sha256(data).hexdigest(), + "content_type": content_type, + "file": { + "type": "file", + "data": base64.b64encode(data).decode("ascii"), + "media_type": content_type, + }, + } + # ─── scanning ───────────────────────────────────────────────────── def _scan(self) -> Snapshot: @@ -308,8 +465,9 @@ def _scan(self) -> Snapshot: self._gitignore_patterns = self._collect_gitignore_patterns() self._gitignore_loaded = True + exclude_patterns = self._exclude_patterns + tuple(self._gitignore_patterns) if self._root.is_dir(): - self._walk_directory(str(self._root), files, budget) + self._walk_directory(str(self._root), files, budget, exclude_patterns) return Snapshot( timestamp=time.time(), @@ -318,7 +476,11 @@ def _scan(self) -> Snapshot: ) def _walk_directory( - self, abs_dir: str, files: dict[str, FileEntry], budget: ScanBudget + self, + abs_dir: str, + files: dict[str, FileEntry], + budget: ScanBudget, + exclude_patterns: tuple[str, ...], ) -> None: try: scanner = os.scandir(abs_dir) @@ -334,11 +496,11 @@ def _walk_directory( rel = os.path.relpath(entry.path, root_str).replace(os.sep, "/") is_dir = entry.is_dir(follow_symlinks=False) - if self._should_exclude(rel, is_dir): + if self._matches(rel, is_dir, exclude_patterns): continue if is_dir: - self._walk_directory(entry.path, files, budget) + self._walk_directory(entry.path, files, budget, exclude_patterns) continue if not entry.is_file(follow_symlinks=False): continue @@ -370,7 +532,7 @@ def _build_entry( redacted=prev.redacted, ) - if self._is_secret(rel): + if self._matches(rel, False, self._secret_deny_patterns): # Credential-shaped: detect change via fingerprint, never read content. return FileEntry( rel_path=rel, @@ -399,12 +561,6 @@ def _build_entry( lines=lines, ) - def _should_exclude(self, rel: str, is_dir: bool) -> bool: - return self._matches(rel, is_dir, self._exclude_patterns + tuple(self._gitignore_patterns)) - - def _is_secret(self, rel: str) -> bool: - return self._matches(rel, False, self._secret_deny_patterns) - @staticmethod def _matches(rel: str, is_dir: bool, patterns: tuple[str, ...]) -> bool: path = f"/{rel}" @@ -433,16 +589,9 @@ def _collect_gitignore_patterns(self) -> list[str]: root_gitignore = self._root / ".gitignore" if not root_gitignore.is_file(): return [] - patterns = self._parse_gitignore(root_gitignore) - if patterns: - LOGGER.info("Loaded %d gitignore patterns", len(patterns)) - return patterns - - @staticmethod - def _parse_gitignore(path: Path) -> list[str]: patterns: list[str] = [] try: - with path.open("r", encoding="utf-8", errors="replace") as f: + with root_gitignore.open("r", encoding="utf-8", errors="replace") as f: for raw in f: line = raw.strip() # Skip comments, blanks, and negations (unsupported). @@ -450,7 +599,9 @@ def _parse_gitignore(path: Path) -> list[str]: continue patterns.append(line.lstrip("/")) except (PermissionError, OSError) as exc: - LOGGER.debug("Cannot read gitignore %s: %s", path, exc) + LOGGER.debug("Cannot read gitignore %s: %s", root_gitignore, exc) + if patterns: + LOGGER.info("Loaded %d gitignore patterns", len(patterns)) return patterns @staticmethod @@ -555,12 +706,6 @@ def _patch_text( return f"File too large to diff ({size_before} -> {size_after} bytes): {path}\n" old_lines = old_entry.lines if old_entry else () new_lines = new_entry.lines if new_entry else () - return self._unified_diff(path, old_lines, new_lines) - - @staticmethod - def _unified_diff( - path: str, old_lines: tuple[str, ...] | None, new_lines: tuple[str, ...] | None - ) -> str: if old_lines is None or new_lines is None: return f"Binary file changed: {path}\n" return "\n".join( @@ -574,7 +719,60 @@ def _unified_diff( ) +class _FileTrackingHandler: + """Per-server dispatcher; one tracker shared across connections under a lock.""" + + def __init__(self, tracker: FileTracker) -> None: + self._tracker = tracker + self._lock = asyncio.Lock() + + async def handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + loop = asyncio.get_running_loop() + try: + while (msg := await read_frame(reader)) is not None: + msg_id = msg.get("id") + method = msg.get("method", "") + try: + async with self._lock: + if method == "diff": + diff = await loop.run_in_executor(None, self._tracker.take_snapshot) + result = diff.to_dict() + elif method == "snapshot": + manifest = self._tracker.current_manifest() + result = {"files": manifest, "files_scanned": len(manifest)} + elif method == "advance": + await loop.run_in_executor(None, self._tracker.advance_baseline) + result = {"advanced": True} + elif method == "flush": + result = await loop.run_in_executor(None, self._tracker.flush_changes) + else: + raise ValueError(f"unknown filetracking method: {method!r}") + except Exception as exc: # never tear the connection down on one bad call + LOGGER.debug("filetracking %s failed: %s", method, exc) + if isinstance(msg_id, int): + await send_frame(writer, error(msg_id, -32000, str(exc))) + continue + if isinstance(msg_id, int): + await send_frame(writer, reply(msg_id, result)) + except (ConnectionError, OSError): + pass + finally: + writer.close() + + +async def serve_file_tracking( + tracker: FileTracker, host: str = "127.0.0.1", port: int = 0 +) -> asyncio.Server: + """Bind a ``filetracking/1`` server for ``tracker``. Caller drives the port.""" + handler = _FileTrackingHandler(tracker) + server = await asyncio.start_server(handler.handle, host, port) + sock = server.sockets[0].getsockname() + LOGGER.info("filetracking bound on %s:%s", sock[0], sock[1]) + return server + + __all__ = [ + "DEFAULT_CAPTURE_EXTENSIONS", "DEFAULT_EXCLUDE_PATTERNS", "DEFAULT_MAX_FILE_SIZE", "SECRET_DENY_PATTERNS", @@ -583,4 +781,5 @@ def _unified_diff( "FileTracker", "PatchEntry", "Snapshot", + "serve_file_tracking", ] diff --git a/hud/environment/file_tracking.py b/hud/environment/file_tracking.py deleted file mode 100644 index 5adb4a939..000000000 --- a/hud/environment/file_tracking.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Serving layer for :class:`~hud.environment.file_tracker.FileTracker`. - -Exposes one tracker over the ``filetracking/1`` wire: a framed-JSON request -loop handling ``diff`` / ``snapshot`` / ``advance``. Scans run in a thread -executor (CPU-bound directory walks must not block the event loop) and are -serialized by a lock, since the tracker's baseline is mutable state. -""" - -from __future__ import annotations - -import asyncio -import logging -from typing import TYPE_CHECKING, Any - -from .utils import error, read_frame, reply, send_frame - -if TYPE_CHECKING: - from .file_tracker import FileTracker - -LOGGER = logging.getLogger("hud.environment.file_tracking") - - -class _FileTrackingHandler: - """Per-server dispatcher; one tracker shared across connections under a lock.""" - - def __init__(self, tracker: FileTracker) -> None: - self._tracker = tracker - self._lock = asyncio.Lock() - - async def handle(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: - try: - while (msg := await read_frame(reader)) is not None: - msg_id = msg.get("id") - method = msg.get("method", "") - try: - result = await self._dispatch(method) - except Exception as exc: # never tear the connection down on one bad call - LOGGER.debug("filetracking %s failed: %s", method, exc) - if isinstance(msg_id, int): - await send_frame(writer, error(msg_id, -32000, str(exc))) - continue - if isinstance(msg_id, int): - await send_frame(writer, reply(msg_id, result)) - except (ConnectionError, OSError): - pass - finally: - writer.close() - - async def _dispatch(self, method: str) -> dict[str, Any]: - loop = asyncio.get_running_loop() - async with self._lock: - if method == "diff": - diff = await loop.run_in_executor(None, self._tracker.take_snapshot) - return diff.to_dict() - if method == "snapshot": - manifest = self._tracker.current_manifest() - return {"files": manifest, "files_scanned": len(manifest)} - if method == "advance": - await loop.run_in_executor(None, self._tracker.advance_baseline) - return {"advanced": True} - raise ValueError(f"unknown filetracking method: {method!r}") - - -async def serve_file_tracking( - tracker: FileTracker, host: str = "127.0.0.1", port: int = 0 -) -> asyncio.Server: - """Bind a ``filetracking/1`` server for ``tracker``. Caller drives the port.""" - handler = _FileTrackingHandler(tracker) - server = await asyncio.start_server(handler.handle, host, port) - sock = server.sockets[0].getsockname() - LOGGER.info("filetracking bound on %s:%s", sock[0], sock[1]) - return server - - -__all__ = ["serve_file_tracking"] diff --git a/hud/environment/tests/test_file_tracker.py b/hud/environment/tests/test_file_tracker.py index e3695944e..9c28376ff 100644 --- a/hud/environment/tests/test_file_tracker.py +++ b/hud/environment/tests/test_file_tracker.py @@ -2,6 +2,7 @@ from __future__ import annotations +import base64 from typing import TYPE_CHECKING from hud.environment.file_tracker import FileTracker @@ -137,6 +138,73 @@ def test_vcs_config_files_are_redacted(tmp_path: Path) -> None: assert "new-token@example.com" not in patch.patch +def test_capture_changed_deliverables_since_advanced_baseline(tmp_path: Path) -> None: + (tmp_path / "setup.xlsx").write_bytes(b"setup workbook") + tracker = FileTracker(tmp_path) + tracker.take_baseline() + tracker.advance_baseline() + + (tmp_path / "analysis.py").write_text("print('tracked by diff')\n") + (tmp_path / "notes.txt").write_text("tracked by diff\n") + (tmp_path / "deliverable.xlsx").write_bytes(b"workbook bytes") + (tmp_path / "report.html").write_text("report\n") + + payload = tracker.flush_changes()["capture"] + + assert payload["files_changed"] == 4 + assert payload["files_eligible"] == 2 + assert payload["files_captured"] == 2 + assert payload["files_skipped"] == 0 + files = {file["path"]: file for file in payload["files"]} + + xlsx = files["deliverable.xlsx"] + assert xlsx["status"] == "added" + assert ( + xlsx["content_type"] + == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ) + assert len(xlsx["content_hash"]) == 64 + assert base64.b64decode(xlsx["file"]["data"]) == b"workbook bytes" + assert xlsx["file"]["type"] == "file" + assert xlsx["file"]["media_type"] == xlsx["content_type"] + + html = files["report.html"] + assert html["status"] == "added" + assert html["content_type"] == "text/html" + assert base64.b64decode(html["file"]["data"]) == b"report\n" + assert html["file"]["media_type"] == "text/html" + + +def test_capture_skips_secret_shaped_files(tmp_path: Path) -> None: + (tmp_path / ".env.xlsx").write_bytes(b"API_KEY=before") + tracker = FileTracker(tmp_path) + tracker.take_baseline() + + (tmp_path / ".env.xlsx").write_bytes(b"API_KEY=after") + payload = tracker.flush_changes()["capture"] + + assert payload["files_changed"] == 1 + assert payload["files_eligible"] == 0 + assert payload["files_captured"] == 0 + assert payload["files"] == [] + + +def test_capture_marks_cap_skips_as_truncated( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(FileTracker, "_MAX_CAPTURE_FILE_BYTES", 4) + tracker = FileTracker(tmp_path) + tracker.take_baseline() + + (tmp_path / "large.pdf").write_bytes(b"12345") + payload = tracker.flush_changes()["capture"] + + assert payload["files_eligible"] == 1 + assert payload["files_captured"] == 0 + assert payload["files_skipped"] == 1 + assert payload["truncated"] is True + + def test_per_file_diff_cap_emits_a_placeholder( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/hud/environment/tests/test_file_tracking.py b/hud/environment/tests/test_file_tracking.py index e12fbf181..b8ff78043 100644 --- a/hud/environment/tests/test_file_tracking.py +++ b/hud/environment/tests/test_file_tracking.py @@ -3,10 +3,10 @@ from __future__ import annotations import asyncio +import base64 from typing import TYPE_CHECKING, Any -from hud.environment.file_tracker import FileTracker -from hud.environment.file_tracking import serve_file_tracking +from hud.environment.file_tracker import FileTracker, serve_file_tracking from hud.environment.utils import read_frame, send_frame if TYPE_CHECKING: @@ -59,6 +59,15 @@ async def test_diff_snapshot_advance_roundtrip(tmp_path: Path) -> None: (tmp_path / "b.txt").write_text("z\n") await _call(reader, writer, "advance") assert (await _call(reader, writer, "diff"))["files_changed"] == 0 + + # flush() returns the trailing diff plus bounded changed deliverables. + (tmp_path / "report.xlsx").write_bytes(b"spreadsheet") + flush = await _call(reader, writer, "flush") + assert flush["diff"]["files_changed"] == 1 + capture = flush["capture"] + assert capture["files_captured"] == 1 + assert capture["files"][0]["path"] == "report.xlsx" + assert base64.b64decode(capture["files"][0]["file"]["data"]) == b"spreadsheet" finally: writer.close() await writer.wait_closed() diff --git a/hud/environment/workspace.py b/hud/environment/workspace.py index 096e83142..91b1487fc 100644 --- a/hud/environment/workspace.py +++ b/hud/environment/workspace.py @@ -249,8 +249,7 @@ async def start(self) -> None: async def _start_file_tracking(self) -> None: """Take the baseline snapshot and bind the filetracking/1 server.""" - from .file_tracker import FileTracker - from .file_tracking import serve_file_tracking + from .file_tracker import FileTracker, serve_file_tracking tracker = FileTracker(self.root) # The baseline walk is CPU-bound; keep it off the event loop. diff --git a/hud/eval/file_tracking.py b/hud/eval/file_tracking.py index 2cc5e4ecc..f55c21f11 100644 --- a/hud/eval/file_tracking.py +++ b/hud/eval/file_tracking.py @@ -2,9 +2,9 @@ Wraps the agent loop: if the env published a ``filetracking/1`` capability and file tracking is on, open it, skip the scenario-setup churn, then sample diffs -on a fixed interval and emit each as a ``hud.filetracking.v1`` span. Decoupled -from the tool loop — spans are self-timestamped and the viewer correlates them -to steps by time. +on a fixed interval. On teardown it flushes the trailing diff plus changed +deliverable artifacts. Decoupled from the tool loop — spans are self-timestamped +and the viewer correlates them to steps by time. """ from __future__ import annotations @@ -14,10 +14,19 @@ import json import logging from contextlib import asynccontextmanager -from typing import TYPE_CHECKING, Any, Self +from typing import TYPE_CHECKING, Any, Literal, Self, TypeAlias from urllib.parse import urlsplit -from hud.telemetry.filetracking import emit_file_diff, emit_file_snapshot +from hud.telemetry.context import get_current_trace_id +from hud.telemetry.exporter import queue_span +from hud.telemetry.span import ( + PAYLOAD_ATTRIBUTE, + SCHEMA_ATTRIBUTE, + TASK_RUN_ID_ATTRIBUTE, + Span, + new_span_id, + normalize_trace_id, +) from hud.utils.time import now_iso if TYPE_CHECKING: @@ -29,18 +38,23 @@ logger = logging.getLogger("hud.eval.file_tracking") _DRAIN_TIMEOUT = 10.0 +_FRAME_LIMIT_BYTES = 32 * 1024 * 1024 +_FILETRACKING_SCHEMA = "hud.filetracking.v1" +_FileTrackingSpanName: TypeAlias = Literal[ + "filetracking.capture", + "filetracking.diff", + "filetracking.snapshot", +] class FileTrackingClient: - """Live ``filetracking/1`` connection: ``diff`` / ``snapshot`` / ``advance``.""" + """Live ``filetracking/1`` connection.""" def __init__( self, - capability: Capability, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, ) -> None: - self.capability = capability self._reader = reader self._writer = writer self._id = 0 @@ -51,24 +65,19 @@ async def connect(cls, cap: Capability) -> Self: parts = urlsplit(cap.url) if parts.hostname is None or parts.port is None: raise ValueError(f"filetracking capability missing host or port: {cap.url!r}") - reader, writer = await asyncio.open_connection(parts.hostname, parts.port) - return cls(cap, reader, writer) - - async def diff(self) -> dict[str, Any]: - return await self._call("diff") - - async def snapshot(self) -> dict[str, Any]: - return await self._call("snapshot") - - async def advance(self) -> None: - await self._call("advance") + reader, writer = await asyncio.open_connection( + parts.hostname, + parts.port, + limit=_FRAME_LIMIT_BYTES, + ) + return cls(reader, writer) async def close(self) -> None: self._writer.close() with contextlib.suppress(OSError): await self._writer.wait_closed() - async def _call(self, method: str) -> dict[str, Any]: + async def call(self, method: str) -> dict[str, Any]: async with self._lock: self._id += 1 msg_id = self._id @@ -93,7 +102,7 @@ async def _call(self, method: str) -> dict[str, Any]: @asynccontextmanager async def file_tracking_observer(client: HudClient) -> AsyncIterator[None]: - """Stream workspace diffs to telemetry for the duration of the ``with`` block. + """Stream workspace diffs and final artifacts during the ``with`` block. A no-op unless telemetry is enabled and the manifest has a ``filetracking`` binding. The binding's presence is the authoritative opt-in: it is published @@ -121,8 +130,12 @@ async def file_tracking_observer(client: HudClient) -> AsyncIterator[None]: ft: FileTrackingClient | None = None try: ft = await FileTrackingClient.connect(cap) - await ft.advance() - emit_file_snapshot(await ft.snapshot(), started_at=now_iso()) + await ft.call("advance") + _emit_file_tracking( + "filetracking.snapshot", + await ft.call("snapshot"), + started_at=now_iso(), + ) except Exception as exc: if ft is not None: with contextlib.suppress(Exception): @@ -132,7 +145,23 @@ async def file_tracking_observer(client: HudClient) -> AsyncIterator[None]: return stop = asyncio.Event() - task = asyncio.create_task(_poll(ft, settings.file_tracking_interval, stop)) + + async def poll_diffs() -> None: + while not stop.is_set(): + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for(stop.wait(), timeout=settings.file_tracking_interval) + if stop.is_set(): + return + started_at = now_iso() + try: + diff = await ft.call("diff") + except Exception as exc: + logger.debug("file tracking diff failed: %s", exc) + continue + if diff.get("files_changed"): + _emit_file_tracking("filetracking.diff", diff, started_at=started_at) + + task = asyncio.create_task(poll_diffs()) try: yield finally: @@ -145,34 +174,52 @@ async def file_tracking_observer(client: HudClient) -> AsyncIterator[None]: task.cancel() with contextlib.suppress(asyncio.CancelledError): await task - # Trailing diff: edits since the last successful sample. Attempt it in - # both paths (clean drain or forced cancel); bound it so a connection - # desynced by the cancel above can't wedge teardown. ``_emit_once`` logs - # and swallows its own failures. - with contextlib.suppress(asyncio.TimeoutError): - await asyncio.wait_for(_emit_once(ft), _DRAIN_TIMEOUT) + # Trailing diff + artifacts since the last successful sample. Bound it so + # a connection desynced by the cancel above can't wedge teardown. + try: + flush = await asyncio.wait_for(ft.call("flush"), _DRAIN_TIMEOUT) + except TimeoutError: + pass + except Exception as exc: + logger.debug("file tracking flush failed: %s", exc) + else: + started_at = now_iso() + diff = flush.get("diff") + if isinstance(diff, dict) and diff.get("files_changed"): + _emit_file_tracking("filetracking.diff", diff, started_at=started_at) + + capture = flush.get("capture") + if isinstance(capture, dict) and capture.get("files_captured"): + _emit_file_tracking("filetracking.capture", capture, started_at=started_at) with contextlib.suppress(Exception): await ft.close() -async def _poll(ft: FileTrackingClient, interval: float, stop: asyncio.Event) -> None: - while not stop.is_set(): - with contextlib.suppress(asyncio.TimeoutError): - await asyncio.wait_for(stop.wait(), timeout=interval) - if stop.is_set(): - return - await _emit_once(ft) - - -async def _emit_once(ft: FileTrackingClient) -> None: - started_at = now_iso() - try: - result = await ft.diff() - except Exception as exc: - logger.debug("file tracking diff failed: %s", exc) - return - if result.get("files_changed"): - emit_file_diff(result, started_at=started_at) +def _emit_file_tracking( + span_name: _FileTrackingSpanName, + payload: dict[str, Any], + *, + started_at: str, + ended_at: str | None = None, +) -> bool: + task_run_id = get_current_trace_id() + if task_run_id is None: + return False + span = Span( + name=span_name, + trace_id=normalize_trace_id(task_run_id), + span_id=new_span_id(), + start_time=started_at, + end_time=ended_at or now_iso(), + status_code="OK", + attributes={ + SCHEMA_ATTRIBUTE: _FILETRACKING_SCHEMA, + TASK_RUN_ID_ATTRIBUTE: task_run_id, + PAYLOAD_ATTRIBUTE: payload, + }, + ) + queue_span(span.model_dump(mode="json")) + return True __all__ = ["file_tracking_observer"] diff --git a/hud/eval/tests/test_file_tracking_observer.py b/hud/eval/tests/test_file_tracking_observer.py index 522850be5..b66627b36 100644 --- a/hud/eval/tests/test_file_tracking_observer.py +++ b/hud/eval/tests/test_file_tracking_observer.py @@ -1,141 +1,239 @@ -"""file_tracking_observer: setup must gate polling. - -The observer re-baselines past scenario setup and emits the manifest anchor -before it starts streaming diffs. If that setup fails, it must not poll — a -stale baseline would misattribute scenario-setup edits to the agent. -""" - -from __future__ import annotations - -import asyncio -from typing import TYPE_CHECKING, Any - -from hud.capabilities import Capability -from hud.eval import file_tracking as observer - -if TYPE_CHECKING: - import pytest - -_CAP = Capability(name="filetracking", protocol="filetracking/1", url="tcp://127.0.0.1:1") - - -class _FakeFt: - def __init__(self, *, advance_raises: bool = False) -> None: - self.advance_raises = advance_raises - self.advance_calls = 0 - self.snapshot_calls = 0 - self.diff_calls = 0 - self.close_calls = 0 - - async def advance(self) -> dict[str, Any]: - self.advance_calls += 1 - if self.advance_raises: - raise RuntimeError("advance boom") - return {"advanced": True} - - async def snapshot(self) -> dict[str, Any]: - self.snapshot_calls += 1 - return {"files": [], "files_scanned": 0} - - async def diff(self) -> dict[str, Any]: - self.diff_calls += 1 - return {"files_changed": 1, "patches": [{"path": "a.txt"}]} - - async def close(self) -> None: - self.close_calls += 1 - - -class _BoundClient: - def binding(self, name: str) -> object: - return _CAP - - -def _record_emitters(monkeypatch: pytest.MonkeyPatch) -> tuple[list[Any], list[Any]]: - diffs: list[Any] = [] - snapshots: list[Any] = [] - - def _diff(payload: Any, *, started_at: str, ended_at: str | None = None) -> bool: - diffs.append(payload) - return True - - def _snapshot(payload: Any, *, started_at: str, ended_at: str | None = None) -> bool: - snapshots.append(payload) - return True - - monkeypatch.setattr(observer, "emit_file_diff", _diff) - monkeypatch.setattr(observer, "emit_file_snapshot", _snapshot) - return diffs, snapshots - - -def _connects_to(monkeypatch: pytest.MonkeyPatch, ft: _FakeFt) -> None: - class _FakeFileTrackingClient: - @classmethod - async def connect(cls, cap: Capability) -> _FakeFt: - assert cap == _CAP - return ft - - monkeypatch.setattr(observer, "FileTrackingClient", _FakeFileTrackingClient) - - -async def test_setup_failure_skips_polling(monkeypatch: pytest.MonkeyPatch) -> None: - from hud.settings import settings - - monkeypatch.setattr(settings, "telemetry_enabled", True) - monkeypatch.setattr(settings, "file_tracking_interval", 0.01) - diffs, snapshots = _record_emitters(monkeypatch) - ft = _FakeFt(advance_raises=True) - _connects_to(monkeypatch, ft) - - async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] - await asyncio.sleep(0.05) - - assert ft.advance_calls == 1 - # advance() raised, so the anchor snapshot and all diff polling are skipped. - assert ft.snapshot_calls == 0 - assert ft.diff_calls == 0 - assert ft.close_calls == 1 - assert diffs == [] - assert snapshots == [] - - -async def test_successful_setup_anchors_and_polls(monkeypatch: pytest.MonkeyPatch) -> None: - from hud.settings import settings - - monkeypatch.setattr(settings, "telemetry_enabled", True) - monkeypatch.setattr(settings, "file_tracking_interval", 0.01) - diffs, snapshots = _record_emitters(monkeypatch) - ft = _FakeFt() - _connects_to(monkeypatch, ft) - - async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] - await asyncio.sleep(0.05) - - assert ft.advance_calls == 1 - assert len(snapshots) == 1 # manifest anchor emitted once - assert ft.diff_calls >= 1 - assert ft.close_calls == 1 - assert diffs # at least one diff streamed - - -async def test_connect_failure_does_not_break_the_rollout(monkeypatch: pytest.MonkeyPatch) -> None: - from hud.settings import settings - - monkeypatch.setattr(settings, "telemetry_enabled", True) - diffs, snapshots = _record_emitters(monkeypatch) - - class _FailingFileTrackingClient: - @classmethod - async def connect(cls, cap: Capability) -> object: - assert cap == _CAP - raise ConnectionError("tunnel refused") - - monkeypatch.setattr(observer, "FileTrackingClient", _FailingFileTrackingClient) - - # A failed connection must degrade to a no-op, not raise into the agent loop. - ran = False - async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] - ran = True - - assert ran - assert diffs == [] - assert snapshots == [] +"""file_tracking_observer: setup must gate polling. + +The observer re-baselines past scenario setup and emits the manifest anchor +before it starts streaming diffs. If that setup fails, it must not poll — a +stale baseline would misattribute scenario-setup edits to the agent. +""" + +from __future__ import annotations + +import asyncio +import json +from typing import TYPE_CHECKING, Any + +from hud.capabilities import Capability +from hud.eval import file_tracking as observer +from hud.telemetry.context import set_trace_context + +if TYPE_CHECKING: + import pytest + +_CAP = Capability(name="filetracking", protocol="filetracking/1", url="tcp://127.0.0.1:1") + + +class _FakeFt: + def __init__(self, *, advance_raises: bool = False) -> None: + self.advance_raises = advance_raises + self.advance_calls = 0 + self.snapshot_calls = 0 + self.diff_calls = 0 + self.flush_calls = 0 + self.close_calls = 0 + + async def close(self) -> None: + self.close_calls += 1 + + async def call(self, method: str) -> dict[str, Any]: + if method == "advance": + self.advance_calls += 1 + if self.advance_raises: + raise RuntimeError("advance boom") + return {"advanced": True} + if method == "snapshot": + self.snapshot_calls += 1 + return {"files": [], "files_scanned": 0} + if method == "diff": + self.diff_calls += 1 + return {"files_changed": 1, "patches": [{"path": "a.txt"}]} + if method == "flush": + self.flush_calls += 1 + return { + "diff": {"files_changed": 1, "patches": [{"path": "final.txt"}]}, + "capture": {"files_captured": 1, "files": [{"path": "report.xlsx"}]}, + } + raise ValueError(method) + + +class _BoundClient: + def binding(self, name: str) -> object: + return _CAP + + +def _record_emitters( + monkeypatch: pytest.MonkeyPatch, +) -> tuple[list[Any], list[Any], list[Any]]: + diffs: list[Any] = [] + snapshots: list[Any] = [] + captures: list[Any] = [] + + def _emit( + span_name: str, + payload: Any, + *, + started_at: str, + ended_at: str | None = None, + ) -> bool: + if span_name == "filetracking.diff": + diffs.append(payload) + elif span_name == "filetracking.snapshot": + snapshots.append(payload) + elif span_name == "filetracking.capture": + captures.append(payload) + return True + + monkeypatch.setattr(observer, "_emit_file_tracking", _emit) + return diffs, snapshots, captures + + +def _connects_to(monkeypatch: pytest.MonkeyPatch, ft: _FakeFt) -> None: + class _FakeFileTrackingClient: + @classmethod + async def connect(cls, cap: Capability) -> _FakeFt: + assert cap == _CAP + return ft + + monkeypatch.setattr(observer, "FileTrackingClient", _FakeFileTrackingClient) + + +async def test_setup_failure_skips_polling(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.settings import settings + + monkeypatch.setattr(settings, "telemetry_enabled", True) + monkeypatch.setattr(settings, "file_tracking_interval", 0.01) + diffs, snapshots, captures = _record_emitters(monkeypatch) + ft = _FakeFt(advance_raises=True) + _connects_to(monkeypatch, ft) + + async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] + await asyncio.sleep(0.05) + + assert ft.advance_calls == 1 + # advance() raised, so the anchor snapshot and all diff polling are skipped. + assert ft.snapshot_calls == 0 + assert ft.diff_calls == 0 + assert ft.flush_calls == 0 + assert ft.close_calls == 1 + assert diffs == [] + assert snapshots == [] + assert captures == [] + + +async def test_successful_setup_anchors_and_polls(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.settings import settings + + monkeypatch.setattr(settings, "telemetry_enabled", True) + monkeypatch.setattr(settings, "file_tracking_interval", 0.01) + diffs, snapshots, captures = _record_emitters(monkeypatch) + ft = _FakeFt() + _connects_to(monkeypatch, ft) + + async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] + await asyncio.sleep(0.05) + + assert ft.advance_calls == 1 + assert len(snapshots) == 1 # manifest anchor emitted once + assert ft.diff_calls >= 1 + assert ft.flush_calls == 1 + assert ft.close_calls == 1 + assert diffs # at least one diff streamed + assert {"files_changed": 1, "patches": [{"path": "final.txt"}]} in diffs + assert captures == [{"files_captured": 1, "files": [{"path": "report.xlsx"}]}] + + +async def test_connect_failure_does_not_break_the_rollout(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.settings import settings + + monkeypatch.setattr(settings, "telemetry_enabled", True) + diffs, snapshots, captures = _record_emitters(monkeypatch) + + class _FailingFileTrackingClient: + @classmethod + async def connect(cls, cap: Capability) -> object: + assert cap == _CAP + raise ConnectionError("tunnel refused") + + monkeypatch.setattr(observer, "FileTrackingClient", _FailingFileTrackingClient) + + # A failed connection must degrade to a no-op, not raise into the agent loop. + ran = False + async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] + ran = True + + assert ran + assert diffs == [] + assert snapshots == [] + assert captures == [] + + +async def test_file_tracking_client_reads_large_snapshot_frame() -> None: + large_path = "a" * (80 * 1024) + + async def handle( + reader: asyncio.StreamReader, + writer: asyncio.StreamWriter, + ) -> None: + await reader.readline() + writer.write( + json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "result": { + "files_scanned": 1, + "files": [{"path": large_path, "sha256": "0" * 64, "size": 1}], + }, + }, + separators=(",", ":"), + ).encode("utf-8") + + b"\n" + ) + await writer.drain() + writer.close() + await writer.wait_closed() + + server = await asyncio.start_server(handle, "127.0.0.1", 0) + host, port = server.sockets[0].getsockname()[:2] + cap = Capability(name="filetracking", protocol="filetracking/1", url=f"tcp://{host}:{port}") + try: + ft = await observer.FileTrackingClient.connect(cap) + snapshot = await ft.call("snapshot") + assert snapshot["files"][0]["path"] == large_path + finally: + server.close() + await server.wait_closed() + + +def test_emit_file_tracking_noops_without_a_trace_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: list[dict[str, Any]] = [] + monkeypatch.setattr(observer, "queue_span", captured.append) + + emitted = observer._emit_file_tracking( + "filetracking.diff", + {"files_changed": 1}, + started_at="2026-06-18T22:00:00Z", + ) + + assert emitted is False + assert captured == [] + + +def test_emit_file_tracking_builds_schema_tagged_span(monkeypatch: pytest.MonkeyPatch) -> None: + captured: list[dict[str, Any]] = [] + monkeypatch.setattr(observer, "queue_span", captured.append) + + with set_trace_context("run-abc-123"): + emitted = observer._emit_file_tracking( + "filetracking.capture", + {"files_captured": 1, "files": [{"path": "report.xlsx"}]}, + started_at="2026-06-18T22:00:00Z", + ) + + assert emitted is True + span = captured[0] + assert span["name"] == "filetracking.capture" + assert span["attributes"]["hud.schema"] == "hud.filetracking.v1" + assert span["attributes"]["hud.task_run_id"] == "run-abc-123" + assert span["attributes"]["hud.payload"]["files"][0]["path"] == "report.xlsx" + assert len(span["trace_id"]) == 32 diff --git a/hud/settings.py b/hud/settings.py index 71ffb8f15..d7c1757fe 100644 --- a/hud/settings.py +++ b/hud/settings.py @@ -162,8 +162,8 @@ def settings_customise_sources( file_tracking_enabled: bool = Field( default=True, description="Publish a workspace's filetracking/1 capability and stream file-change " - "diffs to telemetry during a rollout. Enabled by default; set " - "HUD_FILE_TRACKING_ENABLED=false to opt out.", + "diffs plus artifacts to telemetry during a rollout. " + "Enabled by default; set HUD_FILE_TRACKING_ENABLED=false to opt out.", validation_alias="HUD_FILE_TRACKING_ENABLED", ) diff --git a/hud/telemetry/filetracking.py b/hud/telemetry/filetracking.py deleted file mode 100644 index b943bc523..000000000 --- a/hud/telemetry/filetracking.py +++ /dev/null @@ -1,76 +0,0 @@ -"""Emit file-tracking observations as ``hud.filetracking.v1`` telemetry spans. - -A standalone span schema — *not* attached to tool-call results. The rollout -observer pulls a diff from the workspace's ``filetracking/1`` capability and -calls :func:`emit_file_diff`, which builds an OTel-shaped span under the active -trace context and hands it to the exporter. Each span is self-timestamped so -the viewer correlates file changes to the step timeline by time. -""" - -from __future__ import annotations - -from typing import Any - -from hud.telemetry.context import get_current_trace_id -from hud.telemetry.exporter import queue_span -from hud.telemetry.span import ( - PAYLOAD_ATTRIBUTE, - SCHEMA_ATTRIBUTE, - TASK_RUN_ID_ATTRIBUTE, - Span, - new_span_id, - normalize_trace_id, -) -from hud.utils.time import now_iso - -#: Schema tag the platform projector dispatches on to build ``file_change`` / -#: ``file_snapshot`` events. -FILETRACKING_SCHEMA = "hud.filetracking.v1" - -DIFF_SPAN_NAME = "filetracking.diff" -SNAPSHOT_SPAN_NAME = "filetracking.snapshot" - - -def _emit(payload: dict[str, Any], name: str, started_at: str, ended_at: str | None) -> bool: - """Build and queue one file-tracking span. No-ops outside a rollout.""" - task_run_id = get_current_trace_id() - if task_run_id is None: - return False - span = Span( - name=name, - trace_id=normalize_trace_id(task_run_id), - span_id=new_span_id(), - start_time=started_at, - end_time=ended_at or now_iso(), - status_code="OK", - attributes={ - SCHEMA_ATTRIBUTE: FILETRACKING_SCHEMA, - TASK_RUN_ID_ATTRIBUTE: task_run_id, - PAYLOAD_ATTRIBUTE: payload, - }, - ) - queue_span(span.model_dump(mode="json")) - return True - - -def emit_file_diff( - payload: dict[str, Any], *, started_at: str, ended_at: str | None = None -) -> bool: - """Emit a per-scan diff (``DiffResult.to_dict()`` shape); ``False`` outside a rollout.""" - return _emit(payload, DIFF_SPAN_NAME, started_at, ended_at) - - -def emit_file_snapshot( - payload: dict[str, Any], *, started_at: str, ended_at: str | None = None -) -> bool: - """Emit the baseline manifest (``{files, files_scanned}``). The reconstruction anchor.""" - return _emit(payload, SNAPSHOT_SPAN_NAME, started_at, ended_at) - - -__all__ = [ - "DIFF_SPAN_NAME", - "FILETRACKING_SCHEMA", - "SNAPSHOT_SPAN_NAME", - "emit_file_diff", - "emit_file_snapshot", -] diff --git a/hud/telemetry/tests/test_filetracking.py b/hud/telemetry/tests/test_filetracking.py deleted file mode 100644 index b902a797c..000000000 --- a/hud/telemetry/tests/test_filetracking.py +++ /dev/null @@ -1,60 +0,0 @@ -"""The hud.filetracking.v1 span emitter.""" - -from __future__ import annotations - -from typing import Any -from unittest.mock import patch - -from hud.telemetry.context import set_trace_context -from hud.telemetry.filetracking import ( - DIFF_SPAN_NAME, - FILETRACKING_SCHEMA, - SNAPSHOT_SPAN_NAME, - emit_file_diff, - emit_file_snapshot, -) - -_PAYLOAD = {"files_changed": 1, "patches": [{"path": "a.txt", "status": "modified"}]} - - -def test_emit_noops_without_a_trace_context() -> None: - with patch("hud.telemetry.filetracking.queue_span") as queue_span: - emitted = emit_file_diff(_PAYLOAD, started_at="2026-06-18T22:00:00Z") - assert emitted is False - queue_span.assert_not_called() - - -def test_emit_builds_a_schema_tagged_span() -> None: - captured: list[dict[str, Any]] = [] - with ( - patch("hud.telemetry.filetracking.queue_span", side_effect=captured.append), - set_trace_context("run-abc-123"), - ): - emitted = emit_file_diff(_PAYLOAD, started_at="2026-06-18T22:00:00Z") - - assert emitted is True - assert len(captured) == 1 - span = captured[0] - assert span["name"] == DIFF_SPAN_NAME - attributes = span["attributes"] - assert attributes["hud.schema"] == FILETRACKING_SCHEMA - assert attributes["hud.task_run_id"] == "run-abc-123" - assert attributes["hud.payload"]["files_changed"] == 1 - # trace_id is the normalized 32-hex telemetry id, not the raw run id. - assert len(span["trace_id"]) == 32 - - -def test_emit_snapshot_uses_the_snapshot_span_name() -> None: - captured: list[dict[str, Any]] = [] - manifest = {"files_scanned": 2, "files": [{"path": "a", "size": 1, "content_hash": "h"}]} - with ( - patch("hud.telemetry.filetracking.queue_span", side_effect=captured.append), - set_trace_context("run-abc-123"), - ): - emitted = emit_file_snapshot(manifest, started_at="2026-06-18T22:00:00Z") - - assert emitted is True - span = captured[0] - assert span["name"] == SNAPSHOT_SPAN_NAME - assert span["attributes"]["hud.schema"] == FILETRACKING_SCHEMA - assert span["attributes"]["hud.payload"]["files_scanned"] == 2 From 6b6cd1d048a1a262c8bbc4ad3f44c2f32a407b6a Mon Sep 17 00:00:00 2001 From: Jaideep <67646710+jdchawla29@users.noreply.github.com> Date: Thu, 25 Jun 2026 12:53:24 -0700 Subject: [PATCH 2/3] style: format file tracking tests --- hud/environment/tests/test_file_tracker.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/hud/environment/tests/test_file_tracker.py b/hud/environment/tests/test_file_tracker.py index 9c28376ff..ddc2798eb 100644 --- a/hud/environment/tests/test_file_tracker.py +++ b/hud/environment/tests/test_file_tracker.py @@ -160,8 +160,7 @@ def test_capture_changed_deliverables_since_advanced_baseline(tmp_path: Path) -> xlsx = files["deliverable.xlsx"] assert xlsx["status"] == "added" assert ( - xlsx["content_type"] - == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + xlsx["content_type"] == "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" ) assert len(xlsx["content_hash"]) == 64 assert base64.b64decode(xlsx["file"]["data"]) == b"workbook bytes" From 014529c69d294a8113085383ec5cc3e4ba157b76 Mon Sep 17 00:00:00 2001 From: Jaideep <67646710+jdchawla29@users.noreply.github.com> Date: Thu, 25 Jun 2026 13:08:05 -0700 Subject: [PATCH 3/3] fix: preserve capped filetracking captures --- hud/eval/file_tracking.py | 10 ++++- hud/eval/tests/test_file_tracking_observer.py | 43 ++++++++++++++++++- 2 files changed, 50 insertions(+), 3 deletions(-) diff --git a/hud/eval/file_tracking.py b/hud/eval/file_tracking.py index f55c21f11..c85d07323 100644 --- a/hud/eval/file_tracking.py +++ b/hud/eval/file_tracking.py @@ -38,7 +38,8 @@ logger = logging.getLogger("hud.eval.file_tracking") _DRAIN_TIMEOUT = 10.0 -_FRAME_LIMIT_BYTES = 32 * 1024 * 1024 +# flush can carry a 50 MiB diff plus base64 capture and JSON escaping overhead. +_FRAME_LIMIT_BYTES = 160 * 1024 * 1024 _FILETRACKING_SCHEMA = "hud.filetracking.v1" _FileTrackingSpanName: TypeAlias = Literal[ "filetracking.capture", @@ -189,7 +190,12 @@ async def poll_diffs() -> None: _emit_file_tracking("filetracking.diff", diff, started_at=started_at) capture = flush.get("capture") - if isinstance(capture, dict) and capture.get("files_captured"): + if isinstance(capture, dict) and ( + capture.get("files_captured") + or capture.get("files_skipped") + or capture.get("files_eligible") + or capture.get("truncated") + ): _emit_file_tracking("filetracking.capture", capture, started_at=started_at) with contextlib.suppress(Exception): await ft.close() diff --git a/hud/eval/tests/test_file_tracking_observer.py b/hud/eval/tests/test_file_tracking_observer.py index b66627b36..5011e9bc3 100644 --- a/hud/eval/tests/test_file_tracking_observer.py +++ b/hud/eval/tests/test_file_tracking_observer.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any from hud.capabilities import Capability +from hud.environment.file_tracker import FileTracker from hud.eval import file_tracking as observer from hud.telemetry.context import set_trace_context @@ -22,8 +23,14 @@ class _FakeFt: - def __init__(self, *, advance_raises: bool = False) -> None: + def __init__( + self, + *, + advance_raises: bool = False, + flush_result: dict[str, Any] | None = None, + ) -> None: self.advance_raises = advance_raises + self.flush_result = flush_result self.advance_calls = 0 self.snapshot_calls = 0 self.diff_calls = 0 @@ -47,6 +54,8 @@ async def call(self, method: str) -> dict[str, Any]: return {"files_changed": 1, "patches": [{"path": "a.txt"}]} if method == "flush": self.flush_calls += 1 + if self.flush_result is not None: + return self.flush_result return { "diff": {"files_changed": 1, "patches": [{"path": "final.txt"}]}, "capture": {"files_captured": 1, "files": [{"path": "report.xlsx"}]}, @@ -140,6 +149,32 @@ async def test_successful_setup_anchors_and_polls(monkeypatch: pytest.MonkeyPatc assert captures == [{"files_captured": 1, "files": [{"path": "report.xlsx"}]}] +async def test_flush_emits_skipped_capture_metadata(monkeypatch: pytest.MonkeyPatch) -> None: + from hud.settings import settings + + capture = { + "files_changed": 1, + "files_eligible": 1, + "files_captured": 0, + "files_skipped": 1, + "truncated": True, + "files": [], + } + monkeypatch.setattr(settings, "telemetry_enabled", True) + monkeypatch.setattr(settings, "file_tracking_interval", 60.0) + diffs, snapshots, captures = _record_emitters(monkeypatch) + ft = _FakeFt(flush_result={"diff": {"files_changed": 0, "patches": []}, "capture": capture}) + _connects_to(monkeypatch, ft) + + async with observer.file_tracking_observer(_BoundClient()): # type: ignore[arg-type] + pass + + assert ft.flush_calls == 1 + assert len(snapshots) == 1 + assert diffs == [] + assert captures == [capture] + + async def test_connect_failure_does_not_break_the_rollout(monkeypatch: pytest.MonkeyPatch) -> None: from hud.settings import settings @@ -203,6 +238,12 @@ async def handle( await server.wait_closed() +def test_file_tracking_frame_limit_covers_flush_caps() -> None: + encoded_capture_cap = 4 * ((FileTracker._MAX_CAPTURE_TOTAL_BYTES + 2) // 3) + escaped_diff_cap = 2 * FileTracker._MAX_DIFF_BYTES + assert escaped_diff_cap + encoded_capture_cap <= observer._FRAME_LIMIT_BYTES + + def test_emit_file_tracking_noops_without_a_trace_context( monkeypatch: pytest.MonkeyPatch, ) -> None: