Tooling: Windows portability for research onboarding scripts and tests (no new kappa)

scripts/choose_research_task.py

@@ -135,7 +135,7 @@ def main():
         try:
             base = subprocess.run(
                 ["git", "rev-parse", "HEAD"], cwd=ROOT, check=True,
-                capture_output=True, text=True,
+                capture_output=True, text=True, encoding="utf-8",
             ).stdout.strip()
         except (OSError, subprocess.CalledProcessError):
             base = None

scripts/find_overlap.py

@@ -12,6 +12,7 @@
 import os
 from pathlib import Path
 import subprocess
+import sys
 import tempfile
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -25,7 +26,7 @@ def now():
 def gh_json(args, payload=None):
     result = subprocess.run(
         ["gh", "api", *args], input=None if payload is None else json.dumps(payload),
-        text=True, capture_output=True, timeout=90, check=True,
+        text=True, encoding="utf-8", capture_output=True, timeout=90, check=True,
     )
     value = json.loads(result.stdout)
     if isinstance(value, dict) and value.get("errors"):
@@ -100,7 +101,7 @@ def fetch_discussions():
 
 def atomic_json(path, data):
     path.parent.mkdir(parents=True, exist_ok=True)
-    with tempfile.NamedTemporaryFile(mode="w", dir=path.parent, delete=False) as out:
+    with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent, delete=False) as out:
         tmp = Path(out.name)
         json.dump(data, out)
         out.write("\n")
@@ -119,7 +120,7 @@ def error_summary(exc):
 def remote_snapshot(cache, offline=False, refresh=False, max_age=300):
     warnings = []
     try:
-        snapshot = json.loads(cache.read_text())
+        snapshot = json.loads(cache.read_text(encoding="utf-8"))
         if snapshot.get("schema_version") != 1 or snapshot.get("repository") != REPO:
             raise ValueError("cache schema or repository mismatch")
     except (OSError, ValueError):
@@ -212,7 +213,7 @@ def main():
         ap.error("keep the cache outside the repository")
     terms = list(args.terms)
     if args.task:
-        catalog = json.loads((ROOT / "docs/research/agent-task-catalog.json").read_text())
+        catalog = json.loads((ROOT / "docs/research/agent-task-catalog.json").read_text(encoding="utf-8"))
         cards = [t for tr in catalog["tracks"] for t in tr["tasks"] if t["id"] == args.task]
         if not cards:
             ap.error("unknown task")
@@ -238,6 +239,9 @@ def main():
     if args.json_output:
         print(json.dumps(report, indent=2))
     else:
+        # Excerpts quote arbitrary UTF-8; a cp1252 pipe must escape them, not crash.
+        if hasattr(sys.stdout, "reconfigure"):
+            sys.stdout.reconfigure(errors="backslashreplace")
         print(report["status"])
         print("Coverage: " + json.dumps(coverage))
         print(report["limits"])

scripts/new_research_package.py

@@ -95,7 +95,7 @@ def create(root, slug, task_id, author, sources=(), base="HEAD", budget=60):
     temporary = Path(tempfile.mkdtemp(prefix=".research-draft-", dir=destination.parent))
     try:
         for name, content in files.items():
-            (temporary / name).write_text(content)
+            (temporary / name).write_text(content, encoding="utf-8", newline="\n")
         # mkdir is the exclusive claim; rename alone can replace an empty
         # directory that another agent created after the initial existence check.
         destination.mkdir()

scripts/research_package.py

@@ -45,7 +45,7 @@ def validate(root, package, check_pins=True):
             "package must be research/<slug>")
     data = {}
     for name in ("TASK.json", "SOURCES.json", "VERIFY.json"):
-        data[name] = json.loads(relative_file(root, package + "/" + name).read_text())
+        data[name] = json.loads(relative_file(root, package + "/" + name).read_text(encoding="utf-8"))
         require(data[name].get("schema_version") == 1, "unsupported schema: " + name)
     task, sources, manifest = (data[n] for n in ("TASK.json", "SOURCES.json", "VERIFY.json"))
     require(task["scope_status"] in ("unchanged", "refined", "changed"), "declare scope_status")
@@ -64,7 +64,7 @@ def validate(root, package, check_pins=True):
     for question in questions:
         for field in ("question", "why_it_matters", "first_step", "blocked_by"):
             require(meaningful(question.get(field)), "complete next-question field: " + field)
-    readme = relative_file(root, package + "/README.md").read_text()
+    readme = relative_file(root, package + "/README.md").read_text(encoding="utf-8")
     require("## Open next questions" in readme, "README needs Open next questions")
     require(manifest.get("status") == "ready", "VERIFY.json is a draft; complete it before requesting replay")
     require(manifest.get("python") == "3.12", "generic replay currently uses Python 3.12")
@@ -120,7 +120,10 @@ def run_command(argv, cwd, timeout):
     try:
         code = process.wait(timeout=timeout)
     except subprocess.TimeoutExpired:
-        os.killpg(process.pid, signal.SIGKILL)
+        if hasattr(os, "killpg"):
+            os.killpg(process.pid, signal.SIGKILL)
+        else:  # Windows has no process groups to signal; end the whole child tree instead.
+            subprocess.run(["taskkill", "/F", "/T", "/PID", str(process.pid)], capture_output=True)
         process.wait()
         raise ValueError("verification timed out")
     require(code == 0, f"verification command exited {code}")
@@ -146,7 +149,7 @@ def plan(root, changed):
     packages = []
     for path in sorted((root / "research").glob("*/VERIFY.json")):
         package = path.parent.relative_to(root).as_posix()
-        manifest = json.loads(path.read_text())
+        manifest = json.loads(path.read_text(encoding="utf-8"))
         # Parse only for selection; validate fully in the isolated replay job.
         inputs = manifest.get("inputs", [])
         if (changed.intersection(inputs) or any(p.startswith(package + "/") for p in changed)
@@ -174,7 +177,7 @@ def main():
             matrix = json.dumps({"include": [{"package": p} for p in packages]})
             print(matrix)
             if args.github_output:
-                with args.github_output.open("a") as out:
+                with args.github_output.open("a", encoding="utf-8") as out:
                     out.write(f"matrix={matrix}\ncount={len(packages)}\n")
         elif args.package:
             if args.check_only:

tests/test_research_assignment.py

@@ -13,15 +13,15 @@
 SPEC = importlib.util.spec_from_file_location("choose_research_task", ROOT / "scripts/choose_research_task.py")
 assignment = importlib.util.module_from_spec(SPEC)
 SPEC.loader.exec_module(assignment)
-CATALOG = json.loads((ROOT / assignment.CATALOG).read_text())
+CATALOG = json.loads((ROOT / assignment.CATALOG).read_text(encoding="utf-8"))
 
 
 class ResearchAssignmentTests(unittest.TestCase):
     def test_catalog_sources_and_weights(self):
         assignment.validate_catalog(CATALOG, ROOT)
 
     def test_documented_allocation_matches_catalog(self):
-        guide = (ROOT / "autoresearch.md").read_text()
+        guide = (ROOT / "autoresearch.md").read_text(encoding="utf-8")
         rows = re.findall(r"^\| [^|\n]*\(`([a-z0-9-]+)`\) \| (\d+)% \|", guide, re.MULTILINE)
         self.assertEqual(len(rows), len(CATALOG["tracks"]))
         self.assertEqual(
@@ -97,7 +97,7 @@ def test_cli_dynamic_selected_record_and_receipt(self):
             cwd=ROOT, capture_output=True, text=True, check=True,
         )
         receipt = json.loads(result.stdout)
-        selected = json.loads((ROOT / "certificates/selected-result.json").read_text())
+        selected = json.loads((ROOT / "certificates/selected-result.json").read_text(encoding="utf-8"))
         self.assertEqual(receipt["selected"]["reproduce"], selected["reproduce"])
         self.assertEqual(receipt["selected"]["review"], selected["review"])
         self.assertEqual(receipt["status"], "SUGGESTIONS_ONLY_NOT_RESERVED")

tests/test_research_onboarding.py

@@ -1,6 +1,8 @@
 """Failure controls for overlap coverage, package provenance and bounded replay."""
+import ast
 import importlib.util
 import json
+import os
 from pathlib import Path
 import subprocess
 import sys
@@ -26,6 +28,14 @@ def test_body_only_matches_and_no_clean_certificate(self):
         self.assertEqual(matches[0]["head"], "a" * 40)
         self.assertEqual(overlap.search(records, ["missing"]), [])
 
+    def test_report_survives_a_cp1252_stdout(self):
+        # Local package READMEs quote kappa; forcing cp1252 reproduces a Windows pipe on any host.
+        env = {**os.environ, "PYTHONIOENCODING": "cp1252", "PYTHONUTF8": "0"}
+        result = subprocess.run([sys.executable, "-B", str(ROOT / "scripts/find_overlap.py"), "κ", "--offline"],
+                                cwd=ROOT, env=env, capture_output=True, timeout=120)
+        self.assertEqual(result.returncode, 0, result.stderr.decode("utf-8", "replace"))
+        self.assertIn(b"\\u03ba", result.stdout)
+
     def test_pr_pagination_normalization(self):
         def row(n):
             return {"number": n, "title": "t", "body": "body", "html_url": f"url{n}",
@@ -98,7 +108,10 @@ def setUp(self):
         self.addCleanup(self.temp.cleanup)
         self.root = Path(self.temp.name)
         subprocess.run(["git", "init", "-q", str(self.root)], check=True)
-        for key, value in (("user.name", "Fixture"), ("user.email", "fixture@example.invalid")):
+        # Pin line-ending handling so committed blobs equal working bytes on every host
+        # (Git for Windows defaults to core.autocrlf=true).
+        for key, value in (("user.name", "Fixture"), ("user.email", "fixture@example.invalid"),
+                           ("core.autocrlf", "false")):
             package.git(self.root, "config", key, value)
         (self.root / "docs/research").mkdir(parents=True)
         (self.root / "certificates").mkdir()
@@ -152,6 +165,17 @@ def test_scope_drift_requires_reason(self):
         path.write_text(json.dumps(task))
         package.validate(self.root, "research/fixture")
 
+    def test_utf8_metadata_validates_under_any_locale(self):
+        # Byte 0x9d (in U+201D) is undefined in cp1252, the default Windows codec.
+        task, _ = self.ready()
+        task["evidence"] = "Exact κ moment — “priced”"
+        (self.target / "TASK.json").write_text(json.dumps(task, ensure_ascii=False), encoding="utf-8")
+        readme = self.target / "README.md"
+        readme.write_text(readme.read_text(encoding="utf-8") + "\n## Open next questions\n\n“Which κ?”\n",
+                          encoding="utf-8")
+        validated, _ = package.validate(self.root, "research/fixture")
+        self.assertEqual(validated["evidence"], task["evidence"])
+
     def test_live_input_drift_but_provenance_survives_priority_change(self):
         self.ready()
         (self.root / "docs/research/agent-task-catalog.json").write_text('{"new":"priority"}')
@@ -200,6 +224,29 @@ def test_ready_package_replays_and_mutation_fails(self):
             package.run_package(self.root, "research/fixture")
 
 
+class TextEncodingTests(unittest.TestCase):
+    SCRIPTS = ("choose_research_task.py", "find_overlap.py", "new_research_package.py",
+               "research_baseline.py", "research_package.py")
+
+    def test_onboarding_text_io_does_not_depend_on_host_locale(self):
+        # A UTF-8 CI host cannot observe locale-codec bugs, so check the calls statically.
+        missing = []
+        for name in self.SCRIPTS:
+            for node in ast.walk(ast.parse((ROOT / "scripts" / name).read_text(encoding="utf-8"))):
+                if not isinstance(node, ast.Call):
+                    continue
+                func = getattr(node.func, "attr", getattr(node.func, "id", None))
+                keywords = {k.arg: k.value for k in node.keywords}
+                mode = keywords.get("mode", node.args[0] if func == "open" and node.args else None)
+                binary = isinstance(mode, ast.Constant) and "b" in str(mode.value)
+                text = any(isinstance(keywords.get(k), ast.Constant) and keywords[k].value is True
+                           for k in ("text", "universal_newlines"))
+                if ((func in ("read_text", "write_text", "open", "NamedTemporaryFile") and not binary
+                        or func in ("run", "Popen", "check_output") and text) and "encoding" not in keywords):
+                    missing.append(f"{name}:{node.lineno}")
+        self.assertEqual(missing, [])
+
+
 class BaselineTests(unittest.TestCase):
     def test_binding_and_target_are_derived(self):
         record = {"kappa": "1/1000", "review": "r", "certificate": "c", "reproduce": "command"}