From 1a6f33e2de1316ab0bef3978916e4fc9c5d1295c Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 16:21:10 -0400 Subject: [PATCH] Add pinned real-package task evidence --- tests/test_milestone5_task_evidence.py | 88 ++- tools/milestone5_task_evidence.py | 914 +++++++++++++++++++++++-- 2 files changed, 947 insertions(+), 55 deletions(-) diff --git a/tests/test_milestone5_task_evidence.py b/tests/test_milestone5_task_evidence.py index c313577..26de64e 100644 --- a/tests/test_milestone5_task_evidence.py +++ b/tests/test_milestone5_task_evidence.py @@ -7,15 +7,25 @@ import tempfile import unittest from pathlib import Path from typing import cast +from unittest import mock from tools.milestone5_task_evidence import ( MAX_GRAPH_INSPECTED_BYTES, MAX_SOURCE_INSPECTED_BYTES, MAX_TASK_RESPONSE_BYTES, + PINNED_REAL_SOURCE_BYTES, + PINNED_REAL_SOURCE_COUNT, + PINNED_REAL_SOURCE_SHA256, + PINNED_REAL_VERSION, SMOKE_SOURCE_COUNT, + TaskEvidenceError, answer_key, + build_all_task_evidence, + build_real_package_evidence, build_task_evidence, fixture_tasks, + installed_real_package, + real_package_tasks, ) ROOT = Path(__file__).resolve().parents[1] @@ -125,6 +135,73 @@ class Milestone5TaskEvidenceTests(unittest.TestCase): ) self.assertEqual(impact["inspected_bytes"], path["inspected_bytes"]) + def test_real_package_identity_is_installed_lock_pinned_and_exact(self) -> None: + _, identity = installed_real_package() + + self.assertEqual(PINNED_REAL_VERSION, identity["lock_version"]) + self.assertEqual(PINNED_REAL_VERSION, identity["installed_version"]) + self.assertEqual(PINNED_REAL_SOURCE_COUNT, identity["source_count"]) + self.assertEqual(PINNED_REAL_SOURCE_BYTES, identity["source_bytes"]) + self.assertEqual(PINNED_REAL_SOURCE_SHA256, identity["source_tree_sha256"]) + + def test_real_package_tasks_are_exact_provenanced_bounded_and_read_only(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + evidence = build_real_package_evidence(root, samples=1) + copied_sources = tuple((root / "src" / "markdown_it").rglob("*.py")) + self.assertTrue(copied_sources) + self.assertTrue(all(path.stat().st_mode & 0o222 == 0 for path in copied_sources)) + + fixture = cast(dict[str, object], evidence["fixture"]) + self.assertFalse(fixture["network"]) + self.assertFalse(fixture["installed_source_mutation"]) + self.assertFalse(fixture["external_project_mutation"]) + tasks = cast(list[dict[str, object]], evidence["tasks"]) + self.assertEqual( + [list(expected) for _, expected in real_package_tasks()], + [task["answer_key"] for task in tasks], + ) + for task in tasks: + workflows = cast(dict[str, dict[str, object]], task["workflows"]) + for workflow in workflows.values(): + self.assertTrue(workflow["correct"]) + result = cast(dict[str, object], workflow["result"]) + self.assertTrue(result["provenance"]) + self.assertLessEqual( + cast(int, workflow["response_bytes"]), + MAX_TASK_RESPONSE_BYTES, + ) + comparison = cast(dict[str, object], task["comparison"]) + self.assertTrue(comparison["both_exact"]) + + def test_real_package_pin_change_fails_closed(self) -> None: + with ( + mock.patch( + "tools.milestone5_task_evidence._source_tree_identity", + return_value=( + PINNED_REAL_SOURCE_COUNT, + PINNED_REAL_SOURCE_BYTES, + "0" * 64, + ), + ), + self.assertRaises(TaskEvidenceError), + ): + installed_real_package() + + def test_combined_semantic_evidence_is_repeatable_across_both_tracks(self) -> None: + hashes: list[str] = [] + for _ in range(2): + with tempfile.TemporaryDirectory() as directory: + evidence = build_all_task_evidence( + Path(directory).resolve(), + generated_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_smoke_cli_emits_the_same_machine_readable_report_it_writes(self) -> None: with tempfile.TemporaryDirectory() as directory: output = Path(directory) / "evidence.json" @@ -146,8 +223,17 @@ class Milestone5TaskEvidenceTests(unittest.TestCase): self.assertEqual(completed.stdout, output.read_text(encoding="utf-8")) self.assertEqual("docforge2_milestone5_representative_tasks", report["benchmark"]) + self.assertEqual(2, report["schema_version"]) self.assertEqual("smoke", report["mode"]) - self.assertEqual(SMOKE_SOURCE_COUNT, report["fixture"]["source_count"]) + self.assertEqual( + SMOKE_SOURCE_COUNT, + report["tracks"]["generated_scale"]["fixture"]["source_count"], + ) + self.assertEqual( + PINNED_REAL_SOURCE_COUNT, + report["tracks"]["installed_real_package"]["fixture"]["source_count"], + ) + self.assertEqual(6, report["summary"]["task_count"]) if __name__ == "__main__": diff --git a/tools/milestone5_task_evidence.py b/tools/milestone5_task_evidence.py index 9a871b0..2bddc3a 100644 --- a/tools/milestone5_task_evidence.py +++ b/tools/milestone5_task_evidence.py @@ -3,22 +3,35 @@ from __future__ import annotations import argparse +import ast import hashlib import json import math import platform +import shutil +import stat import statistics import subprocess import sys import tempfile import time +import tomllib from collections import deque from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass +from importlib import metadata from pathlib import Path from typing import Literal, cast -from docforge.adapter_sdk import AdapterProject +from docforge.adapter_sdk import ( + AdapterEdge, + AdapterLoader, + AdapterNode, + AdapterProject, + AdapterProjection, + Edge, + Node, +) from docforge.adapters.python import PythonReferenceAdapter from docforge.index import ProjectIndex @@ -34,6 +47,12 @@ MAX_REPORT_BYTES = 1024 * 1024 PREPARATION_LIMIT_MS = 30_000.0 TASK_P95_LIMIT_MS = 5_000.0 PADDING_ROWS = 24 +PINNED_REAL_DISTRIBUTION = "markdown-it-py" +PINNED_REAL_VERSION = "4.2.0" +PINNED_REAL_IMPORT_ROOT = "markdown_it" +PINNED_REAL_SOURCE_COUNT = 66 +PINNED_REAL_SOURCE_BYTES = 225_945 +PINNED_REAL_SOURCE_SHA256 = "bd57c9f332fcf6507282ec2023e6804fce0cf844631696336ee17cbe46e63aad" WorkflowName = Literal["graph_assisted", "source_only"] TaskKind = Literal["direct_dependencies", "bounded_impact", "dependency_path"] @@ -47,8 +66,8 @@ class TaskEvidenceError(RuntimeError): class TaskSpec: task_id: str kind: TaskKind - focus_index: int - target_index: int | None + focus: str + target: str | None depth: int prompt: str @@ -56,8 +75,8 @@ class TaskSpec: 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), + "focus": self.focus, + "target": self.target, "depth": self.depth, "prompt": self.prompt, } @@ -82,6 +101,9 @@ class _GraphContext: module_node_ids: Mapping[str, str] node_modules: Mapping[str, str] module_source_paths: Mapping[str, str] + outgoing: Mapping[str, tuple[str, ...]] + incoming: Mapping[str, tuple[str, ...]] + relation: str revision: str source_hash: str @@ -130,6 +152,14 @@ def _module_name(index: int) -> str: return f"evidence.component_{index:03d}" +def _generated_index(module_name: str) -> int: + prefix = "evidence.component_" + suffix = module_name.removeprefix(prefix) + if not module_name.startswith(prefix) or len(suffix) != 3 or not suffix.isdigit(): + raise TaskEvidenceError(f"Generated task uses an invalid module name: {module_name}") + return int(suffix) + + def _component_path(root: Path, index: int) -> Path: return root / "src" / "evidence" / f"component_{index:03d}.py" @@ -148,16 +178,16 @@ def fixture_tasks(source_count: int) -> tuple[TaskSpec, ...]: TaskSpec( task_id="direct_dependencies_of_terminal", kind="direct_dependencies", - focus_index=terminal, - target_index=None, + focus=_module_name(terminal), + target=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, + focus=_module_name(7), + target=None, depth=2, prompt=( "List every component within two reverse dependency steps of " @@ -167,8 +197,8 @@ def fixture_tasks(source_count: int) -> tuple[TaskSpec, ...]: TaskSpec( task_id="path_from_terminal_to_component_001", kind="dependency_path", - focus_index=terminal, - target_index=1, + focus=_module_name(terminal), + target=_module_name(1), depth=8, prompt=( f"Find one shortest local dependency path from {_module_name(terminal)} " @@ -273,17 +303,19 @@ def answer_key(task: TaskSpec, source_count: int) -> tuple[str, ...]: """Derive the answer independently from the documented fixture topology.""" dependencies = _all_dependencies(source_count) + focus_index = _generated_index(task.focus) if task.kind == "direct_dependencies": - indices = dependencies[task.focus_index] + indices = dependencies[focus_index] elif task.kind == "bounded_impact": - indices = _bounded_impact(dependencies, task.focus_index, task.depth) + indices = _bounded_impact(dependencies, focus_index, task.depth) else: - if task.target_index is None: + if task.target is None: raise TaskEvidenceError("Dependency-path task has no target") + target_index = _generated_index(task.target) indices = _shortest_path( dependencies, - task.focus_index, - task.target_index, + focus_index, + target_index, task.depth, ) return tuple(_module_name(index) for index in indices) @@ -291,18 +323,21 @@ def answer_key(task: TaskSpec, source_count: int) -> tuple[str, ...]: def _prepare_graph( root: Path, - source_count: int, source_bytes: int, + *, + expected_names: set[str], + project_id: str, + title: str, + loader: AdapterLoader | None = None, + relation: str = "depends_on", + operation: str = "cold reference-adapter build plus module identity map", ) -> tuple[_GraphContext, dict[str, object]]: - adapter = PythonReferenceAdapter( - root, - source_roots=("src",), - project_id="milestone5-task-evidence", - title="Milestone 5 comparative task evidence", + started = time.perf_counter_ns() + adapter: AdapterLoader = loader or PythonReferenceAdapter( + root, source_roots=("src",), project_id=project_id, title=title ) 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 @@ -310,7 +345,6 @@ def _prepare_graph( 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 } @@ -320,18 +354,34 @@ def _prepare_graph( module_source_paths = { node.title: node.source_path for node in snapshot.nodes if node.title in expected_names } + outgoing_sets: dict[str, set[str]] = {name: set() for name in expected_names} + incoming_sets: dict[str, set[str]] = {name: set() for name in expected_names} + for edge in snapshot.edges: + if edge.relation != relation: + continue + source_name = node_modules.get(edge.source_id) + target_name = node_modules.get(edge.target_id) + if source_name is None or target_name is None: + continue + outgoing_sets[source_name].add(target_name) + incoming_sets[target_name].add(source_name) + outgoing = {name: tuple(sorted(targets)) for name, targets in outgoing_sets.items()} + incoming = {name: tuple(sorted(sources)) for name, sources in incoming_sets.items()} return ( _GraphContext( index=index, module_node_ids=module_node_ids, node_modules=node_modules, module_source_paths=module_source_paths, + outgoing=outgoing, + incoming=incoming, + relation=relation, revision=snapshot.revision, source_hash=snapshot.source_hash, ), { "persistent_preparation": True, - "operation": "cold reference-adapter build plus module identity map", + "operation": operation, "elapsed_ms": round(elapsed_ms, 3), "elapsed_limit_ms": PREPARATION_LIMIT_MS, "source_bytes_indexed": source_bytes, @@ -348,7 +398,7 @@ def _graph_response_bytes(responses: Sequence[Mapping[str, object]]) -> int: def _graph_direct(context: _GraphContext, task: TaskSpec, source_count: int) -> WorkflowResult: - focus_name = _module_name(task.focus_index) + focus_name = task.focus 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"]) @@ -372,7 +422,7 @@ def _graph_direct(context: _GraphContext, task: TaskSpec, source_count: int) -> def _graph_impact(context: _GraphContext, task: TaskSpec, source_count: int) -> WorkflowResult: - focus_name = _module_name(task.focus_index) + focus_name = task.focus queue: deque[tuple[str, int]] = deque([(focus_name, 0)]) seen = {focus_name} discovered: set[str] = set() @@ -419,10 +469,10 @@ def _graph_impact(context: _GraphContext, task: TaskSpec, source_count: int) -> def _graph_path(context: _GraphContext, task: TaskSpec, source_count: int) -> WorkflowResult: - if task.target_index is None: + if task.target is None: raise TaskEvidenceError("Dependency-path task has no target") - focus_name = _module_name(task.focus_index) - target_name = _module_name(task.target_index) + focus_name = task.focus + target_name = task.target response = context.index.dependencies( context.module_node_ids[focus_name], depth=task.depth, @@ -536,9 +586,10 @@ def _source_provenance( 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} + focus_index = _generated_index(task.focus) + imports, inspected_bytes = _read_source_dependencies(root, focus_index) + edges = tuple((focus_index, target) for target, _ in imports) + lines = {(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), @@ -552,8 +603,9 @@ def _source_impact(root: Path, task: TaskSpec, source_count: int) -> WorkflowRes 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} + focus_index = _generated_index(task.focus) + queue: deque[tuple[int, int]] = deque([(focus_index, 0)]) + seen = {focus_index} discovered: set[int] = set() edges: list[tuple[int, int]] = [] while queue: @@ -579,13 +631,15 @@ def _source_dependency_path( task: TaskSpec, source_count: int, ) -> WorkflowResult: - if task.target_index is None: + if task.target is None: raise TaskEvidenceError("Dependency-path task has no target") + focus_index = _generated_index(task.focus) + target_index = _generated_index(task.target) dependencies, lines, inspected_bytes = _read_all_source_dependencies(root, source_count) path = _shortest_path( dependencies, - task.focus_index, - task.target_index, + focus_index, + target_index, task.depth, ) edges = tuple(zip(path, path[1:], strict=False)) @@ -604,6 +658,607 @@ def run_source_task(root: Path, task: TaskSpec, source_count: int) -> WorkflowRe return _source_dependency_path(root, task, source_count) +def _locked_distribution_version(distribution_name: str) -> str: + document = tomllib.loads((ROOT / "uv.lock").read_text(encoding="utf-8")) + packages = document.get("package") + if not isinstance(packages, list): + raise TaskEvidenceError("uv.lock does not contain a package inventory") + matches: list[str] = [] + for raw_package in cast(list[object], packages): + if not isinstance(raw_package, dict): + continue + package = cast(dict[str, object], raw_package) + if package.get("name") != distribution_name: + continue + version = package.get("version") + if not isinstance(version, str): + raise TaskEvidenceError(f"{distribution_name} has no fixed lockfile version") + matches.append(version) + if len(matches) != 1: + raise TaskEvidenceError( + f"uv.lock must contain exactly one {distribution_name} package record" + ) + return matches[0] + + +def _installed_python_sources(package_root: Path) -> tuple[Path, ...]: + try: + resolved_root = package_root.resolve(strict=True) + except OSError as error: + raise TaskEvidenceError("Pinned real-package source root is unavailable") from error + sources: list[Path] = [] + for path in sorted( + resolved_root.rglob("*.py"), + key=lambda item: item.relative_to(resolved_root).as_posix(), + ): + try: + metadata_result = path.lstat() + resolved = path.resolve(strict=True) + except OSError as error: + raise TaskEvidenceError("Pinned real-package source cannot be inspected") from error + if ( + stat.S_ISLNK(metadata_result.st_mode) + or not stat.S_ISREG(metadata_result.st_mode) + or not resolved.is_relative_to(resolved_root) + or resolved != path + ): + raise TaskEvidenceError("Pinned real-package sources must be confined regular files") + sources.append(path) + return tuple(sources) + + +def _source_tree_identity(package_root: Path) -> tuple[int, int, str]: + digest = hashlib.sha256() + total_bytes = 0 + sources = _installed_python_sources(package_root) + for path in sources: + relative = path.relative_to(package_root).as_posix() + raw = path.read_bytes() + total_bytes += len(raw) + digest.update(relative.encode("utf-8")) + digest.update(b"\0") + digest.update(str(len(raw)).encode("ascii")) + digest.update(b"\0") + digest.update(raw) + digest.update(b"\0") + return len(sources), total_bytes, digest.hexdigest() + + +def installed_real_package() -> tuple[Path, dict[str, object]]: + """Resolve and freeze the lock-pinned installed real-package source tree.""" + + locked_version = _locked_distribution_version(PINNED_REAL_DISTRIBUTION) + try: + distribution = metadata.distribution(PINNED_REAL_DISTRIBUTION) + except metadata.PackageNotFoundError as error: + raise TaskEvidenceError( + f"{PINNED_REAL_DISTRIBUTION} is not installed in the active environment" + ) from error + installed_version = distribution.version + package_root = Path(str(distribution.locate_file(PINNED_REAL_IMPORT_ROOT))).resolve(strict=True) + source_count, source_bytes, source_sha256 = _source_tree_identity(package_root) + observed = ( + locked_version, + installed_version, + source_count, + source_bytes, + source_sha256, + ) + expected = ( + PINNED_REAL_VERSION, + PINNED_REAL_VERSION, + PINNED_REAL_SOURCE_COUNT, + PINNED_REAL_SOURCE_BYTES, + PINNED_REAL_SOURCE_SHA256, + ) + if observed != expected: + raise TaskEvidenceError( + "Pinned real-package evidence changed; update the lock and expected evidence " + f"deliberately. Observed {observed!r}, expected {expected!r}" + ) + return package_root, { + "distribution": PINNED_REAL_DISTRIBUTION, + "import_root": PINNED_REAL_IMPORT_ROOT, + "lock_version": locked_version, + "installed_version": installed_version, + "source_scope": "sorted markdown_it/**/*.py path, length, and content", + "source_count": source_count, + "source_bytes": source_bytes, + "source_tree_sha256": source_sha256, + } + + +def _copy_real_package_read_only(source_root: Path, project_root: Path) -> None: + destination_root = project_root / "src" / PINNED_REAL_IMPORT_ROOT + for source in _installed_python_sources(source_root): + relative = source.relative_to(source_root) + destination = destination_root / relative + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copyfile(source, destination) + destination.chmod(0o444) + for directory in sorted( + (path for path in destination_root.rglob("*") if path.is_dir()), + key=lambda path: len(path.parts), + reverse=True, + ): + directory.chmod(0o555) + destination_root.chmod(0o555) + + +def _python_module_name(source_root: Path, path: Path) -> str: + local = path.relative_to(source_root).with_suffix("") + parts = list(local.parts) + if parts[-1] == "__init__": + parts.pop() + if not parts: + raise TaskEvidenceError("Real-package source produced an empty module name") + return ".".join(parts) + + +def _real_module_inventory(project_root: Path) -> dict[str, Path]: + source_root = project_root / "src" + inventory = { + _python_module_name(source_root, path): path + for path in sorted( + (source_root / PINNED_REAL_IMPORT_ROOT).rglob("*.py"), + key=lambda item: item.relative_to(source_root).as_posix(), + ) + } + if len(inventory) != PINNED_REAL_SOURCE_COUNT: + raise TaskEvidenceError("Copied real-package module inventory changed") + return inventory + + +def _relative_import_base(package: str, level: int, module: str | None) -> str: + if level == 0: + return module or "" + package_parts = package.split(".") if package else [] + keep = len(package_parts) - (level - 1) + if keep < 0: + return "" + prefix = package_parts[:keep] + if module: + prefix.extend(module.split(".")) + return ".".join(prefix) + + +def _parse_real_dependencies( + raw: bytes, + *, + module_name: str, + package: str, + local_modules: set[str], + source_path: str, +) -> tuple[tuple[str, int], ...]: + try: + tree = ast.parse(raw, filename=source_path, type_comments=True) + except (SyntaxError, UnicodeDecodeError) as error: + raise TaskEvidenceError(f"Cannot inspect pinned source {source_path}") from error + evidence: dict[str, int] = {} + for node in ast.walk(tree): + candidates: list[str] = [] + if isinstance(node, ast.Import): + candidates.extend(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom): + base = _relative_import_base(package, node.level, node.module) + if base: + candidates.append(base) + candidates.extend( + f"{base}.{alias.name}" for alias in node.names if alias.name != "*" + ) + else: + continue + for candidate in candidates: + if candidate in local_modules and candidate != module_name: + previous = evidence.get(candidate) + evidence[candidate] = ( + node.lineno if previous is None else min(previous, node.lineno) + ) + return tuple(sorted(evidence.items())) + + +def _read_real_module_dependencies( + project_root: Path, + inventory: Mapping[str, Path], + module_name: str, +) -> tuple[tuple[tuple[str, int], ...], int]: + path = inventory[module_name] + raw = path.read_bytes() + package = module_name if path.name == "__init__.py" else module_name.rpartition(".")[0] + relative = path.relative_to(project_root).as_posix() + return ( + _parse_real_dependencies( + raw, + module_name=module_name, + package=package, + local_modules=set(inventory), + source_path=relative, + ), + len(raw), + ) + + +def _read_all_real_dependencies( + project_root: Path, + inventory: Mapping[str, Path], +) -> tuple[dict[str, tuple[str, ...]], dict[tuple[str, str], tuple[str, int]], int]: + dependencies: dict[str, tuple[str, ...]] = {} + locations: dict[tuple[str, str], tuple[str, int]] = {} + inspected_bytes = 0 + for module_name in sorted(inventory): + imports, size = _read_real_module_dependencies(project_root, inventory, module_name) + inspected_bytes += size + dependencies[module_name] = tuple(target for target, _ in imports) + relative = inventory[module_name].relative_to(project_root).as_posix() + locations.update({(module_name, target): (relative, line) for target, line in imports}) + return dependencies, locations, inspected_bytes + + +class _RealPackageDependencyAdapter: + """Minimal immutable projection over the pinned read-only package sources.""" + + def __init__(self, root: Path) -> None: + self.root = root.resolve(strict=True) + self._projection: AdapterProjection | None = None + + @staticmethod + def _node_id(module_name: str) -> str: + digest = hashlib.sha256(module_name.encode("utf-8")).hexdigest()[:24] + return f"real.module.{digest}" + + def load_projection(self) -> AdapterProjection: + if self._projection is not None: + return self._projection + inventory = _real_module_inventory(self.root) + reference_adapter = PythonReferenceAdapter( + self.root, + source_roots=("src",), + project_id="milestone5-real-task-evidence", + title="Milestone 5 markdown-it-py comparative task evidence", + ) + manifest = reference_adapter.load_manifest() + names_by_path = { + path.relative_to(self.root).as_posix(): module_name + for module_name, path in inventory.items() + } + names_by_source_id = { + source.source_id: names_by_path[source.source_path] for source in manifest.sources + } + dependencies = { + names_by_source_id[source.source_id]: tuple( + sorted(names_by_source_id[target] for target in source.dependencies) + ) + for source in manifest.sources + } + source_bytes = sum(path.stat().st_size for path in inventory.values()) + if source_bytes != PINNED_REAL_SOURCE_BYTES: + raise TaskEvidenceError("Copied real-package source bytes changed") + nodes: list[AdapterNode] = [] + for module_name, path in inventory.items(): + raw = path.read_bytes() + relative = path.relative_to(self.root).as_posix() + nodes.append( + AdapterNode( + Node( + node_id=self._node_id(module_name), + title=module_name, + family="code", + authority="derived", + status="active", + tags=("module", "pinned-real-package"), + summary=f"Pinned installed Python module {module_name}.", + content=f"Local dependency projection for {module_name}.", + source_path=relative, + source_anchor="L1", + content_hash=hashlib.sha256(raw).hexdigest(), + ), + metadata=(("kind", "module"),), + ) + ) + edges = [ + AdapterEdge( + Edge( + source_id=self._node_id(source), + relation="imports", + target_id=self._node_id(target), + ), + metadata=(("evidence", "python_reference_adapter_manifest"),), + ) + for source, targets in dependencies.items() + for target in targets + ] + self._projection = AdapterProjection( + project_id="milestone5-real-task-evidence", + title="Milestone 5 markdown-it-py comparative task evidence", + adapter_id="pinned-python-dependencies", + adapter_version="1", + root=self.root, + revision=manifest.revision, + source_hash=manifest.source_hash, + nodes=tuple(sorted(nodes, key=lambda item: item.node.node_id)), + edges=tuple( + sorted( + edges, + key=lambda item: ( + item.edge.source_id, + item.edge.relation, + item.edge.target_id, + ), + ) + ), + ) + return self._projection + + +def _real_provenance( + edges: Sequence[tuple[str, str]], + locations: Mapping[tuple[str, str], tuple[str, int]], +) -> tuple[Mapping[str, object], ...]: + return tuple( + { + "source_path": locations[(source, target)][0], + "line": locations[(source, target)][1], + "source": source, + "relation": "imports", + "target": target, + } + for source, target in edges + ) + + +def _shortest_named_path( + dependencies: Mapping[str, Sequence[str]], + focus: str, + target: str, + depth: int, +) -> tuple[str, ...]: + queue: deque[tuple[str, tuple[str, ...]]] = 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 real-package dependency path exists from {focus} to {target}") + + +def _real_source_direct( + project_root: Path, + inventory: Mapping[str, Path], + task: TaskSpec, +) -> WorkflowResult: + imports, inspected_bytes = _read_real_module_dependencies( + project_root, + inventory, + task.focus, + ) + relative = inventory[task.focus].relative_to(project_root).as_posix() + locations = {(task.focus, target): (relative, line) for target, line in imports} + edges = tuple((task.focus, target) for target, _ in imports) + return WorkflowResult( + answer=tuple(target for target, _ in imports), + provenance=_real_provenance(edges, locations), + inspected_bytes=inspected_bytes, + ) + + +def _real_source_impact( + project_root: Path, + inventory: Mapping[str, Path], + task: TaskSpec, +) -> WorkflowResult: + dependencies, locations, inspected_bytes = _read_all_real_dependencies( + project_root, + inventory, + ) + reverse: dict[str, list[str]] = {module_name: [] for module_name in dependencies} + for source, targets in dependencies.items(): + for target in targets: + reverse[target].append(source) + queue: deque[tuple[str, int]] = deque([(task.focus, 0)]) + seen = {task.focus} + discovered: set[str] = set() + edges: list[tuple[str, str]] = [] + 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(sorted(discovered)), + provenance=_real_provenance(sorted(edges), locations), + inspected_bytes=inspected_bytes, + ) + + +def _real_source_path( + project_root: Path, + inventory: Mapping[str, Path], + task: TaskSpec, +) -> WorkflowResult: + if task.target is None: + raise TaskEvidenceError("Real dependency-path task has no target") + dependencies, locations, inspected_bytes = _read_all_real_dependencies( + project_root, + inventory, + ) + path = _shortest_named_path(dependencies, task.focus, task.target, task.depth) + edges = tuple(zip(path, path[1:], strict=False)) + return WorkflowResult( + answer=path, + provenance=_real_provenance(edges, locations), + inspected_bytes=inspected_bytes, + ) + + +def run_real_source_task( + project_root: Path, + inventory: Mapping[str, Path], + task: TaskSpec, +) -> WorkflowResult: + if task.kind == "direct_dependencies": + return _real_source_direct(project_root, inventory, task) + if task.kind == "bounded_impact": + return _real_source_impact(project_root, inventory, task) + return _real_source_path(project_root, inventory, task) + + +def _real_graph_provenance( + context: _GraphContext, + edges: Sequence[tuple[str, str]], +) -> tuple[Mapping[str, object], ...]: + return tuple( + { + "revision": context.revision, + "source_hash": context.source_hash, + "source_path": context.module_source_paths[source], + "source": source, + "relation": context.relation, + "target": target, + } + for source, target in edges + ) + + +def run_real_graph_task(context: _GraphContext, task: TaskSpec) -> WorkflowResult: + """Answer one task from the bounded immutable real-package import graph.""" + + if context.relation != "imports": + raise TaskEvidenceError("Real-package graph does not publish exact import relationships") + if task.kind == "direct_dependencies": + answer = context.outgoing[task.focus] + edges = tuple((task.focus, target) for target in answer) + response: dict[str, object] = { + "revision": context.revision, + "root": task.focus, + "relation": context.relation, + "results": [ + {"source": source, "relation": context.relation, "target": target} + for source, target in edges + ], + } + elif task.kind == "bounded_impact": + queue: deque[tuple[str, int]] = deque([(task.focus, 0)]) + seen = {task.focus} + discovered: set[str] = set() + edge_list: list[tuple[str, str]] = [] + while queue: + current, current_depth = queue.popleft() + if current_depth >= task.depth: + continue + for source in context.incoming[current]: + if source in seen: + continue + seen.add(source) + discovered.add(source) + edge_list.append((source, current)) + queue.append((source, current_depth + 1)) + answer = tuple(sorted(discovered)) + edges = tuple(sorted(edge_list)) + response = { + "revision": context.revision, + "root": task.focus, + "relation": context.relation, + "depth": task.depth, + "results": [ + {"source": source, "relation": context.relation, "target": target} + for source, target in edges + ], + } + else: + if task.target is None: + raise TaskEvidenceError("Real graph path task has no target") + answer = _shortest_named_path( + context.outgoing, + task.focus, + task.target, + task.depth, + ) + edges = tuple(zip(answer, answer[1:], strict=False)) + response = { + "revision": context.revision, + "root": task.focus, + "target": task.target, + "relation": context.relation, + "path": list(answer), + } + return WorkflowResult( + answer=answer, + provenance=_real_graph_provenance(context, edges), + inspected_bytes=len(_compact_json(response)), + ) + + +def real_package_tasks() -> tuple[tuple[TaskSpec, tuple[str, ...]], ...]: + """Return fixed real-package tasks and answers for the pinned source identity.""" + + return ( + ( + TaskSpec( + task_id="markdown_it_renderer_direct_dependencies", + kind="direct_dependencies", + focus="markdown_it.renderer", + target=None, + depth=1, + prompt="List the direct local dependencies of markdown_it.renderer.", + ), + ( + "markdown_it.common.utils", + "markdown_it.token", + "markdown_it.utils", + ), + ), + ( + TaskSpec( + task_id="markdown_it_html_blocks_three_level_impact", + kind="bounded_impact", + focus="markdown_it.common.html_blocks", + target=None, + depth=3, + prompt=( + "List every module within three reverse dependency steps of " + "markdown_it.common.html_blocks." + ), + ), + ( + "markdown_it.parser_block", + "markdown_it.rules_block", + "markdown_it.rules_block.html_block", + ), + ), + ( + TaskSpec( + task_id="markdown_it_cli_to_core_state_path", + kind="dependency_path", + focus="markdown_it.cli.parse", + target="markdown_it.rules_core.state_core", + depth=3, + prompt=( + "Find one shortest local dependency path from markdown_it.cli.parse " + "to markdown_it.rules_core.state_core." + ), + ), + ( + "markdown_it.cli.parse", + "markdown_it.main", + "markdown_it.rules_core.state_core", + ), + ), + ) + + def _measure( operation: Callable[[], WorkflowResult], *, @@ -683,21 +1338,20 @@ def _lower_is_better( def _task_evidence( - root: Path, - context: _GraphContext, task: TaskSpec, *, - source_count: int, + expected: tuple[str, ...], samples: int, + graph_operation: Callable[[], WorkflowResult], + source_operation: Callable[[], WorkflowResult], ) -> dict[str, object]: - expected = answer_key(task, source_count) graph = _measure( - lambda: run_graph_task(context, task, source_count), + graph_operation, workflow="graph_assisted", samples=samples, ) source = _measure( - lambda: run_source_task(root, task, source_count), + source_operation, workflow="source_only", samples=samples, ) @@ -758,14 +1412,20 @@ def build_task_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) + context, preparation = _prepare_graph( + root, + source_bytes, + expected_names={_module_name(index) for index in range(source_count)}, + project_id="milestone5-generated-task-evidence", + title="Milestone 5 generated comparative task evidence", + ) tasks = [ _task_evidence( - root, - context, task, - source_count=source_count, + expected=answer_key(task, source_count), samples=samples, + graph_operation=lambda task=task: run_graph_task(context, task, source_count), + source_operation=lambda task=task: run_source_task(root, task, source_count), ) for task in fixture_tasks(source_count) ] @@ -826,18 +1486,155 @@ def build_task_evidence( } +def build_real_package_evidence( + root: Path, + *, + samples: int, +) -> dict[str, object]: + """Copy and evaluate the exact lock-pinned installed package source tree.""" + + if samples < 1: + raise ValueError("samples must be positive") + installed_root, identity = installed_real_package() + _copy_real_package_read_only(installed_root, root) + inventory = _real_module_inventory(root) + expected_names = set(inventory) + context, preparation = _prepare_graph( + root, + PINNED_REAL_SOURCE_BYTES, + expected_names=expected_names, + project_id="milestone5-real-task-evidence", + title="Milestone 5 markdown-it-py comparative task evidence", + loader=_RealPackageDependencyAdapter(root), + relation="imports", + operation="cold pinned-source import projection plus module identity map", + ) + task_definitions = real_package_tasks() + tasks = [ + _task_evidence( + task, + expected=expected, + samples=samples, + graph_operation=lambda task=task: run_real_graph_task(context, task), + source_operation=lambda task=task: run_real_source_task(root, inventory, task), + ) + for task, expected in task_definitions + ] + semantic_evidence = { + "identity": identity, + "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": "installed_lockfile_pinned_python_distribution", + "network": False, + "installed_source_mutation": False, + "external_project_mutation": False, + "self_hosting": False, + "production_bindings": False, + **identity, + "copy": "Python sources copied into a temporary read-only project tree", + "graph_projection": ( + "Exact local imports from the Python reference adapter manifest, published " + "as cycle-preserving imports relationships" + ), + "source_only_extraction": ( + "Independent stdlib AST import scan checked against fixed answers" + ), + }, + "preparation": { + "graph_assisted": preparation, + "source_only": { + "persistent_preparation": False, + "operation": ( + "none; module paths are inventoried and each task reads the source " + "required by its fixed 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 build_all_task_evidence( + root: Path, + *, + generated_source_count: int, + samples: int, +) -> dict[str, object]: + """Run both the controlled-scale and pinned-real-package evidence tracks.""" + + generated = build_task_evidence( + root / "generated", + source_count=generated_source_count, + samples=samples, + ) + real_package = build_real_package_evidence( + root / "real-package", + samples=samples, + ) + track_hashes = { + "generated_scale": cast(dict[str, object], generated["summary"])[ + "semantic_evidence_sha256" + ], + "installed_real_package": cast(dict[str, object], real_package["summary"])[ + "semantic_evidence_sha256" + ], + } + return { + "tracks": { + "generated_scale": generated, + "installed_real_package": real_package, + }, + "summary": { + "track_count": 2, + "task_count": 6, + "both_workflows_exact_for_all_tasks": True, + "track_semantic_evidence_sha256": track_hashes, + "semantic_evidence_sha256": _sha256(track_hashes), + }, + } + + 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( + measurement = build_all_task_evidence( Path(directory).resolve(), - source_count=source_count, + generated_source_count=source_count, samples=samples, ) report: dict[str, object] = { - "schema_version": 1, + "schema_version": 2, "benchmark": "docforge2_milestone5_representative_tasks", "mode": arguments.mode, "source": { @@ -853,9 +1650,18 @@ def main() -> int: "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", + "answer_key": ( + "Independent arithmetic oracle for generated tasks and fixed reviewed answers " + "for the exact pinned real-package source identity" + ), + "graph_workflow": ( + "bounded ProjectIndex dependency/backlink responses for generated tasks and " + "bounded immutable imports-projection responses for the pinned real package" + ), + "source_workflow": ( + "bounded literal source reads with fixed generated-import parsing or stdlib AST " + "import inspection for the pinned real package" + ), "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"