emitter/bundle: record executable-intent explicitly, not via stat()

src/repo2rlenv/cli.py

@@ -498,6 +498,35 @@ def cmd_validate(args: argparse.Namespace) -> int:
     return 0 if failures == 0 else 1
 
 
+def cmd_migrate(args: argparse.Namespace) -> int:
+    from repo2rlenv.pipelines.pr_diff import migrate_pr_diff_task
+
+    dataset_dir = Path(args.path).expanduser().resolve()
+    task_dirs = sorted({tf.parent for tf in dataset_dir.rglob("task.toml")})
+    if not task_dirs:
+        console.error(f"no task.toml files found under {dataset_dir}")
+        return 1
+
+    apply = getattr(args, "apply", False)
+    label = "Migrating" if apply else "Auditing (dry run — pass --apply to write)"
+    counts: dict[str, int] = {}
+    errors: list[dict[str, str]] = []
+    with console.section(f"{label} pr_diff tasks under {dataset_dir}"):
+        for task_dir in task_dirs:
+            result = migrate_pr_diff_task(task_dir, dry_run=not apply)
+            action = result["action"]
+            counts[action] = counts.get(action, 0) + 1
+            if action == "error":
+                errors.append(result)
+            elif action in ("migrated", "would_migrate"):
+                console.success(f"{result['task']}: {action}")
+        for err in errors:
+            console.error(f"{err['task']}: {err['detail']}")
+
+    console.kv(counts, title="pr_diff migration" if apply else "pr_diff migration (dry run)")
+    return 0 if not errors else 1
+
+
 class _Backend:
     """Which source/destination a URI points at."""
 
@@ -1101,6 +1130,22 @@ def _dispatch(argv: list[str]) -> int:
     )
     v.set_defaults(func=cmd_validate)
 
+    # migrate
+    mg = sub.add_parser("migrate", help="Repair already-emitted tasks after a generator fix")
+    mg_sub = mg.add_subparsers(dest="migration", required=True)
+    mg_pr_diff = mg_sub.add_parser(
+        "pr-diff",
+        help=(
+            "repair pr_diff tasks emitted before #145's oracle-isolation fix "
+            "(oracle baked into the agent's own image)"
+        ),
+    )
+    mg_pr_diff.add_argument("path", help="dataset or task directory")
+    mg_pr_diff.add_argument(
+        "--apply", action="store_true", help="write the fix (default: dry-run report only)"
+    )
+    mg_pr_diff.set_defaults(func=cmd_migrate)
+
     # push
     p = sub.add_parser("push", help="Push a local dataset directory to HF Hub")
     p.add_argument(

src/repo2rlenv/emitter/bundle.py

@@ -11,6 +11,7 @@
 import os
 import re
 import shutil
+import sys
 import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path, PurePosixPath
@@ -21,6 +22,8 @@
 _SLUG = re.compile(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}\Z")
 _ROLES = frozenset({"environment", "solution", "tests"})
 _HASH_FIELD = "bundle_hash"
+_TRACKED_FIELD = "tracked_files"
+_EXECUTABLE_FIELD = "executable_files"
 
 
 def relative_asset_path(value: str) -> PurePosixPath:
@@ -108,34 +111,82 @@ def configuration(self) -> dict[str, Any]:
         }
 
 
+def _tracked_files(files: dict[str, TaskFile]) -> list[str]:
+    """Every role-scoped path this bundle originally declared, regardless of
+    executable-ness. instruction.md/task.toml are handled separately and
+    never appear here — see relative_asset_path."""
+    return sorted(name for name in files if name != "instruction.md")
+
+
+def _executable_files(files: dict[str, TaskFile]) -> list[str]:
+    return sorted(name for name, asset in files.items() if asset.executable)
+
+
 def _identity(configuration: dict, files: dict[str, TaskFile]) -> str:
     # TOML parsing normalizes the stored representation before hashing on either
     # side, so dictionary insertion order and serialization layout cannot matter.
     import tomllib
 
     normalized = tomllib.loads(tomli_w.dumps(configuration))
-    normalized.get("metadata", {}).get("repo2env", {}).pop(_HASH_FIELD, None)
+    repo2env = normalized.get("metadata", {}).get("repo2env", {})
+    repo2env.pop(_HASH_FIELD, None)
     # Evaluation is an advisory overlay, not executable task content. Excluding
     # only this namespace preserves existing unlabeled bundle identities; every
     # other configuration field and every task file remains bound to the hash.
-    normalized.get("metadata", {}).get("repo2env", {}).pop("evaluation", None)
-    record = {
-        "configuration": normalized,
-        "files": {
-            name: {
-                "sha256": hashlib.sha256(asset.content).hexdigest(),
-                "mode": asset.mode,
-            }
+    repo2env.pop("evaluation", None)
+    if _TRACKED_FIELD in repo2env:
+        # tracked_files/executable_files are themselves part of `normalized`
+        # (hashed below), so a per-file "mode" would be redundant for a file
+        # they cover — and re-deriving it from a stat() on the read side is
+        # exactly the lossy round-trip these fields replace (NTFS has no
+        # POSIX execute bit for regular files at all). A file NOT in
+        # tracked_files (added to the directory after emission by other
+        # tooling, e.g. the quality loop) isn't covered by either field, so
+        # its mode still needs to ride along in the hash exactly as before —
+        # otherwise a real chmod tamper on that file would go undetected.
+        tracked_set = set(repo2env.get(_TRACKED_FIELD, []))
+        file_record = {}
+        for name, asset in sorted(files.items()):
+            entry: dict[str, Any] = {"sha256": hashlib.sha256(asset.content).hexdigest()}
+            if name not in tracked_set:
+                entry["mode"] = asset.mode
+            file_record[name] = entry
+    else:
+        # Legacy bundle (emitted before tracked_files existed): keep the
+        # exact original hash shape so already-published bundle identities
+        # never change.
+        file_record = {
+            name: {"sha256": hashlib.sha256(asset.content).hexdigest(), "mode": asset.mode}
             for name, asset in sorted(files.items())
-        },
-    }
+        }
+    record = {"configuration": normalized, "files": file_record}
     raw = json.dumps(record, sort_keys=True, separators=(",", ":"), default=str)
     return "sha256:" + hashlib.sha256(raw.encode()).hexdigest()
 
 
 def bundle_hash(bundle: TaskBundle) -> str:
     files = {"instruction.md": TaskFile.text(bundle.instruction), **bundle.files}
-    return _identity(bundle.configuration(), files)
+    configuration = bundle.configuration()
+    configuration["metadata"]["repo2env"][_TRACKED_FIELD] = _tracked_files(files)
+    configuration["metadata"]["repo2env"][_EXECUTABLE_FIELD] = _executable_files(files)
+    return _identity(configuration, files)
+
+
+def _validated_asset_path_set(raw: object, *, field: str) -> set[str]:
+    """Validate a task.toml path-list field before trusting it.
+
+    Each entry must be a well-formed, role-scoped asset path (the same shape
+    `relative_asset_path` enforces for every other asset), and the list must
+    not contain duplicates — either would indicate a hand-edited or corrupted
+    manifest, not a real bundle emitted by `write_bundle`.
+    """
+    if not isinstance(raw, list) or not all(isinstance(item, str) for item in raw):
+        raise ValueError(f"{field} must be a list of strings")
+    if len(set(raw)) != len(raw):
+        raise ValueError(f"{field} contains duplicate entries")
+    for entry in raw:
+        relative_asset_path(entry)
+    return set(raw)
 
 
 def inspect_bundle(path: Path) -> dict[str, Any]:
@@ -144,6 +195,19 @@ def inspect_bundle(path: Path) -> dict[str, Any]:
 
     if path.is_symlink() or not path.is_dir():
         raise ValueError("Task directory must be a real directory")
+    configuration = tomllib.loads((path / "task.toml").read_text(encoding="utf-8"))
+    repo2env = configuration.get("metadata", {}).get("repo2env", {})
+    raw_tracked = repo2env.get(_TRACKED_FIELD)
+    if raw_tracked is None:
+        tracked = executable = None
+    else:
+        tracked = _validated_asset_path_set(raw_tracked, field=_TRACKED_FIELD)
+        executable = _validated_asset_path_set(
+            repo2env.get(_EXECUTABLE_FIELD, []), field=_EXECUTABLE_FIELD
+        )
+        if not executable <= tracked:
+            raise ValueError(f"{_EXECUTABLE_FIELD} contains a path outside {_TRACKED_FIELD}")
+
     files: dict[str, TaskFile] = {}
     for item in sorted(path.rglob("*")):
         if item.is_symlink():
@@ -153,33 +217,63 @@ def inspect_bundle(path: Path) -> dict[str, Any]:
                 continue
             raise ValueError(f"Task contains a special file: {item.relative_to(path)}")
         relative = item.relative_to(path).as_posix()
-        mode = item.stat().st_mode & 0o7777
-        if mode not in {0o644, 0o755}:
-            raise ValueError(f"Unexpected task asset mode: {relative}: {oct(mode)}")
         if relative not in {"instruction.md", "task.toml"}:
             relative_asset_path(relative)
+        if tracked is not None and (
+            relative in tracked or relative in {"instruction.md", "task.toml"}
+        ):
+            # A file this bundle originally declared (instruction.md and
+            # task.toml are always non-executable by construction, never
+            # listed in tracked_files, but get the same platform-safe
+            # treatment): executable-intent comes from the (hashed,
+            # tamper-evident) manifest, not a stat() call that can't answer
+            # correctly on every platform.
+            is_executable = relative in executable
+            if sys.platform != "win32":
+                # Real chmod-tamper detection, exactly as strict as the
+                # legacy check below — NTFS can't persist this bit at all, so
+                # there's nothing meaningful to cross-check on Windows; the
+                # hash still covers executable-intent via the manifest above.
+                mode = item.stat().st_mode & 0o7777
+                expected = 0o755 if is_executable else 0o644
+                if mode != expected:
+                    raise ValueError(f"Unexpected task asset mode: {relative}: {oct(mode)}")
+        else:
+            # Legacy bundle, OR a file added to this directory after emission
+            # by other tooling (e.g. the quality loop appending an evidence
+            # artifact) that this manifest was never asked to track. Fall
+            # back to the original stat()-derived check, unaffected by this
+            # fix either way.
+            mode = item.stat().st_mode & 0o7777
+            if mode not in {0o644, 0o755}:
+                raise ValueError(f"Unexpected task asset mode: {relative}: {oct(mode)}")
+            is_executable = mode == 0o755
         if relative != "task.toml":
-            files[relative] = TaskFile(item.read_bytes(), mode == 0o755)
-    configuration = tomllib.loads((path / "task.toml").read_text())
-    claimed = configuration.get("metadata", {}).get("repo2env", {}).get(_HASH_FIELD)
+            files[relative] = TaskFile(item.read_bytes(), is_executable)
+    claimed = repo2env.get(_HASH_FIELD)
     actual = _identity(configuration, files)
     return {"bundle_hash": actual, "claimed_hash": claimed, "integrity_passed": claimed == actual}
 
 
 def write_bundle(bundle: TaskBundle, destination: Path, *, resume: bool = False) -> Path:
     """Publish a fully materialized task, refusing existing targets or collisions."""
     configuration = bundle.configuration()
-    configuration["metadata"]["repo2env"][_HASH_FIELD] = bundle_hash(bundle)
+    files = {
+        "instruction.md": TaskFile.text(bundle.instruction),
+        **bundle.files,
+    }
+    # Computed against the SAME configuration + files that get written below
+    # (rather than via a second bundle_hash(bundle) call building its own
+    # configuration) so what's hashed and what's on disk can never diverge.
+    configuration["metadata"]["repo2env"][_TRACKED_FIELD] = _tracked_files(files)
+    configuration["metadata"]["repo2env"][_EXECUTABLE_FIELD] = _executable_files(files)
+    configuration["metadata"]["repo2env"][_HASH_FIELD] = _identity(configuration, files)
     from repo2rlenv.emitter.evaluation import generated_evaluation
 
     configuration["metadata"]["repo2env"]["evaluation"] = generated_evaluation(
         subject_bundle_hash=configuration["metadata"]["repo2env"][_HASH_FIELD]
     )
-    files = {
-        "instruction.md": TaskFile.text(bundle.instruction),
-        "task.toml": TaskFile.text(tomli_w.dumps(configuration)),
-        **bundle.files,
-    }
+    files["task.toml"] = TaskFile.text(tomli_w.dumps(configuration))
     destination.mkdir(parents=True, exist_ok=True)
     target = destination / bundle.name
     lock = destination / f".{bundle.name}.lock"