From ad67f71eae70714a7878b06982c75041814fa8f4 Mon Sep 17 00:00:00 2001 From: appleweiping Date: Sat, 3 Oct 2026 22:51:36 -0500 Subject: [PATCH 1/2] Report completion coverage for legacy assigner runs --- .github/workflows/lite-configs.yml | 15 ++ README.md | 24 +++ src/assigner.py | 2 + src/run_coverage.py | 279 +++++++++++++++++++++++++++++ tests/test_assigner_coverage.py | 85 +++++++++ tests/test_run_coverage.py | 193 ++++++++++++++++++++ 6 files changed, 598 insertions(+) create mode 100644 src/run_coverage.py create mode 100644 tests/test_assigner_coverage.py create mode 100644 tests/test_run_coverage.py diff --git a/.github/workflows/lite-configs.yml b/.github/workflows/lite-configs.yml index a1831e85..b1ccf5ae 100644 --- a/.github/workflows/lite-configs.yml +++ b/.github/workflows/lite-configs.yml @@ -6,6 +6,21 @@ on: branches: [ main ] jobs: + legacy-run-coverage: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ['3.9', '3.12'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install legacy runner core dependencies + run: python -m pip install 'pydantic<2' 'numpy<2' requests tqdm pyyaml + - name: Check coverage report and assigner resume + run: python -m unittest discover -s tests -v + validate: runs-on: ubuntu-latest steps: diff --git a/README.md b/README.md index dce14d55..ac59096d 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,30 @@ If you started the task server with the lite preset, you can also run the lite e python -m src.assigner --config configs/assignments/lite.yaml ``` +### Inspect incomplete legacy runs + +The assigner above writes `sample_manifest.json` for each new agent and task pair, +using the indices returned by its task client. To inspect both complete and partial +legacy runs without changing their scores: + +```bash +python -m src.run_coverage --output outputs --save analysis/coverage.json +``` + +This command needs only Python. It includes directories without `overall.json` +and reports unique terminal samples, pending indices, infrastructure errors, +duplicate records, and malformed log lines. A normal terminal model failure still +counts as an evaluated sample. Coverage is not accuracy; no score is calculated +and existing `overall.json` files are not changed. + +Older runs without a manifest have an unknown expected count. Recorded attempt +counts are log-entry counts, not total executions: the legacy assigner does not +log `NOT_AVAILABLE` callbacks. A resumed run with changed task indices must use a +new output directory. Conflicting terminal records retain the first result for +diagnosis and are not labelled complete. + +This reports the legacy assigner's JSONL format, not AgentRL FC evaluation logs. + ## Next Steps If you wish to launch more tasks or use other models, you can refer to the content diff --git a/src/assigner.py b/src/assigner.py index 3f56568f..edbf665c 100644 --- a/src/assigner.py +++ b/src/assigner.py @@ -15,6 +15,7 @@ from src.client.task import TaskError from .client import TaskClient, AgentClient +from .run_coverage import save_sample_manifest from .configs import ConfigLoader from .typings import AssignmentConfig, SampleIndex, TaskOutput, TaskClientOutput from .utils import ColorMessage @@ -92,6 +93,7 @@ def __init__(self, config: AssignmentConfig, auto_retry: bool = True) -> None: print(ColorMessage.green(f"creating {task} client...")) self.tasks[task] = self.config.definition.task[task].create() self.task_indices[task] = self.tasks[task].get_indices() + save_sample_manifest(self.get_output_dir(agent, task), self.task_indices[task]) self.remaining_tasks[agent][task] = self.task_indices[task].copy() if not os.path.exists(runs_file): continue diff --git a/src/run_coverage.py b/src/run_coverage.py new file mode 100644 index 00000000..db6e3f9b --- /dev/null +++ b/src/run_coverage.py @@ -0,0 +1,279 @@ +"""Report legacy assigner log coverage without calculating benchmark scores.""" + +import argparse +import hashlib +import json +import os +import tempfile +from pathlib import Path + +# SampleStatus values serialized by the legacy TaskClient. +TERMINAL_STATUSES = { + "completed", + "agent context limit", + "agent validation failed", + "agent invalid action", + "task limit reached", + "unknown", + "task error", +} +MANIFEST = "sample_manifest.json" +LOG_FILES = {MANIFEST, "runs.jsonl", "error.jsonl", "overall.json"} + + +def sample_key(index): + if type(index) not in (int, str): + raise ValueError("sample indices must be integers or strings") + return type(index).__name__, index + + +def validate_indices(indices): + if not isinstance(indices, list): + raise TypeError("expected_ids must be a list") + keys = [sample_key(index) for index in indices] + if len(set(keys)) != len(keys): + raise ValueError("expected_ids contains duplicate indices") + return keys + + +def save_sample_manifest(directory, indices): + """Persist the actual get_indices result; refuse changed indices on resume.""" + validate_indices(indices) + directory = Path(directory) + directory.mkdir(parents=True, exist_ok=True) + path = directory / MANIFEST + value = {"schema_version": 1, "expected_ids": indices} + if path.exists(): + existing = json.loads(path.read_text(encoding="utf-8")) + if existing != value: + raise ValueError( + f"Task indices changed since {path} was created; use a new output directory" + ) + return + temporary = None + try: + with tempfile.NamedTemporaryFile( + "w", encoding="utf-8", dir=directory, delete=False + ) as stream: + temporary = stream.name + json.dump(value, stream, ensure_ascii=False) + stream.write("\n") + os.replace(temporary, path) + finally: + if temporary is not None and os.path.exists(temporary): + os.unlink(temporary) + + +def reject_constant(value): + raise ValueError(f"non-finite JSON constant: {value}") + + +def report_task(directory): + directory = Path(directory) + diagnostics = [] + expected = None + manifest_hash = None + manifest_path = directory / MANIFEST + if manifest_path.exists(): + try: + raw = manifest_path.read_bytes() + manifest_hash = hashlib.sha256(raw).hexdigest() + manifest = json.loads(raw, parse_constant=reject_constant) + if ( + not isinstance(manifest, dict) + or type(manifest.get("schema_version")) is not int + or manifest["schema_version"] != 1 + ): + raise ValueError("unsupported manifest schema") + expected = manifest["expected_ids"] + validate_indices(expected) + except (ValueError, TypeError, KeyError, OSError) as exc: + diagnostics.append({"file": MANIFEST, "message": str(exc)}) + expected = None + samples = {} + duplicates = [] + conflicts = [] + for filename in ("runs.jsonl", "error.jsonl"): + path = directory / filename + if not path.exists(): + continue + try: + with path.open(encoding="utf-8") as stream: + for line_number, line in enumerate(stream, 1): + if not line.strip(): + continue + try: + record = json.loads(line, parse_constant=reject_constant) + if not isinstance(record, dict): + raise TypeError("log record must be an object") + key = sample_key(record["index"]) + error = record.get("error") + output = record.get("output") + if error is not None: + if not isinstance(error, str) or not error: + raise ValueError( + "error must be a nonempty string or null" + ) + else: + if not isinstance(output, dict): + raise ValueError("missing TaskOutput") + status = output.get("status") + if status not in TERMINAL_STATUSES | {"running"}: + raise ValueError("unknown TaskOutput status") + if ( + output.get("index") is not None + and sample_key(output["index"]) != key + ): + raise ValueError( + "top-level and TaskOutput indices disagree" + ) + if (filename == "error.jsonl") != (error is not None): + raise ValueError("record stored in the wrong log file") + except (ValueError, KeyError, TypeError) as exc: + diagnostics.append( + {"file": filename, "line": line_number, "message": str(exc)} + ) + continue + sample = samples.setdefault( + key, + { + "index": record["index"], + "recorded_attempt_count": 0, + "infrastructure_error_count": 0, + "terminal_status": None, + }, + ) + sample["recorded_attempt_count"] += 1 + if error is not None: + sample["infrastructure_error_count"] += 1 + elif status in TERMINAL_STATUSES: + output_digest = hashlib.sha256( + json.dumps(output, sort_keys=True).encode() + ).hexdigest() + if "first_output_digest" in sample: + duplicates.append(record["index"]) + if output_digest != sample["first_output_digest"]: + conflicts.append(record["index"]) + else: + # Same first valid terminal record policy as assigner resume. + sample["first_output_digest"] = output_digest + sample["terminal_status"] = status + except (OSError, UnicodeError) as exc: + diagnostics.append({"file": filename, "message": str(exc)}) + expected_keys = None if expected is None else set(validate_indices(expected)) + completed_keys = { + key for key, sample in samples.items() if sample["terminal_status"] is not None + } + unexpected = ( + [] + if expected_keys is None + else [s["index"] for key, s in samples.items() if key not in expected_keys] + ) + completed = ( + completed_keys if expected_keys is None else completed_keys & expected_keys + ) + pending = ( + None + if expected is None + else [i for i in expected if sample_key(i) not in completed] + ) + infra_pending = [ + s["index"] + for key, s in samples.items() + if s["infrastructure_error_count"] + and key not in completed_keys + and (expected_keys is None or key in expected_keys) + ] + overall_exists = (directory / "overall.json").is_file() + if overall_exists: + try: + overall = json.loads( + (directory / "overall.json").read_text(encoding="utf-8"), + parse_constant=reject_constant, + ) + if not isinstance(overall, dict): + raise TypeError("overall must be an object") + except (ValueError, TypeError, OSError) as exc: + diagnostics.append({"file": "overall.json", "message": str(exc)}) + complete = ( + None + if expected is None + else not pending and not unexpected and not conflicts and not diagnostics + ) + if not overall_exists: + score_status = "missing_overall" + elif expected is None: + score_status = "unknown_coverage" + elif diagnostics or conflicts or unexpected: + score_status = "inconsistent_logs" + elif not complete: + score_status = "incomplete" + else: + score_status = "complete" + return { + "directory": str(directory.resolve()), + "agent": directory.parent.name, + "task": directory.name, + "manifest_sha256": manifest_hash, + "expected_ids": expected, + "expected_count": None if expected is None else len(expected), + "terminal_count": len(completed), + "coverage_fraction": None if not expected else len(completed) / len(expected), + "pending_ids": pending, + "infrastructure_error_ids": infra_pending, + "unexpected_ids": unexpected, + "duplicate_terminal_ids": duplicates, + "conflicting_terminal_ids": conflicts, + "diagnostics": diagnostics, + "is_complete": complete, + "overall_exists": overall_exists, + "score_status": score_status, + "samples": [ + {k: v for k, v in sample.items() if k != "first_output_digest"} + for sample in samples.values() + ], + } + + +def analyze_coverage(output): + tasks = [] + for root, _, files in os.walk(output): + if LOG_FILES.intersection(files): + tasks.append(report_task(root)) + return { + "schema_version": 1, + "tasks": tasks, + "attempt_count_note": "Counts valid recorded log entries, including duplicates. NOT_AVAILABLE callbacks are not logged by the legacy assigner. This is not a total execution-attempt count.", + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("-o", "--output", default="outputs") + parser.add_argument("-s", "--save", default="analysis/coverage.json") + args = parser.parse_args() + if not Path(args.output).is_dir(): + parser.error("output directory does not exist") + report = analyze_coverage(args.output) + target = Path(args.save) + # A report must never overwrite a source log, manifest, or benchmark score. + if target.name in LOG_FILES or target.resolve().is_relative_to( + Path(args.output).resolve() + ): + parser.error("save the coverage report outside the source output directory") + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text( + json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" + ) + for task in report["tasks"]: + denominator = ( + task["expected_count"] if task["expected_count"] is not None else "unknown" + ) + print( + f"{task['agent']} {task['task']}: {task['terminal_count']} of {denominator} terminal; {task['score_status']}" + ) + print(f"Coverage report saved to {target}") + + +if __name__ == "__main__": + main() diff --git a/tests/test_assigner_coverage.py b/tests/test_assigner_coverage.py new file mode 100644 index 00000000..248bbe36 --- /dev/null +++ b/tests/test_assigner_coverage.py @@ -0,0 +1,85 @@ +"""Exercise the real assigner initializer with in-process task/agent clients.""" + +import importlib +import importlib.util +import json +import sys +import tempfile +import types +import unittest +from pathlib import Path +from unittest.mock import patch + +from src.run_coverage import MANIFEST, report_task + + +class AssignerCoverageTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + from src.typings import AssignmentConfig, InstanceFactory + + cls.config_class = AssignmentConfig + cls.factory_class = InstanceFactory + # Do not import src.client.agents (optional fastchat/model dependencies). + package = types.ModuleType("src.client") + package.__path__ = [str(Path(__file__).resolve().parents[1] / "src" / "client")] + with patch.dict(sys.modules, {"src.client": package}): + package.TaskClient = importlib.import_module("src.client.task").TaskClient + package.AgentClient = importlib.import_module( + "src.client.agent" + ).AgentClient + spec = importlib.util.spec_from_file_location( + "src.assigner_coverage_test", + Path(__file__).resolve().parents[1] / "src" / "assigner.py", + ) + cls.module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(cls.module) + + def test_actual_indices_are_saved_before_run_and_preserved_on_resume(self): + with tempfile.TemporaryDirectory() as temp: + output = Path(temp) / "outputs" + config = self.config_class.parse_obj( + { + "assignments": [{"agent": "model", "task": "dbbench"}], + "concurrency": {"agent": {"model": 1}, "task": {"dbbench": 1}}, + "definition": { + "agent": {"model": {"module": "fake.Agent"}}, + "task": {"dbbench": {"module": "fake.Task"}}, + }, + "output": str(output), + } + ) + fake_task = types.SimpleNamespace(get_indices=lambda: [2, 5, 8]) + + def create(factory): + return fake_task if factory.module == "fake.Task" else object() + + with patch.object(self.factory_class, "create", create): + assigner = self.module.Assigner(config) + directory = output / "model" / "dbbench" + manifest_before = (directory / MANIFEST).read_bytes() + self.assertEqual(json.loads(manifest_before)["expected_ids"], [2, 5, 8]) + self.assertEqual( + assigner.remaining_tasks["model"]["dbbench"], [2, 5, 8] + ) + record = { + "index": 5, + "error": None, + "output": {"index": 5, "status": "completed", "result": 0}, + "time": {}, + } + (directory / "runs.jsonl").write_text( + json.dumps(record) + "\n", encoding="utf-8" + ) + resumed = self.module.Assigner(config) + self.assertEqual(resumed.remaining_tasks["model"]["dbbench"], [2, 8]) + self.assertEqual((directory / MANIFEST).read_bytes(), manifest_before) + self.assertEqual(report_task(directory)["terminal_count"], 1) + fake_task.get_indices = lambda: [2, 5, 8, 9] + with self.assertRaisesRegex(ValueError, "Task indices changed"): + self.module.Assigner(config) + self.assertEqual((directory / MANIFEST).read_bytes(), manifest_before) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_run_coverage.py b/tests/test_run_coverage.py new file mode 100644 index 00000000..69cb503e --- /dev/null +++ b/tests/test_run_coverage.py @@ -0,0 +1,193 @@ +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +from src.run_coverage import ( + MANIFEST, + analyze_coverage, + report_task, + save_sample_manifest, +) + + +class CoverageTest(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.task = self.root / "outputs" / "run" / "model" / "dbbench" + self.task.mkdir(parents=True) + + def log(self, index, status="completed", error=None, result=None): + record = { + "index": index, + "error": error, + "output": None + if error + else {"index": index, "status": status, "result": result}, + "time": {"timestamp": 1}, + } + with (self.task / ("error.jsonl" if error else "runs.jsonl")).open( + "a", encoding="utf-8" + ) as stream: + stream.write(json.dumps(record) + "\n") + + def test_partial_ten_samples_visible_without_overall(self): + save_sample_manifest(self.task, list(range(10))) + for i in range(8): + self.log(i) + for i in (8, 9): + self.log(i, error="NETWORK_ERROR") + report = analyze_coverage(self.root / "outputs")["tasks"][0] + self.assertEqual(report["terminal_count"], 8) + self.assertEqual(report["coverage_fraction"], 0.8) + self.assertEqual(report["infrastructure_error_ids"], [8, 9]) + self.assertEqual(report["pending_ids"], [8, 9]) + self.assertFalse(report["is_complete"]) + self.assertEqual(report["score_status"], "missing_overall") + + def test_complete_run_preserves_existing_score_bytes(self): + save_sample_manifest(self.task, [0, 1]) + self.log(0) + self.log(1, status="agent invalid action") + score = b'{"custom": {"accuracy": 0.5}}\n' + (self.task / "overall.json").write_bytes(score) + report = report_task(self.task) + self.assertTrue(report["is_complete"]) + self.assertEqual(report["score_status"], "complete") + self.assertEqual((self.task / "overall.json").read_bytes(), score) + + def test_retries_then_success_only_one_terminal_sample(self): + save_sample_manifest(self.task, ["one"]) + self.log("one", error="NETWORK_ERROR") + self.log("one", error="START_FAILED") + self.log("one", result=0) + report = report_task(self.task) + self.assertEqual(report["terminal_count"], 1) + self.assertEqual(report["samples"][0]["recorded_attempt_count"], 3) + self.assertEqual(report["infrastructure_error_ids"], []) + self.assertTrue(report["is_complete"]) + + def test_duplicate_resume_does_not_double_count(self): + save_sample_manifest(self.task, [0]) + self.log(0, result=0) + self.log(0, result=0) + report = report_task(self.task) + self.assertEqual(report["terminal_count"], 1) + self.assertEqual(report["duplicate_terminal_ids"], [0]) + self.assertTrue(report["is_complete"]) + + def test_conflicting_terminal_records_do_not_choose_best_result(self): + save_sample_manifest(self.task, [0]) + self.log(0, status="agent invalid action", result=0) + self.log(0, result=1) + report = report_task(self.task) + self.assertEqual( + report["samples"][0]["terminal_status"], "agent invalid action" + ) + self.assertEqual(report["conflicting_terminal_ids"], [0]) + self.assertFalse(report["is_complete"]) + + def test_malformed_tail_is_reported(self): + save_sample_manifest(self.task, [0, 1]) + self.log(0) + with (self.task / "runs.jsonl").open("a") as stream: + stream.write('{"index":1') + report = report_task(self.task) + self.assertEqual(report["terminal_count"], 1) + self.assertEqual(report["diagnostics"][0]["line"], 2) + self.assertFalse(report["is_complete"]) + + def test_missing_or_invalid_manifest_never_guesses_denominator(self): + self.log(99) + for content in (None, '{"schema_version":1,"expected_ids":[true]}', "{}"): + if content is not None: + (self.task / MANIFEST).write_text(content) + report = report_task(self.task) + self.assertIsNone(report["expected_count"]) + self.assertIsNone(report["coverage_fraction"]) + self.assertIsNone(report["is_complete"]) + + def test_integer_and_string_indices_are_distinct(self): + save_sample_manifest(self.task, [0, "0"]) + self.log(0) + report = report_task(self.task) + self.assertEqual(report["pending_ids"], ["0"]) + self.assertEqual(report["coverage_fraction"], 0.5) + + def test_malformed_overall_and_invalid_log_schema_are_diagnosed(self): + save_sample_manifest(self.task, [0]) + self.log(0) + (self.task / "overall.json").write_text("[]") + with (self.task / "runs.jsonl").open("a") as stream: + stream.write('{"index":false,"output":{"status":"completed"}}\n') + stream.write('{"index":1,"output":{"status":"completed","result":NaN}}\n') + report = report_task(self.task) + self.assertEqual(len(report["diagnostics"]), 3) + self.assertEqual(report["score_status"], "inconsistent_logs") + self.assertFalse(report["is_complete"]) + + def test_unstarted_manifest_and_empty_plan(self): + save_sample_manifest(self.task, [0]) + report = report_task(self.task) + self.assertEqual(report["coverage_fraction"], 0.0) + self.assertFalse(report["is_complete"]) + empty = self.task.parent / "empty" + save_sample_manifest(empty, []) + self.assertTrue(report_task(empty)["is_complete"]) + self.assertIsNone(report_task(empty)["coverage_fraction"]) + + def test_running_and_unexpected_results_cannot_complete_plan(self): + save_sample_manifest(self.task, [0]) + self.log(0, status="running") + self.log(1) + report = report_task(self.task) + self.assertEqual(report["terminal_count"], 0) + self.assertEqual(report["unexpected_ids"], [1]) + self.assertFalse(report["is_complete"]) + + def test_manifest_is_stable_and_changed_resume_rejected(self): + save_sample_manifest(self.task, [0]) + before = (self.task / MANIFEST).read_bytes() + save_sample_manifest(self.task, [0]) + self.assertEqual((self.task / MANIFEST).read_bytes(), before) + with self.assertRaises(ValueError): + save_sample_manifest(self.task, [1]) + with self.assertRaises(ValueError): + save_sample_manifest(self.task, [0, 0]) + + def test_public_cli_writes_sidecar_and_protects_sources(self): + save_sample_manifest(self.task, [0, 1]) + self.log(0) + target = self.root / "coverage.json" + command = [ + sys.executable, + "-m", + "src.run_coverage", + "--output", + str(self.root / "outputs"), + ] + result = subprocess.run( + command + ["--save", str(target)], + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn("1 of 2 terminal", result.stdout) + self.assertEqual(json.loads(target.read_text())["tasks"][0]["pending_ids"], [1]) + before = (self.task / "runs.jsonl").read_bytes() + result = subprocess.run( + command + ["--save", str(self.task / "runs.jsonl")], + capture_output=True, + check=False, + ) + self.assertNotEqual(result.returncode, 0) + self.assertEqual((self.task / "runs.jsonl").read_bytes(), before) + + +if __name__ == "__main__": + unittest.main() From ef93ff8db41aca48dda1413626509ed0af1b5620 Mon Sep 17 00:00:00 2001 From: appleweiping Date: Sat, 3 Oct 2026 22:53:33 -0500 Subject: [PATCH 2/2] Reject malformed manifests before resuming a run --- src/run_coverage.py | 11 ++++++++++- tests/test_run_coverage.py | 3 +++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/run_coverage.py b/src/run_coverage.py index db6e3f9b..a3e618a2 100644 --- a/src/run_coverage.py +++ b/src/run_coverage.py @@ -44,7 +44,16 @@ def save_sample_manifest(directory, indices): path = directory / MANIFEST value = {"schema_version": 1, "expected_ids": indices} if path.exists(): - existing = json.loads(path.read_text(encoding="utf-8")) + existing = json.loads( + path.read_text(encoding="utf-8"), parse_constant=reject_constant + ) + if ( + not isinstance(existing, dict) + or type(existing.get("schema_version")) is not int + or existing["schema_version"] != 1 + ): + raise ValueError(f"Invalid manifest at {path}") + validate_indices(existing.get("expected_ids")) if existing != value: raise ValueError( f"Task indices changed since {path} was created; use a new output directory" diff --git a/tests/test_run_coverage.py b/tests/test_run_coverage.py index 69cb503e..a63df543 100644 --- a/tests/test_run_coverage.py +++ b/tests/test_run_coverage.py @@ -158,6 +158,9 @@ def test_manifest_is_stable_and_changed_resume_rejected(self): save_sample_manifest(self.task, [1]) with self.assertRaises(ValueError): save_sample_manifest(self.task, [0, 0]) + (self.task / MANIFEST).write_text('{"schema_version":1,"expected_ids":[false]}') + with self.assertRaises(ValueError): + save_sample_manifest(self.task, [0]) def test_public_cli_writes_sidecar_and_protects_sources(self): save_sample_manifest(self.task, [0, 1])