From 9071bcc0bc8c657f7f063bc510bc72b16b664b0c Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 15:59:55 -0400 Subject: [PATCH] Add comparative Milestone 5 task evidence --- tests/test_milestone5_task_evidence.py | 154 +++++ tools/milestone5_task_evidence.py | 886 +++++++++++++++++++++++++ 2 files changed, 1040 insertions(+) create mode 100644 tests/test_milestone5_task_evidence.py create mode 100644 tools/milestone5_task_evidence.py diff --git a/tests/test_milestone5_task_evidence.py b/tests/test_milestone5_task_evidence.py new file mode 100644 index 0000000..c313577 --- /dev/null +++ b/tests/test_milestone5_task_evidence.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from typing import cast + +from tools.milestone5_task_evidence import ( + MAX_GRAPH_INSPECTED_BYTES, + MAX_SOURCE_INSPECTED_BYTES, + MAX_TASK_RESPONSE_BYTES, + SMOKE_SOURCE_COUNT, + answer_key, + build_task_evidence, + fixture_tasks, +) + +ROOT = Path(__file__).resolve().parents[1] + + +class Milestone5TaskEvidenceTests(unittest.TestCase): + def test_answer_keys_are_fixed_and_independent_of_both_workflows(self) -> None: + tasks = fixture_tasks(SMOKE_SOURCE_COUNT) + + self.assertEqual( + ("evidence.component_000", "evidence.component_008"), + answer_key(tasks[0], SMOKE_SOURCE_COUNT), + ) + self.assertEqual( + ("evidence.component_014", "evidence.component_015"), + answer_key(tasks[1], SMOKE_SOURCE_COUNT), + ) + self.assertEqual( + ( + "evidence.component_017", + "evidence.component_008", + "evidence.component_004", + "evidence.component_002", + "evidence.component_001", + ), + answer_key(tasks[2], SMOKE_SOURCE_COUNT), + ) + + def test_smoke_evidence_is_exact_provenanced_and_bounded(self) -> None: + with tempfile.TemporaryDirectory() as directory: + evidence = build_task_evidence( + Path(directory).resolve(), + source_count=SMOKE_SOURCE_COUNT, + samples=1, + ) + + fixture = cast(dict[str, object], evidence["fixture"]) + self.assertFalse(fixture["external_projects"]) + self.assertFalse(fixture["self_hosting"]) + self.assertFalse(fixture["production_bindings"]) + tasks = cast(list[dict[str, object]], evidence["tasks"]) + self.assertEqual(3, len(tasks)) + for task in tasks: + answer = cast(list[str], task["answer_key"]) + workflows = cast(dict[str, dict[str, object]], task["workflows"]) + self.assertTrue(answer) + for name, workflow in workflows.items(): + with self.subTest(task=task["task"], workflow=name): + self.assertTrue(workflow["correct"]) + result = cast(dict[str, object], workflow["result"]) + self.assertEqual(answer, result["answer"]) + self.assertTrue(result["provenance"]) + self.assertLessEqual( + cast(int, workflow["response_bytes"]), + MAX_TASK_RESPONSE_BYTES, + ) + inspected_limit = ( + MAX_GRAPH_INSPECTED_BYTES + if name == "graph_assisted" + else MAX_SOURCE_INSPECTED_BYTES + ) + self.assertLessEqual( + cast(int, workflow["inspected_bytes"]), + inspected_limit, + ) + comparison = cast(dict[str, object], task["comparison"]) + self.assertTrue(comparison["both_exact"]) + + def test_semantic_evidence_is_repeatable_while_timings_remain_measurements(self) -> None: + hashes: list[str] = [] + for _ in range(2): + with tempfile.TemporaryDirectory() as directory: + evidence = build_task_evidence( + Path(directory).resolve(), + source_count=SMOKE_SOURCE_COUNT, + samples=1, + ) + summary = cast(dict[str, object], evidence["summary"]) + hashes.append(cast(str, summary["semantic_evidence_sha256"])) + + self.assertEqual(hashes[0], hashes[1]) + + def test_source_only_scope_reflects_each_task_algorithm(self) -> None: + with tempfile.TemporaryDirectory() as directory: + evidence = build_task_evidence( + Path(directory).resolve(), + source_count=SMOKE_SOURCE_COUNT, + samples=1, + ) + + tasks = cast(list[dict[str, object]], evidence["tasks"]) + direct = cast( + dict[str, object], + cast(dict[str, object], tasks[0]["workflows"])["source_only"], + ) + impact = cast( + dict[str, object], + cast(dict[str, object], tasks[1]["workflows"])["source_only"], + ) + path = cast( + dict[str, object], + cast(dict[str, object], tasks[2]["workflows"])["source_only"], + ) + self.assertLess( + cast(int, direct["inspected_bytes"]), + cast(int, impact["inspected_bytes"]), + ) + self.assertEqual(impact["inspected_bytes"], path["inspected_bytes"]) + + def test_smoke_cli_emits_the_same_machine_readable_report_it_writes(self) -> None: + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "evidence.json" + completed = subprocess.run( + [ + sys.executable, + "tools/milestone5_task_evidence.py", + "--mode", + "smoke", + "--output", + str(output), + ], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ) + report = json.loads(completed.stdout) + + self.assertEqual(completed.stdout, output.read_text(encoding="utf-8")) + self.assertEqual("docforge2_milestone5_representative_tasks", report["benchmark"]) + self.assertEqual("smoke", report["mode"]) + self.assertEqual(SMOKE_SOURCE_COUNT, report["fixture"]["source_count"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/milestone5_task_evidence.py b/tools/milestone5_task_evidence.py new file mode 100644 index 0000000..9a871b0 --- /dev/null +++ b/tools/milestone5_task_evidence.py @@ -0,0 +1,886 @@ +"""Reproducible comparative task evidence for the Milestone 5 release gate.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import math +import platform +import statistics +import subprocess +import sys +import tempfile +import time +from collections import deque +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from pathlib import Path +from typing import Literal, cast + +from docforge.adapter_sdk import AdapterProject +from docforge.adapters.python import PythonReferenceAdapter +from docforge.index import ProjectIndex + +ROOT = Path(__file__).resolve().parents[1] +SMOKE_SOURCE_COUNT = 18 +FULL_SOURCE_COUNT = 72 +SMOKE_SAMPLES = 1 +FULL_SAMPLES = 5 +MAX_TASK_RESPONSE_BYTES = 64 * 1024 +MAX_GRAPH_INSPECTED_BYTES = 512 * 1024 +MAX_SOURCE_INSPECTED_BYTES = 4 * 1024 * 1024 +MAX_REPORT_BYTES = 1024 * 1024 +PREPARATION_LIMIT_MS = 30_000.0 +TASK_P95_LIMIT_MS = 5_000.0 +PADDING_ROWS = 24 + +WorkflowName = Literal["graph_assisted", "source_only"] +TaskKind = Literal["direct_dependencies", "bounded_impact", "dependency_path"] + + +class TaskEvidenceError(RuntimeError): + """The maintained comparative evidence gate was not satisfied.""" + + +@dataclass(frozen=True) +class TaskSpec: + task_id: str + kind: TaskKind + focus_index: int + target_index: int | None + depth: int + prompt: str + + def as_dict(self) -> dict[str, object]: + return { + "task_id": self.task_id, + "kind": self.kind, + "focus": _module_name(self.focus_index), + "target": None if self.target_index is None else _module_name(self.target_index), + "depth": self.depth, + "prompt": self.prompt, + } + + +@dataclass(frozen=True) +class WorkflowResult: + answer: tuple[str, ...] + provenance: tuple[Mapping[str, object], ...] + inspected_bytes: int + + def payload(self) -> dict[str, object]: + return { + "answer": list(self.answer), + "provenance": [dict(item) for item in self.provenance], + } + + +@dataclass(frozen=True) +class _GraphContext: + index: ProjectIndex + module_node_ids: Mapping[str, str] + node_modules: Mapping[str, str] + module_source_paths: Mapping[str, str] + revision: str + source_hash: str + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Compare graph-assisted and source-only workflows on fixed, answer-keyed tasks." + ) + ) + parser.add_argument("--mode", choices=("smoke", "full"), default="full") + parser.add_argument("--output", type=Path) + return parser + + +def encode_report(value: object) -> str: + """Serialize evidence using the repository's deterministic JSON convention.""" + + return json.dumps(value, sort_keys=True, indent=2, ensure_ascii=False) + "\n" + + +def _compact_json(value: object) -> bytes: + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf-8") + + +def _sha256(value: object) -> str: + return hashlib.sha256(_compact_json(value)).hexdigest() + + +def _git(arguments: list[str]) -> str: + return subprocess.run( + ["git", *arguments], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def _module_name(index: int) -> str: + return f"evidence.component_{index:03d}" + + +def _component_path(root: Path, index: int) -> Path: + return root / "src" / "evidence" / f"component_{index:03d}.py" + + +def _dependency_indices(index: int) -> tuple[int, ...]: + if index == 0: + return () + return tuple(sorted({0, index // 2})) + + +def fixture_tasks(source_count: int) -> tuple[TaskSpec, ...]: + """Return the fixed task inventory for one supported fixture size.""" + + terminal = source_count - 1 + return ( + TaskSpec( + task_id="direct_dependencies_of_terminal", + kind="direct_dependencies", + focus_index=terminal, + target_index=None, + depth=1, + prompt=f"List the direct local dependencies of {_module_name(terminal)}.", + ), + TaskSpec( + task_id="two_level_impact_of_component_007", + kind="bounded_impact", + focus_index=7, + target_index=None, + depth=2, + prompt=( + "List every component within two reverse dependency steps of " + "evidence.component_007." + ), + ), + TaskSpec( + task_id="path_from_terminal_to_component_001", + kind="dependency_path", + focus_index=terminal, + target_index=1, + depth=8, + prompt=( + f"Find one shortest local dependency path from {_module_name(terminal)} " + "to evidence.component_001." + ), + ), + ) + + +def write_fixture(root: Path, source_count: int) -> int: + """Create the bounded deterministic corpus and return its exact byte size.""" + + if not SMOKE_SOURCE_COUNT <= source_count <= FULL_SOURCE_COUNT: + raise ValueError( + f"source_count must be between {SMOKE_SOURCE_COUNT} and {FULL_SOURCE_COUNT}" + ) + package = root / "src" / "evidence" + package.mkdir(parents=True) + total_bytes = 0 + for index in range(source_count): + imports = "".join( + f"from evidence.component_{dependency:03d} import compute_{dependency:03d}\n" + for dependency in _dependency_indices(index) + ) + padding = "\n".join( + f' "component-{index:03d}-evidence-row-{row:02d}-{"x" * 52}",' + for row in range(PADDING_ROWS) + ) + expression = " + ".join( + f"compute_{dependency:03d}(value)" for dependency in _dependency_indices(index) + ) + if not expression: + expression = "value" + source = ( + f'"""Deterministic comparative evidence component {index:03d}."""\n\n' + f"{imports}\n" + f"PADDING = (\n{padding}\n)\n\n" + f"def compute_{index:03d}(value: int) -> int:\n" + f' """Return the bounded component {index:03d} result."""\n' + f" return ({expression}) + {index}\n" + ) + raw = source.encode("utf-8") + _component_path(root, index).write_bytes(raw) + total_bytes += len(raw) + return total_bytes + + +def _all_dependencies(source_count: int) -> dict[int, tuple[int, ...]]: + return {index: _dependency_indices(index) for index in range(source_count)} + + +def _bounded_impact( + dependencies: Mapping[int, Sequence[int]], + focus: int, + depth: int, +) -> tuple[int, ...]: + reverse: dict[int, list[int]] = {index: [] for index in dependencies} + for source, targets in dependencies.items(): + for target in targets: + reverse[target].append(source) + queue: deque[tuple[int, int]] = deque([(focus, 0)]) + seen = {focus} + results: set[int] = set() + while queue: + current, current_depth = queue.popleft() + if current_depth >= depth: + continue + for source in sorted(reverse[current]): + if source in seen: + continue + seen.add(source) + results.add(source) + queue.append((source, current_depth + 1)) + return tuple(sorted(results)) + + +def _shortest_path( + dependencies: Mapping[int, Sequence[int]], + focus: int, + target: int, + depth: int, +) -> tuple[int, ...]: + queue: deque[tuple[int, tuple[int, ...]]] = deque([(focus, (focus,))]) + seen = {focus} + while queue: + current, path = queue.popleft() + if current == target: + return path + if len(path) - 1 >= depth: + continue + for dependency in sorted(dependencies[current]): + if dependency in seen: + continue + seen.add(dependency) + queue.append((dependency, (*path, dependency))) + raise TaskEvidenceError( + f"No answer-key path exists from {_module_name(focus)} to {_module_name(target)}" + ) + + +def answer_key(task: TaskSpec, source_count: int) -> tuple[str, ...]: + """Derive the answer independently from the documented fixture topology.""" + + dependencies = _all_dependencies(source_count) + if task.kind == "direct_dependencies": + indices = dependencies[task.focus_index] + elif task.kind == "bounded_impact": + indices = _bounded_impact(dependencies, task.focus_index, task.depth) + else: + if task.target_index is None: + raise TaskEvidenceError("Dependency-path task has no target") + indices = _shortest_path( + dependencies, + task.focus_index, + task.target_index, + task.depth, + ) + return tuple(_module_name(index) for index in indices) + + +def _prepare_graph( + root: Path, + source_count: int, + source_bytes: int, +) -> tuple[_GraphContext, dict[str, object]]: + adapter = PythonReferenceAdapter( + root, + source_roots=("src",), + project_id="milestone5-task-evidence", + title="Milestone 5 comparative task evidence", + ) + project = AdapterProject(adapter, cache_root=root / ".docforge" / "task-evidence") + index = ProjectIndex(project) + started = time.perf_counter_ns() + build = index.build() + snapshot = project.load() + elapsed_ms = (time.perf_counter_ns() - started) / 1_000_000 + if elapsed_ms > PREPARATION_LIMIT_MS: + raise TaskEvidenceError( + f"Graph preparation {elapsed_ms:.3f} ms exceeded {PREPARATION_LIMIT_MS:.3f} ms" + ) + expected_names = {_module_name(index) for index in range(source_count)} + module_node_ids = { + node.title: node.node_id for node in snapshot.nodes if node.title in expected_names + } + if set(module_node_ids) != expected_names: + raise TaskEvidenceError("Graph preparation did not publish every fixture module") + node_modules = {node_id: name for name, node_id in module_node_ids.items()} + module_source_paths = { + node.title: node.source_path for node in snapshot.nodes if node.title in expected_names + } + return ( + _GraphContext( + index=index, + module_node_ids=module_node_ids, + node_modules=node_modules, + module_source_paths=module_source_paths, + revision=snapshot.revision, + source_hash=snapshot.source_hash, + ), + { + "persistent_preparation": True, + "operation": "cold reference-adapter build plus module identity map", + "elapsed_ms": round(elapsed_ms, 3), + "elapsed_limit_ms": PREPARATION_LIMIT_MS, + "source_bytes_indexed": source_bytes, + "node_count": build["node_count"], + "edge_count": build["edge_count"], + "revision": snapshot.revision, + "source_hash": snapshot.source_hash, + }, + ) + + +def _graph_response_bytes(responses: Sequence[Mapping[str, object]]) -> int: + return sum(len(_compact_json(response)) for response in responses) + + +def _graph_direct(context: _GraphContext, task: TaskSpec, source_count: int) -> WorkflowResult: + focus_name = _module_name(task.focus_index) + focus_id = context.module_node_ids[focus_name] + response = context.index.dependencies(focus_id, depth=1, limit=source_count) + results = cast(list[dict[str, object]], response["results"]) + names = tuple(sorted(context.node_modules[cast(str, item["node_id"])] for item in results)) + provenance = tuple( + { + "revision": context.revision, + "source_hash": context.source_hash, + "source_path": context.module_source_paths[focus_name], + "source": focus_name, + "relation": "depends_on", + "target": name, + } + for name in names + ) + return WorkflowResult( + answer=names, + provenance=provenance, + inspected_bytes=_graph_response_bytes((response,)), + ) + + +def _graph_impact(context: _GraphContext, task: TaskSpec, source_count: int) -> WorkflowResult: + focus_name = _module_name(task.focus_index) + queue: deque[tuple[str, int]] = deque([(focus_name, 0)]) + seen = {focus_name} + discovered: set[str] = set() + supporting_edges: list[dict[str, object]] = [] + responses: list[Mapping[str, object]] = [] + while queue: + current_name, current_depth = queue.popleft() + if current_depth >= task.depth: + continue + response = context.index.backlinks( + context.module_node_ids[current_name], + relation="depends_on", + limit=source_count, + ) + responses.append(response) + edges = cast(list[dict[str, object]], response["edges"]) + for edge in edges: + source_name = context.node_modules[cast(str, edge["source_id"])] + if source_name in seen: + continue + seen.add(source_name) + discovered.add(source_name) + supporting_edges.append( + { + "revision": context.revision, + "source_hash": context.source_hash, + "source_path": context.module_source_paths[source_name], + "source": source_name, + "relation": "depends_on", + "target": current_name, + } + ) + queue.append((source_name, current_depth + 1)) + return WorkflowResult( + answer=tuple(sorted(discovered)), + provenance=tuple( + sorted( + supporting_edges, + key=lambda item: (cast(str, item["source"]), cast(str, item["target"])), + ) + ), + inspected_bytes=_graph_response_bytes(responses), + ) + + +def _graph_path(context: _GraphContext, task: TaskSpec, source_count: int) -> WorkflowResult: + if task.target_index is None: + raise TaskEvidenceError("Dependency-path task has no target") + focus_name = _module_name(task.focus_index) + target_name = _module_name(task.target_index) + response = context.index.dependencies( + context.module_node_ids[focus_name], + depth=task.depth, + limit=source_count, + ) + results = cast(list[dict[str, object]], response["results"]) + match = next( + ( + item + for item in results + if context.node_modules[cast(str, item["node_id"])] == target_name + ), + None, + ) + if match is None: + raise TaskEvidenceError("Graph-assisted workflow did not find the answer-key path") + path_ids = cast(list[str] | tuple[str, ...], match["path"]) + path = tuple(context.node_modules[node_id] for node_id in path_ids) + provenance = tuple( + { + "revision": context.revision, + "source_hash": context.source_hash, + "source_path": context.module_source_paths[source], + "source": source, + "relation": "depends_on", + "target": target, + } + for source, target in zip(path, path[1:], strict=False) + ) + return WorkflowResult( + answer=path, + provenance=provenance, + inspected_bytes=_graph_response_bytes((response,)), + ) + + +def run_graph_task( + context: _GraphContext, + task: TaskSpec, + source_count: int, +) -> WorkflowResult: + if task.kind == "direct_dependencies": + return _graph_direct(context, task, source_count) + if task.kind == "bounded_impact": + return _graph_impact(context, task, source_count) + return _graph_path(context, task, source_count) + + +def _parse_source_dependencies( + raw: bytes, + source_index: int, +) -> tuple[tuple[int, int], ...]: + prefix = "from evidence.component_" + suffix = " import compute_" + imports: list[tuple[int, int]] = [] + for line_number, line in enumerate(raw.decode("utf-8").splitlines(), start=1): + if not line.startswith(prefix) or suffix not in line: + continue + module_text, function_text = line[len(prefix) :].split(suffix, maxsplit=1) + if not ( + len(module_text) == 3 + and module_text.isdigit() + and len(function_text) == 3 + and function_text.isdigit() + and module_text == function_text + ): + raise TaskEvidenceError( + f"Source-only parser rejected component {source_index:03d} import syntax" + ) + imports.append((int(module_text), line_number)) + return tuple(imports) + + +def _read_source_dependencies( + root: Path, + source_index: int, +) -> tuple[tuple[tuple[int, int], ...], int]: + raw = _component_path(root, source_index).read_bytes() + return _parse_source_dependencies(raw, source_index), len(raw) + + +def _read_all_source_dependencies( + root: Path, + source_count: int, +) -> tuple[dict[int, tuple[int, ...]], dict[tuple[int, int], int], int]: + dependencies: dict[int, tuple[int, ...]] = {} + lines: dict[tuple[int, int], int] = {} + inspected_bytes = 0 + for source in range(source_count): + imports, size = _read_source_dependencies(root, source) + inspected_bytes += size + dependencies[source] = tuple(target for target, _ in imports) + lines.update({(source, target): line for target, line in imports}) + return dependencies, lines, inspected_bytes + + +def _source_provenance( + edges: Sequence[tuple[int, int]], + lines: Mapping[tuple[int, int], int], +) -> tuple[Mapping[str, object], ...]: + return tuple( + { + "source_path": f"src/evidence/component_{source:03d}.py", + "line": lines[(source, target)], + "source": _module_name(source), + "relation": "depends_on", + "target": _module_name(target), + } + for source, target in edges + ) + + +def _source_direct(root: Path, task: TaskSpec) -> WorkflowResult: + imports, inspected_bytes = _read_source_dependencies(root, task.focus_index) + edges = tuple((task.focus_index, target) for target, _ in imports) + lines = {(task.focus_index, target): line for target, line in imports} + return WorkflowResult( + answer=tuple(_module_name(target) for target, _ in imports), + provenance=_source_provenance(edges, lines), + inspected_bytes=inspected_bytes, + ) + + +def _source_impact(root: Path, task: TaskSpec, source_count: int) -> WorkflowResult: + dependencies, lines, inspected_bytes = _read_all_source_dependencies(root, source_count) + reverse: dict[int, list[int]] = {index: [] for index in dependencies} + for source, targets in dependencies.items(): + for target in targets: + reverse[target].append(source) + queue: deque[tuple[int, int]] = deque([(task.focus_index, 0)]) + seen = {task.focus_index} + discovered: set[int] = set() + edges: list[tuple[int, int]] = [] + while queue: + current, current_depth = queue.popleft() + if current_depth >= task.depth: + continue + for source in sorted(reverse[current]): + if source in seen: + continue + seen.add(source) + discovered.add(source) + edges.append((source, current)) + queue.append((source, current_depth + 1)) + return WorkflowResult( + answer=tuple(_module_name(index) for index in sorted(discovered)), + provenance=_source_provenance(sorted(edges), lines), + inspected_bytes=inspected_bytes, + ) + + +def _source_dependency_path( + root: Path, + task: TaskSpec, + source_count: int, +) -> WorkflowResult: + if task.target_index is None: + raise TaskEvidenceError("Dependency-path task has no target") + dependencies, lines, inspected_bytes = _read_all_source_dependencies(root, source_count) + path = _shortest_path( + dependencies, + task.focus_index, + task.target_index, + task.depth, + ) + edges = tuple(zip(path, path[1:], strict=False)) + return WorkflowResult( + answer=tuple(_module_name(index) for index in path), + provenance=_source_provenance(edges, lines), + inspected_bytes=inspected_bytes, + ) + + +def run_source_task(root: Path, task: TaskSpec, source_count: int) -> WorkflowResult: + if task.kind == "direct_dependencies": + return _source_direct(root, task) + if task.kind == "bounded_impact": + return _source_impact(root, task, source_count) + return _source_dependency_path(root, task, source_count) + + +def _measure( + operation: Callable[[], WorkflowResult], + *, + workflow: WorkflowName, + samples: int, +) -> dict[str, object]: + durations: list[float] = [] + stable_result: WorkflowResult | None = None + stable_payload: dict[str, object] | None = None + response_bytes = 0 + for _ in range(samples): + started = time.perf_counter_ns() + current = operation() + elapsed_ms = (time.perf_counter_ns() - started) / 1_000_000 + payload = current.payload() + current_response_bytes = len(_compact_json(payload)) + if stable_payload is None or stable_result is None: + stable_result = current + stable_payload = payload + response_bytes = current_response_bytes + elif ( + payload != stable_payload + or current.inspected_bytes != stable_result.inspected_bytes + or current_response_bytes != response_bytes + ): + raise TaskEvidenceError(f"{workflow} task output changed between samples") + durations.append(elapsed_ms) + assert stable_result is not None + assert stable_payload is not None + ordered = sorted(durations) + p95_index = max(0, math.ceil(len(ordered) * 0.95) - 1) + p95_ms = ordered[p95_index] + inspected_limit = ( + MAX_GRAPH_INSPECTED_BYTES if workflow == "graph_assisted" else MAX_SOURCE_INSPECTED_BYTES + ) + if p95_ms > TASK_P95_LIMIT_MS: + raise TaskEvidenceError( + f"{workflow} task p95 {p95_ms:.3f} ms exceeded {TASK_P95_LIMIT_MS:.3f} ms" + ) + if stable_result.inspected_bytes > inspected_limit: + raise TaskEvidenceError( + f"{workflow} inspected {stable_result.inspected_bytes} bytes, limit {inspected_limit}" + ) + if response_bytes > MAX_TASK_RESPONSE_BYTES: + raise TaskEvidenceError( + f"{workflow} response was {response_bytes} bytes, limit {MAX_TASK_RESPONSE_BYTES}" + ) + return { + "samples": samples, + "median_ms": round(statistics.median(ordered), 3), + "p95_ms": round(p95_ms, 3), + "p95_limit_ms": TASK_P95_LIMIT_MS, + "inspected_bytes": stable_result.inspected_bytes, + "inspected_bytes_limit": inspected_limit, + "response_bytes": response_bytes, + "response_bytes_limit": MAX_TASK_RESPONSE_BYTES, + "result": stable_payload, + } + + +def _lower_is_better( + graph_value: int | float, + source_value: int | float, +) -> dict[str, object]: + if graph_value < source_value: + lower: str = "graph_assisted" + elif source_value < graph_value: + lower = "source_only" + else: + lower = "tie" + return { + "graph_assisted": graph_value, + "source_only": source_value, + "lower_workflow": lower, + "absolute_difference": round(abs(graph_value - source_value), 3), + } + + +def _task_evidence( + root: Path, + context: _GraphContext, + task: TaskSpec, + *, + source_count: int, + samples: int, +) -> dict[str, object]: + expected = answer_key(task, source_count) + graph = _measure( + lambda: run_graph_task(context, task, source_count), + workflow="graph_assisted", + samples=samples, + ) + source = _measure( + lambda: run_source_task(root, task, source_count), + workflow="source_only", + samples=samples, + ) + graph_result = cast(dict[str, object], graph["result"]) + source_result = cast(dict[str, object], source["result"]) + graph_correct = graph_result["answer"] == list(expected) + source_correct = source_result["answer"] == list(expected) + if not graph_correct or not source_correct: + raise TaskEvidenceError(f"Workflow answer did not match key for {task.task_id}") + comparisons = { + "task_elapsed_ms": _lower_is_better( + cast(float, graph["median_ms"]), + cast(float, source["median_ms"]), + ), + "inspected_bytes": _lower_is_better( + cast(int, graph["inspected_bytes"]), + cast(int, source["inspected_bytes"]), + ), + "response_bytes": _lower_is_better( + cast(int, graph["response_bytes"]), + cast(int, source["response_bytes"]), + ), + } + advantages = [ + { + "metric": metric, + "lower_workflow": comparison["lower_workflow"], + "absolute_difference": comparison["absolute_difference"], + "scope": "this fixed task; graph cold preparation excluded", + } + for metric, comparison in comparisons.items() + if comparison["lower_workflow"] != "tie" + ] + return { + "task": task.as_dict(), + "answer_key": list(expected), + "answer_key_sha256": _sha256(list(expected)), + "workflows": { + "graph_assisted": {**graph, "correct": graph_correct}, + "source_only": {**source, "correct": source_correct}, + }, + "comparison": { + "both_exact": True, + "metrics": comparisons, + "measured_advantages": advantages, + }, + } + + +def build_task_evidence( + root: Path, + *, + source_count: int, + samples: int, +) -> dict[str, object]: + """Build the fixture, execute both workflows, and return gated evidence.""" + + if samples < 1: + raise ValueError("samples must be positive") + source_bytes = write_fixture(root, source_count) + context, preparation = _prepare_graph(root, source_count, source_bytes) + tasks = [ + _task_evidence( + root, + context, + task, + source_count=source_count, + samples=samples, + ) + for task in fixture_tasks(source_count) + ] + semantic_evidence = { + "fixture": { + "source_count": source_count, + "source_bytes": source_bytes, + "topology": "component N depends on component 0 and floor(N/2), deduplicated", + }, + "graph_identity": { + "revision": context.revision, + "source_hash": context.source_hash, + }, + "tasks": [ + { + "task": task["task"], + "answer_key": task["answer_key"], + "answer_key_sha256": task["answer_key_sha256"], + "graph_result": cast( + dict[str, object], + cast(dict[str, object], task["workflows"])["graph_assisted"], + )["result"], + "source_result": cast( + dict[str, object], + cast(dict[str, object], task["workflows"])["source_only"], + )["result"], + } + for task in tasks + ], + } + return { + "fixture": { + "kind": "deterministic_generated_python_dependency_corpus", + "external_projects": False, + "self_hosting": False, + "production_bindings": False, + "source_count": source_count, + "source_bytes": source_bytes, + "padding_rows_per_source": PADDING_ROWS, + "topology": "component N depends on component 0 and floor(N/2), deduplicated", + }, + "preparation": { + "graph_assisted": preparation, + "source_only": { + "persistent_preparation": False, + "operation": "none; each task reads only the source needed by its algorithm", + }, + }, + "tasks": tasks, + "summary": { + "task_count": len(tasks), + "both_workflows_exact_for_all_tasks": True, + "comparison_scope": ( + "Per-task latency excludes the separately reported one-time graph preparation." + ), + "semantic_evidence_sha256": _sha256(semantic_evidence), + }, + } + + +def main() -> int: + arguments = _parser().parse_args() + source_count = FULL_SOURCE_COUNT if arguments.mode == "full" else SMOKE_SOURCE_COUNT + samples = FULL_SAMPLES if arguments.mode == "full" else SMOKE_SAMPLES + with tempfile.TemporaryDirectory(prefix="docforge-milestone5-task-evidence-") as directory: + measurement = build_task_evidence( + Path(directory).resolve(), + source_count=source_count, + samples=samples, + ) + report: dict[str, object] = { + "schema_version": 1, + "benchmark": "docforge2_milestone5_representative_tasks", + "mode": arguments.mode, + "source": { + "revision": _git(["rev-parse", "HEAD"]), + "dirty": bool(_git(["status", "--porcelain"])), + }, + "environment": { + "platform": platform.platform(), + "machine": platform.machine(), + "python": platform.python_version(), + "implementation": platform.python_implementation(), + }, + "method": { + "clock": "time.perf_counter_ns", + "samples": samples, + "answer_key": "independent arithmetic oracle over the declared fixture topology", + "graph_workflow": "bounded ProjectIndex dependency and backlink responses", + "source_workflow": "bounded literal source reads and fixed import syntax parsing", + "inspected_bytes": ( + "agent-visible compact graph response bytes or exact source bytes inspected " + "by the source-only task; internal storage I/O is not compared" + ), + "response_bytes": "UTF-8 bytes of compact sorted answer-and-provenance JSON", + "comparison": ( + "Direction-neutral lower-is-better measurements for each fixed task. " + "One-time graph preparation is reported separately." + ), + }, + **measurement, + } + encoded = encode_report(report) + report_bytes = len(encoded.encode("utf-8")) + if report_bytes > MAX_REPORT_BYTES: + raise TaskEvidenceError( + f"Evidence report was {report_bytes} bytes, limit {MAX_REPORT_BYTES}" + ) + if arguments.output is not None: + output = arguments.output.resolve() + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(encoded, encoding="utf-8") + sys.stdout.write(encoded) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())