fix(platform): harden win32 atomic replaces and path segments

src/bmad_loop/data/plugins/unity/unity_setup.py

@@ -304,7 +304,7 @@ def _local_url(worktree: Path) -> str | None:
         return override.strip()
     cfg = worktree / ".mcp.json"
     try:
-        data = json.loads(cfg.read_text())
+        data = json.loads(cfg.read_text(encoding="utf-8-sig"))
         servers = data.get("mcpServers", {})
         entry = servers.get(_MCP_SERVER_NAME) or next(iter(servers.values()), {})
         url = entry.get("url")
@@ -425,10 +425,10 @@ def _verify_agent_isolation(agent: str, worktree: Path, url: str) -> bool:
     port = url.rsplit(":", 1)[-1].strip("/")
     cfg = worktree / rel
     try:
-        text = cfg.read_text()
-    except OSError:
+        text = cfg.read_text(encoding="utf-8-sig")
+    except (OSError, UnicodeDecodeError):
         print(
-            f"unity_setup: agent {agent!r} MCP config {rel} missing after setup; "
+            f"unity_setup: agent {agent!r} MCP config {rel} missing or unreadable after setup; "
             "cannot guarantee per-worktree isolation",
             file=sys.stderr,
         )

src/bmad_loop/data/plugins/unity/unity_teardown.py

@@ -176,11 +176,20 @@ def _force_kill_lingering(worktree: Path) -> int:
             break
         time.sleep(0.5)
     for pid in pids:
-        if host.is_alive(pid) and ident[pid] is not None and host.identity(pid) == ident[pid]:
-            try:
-                host.force_kill(pid)
-            except OSError:
-                pass
+        if not host.is_alive(pid):
+            continue
+        if ident[pid] is None or host.identity(pid) != ident[pid]:
+            print(
+                f"unity_teardown: pid {pid} survived terminate but its identity is "
+                "missing or changed since teardown began; skipping force-kill to "
+                "avoid hitting a reused pid",
+                file=sys.stderr,
+            )
+            continue
+        try:
+            host.force_kill(pid)
+        except OSError:
+            pass
     return len(pids)
 
 

src/bmad_loop/decisions.py

@@ -27,6 +27,7 @@
 from pathlib import Path
 
 from . import bmadconfig, deferredwork, runs, verify
+from .platform_util import atomic_replace
 from .sweep import Decision, DecisionOption, validate_triage
 
 STORE_REL = Path(".bmad-loop") / "decisions.json"
@@ -58,7 +59,7 @@ def _write_store(project: Path, data: dict) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     tmp = path.with_suffix(".tmp")
     tmp.write_text(json.dumps(data, indent=2, sort_keys=True), encoding="utf-8")
-    tmp.replace(path)
+    atomic_replace(tmp, path)
 
 
 def record_pre_answer(project: Path, dw_id: str, option: DecisionOption, *, date: str) -> None:

src/bmad_loop/engine.py

@@ -39,6 +39,7 @@
     SessionRecord,
     StoryTask,
 )
+from .platform_util import safe_segment
 from .plugins import HookBus, HookContext, PluginRegistry
 from .policy import Policy
 from .runs import kill_session
@@ -1175,7 +1176,7 @@ def _resumable_session(self, task: StoryTask) -> tuple[str, SessionResult] | Non
             role, seq = "review", task.review_cycle
         else:
             return None
-        task_id = f"{task.story_key}-{role}-{seq}"
+        task_id = f"{safe_segment(task.story_key)}-{role}-{seq}"
         for record in reversed(task.sessions):
             if record.task_id != task_id:
                 continue
@@ -1992,7 +1993,9 @@ def _run_session(
     ) -> SessionResult:
         # ``label`` names a non-standard session (a plugin-provided workflow) so
         # its task_id stays distinct from the role's own dev/review attempts.
-        task_id = f"{task.story_key}-{label or role}-{seq}"
+        # Sanitize the whole composition, not the parts: two individually capped
+        # parts can still compose past a Windows filename segment limit.
+        task_id = safe_segment(f"{task.story_key}-{label if label else role}-{seq}")
         adapter = self.adapters[role]
         cfg = self.policy.adapter.resolved(role)
         env = {
@@ -2167,7 +2170,7 @@ def _reset_spec_for_repair(self, task: StoryTask) -> None:
     def _write_feedback(self, task: StoryTask, reason: str) -> Path:
         """Persist a verification failure where the next session can read it —
         deterministic evidence must reach the LLM, not just the journal."""
-        path = self.run_dir / "feedback" / f"{task.story_key}-{len(task.sessions)}.md"
+        path = self.run_dir / "feedback" / f"{safe_segment(task.story_key)}-{len(task.sessions)}.md"
         path.parent.mkdir(parents=True, exist_ok=True)
         path.write_text(
             f"# Verification feedback: {task.story_key}\n\n"
@@ -2329,7 +2332,7 @@ def _stash_deferred_artifacts(self, task: StoryTask) -> None:
         spec_path = Path(task.spec_file)
         if not spec_path.is_file():
             return
-        dest = self.run_dir / "deferred" / task.story_key
+        dest = self.run_dir / "deferred" / safe_segment(task.story_key)
         dest.mkdir(parents=True, exist_ok=True)
         shutil.move(str(spec_path), str(dest / spec_path.name))
         self.journal.append(

src/bmad_loop/journal.py

@@ -3,12 +3,12 @@
 from __future__ import annotations
 
 import json
-import os
 import time
 from pathlib import Path
 from typing import Any
 
 from .model import RunState
+from .platform_util import atomic_replace
 
 STATE_FILE = "state.json"
 JOURNAL_FILE = "journal.jsonl"
@@ -63,7 +63,7 @@ def save_state(run_dir: Path, state: RunState) -> None:
     target = run_dir / STATE_FILE
     tmp = target.with_suffix(".json.tmp")
     tmp.write_text(json.dumps(state.to_dict(), indent=2), encoding="utf-8")
-    os.replace(tmp, target)
+    atomic_replace(tmp, target)
 
 
 def load_state(run_dir: Path) -> RunState:

src/bmad_loop/platform_util.py

@@ -5,17 +5,47 @@
 delegate to it. ``detach_kwargs`` stays a real implementation — it is spawn
 configuration, not a process-lifecycle primitive, so it does not belong on the
 host. On Linux/macOS — and WSL, which *is* Linux — these preserve today's exact
-behavior; the Windows branch degrades gracefully and is not yet exercised.
+behavior. The win32 file-replace and path-segment helpers below (``atomic_replace``,
+``safe_segment``) are exercised by the platform tests; the pid kill/liveness
+Windows branch degrades gracefully and is not yet exercised.
 """
 
 from __future__ import annotations
 
+import hashlib
+import os
+import random
+import re
 import subprocess
 import sys
+import time
 from pathlib import Path, PurePosixPath, PureWindowsPath
 
 from .process_host import get_process_host
 
+# Windows-only: os.replace (MoveFileExW) fails with ERROR_ACCESS_DENIED (5) or
+# ERROR_SHARING_VIOLATION (32) when a concurrent reader holds a handle on the
+# target — Python's open() grants no FILE_SHARE_DELETE, so renaming over the open
+# file is denied. Readers hold their handle briefly, so a jittered backoff clears
+# it; an anti-virus / indexer touch can hold longer, hence the ~5 s worst case.
+# POSIX rename-over-open never raises this, so the retry stays win32-gated.
+_REPLACE_ATTEMPTS = 12
+_REPLACE_BASE_S = 0.02
+_REPLACE_CAP_S = 0.7
+
+# Reserved on Windows regardless of extension: CON.txt is as illegal as CON. The
+# COM0/LPT0 and superscript (COM¹/COM²/COM³) forms are reserved by the same rule,
+# as are the console device names CONIN$/CONOUT$.
+_RESERVED_BASENAMES = frozenset(
+    {"CON", "PRN", "AUX", "NUL", "CONIN$", "CONOUT$"}
+    | {f"COM{i}" for i in range(10)}
+    | {f"LPT{i}" for i in range(10)}
+    | {f"COM{s}" for s in "¹²³"}
+    | {f"LPT{s}" for s in "¹²³"}
+)
+_ILLEGAL_SEGMENT_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
+_MAX_SEGMENT = 120  # keep segment (incl. any collision suffix) well under the 255 limit
+
 
 def terminate_pid(pid: int) -> None:
     """Politely terminate ``pid``. Back-compat shim over
@@ -68,3 +98,63 @@ def has_parent_ref(value: str | Path) -> bool:
     the two for a complete "must stay inside the project" guard."""
     text = str(value)
     return ".." in PurePosixPath(text).parts or ".." in PureWindowsPath(text).parts
+
+
+def atomic_replace(tmp: Path, target: Path) -> None:
+    """``os.replace(tmp, target)``, retried on the transient Windows sharing
+    violation a concurrent reader of ``target`` triggers (WinError 5/32). Gated to
+    win32 so a real POSIX EACCES/EPERM surfaces immediately instead of after a
+    pointless backoff. Worst-case total wait is ~5 s of jittered exponential
+    backoff before the final failure propagates."""
+    for attempt in range(_REPLACE_ATTEMPTS):
+        try:
+            os.replace(tmp, target)
+            return
+        except OSError as exc:
+            last = attempt == _REPLACE_ATTEMPTS - 1
+            # a retryable rename-over-open denial, not a genuine permission fault
+            winerror = getattr(exc, "winerror", None)
+            retryable = isinstance(exc, PermissionError) or winerror in (5, 32)
+            # portability: only Windows raises this on rename-over-open; elsewhere a
+            # permission error is real and must surface at once.
+            if sys.platform != "win32" or last or not retryable:
+                raise
+            delay = min(_REPLACE_CAP_S, _REPLACE_BASE_S * 2**attempt)
+            time.sleep(delay + random.uniform(0, _REPLACE_BASE_S))  # nosec B311 - retry jitter
+
+
+def _is_reserved_basename(seg: str) -> bool:
+    """True if ``seg``'s basename (before the first dot, trailing spaces trimmed —
+    ``CON .txt`` counts) is a Windows reserved device name."""
+    stem = seg.split(".", 1)[0].rstrip(" ")
+    return stem.upper() in _RESERVED_BASENAMES
+
+
+def safe_segment(name: str) -> str:
+    """Coerce ``name`` into a single Windows-legal path segment, returning legal
+    input unchanged (identity for clean keys — the common case, e.g. a story key
+    like ``3-2-digest-delivery``).
+
+    Replaces the reserved characters ``<>:"/\\|?*`` and control chars with ``_``,
+    strips trailing dots and spaces (Windows silently drops them), caps the length,
+    and defuses the reserved device basenames (CON, PRN, AUX, NUL, COM0-9, LPT0-9
+    and their superscript ¹²³ forms — case-insensitive, with or without an
+    extension). Whenever anything is changed a short digest of the raw input is
+    appended, giving practical (probabilistic, not absolute) collision resistance
+    between distinct raw names: clean-key identity is the stronger contract, so a
+    clean name that happens to look like a sanitized-plus-digest name passes
+    through verbatim, and case-insensitive NTFS collisions between clean names
+    remain the caller's concern. Never raises."""
+    cleaned = _ILLEGAL_SEGMENT_CHARS.sub("_", name).rstrip(". ")[:_MAX_SEGMENT]
+    if _is_reserved_basename(cleaned):
+        cleaned = "_" + cleaned
+    if not cleaned:
+        cleaned = "_"
+    if cleaned == name:
+        return name  # already a legal segment — keep it byte-identical
+    digest = hashlib.sha1(
+        name.encode("utf-8", "surrogatepass"),
+        usedforsecurity=False,  # collision-resistance suffix, not a credential
+    ).hexdigest()
+    suffix = "-" + digest[:8]
+    return cleaned[: _MAX_SEGMENT - len(suffix)] + suffix

src/bmad_loop/resolve.py

@@ -22,12 +22,13 @@
 
 from .adapters.base import SessionSpec
 from .model import RunState
+from .platform_util import safe_segment
 
 RESOLVE_DIR = "resolve"
 
 
 def _story_dir(run_dir: Path, story_key: str) -> Path:
-    return run_dir / RESOLVE_DIR / story_key
+    return run_dir / RESOLVE_DIR / safe_segment(story_key)
 
 
 def context_path(run_dir: Path, story_key: str) -> Path:
@@ -173,7 +174,7 @@ def run_session(adapter, project: Path, run_dir: Path, story_key: str, *, model:
     resolution marker. The context file must already be written (build_context).
     """
     spec = SessionSpec(
-        task_id=f"{story_key}-resolve-1",
+        task_id=f"{safe_segment(story_key)}-resolve-1",
         role="dev",
         prompt=f"/bmad-loop-resolve {story_key}",
         cwd=project,

src/bmad_loop/runs.py

@@ -14,6 +14,7 @@
 from .adapters.multiplexer import get_multiplexer
 from .journal import STATE_FILE, Journal, load_state, save_state
 from .model import PAUSE_ESCALATION, Phase
+from .platform_util import atomic_replace
 from .process_host import get_process_host
 
 RUNS_DIR = Path(".bmad-loop") / "runs"
@@ -357,15 +358,15 @@ def delete_run(run_dir: Path) -> None:
 
 def archive_run(project: Path, run_dir: Path) -> Path:
     """Compress a run dir into .bmad-loop/archive/<id>.tar.gz and remove the
-    original. The tarball is written to a temp path then os.replace'd into place
-    so a partial archive never appears. Callers enforce the live guard."""
+    original. The tarball is written to a temp path then atomically replaced into
+    place so a partial archive never appears. Callers enforce the live guard."""
     archive_dir = project / ARCHIVE_DIR
     archive_dir.mkdir(parents=True, exist_ok=True)
     dest = archive_dir / f"{run_dir.name}.tar.gz"
     tmp = dest.with_suffix(".tar.gz.tmp")
     with tarfile.open(tmp, "w:gz") as tar:
         tar.add(run_dir, arcname=run_dir.name)
-    os.replace(tmp, dest)
+    atomic_replace(tmp, dest)
     shutil.rmtree(run_dir)
     return dest
 

src/bmad_loop/sweep.py

@@ -22,6 +22,7 @@
 from .engine import Engine
 from .escalation import critical_escalations
 from .model import Phase, StoryTask
+from .platform_util import atomic_replace
 from .statemachine import advance
 from .workspace import discard_worktree
 
@@ -793,7 +794,7 @@ def _decisions_phase(self, plan: TriagePlan) -> tuple[dict[str, dict[str, str]],
         if seeded:
             tmp = decisions_path.with_suffix(".tmp")
             tmp.write_text(json.dumps(answers, indent=2), encoding="utf-8")
-            tmp.replace(decisions_path)
+            atomic_replace(tmp, decisions_path)
         pending = [d for d in plan.decisions if d.id not in answers]
         answered_interactively = False
         if not self.prompting:
@@ -831,7 +832,7 @@ def _decisions_phase(self, plan: TriagePlan) -> tuple[dict[str, dict[str, str]],
                 }
                 tmp = decisions_path.with_suffix(".tmp")
                 tmp.write_text(json.dumps(answers, indent=2), encoding="utf-8")
-                tmp.replace(decisions_path)
+                atomic_replace(tmp, decisions_path)
                 self.journal.append(
                     "decision-answered",
                     dw_id=decision.id,

src/bmad_loop/tui/settings.py

@@ -9,13 +9,13 @@
 
 from __future__ import annotations
 
-import os
 from pathlib import Path
 from typing import Any
 
 import tomlkit
 
 from .. import policy as policy_mod
+from ..platform_util import atomic_replace
 
 STAGES = ("dev", "review", "triage")
 
@@ -98,4 +98,4 @@ def save(self, path: Path) -> None:
         path.parent.mkdir(parents=True, exist_ok=True)
         tmp = path.with_suffix(".toml.tmp")
         tmp.write_text(self.dumps(), encoding="utf-8")
-        os.replace(tmp, path)
+        atomic_replace(tmp, path)

src/bmad_loop/workspace.py

@@ -21,6 +21,7 @@
 
 from . import verify
 from .bmadconfig import ProjectPaths
+from .platform_util import safe_segment
 
 # Per-unit worktrees live under the run dir (.bmad-loop/runs/<run_id>/worktrees/),
 # which `bmad-loop init` already gitignores — so unit checkouts never show up as
@@ -85,7 +86,7 @@ def open_unit_workspace(
     keeps the commits earlier units already landed on it.
     """
     branch = unit_branch_name(run_id, unit_key, branch_per)
-    wt = (unit_worktrees_dir(run_dir) / unit_key).resolve()
+    wt = (unit_worktrees_dir(run_dir) / safe_segment(unit_key)).resolve()
     wt.parent.mkdir(parents=True, exist_ok=True)
     if verify.branch_exists(repo_root, branch):
         verify.worktree_add(repo_root, wt, branch, create=False)
@@ -134,7 +135,7 @@ def close_unit_workspace(
         except verify.GitError:
             diff = ""
         if diff:
-            patch = run_dir / "failed" / unit_key / "changes.patch"
+            patch = run_dir / "failed" / safe_segment(unit_key) / "changes.patch"
             patch.parent.mkdir(parents=True, exist_ok=True)
             patch.write_text(diff, encoding="utf-8")
         if keep_failed: