diff --git a/examples/harbor/README.md b/examples/harbor/README.md index 834af879..2e6a6129 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,10 @@ 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=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. + +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`: ```dotenv diff --git a/py/src/braintrust/integrations/harbor/atif.py b/py/src/braintrust/integrations/harbor/atif.py index 3b203d71..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" @@ -226,11 +226,6 @@ def _content( complete = False result.append(_bounded(part, config, notes, part_context).value) continue - if len(data) > config.max_attachment_bytes: - complete = False - notes.add(f"{part_context}: image omitted because it exceeds max_attachment_bytes") - result.append({"type": "text", "text": "[image omitted: size limit]"}) - continue result.append( { "type": "image_url", @@ -276,11 +271,14 @@ 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: + 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() @@ -325,11 +323,16 @@ def import_trajectory( 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: + 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",)) 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/config.py b/py/src/braintrust/integrations/harbor/config.py index 97254444..c4a658d9 100644 --- a/py/src/braintrust/integrations/harbor/config.py +++ b/py/src/braintrust/integrations/harbor/config.py @@ -98,11 +98,10 @@ 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_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"): @@ -157,12 +155,10 @@ 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'") - 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..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")) @@ -118,7 +130,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 +140,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 +191,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 29758db8..afca942a 100644 --- a/py/src/braintrust/integrations/harbor/plugin.py +++ b/py/src/braintrust/integrations/harbor/plugin.py @@ -9,6 +9,7 @@ import json import logging import os +import stat from dataclasses import dataclass, field, fields from datetime import datetime from pathlib import Path @@ -26,6 +27,7 @@ reward_details_paths, snapshot_job, trajectory_paths, + verifier_output_paths, ) from .config import _UNSET, PluginConfig from .identity import ( @@ -37,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 ( @@ -158,6 +161,44 @@ 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_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" + 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() + except FileNotFoundError: + return None, None + except OSError as exc: + return None, str(exc) + 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] = [] @@ -170,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: @@ -183,7 +228,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(): @@ -202,16 +246,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}") + 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, @@ -225,50 +264,134 @@ 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 - try: - data = path.read_bytes() - except FileNotFoundError: + 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 - 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") - continue - try: - parsed = json.loads(data) - except json.JSONDecodeError: + parsed, parsed_ok = try_parse_json(data) + if not parsed_ok: 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() - # One serialized payload is bounded by the per-file limit, not the job total. - if len(attachment_data) > config.max_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, - warnings, + return _json_attachment(complete, "details", filename, warnings) + + +@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. 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 + + +_VERIFIER_OUTPUT_FILES = ( + _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), +) + + +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, [] + 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, + redact_patterns=config.redact_patterns, + max_depth=20, + redact_absolute_paths=False, ) + return normalized.value, [*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] = [] + for step_name, verifier_dir in verifier_output_paths(result): + step_output: dict[str, Any] = {} + 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) + 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[output_file.key] = value + if step_output: + outputs.append((step_name, step_output)) + return _json_attachment(outputs, "verifier", "verifier-output.json", warnings) + + +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, + max_bytes=config.max_content_bytes, + redact_patterns=config.redact_patterns, + max_depth=20, + redact_absolute_paths=redact_absolute_paths, + ).value + + +def _serialized_bytes(attachment: Attachment) -> int: + # _json_attachment appends a trailing newline that canonical sizing omits. + return len(attachment.data) - 1 + + +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: @@ -295,9 +418,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, @@ -322,9 +444,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, @@ -626,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( @@ -636,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 @@ -746,7 +868,18 @@ def _sync_final_result(self, result: Any) -> None: 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_output, verifier_warnings = _verifier_evidence(result, self.config) + metadata["harbor"]["warnings"].extend(verifier_warnings) + self._start_phase( + task, + result, + "verification", + "verifier", + trial_id, + root_start, + root_end, + output=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) @@ -816,15 +949,11 @@ def _sync_final_result(self, result: Any) -> None: 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 - 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( @@ -842,6 +971,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..c34e7342 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,18 @@ semantic_agent_config, ) from braintrust.integrations.harbor.plugin import ( + DatasetBinding, HarborPlugin, + Partition, RuntimeState, _artifact_attachments, _attachment, + _read_safe_file, _resolve_project, _seconds, _timing, + _verifier_evidence, + _verifier_output_attachment, ) from braintrust.integrations.harbor.rewards import classify_rewards, validate_classifications from braintrust.integrations.harbor.state import ( @@ -47,13 +58,20 @@ reduce_job, reduce_trial, ) +from braintrust.test_helpers import ( # noqa: F401 + find_span_by_name, + find_spans_by_type, + 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 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]+)") @@ -125,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 @@ -421,6 +446,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")): @@ -441,9 +474,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" @@ -451,13 +482,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 @@ -466,6 +498,397 @@ 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, verifier_dir = _verifier_trial(tmp_path) + # 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( + { + "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(attachments="all", 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, 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(redact_patterns=(r"opaque-secret",)) + ) + + assert attachment is not None + 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", + 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" + if kind == "symlink": + secret = tmp_path / "host-secret" + secret.write_text("must not escape") + stdout.symlink_to(secret) + else: + os.mkfifo(stdout) + + config = PluginConfig.from_options(attachments="all") + attachment, summary, warnings = _verifier_output_attachment(result, config) + + assert attachment is None + assert summary is None + assert warnings == ["test-stdout.txt omitted: unsafe file type"] + + +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") + 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_safe_file(expected) == (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_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")) + + assert summary == { + "first": {"stdout": "first output\n", "ctrf": {"step": "first"}}, + "second": {"stdout": "second output\n", "ctrf": {"step": "second"}}, + } + assert attachment is not None + assert warnings == [] + + # 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 == 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, []) + + +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, 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()) + + assert attachment is not None + assert summary["ctrf"].startswith("[[[") + 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) + + attachment, summary, warnings = _attachment([(None, path)], PluginConfig.from_options()) + + 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", + 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", trials_dir, None, None, (plan,)) + partition = Partition("partition", "experiment", "scope", experiment=experiment) + 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) + + +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 ())] + + +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_bytes(b"assert 1 == 2\n") + (verifier_dir / "ctrf.json").write_text(json.dumps({"failed": 1})) + return result + + +@pytest.mark.parametrize( + ("attachments", "expected_summary"), + [ + ("structured", {"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 = _scored_verifier_trial(tmp_path) + 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) + + verification = find_span_by_name(spans, "verification") + 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. + 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_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. + 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 find_spans_by_type(spans, "score") + 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"}}, + ], + } + ] + } + ) + ) + + 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(): plugin = HarborPlugin(project_name="unused") # Reproduce the ordering that makes this reachable: the runtime is built, then