From 749f7e552e8db9282ac49e0bb64d509fb16ab09a Mon Sep 17 00:00:00 2001 From: David Leen Date: Thu, 27 Aug 2026 18:44:58 -0700 Subject: [PATCH 1/5] feat(harbor): upload standard verifier output --- examples/harbor/README.md | 4 +- py/src/braintrust/integrations/harbor/atif.py | 69 +++- .../braintrust/integrations/harbor/compat.py | 4 + .../braintrust/integrations/harbor/plugin.py | 184 +++++++++-- .../integrations/harbor/test_harbor.py | 301 +++++++++++++++++- 5 files changed, 527 insertions(+), 35 deletions(-) diff --git a/examples/harbor/README.md b/examples/harbor/README.md index 834af879..0189a7d0 100644 --- a/examples/harbor/README.md +++ b/examples/harbor/README.md @@ -1,6 +1,6 @@ # Harbor + Braintrust -Runs a small, self-contained [Harbor](https://harborframework.com/) evaluation and uses Harbor's native Braintrust job plugin to sync the result. Braintrust receives a managed dataset, an experiment row for the final trial, verifier rewards, and the Harbor lifecycle and ATIF trace. +Runs a small, self-contained [Harbor](https://harborframework.com/) evaluation and uses Harbor's native Braintrust job plugin to sync the result. Braintrust receives a managed dataset, an experiment row for the final trial, verifier rewards and standard verifier output, and the Harbor lifecycle and ATIF trace. The plugin is discovered automatically through Harbor's `braintrust` entry point. The Braintrust API key remains in the host process; it is not passed into the task container. @@ -39,6 +39,8 @@ uv run harbor run \ The agent solves the task in `task/`, and Harbor's verifier emits a normalized `reward` plus an `answer_length` metric. The plugin creates `jobs/braintrust-harbor-example/braintrust-sync.json` after synchronization. +With the default `attachments=verifier-details` mode, the verification span and each score also include Harbor's captured `test-stdout.txt`, optional `test-stderr.txt`, and conventional `ctrf.json` output when present. The size-bounded summary and complete redacted `verifier-output.json` attachment appear together in the span output. Structured CTRF fields with sensitive key names use the plugin's standard redaction, and configured `redact_patterns` apply to both structured CTRF strings and raw verifier text. + By default, the plugin uses `Harbor` as the Braintrust project name. Override it with `--plugin-kwarg project_name=example-harbor` or through `.env`: ```dotenv diff --git a/py/src/braintrust/integrations/harbor/atif.py b/py/src/braintrust/integrations/harbor/atif.py index 3b203d71..2520d923 100644 --- a/py/src/braintrust/integrations/harbor/atif.py +++ b/py/src/braintrust/integrations/harbor/atif.py @@ -61,6 +61,28 @@ class ATIFImportResult: repairs: tuple[str, ...] = () imported_llm_spans: int = 0 imported_tool_spans: int = 0 + attachment_bytes: int = 0 + + +@dataclass +class _AttachmentBudget: + remaining: int + used: int = 0 + + def consume(self, size: int) -> None: + self.remaining -= size + self.used += size + + +def _resolve_attachment_budget( + config: PluginConfig, + available_bytes: int | None, + shared_budget: _AttachmentBudget | None, +) -> _AttachmentBudget: + if shared_budget is not None: + return shared_budget + limit = config.max_total_attachment_bytes if available_bytes is None else available_bytes + return _AttachmentBudget(max(0, limit)) def _timestamp(value: Any) -> tuple[float | None, bool]: @@ -193,14 +215,16 @@ def _content( config: PluginConfig, notes: _Notes, context: str, -) -> tuple[Any, bool]: + attachment_budget: _AttachmentBudget, +) -> tuple[Any, bool, int]: if isinstance(value, str) or value is None: bounded = _bounded(value, config, notes, context) - return bounded.value, bounded.complete + return bounded.value, bounded.complete, 0 if not isinstance(value, list): - return _bounded(value, config, notes, context).value, False + return _bounded(value, config, notes, context).value, False, 0 result: list[Any] = [] complete = True + attachment_bytes = 0 trajectory_root = trajectory_dir.resolve() for index, part in enumerate(value): part_context = f"{context}[{index}]" @@ -226,9 +250,9 @@ def _content( complete = False result.append(_bounded(part, config, notes, part_context).value) continue - if len(data) > config.max_attachment_bytes: + if len(data) > min(config.max_attachment_bytes, attachment_budget.remaining): complete = False - notes.add(f"{part_context}: image omitted because it exceeds max_attachment_bytes") + notes.add(f"{part_context}: image omitted because it exceeds the attachment size limit") result.append({"type": "text", "text": "[image omitted: size limit]"}) continue result.append( @@ -243,10 +267,12 @@ def _content( }, } ) + attachment_budget.consume(len(data)) + attachment_bytes += len(data) continue complete = False result.append(_bounded(part, config, notes, part_context).value) - return result, complete + return result, complete, attachment_bytes def _step_observations(step: dict[str, Any]) -> dict[str, Any]: @@ -319,6 +345,8 @@ def import_trajectory( phase_end: float, config: PluginConfig, _trajectory_data: dict[str, Any] | None = None, + _available_attachment_bytes: int | None = None, + _shared_attachment_budget: _AttachmentBudget | None = None, ) -> ATIFImportResult: notes = _Notes() if _trajectory_data is not None: @@ -333,6 +361,11 @@ def import_trajectory( if not isinstance(trajectory, dict) or not isinstance(trajectory.get("steps"), list): return ATIFImportResult(warnings=("trajectory malformed: steps must be an array",)) + attachment_budget = _resolve_attachment_budget( + config, + _available_attachment_bytes, + _shared_attachment_budget, + ) steps = [step for step in trajectory["steps"] if isinstance(step, dict)] times, repairs = _step_times(steps, phase_start, phase_end) agent = trajectory.get("agent") if isinstance(trajectory.get("agent"), dict) else {} @@ -354,11 +387,18 @@ def import_trajectory( final_message: Any = None llm_count = 0 tool_count = 0 + attachment_bytes = 0 for index, step in enumerate(steps): source = step.get("source") - content, content_complete = _content( - step.get("message"), trajectory_path.parent, config, notes, f"step {index + 1} message" + content, content_complete, content_bytes = _content( + step.get("message"), + trajectory_path.parent, + config, + notes, + f"step {index + 1} message", + attachment_budget, ) + attachment_bytes += content_bytes if source in {"system", "user"}: if config.content_mode != "metadata": messages.append({"role": source, "content": content}) @@ -456,9 +496,15 @@ def import_trajectory( and isinstance(result, dict) ): tool_context = f"step {index + 1} tool {call_id}" - tool_output, tool_complete = _content( - result.get("content"), trajectory_path.parent, config, notes, f"{tool_context} result" + tool_output, tool_complete, tool_bytes = _content( + result.get("content"), + trajectory_path.parent, + config, + notes, + f"{tool_context} result", + attachment_budget, ) + attachment_bytes += tool_bytes tool_input = _bounded(arguments, config, notes, f"{tool_context} arguments") result_extra = result.get("extra") if isinstance(result.get("extra"), dict) else {} tool_error = result_extra.get("error") if isinstance(result_extra.get("error"), str) else None @@ -509,6 +555,7 @@ def import_trajectory( phase_end=phase_end, config=config, _trajectory_data=subagent, + _shared_attachment_budget=attachment_budget, ) sub_parent.end(end_time=phase_end) # Step numbers restart inside a subagent, so namespace its warnings the way @@ -518,6 +565,7 @@ def import_trajectory( repairs.extend(f"subagent {sub_index}: {repair}" for repair in imported.repairs) llm_count += imported.imported_llm_spans tool_count += imported.imported_tool_spans + attachment_bytes += imported.attachment_bytes extra = trajectory.get("extra") if isinstance(trajectory.get("extra"), dict) else None root_extra = dict(extra or {}) @@ -531,4 +579,5 @@ def import_trajectory( repairs=tuple(repairs), imported_llm_spans=llm_count, imported_tool_spans=tool_count, + attachment_bytes=attachment_bytes, ) diff --git a/py/src/braintrust/integrations/harbor/compat.py b/py/src/braintrust/integrations/harbor/compat.py index 66e6546e..ff8f8ce7 100644 --- a/py/src/braintrust/integrations/harbor/compat.py +++ b/py/src/braintrust/integrations/harbor/compat.py @@ -194,6 +194,10 @@ def reward_details_paths(result: Any) -> list[tuple[str | None, Path]]: return _step_paths(result, "verifier", "reward-details.json") +def verifier_output_paths(result: Any) -> list[tuple[str | None, Path]]: + return _step_paths(result, "verifier") + + def artifact_manifest_paths(result: Any) -> list[tuple[str | None, Path]]: return _step_paths(result, "artifacts", "manifest.json") diff --git a/py/src/braintrust/integrations/harbor/plugin.py b/py/src/braintrust/integrations/harbor/plugin.py index 29758db8..d176dab6 100644 --- a/py/src/braintrust/integrations/harbor/plugin.py +++ b/py/src/braintrust/integrations/harbor/plugin.py @@ -9,7 +9,8 @@ import json import logging import os -from dataclasses import dataclass, field, fields +import stat +from dataclasses import dataclass, field, fields, replace from datetime import datetime from pathlib import Path from typing import Any @@ -26,6 +27,7 @@ reward_details_paths, snapshot_job, trajectory_paths, + verifier_output_paths, ) from .config import _UNSET, PluginConfig from .identity import ( @@ -158,6 +160,31 @@ def _step_label(step_name: str | None, path: Path) -> str: return path.name if step_name is None else f"{step_name}/{path.name}" +def _read_bounded_file(path: Path, max_bytes: int) -> tuple[bytes | None, str | None]: + """Read at most max_bytes from a file that may be controlled by a task.""" + if max_bytes < 0: + return None, "attachment size limit" + try: + before = path.lstat() + if not stat.S_ISREG(before.st_mode): + return None, "unsafe file type" + if before.st_size > max_bytes: + return None, "attachment size limit" + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) + with os.fdopen(os.open(path, flags), "rb") as file_obj: + opened = os.fstat(file_obj.fileno()) + if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): + return None, "unsafe file type" + data = file_obj.read(max_bytes + 1) + except FileNotFoundError: + return None, None + except OSError as exc: + return None, str(exc) + if opened.st_size > max_bytes or len(data) > max_bytes: + return None, "attachment size limit" + return data, None + + def _read_json_summary(entries: list[tuple[str | None, Path]], max_bytes: int) -> tuple[Any, list[str]]: summaries: list[tuple[str | None, Any]] = [] warnings: list[str] = [] @@ -202,14 +229,11 @@ def _artifact_attachments(result: Any, config: PluginConfig) -> tuple[dict[str, # Each step has its own artifacts root, so the relative path alone # collides whenever two steps collect the same file name. key = relative if step_name is None else f"{step_name}/{relative}" - try: - size = resolved.stat().st_size - if size > config.max_attachment_bytes or total + size > config.max_total_attachment_bytes: - warnings.append(f"artifact {key} omitted: attachment size limit") - continue - data = resolved.read_bytes() - except OSError as exc: - warnings.append(f"artifact {key} omitted: {exc}") + limit = min(config.max_attachment_bytes, config.max_total_attachment_bytes - total) + data, read_warning = _read_bounded_file(resolved, limit) + if data is None: + if read_warning is not None: + warnings.append(f"artifact {key} omitted: {read_warning}") continue total += len(data) attachments[key] = Attachment( @@ -232,15 +256,11 @@ def _attachment( for step_name, path in entries: label = _step_label(step_name, path) filename = path.name - try: - data = path.read_bytes() - except FileNotFoundError: - continue - except OSError as exc: - warnings.append(f"could not read {label}: {exc}") - continue - if len(data) > config.max_attachment_bytes or total + len(data) > config.max_total_attachment_bytes: - warnings.append(f"{label} omitted: attachment size limit") + limit = min(config.max_attachment_bytes, config.max_total_attachment_bytes - total) + data, read_warning = _read_bounded_file(path, limit) + if data is None: + if read_warning is not None: + warnings.append(f"{label} omitted: {read_warning}") continue try: parsed = json.loads(data) @@ -260,8 +280,8 @@ def _attachment( if summary is None: return None, None, warnings attachment_data = (canonical_json(summary) + "\n").encode() - # One serialized payload is bounded by the per-file limit, not the job total. - if len(attachment_data) > config.max_attachment_bytes: + # The merged payload must fit both the per-file and remaining trial limits. + if len(attachment_data) > min(config.max_attachment_bytes, config.max_total_attachment_bytes): warnings.append(f"{filename} omitted after redaction: attachment size limit") return None, summary, warnings return ( @@ -271,6 +291,106 @@ def _attachment( ) +_VERIFIER_OUTPUT_FILES = ( + ("stdout", "test-stdout.txt", False), + ("stderr", "test-stderr.txt", False), + ("ctrf", "ctrf.json", True), +) + + +def _decode_verifier_output(data: bytes, parse_json: bool) -> tuple[Any, list[str]]: + if not parse_json: + return data.decode("utf-8", errors="replace"), [] + try: + return json.loads(data), [] + except (UnicodeDecodeError, json.JSONDecodeError): + return data.decode("utf-8", errors="replace"), ["is not valid JSON"] + + +def _read_verifier_output( + path: Path, parse_json: bool, config: PluginConfig, max_bytes: int +) -> tuple[Any | None, int, list[str]]: + data, read_warning = _read_bounded_file(path, max_bytes) + if data is None: + return None, 0, [] if read_warning is None else [f"omitted: {read_warning}"] + if not data: + return None, 0, [] + value, warnings = _decode_verifier_output(data, parse_json) + normalized = normalize_json( + value, + max_bytes=config.max_attachment_bytes, + redact_patterns=config.redact_patterns, + max_depth=20, + redact_absolute_paths=False, + ) + return normalized.value, len(data), [*warnings, *normalized.warnings] + + +def _verifier_output_attachment(result: Any, config: PluginConfig) -> tuple[Attachment | None, Any, list[str]]: + if config.attachments == "none": + return None, None, [] + outputs: list[tuple[str | None, dict[str, Any]]] = [] + warnings: list[str] = [] + total = 0 + for step_name, verifier_dir in verifier_output_paths(result): + step_output: dict[str, Any] = {} + for key, filename, parse_json in _VERIFIER_OUTPUT_FILES: + path = verifier_dir / filename + label = _step_label(step_name, path) + limit = min(config.max_attachment_bytes, config.max_total_attachment_bytes - total) + value, size, file_warnings = _read_verifier_output(path, parse_json, config, limit) + warnings.extend(f"{label} {warning}" for warning in file_warnings) + if value is not None: + step_output[key] = value + total += size + if step_output: + outputs.append((step_name, step_output)) + + summary = _by_step(outputs, "verifier") + if summary is None: + return None, None, warnings + attachment_data = (canonical_json(summary) + "\n").encode() + if len(attachment_data) > min(config.max_attachment_bytes, config.max_total_attachment_bytes): + warnings.append("verifier-output.json omitted after redaction: attachment size limit") + return None, summary, warnings + return ( + Attachment(data=attachment_data, filename="verifier-output.json", content_type="application/json"), + summary, + warnings, + ) + + +def _remaining_attachment_config(config: PluginConfig, used_bytes: int) -> PluginConfig: + return replace(config, max_total_attachment_bytes=_remaining_attachment_bytes(config, used_bytes)) + + +def _remaining_attachment_bytes(config: PluginConfig, used_bytes: int) -> int: + return max(0, config.max_total_attachment_bytes - used_bytes) + + +def _attachments_size(attachments: dict[str, Attachment]) -> int: + return sum(len(attachment.data) for attachment in attachments.values()) + + +def _verifier_evidence(result: Any, config: PluginConfig) -> tuple[dict[str, Any], list[str], int]: + attachment, summary, warnings = _verifier_output_attachment(result, config) + output: dict[str, Any] = {} + if summary is not None: + output["verifier_output_summary"] = normalize_json( + summary, + max_bytes=config.max_content_bytes, + redact_patterns=config.redact_patterns, + redact_absolute_paths=False, + ).value + if attachment is not None: + output["verifier_output"] = attachment + return output, warnings, 0 if attachment is None else len(attachment.data) + + +def _output_event(output: dict[str, Any]) -> dict[str, Any]: + return {} if not output else {"output": output} + + class HarborPlugin: """Harbor plugin that reconciles final trials into Braintrust experiments.""" @@ -714,6 +834,7 @@ def _sync_final_result(self, result: Any) -> None: execution_input["extra_instructions"] = extra_instructions selected_artifacts, artifact_attachment_warnings = _artifact_attachments(result, self.config) metadata["harbor"]["warnings"].extend(artifact_attachment_warnings) + attachment_bytes = _attachments_size(selected_artifacts) agent_span = task.start_span( name="agent_execution", type="task", @@ -741,12 +862,27 @@ def _sync_final_result(self, result: Any) -> None: phase_start=agent_start, phase_end=agent_end, config=self.config, + _available_attachment_bytes=_remaining_attachment_bytes(self.config, attachment_bytes), ) atif_results.append((step_name, imported)) + attachment_bytes += imported.attachment_bytes if selected_artifacts: agent_span.log(output={"artifacts": selected_artifacts}) agent_span.end(end_time=agent_end) - self._start_phase(task, result, "verification", "verifier", trial_id, root_start, root_end) + verifier_config = _remaining_attachment_config(self.config, attachment_bytes) + verifier_output, verifier_warnings, verifier_attachment_bytes = _verifier_evidence(result, verifier_config) + metadata["harbor"]["warnings"].extend(verifier_warnings) + attachment_bytes += verifier_attachment_bytes + self._start_phase( + task, + result, + "verification", + "verifier", + trial_id, + root_start, + root_end, + **_output_event(verifier_output), + ) for step in getattr(result, "step_results", None) or []: step_start, step_end = _timing(getattr(step, "agent_execution", None), root_start, root_end) @@ -813,7 +949,10 @@ def _sync_final_result(self, result: Any) -> None: task.log(metadata={"harbor": {"warnings": trajectory_warnings}}) task.end(end_time=root_end) - details_attachment, details_summary, detail_warnings = _attachment(reward_details_paths(result), self.config) + details_config = _remaining_attachment_config(self.config, attachment_bytes) + details_attachment, details_summary, detail_warnings = _attachment( + reward_details_paths(result), details_config + ) metadata["harbor"]["warnings"].extend(detail_warnings) # The summary is the same for every score, so bound it once rather than # re-normalizing a payload up to max_attachment_bytes per scorer span. @@ -842,6 +981,7 @@ def _sync_final_result(self, result: Any) -> None: scorer_output["reward_details_summary"] = bounded_details if details_attachment is not None: scorer_output["reward_details"] = details_attachment + scorer_output.update(verifier_output) scorer.log(output=scorer_output, scores={score.name: score.value}) scorer.end(end_time=root_end) diff --git a/py/src/braintrust/integrations/harbor/test_harbor.py b/py/src/braintrust/integrations/harbor/test_harbor.py index 848ae0b4..51a86019 100644 --- a/py/src/braintrust/integrations/harbor/test_harbor.py +++ b/py/src/braintrust/integrations/harbor/test_harbor.py @@ -15,7 +15,13 @@ from braintrust.conftest import get_vcr_config from braintrust.git_fields import GitMetadataSettings from braintrust.integrations.harbor.atif import _usage_metrics, import_trajectory, summarize_trajectory -from braintrust.integrations.harbor.compat import artifact_manifest_paths, load_backfill_snapshot +from braintrust.integrations.harbor.compat import ( + JobSnapshot, + TaskData, + TrialPlan, + artifact_manifest_paths, + load_backfill_snapshot, +) from braintrust.integrations.harbor.config import PluginConfig from braintrust.integrations.harbor.identity import ( child_span_id, @@ -27,13 +33,17 @@ semantic_agent_config, ) from braintrust.integrations.harbor.plugin import ( + DatasetBinding, HarborPlugin, + Partition, RuntimeState, _artifact_attachments, _attachment, + _read_bounded_file, _resolve_project, _seconds, _timing, + _verifier_output_attachment, ) from braintrust.integrations.harbor.rewards import classify_rewards, validate_classifications from braintrust.integrations.harbor.state import ( @@ -53,7 +63,7 @@ from harbor.models.task.id import LocalTaskId from harbor.models.trajectories.trajectory import Trajectory from harbor.models.trial.config import AgentConfig, EnvironmentConfig, TaskConfig, TrialConfig -from harbor.models.trial.result import AgentInfo, StepResult, TimingInfo, TrialResult +from harbor.models.trial.result import AgentInfo, StepResult, TimingInfo, TrialResult, VerifierResult _ABSOLUTE_PATH_RE = re.compile(r"(?:/(?:Users|private|home)/[^\"\\\\\s]+|[A-Za-z]:\\\\[^\"\\\\\s]+)") @@ -466,6 +476,293 @@ def test_reward_details_attachment_uses_the_per_file_limit(tmp_path): assert warnings == [] +def test_verifier_output_attachment_collects_standard_harbor_evidence(tmp_path): + result = _trial_result(tmp_path, "trial-1", "task-a") + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "test-stdout.txt").write_text("FAILED test_answer.py::test_count - assert 27 == 28\n") + (verifier_dir / "test-stderr.txt").write_text("token=secret-value\n") + (verifier_dir / "ctrf.json").write_text( + json.dumps( + { + "results": { + "summary": {"tests": 1, "passed": 0, "failed": 1}, + "tests": [ + { + "name": "test_answer.py::test_count", + "status": "failed", + "message": "assert 27 == 28", + "trace": "Authorization: Bearer verifier-secret", + } + ], + } + } + ) + ) + + config = PluginConfig.from_options(redact_patterns=(r"secret-value|Bearer verifier-secret",)) + attachment, summary, warnings = _verifier_output_attachment(result, config) + + assert summary == { + "stdout": "FAILED test_answer.py::test_count - assert 27 == 28\n", + "stderr": "token=[REDACTED]\n", + "ctrf": { + "results": { + "summary": {"tests": 1, "passed": 0, "failed": 1}, + "tests": [ + { + "name": "test_answer.py::test_count", + "status": "failed", + "message": "assert 27 == 28", + "trace": "Authorization: [REDACTED]", + } + ], + } + }, + } + assert attachment is not None + assert attachment.reference["filename"] == "verifier-output.json" + assert json.loads(attachment.data) == summary + assert warnings == [] + + +def test_verifier_output_attachment_handles_invalid_utf8_and_configured_redaction(tmp_path): + result = _trial_result(tmp_path, "trial-1", "task-a") + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "test-stdout.txt").write_text("Authorization: Bearer header-secret\n") + (verifier_dir / "ctrf.json").write_bytes(b"\xff token=opaque-secret\n") + + attachment, summary, warnings = _verifier_output_attachment( + result, + PluginConfig.from_options(redact_patterns=(r"(?:header|opaque)-secret",)), + ) + + assert attachment is not None + assert summary == { + "stdout": "Authorization: Bearer [REDACTED]\n", + "ctrf": "\ufffd token=[REDACTED]\n", + } + assert any("ctrf.json is not valid JSON" in warning for warning in warnings) + + +def test_verifier_output_attachment_rejects_oversized_file_before_reading_it(tmp_path, monkeypatch): + result = _trial_result(tmp_path, "trial-1", "task-a") + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + stdout = verifier_dir / "test-stdout.txt" + stdout.write_text("too large") + + def fail_open(*_args, **_kwargs): + raise AssertionError("oversized output must be rejected from stat metadata") + + monkeypatch.setattr(os, "open", fail_open) + attachment, summary, warnings = _verifier_output_attachment( + result, + PluginConfig.from_options(max_attachment_bytes=4), + ) + + assert attachment is None + assert summary is None + assert warnings == ["test-stdout.txt omitted: attachment size limit"] + + +@pytest.mark.parametrize("kind", ["symlink", "fifo"]) +def test_verifier_output_attachment_rejects_unsafe_file_types(tmp_path, kind): + result = _trial_result(tmp_path, "trial-1", "task-a") + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + stdout = verifier_dir / "test-stdout.txt" + if kind == "symlink": + secret = tmp_path / "host-secret" + secret.write_text("must not escape") + stdout.symlink_to(secret) + else: + os.mkfifo(stdout) + + attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options()) + + assert attachment is None + assert summary is None + assert warnings == ["test-stdout.txt omitted: unsafe file type"] + + +def test_bounded_file_read_rejects_replacement_between_inspection_and_open(tmp_path, monkeypatch): + expected = tmp_path / "expected" + replacement = tmp_path / "replacement" + expected.write_text("safe") + replacement.write_text("must not escape") + real_open = os.open + + def swap_after_inspection(_path, flags): + return real_open(replacement, flags) + + monkeypatch.setattr(os, "open", swap_after_inspection) + + assert _read_bounded_file(expected, 100) == (None, "unsafe file type") + + +def test_verifier_output_attachment_scopes_steps_and_respects_attachment_mode(tmp_path): + result = _trial_result(tmp_path, "trial-1", "task-a", step_names=("first", "second")) + for step_name in ("first", "second"): + verifier_dir = tmp_path / "trial-1" / "steps" / step_name / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "test-stdout.txt").write_text(f"{step_name} output\n") + + attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options()) + + assert summary == { + "first": {"stdout": "first output\n"}, + "second": {"stdout": "second output\n"}, + } + assert attachment is not None + assert warnings == [] + assert _verifier_output_attachment(result, PluginConfig.from_options(attachments="none")) == (None, None, []) + + +@pytest.mark.parametrize("attachments", ["verifier-details", "none"]) +def test_final_sync_wires_verifier_evidence_to_verification_and_score_spans(tmp_path, attachments): + class RecordingSpan: + def __init__(self, **event): + self.event = event + self.children = [] + self.logs = [] + + def start_span(self, **event): + child = RecordingSpan(**event) + self.children.append(child) + return child + + def log(self, **event): + self.logs.append(event) + + def end(self, **_event): + return None + + class RecordingExperiment: + def __init__(self): + self.children = [] + + def start_span(self, **event): + span = RecordingSpan(**event) + self.children.append(span) + return span + + result = _trial_result(tmp_path, "trial-1", "task-a") + result.verifier_result = VerifierResult(rewards={"reward": 0.25}) + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "test-stdout.txt").write_text("assert 1 == 2\n") + task = TaskData( + logical_key="task-key", + source="suite", + name="task-a", + input={"instruction": "solve"}, + expected=None, + metadata={"harbor": {"custom": {}}}, + digest=None, + schema_version=None, + task_dir=None, + ) + plan = TrialPlan(result.trial_name, result.config, None, task, 0) + snapshot = JobSnapshot("job-id", "job", tmp_path, None, None, (plan,)) + experiment = RecordingExperiment() + partition = Partition("partition", "experiment", "scope", experiment=experiment) + plugin = HarborPlugin(attachments=attachments) + plugin._runtime = RuntimeState( + snapshot, + {result.trial_name: plan}, + {result.trial_name: partition}, + {"scope": DatasetBinding("scope")}, + {"partition": partition}, + ) + plugin._trial_machines[result.trial_name] = TrialMachine(result.trial_name) + + plugin._sync_final_result(result) + + root = experiment.children[0] + task_span = next(span for span in root.children if span.event["name"] == "task") + verification = next(span for span in task_span.children if span.event["name"] == "verification") + scorer = next(span for span in root.children if span.event["type"] == "score") + if attachments == "none": + assert "output" not in verification.event + assert "verifier_output_summary" not in scorer.logs[0]["output"] + assert "verifier_output" not in scorer.logs[0]["output"] + else: + assert verification.event["output"]["verifier_output_summary"] == {"stdout": "assert 1 == 2\n"} + verifier_attachment = verification.event["output"]["verifier_output"] + assert verifier_attachment.reference["filename"] == "verifier-output.json" + assert ( + scorer.logs[0]["output"]["verifier_output_summary"] + == verification.event["output"]["verifier_output_summary"] + ) + assert scorer.logs[0]["output"]["verifier_output"] is verifier_attachment + + # The verification span owns the attachment, so unevaluated trials do not + # lose their complete evidence merely because no score span is created. + experiment.children.clear() + result.verifier_result = VerifierResult(rewards=None) + plugin._sync_final_result(result) + scoreless_root = experiment.children[0] + scoreless_task = next(span for span in scoreless_root.children if span.event["name"] == "task") + scoreless_verification = next(span for span in scoreless_task.children if span.event["name"] == "verification") + assert not any(span.event["type"] == "score" for span in scoreless_root.children) + assert scoreless_verification.event["output"]["verifier_output"] is not None + + # ATIF images consume the same trial attachment budget as verifier + # evidence, even though the image is logged on an agent child span. + agent_dir = tmp_path / "trial-1" / "agent" + agent_dir.mkdir(exist_ok=True) + (agent_dir / "first.png").write_bytes(b"i" * 600) + (agent_dir / "second.png").write_bytes(b"j" * 600) + (agent_dir / "trajectory.json").write_text( + json.dumps( + { + "steps": [ + { + "step_id": 1, + "source": "agent", + "message": [ + { + "type": "image", + "source": {"path": "first.png", "media_type": "image/png"}, + }, + { + "type": "image", + "source": {"path": "second.png", "media_type": "image/png"}, + }, + ], + } + ] + } + ) + ) + (verifier_dir / "test-stdout.txt").write_text("v" * 100) + experiment.children.clear() + result.verifier_result = VerifierResult(rewards={"reward": 0.25}) + budgeted_plugin = HarborPlugin(max_attachment_bytes=1000, max_total_attachment_bytes=1000) + budgeted_plugin._runtime = RuntimeState( + snapshot, + {result.trial_name: plan}, + {result.trial_name: partition}, + {"scope": DatasetBinding("scope")}, + {"partition": partition}, + ) + budgeted_plugin._trial_machines[result.trial_name] = TrialMachine(result.trial_name) + + budgeted_plugin._sync_final_result(result) + + budgeted_root = experiment.children[0] + budgeted_task = next(span for span in budgeted_root.children if span.event["name"] == "task") + budgeted_agent = next(span for span in budgeted_task.children if span.event["name"] == "agent_execution") + trajectory_step = budgeted_agent.children[0] + trajectory_message = trajectory_step.logs[0]["output"]["message"] + assert trajectory_message[0]["type"] == "image_url" + assert trajectory_message[1] == {"type": "text", "text": "[image omitted: size limit]"} + budgeted_verification = next(span for span in budgeted_task.children if span.event["name"] == "verification") + assert budgeted_verification.event["output"]["verifier_output_summary"] == {"stdout": "v" * 100} + + def test_disabled_plugin_does_not_reconcile_or_write_spans(): plugin = HarborPlugin(project_name="unused") # Reproduce the ordering that makes this reachable: the runtime is built, then From 5ec775a6637ee167201484d5f4f38f2cd8a0a3ce Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Mon, 31 Aug 2026 12:31:19 -0400 Subject: [PATCH 2/5] fix(harbor): remove attachment size caps and gate raw verifier logs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `max_total_attachment_bytes` was one shared pot consumed in source order, so artifacts and trajectory images could starve verifier output and reward-details. Same trial, one bigger screenshot, different evidence. Attachments are meant to be large and `Attachment` uploads lazily from a path, so both caps and the budget plumbing are gone; `_read_bounded_file` becomes `_read_safe_file`, keeping the symlink, file-type and dev/ino checks. `trajectory.json` is parsed in-process rather than uploaded, so it keeps a bound of its own via `max_trajectory_bytes`. Two fixes from the same review: - Raw `test-stdout.txt`/`test-stderr.txt` now require `attachments=all`. `normalize_json` redacts by key name, which raw text lacks, so `redact_patterns` — empty by default — was the only protection on a payload the default config uploaded. `ctrf.json` stays in `verifier-details`. - `json.loads` raises `RecursionError`, not `JSONDecodeError`, on deeply nested input. A 400 KB `ctrf.json` aborted `_sync_final_result` after the spans started but before output, scores and metadata, leaving a half-written row. Both decoders now fall back to text. Tests use the real span harness (`init_test_exp` + `with_memory_logger`) instead of `RecordingSpan` fakes, so they assert the stored `AttachmentReference` and the queued upload. BREAKING CHANGE: `max_attachment_bytes` and `max_total_attachment_bytes` are removed, and unknown options are silently ignored, so a config setting them now uploads with no ceiling. `verifier-details` no longer ships raw verifier stdout/stderr; use `attachments=all` with `redact_patterns`. --- examples/harbor/README.md | 4 +- py/src/braintrust/integrations/harbor/atif.py | 64 +--- .../braintrust/integrations/harbor/config.py | 8 +- .../integrations/harbor/identity.py | 7 +- .../braintrust/integrations/harbor/plugin.py | 123 +++---- .../integrations/harbor/test_harbor.py | 343 ++++++++++-------- 6 files changed, 267 insertions(+), 282 deletions(-) diff --git a/examples/harbor/README.md b/examples/harbor/README.md index 0189a7d0..cbfeddc7 100644 --- a/examples/harbor/README.md +++ b/examples/harbor/README.md @@ -39,7 +39,9 @@ uv run harbor run \ The agent solves the task in `task/`, and Harbor's verifier emits a normalized `reward` plus an `answer_length` metric. The plugin creates `jobs/braintrust-harbor-example/braintrust-sync.json` after synchronization. -With the default `attachments=verifier-details` mode, the verification span and each score also include Harbor's captured `test-stdout.txt`, optional `test-stderr.txt`, and conventional `ctrf.json` output when present. The size-bounded summary and complete redacted `verifier-output.json` attachment appear together in the span output. Structured CTRF fields with sensitive key names use the plugin's standard redaction, and configured `redact_patterns` apply to both structured CTRF strings and raw verifier text. +With the default `attachments=verifier-details` mode, the verification span and each score also include Harbor's conventional `ctrf.json` output when present. The size-bounded summary and complete redacted `verifier-output.json` attachment appear together in the span output. Structured CTRF fields with sensitive key names use the plugin's standard redaction. + +Harbor's captured `test-stdout.txt` and `test-stderr.txt` require `attachments=all`. Redaction works from key names, which raw text does not have, so **configured `redact_patterns` are the only redaction applied to raw verifier logs** — and `redact_patterns` is empty by default. Verifier logs are a common place for environment dumps, tokens in URLs, and other credentials, so set `redact_patterns` before enabling this, or leave the logs out and rely on `ctrf.json` for structured failure detail. By default, the plugin uses `Harbor` as the Braintrust project name. Override it with `--plugin-kwarg project_name=example-harbor` or through `.env`: diff --git a/py/src/braintrust/integrations/harbor/atif.py b/py/src/braintrust/integrations/harbor/atif.py index 2520d923..cb5dc5d5 100644 --- a/py/src/braintrust/integrations/harbor/atif.py +++ b/py/src/braintrust/integrations/harbor/atif.py @@ -61,28 +61,6 @@ class ATIFImportResult: repairs: tuple[str, ...] = () imported_llm_spans: int = 0 imported_tool_spans: int = 0 - attachment_bytes: int = 0 - - -@dataclass -class _AttachmentBudget: - remaining: int - used: int = 0 - - def consume(self, size: int) -> None: - self.remaining -= size - self.used += size - - -def _resolve_attachment_budget( - config: PluginConfig, - available_bytes: int | None, - shared_budget: _AttachmentBudget | None, -) -> _AttachmentBudget: - if shared_budget is not None: - return shared_budget - limit = config.max_total_attachment_bytes if available_bytes is None else available_bytes - return _AttachmentBudget(max(0, limit)) def _timestamp(value: Any) -> tuple[float | None, bool]: @@ -215,16 +193,14 @@ def _content( config: PluginConfig, notes: _Notes, context: str, - attachment_budget: _AttachmentBudget, -) -> tuple[Any, bool, int]: +) -> tuple[Any, bool]: if isinstance(value, str) or value is None: bounded = _bounded(value, config, notes, context) - return bounded.value, bounded.complete, 0 + return bounded.value, bounded.complete if not isinstance(value, list): - return _bounded(value, config, notes, context).value, False, 0 + return _bounded(value, config, notes, context).value, False result: list[Any] = [] complete = True - attachment_bytes = 0 trajectory_root = trajectory_dir.resolve() for index, part in enumerate(value): part_context = f"{context}[{index}]" @@ -250,11 +226,6 @@ def _content( complete = False result.append(_bounded(part, config, notes, part_context).value) continue - if len(data) > min(config.max_attachment_bytes, attachment_budget.remaining): - complete = False - notes.add(f"{part_context}: image omitted because it exceeds the attachment size limit") - result.append({"type": "text", "text": "[image omitted: size limit]"}) - continue result.append( { "type": "image_url", @@ -267,12 +238,10 @@ def _content( }, } ) - attachment_budget.consume(len(data)) - attachment_bytes += len(data) continue complete = False result.append(_bounded(part, config, notes, part_context).value) - return result, complete, attachment_bytes + return result, complete def _step_observations(step: dict[str, Any]) -> dict[str, Any]: @@ -302,7 +271,7 @@ def _end_time(times: list[float], index: int, phase_end: float) -> float: def summarize_trajectory(trajectory_path: Path, config: PluginConfig) -> ATIFImportResult: """Read bounded trajectory summary data without creating detailed leaves.""" try: - if trajectory_path.stat().st_size > config.max_total_attachment_bytes: + if trajectory_path.stat().st_size > config.max_trajectory_bytes: return ATIFImportResult(warnings=("trajectory omitted: size limit",)) trajectory = json.loads(trajectory_path.read_text()) except (OSError, json.JSONDecodeError) as exc: @@ -345,15 +314,15 @@ def import_trajectory( phase_end: float, config: PluginConfig, _trajectory_data: dict[str, Any] | None = None, - _available_attachment_bytes: int | None = None, - _shared_attachment_budget: _AttachmentBudget | None = None, ) -> ATIFImportResult: notes = _Notes() if _trajectory_data is not None: trajectory = _trajectory_data else: try: - if trajectory_path.stat().st_size > config.max_total_attachment_bytes: + # Unlike an attachment, this document is parsed into the host process + # rather than handed to object storage, so its size is bounded. + if trajectory_path.stat().st_size > config.max_trajectory_bytes: return ATIFImportResult(warnings=("trajectory omitted: size limit",)) trajectory = json.loads(trajectory_path.read_text()) except (OSError, json.JSONDecodeError) as exc: @@ -361,11 +330,6 @@ def import_trajectory( if not isinstance(trajectory, dict) or not isinstance(trajectory.get("steps"), list): return ATIFImportResult(warnings=("trajectory malformed: steps must be an array",)) - attachment_budget = _resolve_attachment_budget( - config, - _available_attachment_bytes, - _shared_attachment_budget, - ) steps = [step for step in trajectory["steps"] if isinstance(step, dict)] times, repairs = _step_times(steps, phase_start, phase_end) agent = trajectory.get("agent") if isinstance(trajectory.get("agent"), dict) else {} @@ -387,18 +351,15 @@ def import_trajectory( final_message: Any = None llm_count = 0 tool_count = 0 - attachment_bytes = 0 for index, step in enumerate(steps): source = step.get("source") - content, content_complete, content_bytes = _content( + content, content_complete = _content( step.get("message"), trajectory_path.parent, config, notes, f"step {index + 1} message", - attachment_budget, ) - attachment_bytes += content_bytes if source in {"system", "user"}: if config.content_mode != "metadata": messages.append({"role": source, "content": content}) @@ -496,15 +457,13 @@ def import_trajectory( and isinstance(result, dict) ): tool_context = f"step {index + 1} tool {call_id}" - tool_output, tool_complete, tool_bytes = _content( + tool_output, tool_complete = _content( result.get("content"), trajectory_path.parent, config, notes, f"{tool_context} result", - attachment_budget, ) - attachment_bytes += tool_bytes tool_input = _bounded(arguments, config, notes, f"{tool_context} arguments") result_extra = result.get("extra") if isinstance(result.get("extra"), dict) else {} tool_error = result_extra.get("error") if isinstance(result_extra.get("error"), str) else None @@ -555,7 +514,6 @@ def import_trajectory( phase_end=phase_end, config=config, _trajectory_data=subagent, - _shared_attachment_budget=attachment_budget, ) sub_parent.end(end_time=phase_end) # Step numbers restart inside a subagent, so namespace its warnings the way @@ -565,7 +523,6 @@ def import_trajectory( repairs.extend(f"subagent {sub_index}: {repair}" for repair in imported.repairs) llm_count += imported.imported_llm_spans tool_count += imported.imported_tool_spans - attachment_bytes += imported.attachment_bytes extra = trajectory.get("extra") if isinstance(trajectory.get("extra"), dict) else None root_extra = dict(extra or {}) @@ -579,5 +536,4 @@ def import_trajectory( repairs=tuple(repairs), imported_llm_spans=llm_count, imported_tool_spans=tool_count, - attachment_bytes=attachment_bytes, ) diff --git a/py/src/braintrust/integrations/harbor/config.py b/py/src/braintrust/integrations/harbor/config.py index 97254444..d5a2757e 100644 --- a/py/src/braintrust/integrations/harbor/config.py +++ b/py/src/braintrust/integrations/harbor/config.py @@ -100,9 +100,8 @@ class PluginConfig: include_tracebacks: bool = False attachments: str = "verifier-details" artifact_include: tuple[str, ...] = () - max_attachment_bytes: int = 5_000_000 - max_total_attachment_bytes: int = 20_000_000 max_content_bytes: int = 20_000 + max_trajectory_bytes: int = 20_000_000 log_retry_attempts: bool = False strict: bool = False redact_patterns: tuple[str, ...] = () @@ -124,9 +123,8 @@ def from_options(cls, **options: Any) -> "PluginConfig": values[name] = _parse_bool(values[name], name) for name in ( "max_custom_metadata_bytes", - "max_attachment_bytes", - "max_total_attachment_bytes", "max_content_bytes", + "max_trajectory_bytes", ): values[name] = _parse_int(values[name], name) for name in ("score_keys", "metric_keys", "artifact_include", "redact_patterns"): @@ -161,8 +159,6 @@ def validate(self) -> None: raise ValueError("attachments must be 'none', 'verifier-details', or 'all'") if self.artifact_include and self.attachments != "all": raise ValueError("artifact_include requires attachments='all'") - if self.max_total_attachment_bytes < self.max_attachment_bytes: - raise ValueError("max_total_attachment_bytes must be at least max_attachment_bytes") for score_pattern in self.score_keys: for metric_pattern in self.metric_keys: diff --git a/py/src/braintrust/integrations/harbor/identity.py b/py/src/braintrust/integrations/harbor/identity.py index 5eabe6b1..c47ce28c 100644 --- a/py/src/braintrust/integrations/harbor/identity.py +++ b/py/src/braintrust/integrations/harbor/identity.py @@ -118,7 +118,7 @@ def _json_size(value: Any) -> int: def normalize_json( value: Any, *, - max_bytes: int, + max_bytes: int | None, redact_patterns: tuple[str, ...] = (), max_depth: int = 8, redact_absolute_paths: bool = True, @@ -128,6 +128,9 @@ def normalize_json( Set ``redact_absolute_paths=False`` for payloads produced inside the task sandbox: their absolute paths are container paths the agent actually operated on, so redacting them erases the substance of filesystem tool calls. + + ``max_bytes=None`` redacts without truncating, for payloads bound for an + attachment rather than a span field. """ warnings: list[str] = [] compiled_patterns = tuple(re.compile(pattern) for pattern in redact_patterns) @@ -176,7 +179,7 @@ def walk(item: Any, path: str, depth: int, key: str | None = None) -> Any: return f"[DROPPED: {type(item).__name__}]" normalized = walk(value, "", 0) - if _json_size(normalized) <= max_bytes: + if max_bytes is None or _json_size(normalized) <= max_bytes: return NormalizedValue(normalized, tuple(warnings)) # Fitting a container to a byte budget needs each entry's serialized size, not diff --git a/py/src/braintrust/integrations/harbor/plugin.py b/py/src/braintrust/integrations/harbor/plugin.py index d176dab6..34a0d7eb 100644 --- a/py/src/braintrust/integrations/harbor/plugin.py +++ b/py/src/braintrust/integrations/harbor/plugin.py @@ -10,7 +10,7 @@ import logging import os import stat -from dataclasses import dataclass, field, fields, replace +from dataclasses import dataclass, field, fields from datetime import datetime from pathlib import Path from typing import Any @@ -160,28 +160,26 @@ def _step_label(step_name: str | None, path: Path) -> str: return path.name if step_name is None else f"{step_name}/{path.name}" -def _read_bounded_file(path: Path, max_bytes: int) -> tuple[bytes | None, str | None]: - """Read at most max_bytes from a file that may be controlled by a task.""" - if max_bytes < 0: - return None, "attachment size limit" +def _read_safe_file(path: Path) -> tuple[bytes | None, str | None]: + """Read a file that may be controlled by a task, refusing anything but a regular file. + + Attachments are the escape hatch for large payloads, so size is deliberately + unbounded here; only the file's type and identity are checked. + """ try: before = path.lstat() if not stat.S_ISREG(before.st_mode): return None, "unsafe file type" - if before.st_size > max_bytes: - return None, "attachment size limit" flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0) with os.fdopen(os.open(path, flags), "rb") as file_obj: opened = os.fstat(file_obj.fileno()) if not stat.S_ISREG(opened.st_mode) or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino): return None, "unsafe file type" - data = file_obj.read(max_bytes + 1) + data = file_obj.read() except FileNotFoundError: return None, None except OSError as exc: return None, str(exc) - if opened.st_size > max_bytes or len(data) > max_bytes: - return None, "attachment size limit" return data, None @@ -210,7 +208,6 @@ def _artifact_attachments(result: Any, config: PluginConfig) -> tuple[dict[str, return {}, [] attachments: dict[str, Attachment] = {} warnings: list[str] = [] - total = 0 for step_name, manifest_path in artifact_manifest_paths(result): root = manifest_path.parent.resolve() if not root.exists(): @@ -229,13 +226,11 @@ def _artifact_attachments(result: Any, config: PluginConfig) -> tuple[dict[str, # Each step has its own artifacts root, so the relative path alone # collides whenever two steps collect the same file name. key = relative if step_name is None else f"{step_name}/{relative}" - limit = min(config.max_attachment_bytes, config.max_total_attachment_bytes - total) - data, read_warning = _read_bounded_file(resolved, limit) + data, read_warning = _read_safe_file(resolved) if data is None: if read_warning is not None: warnings.append(f"artifact {key} omitted: {read_warning}") continue - total += len(data) attachments[key] = Attachment( data=data, filename=resolved.name, @@ -249,41 +244,34 @@ def _attachment( ) -> tuple[Attachment | None, Any, list[str]]: if config.attachments == "none": return None, None, [] - total = 0 complete: list[tuple[str | None, Any]] = [] warnings: list[str] = [] filename = "details.json" for step_name, path in entries: label = _step_label(step_name, path) filename = path.name - limit = min(config.max_attachment_bytes, config.max_total_attachment_bytes - total) - data, read_warning = _read_bounded_file(path, limit) + data, read_warning = _read_safe_file(path) if data is None: if read_warning is not None: warnings.append(f"{label} omitted: {read_warning}") continue try: parsed = json.loads(data) - except json.JSONDecodeError: + except (json.JSONDecodeError, RecursionError): warnings.append(f"{label} is not valid JSON") continue normalized = normalize_json( parsed, - max_bytes=config.max_attachment_bytes, + max_bytes=None, redact_patterns=config.redact_patterns, max_depth=20, ) warnings.extend(normalized.warnings) complete.append((step_name, normalized.value)) - total += len(data) summary = _by_step(complete, "details") if summary is None: return None, None, warnings attachment_data = (canonical_json(summary) + "\n").encode() - # The merged payload must fit both the per-file and remaining trial limits. - if len(attachment_data) > min(config.max_attachment_bytes, config.max_total_attachment_bytes): - warnings.append(f"{filename} omitted after redaction: attachment size limit") - return None, summary, warnings return ( Attachment(data=attachment_data, filename=filename, content_type="application/json"), summary, @@ -291,10 +279,22 @@ def _attachment( ) +@dataclass(frozen=True) +class _VerifierOutputFile: + key: str + filename: str + parse_json: bool + # normalize_json redacts by key name, which a structured document supplies and + # raw text does not: raw verifier output is only covered by configured + # redact_patterns. Treat it like any other arbitrary byte stream the task + # wrote and require the same explicit opt-in that artifact_include does. + requires_all: bool + + _VERIFIER_OUTPUT_FILES = ( - ("stdout", "test-stdout.txt", False), - ("stderr", "test-stderr.txt", False), - ("ctrf", "ctrf.json", True), + _VerifierOutputFile("stdout", "test-stdout.txt", parse_json=False, requires_all=True), + _VerifierOutputFile("stderr", "test-stderr.txt", parse_json=False, requires_all=True), + _VerifierOutputFile("ctrf", "ctrf.json", parse_json=True, requires_all=False), ) @@ -303,27 +303,26 @@ def _decode_verifier_output(data: bytes, parse_json: bool) -> tuple[Any, list[st return data.decode("utf-8", errors="replace"), [] try: return json.loads(data), [] - except (UnicodeDecodeError, json.JSONDecodeError): + except (UnicodeDecodeError, json.JSONDecodeError, RecursionError): + # RecursionError is what json.loads raises for a deeply nested document. return data.decode("utf-8", errors="replace"), ["is not valid JSON"] -def _read_verifier_output( - path: Path, parse_json: bool, config: PluginConfig, max_bytes: int -) -> tuple[Any | None, int, list[str]]: - data, read_warning = _read_bounded_file(path, max_bytes) +def _read_verifier_output(path: Path, parse_json: bool, config: PluginConfig) -> tuple[Any | None, list[str]]: + data, read_warning = _read_safe_file(path) if data is None: - return None, 0, [] if read_warning is None else [f"omitted: {read_warning}"] + return None, [] if read_warning is None else [f"omitted: {read_warning}"] if not data: - return None, 0, [] + return None, [] value, warnings = _decode_verifier_output(data, parse_json) normalized = normalize_json( value, - max_bytes=config.max_attachment_bytes, + max_bytes=None, redact_patterns=config.redact_patterns, max_depth=20, redact_absolute_paths=False, ) - return normalized.value, len(data), [*warnings, *normalized.warnings] + return normalized.value, [*warnings, *normalized.warnings] def _verifier_output_attachment(result: Any, config: PluginConfig) -> tuple[Attachment | None, Any, list[str]]: @@ -331,18 +330,17 @@ def _verifier_output_attachment(result: Any, config: PluginConfig) -> tuple[Atta return None, None, [] outputs: list[tuple[str | None, dict[str, Any]]] = [] warnings: list[str] = [] - total = 0 for step_name, verifier_dir in verifier_output_paths(result): step_output: dict[str, Any] = {} - for key, filename, parse_json in _VERIFIER_OUTPUT_FILES: - path = verifier_dir / filename + for output_file in _VERIFIER_OUTPUT_FILES: + if output_file.requires_all and config.attachments != "all": + continue + path = verifier_dir / output_file.filename label = _step_label(step_name, path) - limit = min(config.max_attachment_bytes, config.max_total_attachment_bytes - total) - value, size, file_warnings = _read_verifier_output(path, parse_json, config, limit) + value, file_warnings = _read_verifier_output(path, output_file.parse_json, config) warnings.extend(f"{label} {warning}" for warning in file_warnings) if value is not None: - step_output[key] = value - total += size + step_output[output_file.key] = value if step_output: outputs.append((step_name, step_output)) @@ -350,9 +348,6 @@ def _verifier_output_attachment(result: Any, config: PluginConfig) -> tuple[Atta if summary is None: return None, None, warnings attachment_data = (canonical_json(summary) + "\n").encode() - if len(attachment_data) > min(config.max_attachment_bytes, config.max_total_attachment_bytes): - warnings.append("verifier-output.json omitted after redaction: attachment size limit") - return None, summary, warnings return ( Attachment(data=attachment_data, filename="verifier-output.json", content_type="application/json"), summary, @@ -360,19 +355,7 @@ def _verifier_output_attachment(result: Any, config: PluginConfig) -> tuple[Atta ) -def _remaining_attachment_config(config: PluginConfig, used_bytes: int) -> PluginConfig: - return replace(config, max_total_attachment_bytes=_remaining_attachment_bytes(config, used_bytes)) - - -def _remaining_attachment_bytes(config: PluginConfig, used_bytes: int) -> int: - return max(0, config.max_total_attachment_bytes - used_bytes) - - -def _attachments_size(attachments: dict[str, Attachment]) -> int: - return sum(len(attachment.data) for attachment in attachments.values()) - - -def _verifier_evidence(result: Any, config: PluginConfig) -> tuple[dict[str, Any], list[str], int]: +def _verifier_evidence(result: Any, config: PluginConfig) -> tuple[dict[str, Any], list[str]]: attachment, summary, warnings = _verifier_output_attachment(result, config) output: dict[str, Any] = {} if summary is not None: @@ -384,7 +367,7 @@ def _verifier_evidence(result: Any, config: PluginConfig) -> tuple[dict[str, Any ).value if attachment is not None: output["verifier_output"] = attachment - return output, warnings, 0 if attachment is None else len(attachment.data) + return output, warnings def _output_event(output: dict[str, Any]) -> dict[str, Any]: @@ -415,9 +398,8 @@ def __init__( include_tracebacks: Any = _UNSET, attachments: Any = _UNSET, artifact_include: Any = _UNSET, - max_attachment_bytes: Any = _UNSET, - max_total_attachment_bytes: Any = _UNSET, max_content_bytes: Any = _UNSET, + max_trajectory_bytes: Any = _UNSET, log_retry_attempts: Any = _UNSET, strict: Any = _UNSET, **kwargs: Any, @@ -442,9 +424,8 @@ def __init__( "include_tracebacks": include_tracebacks, "attachments": attachments, "artifact_include": artifact_include, - "max_attachment_bytes": max_attachment_bytes, - "max_total_attachment_bytes": max_total_attachment_bytes, "max_content_bytes": max_content_bytes, + "max_trajectory_bytes": max_trajectory_bytes, "log_retry_attempts": log_retry_attempts, "strict": strict, **kwargs, @@ -834,7 +815,6 @@ def _sync_final_result(self, result: Any) -> None: execution_input["extra_instructions"] = extra_instructions selected_artifacts, artifact_attachment_warnings = _artifact_attachments(result, self.config) metadata["harbor"]["warnings"].extend(artifact_attachment_warnings) - attachment_bytes = _attachments_size(selected_artifacts) agent_span = task.start_span( name="agent_execution", type="task", @@ -862,17 +842,13 @@ def _sync_final_result(self, result: Any) -> None: phase_start=agent_start, phase_end=agent_end, config=self.config, - _available_attachment_bytes=_remaining_attachment_bytes(self.config, attachment_bytes), ) atif_results.append((step_name, imported)) - attachment_bytes += imported.attachment_bytes if selected_artifacts: agent_span.log(output={"artifacts": selected_artifacts}) agent_span.end(end_time=agent_end) - verifier_config = _remaining_attachment_config(self.config, attachment_bytes) - verifier_output, verifier_warnings, verifier_attachment_bytes = _verifier_evidence(result, verifier_config) + verifier_output, verifier_warnings = _verifier_evidence(result, self.config) metadata["harbor"]["warnings"].extend(verifier_warnings) - attachment_bytes += verifier_attachment_bytes self._start_phase( task, result, @@ -949,13 +925,10 @@ def _sync_final_result(self, result: Any) -> None: task.log(metadata={"harbor": {"warnings": trajectory_warnings}}) task.end(end_time=root_end) - details_config = _remaining_attachment_config(self.config, attachment_bytes) - details_attachment, details_summary, detail_warnings = _attachment( - reward_details_paths(result), details_config - ) + details_attachment, details_summary, detail_warnings = _attachment(reward_details_paths(result), self.config) metadata["harbor"]["warnings"].extend(detail_warnings) # The summary is the same for every score, so bound it once rather than - # re-normalizing a payload up to max_attachment_bytes per scorer span. + # re-normalizing an unbounded payload per scorer span. bounded_details = ( None if details_summary is None diff --git a/py/src/braintrust/integrations/harbor/test_harbor.py b/py/src/braintrust/integrations/harbor/test_harbor.py index 51a86019..5ad80561 100644 --- a/py/src/braintrust/integrations/harbor/test_harbor.py +++ b/py/src/braintrust/integrations/harbor/test_harbor.py @@ -39,7 +39,7 @@ RuntimeState, _artifact_attachments, _attachment, - _read_bounded_file, + _read_safe_file, _resolve_project, _seconds, _timing, @@ -57,6 +57,12 @@ reduce_job, reduce_trial, ) +from braintrust.test_helpers import ( # noqa: F401 + find_span_by_name, + init_test_exp, + with_memory_logger, + with_simulate_login, +) from harbor.models.job.config import JobConfig, RetryConfig from harbor.models.job.lock import AgentSkillLock, JobLock, TaskLock, TrialLock from harbor.models.job.result import JobResult, JobStats @@ -451,9 +457,7 @@ def test_artifact_attachments_are_scoped_per_step(tmp_path): assert [step for step, _ in artifact_manifest_paths(result)] == ["first", "second"] -def test_reward_details_attachment_uses_the_per_file_limit(tmp_path): - # A multi-step trial merges one reward-details file per step, so the combined - # payload can exceed the per-file limit while every source file fits. +def test_reward_details_attachment_merges_every_step(tmp_path): entries = [] for step in ("first", "second"): path = tmp_path / step / "reward-details.json" @@ -461,13 +465,14 @@ def test_reward_details_attachment_uses_the_per_file_limit(tmp_path): path.write_text(json.dumps({"criteria": "x" * 400})) entries.append((step, path)) - config = PluginConfig.from_options(max_attachment_bytes=600, max_total_attachment_bytes=100_000) + config = PluginConfig.from_options() attachment, summary, warnings = _attachment(entries, config) - assert attachment is None + assert attachment is not None # The summary keys each step's details, so a score cannot misattribute them. assert sorted(summary) == ["first", "second"] - assert any("attachment size limit" in warning for warning in warnings) + assert json.loads(attachment.data) == summary + assert warnings == [] attachment, summary, warnings = _attachment([(None, entries[0][1])], config) assert attachment is not None @@ -476,6 +481,33 @@ def test_reward_details_attachment_uses_the_per_file_limit(tmp_path): assert warnings == [] +def test_oversized_trajectory_is_refused_before_it_is_parsed(tmp_path): + # trajectory.json is parsed into the host process rather than handed to + # object storage, so it keeps a bound of its own now that attachments do not. + trajectory_path = tmp_path / "trajectory.json" + trajectory_path.write_text(json.dumps({"steps": [{"step_id": 1, "source": "agent", "message": "hello"}]})) + oversized = PluginConfig.from_options(max_trajectory_bytes=4) + + summarized = summarize_trajectory(trajectory_path, oversized) + assert summarized.warnings == ("trajectory omitted: size limit",) + assert summarized.final_message is None + + imported = import_trajectory( + parent=None, + trajectory_path=trajectory_path, + trial_id="trial-1", + semantic_prefix="prefix", + phase_start=0.0, + phase_end=1.0, + config=oversized, + ) + assert imported.warnings == ("trajectory omitted: size limit",) + assert imported.imported_llm_spans == 0 + + # The same document is imported normally under the default bound. + assert summarize_trajectory(trajectory_path, PluginConfig.from_options()).warnings == () + + def test_verifier_output_attachment_collects_standard_harbor_evidence(tmp_path): result = _trial_result(tmp_path, "trial-1", "task-a") verifier_dir = tmp_path / "trial-1" / "verifier" @@ -500,7 +532,7 @@ def test_verifier_output_attachment_collects_standard_harbor_evidence(tmp_path): ) ) - config = PluginConfig.from_options(redact_patterns=(r"secret-value|Bearer verifier-secret",)) + config = PluginConfig.from_options(attachments="all", redact_patterns=(r"secret-value|Bearer verifier-secret",)) attachment, summary, warnings = _verifier_output_attachment(result, config) assert summary == { @@ -535,7 +567,7 @@ def test_verifier_output_attachment_handles_invalid_utf8_and_configured_redactio attachment, summary, warnings = _verifier_output_attachment( result, - PluginConfig.from_options(redact_patterns=(r"(?:header|opaque)-secret",)), + PluginConfig.from_options(attachments="all", redact_patterns=(r"(?:header|opaque)-secret",)), ) assert attachment is not None @@ -546,27 +578,6 @@ def test_verifier_output_attachment_handles_invalid_utf8_and_configured_redactio assert any("ctrf.json is not valid JSON" in warning for warning in warnings) -def test_verifier_output_attachment_rejects_oversized_file_before_reading_it(tmp_path, monkeypatch): - result = _trial_result(tmp_path, "trial-1", "task-a") - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) - stdout = verifier_dir / "test-stdout.txt" - stdout.write_text("too large") - - def fail_open(*_args, **_kwargs): - raise AssertionError("oversized output must be rejected from stat metadata") - - monkeypatch.setattr(os, "open", fail_open) - attachment, summary, warnings = _verifier_output_attachment( - result, - PluginConfig.from_options(max_attachment_bytes=4), - ) - - assert attachment is None - assert summary is None - assert warnings == ["test-stdout.txt omitted: attachment size limit"] - - @pytest.mark.parametrize("kind", ["symlink", "fifo"]) def test_verifier_output_attachment_rejects_unsafe_file_types(tmp_path, kind): result = _trial_result(tmp_path, "trial-1", "task-a") @@ -580,7 +591,8 @@ def test_verifier_output_attachment_rejects_unsafe_file_types(tmp_path, kind): else: os.mkfifo(stdout) - attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options()) + config = PluginConfig.from_options(attachments="all") + attachment, summary, warnings = _verifier_output_attachment(result, config) assert attachment is None assert summary is None @@ -599,7 +611,7 @@ def swap_after_inspection(_path, flags): monkeypatch.setattr(os, "open", swap_after_inspection) - assert _read_bounded_file(expected, 100) == (None, "unsafe file type") + assert _read_safe_file(expected) == (None, "unsafe file type") def test_verifier_output_attachment_scopes_steps_and_respects_attachment_mode(tmp_path): @@ -608,51 +620,55 @@ def test_verifier_output_attachment_scopes_steps_and_respects_attachment_mode(tm verifier_dir = tmp_path / "trial-1" / "steps" / step_name / "verifier" verifier_dir.mkdir(parents=True) (verifier_dir / "test-stdout.txt").write_text(f"{step_name} output\n") + (verifier_dir / "ctrf.json").write_text(json.dumps({"step": step_name})) - attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options()) + attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options(attachments="all")) assert summary == { - "first": {"stdout": "first output\n"}, - "second": {"stdout": "second output\n"}, + "first": {"stdout": "first output\n", "ctrf": {"step": "first"}}, + "second": {"stdout": "second output\n", "ctrf": {"step": "second"}}, } assert attachment is not None assert warnings == [] + + # Raw verifier text is only covered by configured redact_patterns, so the + # default tier ships the structured report and leaves the logs behind. + _attachment_default, default_summary, default_warnings = _verifier_output_attachment( + result, PluginConfig.from_options() + ) + assert default_summary == {"first": {"ctrf": {"step": "first"}}, "second": {"ctrf": {"step": "second"}}} + assert default_warnings == [] + assert _verifier_output_attachment(result, PluginConfig.from_options(attachments="none")) == (None, None, []) -@pytest.mark.parametrize("attachments", ["verifier-details", "none"]) -def test_final_sync_wires_verifier_evidence_to_verification_and_score_spans(tmp_path, attachments): - class RecordingSpan: - def __init__(self, **event): - self.event = event - self.children = [] - self.logs = [] +def test_deeply_nested_verifier_json_is_reported_not_raised(tmp_path): + # json.loads raises RecursionError rather than JSONDecodeError here, and it + # used to escape the sync between starting a trial's spans and logging them. + result = _trial_result(tmp_path, "trial-1", "task-a") + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "ctrf.json").write_bytes(b"[" * 200_000 + b"]" * 200_000) - def start_span(self, **event): - child = RecordingSpan(**event) - self.children.append(child) - return child + attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options()) - def log(self, **event): - self.logs.append(event) + assert attachment is not None + assert summary["ctrf"].startswith("[[[") + assert warnings == ["ctrf.json is not valid JSON"] - def end(self, **_event): - return None - class RecordingExperiment: - def __init__(self): - self.children = [] +def test_deeply_nested_reward_details_are_reported_not_raised(tmp_path): + path = tmp_path / "reward-details.json" + path.write_bytes(b"[" * 200_000 + b"]" * 200_000) - def start_span(self, **event): - span = RecordingSpan(**event) - self.children.append(span) - return span + attachment, summary, warnings = _attachment([(None, path)], PluginConfig.from_options()) - result = _trial_result(tmp_path, "trial-1", "task-a") - result.verifier_result = VerifierResult(rewards={"reward": 0.25}) - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) - (verifier_dir / "test-stdout.txt").write_text("assert 1 == 2\n") + assert (attachment, summary) == (None, None) + assert warnings == ["reward-details.json is not valid JSON"] + + +def _install_runtime(plugin, result, trials_dir, experiment): + """Point a plugin at one trial writing into a real Braintrust experiment.""" task = TaskData( logical_key="task-key", source="suite", @@ -665,10 +681,8 @@ def start_span(self, **event): task_dir=None, ) plan = TrialPlan(result.trial_name, result.config, None, task, 0) - snapshot = JobSnapshot("job-id", "job", tmp_path, None, None, (plan,)) - experiment = RecordingExperiment() + snapshot = JobSnapshot("job-id", "job", trials_dir, None, None, (plan,)) partition = Partition("partition", "experiment", "scope", experiment=experiment) - plugin = HarborPlugin(attachments=attachments) plugin._runtime = RuntimeState( snapshot, {result.trial_name: plan}, @@ -678,89 +692,130 @@ def start_span(self, **event): ) plugin._trial_machines[result.trial_name] = TrialMachine(result.trial_name) + +def _sync_spans(plugin, result, experiment, memory_logger): + """Run one trial through the plugin and return its flushed spans.""" plugin._sync_final_result(result) + experiment.flush() + return memory_logger.pop() + + +def _child_spans(spans, parent): + # Root spans carry span_parents=None rather than an empty list. + return [span for span in spans if parent["span_id"] in (span["span_parents"] or ())] + + +@pytest.mark.parametrize( + ("attachments", "expected_summary"), + [ + ("none", None), + # Raw verifier text needs the explicit opt-in; the structured report does not. + ("verifier-details", {"ctrf": {"failed": 1}}), + ("all", {"ctrf": {"failed": 1}, "stdout": "assert 1 == 2\n"}), + ], +) +def test_final_sync_logs_verifier_evidence_on_verification_and_score_spans( + tmp_path, attachments, expected_summary, with_memory_logger, with_simulate_login +): + result = _trial_result(tmp_path, "trial-1", "task-a") + result.verifier_result = VerifierResult(rewards={"reward": 0.25}) + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "test-stdout.txt").write_text("assert 1 == 2\n") + (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) + + experiment = init_test_exp("harbor-verifier-evidence", "harbor") + plugin = HarborPlugin(attachments=attachments) + _install_runtime(plugin, result, tmp_path, experiment) + + spans = _sync_spans(plugin, result, experiment, with_memory_logger) - root = experiment.children[0] - task_span = next(span for span in root.children if span.event["name"] == "task") - verification = next(span for span in task_span.children if span.event["name"] == "verification") - scorer = next(span for span in root.children if span.event["type"] == "score") + verification = find_span_by_name(spans, "verification") + scorer = next(span for span in spans if span["span_attributes"].get("type") == "score") if attachments == "none": - assert "output" not in verification.event - assert "verifier_output_summary" not in scorer.logs[0]["output"] - assert "verifier_output" not in scorer.logs[0]["output"] - else: - assert verification.event["output"]["verifier_output_summary"] == {"stdout": "assert 1 == 2\n"} - verifier_attachment = verification.event["output"]["verifier_output"] - assert verifier_attachment.reference["filename"] == "verifier-output.json" - assert ( - scorer.logs[0]["output"]["verifier_output_summary"] - == verification.event["output"]["verifier_output_summary"] - ) - assert scorer.logs[0]["output"]["verifier_output"] is verifier_attachment - - # The verification span owns the attachment, so unevaluated trials do not - # lose their complete evidence merely because no score span is created. - experiment.children.clear() - result.verifier_result = VerifierResult(rewards=None) - plugin._sync_final_result(result) - scoreless_root = experiment.children[0] - scoreless_task = next(span for span in scoreless_root.children if span.event["name"] == "task") - scoreless_verification = next(span for span in scoreless_task.children if span.event["name"] == "verification") - assert not any(span.event["type"] == "score" for span in scoreless_root.children) - assert scoreless_verification.event["output"]["verifier_output"] is not None - - # ATIF images consume the same trial attachment budget as verifier - # evidence, even though the image is logged on an agent child span. - agent_dir = tmp_path / "trial-1" / "agent" - agent_dir.mkdir(exist_ok=True) - (agent_dir / "first.png").write_bytes(b"i" * 600) - (agent_dir / "second.png").write_bytes(b"j" * 600) - (agent_dir / "trajectory.json").write_text( - json.dumps( - { - "steps": [ - { - "step_id": 1, - "source": "agent", - "message": [ - { - "type": "image", - "source": {"path": "first.png", "media_type": "image/png"}, - }, - { - "type": "image", - "source": {"path": "second.png", "media_type": "image/png"}, - }, - ], - } - ] - } - ) - ) - (verifier_dir / "test-stdout.txt").write_text("v" * 100) - experiment.children.clear() - result.verifier_result = VerifierResult(rewards={"reward": 0.25}) - budgeted_plugin = HarborPlugin(max_attachment_bytes=1000, max_total_attachment_bytes=1000) - budgeted_plugin._runtime = RuntimeState( - snapshot, - {result.trial_name: plan}, - {result.trial_name: partition}, - {"scope": DatasetBinding("scope")}, - {"partition": partition}, + assert "verifier_output_summary" not in (verification.get("output") or {}) + assert "verifier_output" not in scorer["output"] + assert with_memory_logger.upload_attempts == [] + return + + assert verification["output"]["verifier_output_summary"] == expected_summary + # The real serializer replaces the attachment with the reference that is + # stored on the span, and queues the payload itself for upload. + reference = verification["output"]["verifier_output"] + assert reference["type"] == "braintrust_attachment" + assert reference["filename"] == "verifier-output.json" + assert scorer["output"]["verifier_output_summary"] == verification["output"]["verifier_output_summary"] + assert scorer["output"]["verifier_output"] == reference + uploaded = {attachment.reference["key"] for attachment in with_memory_logger.upload_attempts} + assert reference["key"] in uploaded + + +def test_verification_span_keeps_verifier_evidence_without_scores(tmp_path, with_memory_logger, with_simulate_login): + # The verification span owns the attachment, so unevaluated trials do not + # lose their complete evidence merely because no score span is created. + result = _trial_result(tmp_path, "trial-1", "task-a") + result.verifier_result = VerifierResult(rewards=None) + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) + + experiment = init_test_exp("harbor-verifier-evidence", "harbor") + plugin = HarborPlugin() + _install_runtime(plugin, result, tmp_path, experiment) + + spans = _sync_spans(plugin, result, experiment, with_memory_logger) + + assert not any(span["span_attributes"].get("type") == "score" for span in spans) + verification = find_span_by_name(spans, "verification") + assert verification["output"]["verifier_output"]["type"] == "braintrust_attachment" + + +def test_trajectory_images_and_verifier_evidence_are_both_logged(tmp_path, with_memory_logger, with_simulate_login): + # ATIF images and verifier evidence are independent: neither competes with + # the other for a shared budget. + result = _trial_result(tmp_path, "trial-1", "task-a") + result.verifier_result = VerifierResult(rewards={"reward": 0.25}) + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) + agent_dir = tmp_path / "trial-1" / "agent" + agent_dir.mkdir(parents=True) + (agent_dir / "first.png").write_bytes(b"i" * 600) + (agent_dir / "second.png").write_bytes(b"j" * 600) + (agent_dir / "trajectory.json").write_text( + json.dumps( + { + "steps": [ + { + "step_id": 1, + "source": "agent", + "message": [ + {"type": "image", "source": {"path": "first.png", "media_type": "image/png"}}, + {"type": "image", "source": {"path": "second.png", "media_type": "image/png"}}, + ], + } + ] + } ) - budgeted_plugin._trial_machines[result.trial_name] = TrialMachine(result.trial_name) - - budgeted_plugin._sync_final_result(result) - - budgeted_root = experiment.children[0] - budgeted_task = next(span for span in budgeted_root.children if span.event["name"] == "task") - budgeted_agent = next(span for span in budgeted_task.children if span.event["name"] == "agent_execution") - trajectory_step = budgeted_agent.children[0] - trajectory_message = trajectory_step.logs[0]["output"]["message"] - assert trajectory_message[0]["type"] == "image_url" - assert trajectory_message[1] == {"type": "text", "text": "[image omitted: size limit]"} - budgeted_verification = next(span for span in budgeted_task.children if span.event["name"] == "verification") - assert budgeted_verification.event["output"]["verifier_output_summary"] == {"stdout": "v" * 100} + ) + + experiment = init_test_exp("harbor-verifier-evidence", "harbor") + plugin = HarborPlugin() + _install_runtime(plugin, result, tmp_path, experiment) + + spans = _sync_spans(plugin, result, experiment, with_memory_logger) + + agent_execution = find_span_by_name(spans, "agent_execution") + trajectory_step = _child_spans(spans, agent_execution)[0] + message = trajectory_step["output"]["message"] + assert [part["type"] for part in message] == ["image_url", "image_url"] + assert all(part["image_url"]["url"]["type"] == "braintrust_attachment" for part in message) + assert find_span_by_name(spans, "verification")["output"]["verifier_output_summary"] == {"ctrf": {"failed": 1}} + # Both images and the verifier payload reach the uploader. + assert sorted(a.data for a in with_memory_logger.upload_attempts if a.data in (b"i" * 600, b"j" * 600)) == [ + b"i" * 600, + b"j" * 600, + ] def test_disabled_plugin_does_not_reconcile_or_write_spans(): From d55bf5c64345f47724ea2044c8b38e5ab688d932 Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Mon, 31 Aug 2026 15:55:52 -0400 Subject: [PATCH 3/5] fix(harbor): guard every task-controlled json parse, align preview depth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to 3306c05d, from a quality pass over the branch. `RecursionError` from `json.loads` was guarded at two call sites but not on task-written `manifest.json` or either `trajectory.json` reader. The guard is now `identity.try_parse_json`, used at all five. Those readers also moved to `read_bytes`, closing a second hole: `read_text` raises `UnicodeDecodeError`, a `ValueError`, which the old `except (OSError, JSONDecodeError)` never caught either. Span previews used `normalize_json`'s default `max_depth=8` while their attachment used 20, so a 12-deep `ctrf.json` came out `[DROPPED: depth limit]` in the span and whole in the attachment. That default was incidental. `_bounded_summary` now passes 20 for both verifier output and reward details, and skips the walk when the payload already fits — its serialized size is known from the attachment, so re-walking and re-measuring a multi-MB payload bought nothing. --- py/src/braintrust/integrations/harbor/atif.py | 28 ++-- .../integrations/harbor/identity.py | 12 ++ .../braintrust/integrations/harbor/plugin.py | 133 ++++++++++-------- .../integrations/harbor/test_harbor.py | 123 +++++++++++----- 4 files changed, 184 insertions(+), 112 deletions(-) diff --git a/py/src/braintrust/integrations/harbor/atif.py b/py/src/braintrust/integrations/harbor/atif.py index cb5dc5d5..75ddcaa7 100644 --- a/py/src/braintrust/integrations/harbor/atif.py +++ b/py/src/braintrust/integrations/harbor/atif.py @@ -11,7 +11,7 @@ from braintrust.logger import Attachment from .config import PluginConfig -from .identity import NormalizedValue, child_span_id, normalize_json +from .identity import NormalizedValue, child_span_id, normalize_json, try_parse_json _INSTRUMENTATION = "braintrust.plugin.harbor" @@ -273,9 +273,12 @@ def summarize_trajectory(trajectory_path: Path, config: PluginConfig) -> ATIFImp try: if trajectory_path.stat().st_size > config.max_trajectory_bytes: return ATIFImportResult(warnings=("trajectory omitted: size limit",)) - trajectory = json.loads(trajectory_path.read_text()) - except (OSError, json.JSONDecodeError) as exc: + data = trajectory_path.read_bytes() + except OSError as exc: return ATIFImportResult(warnings=(f"trajectory unavailable or malformed: {exc}",)) + trajectory, parsed_ok = try_parse_json(data) + if not parsed_ok: + return ATIFImportResult(warnings=("trajectory unavailable or malformed: not valid JSON",)) if not isinstance(trajectory, dict) or not isinstance(trajectory.get("steps"), list): return ATIFImportResult(warnings=("trajectory malformed: steps must be an array",)) notes = _Notes() @@ -324,9 +327,12 @@ def import_trajectory( # rather than handed to object storage, so its size is bounded. if trajectory_path.stat().st_size > config.max_trajectory_bytes: return ATIFImportResult(warnings=("trajectory omitted: size limit",)) - trajectory = json.loads(trajectory_path.read_text()) - except (OSError, json.JSONDecodeError) as exc: + data = trajectory_path.read_bytes() + except OSError as exc: return ATIFImportResult(warnings=(f"trajectory unavailable or malformed: {exc}",)) + trajectory, parsed_ok = try_parse_json(data) + if not parsed_ok: + return ATIFImportResult(warnings=("trajectory unavailable or malformed: not valid JSON",)) if not isinstance(trajectory, dict) or not isinstance(trajectory.get("steps"), list): return ATIFImportResult(warnings=("trajectory malformed: steps must be an array",)) @@ -354,11 +360,7 @@ def import_trajectory( for index, step in enumerate(steps): source = step.get("source") content, content_complete = _content( - step.get("message"), - trajectory_path.parent, - config, - notes, - f"step {index + 1} message", + step.get("message"), trajectory_path.parent, config, notes, f"step {index + 1} message" ) if source in {"system", "user"}: if config.content_mode != "metadata": @@ -458,11 +460,7 @@ def import_trajectory( ): tool_context = f"step {index + 1} tool {call_id}" tool_output, tool_complete = _content( - result.get("content"), - trajectory_path.parent, - config, - notes, - f"{tool_context} result", + result.get("content"), trajectory_path.parent, config, notes, f"{tool_context} result" ) tool_input = _bounded(arguments, config, notes, f"{tool_context} arguments") result_extra = result.get("extra") if isinstance(result.get("extra"), dict) else {} diff --git a/py/src/braintrust/integrations/harbor/identity.py b/py/src/braintrust/integrations/harbor/identity.py index c47ce28c..f4bf76cf 100644 --- a/py/src/braintrust/integrations/harbor/identity.py +++ b/py/src/braintrust/integrations/harbor/identity.py @@ -108,6 +108,18 @@ def _is_secret_key(key: str, value: Any) -> bool: return _key_segments(key).isdisjoint(_COUNTER_SEGMENTS) +def try_parse_json(data: bytes) -> tuple[Any, bool]: + """Parse a task-controlled document, reporting failure rather than raising. + + json.loads raises RecursionError, not JSONDecodeError, for a deeply nested + document, and a task can write one to any file this package reads. + """ + try: + return json.loads(data), True + except (UnicodeDecodeError, json.JSONDecodeError, RecursionError): + return None, False + + def _json_size(value: Any) -> int: try: return len(canonical_json(value).encode("utf-8")) diff --git a/py/src/braintrust/integrations/harbor/plugin.py b/py/src/braintrust/integrations/harbor/plugin.py index 34a0d7eb..880b0989 100644 --- a/py/src/braintrust/integrations/harbor/plugin.py +++ b/py/src/braintrust/integrations/harbor/plugin.py @@ -39,6 +39,7 @@ normalize_json, partition_key, semantic_agent_config, + try_parse_json, ) from .rewards import classify_rewards, extract_json_path, validate_classifications from .state import ( @@ -183,6 +184,21 @@ def _read_safe_file(path: Path) -> tuple[bytes | None, str | None]: return data, None +def _json_attachment( + items: list[tuple[str | None, Any]], default_key: str, filename: str, warnings: list[str] +) -> tuple[Attachment | None, Any, list[str]]: + """Merge per-step values into one JSON attachment plus the summary it holds.""" + summary = _by_step(items, default_key) + if summary is None: + return None, None, warnings + attachment_data = (canonical_json(summary) + "\n").encode() + return ( + Attachment(data=attachment_data, filename=filename, content_type="application/json"), + summary, + warnings, + ) + + def _read_json_summary(entries: list[tuple[str | None, Path]], max_bytes: int) -> tuple[Any, list[str]]: summaries: list[tuple[str | None, Any]] = [] warnings: list[str] = [] @@ -195,7 +211,11 @@ def _read_json_summary(entries: list[tuple[str | None, Path]], max_bytes: int) - if len(data) > max_bytes: warnings.append(f"{label} omitted: size limit") continue - summaries.append((step_name, json.loads(data))) + parsed, parsed_ok = try_parse_json(data) + if not parsed_ok: + warnings.append(f"{label} is not valid JSON") + continue + summaries.append((step_name, parsed)) except FileNotFoundError: continue except (OSError, json.JSONDecodeError) as exc: @@ -255,9 +275,8 @@ def _attachment( if read_warning is not None: warnings.append(f"{label} omitted: {read_warning}") continue - try: - parsed = json.loads(data) - except (json.JSONDecodeError, RecursionError): + parsed, parsed_ok = try_parse_json(data) + if not parsed_ok: warnings.append(f"{label} is not valid JSON") continue normalized = normalize_json( @@ -268,15 +287,7 @@ def _attachment( ) warnings.extend(normalized.warnings) complete.append((step_name, normalized.value)) - summary = _by_step(complete, "details") - if summary is None: - return None, None, warnings - attachment_data = (canonical_json(summary) + "\n").encode() - return ( - Attachment(data=attachment_data, filename=filename, content_type="application/json"), - summary, - warnings, - ) + return _json_attachment(complete, "details", filename, warnings) @dataclass(frozen=True) @@ -286,8 +297,10 @@ class _VerifierOutputFile: parse_json: bool # normalize_json redacts by key name, which a structured document supplies and # raw text does not: raw verifier output is only covered by configured - # redact_patterns. Treat it like any other arbitrary byte stream the task - # wrote and require the same explicit opt-in that artifact_include does. + # redact_patterns. Keep it out of the default tier so shipping unredactable + # bytes is always a deliberate choice. Note this is a weaker gate than the one + # on artifact_include, which needs attachments="all" *and* an explicit glob: + # attachments="all" alone is enough to opt into raw verifier logs. requires_all: bool @@ -298,23 +311,20 @@ class _VerifierOutputFile: ) -def _decode_verifier_output(data: bytes, parse_json: bool) -> tuple[Any, list[str]]: - if not parse_json: - return data.decode("utf-8", errors="replace"), [] - try: - return json.loads(data), [] - except (UnicodeDecodeError, json.JSONDecodeError, RecursionError): - # RecursionError is what json.loads raises for a deeply nested document. - return data.decode("utf-8", errors="replace"), ["is not valid JSON"] - - def _read_verifier_output(path: Path, parse_json: bool, config: PluginConfig) -> tuple[Any | None, list[str]]: data, read_warning = _read_safe_file(path) if data is None: return None, [] if read_warning is None else [f"omitted: {read_warning}"] if not data: return None, [] - value, warnings = _decode_verifier_output(data, parse_json) + warnings: list[str] = [] + value: Any = data.decode("utf-8", errors="replace") + if parse_json: + parsed, parsed_ok = try_parse_json(data) + if parsed_ok: + value = parsed + else: + warnings.append("is not valid JSON") normalized = normalize_json( value, max_bytes=None, @@ -343,35 +353,45 @@ def _verifier_output_attachment(result: Any, config: PluginConfig) -> tuple[Atta step_output[output_file.key] = value if step_output: outputs.append((step_name, step_output)) + return _json_attachment(outputs, "verifier", "verifier-output.json", warnings) - summary = _by_step(outputs, "verifier") - if summary is None: - return None, None, warnings - attachment_data = (canonical_json(summary) + "\n").encode() - return ( - Attachment(data=attachment_data, filename="verifier-output.json", content_type="application/json"), + +def _bounded_summary( + summary: Any, serialized_bytes: int, config: PluginConfig, *, redact_absolute_paths: bool = True +) -> Any: + """Bound an attachment's summary for the span field that previews it. + + The attachment payload is this same value already normalized with the same + patterns and depth, so its serialized length is the summary's size and one + that already fits needs no second walk. Keep max_depth in step with the + attachment's, or the preview would truncate structure the attachment kept. + """ + if serialized_bytes <= config.max_content_bytes: + return summary + return normalize_json( summary, - warnings, - ) + max_bytes=config.max_content_bytes, + redact_patterns=config.redact_patterns, + max_depth=20, + redact_absolute_paths=redact_absolute_paths, + ).value -def _verifier_evidence(result: Any, config: PluginConfig) -> tuple[dict[str, Any], list[str]]: - attachment, summary, warnings = _verifier_output_attachment(result, config) - output: dict[str, Any] = {} - if summary is not None: - output["verifier_output_summary"] = normalize_json( - summary, - max_bytes=config.max_content_bytes, - redact_patterns=config.redact_patterns, - redact_absolute_paths=False, - ).value - if attachment is not None: - output["verifier_output"] = attachment - return output, warnings +def _serialized_bytes(attachment: Attachment) -> int: + # _json_attachment appends a trailing newline that canonical sizing omits. + return len(attachment.data) - 1 -def _output_event(output: dict[str, Any]) -> dict[str, Any]: - return {} if not output else {"output": output} +def _verifier_evidence(result: Any, config: PluginConfig) -> tuple[dict[str, Any], list[str]]: + attachment, summary, warnings = _verifier_output_attachment(result, config) + if attachment is None: + return {}, warnings + return { + "verifier_output_summary": _bounded_summary( + summary, _serialized_bytes(attachment), config, redact_absolute_paths=False + ), + "verifier_output": attachment, + }, warnings class HarborPlugin: @@ -727,7 +747,7 @@ def _start_phase( trial_id: str, root_start: float, root_end: float, - **event: Any, + output: dict[str, Any] | None = None, ) -> Any: start, end = _timing(getattr(result, timing_name, None), root_start, root_end) span = task_span.start_span( @@ -737,7 +757,8 @@ def _start_phase( start_time=start, set_current=False, internal={"instrumentation": _INSTRUMENTATION}, - **event, + # A phase with nothing to report must not log an empty output field. + **({"output": output} if output else {}), ) span.end(end_time=end) return span @@ -857,7 +878,7 @@ def _sync_final_result(self, result: Any) -> None: trial_id, root_start, root_end, - **_output_event(verifier_output), + output=verifier_output, ) for step in getattr(result, "step_results", None) or []: @@ -931,12 +952,8 @@ def _sync_final_result(self, result: Any) -> None: # re-normalizing an unbounded payload per scorer span. bounded_details = ( None - if details_summary is None - else normalize_json( - details_summary, - max_bytes=self.config.max_content_bytes, - redact_patterns=self.config.redact_patterns, - ).value + if details_attachment is None + else _bounded_summary(details_summary, _serialized_bytes(details_attachment), self.config) ) for score in conversion.scores: scorer = root.start_span( diff --git a/py/src/braintrust/integrations/harbor/test_harbor.py b/py/src/braintrust/integrations/harbor/test_harbor.py index 5ad80561..dece7211 100644 --- a/py/src/braintrust/integrations/harbor/test_harbor.py +++ b/py/src/braintrust/integrations/harbor/test_harbor.py @@ -43,6 +43,7 @@ _resolve_project, _seconds, _timing, + _verifier_evidence, _verifier_output_attachment, ) from braintrust.integrations.harbor.rewards import classify_rewards, validate_classifications @@ -59,6 +60,7 @@ ) from braintrust.test_helpers import ( # noqa: F401 find_span_by_name, + find_spans_by_type, init_test_exp, with_memory_logger, with_simulate_login, @@ -437,6 +439,14 @@ def _trial_result(trials_dir, trial_name, task_name, step_names=()): ) +def _verifier_trial(tmp_path): + """Build a single-phase trial with an empty verifier output directory.""" + result = _trial_result(tmp_path, "trial-1", "task-a") + verifier_dir = tmp_path / "trial-1" / "verifier" + verifier_dir.mkdir(parents=True) + return result, verifier_dir + + def test_artifact_attachments_are_scoped_per_step(tmp_path): result = _trial_result(tmp_path, "trial-1", "task-a", step_names=("first", "second")) for step_name, contents in (("first", b"first output"), ("second", b"second output")): @@ -509,9 +519,7 @@ def test_oversized_trajectory_is_refused_before_it_is_parsed(tmp_path): def test_verifier_output_attachment_collects_standard_harbor_evidence(tmp_path): - result = _trial_result(tmp_path, "trial-1", "task-a") - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) + result, verifier_dir = _verifier_trial(tmp_path) (verifier_dir / "test-stdout.txt").write_text("FAILED test_answer.py::test_count - assert 27 == 28\n") (verifier_dir / "test-stderr.txt").write_text("token=secret-value\n") (verifier_dir / "ctrf.json").write_text( @@ -559,30 +567,21 @@ def test_verifier_output_attachment_collects_standard_harbor_evidence(tmp_path): def test_verifier_output_attachment_handles_invalid_utf8_and_configured_redaction(tmp_path): - result = _trial_result(tmp_path, "trial-1", "task-a") - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) - (verifier_dir / "test-stdout.txt").write_text("Authorization: Bearer header-secret\n") + result, verifier_dir = _verifier_trial(tmp_path) (verifier_dir / "ctrf.json").write_bytes(b"\xff token=opaque-secret\n") attachment, summary, warnings = _verifier_output_attachment( - result, - PluginConfig.from_options(attachments="all", redact_patterns=(r"(?:header|opaque)-secret",)), + result, PluginConfig.from_options(redact_patterns=(r"opaque-secret",)) ) assert attachment is not None - assert summary == { - "stdout": "Authorization: Bearer [REDACTED]\n", - "ctrf": "\ufffd token=[REDACTED]\n", - } + assert summary == {"ctrf": "\ufffd token=[REDACTED]\n"} assert any("ctrf.json is not valid JSON" in warning for warning in warnings) @pytest.mark.parametrize("kind", ["symlink", "fifo"]) def test_verifier_output_attachment_rejects_unsafe_file_types(tmp_path, kind): - result = _trial_result(tmp_path, "trial-1", "task-a") - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) + result, verifier_dir = _verifier_trial(tmp_path) stdout = verifier_dir / "test-stdout.txt" if kind == "symlink": secret = tmp_path / "host-secret" @@ -599,7 +598,7 @@ def test_verifier_output_attachment_rejects_unsafe_file_types(tmp_path, kind): assert warnings == ["test-stdout.txt omitted: unsafe file type"] -def test_bounded_file_read_rejects_replacement_between_inspection_and_open(tmp_path, monkeypatch): +def test_safe_file_read_rejects_replacement_between_inspection_and_open(tmp_path, monkeypatch): expected = tmp_path / "expected" replacement = tmp_path / "replacement" expected.write_text("safe") @@ -633,21 +632,34 @@ def test_verifier_output_attachment_scopes_steps_and_respects_attachment_mode(tm # Raw verifier text is only covered by configured redact_patterns, so the # default tier ships the structured report and leaves the logs behind. - _attachment_default, default_summary, default_warnings = _verifier_output_attachment( - result, PluginConfig.from_options() - ) + _, default_summary, default_warnings = _verifier_output_attachment(result, PluginConfig.from_options()) assert default_summary == {"first": {"ctrf": {"step": "first"}}, "second": {"ctrf": {"step": "second"}}} assert default_warnings == [] assert _verifier_output_attachment(result, PluginConfig.from_options(attachments="none")) == (None, None, []) +def test_verifier_summary_keeps_the_structure_the_attachment_kept(tmp_path): + # The span preview and the attachment are the same value, so a document + # nested deeper than normalize_json's default limit but within the limit the + # attachment uses must not be truncated in one and whole in the other. + result, verifier_dir = _verifier_trial(tmp_path) + nested = {"leaf": "kept"} + for _ in range(12): + nested = {"a": nested} + (verifier_dir / "ctrf.json").write_text(json.dumps(nested)) + + output, warnings = _verifier_evidence(result, PluginConfig.from_options()) + + assert warnings == [] + assert output["verifier_output_summary"] == {"ctrf": nested} + assert json.loads(output["verifier_output"].data) == output["verifier_output_summary"] + + def test_deeply_nested_verifier_json_is_reported_not_raised(tmp_path): # json.loads raises RecursionError rather than JSONDecodeError here, and it # used to escape the sync between starting a trial's spans and logging them. - result = _trial_result(tmp_path, "trial-1", "task-a") - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) + result, verifier_dir = _verifier_trial(tmp_path) (verifier_dir / "ctrf.json").write_bytes(b"[" * 200_000 + b"]" * 200_000) attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options()) @@ -657,6 +669,28 @@ def test_deeply_nested_verifier_json_is_reported_not_raised(tmp_path): assert warnings == ["ctrf.json is not valid JSON"] +def test_deeply_nested_trajectory_is_reported_not_raised(tmp_path): + # trajectory.json is task-controlled too, and both readers parse it in-process. + trajectory_path = tmp_path / "trajectory.json" + trajectory_path.write_bytes(b"[" * 200_000 + b"]" * 200_000) + config = PluginConfig.from_options() + + assert summarize_trajectory(trajectory_path, config).warnings == ( + "trajectory unavailable or malformed: not valid JSON", + ) + + imported = import_trajectory( + parent=None, + trajectory_path=trajectory_path, + trial_id="trial-1", + semantic_prefix="prefix", + phase_start=0.0, + phase_end=1.0, + config=config, + ) + assert imported.warnings == ("trajectory unavailable or malformed: not valid JSON",) + + def test_deeply_nested_reward_details_are_reported_not_raised(tmp_path): path = tmp_path / "reward-details.json" path.write_bytes(b"[" * 200_000 + b"]" * 200_000) @@ -705,10 +739,17 @@ def _child_spans(spans, parent): return [span for span in spans if parent["span_id"] in (span["span_parents"] or ())] +def _scored_verifier_trial(tmp_path): + result, verifier_dir = _verifier_trial(tmp_path) + result.verifier_result = VerifierResult(rewards={"reward": 0.25}) + (verifier_dir / "test-stdout.txt").write_text("assert 1 == 2\n") + (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) + return result + + @pytest.mark.parametrize( ("attachments", "expected_summary"), [ - ("none", None), # Raw verifier text needs the explicit opt-in; the structured report does not. ("verifier-details", {"ctrf": {"failed": 1}}), ("all", {"ctrf": {"failed": 1}, "stdout": "assert 1 == 2\n"}), @@ -717,13 +758,7 @@ def _child_spans(spans, parent): def test_final_sync_logs_verifier_evidence_on_verification_and_score_spans( tmp_path, attachments, expected_summary, with_memory_logger, with_simulate_login ): - result = _trial_result(tmp_path, "trial-1", "task-a") - result.verifier_result = VerifierResult(rewards={"reward": 0.25}) - verifier_dir = tmp_path / "trial-1" / "verifier" - verifier_dir.mkdir(parents=True) - (verifier_dir / "test-stdout.txt").write_text("assert 1 == 2\n") - (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) - + result = _scored_verifier_trial(tmp_path) experiment = init_test_exp("harbor-verifier-evidence", "harbor") plugin = HarborPlugin(attachments=attachments) _install_runtime(plugin, result, tmp_path, experiment) @@ -731,13 +766,7 @@ def test_final_sync_logs_verifier_evidence_on_verification_and_score_spans( spans = _sync_spans(plugin, result, experiment, with_memory_logger) verification = find_span_by_name(spans, "verification") - scorer = next(span for span in spans if span["span_attributes"].get("type") == "score") - if attachments == "none": - assert "verifier_output_summary" not in (verification.get("output") or {}) - assert "verifier_output" not in scorer["output"] - assert with_memory_logger.upload_attempts == [] - return - + scorer = find_spans_by_type(spans, "score")[0] assert verification["output"]["verifier_output_summary"] == expected_summary # The real serializer replaces the attachment with the reference that is # stored on the span, and queues the payload itself for upload. @@ -750,6 +779,22 @@ def test_final_sync_logs_verifier_evidence_on_verification_and_score_spans( assert reference["key"] in uploaded +def test_final_sync_logs_no_verifier_evidence_when_attachments_are_disabled( + tmp_path, with_memory_logger, with_simulate_login +): + result = _scored_verifier_trial(tmp_path) + experiment = init_test_exp("harbor-verifier-evidence", "harbor") + plugin = HarborPlugin(attachments="none") + _install_runtime(plugin, result, tmp_path, experiment) + + spans = _sync_spans(plugin, result, experiment, with_memory_logger) + + verification = find_span_by_name(spans, "verification") + assert "verifier_output_summary" not in (verification.get("output") or {}) + assert "verifier_output" not in find_spans_by_type(spans, "score")[0]["output"] + assert with_memory_logger.upload_attempts == [] + + def test_verification_span_keeps_verifier_evidence_without_scores(tmp_path, with_memory_logger, with_simulate_login): # The verification span owns the attachment, so unevaluated trials do not # lose their complete evidence merely because no score span is created. @@ -765,7 +810,7 @@ def test_verification_span_keeps_verifier_evidence_without_scores(tmp_path, with spans = _sync_spans(plugin, result, experiment, with_memory_logger) - assert not any(span["span_attributes"].get("type") == "score" for span in spans) + assert not find_spans_by_type(spans, "score") verification = find_span_by_name(spans, "verification") assert verification["output"]["verifier_output"]["type"] == "braintrust_attachment" From dd4aeb958c20b567fe46ad7134de155b14eff4f3 Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Mon, 31 Aug 2026 16:33:53 -0400 Subject: [PATCH 4/5] fix(harbor): make verifier evidence tests platform-independent The verifier-output tests wrote raw stdout/stderr fixtures with Path.write_text, which translates "\n" to "\r\n" in text mode on Windows. The plugin decodes the file's bytes verbatim, so the summary carried CRLF and the assertions failed on Windows only. Write those fixtures as bytes so the on-disk content matches what is asserted. Also skip the fifo parametrization of the unsafe-file-type test where os.mkfifo does not exist. Co-Authored-By: Claude Opus 5 (1M context) --- .../integrations/harbor/test_harbor.py | 21 ++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/py/src/braintrust/integrations/harbor/test_harbor.py b/py/src/braintrust/integrations/harbor/test_harbor.py index dece7211..70345024 100644 --- a/py/src/braintrust/integrations/harbor/test_harbor.py +++ b/py/src/braintrust/integrations/harbor/test_harbor.py @@ -520,8 +520,10 @@ def test_oversized_trajectory_is_refused_before_it_is_parsed(tmp_path): def test_verifier_output_attachment_collects_standard_harbor_evidence(tmp_path): result, verifier_dir = _verifier_trial(tmp_path) - (verifier_dir / "test-stdout.txt").write_text("FAILED test_answer.py::test_count - assert 27 == 28\n") - (verifier_dir / "test-stderr.txt").write_text("token=secret-value\n") + # write_bytes, not write_text: on Windows text mode would rewrite "\n" as + # "\r\n", and the plugin decodes the file's bytes exactly as written. + (verifier_dir / "test-stdout.txt").write_bytes(b"FAILED test_answer.py::test_count - assert 27 == 28\n") + (verifier_dir / "test-stderr.txt").write_bytes(b"token=secret-value\n") (verifier_dir / "ctrf.json").write_text( json.dumps( { @@ -579,7 +581,16 @@ def test_verifier_output_attachment_handles_invalid_utf8_and_configured_redactio assert any("ctrf.json is not valid JSON" in warning for warning in warnings) -@pytest.mark.parametrize("kind", ["symlink", "fifo"]) +@pytest.mark.parametrize( + "kind", + [ + "symlink", + pytest.param( + "fifo", + marks=pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="os.mkfifo is not available on Windows"), + ), + ], +) def test_verifier_output_attachment_rejects_unsafe_file_types(tmp_path, kind): result, verifier_dir = _verifier_trial(tmp_path) stdout = verifier_dir / "test-stdout.txt" @@ -618,7 +629,7 @@ def test_verifier_output_attachment_scopes_steps_and_respects_attachment_mode(tm for step_name in ("first", "second"): verifier_dir = tmp_path / "trial-1" / "steps" / step_name / "verifier" verifier_dir.mkdir(parents=True) - (verifier_dir / "test-stdout.txt").write_text(f"{step_name} output\n") + (verifier_dir / "test-stdout.txt").write_bytes(f"{step_name} output\n".encode()) (verifier_dir / "ctrf.json").write_text(json.dumps({"step": step_name})) attachment, summary, warnings = _verifier_output_attachment(result, PluginConfig.from_options(attachments="all")) @@ -742,7 +753,7 @@ def _child_spans(spans, parent): def _scored_verifier_trial(tmp_path): result, verifier_dir = _verifier_trial(tmp_path) result.verifier_result = VerifierResult(rewards={"reward": 0.25}) - (verifier_dir / "test-stdout.txt").write_text("assert 1 == 2\n") + (verifier_dir / "test-stdout.txt").write_bytes(b"assert 1 == 2\n") (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) return result From c3f12ea0390adb4cfa7d360ef6e50b548019e17b Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Tue, 1 Sep 2026 11:06:32 -0400 Subject: [PATCH 5/5] rename --- examples/harbor/README.md | 4 +-- .../braintrust/integrations/harbor/config.py | 6 ++--- .../braintrust/integrations/harbor/plugin.py | 8 +++--- .../integrations/harbor/test_harbor.py | 25 +++++++++++++++---- 4 files changed, 29 insertions(+), 14 deletions(-) diff --git a/examples/harbor/README.md b/examples/harbor/README.md index cbfeddc7..2e6a6129 100644 --- a/examples/harbor/README.md +++ b/examples/harbor/README.md @@ -39,9 +39,9 @@ uv run harbor run \ The agent solves the task in `task/`, and Harbor's verifier emits a normalized `reward` plus an `answer_length` metric. The plugin creates `jobs/braintrust-harbor-example/braintrust-sync.json` after synchronization. -With the default `attachments=verifier-details` mode, the verification span and each score also include Harbor's conventional `ctrf.json` output when present. The size-bounded summary and complete redacted `verifier-output.json` attachment appear together in the span output. Structured CTRF fields with sensitive key names use the plugin's standard redaction. +With the default `attachments=all` mode, the verification span and each score also include Harbor's captured `test-stdout.txt`, optional `test-stderr.txt`, and conventional `ctrf.json` output when present. The size-bounded summary and complete redacted `verifier-output.json` attachment appear together in the span output. Structured CTRF fields with sensitive key names use the plugin's standard redaction, and configured `redact_patterns` apply to both structured CTRF strings and raw verifier text. -Harbor's captured `test-stdout.txt` and `test-stderr.txt` require `attachments=all`. Redaction works from key names, which raw text does not have, so **configured `redact_patterns` are the only redaction applied to raw verifier logs** — and `redact_patterns` is empty by default. Verifier logs are a common place for environment dumps, tokens in URLs, and other credentials, so set `redact_patterns` before enabling this, or leave the logs out and rely on `ctrf.json` for structured failure detail. +Use `attachments=structured` to keep only structured verifier evidence such as `ctrf.json`, or `attachments=none` to disable verifier and reward-detail attachments. Redaction works from key names, which raw text does not have, so **configured `redact_patterns` are the only redaction applied to raw verifier logs** — and `redact_patterns` is empty by default. Verifier logs are a common place for environment dumps, tokens in URLs, and other credentials, so configure `redact_patterns` when the evaluation environment contains sensitive values. By default, the plugin uses `Harbor` as the Braintrust project name. Override it with `--plugin-kwarg project_name=example-harbor` or through `.env`: diff --git a/py/src/braintrust/integrations/harbor/config.py b/py/src/braintrust/integrations/harbor/config.py index d5a2757e..c4a658d9 100644 --- a/py/src/braintrust/integrations/harbor/config.py +++ b/py/src/braintrust/integrations/harbor/config.py @@ -98,7 +98,7 @@ class PluginConfig: classifier_rules: dict[str, str] = field(default_factory=dict) invalid_score_policy: str = "metric" include_tracebacks: bool = False - attachments: str = "verifier-details" + attachments: str = "all" artifact_include: tuple[str, ...] = () max_content_bytes: int = 20_000 max_trajectory_bytes: int = 20_000_000 @@ -155,8 +155,8 @@ def validate(self) -> None: raise ValueError("log_retry_attempts=True is not implemented; only the final attempt is logged") if self.invalid_score_policy not in {"metric", "drop", "error"}: raise ValueError("invalid_score_policy must be 'metric', 'drop', or 'error'") - if self.attachments not in {"none", "verifier-details", "all"}: - raise ValueError("attachments must be 'none', 'verifier-details', or 'all'") + if self.attachments not in {"none", "structured", "all"}: + raise ValueError("attachments must be 'none', 'structured', or 'all'") if self.artifact_include and self.attachments != "all": raise ValueError("artifact_include requires attachments='all'") diff --git a/py/src/braintrust/integrations/harbor/plugin.py b/py/src/braintrust/integrations/harbor/plugin.py index 880b0989..afca942a 100644 --- a/py/src/braintrust/integrations/harbor/plugin.py +++ b/py/src/braintrust/integrations/harbor/plugin.py @@ -297,10 +297,10 @@ class _VerifierOutputFile: parse_json: bool # normalize_json redacts by key name, which a structured document supplies and # raw text does not: raw verifier output is only covered by configured - # redact_patterns. Keep it out of the default tier so shipping unredactable - # bytes is always a deliberate choice. Note this is a weaker gate than the one - # on artifact_include, which needs attachments="all" *and* an explicit glob: - # attachments="all" alone is enough to opt into raw verifier logs. + # redact_patterns. The structured tier excludes it for callers that do not + # want raw eval logs. Note this is a weaker gate than the one on + # artifact_include, which needs attachments="all" *and* an explicit glob: + # attachments="all" alone includes raw verifier logs. requires_all: bool diff --git a/py/src/braintrust/integrations/harbor/test_harbor.py b/py/src/braintrust/integrations/harbor/test_harbor.py index 70345024..c34e7342 100644 --- a/py/src/braintrust/integrations/harbor/test_harbor.py +++ b/py/src/braintrust/integrations/harbor/test_harbor.py @@ -143,6 +143,13 @@ def test_plugin_defaults_project_to_harbor(): assert _resolve_project(PluginConfig.from_options(project_id="project-id")) == (None, "project-id") +def test_attachment_modes_default_to_all_and_name_the_structured_only_tier(): + assert PluginConfig.from_options().attachments == "all" + assert PluginConfig.from_options(attachments="structured").attachments == "structured" + with pytest.raises(ValueError, match="attachments must be 'none', 'structured', or 'all'"): + PluginConfig.from_options(attachments="verifier-details") + + def test_harbor_resolves_the_plugin_through_its_entry_point(): # Users select this plugin with `--plugin braintrust`, which Harbor resolves # through the harbor.plugins entry-point group. Nothing else in the test suite @@ -641,12 +648,21 @@ def test_verifier_output_attachment_scopes_steps_and_respects_attachment_mode(tm assert attachment is not None assert warnings == [] - # Raw verifier text is only covered by configured redact_patterns, so the - # default tier ships the structured report and leaves the logs behind. + # Eval runs keep the full verifier evidence by default. The explicitly + # structured tier retains the machine-readable report without raw logs. _, default_summary, default_warnings = _verifier_output_attachment(result, PluginConfig.from_options()) - assert default_summary == {"first": {"ctrf": {"step": "first"}}, "second": {"ctrf": {"step": "second"}}} + assert default_summary == summary assert default_warnings == [] + _, structured_summary, structured_warnings = _verifier_output_attachment( + result, PluginConfig.from_options(attachments="structured") + ) + assert structured_summary == { + "first": {"ctrf": {"step": "first"}}, + "second": {"ctrf": {"step": "second"}}, + } + assert structured_warnings == [] + assert _verifier_output_attachment(result, PluginConfig.from_options(attachments="none")) == (None, None, []) @@ -761,8 +777,7 @@ def _scored_verifier_trial(tmp_path): @pytest.mark.parametrize( ("attachments", "expected_summary"), [ - # Raw verifier text needs the explicit opt-in; the structured report does not. - ("verifier-details", {"ctrf": {"failed": 1}}), + ("structured", {"ctrf": {"failed": 1}}), ("all", {"ctrf": {"failed": 1}, "stdout": "assert 1 == 2\n"}), ], )