diff --git a/CHANGELOG.md b/CHANGELOG.md index 0815d4f..b055041 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,3 +29,7 @@ - Add `gpt-5.6-luna` and `gpt-5.6-terra` to the native evaluation catalog. - Report repair-overlay task provenance and original-versus-repaired score sensitivity when the original job is available. +- Export native run metadata into aggregate task rows and add task-matrix + variance analysis with paired harness/reasoning diagnostics and SVG plots. +- Add private S3 trace-bundle publishing through the standard AWS credential + chain with SHA-256 verification and resumable object checks. diff --git a/README.md b/README.md index 04cdb61..798e239 100644 --- a/README.md +++ b/README.md @@ -435,6 +435,53 @@ clawbench dynamics-report \ # results/gptoss_dynamics/dynamics.json ``` +### Analyzing a native ShellBench matrix + +Native aggregation now preserves the task revision, harness, provider model, +reasoning effort, and parent-run score/eligibility on every task row. Build +task-level variance diagnostics from those normalized exports: + +```bash +python scripts/native_eval/aggregate.py runs-full-YYYYMMDD results/native_summaries + +# Install the optional plotting dependency, or pass --no-plots. +pip install -e '.[analysis]' +clawbench task-analysis \ + --summaries-dir results/native_summaries \ + --output-dir results/task_analysis +``` + +The report writes task-cell summaries, paired harness and reasoning deltas, +task variance rankings, a machine-readable JSON summary, `ANALYSIS.md`, and +SVG box plots. Different task revisions and task counts remain separate +datasets and are never silently pooled. + +### Publishing trace bundles to private S3 + +Trace bundles can be uploaded without placing static credentials in command +arguments or repository files. The command uses the standard AWS credential +chain and does not expose access-key options: + +```bash +pip install -e '.[s3]' + +export AWS_PROFILE="your-private-profile" +export SHELLBENCH_TRACE_BUCKET="your-private-bucket" + +clawbench trace-upload \ + --run-dir runs-full-YYYYMMDD \ + --prefix runs-full-YYYYMMDD +``` + +The uploader selects final and checkpoint artifact archives, trajectory +validation reports, trace-gap audits, and the run index. It sets no public ACL, +uses S3-managed encryption, and records SHA-256 metadata; the destination +bucket policy remains authoritative. Existing objects with matching size and +digest are skipped, and a verified `S3_UPLOAD_MANIFEST.json` is written locally +and uploaded last. + +Use `--dry-run` to inspect the selected files without authenticating. + ### Running locally with small models (Ollama) A single consumer GPU running an open-weight model is enough to develop plugin profiles and validate algorithmic ideas — no API keys or cloud spend required. @@ -518,6 +565,8 @@ clawbench/ │ ├── dynamics.py # Trajectory metrics + sensitivity analysis │ ├── dynamics_archive.py # Cached-run loading + offline report assembly │ ├── dynamics_plots.py # Offline dynamics visualizations +│ ├── task_analysis.py # Native matrix task variance + paired deltas +│ ├── trace_upload.py # Private S3 trace publishing + verification │ └── cli.py # CLI entry points │ ├── tasks-public/ # Core v1 PUBLIC release (19 tasks) diff --git a/clawbench/cli.py b/clawbench/cli.py index 413ebf2..82f45b9 100644 --- a/clawbench/cli.py +++ b/clawbench/cli.py @@ -254,6 +254,111 @@ def dynamics_report( click.echo(f"Saved {len(plots)} plots to {output_dir}/") +@cli.command("task-analysis") +@click.option( + "--summaries-dir", + type=click.Path(exists=True, file_okay=False, path_type=Path), + required=True, + help="Directory containing native aggregate_results.csv and per_task_results.csv.", +) +@click.option( + "--output-dir", + type=click.Path(path_type=Path), + default=Path("results/task_analysis"), + show_default=True, + help="Directory where task diagnostics, Markdown, and plots will be written.", +) +@click.option( + "--no-plots", + is_flag=True, + help="Write CSV, JSON, and Markdown outputs without rendering SVG plots.", +) +def task_analysis(summaries_dir: Path, output_dir: Path, no_plots: bool) -> None: + """Analyze task variance across harness, model, and reasoning cells.""" + from clawbench.task_analysis import analyze_task_matrix + + try: + report = analyze_task_matrix( + summaries_dir, + output_dir, + generate_plots=not no_plots, + ) + except (FileNotFoundError, RuntimeError, ValueError) as exc: + raise click.ClickException(str(exc)) from exc + + click.echo( + f"Analyzed {report['source']['eligible_run_count']} eligible runs " + f"across {len(report['datasets'])} task dataset(s)" + ) + click.echo(f"Task analysis saved to {output_dir}") + + +@cli.command("trace-upload") +@click.option( + "--run-dir", + type=click.Path(exists=True, file_okay=False, path_type=Path), + required=True, + help="Native run directory containing raw/, summaries/, and manifests/.", +) +@click.option( + "--bucket", + envvar="SHELLBENCH_TRACE_BUCKET", + default="", + help="Private destination bucket. May be set with SHELLBENCH_TRACE_BUCKET.", +) +@click.option( + "--prefix", + default=None, + help="Object prefix. Defaults to the run directory name.", +) +@click.option( + "--workers", + type=click.IntRange(1, 32), + default=4, + show_default=True, + help="Number of files to upload concurrently.", +) +@click.option( + "--dry-run", + is_flag=True, + help="Print the bundle plan without authenticating or uploading.", +) +def trace_upload( + run_dir: Path, + bucket: str, + prefix: str | None, + workers: int, + dry_run: bool, +) -> None: + """Publish native trace archives to private S3 with hash verification.""" + from clawbench.trace_upload import trace_bundle_plan, upload_trace_bundle + + try: + if dry_run: + plan = trace_bundle_plan(run_dir) + click.echo(json.dumps(plan, indent=2, sort_keys=True)) + return + if not bucket: + raise ValueError( + "set --bucket or SHELLBENCH_TRACE_BUCKET before uploading" + ) + report = upload_trace_bundle( + run_dir, + bucket=bucket, + prefix=prefix, + workers=workers, + progress=click.echo, + ) + except (FileNotFoundError, RuntimeError, ValueError) as exc: + raise click.ClickException(str(exc)) from exc + + click.echo( + f"Verified {report['file_count']} trace objects " + f"({report['uploaded_count']} uploaded, {report['skipped_count']} unchanged)" + ) + click.echo(f"Upload manifest saved to {report['manifest_path']}") + + def _write_dynamics_report( task_runs: dict[str, list], output_dir: Path, diff --git a/clawbench/task_analysis.py b/clawbench/task_analysis.py new file mode 100644 index 0000000..5ff1377 --- /dev/null +++ b/clawbench/task_analysis.py @@ -0,0 +1,686 @@ +"""Task-level analysis for native ShellBench matrix exports.""" + +from __future__ import annotations + +import csv +import itertools +import json +import re +import statistics +from collections import defaultdict +from pathlib import Path +from typing import Any, Iterable, Sequence + + +RUN_REQUIRED_FIELDS = { + "run_label", + "harness", + "model_slug", + "reasoning_effort", + "task_revision", + "expected_task_count", + "score", + "eligible", +} +TASK_REQUIRED_FIELDS = { + "run_label", + "task_name", + "classification", + "reward", +} +REASONING_ORDER = ("low", "medium", "high", "xhigh") + +CELL_FIELDS = ( + "dataset_key", + "task_revision", + "suite_task_count", + "harness", + "model_slug", + "model_id", + "reasoning_effort", + "task_name", + "repetitions", + "mean_reward", + "reward_stdev", + "min_reward", + "max_reward", + "nonzero_repetitions", + "exact_passes", +) +TASK_VARIANCE_FIELDS = ( + "dataset_key", + "task_revision", + "suite_task_count", + "task_name", + "cell_count", + "nonzero_cells", + "mean_cell_reward", + "cell_reward_stdev", + "min_cell_reward", + "max_cell_reward", + "range_across_cells", + "best_cell", + "worst_cell", +) +HARNESS_DELTA_FIELDS = ( + "dataset_key", + "task_revision", + "suite_task_count", + "model_slug", + "reasoning_effort", + "task_name", + "left_harness", + "right_harness", + "left_reward", + "right_reward", + "delta", +) +REASONING_DELTA_FIELDS = ( + "dataset_key", + "task_revision", + "suite_task_count", + "harness", + "model_slug", + "task_name", + "lower_reasoning", + "higher_reasoning", + "lower_reward", + "higher_reward", + "delta", +) + + +def _read_csv(path: Path, required: set[str]) -> list[dict[str, str]]: + if not path.is_file(): + raise FileNotFoundError(f"required analysis input does not exist: {path}") + with path.open(newline="", encoding="utf-8") as handle: + reader = csv.DictReader(handle) + fields = set(reader.fieldnames or ()) + missing = sorted(required - fields) + if missing: + raise ValueError(f"{path.name} is missing required columns: {', '.join(missing)}") + return list(reader) + + +def _write_csv( + path: Path, + fieldnames: Sequence[str], + rows: Iterable[dict[str, Any]], +) -> None: + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=fieldnames, extrasaction="ignore") + writer.writeheader() + writer.writerows(rows) + + +def _number(value: Any, default: float = 0.0) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _integer(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _boolean(value: Any) -> bool: + return str(value).strip().lower() in {"1", "true", "yes"} + + +def _stdev(values: Sequence[float]) -> float: + return statistics.stdev(values) if len(values) > 1 else 0.0 + + +def _dataset_key(run: dict[str, str]) -> str: + revision = run["task_revision"].strip() or "unknown-revision" + count = _integer(run["expected_task_count"]) + return f"{revision}:{count}" + + +def _cell_label(row: dict[str, Any]) -> str: + return "/".join( + ( + str(row["harness"]), + str(row["model_slug"]), + str(row["reasoning_effort"]), + ) + ) + + +def _reasoning_rank(value: str) -> tuple[int, str]: + try: + return REASONING_ORDER.index(value), value + except ValueError: + return len(REASONING_ORDER), value + + +def _task_cells( + runs: Sequence[dict[str, str]], + tasks: Sequence[dict[str, str]], +) -> tuple[list[dict[str, Any]], set[str]]: + eligible_runs = {row["run_label"]: row for row in runs if _boolean(row["eligible"])} + grouped: dict[tuple[str, str, str, str, str], list[float]] = defaultdict(list) + model_ids: dict[tuple[str, str, str, str, str], set[str]] = defaultdict(set) + + for task in tasks: + run = eligible_runs.get(task["run_label"]) + if run is None: + continue + key = ( + _dataset_key(run), + run["harness"], + run["model_slug"], + run["reasoning_effort"], + task["task_name"], + ) + grouped[key].append(_number(task["reward"])) + if run.get("model_id"): + model_ids[key].add(run["model_id"]) + + rows: list[dict[str, Any]] = [] + for key, rewards in sorted(grouped.items()): + dataset_key, harness, model_slug, reasoning, task_name = key + revision, count = dataset_key.rsplit(":", 1) + ids = sorted(model_ids[key]) + rows.append( + { + "dataset_key": dataset_key, + "task_revision": revision, + "suite_task_count": int(count), + "harness": harness, + "model_slug": model_slug, + "model_id": ",".join(ids), + "reasoning_effort": reasoning, + "task_name": task_name, + "repetitions": len(rewards), + "mean_reward": statistics.mean(rewards), + "reward_stdev": _stdev(rewards), + "min_reward": min(rewards), + "max_reward": max(rewards), + "nonzero_repetitions": sum(value > 0 for value in rewards), + "exact_passes": sum(value >= 1 for value in rewards), + } + ) + return rows, set(eligible_runs) + + +def _task_variance(cells: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + grouped: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) + for cell in cells: + grouped[(str(cell["dataset_key"]), str(cell["task_name"]))].append(cell) + + rows = [] + for (dataset_key, task_name), task_cells in grouped.items(): + values = [float(cell["mean_reward"]) for cell in task_cells] + best_reward = max(values) + worst_reward = min(values) + best_cells = sorted( + _cell_label(cell) + for cell in task_cells + if float(cell["mean_reward"]) == best_reward + ) + worst_cells = sorted( + _cell_label(cell) + for cell in task_cells + if float(cell["mean_reward"]) == worst_reward + ) + revision, count = dataset_key.rsplit(":", 1) + rows.append( + { + "dataset_key": dataset_key, + "task_revision": revision, + "suite_task_count": int(count), + "task_name": task_name, + "cell_count": len(values), + "nonzero_cells": sum(value > 0 for value in values), + "mean_cell_reward": statistics.mean(values), + "cell_reward_stdev": _stdev(values), + "min_cell_reward": min(values), + "max_cell_reward": max(values), + "range_across_cells": max(values) - min(values), + "best_cell": ";".join(best_cells), + "worst_cell": ";".join(worst_cells), + } + ) + return sorted( + rows, + key=lambda row: ( + str(row["dataset_key"]), + -float(row["range_across_cells"]), + -float(row["cell_reward_stdev"]), + str(row["task_name"]), + ), + ) + + +def _harness_deltas(cells: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + grouped: dict[tuple[str, str, str, str], dict[str, float]] = defaultdict(dict) + for cell in cells: + key = ( + str(cell["dataset_key"]), + str(cell["model_slug"]), + str(cell["reasoning_effort"]), + str(cell["task_name"]), + ) + grouped[key][str(cell["harness"])] = float(cell["mean_reward"]) + + rows = [] + for (dataset_key, model, reasoning, task_name), values in sorted(grouped.items()): + revision, count = dataset_key.rsplit(":", 1) + for left, right in itertools.combinations(sorted(values), 2): + rows.append( + { + "dataset_key": dataset_key, + "task_revision": revision, + "suite_task_count": int(count), + "model_slug": model, + "reasoning_effort": reasoning, + "task_name": task_name, + "left_harness": left, + "right_harness": right, + "left_reward": values[left], + "right_reward": values[right], + "delta": values[left] - values[right], + } + ) + return rows + + +def _reasoning_deltas(cells: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + grouped: dict[tuple[str, str, str, str], dict[str, float]] = defaultdict(dict) + for cell in cells: + key = ( + str(cell["dataset_key"]), + str(cell["harness"]), + str(cell["model_slug"]), + str(cell["task_name"]), + ) + grouped[key][str(cell["reasoning_effort"])] = float(cell["mean_reward"]) + + rows = [] + for (dataset_key, harness, model, task_name), values in sorted(grouped.items()): + revision, count = dataset_key.rsplit(":", 1) + reasoning = sorted(values, key=_reasoning_rank) + for lower, higher in zip(reasoning, reasoning[1:]): + rows.append( + { + "dataset_key": dataset_key, + "task_revision": revision, + "suite_task_count": int(count), + "harness": harness, + "model_slug": model, + "task_name": task_name, + "lower_reasoning": lower, + "higher_reasoning": higher, + "lower_reward": values[lower], + "higher_reward": values[higher], + "delta": values[higher] - values[lower], + } + ) + return rows + + +def _safe_name(value: str) -> str: + return re.sub(r"[^a-zA-Z0-9._-]+", "-", value).strip("-").lower() + + +def _short_cell_list(value: Any, limit: int = 3) -> str: + cells = str(value).split(";") + if len(cells) <= limit: + return "; ".join(cells) + return f"{'; '.join(cells[:limit])}; +{len(cells) - limit} tied" + + +def _plot_box( + values_by_label: dict[str, list[float]], + *, + title: str, + ylabel: str, + path: Path, +) -> None: + import matplotlib + + matplotlib.use("Agg") + import matplotlib.pyplot as plt + + labels = sorted(values_by_label) + values = [values_by_label[label] for label in labels] + width = min(24, max(9, len(labels) * 0.55)) + fig, ax = plt.subplots(figsize=(width, 6)) + ax.boxplot(values, tick_labels=labels, showmeans=True, patch_artist=True) + ax.axhline(0, color="#374151", linewidth=0.8) + ax.set_title(title) + ax.set_ylabel(ylabel) + ax.tick_params(axis="x", rotation=45) + fig.tight_layout() + fig.savefig(path, format="svg", bbox_inches="tight") + plt.close(fig) + + +def _generate_plots( + runs: Sequence[dict[str, str]], + cells: Sequence[dict[str, Any]], + harness_deltas: Sequence[dict[str, Any]], + reasoning_deltas: Sequence[dict[str, Any]], + output_dir: Path, +) -> list[Path]: + try: + import matplotlib # noqa: F401 + except ImportError as exc: + raise RuntimeError( + "plot generation requires matplotlib; install the project with .[analysis]" + ) from exc + + paths = [] + datasets = sorted({str(cell["dataset_key"]) for cell in cells}) + for dataset_key in datasets: + prefix = _safe_name(dataset_key) + run_scores: dict[str, list[float]] = defaultdict(list) + for run in runs: + if _boolean(run["eligible"]) and _dataset_key(run) == dataset_key: + label = "/".join( + (run["harness"], run["model_slug"], run["reasoning_effort"]) + ) + run_scores[label].append(_number(run["score"])) + path = output_dir / f"{prefix}-run-score-distributions.svg" + _plot_box( + run_scores, + title=f"Run score distributions ({dataset_key})", + ylabel="Score", + path=path, + ) + paths.append(path) + + harness_values: dict[str, list[float]] = defaultdict(list) + for row in harness_deltas: + if row["dataset_key"] == dataset_key: + label = f"{row['left_harness']} - {row['right_harness']}" + harness_values[label].append(float(row["delta"])) + if harness_values: + path = output_dir / f"{prefix}-harness-task-deltas.svg" + _plot_box( + harness_values, + title=f"Paired harness task deltas ({dataset_key})", + ylabel="Left minus right task reward", + path=path, + ) + paths.append(path) + + reasoning_values: dict[str, list[float]] = defaultdict(list) + for row in reasoning_deltas: + if row["dataset_key"] == dataset_key: + label = f"{row['lower_reasoning']} -> {row['higher_reasoning']}" + reasoning_values[label].append(float(row["delta"])) + if reasoning_values: + path = output_dir / f"{prefix}-reasoning-task-deltas.svg" + _plot_box( + reasoning_values, + title=f"Paired reasoning task deltas ({dataset_key})", + ylabel="Higher minus lower task reward", + path=path, + ) + paths.append(path) + + task_values: dict[str, list[float]] = defaultdict(list) + dataset_cells = [cell for cell in cells if cell["dataset_key"] == dataset_key] + ranges: dict[str, float] = defaultdict(float) + for cell in dataset_cells: + task_values[str(cell["task_name"])].append(float(cell["mean_reward"])) + for task_name, values in task_values.items(): + ranges[task_name] = max(values) - min(values) + top_tasks = { + task: task_values[task] + for task in sorted(ranges, key=lambda task: (-ranges[task], task))[:20] + } + if top_tasks: + path = output_dir / f"{prefix}-highest-variance-tasks.svg" + _plot_box( + top_tasks, + title=f"Highest-variance task cell distributions ({dataset_key})", + ylabel="Mean reward per experiment cell", + path=path, + ) + paths.append(path) + return paths + + +def _dataset_summaries( + runs: Sequence[dict[str, str]], + cells: Sequence[dict[str, Any]], + task_variance: Sequence[dict[str, Any]], + harness_deltas: Sequence[dict[str, Any]], + reasoning_deltas: Sequence[dict[str, Any]], +) -> list[dict[str, Any]]: + summaries = [] + for dataset_key in sorted({_dataset_key(run) for run in runs}): + dataset_runs = [run for run in runs if _dataset_key(run) == dataset_key] + dataset_cells = [cell for cell in cells if cell["dataset_key"] == dataset_key] + variance_rows = [ + row for row in task_variance if row["dataset_key"] == dataset_key + ] + harness_rows = [ + row for row in harness_deltas if row["dataset_key"] == dataset_key + ] + reasoning_rows = [ + row for row in reasoning_deltas if row["dataset_key"] == dataset_key + ] + revision, count = dataset_key.rsplit(":", 1) + summaries.append( + { + "dataset_key": dataset_key, + "task_revision": revision, + "suite_task_count": int(count), + "total_runs": len(dataset_runs), + "eligible_runs": sum(_boolean(run["eligible"]) for run in dataset_runs), + "excluded_runs": sum(not _boolean(run["eligible"]) for run in dataset_runs), + "experiment_cells": len( + { + ( + cell["harness"], + cell["model_slug"], + cell["reasoning_effort"], + ) + for cell in dataset_cells + } + ), + "tasks": len({cell["task_name"] for cell in dataset_cells}), + "high_variance_tasks": sum( + float(row["range_across_cells"]) >= 0.5 for row in variance_rows + ), + "harness_task_comparisons": len(harness_rows), + "large_harness_deltas": sum( + abs(float(row["delta"])) >= 0.5 for row in harness_rows + ), + "reasoning_task_comparisons": len(reasoning_rows), + "reasoning_improvements": sum( + float(row["delta"]) > 0 for row in reasoning_rows + ), + "reasoning_regressions": sum( + float(row["delta"]) < 0 for row in reasoning_rows + ), + } + ) + return summaries + + +def _write_markdown( + path: Path, + datasets: Sequence[dict[str, Any]], + task_variance: Sequence[dict[str, Any]], + plot_paths: Sequence[Path], +) -> None: + lines = [ + "# ShellBench task matrix analysis", + "", + "## TL;DR", + "", + ] + for dataset in datasets: + lines.append( + "- `{dataset_key}`: {eligible_runs}/{total_runs} eligible runs, " + "{experiment_cells} experiment cells, {tasks} tasks, " + "{high_variance_tasks} tasks with a cross-cell range of at least 0.5, and " + "{large_harness_deltas}/{harness_task_comparisons} paired harness " + "comparisons with an absolute delta of at least 0.5.".format(**dataset) + ) + lines.append( + " Reasoning changes improved {reasoning_improvements} matched task cells " + "and regressed {reasoning_regressions}; treat reasoning as an experimental " + "condition, not a monotonic quality ladder.".format(**dataset) + ) + if len(datasets) > 1: + lines.append( + "- Multiple task revisions are present. They are analyzed separately and " + "must not be pooled into one leaderboard." + ) + + lines.extend( + [ + "", + "## Datasets", + "", + "| Revision | Tasks | Eligible runs | Cells | High-variance tasks |", + "| --- | ---: | ---: | ---: | ---: |", + ] + ) + for dataset in datasets: + lines.append( + "| `{task_revision}` | {suite_task_count} | {eligible_runs}/{total_runs} | " + "{experiment_cells} | {high_variance_tasks} |".format(**dataset) + ) + + lines.extend(["", "## Highest-variance tasks", ""]) + for dataset in datasets: + lines.extend( + [ + f"### `{dataset['dataset_key']}`", + "", + "| Task | Cells | Mean | Stdev | Range | Best cell | Worst cell |", + "| --- | ---: | ---: | ---: | ---: | --- | --- |", + ] + ) + rows = [ + row + for row in task_variance + if row["dataset_key"] == dataset["dataset_key"] + ][:20] + for row in rows: + lines.append( + "| {task} | {cells} | {mean:.4f} | {stdev:.4f} | {span:.4f} | " + "{best} | {worst} |".format( + task=row["task_name"], + cells=row["cell_count"], + mean=row["mean_cell_reward"], + stdev=row["cell_reward_stdev"], + span=row["range_across_cells"], + best=_short_cell_list(row["best_cell"]), + worst=_short_cell_list(row["worst_cell"]), + ) + ) + lines.append("") + + if plot_paths: + lines.extend(["", "## Plots", ""]) + for plot_path in plot_paths: + lines.append(f"![{plot_path.stem}]({plot_path.name})") + lines.append("") + + lines.extend( + [ + "## Appendix: how to read this report", + "", + "| Term | Meaning |", + "| --- | --- |", + "| Dataset | One exact task revision and task count. Different datasets are never pooled. |", + "| Experiment cell | One harness, model, and reasoning-effort combination. |", + "| Run | One independent full-suite repetition within an experiment cell. |", + "| Task cell score | Mean reward for one task across eligible repetitions of one experiment cell. |", + "| Harness delta | Paired task-cell score for the left harness minus the right harness, holding model and reasoning fixed. |", + "| Reasoning delta | Paired higher-effort task-cell score minus the adjacent lower-effort score, holding harness and model fixed. |", + "| High variance | A task whose best and worst experiment-cell means differ by at least 0.5. This is a triage signal, not proof of a harness defect. |", + "| Eligible run | A complete run accepted by native aggregation after coverage, infrastructure, model-identity, and trajectory checks. |", + "", + "Raw failed and excluded runs remain in the source aggregate exports; this report " + "uses eligible runs for score distributions and paired task comparisons.", + ] + ) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def analyze_task_matrix( + summaries_dir: str | Path, + output_dir: str | Path, + *, + generate_plots: bool = True, +) -> dict[str, Any]: + """Build task-level diagnostics from native aggregate CSV exports.""" + + source = Path(summaries_dir) + destination = Path(output_dir) + destination.mkdir(parents=True, exist_ok=True) + runs = _read_csv(source / "aggregate_results.csv", RUN_REQUIRED_FIELDS) + tasks = _read_csv(source / "per_task_results.csv", TASK_REQUIRED_FIELDS) + + cells, eligible_run_labels = _task_cells(runs, tasks) + if not cells: + raise ValueError("no task rows belong to eligible runs") + task_variance = _task_variance(cells) + harness_deltas = _harness_deltas(cells) + reasoning_deltas = _reasoning_deltas(cells) + plots = ( + _generate_plots(runs, cells, harness_deltas, reasoning_deltas, destination) + if generate_plots + else [] + ) + datasets = _dataset_summaries( + runs, + cells, + task_variance, + harness_deltas, + reasoning_deltas, + ) + + _write_csv(destination / "task_cell_summary.csv", CELL_FIELDS, cells) + _write_csv( + destination / "task_variance_diagnostics.csv", + TASK_VARIANCE_FIELDS, + task_variance, + ) + _write_csv( + destination / "harness_task_deltas.csv", + HARNESS_DELTA_FIELDS, + harness_deltas, + ) + _write_csv( + destination / "reasoning_task_deltas.csv", + REASONING_DELTA_FIELDS, + reasoning_deltas, + ) + report = { + "schema_version": 1, + "source": { + "summaries_dir": source.name, + "run_count": len(runs), + "task_row_count": len(tasks), + "eligible_run_count": len(eligible_run_labels), + }, + "datasets": datasets, + "outputs": { + "task_cell_summary": "task_cell_summary.csv", + "task_variance_diagnostics": "task_variance_diagnostics.csv", + "harness_task_deltas": "harness_task_deltas.csv", + "reasoning_task_deltas": "reasoning_task_deltas.csv", + "markdown": "ANALYSIS.md", + "plots": [path.name for path in plots], + }, + } + (destination / "analysis_summary.json").write_text( + json.dumps(report, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + _write_markdown(destination / "ANALYSIS.md", datasets, task_variance, plots) + return report diff --git a/clawbench/trace_upload.py b/clawbench/trace_upload.py new file mode 100644 index 0000000..3e98c00 --- /dev/null +++ b/clawbench/trace_upload.py @@ -0,0 +1,263 @@ +"""Verified private S3 publishing for native ShellBench trace bundles.""" + +from __future__ import annotations + +import hashlib +import json +import threading +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable + + +TRACE_AUDIT_FILES = ( + "TRACE_COVERAGE_AUDIT.md", + "trace_gaps.csv", +) +TRACE_VALIDATION_FILES = ( + "trajectory_validation.json", + "trajectory_validation.full.json", +) + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + while chunk := handle.read(8 * 1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + +def collect_trace_bundle(run_dir: str | Path) -> list[tuple[Path, str]]: + """Collect immutable trace archives and their audit metadata.""" + + root = Path(run_dir) + if not root.is_dir(): + raise FileNotFoundError(f"run directory does not exist: {root}") + + selected: dict[str, Path] = {} + raw_dir = root / "raw" + if raw_dir.is_dir(): + for path in sorted(raw_dir.glob("*-artifacts.tar.gz")): + selected[f"raw/{path.name}"] = path + if not any(key.startswith("raw/") for key in selected): + raise ValueError(f"no trace artifact archives found under {raw_dir}") + + summaries_dir = root / "summaries" + for name in TRACE_AUDIT_FILES: + path = summaries_dir / name + if path.is_file(): + selected[f"summaries/{name}"] = path + if summaries_dir.is_dir(): + for path in sorted(summaries_dir.glob("*/trajectory_validation*.json")): + if path.name in TRACE_VALIDATION_FILES: + relative = path.relative_to(root).as_posix() + selected[relative] = path + + run_index = root / "manifests" / "run_index.json" + if run_index.is_file(): + selected["manifests/run_index.json"] = run_index + return [(selected[key], key) for key in sorted(selected)] + + +def trace_bundle_plan(run_dir: str | Path) -> dict[str, Any]: + files = collect_trace_bundle(run_dir) + return { + "source_run": Path(run_dir).name, + "file_count": len(files), + "archive_count": sum(key.startswith("raw/") for _path, key in files), + "total_bytes": sum(path.stat().st_size for path, _key in files), + "files": [ + {"relative_key": key, "size": path.stat().st_size} + for path, key in files + ], + } + + +def _not_found(exc: Exception) -> bool: + response = getattr(exc, "response", {}) + error = response.get("Error", {}) if isinstance(response, dict) else {} + return str(error.get("Code") or "") in {"404", "NoSuchKey", "NotFound"} + + +def _object_matches(client: Any, bucket: str, key: str, size: int, digest: str) -> bool: + try: + head = client.head_object(Bucket=bucket, Key=key) + except Exception as exc: + if _not_found(exc): + return False + raise + return ( + int(head["ContentLength"]) == size + and head.get("Metadata", {}).get("sha256") == digest + ) + + +def _build_client() -> tuple[Any, Any]: + try: + import boto3 + from boto3.s3.transfer import TransferConfig + from botocore.config import Config + except ImportError as exc: + raise RuntimeError( + "S3 trace upload requires boto3; install the project with .[s3]" + ) from exc + + session = boto3.Session() + client = session.client( + "s3", + config=Config( + retries={"max_attempts": 10, "mode": "adaptive"}, + max_pool_connections=32, + ), + ) + transfer = TransferConfig( + multipart_threshold=64 * 1024 * 1024, + multipart_chunksize=64 * 1024 * 1024, + max_concurrency=4, + use_threads=True, + ) + return client, transfer + + +def upload_trace_bundle( + run_dir: str | Path, + *, + bucket: str, + prefix: str | None = None, + workers: int = 4, + client: Any | None = None, + transfer_config: Any | None = None, + progress: Callable[[str], None] | None = None, +) -> dict[str, Any]: + """Upload and verify a private trace bundle using the AWS credential chain.""" + + root = Path(run_dir) + files = collect_trace_bundle(root) + if not bucket.strip(): + raise ValueError("bucket must not be empty") + if workers < 1: + raise ValueError("workers must be at least 1") + object_prefix = (prefix or root.name).strip("/") + if not object_prefix: + raise ValueError("prefix must not be empty") + + if client is None: + client, transfer_config = _build_client() + client.head_bucket(Bucket=bucket) + + records = [] + for path, relative_key in files: + records.append( + { + "path": path, + "relative_key": relative_key, + "size": path.stat().st_size, + "sha256": _sha256(path), + } + ) + + lock = threading.Lock() + completed = 0 + + def upload(record: dict[str, Any]) -> dict[str, Any]: + nonlocal completed + key = f"{object_prefix}/{record['relative_key']}" + status = "skipped" + if not _object_matches( + client, + bucket, + key, + int(record["size"]), + str(record["sha256"]), + ): + kwargs = { + "ExtraArgs": { + "ServerSideEncryption": "AES256", + "Metadata": {"sha256": str(record["sha256"])}, + } + } + if transfer_config is not None: + kwargs["Config"] = transfer_config + client.upload_file(str(record["path"]), bucket, key, **kwargs) + status = "uploaded" + + if not _object_matches( + client, + bucket, + key, + int(record["size"]), + str(record["sha256"]), + ): + raise RuntimeError(f"uploaded object failed verification: {key}") + + with lock: + completed += 1 + if progress is not None: + progress(f"{completed}/{len(records)} {status}: {record['relative_key']}") + return { + "key": key, + "size": record["size"], + "sha256": record["sha256"], + "status": status, + } + + uploaded = [] + with ThreadPoolExecutor(max_workers=workers) as executor: + futures = [executor.submit(upload, record) for record in records] + for future in as_completed(futures): + uploaded.append(future.result()) + uploaded.sort(key=lambda item: str(item["key"])) + + manifest = { + "schema_version": 1, + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "bucket": bucket, + "prefix": object_prefix, + "source_run": root.name, + "file_count": len(uploaded), + "archive_count": sum( + str(item["key"]).startswith(f"{object_prefix}/raw/") + for item in uploaded + ), + "total_bytes": sum(int(item["size"]) for item in uploaded), + "files": uploaded, + } + manifest_path = root / "summaries" / "S3_UPLOAD_MANIFEST.json" + manifest_path.parent.mkdir(parents=True, exist_ok=True) + manifest_path.write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + manifest_body = manifest_path.read_bytes() + manifest_digest = hashlib.sha256(manifest_body).hexdigest() + manifest_key = f"{object_prefix}/S3_UPLOAD_MANIFEST.json" + client.put_object( + Bucket=bucket, + Key=manifest_key, + Body=manifest_body, + ServerSideEncryption="AES256", + Metadata={"sha256": manifest_digest}, + ContentType="application/json", + ) + if not _object_matches( + client, + bucket, + manifest_key, + len(manifest_body), + manifest_digest, + ): + raise RuntimeError("upload manifest failed verification") + + return { + "bucket": bucket, + "prefix": object_prefix, + "file_count": len(uploaded), + "archive_count": manifest["archive_count"], + "total_bytes": manifest["total_bytes"], + "uploaded_count": sum(item["status"] == "uploaded" for item in uploaded), + "skipped_count": sum(item["status"] == "skipped" for item in uploaded), + "manifest_path": str(manifest_path), + "manifest_key": manifest_key, + } diff --git a/pyproject.toml b/pyproject.toml index 3af6729..31e9385 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,6 +37,12 @@ dev = [ mlflow = [ "mlflow>=3.14.0,<4", ] +analysis = [ + "matplotlib>=3.9,<4", +] +s3 = [ + "boto3>=1.40,<2", +] hermes = [ "hermes-agent @ git+https://github.com/NousResearch/hermes-agent.git@main", ] diff --git a/scripts/native_eval/aggregate.py b/scripts/native_eval/aggregate.py index d316090..c17e5dc 100644 --- a/scripts/native_eval/aggregate.py +++ b/scripts/native_eval/aggregate.py @@ -86,6 +86,22 @@ "run_label", "pair_label", "repetition", + "harness", + "harness_version", + "model_slug", + "model_id", + "model_provider", + "reasoning_effort", + "judge_model_id", + "judge_reasoning_effort", + "task_revision", + "task_suite", + "run_expected_task_count", + "run_score", + "run_coverage", + "run_exact_passes", + "run_eligible", + "run_exclusion_reason", "task_name", "task_path", "trial_name", @@ -124,6 +140,16 @@ "run_label", "pair_label", "repetition", + "harness", + "harness_version", + "model_slug", + "model_id", + "model_provider", + "reasoning_effort", + "judge_model_id", + "judge_reasoning_effort", + "task_revision", + "task_suite", "expected_task_count", "result_file_count", "valid_result_count", @@ -437,10 +463,28 @@ def _pair_label(manifest: dict[str, Any], run_label: str) -> str: harness = manifest.get("harness") model_slug = manifest.get("model_slug") if harness and model_slug: + reasoning_effort = manifest.get("reasoning_effort") + if reasoning_effort: + return f"{harness}-{model_slug}-{reasoning_effort}" return f"{harness}-{model_slug}" return _derive_pair_label(run_label) +def _experiment_metadata(manifest: dict[str, Any]) -> dict[str, Any]: + return { + "harness": manifest.get("harness") or "", + "harness_version": manifest.get("harness_version") or "", + "model_slug": manifest.get("model_slug") or "", + "model_id": manifest.get("model_id") or manifest.get("provider_model_id") or "", + "model_provider": manifest.get("model_provider") or "", + "reasoning_effort": manifest.get("reasoning_effort") or "", + "judge_model_id": manifest.get("judge_model_id") or "", + "judge_reasoning_effort": manifest.get("judge_reasoning_effort") or "", + "task_revision": manifest.get("public_tasks_commit") or "", + "task_suite": manifest.get("task_suite") or "", + } + + def _repetition(manifest: dict[str, Any], run_label: str) -> int | None: value = _manifest_value( manifest, @@ -782,6 +826,9 @@ def _load_run( ) ) + metadata = _experiment_metadata(manifest) + for row in rows: + row.update(metadata) rows.sort(key=lambda row: (str(row["task_name"]), str(row["trial_name"]))) summary = _summarize_run( run_label=run_label, @@ -791,6 +838,16 @@ def _load_run( rows=rows, manifest=manifest, ) + run_context = { + "run_expected_task_count": summary["expected_task_count"], + "run_score": summary["score"], + "run_coverage": summary["coverage"], + "run_exact_passes": summary["exact_passes"], + "run_eligible": summary["eligible"], + "run_exclusion_reason": summary["exclusion_reason"], + } + for row in rows: + row.update(run_context) return summary, rows, manifest @@ -944,6 +1001,7 @@ def _summarize_run( "run_label": run_label, "pair_label": pair_label, "repetition": repetition, + **_experiment_metadata(manifest), "expected_task_count": expected_count, "result_file_count": result_file_count, "valid_result_count": valid_result_count, diff --git a/tests/test_native_eval_aggregate.py b/tests/test_native_eval_aggregate.py index c1e26d1..67e8379 100644 --- a/tests/test_native_eval_aggregate.py +++ b/tests/test_native_eval_aggregate.py @@ -210,6 +210,54 @@ def test_aggregate_uses_task_path_for_canonical_task_name(tmp_path: Path): assert row["task_path"] == "/benchmark/tasks/canonical-task" +def test_aggregate_exports_experiment_metadata_and_reasoning_pair_label(tmp_path: Path): + jobs_root = tmp_path / "native" + summaries_dir = tmp_path / "summaries" + _write_run( + jobs_root, + "openclaw-model-medium-r1", + expected_task_count=1, + results=[_result("task-a", reward=1.0)], + pair_label=None, + manifest_extra={ + "harness_version": "2026.7.1", + "model_id": "provider/model", + "model_provider": "provider", + "reasoning_effort": "medium", + "judge_model_id": "judge/model", + "judge_reasoning_effort": "high", + "public_tasks_commit": "abc123", + "task_suite": "combined tasks/tasks", + }, + ) + + aggregate(jobs_root, summaries_dir) + + with (summaries_dir / "aggregate_results.csv").open(newline="") as handle: + run = next(csv.DictReader(handle)) + with (summaries_dir / "per_task_results.csv").open(newline="") as handle: + task = next(csv.DictReader(handle)) + + assert run["pair_label"] == "openclaw-model-medium" + for row in (run, task): + assert row["harness"] == "openclaw" + assert row["harness_version"] == "2026.7.1" + assert row["model_slug"] == "model" + assert row["model_id"] == "provider/model" + assert row["model_provider"] == "provider" + assert row["reasoning_effort"] == "medium" + assert row["judge_model_id"] == "judge/model" + assert row["judge_reasoning_effort"] == "high" + assert row["task_revision"] == "abc123" + assert row["task_suite"] == "combined tasks/tasks" + assert task["run_expected_task_count"] == "1" + assert task["run_score"] == "1.0" + assert task["run_coverage"] == "1.0" + assert task["run_exact_passes"] == "1" + assert task["run_eligible"] == "True" + assert task["run_exclusion_reason"] == "" + + def test_pair_aggregates_exclude_incomplete_and_infra_dominated_runs(tmp_path: Path): jobs_root = tmp_path / "native" summaries_dir = tmp_path / "summaries" diff --git a/tests/test_task_analysis.py b/tests/test_task_analysis.py new file mode 100644 index 0000000..cf5b397 --- /dev/null +++ b/tests/test_task_analysis.py @@ -0,0 +1,195 @@ +import csv +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from clawbench.cli import cli +from clawbench.task_analysis import analyze_task_matrix + + +RUN_FIELDS = ( + "run_label", + "harness", + "model_slug", + "model_id", + "reasoning_effort", + "task_revision", + "expected_task_count", + "score", + "eligible", +) +TASK_FIELDS = ("run_label", "task_name", "classification", "reward") + + +def _write_csv(path: Path, fields: tuple[str, ...], rows: list[dict]) -> None: + with path.open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=fields) + writer.writeheader() + writer.writerows(rows) + + +def _matrix_fixture(tmp_path: Path) -> Path: + summaries = tmp_path / "summaries" + summaries.mkdir() + runs = [] + tasks = [] + + conditions = [ + ("openclaw", "low", {"task-a": 1.0, "task-b": 0.0}), + ("hermes", "low", {"task-a": 0.0, "task-b": 0.5}), + ("openclaw", "medium", {"task-a": 1.0, "task-b": 0.5}), + ] + for harness, reasoning, rewards in conditions: + for repetition in (1, 2): + run_label = f"{harness}-model-{reasoning}-r{repetition}" + runs.append( + { + "run_label": run_label, + "harness": harness, + "model_slug": "model", + "model_id": "provider/model", + "reasoning_effort": reasoning, + "task_revision": "revision-a", + "expected_task_count": 2, + "score": sum(rewards.values()) / 2, + "eligible": "true", + } + ) + for task_name, reward in rewards.items(): + tasks.append( + { + "run_label": run_label, + "task_name": task_name, + "classification": "pass" if reward >= 1 else "partial", + "reward": reward, + } + ) + + runs.append( + { + "run_label": "openclaw-other-low-r1", + "harness": "openclaw", + "model_slug": "other", + "model_id": "provider/other", + "reasoning_effort": "low", + "task_revision": "revision-b", + "expected_task_count": 1, + "score": 1.0, + "eligible": "true", + } + ) + tasks.append( + { + "run_label": "openclaw-other-low-r1", + "task_name": "task-a", + "classification": "pass", + "reward": 1.0, + } + ) + runs.append( + { + "run_label": "excluded-run", + "harness": "codex", + "model_slug": "model", + "model_id": "provider/model", + "reasoning_effort": "low", + "task_revision": "revision-a", + "expected_task_count": 2, + "score": 1.0, + "eligible": "false", + } + ) + tasks.extend( + [ + { + "run_label": "excluded-run", + "task_name": "task-a", + "classification": "pass", + "reward": 1.0, + }, + { + "run_label": "excluded-run", + "task_name": "task-b", + "classification": "pass", + "reward": 1.0, + }, + ] + ) + _write_csv(summaries / "aggregate_results.csv", RUN_FIELDS, runs) + _write_csv(summaries / "per_task_results.csv", TASK_FIELDS, tasks) + return summaries + + +def test_analyze_task_matrix_builds_separate_dataset_diagnostics(tmp_path: Path): + summaries = _matrix_fixture(tmp_path) + output = tmp_path / "analysis" + + report = analyze_task_matrix(summaries, output, generate_plots=False) + + assert report["source"]["eligible_run_count"] == 7 + assert len(report["datasets"]) == 2 + revision_a = next( + dataset + for dataset in report["datasets"] + if dataset["task_revision"] == "revision-a" + ) + assert revision_a["experiment_cells"] == 3 + assert revision_a["tasks"] == 2 + assert revision_a["excluded_runs"] == 1 + assert revision_a["high_variance_tasks"] == 2 + assert revision_a["reasoning_improvements"] == 1 + assert revision_a["reasoning_regressions"] == 0 + + with (output / "task_variance_diagnostics.csv").open(newline="") as handle: + variance = list(csv.DictReader(handle)) + task_a = next( + row + for row in variance + if row["task_revision"] == "revision-a" and row["task_name"] == "task-a" + ) + assert float(task_a["range_across_cells"]) == 1.0 + assert task_a["best_cell"] == "openclaw/model/low;openclaw/model/medium" + assert task_a["worst_cell"] == "hermes/model/low" + + with (output / "harness_task_deltas.csv").open(newline="") as handle: + harness_deltas = list(csv.DictReader(handle)) + assert sorted(float(row["delta"]) for row in harness_deltas) == [-1.0, 0.5] + + markdown = (output / "ANALYSIS.md").read_text(encoding="utf-8") + assert "Multiple task revisions are present" in markdown + assert "Reasoning effort" not in markdown + assert (output / "analysis_summary.json").is_file() + + +def test_task_analysis_cli_supports_no_plots(tmp_path: Path): + summaries = _matrix_fixture(tmp_path) + output = tmp_path / "analysis" + + result = CliRunner().invoke( + cli, + [ + "task-analysis", + "--summaries-dir", + str(summaries), + "--output-dir", + str(output), + "--no-plots", + ], + ) + + assert result.exit_code == 0, result.output + assert "Analyzed 7 eligible runs across 2 task dataset(s)" in result.output + assert list(output.glob("*.svg")) == [] + + +def test_analyze_task_matrix_renders_svg_box_plots(tmp_path: Path): + pytest.importorskip("matplotlib") + summaries = _matrix_fixture(tmp_path) + output = tmp_path / "analysis" + + report = analyze_task_matrix(summaries, output, generate_plots=True) + + plot_names = report["outputs"]["plots"] + assert plot_names + assert all((output / name).read_text(encoding="utf-8").startswith(" None: + self.objects: dict[tuple[str, str], dict] = {} + self.lock = threading.Lock() + + def head_bucket(self, *, Bucket: str) -> None: + assert Bucket + + def head_object(self, *, Bucket: str, Key: str) -> dict: + with self.lock: + try: + stored = self.objects[(Bucket, Key)] + except KeyError as exc: + raise MissingObjectError from exc + return { + "ContentLength": len(stored["body"]), + "Metadata": stored["metadata"], + } + + def upload_file( + self, + filename: str, + bucket: str, + key: str, + *, + ExtraArgs: dict, + Config=None, + ) -> None: + del Config + with self.lock: + self.objects[(bucket, key)] = { + "body": Path(filename).read_bytes(), + "metadata": ExtraArgs["Metadata"], + "encryption": ExtraArgs["ServerSideEncryption"], + } + + def put_object( + self, + *, + Bucket: str, + Key: str, + Body: bytes, + ServerSideEncryption: str, + Metadata: dict, + ContentType: str, + ) -> None: + with self.lock: + self.objects[(Bucket, Key)] = { + "body": Body, + "metadata": Metadata, + "encryption": ServerSideEncryption, + "content_type": ContentType, + } + + +def _trace_fixture(tmp_path: Path) -> Path: + root = tmp_path / "runs-full-example" + raw = root / "raw" + raw.mkdir(parents=True) + (raw / "run-a-checkpoint-0001-artifacts.tar.gz").write_bytes(b"checkpoint") + (raw / "run-a-final-artifacts.tar.gz").write_bytes(b"final") + (raw / "ignore.txt").write_text("not uploaded", encoding="utf-8") + + summaries = root / "summaries" + (summaries / "low").mkdir(parents=True) + (summaries / "TRACE_COVERAGE_AUDIT.md").write_text("audit\n", encoding="utf-8") + (summaries / "trace_gaps.csv").write_text("run_label\n", encoding="utf-8") + (summaries / "low" / "trajectory_validation.full.json").write_text( + "{}\n", + encoding="utf-8", + ) + (summaries / "unrelated.json").write_text("{}\n", encoding="utf-8") + + manifests = root / "manifests" + manifests.mkdir() + (manifests / "run_index.json").write_text('{"runs":[]}\n', encoding="utf-8") + return root + + +def test_collect_trace_bundle_selects_archives_and_audit_metadata(tmp_path: Path): + root = _trace_fixture(tmp_path) + + selected = collect_trace_bundle(root) + + assert [key for _path, key in selected] == [ + "manifests/run_index.json", + "raw/run-a-checkpoint-0001-artifacts.tar.gz", + "raw/run-a-final-artifacts.tar.gz", + "summaries/TRACE_COVERAGE_AUDIT.md", + "summaries/low/trajectory_validation.full.json", + "summaries/trace_gaps.csv", + ] + + +def test_upload_trace_bundle_verifies_and_skips_existing_objects(tmp_path: Path): + root = _trace_fixture(tmp_path) + client = FakeS3Client() + + first = upload_trace_bundle( + root, + bucket="private-bucket", + prefix="matrix/example", + workers=2, + client=client, + ) + second = upload_trace_bundle( + root, + bucket="private-bucket", + prefix="matrix/example", + workers=2, + client=client, + ) + + assert first["file_count"] == 6 + assert first["archive_count"] == 2 + assert first["uploaded_count"] == 6 + assert first["skipped_count"] == 0 + assert second["uploaded_count"] == 0 + assert second["skipped_count"] == 6 + assert all( + stored["encryption"] == "AES256" + for stored in client.objects.values() + ) + + manifest_path = root / "summaries" / "S3_UPLOAD_MANIFEST.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert manifest["bucket"] == "private-bucket" + assert manifest["prefix"] == "matrix/example" + assert manifest["file_count"] == 6 + assert ("private-bucket", "matrix/example/S3_UPLOAD_MANIFEST.json") in client.objects + + +def test_trace_upload_cli_dry_run_does_not_require_aws_authentication(tmp_path: Path): + root = _trace_fixture(tmp_path) + + result = CliRunner().invoke( + cli, + [ + "trace-upload", + "--run-dir", + str(root), + "--dry-run", + ], + ) + + assert result.exit_code == 0, result.output + plan = json.loads(result.output) + assert plan["file_count"] == 6 + assert plan["archive_count"] == 2 + + +def test_trace_upload_cli_exposes_no_static_credential_options(): + result = CliRunner().invoke(cli, ["trace-upload", "--help"]) + + assert result.exit_code == 0 + assert "access-key" not in result.output.lower() + assert "secret-key" not in result.output.lower()