Allow research task selection from sparse checkouts

docs/research/agent-procedure.md

@@ -121,6 +121,13 @@ that a task is open. A deterministic seed is a reproducibility aid, not novelty.
 Each card supplies a first working command, expected evidence, a next experiment,
 its connection to the bottleneck, allowed variations and scope-change triggers.
 
+In a sparse checkout, the helper can validate omitted catalogue references when
+they are regular files in `HEAD` with matching skip-worktree index entries. It
+reports these as `sparse_checkout.omitted_catalogue_paths`. For absent paths,
+ordinary deletions and staged content replacements still fail. Materialize the
+chosen card's inputs before running its commands. This reference check does not
+resolve replay dependencies.
+
 Use the overlap helper before choosing a card and refresh before submission:
 
 ```sh

scripts/choose_research_task.py

@@ -25,15 +25,58 @@ def canonical(value):
     return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
 
 
+def check_catalog_files(root, required):
+    """Allow deliberate sparse omissions, not deletions or changed index inputs."""
+    missing = {}
+    for name, error in required:
+        path = root / name
+        if not path.resolve().is_relative_to(root.resolve()):
+            raise ValueError(error)
+        if path.is_file():
+            continue
+        if path.exists() or path.is_symlink() or path.resolve() != root.resolve() / name:
+            raise ValueError(error)
+        missing[name] = error
+    if not missing:
+        return []
+    omitted = {}
+    try:
+        def git(*args):
+            return subprocess.run(
+                ["git", "--literal-pathspecs", *args, "--", *missing], cwd=root,
+                check=True, capture_output=True, text=True, encoding="utf-8",
+            ).stdout.split("\0")
+        for entry in git("ls-files", "-t", "--stage", "-z"):
+            if not entry:
+                continue
+            metadata, name = entry.split("\t", 1)
+            flag, mode, blob, stage = metadata.split()
+            if flag == "S" and mode in ("100644", "100755") and stage == "0":
+                omitted[name] = blob
+        committed = {}
+        for entry in git("ls-tree", "-r", "-z", "HEAD"):
+            if not entry:
+                continue
+            metadata, name = entry.split("\t", 1)
+            mode, kind, blob = metadata.split()
+            if kind == "blob" and mode in ("100644", "100755"):
+                committed[name] = blob
+    except (OSError, subprocess.CalledProcessError):
+        omitted = {}
+        committed = {}
+    for name, error in missing.items():
+        if name not in omitted or omitted[name] != committed.get(name):
+            raise ValueError(error)
+    return sorted(missing)
+
+
 def validate_catalog(catalog, root):
     if catalog.get("schema_version") != 1:
         raise ValueError("unsupported catalogue schema")
     revision = catalog.get("priority_revision")
     if type(revision) is not int or revision < 1:
         raise ValueError("priority_revision must be a positive integer")
-    log = root / catalog["priority_log"]
-    if not log.resolve().is_relative_to(root.resolve()) or not log.is_file():
-        raise ValueError("missing or external priority log")
+    required = [(catalog["priority_log"], "missing or external priority log")]
     tracks = catalog.get("tracks", [])
     if not tracks:
         raise ValueError("empty catalogue")
@@ -78,14 +121,12 @@ def validate_catalog(catalog, root):
             if not step.get("expected") or not step.get("inspect") or not step.get("command"):
                 raise ValueError("incomplete first step")
             script = step["command"][-1]
-            if not (root / script).is_file():
-                raise ValueError("missing first-step script: " + script)
+            required.append((script, "missing first-step script: " + script))
             for name in task["starting_points"]:
-                path = root / name
-                if not path.resolve().is_relative_to(root.resolve()) or not path.is_file():
-                    raise ValueError(f"missing or external starting point: {name}")
+                required.append((name, f"missing or external starting point: {name}"))
     if sum(t["weight"] for t in tracks) != 100:
         raise ValueError("track weights must sum to 100")
+    return check_catalog_files(root, required)
 
 
 def scope_key(task_id, method):
@@ -149,7 +190,7 @@ def main():
     try:
         catalog_bytes = (ROOT / CATALOG).read_bytes()
         catalog = json.loads(catalog_bytes)
-        validate_catalog(catalog, ROOT)
+        omitted = validate_catalog(catalog, ROOT)
         result = select(catalog, args.seed if args.seed is not None else secrets.token_hex(16), args.track, args.task)
         selected_bytes = (ROOT / "certificates/selected-result.json").read_bytes()
         selected = json.loads(selected_bytes)
@@ -186,6 +227,11 @@ def main():
             "status": "SUGGESTIONS_ONLY_NOT_RESERVED",
             "next_step": "First check #1166 for a suitable open handoff. Use these suggestions if none fits, check live overlap and existing coverage, then record a bounded scope. Claims never exclude independent audits.",
         })
+        if omitted:
+            result["sparse_checkout"] = {
+                "omitted_catalogue_paths": omitted,
+                "scope": "These absent regular files match HEAD and skip-worktree index entries. Materialize the chosen task's inputs before running it; this validates catalogue references, not replay dependencies.",
+            }
     except (OSError, ValueError, KeyError, TypeError) as exc:
         parser.error(str(exc))
     print(json.dumps(result, indent=2))

tests/test_research_assignment.py

@@ -7,7 +7,9 @@
 import re
 import subprocess
 import sys
+import tempfile
 import unittest
+from unittest import mock
 
 ROOT = Path(__file__).resolve().parents[1]
 SPEC = importlib.util.spec_from_file_location("choose_research_task", ROOT / "scripts/choose_research_task.py")
@@ -106,5 +108,135 @@ def test_cli_dynamic_selected_record_and_receipt(self):
         self.assertEqual(len(receipt["catalog_sha256"]), 64)
 
 
+class SparseResearchAssignmentTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name)
+        self.catalog = copy.deepcopy(CATALOG)
+        self.catalog["tracks"] = [self.catalog["tracks"][0]]
+        self.catalog["tracks"][0]["weight"] = 100
+        self.catalog["tracks"][0]["tasks"] = [self.catalog["tracks"][0]["tasks"][0]]
+        self.catalog["priority_log"] = "priority.log"
+        card = self.catalog["tracks"][0]["tasks"][0]
+        card["first_step"]["command"] = ["python3", "-B", "programs/first.py"]
+        card["starting_points"] = ["notes/entry.md"]
+        for name, body in {
+            "priority.log": "fixture log\n",
+            "programs/first.py": "print('fixture')\n",
+            "notes/entry.md": "fixture note\n",
+            str(assignment.CATALOG): json.dumps(self.catalog),
+            "certificates/selected-result.json": json.dumps({
+                "kappa": "1/1000", "review": "review.md",
+                "source_manifest": "source.json", "reproduce": "fixture-only",
+            }),
+            "scripts/choose_research_task.py": (ROOT / "scripts/choose_research_task.py").read_text(),
+        }.items():
+            path = self.root / name
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(body, encoding="utf-8")
+        self.git("init", "-q")
+        self.git("add", ".")
+        self.git("-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid",
+                 "-c", "core.hooksPath=/dev/null", "commit", "-qm", "fixture")
+
+    def git(self, *args):
+        return subprocess.run(["git", *args], cwd=self.root, check=True,
+                              capture_output=True, text=True).stdout
+
+    def sparse(self):
+        self.git("sparse-checkout", "init", "--no-cone")
+        self.git("sparse-checkout", "set", "/scripts/", "/docs/", "/certificates/")
+
+    def receipt(self):
+        result = subprocess.run(
+            [sys.executable, "-B", "scripts/choose_research_task.py", "--seed", "a1" * 16],
+            cwd=self.root, capture_output=True, text=True, check=True,
+        )
+        return json.loads(result.stdout)
+
+    def test_sparse_cli_preserves_assignment_and_reports_omitted_inputs(self):
+        full = self.receipt()
+        self.sparse()
+        self.assertFalse((self.root / "programs/first.py").exists())
+        sparse = self.receipt()
+        self.assertEqual(sparse.pop("sparse_checkout")["omitted_catalogue_paths"],
+                         ["notes/entry.md", "priority.log", "programs/first.py"])
+        self.assertEqual(sparse, full)
+        self.assertFalse((self.root / "programs/first.py").exists())
+
+    def test_cone_sparse_index_also_validates_omitted_files(self):
+        self.git("sparse-checkout", "init", "--cone", "--sparse-index")
+        self.git("sparse-checkout", "set", "scripts", "docs", "certificates")
+        self.assertEqual(assignment.validate_catalog(self.catalog, self.root),
+                         ["notes/entry.md", "programs/first.py"])
+
+    def test_ordinary_deletion_and_unknown_path_still_fail(self):
+        (self.root / "notes/entry.md").unlink()
+        with self.assertRaisesRegex(ValueError, "entry.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+        self.sparse()
+        self.catalog["tracks"][0]["tasks"][0]["starting_points"] = ["unknown.md"]
+        with self.assertRaisesRegex(ValueError, "unknown.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+
+    def test_staged_sparse_change_must_match_head(self):
+        self.sparse()
+        path = self.root / "notes/entry.md"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text("staged replacement\n")
+        self.git("add", "--sparse", "notes/entry.md")
+        self.git("update-index", "--skip-worktree", "notes/entry.md")
+        path.unlink()
+        with self.assertRaisesRegex(ValueError, "entry.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+
+    def test_staged_only_sparse_file_has_no_head_source(self):
+        self.sparse()
+        path = self.root / "new.md"
+        path.write_text("new staged file\n")
+        self.git("add", "--sparse", "new.md")
+        self.git("update-index", "--skip-worktree", "new.md")
+        path.unlink()
+        self.catalog["tracks"][0]["tasks"][0]["starting_points"] = ["new.md"]
+        with self.assertRaisesRegex(ValueError, "new.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+
+    def test_missing_sparse_symlink_is_not_a_regular_source(self):
+        path = self.root / "notes/entry.md"
+        path.unlink()
+        try:
+            path.symlink_to("../priority.log")
+        except OSError:
+            self.skipTest("symlinks unavailable")
+        self.git("add", "notes/entry.md")
+        self.git("-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid",
+                 "-c", "core.hooksPath=/dev/null", "commit", "-qm", "symlink fixture")
+        self.sparse()
+        with self.assertRaisesRegex(ValueError, "entry.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+
+    def test_directory_and_external_alias_cannot_replace_omitted_file(self):
+        self.sparse()
+        path = self.root / "notes/entry.md"
+        path.mkdir(parents=True)
+        with self.assertRaisesRegex(ValueError, "entry.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+        path.rmdir()
+        try:
+            path.symlink_to(self.root.parent / "outside-missing-chooser-input")
+        except OSError:
+            self.skipTest("symlinks unavailable")
+        with self.assertRaisesRegex(ValueError, "entry.md"):
+            assignment.validate_catalog(self.catalog, self.root)
+
+    def test_git_failure_does_not_accept_missing_files(self):
+        self.sparse()
+        with mock.patch.object(assignment.subprocess, "run",
+                               side_effect=subprocess.CalledProcessError(1, "git")):
+            with self.assertRaises(ValueError):
+                assignment.validate_catalog(self.catalog, self.root)
+
+
 if __name__ == "__main__":
     unittest.main()