fix(windows): support long paths in scan tooling

sdk/typescript/_bundled_plugin/scripts/config_preflight.py

@@ -12,6 +12,10 @@
 from pathlib import Path
 from typing import Any
 
+# Some plugin hosts launch Python with safe-path isolation enabled.
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from windows_paths import filesystem_path, portable_path
+
 try:
     import tomllib
 except ModuleNotFoundError:  # pragma: no cover - Python 3.10 only
@@ -121,7 +125,7 @@ def parse_args() -> argparse.Namespace:
 
 def read_toml(path: Path, *, required: bool) -> dict[str, Any]:
     try:
-        with path.open("rb") as file:
+        with filesystem_path(path).open("rb") as file:
             return tomllib.load(file)
     except FileNotFoundError:
         if required:
@@ -231,7 +235,7 @@ def resolve_project_root(cwd: Path, config_layers: list[tuple[Path, dict[str, An
     if not markers:
         return cwd
     for candidate in (cwd, *cwd.parents):
-        if any((candidate / marker).exists() for marker in markers):
+        if any(filesystem_path(candidate / marker).exists() for marker in markers):
             return candidate
     return cwd
 
@@ -243,7 +247,7 @@ def project_trust_level(
         projects = config.get("projects")
         if not isinstance(projects, dict):
             continue
-        project = projects.get(str(project_root))
+        project = projects.get(str(portable_path(project_root)))
         if not isinstance(project, dict):
             continue
         trust_level = project.get("trust_level")
@@ -265,7 +269,7 @@ def project_config_paths(project_root: Path, cwd: Path) -> list[Path]:
 def discover_config_paths(
     *, cwd: Path, profile_layer_path: Path | None
 ) -> tuple[list[Path], dict[str, Any]]:
-    resolved_cwd = cwd.expanduser().resolve()
+    resolved_cwd = filesystem_path(cwd.expanduser()).resolve()
     if not resolved_cwd.is_dir():
         raise ValueError(f"cwd must be a directory, got {str(resolved_cwd)!r}")
 
@@ -279,8 +283,8 @@ def discover_config_paths(
     if trust_level == "trusted":
         paths.extend(project_config_paths(project_root, resolved_cwd))
     return paths, {
-        "cwd": str(resolved_cwd),
-        "project_root": str(project_root),
+        "cwd": str(portable_path(resolved_cwd)),
+        "project_root": str(portable_path(project_root)),
         "project_trust_level": trust_level,
         "project_layers_loaded": trust_level == "trusted",
     }

sdk/typescript/_bundled_plugin/scripts/deep_scan_workbench.py

@@ -26,6 +26,7 @@
     worktree_content_digest,
 )
 from workbench_validation import optional_text, require_uuid, user_text
+from windows_paths import extended_path, filesystem_path, portable_path
 
 DEEP_SCAN_WORKER_KINDS = ("setup", "discovery", "dedup")
 DEEP_SCAN_WORKER_STATUSES = ("queued", "running", "succeeded", "failed", "canceled")
@@ -196,6 +197,10 @@ def require_canonical_scan_directory(scan_dir: Path) -> Path:
     return dependencies().require_canonical_scan_directory(scan_dir)
 
 
+def scan_directory_path(scan: sqlite3.Row, *parts: str) -> Path:
+    return filesystem_path(Path(scan["scan_dir"]).joinpath(*parts))
+
+
 def safe_segment(value: str) -> str:
     return dependencies().safe_segment(value)
 
@@ -264,25 +269,28 @@ def deep_scan_path(
     if not supplied.is_absolute():
         raise SystemExit(f"{label} must be an absolute path inside the scan directory.")
     try:
-        resolved = supplied.resolve(strict=True)
-        scan_dir = require_canonical_scan_directory(Path(scan["scan_dir"]))
-        resolved.relative_to(scan_dir)
+        resolved = type(supplied)(filesystem_path(supplied)).resolve(strict=True)
+        scan_dir = require_canonical_scan_directory(scan_directory_path(scan))
+        portable_resolved = portable_path(resolved)
+        portable_resolved.relative_to(portable_path(scan_dir))
     except (OSError, RuntimeError, ValueError) as exc:
         raise SystemExit(f"{label} must be an existing path inside the scan directory.") from exc
-    if os.path.normcase(resolved) != os.path.normcase(supplied.absolute()):
+    if os.path.normcase(portable_resolved) != os.path.normcase(
+        portable_path(supplied.absolute())
+    ):
         raise SystemExit(f"{label} must be a canonical non-symlink path.")
     if kind == "file" and not resolved.is_file():
         raise SystemExit(f"{label} must be a regular file.")
     if kind == "directory" and not resolved.is_dir():
         raise SystemExit(f"{label} must be a directory.")
-    return str(resolved)
+    return str(portable_resolved)
 
 
 def deep_scan_output_path(scan: sqlite3.Row, value: str, label: str) -> str:
     supplied = Path(value).expanduser()
     if not supplied.is_absolute():
         raise SystemExit(f"{label} must be an absolute path inside the scan directory.")
-    if supplied.exists():
+    if filesystem_path(supplied).exists():
         return deep_scan_path(scan, str(supplied), label, kind="file")
     parent = Path(deep_scan_path(scan, str(supplied.parent), label, kind="directory"))
     output = parent / supplied.name
@@ -291,12 +299,16 @@ def deep_scan_output_path(scan: sqlite3.Row, value: str, label: str) -> str:
     return str(output)
 
 
+def temporary_artifact_path(path: Path, suffix: str) -> Path:
+    return filesystem_path(path.with_name(f".{path.name}.{uuid.uuid4()}.{suffix}"))
+
+
 def promote_staged_file(staged_path: str, output_path: str) -> tuple[Path, Path, Path | None]:
-    staged = Path(staged_path)
-    output = Path(output_path)
+    staged = filesystem_path(Path(staged_path))
+    output = filesystem_path(Path(output_path))
     if staged == output:
         raise SystemExit("A staged Deep Scan artifact must not be its published output path.")
-    backup = output.with_name(f".{output.name}.{uuid.uuid4()}.backup") if output.exists() else None
+    backup = temporary_artifact_path(output, "backup") if output.exists() else None
     if backup is not None:
         os.replace(output, backup)
     try:
@@ -323,10 +335,13 @@ def finish_staged_file(promotion: tuple[Path, Path, Path | None]) -> None:
 
 
 def canonical_discovery_artifacts(scan: sqlite3.Row) -> dict[str, str]:
-    discovery_dir = Path(scan["scan_dir"]) / "artifacts" / "02_discovery"
     artifacts = {
-        "inScopeFilesPath": discovery_dir / "in_scope_files.txt",
-        "candidateLedgerPath": discovery_dir / "candidate_ledger.jsonl",
+        "inScopeFilesPath": scan_directory_path(
+            scan, "artifacts", "02_discovery", "in_scope_files.txt"
+        ),
+        "candidateLedgerPath": scan_directory_path(
+            scan, "artifacts", "02_discovery", "candidate_ledger.jsonl"
+        ),
     }
     labels = {
         "inScopeFilesPath": "Canonical in-scope inventory path",
@@ -364,15 +379,19 @@ def deep_scan_state(connection: sqlite3.Connection, scan_id: str) -> dict[str, A
         run["canonical_inventory_path"] is None
         and run["status"] == "succeeded"
         and run["manifest_path"] is not None
-        and run["manifest_path"] != str(Path(scan["scan_dir"]) / "scan-manifest.json")
-        and (Path(scan["scan_dir"]) / "artifacts" / "02_discovery" / "in_scope_files.txt").exists()
+        and run["manifest_path"]
+        != str(portable_path(scan_directory_path(scan, "scan-manifest.json")))
+        and scan_directory_path(scan, "artifacts", "02_discovery", "in_scope_files.txt").exists()
     ):
         canonical_artifacts = canonical_discovery_artifacts(scan)
         if (
             run["terminal_reason"] == "capped"
             and run["completion_sequence"] == 0
             and deep_scan_deadline_reached(run)
-            and Path(canonical_artifacts["candidateLedgerPath"]).stat().st_size != 0
+            and filesystem_path(Path(canonical_artifacts["candidateLedgerPath"]))
+            .stat()
+            .st_size
+            != 0
         ):
             raise SystemExit(
                 "A capped Deep Scan without completed discoveries requires an empty "
@@ -691,7 +710,7 @@ def begin_deep_scan_for_target(
     target = require_target(args.target_path)
     require_scannable_target(target)
     scope = require_scope(args.scope, "deep", target)
-    target_path = str(target)
+    target_path = str(portable_path(target))
     existing = existing_deep_scan_for_target(connection, thread_id, target_path, scope)
     if existing is not None:
         return begin_deep_scan_for_scan(connection, existing["id"], thread_id, args)
@@ -757,10 +776,17 @@ def begin_deep_scan_for_target(
         if workflow_version is None:
             raise SystemExit("workflow-version is required.")
         root = (
-            Path(args.scan_root).expanduser().resolve() if args.scan_root else state_dir() / "scans"
+            filesystem_path(Path(args.scan_root).expanduser()).resolve()
+            if args.scan_root
+            else state_dir() / "scans"
         )
-        target_root = (root / safe_segment(target.name)).resolve()
-        if target_root == target or target in target_root.parents:
+        target_root = filesystem_path(root / safe_segment(target.name)).resolve()
+        portable_target = portable_path(target)
+        portable_target_root = portable_path(target_root)
+        if (
+            portable_target_root == portable_target
+            or portable_target in portable_target_root.parents
+        ):
             raise SystemExit("The scan artifact directory must be outside the selected target.")
         target_root.mkdir(parents=True, exist_ok=True)
         user_context = user_text(args.user_context)
@@ -770,10 +796,12 @@ def begin_deep_scan_for_target(
         scan_id = str(uuid.uuid4())
         timestamp = now()
         target_id = ensure_security_target(connection, target_path)
-        scan_dir = Path(
-            tempfile.mkdtemp(
-                prefix=f"{safe_segment(revision)}_{compact_timestamp()}_",
-                dir=target_root,
+        scan_dir = filesystem_path(
+            Path(
+                tempfile.mkdtemp(
+                    prefix=f"{safe_segment(revision)}_{compact_timestamp()}_",
+                    dir=extended_path(target_root),
+                )
             )
         ).resolve()
         connection.execute(
@@ -817,7 +845,7 @@ def begin_deep_scan_for_target(
                 scope,
                 user_context,
                 thread_id,
-                str(scan_dir),
+                str(portable_path(scan_dir)),
                 model,
                 reasoning_effort,
                 timestamp,
@@ -886,11 +914,11 @@ def coordinator_lease_is_live(
             seconds=DEEP_SCAN_LEGACY_COORDINATOR_GRACE_SECONDS
         )
     heartbeat_time = datetime.fromisoformat(str(run["updated_at"]))
-    heartbeat_path = (
-        Path(scan["scan_dir"])
-        / "artifacts"
-        / "deep_discovery"
-        / f"coordinator-heartbeat-{run['coordinator_generation']}.json"
+    heartbeat_path = scan_directory_path(
+        scan,
+        "artifacts",
+        "deep_discovery",
+        f"coordinator-heartbeat-{run['coordinator_generation']}.json",
     )
     try:
         heartbeat = json.loads(heartbeat_path.read_text(encoding="utf-8"))
@@ -1047,7 +1075,7 @@ def recover_expired_coordinator(
 
 def recover_candidate_ledger_publication(connection: sqlite3.Connection, scan_id: str) -> None:
     scan = require_scan(connection, scan_id)
-    ledger = Path(scan["scan_dir"]) / "artifacts" / "02_discovery" / "candidate_ledger.jsonl"
+    ledger = scan_directory_path(scan, "artifacts", "02_discovery", "candidate_ledger.jsonl")
     backups = sorted(
         ledger.parent.glob(f".{ledger.name}.*.backup"),
         key=lambda backup: backup.stat().st_mtime_ns,
@@ -1066,7 +1094,7 @@ def recover_candidate_ledger_publication(connection: sqlite3.Connection, scan_id
         (scan_id,),
     )
     for reducer in reducers:
-        snapshot = Path(reducer["artifact_dir"]) / "canonical" / ledger.name
+        snapshot = filesystem_path(Path(reducer["artifact_dir"]) / "canonical" / ledger.name)
         if not snapshot.exists():
             continue
         published = ledger.exists() and ledger.samefile(snapshot)
@@ -1531,7 +1559,7 @@ def commit_deep_scan_dedup_locked(
                 "Staged candidate ledger path",
                 kind="file",
             )
-            discovery_dir = Path(scan["scan_dir"]) / "artifacts" / "02_discovery"
+            discovery_dir = scan_directory_path(scan, "artifacts", "02_discovery")
             deep_scan_path(
                 scan,
                 str(discovery_dir / "in_scope_files.txt"),
@@ -1567,11 +1595,9 @@ def commit_deep_scan_dedup_locked(
         if not inputs or any(row["merge_state"] != "merging" for row in inputs):
             raise SystemExit("Dedup inputs are not in the claimed merging state.")
         if candidate_ledger_path and canonical_candidate_ledger_path:
-            canonical_path = Path(canonical_candidate_ledger_path)
-            publication_copy = canonical_path.with_name(
-                f".{canonical_path.name}.{uuid.uuid4()}.publish"
-            )
-            os.link(candidate_ledger_path, publication_copy)
+            canonical_path = filesystem_path(Path(canonical_candidate_ledger_path))
+            publication_copy = temporary_artifact_path(canonical_path, "publish")
+            os.link(filesystem_path(Path(candidate_ledger_path)), publication_copy)
             promotion = promote_staged_file(
                 str(publication_copy),
                 canonical_candidate_ledger_path,
@@ -1651,7 +1677,9 @@ def finish_deep_scan_locked(
                 scan, args.manifest_path, "Deep Scan coordinator manifest path", kind="file"
             )
         )
-        standard_scan_manifest = manifest_path == str(Path(scan["scan_dir"]) / "scan-manifest.json")
+        standard_scan_manifest = manifest_path == str(
+            portable_path(scan_directory_path(scan, "scan-manifest.json"))
+        )
 
         failure_capped = False
         if (
@@ -1662,12 +1690,12 @@ def finish_deep_scan_locked(
             for artifact_name in ("scan-manifest.json", "findings.json", "coverage.json"):
                 deep_scan_path(
                     scan,
-                    str(Path(scan["scan_dir"]) / artifact_name),
+                    str(scan_directory_path(scan, artifact_name)),
                     f"Canonical parent {artifact_name}",
                     kind="file",
                 )
             coverage = _read_scan_local_json(
-                Path(scan["scan_dir"]), "coverage.json", "Canonical parent coverage.json"
+                scan_directory_path(scan), "coverage.json", "Canonical parent coverage.json"
             )
             deferred = coverage.get("deferred")
             failure_capped = (
@@ -1766,7 +1794,7 @@ def finish_deep_scan_locked(
             for artifact_name in ("scan-manifest.json", "findings.json", "coverage.json"):
                 deep_scan_path(
                     scan,
-                    str(Path(scan["scan_dir"]) / artifact_name),
+                    str(scan_directory_path(scan, artifact_name)),
                     f"Canonical parent {artifact_name}",
                     kind="file",
                 )
@@ -1792,7 +1820,10 @@ def finish_deep_scan_locked(
             and (
                 standard_scan_manifest
                 or canonical_artifacts is not None
-                and Path(canonical_artifacts["candidateLedgerPath"]).stat().st_size == 0
+                and filesystem_path(Path(canonical_artifacts["candidateLedgerPath"]))
+                .stat()
+                .st_size
+                == 0
             )
         )
         if successful_reducer is None and not zero_discovery_deadline:

sdk/typescript/_bundled_plugin/scripts/finalize_scan_contract.py

@@ -24,6 +24,10 @@
 from typing import Any, TextIO
 from urllib.parse import quote, urlsplit
 
+# Some plugin hosts launch Python with safe-path isolation enabled.
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from windows_paths import filesystem_path
+
 SCHEMA_VERSION = "1.0"
 PRODUCER_NAME = "codex-security-plugin"
 FINGERPRINT_ALGORITHM = "codex-security/v1"
@@ -86,7 +90,7 @@ def _loads_json(value: str | bytes) -> Any:
 
 def _read_json(path: Path) -> dict[str, Any]:
     try:
-        payload = _loads_json(path.read_text(encoding="utf-8"))
+        payload = _loads_json(filesystem_path(path).read_text(encoding="utf-8"))
         _require_safe_json_value(payload, str(path))
     except FileNotFoundError as exc:
         raise ContractError(f"missing required contract artifact: {path}") from exc
@@ -230,7 +234,7 @@ def _require_safe_relative_path(value: str, context: str, *, allow_dot: bool = F
 
 
 def _require_scan_directory(scan_dir: Path) -> Path:
-    scan_dir = scan_dir.absolute()
+    scan_dir = filesystem_path(scan_dir.absolute())
     try:
         metadata = scan_dir.lstat()
     except OSError as exc:
@@ -1643,7 +1647,7 @@ def _github_primary_location_line_hash(
         return None
     primary_location = _sarif_primary_location(finding)
     try:
-        source_root = source_root.resolve(strict=True)
+        source_root = filesystem_path(source_root).resolve(strict=True)
     except (OSError, RuntimeError):
         return None
     relative_path = _require_safe_relative_path(primary_location["path"], "SARIF source location")
@@ -1665,7 +1669,7 @@ def _github_line_hash_cache(
     if source_root is None:
         return {}
     try:
-        source_root = source_root.resolve(strict=True)
+        source_root = filesystem_path(source_root).resolve(strict=True)
     except (OSError, RuntimeError):
         return {}
     requested_lines_by_path: dict[str, set[int]] = {}
@@ -1909,7 +1913,7 @@ def build_sarif_projection(
 ) -> dict[str, Any]:
     if source_root is not None:
         try:
-            source_root = source_root.resolve(strict=True)
+            source_root = filesystem_path(source_root).resolve(strict=True)
             source_root_is_directory = source_root.is_dir()
         except (OSError, RuntimeError):
             source_root_is_directory = False
@@ -2052,7 +2056,7 @@ def write_export_output(scan_dir: Path, output: Path, export_format: str, conten
     if export_format not in EXPORT_PATHS:
         raise ContractError(f"unsupported export format: {export_format}")
     scan_dir = _require_scan_directory(scan_dir)
-    output = Path(os.path.abspath(output))
+    output = filesystem_path(Path(os.path.abspath(output)))
     try:
         relative_output = output.relative_to(scan_dir).as_posix()
     except ValueError: