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