"""Per-task git worktree provisioning.

Every okstra task — regardless of task-type/phase — runs inside an
isolated git worktree rooted at
`~/.okstra/worktrees/<project_id>/<task_group>/<task_id>/`. The same
worktree is reused across all phases of one task-key (requirements-
discovery → error-analysis → implementation-option-selection →
implementation-planning → implementation), so phase N picks up exactly
the working-tree state phase N-1 left behind.

A global registry (`worktree_registry.py`) maps task-keys to the
on-disk path + branch and serialises reservations, so two concurrent
okstra runs cannot collide on the same path or branch name.

Pre-conditions handled here:
  - Skip when `project_root` is not a git repo (degrade gracefully).
  - Skip when `project_root` itself is already a non-main worktree
    (caller's tree is already an isolated workspace; reuse it).
  - Re-entry of the same task-key returns the existing worktree.
  - Branch / path collisions across task-keys raise PrepareError-like
    RuntimeError.

Side effects:
  - `git worktree add -b <branch> <path> <base_ref>` invoked in the
    main worktree of `project_root` (NOT `project_root` itself when it
    is also a worktree — base must be the main checkout).
  - Per-task sync dirs (.project-docs, .scratch, graphify-out by
    default) symlinked from the **main worktree** into the new
    worktree so every task sees the same shared state, irrespective of
    which worktree the caller invoked okstra from.
  - The function does NOT chdir.
"""
from __future__ import annotations

import json
import os
import shutil
import stat
import subprocess
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

from okstra_project.dirs import okstra_home, project_json_path

from .ids import _safe_fs_segment
from .json_boundary import JsonBoundaryError, load_owned_object
from . import worktree_registry
from .seeding import (
    SettingsLinkError,
    ensure_project_settings_symlink,
)


# Project-root directories that hold okstra task state, ignored by git, or
# otherwise required for the executor to operate but NOT carried across by
# `git worktree add`. Each is symlinked from the MAIN worktree into the new
# worktree at provision time. Symlinks (not copies) so every task sees the
# live shared state and disk/CPU cost stays near zero; the trade-off is
# that any write through the link reaches the main worktree, which is
# acceptable because okstra writes only to paths it owns: its task-scoped
# subdirectory (e.g. `.okstra/tasks/<task-id>/runs/...`) and, for a
# branch-mode code review, `.project-docs/code-reviews/<branch>/`.
#
# Override precedence (most-specific first):
#   1. `OKSTRA_WORKTREE_SYNC_DIRS` env var — colon-separated list, REPLACES
#      defaults. Empty string disables the feature entirely. One-off
#      operator override.
#   2. `worktreeSyncDirs` array in `.okstra/project.json` —
#      project-level config, persists across runs. Same semantics: array
#      REPLACES defaults, empty array disables.
#   3. The built-in `DEFAULT_WORKTREE_SYNC_DIRS` below.
DEFAULT_WORKTREE_SYNC_DIRS: tuple[str, ...] = (
    ".project-docs",
    ".scratch",
    "graphify-out",
    ".claude",
)


# Sync dirs materialised as a REAL directory whose children are symlinked one
# by one, instead of a single symlink standing in for the whole directory.
#
# Why the split exists: git does not follow a symlink, so a symlinked directory
# is one *file* to it. A project that ignores host config by its contents
# (`.claude/*`) matches every child but never the bare `.claude` path, so the
# symlink lands in `git status` as
# `?? .claude` while the same directory is invisible in the main checkout. Any
# plan step asserting a clean worktree then fails on okstra's own provisioning.
# Linking the children instead reproduces the main checkout's shape, so
# whatever the project's ignore rules do there, they do here too.
#
# Only `.claude` qualifies. The other sync dirs are shared okstra state that
# okstra WRITES into, and a directory symlink is what makes a newly created
# top-level entry land in the main checkout rather than diverging inside the
# worktree. `.claude` is host configuration okstra reads; its one okstra write
# is the fixed `settings.local.json` child seeded by
# `_seed_worktree_settings_symlink`.
CHILD_LINKED_SYNC_DIRS: tuple[str, ...] = (".claude",)


# Project-root-relative FILES (not dirs) symlinked from MAIN → task worktree
# at provision time. Same symlink semantics as `DEFAULT_WORKTREE_SYNC_DIRS`:
# every task sees the live shared file. The split exists because the original
# `_link_sync_dirs` helper only walked directories — a `.env` (or any
# top-level file outside `.git`'s tracking) would otherwise be lost on every
# new worktree, blocking verifier dispatches that depend on environment-
# resolved secrets.
#
# Override precedence mirrors `DEFAULT_WORKTREE_SYNC_DIRS`:
#   1. `OKSTRA_WORKTREE_SYNC_FILES` env var (colon-separated, REPLACES).
#   2. `worktreeSyncFiles` array in `.okstra/project.json`.
#   3. The built-in `DEFAULT_WORKTREE_SYNC_FILES` below.
DEFAULT_WORKTREE_SYNC_FILES: tuple[str, ...] = (
    ".env",
)


# Project-root-relative files COPIED (not symlinked) from MAIN → task worktree
# at provision time, then `chmod 0o444` so the task cannot mutate the
# snapshot. Used for live-mutating fixtures (e.g. `classifications.db`)
# where the verifier needs to reproduce accuracy SQL against the same rows
# the executor saw, without sharing the writable handle that would corrupt
# the main worktree's copy. Default is empty — okstra has no opinion about
# which fixtures any given project relies on; opt in per-project via
# `worktreeSnapshotFiles` in `project.json` (or `OKSTRA_WORKTREE_SNAPSHOT_FILES`
# for one-off operator override).
DEFAULT_WORKTREE_SNAPSHOT_FILES: tuple[str, ...] = ()


# Work-category → branch namespace (slash-prefixed). Mirrors the values
# accepted by `--work-category` (bugfix / feature / refactor / ops /
# improvement); `feature` and `improvement` share the `feature/` namespace,
# and any unset/unrecognised category falls back to `task/`.
_WORK_CATEGORY_NAMESPACE = {
    "feature": "feature",
    "improvement": "feature",
    "bugfix": "fix",
    "refactor": "refactor",
    "ops": "ops",
}


@dataclass
class WorktreeProvision:
    """Result of `provision_task_worktree`.

    status:
      - "created": fresh worktree at `path` on `branch`
      - "reused": registry already had this task-key; same path/branch
        returned and no new `git worktree add` was executed
      - "skipped-in-worktree": project_root is itself a non-main
        worktree; the run reuses `project_root` and no new worktree is
        materialised (registry NOT updated — that caller is already
        isolated by virtue of its own worktree)
      - "skipped-not-git": project_root has no `.git` (worktree path
        cannot be provisioned; degrade gracefully)
    """
    status: str
    path: str = ""          # absolute path of the task worktree (or project_root when reused)
    branch: str = ""        # branch checked out in the worktree (empty when reused / not-git)
    base_ref: str = ""      # commit SHA the worktree was branched from (empty when not created)
    note: str = ""          # human-readable explanation, surfaced in team-state / manifests


@dataclass
class WorktreeDecision:
    """Side-effect-free preview of what `provision_task_worktree` would do.

    status:
      - "new": no active registry entry; a fresh worktree would be created
      - "reused": registry already has this task-key; existing path/branch returned
      - "skipped-in-worktree": project_root is itself a non-main worktree
      - "skipped-not-git": project_root has no .git
    """
    status: str
    path: str              # worktree path (new: prospective; reuse: existing; skip: project_root)
    branch: str = ""       # new: prospective branch; reused: existing branch
    base_ref: str = ""     # new: requested base_ref; reused: existing base


def preview_worktree_decision(
    *,
    project_root,
    project_id: str,
    task_group_segment: str,
    task_id_segment: str,
    work_category: str,
    base_ref: str = "",
) -> "WorktreeDecision":
    """Side-effect-free: what provision_task_worktree WOULD do, without touching disk.

    Mirrors provision's decision branches exactly; reuses the same read-only
    helpers so preview never diverges from the actual provisioning result.
    """
    project_root = Path(project_root)
    if not is_git_work_tree(project_root):
        return WorktreeDecision(status="skipped-not-git", path=str(project_root))
    if _is_inside_non_main_worktree(project_root):
        return WorktreeDecision(status="skipped-in-worktree", path=str(project_root))
    safe_project = _safe_segment(project_id)
    safe_group = _safe_segment(task_group_segment)
    safe_task = _safe_segment(task_id_segment)
    existing = worktree_registry.lookup(safe_project, safe_group, safe_task)
    if existing is not None and existing.status == "active":
        return WorktreeDecision(
            status="reused", path=existing.worktree_path,
            branch=existing.branch, base_ref=existing.base_ref,
        )
    return WorktreeDecision(
        status="new",
        path=str(compute_worktree_path(
            project_id=safe_project, task_group_segment=safe_group,
            task_id_segment=safe_task)),
        branch=compute_branch_name(work_category=work_category, task_id_segment=safe_task),
        base_ref=base_ref,
    )


@dataclass
class StageWorktreeDecision:
    """Side-effect-free decision for one concrete implementation stage."""

    status: str
    path: str
    branch: str = ""
    base_ref: str = ""


def resolve_stage_worktree_decision(
    *,
    project_id: str,
    task_group_segment: str,
    task_id_segment: str,
    work_category: str,
    stage_number: int,
) -> StageWorktreeDecision:
    """Resolve whether one concrete stage worktree is new or reusable."""
    safe_project = _safe_segment(project_id)
    safe_group = _safe_segment(task_group_segment)
    safe_task = _safe_segment(task_id_segment)
    existing = worktree_registry.lookup(
        safe_project, safe_group, safe_task, stage_number=stage_number)
    if existing is not None and _stage_entry_is_reusable(existing):
        return StageWorktreeDecision(
            status="reused",
            path=existing.worktree_path,
            branch=existing.branch,
            base_ref=existing.base_ref,
        )
    return StageWorktreeDecision(
        status="new",
        path=str(compute_worktree_path(
            project_id=safe_project, task_group_segment=safe_group,
            task_id_segment=safe_task, stage_number=stage_number)),
        branch=compute_branch_name(
            work_category=work_category, task_id_segment=safe_task,
            stage_number=stage_number),
    )


def _stage_entry_is_reusable(entry: worktree_registry.WorktreeEntry) -> bool:
    """Whether a registered stage worktree can be entered by this run.

    `active` is the live-run case. `released` with the directory still on disk is
    the fix-run case: a stage whose verifier returned FAIL records a `failed`
    consumers row, which frees the occupancy but deliberately keeps the worktree
    and its branch as the reviewable stack. Re-entry MUST reuse that tree —
    provisioning anew refuses on the existing path and branch. After whole-task
    final-verification removes the directory the entry stops being reusable, so
    the stage provisions from scratch.
    """
    if entry.status == "active":
        return True
    return entry.status == "released" and Path(entry.worktree_path).is_dir()


def _safe_segment(value: str) -> str:
    """Sanitise a single path/branch segment.

    Forbidden chars (`/`, `:`, spaces, anything outside `[a-z0-9-]`)
    are collapsed to `-`. Empty result becomes `_` so we never create
    an empty path component. Delegates to the canonical slugifier in
    `ids.py` to stay in lock-step with run-id / manifest segmentation.
    """
    return _safe_fs_segment(value)


def _work_category_namespace(work_category: str) -> str:
    key = (work_category or "").strip().lower()
    return _WORK_CATEGORY_NAMESPACE.get(key, "task")


def _git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["git", "-C", str(cwd), *args],
        capture_output=True, text=True, check=False,
    )


def _is_inside_non_main_worktree(project_root: Path) -> bool:
    """True iff project_root is inside a git worktree that is NOT the
    repository's main checkout. Detection rule: `--git-dir` (per-worktree
    .git pointer) differs from `--git-common-dir` (shared object store).
    """
    common = _git(project_root, "rev-parse", "--git-common-dir")
    per_tree = _git(project_root, "rev-parse", "--git-dir")
    if common.returncode != 0 or per_tree.returncode != 0:
        return False
    common_abs = (project_root / common.stdout.strip()).resolve()
    per_tree_abs = (project_root / per_tree.stdout.strip()).resolve()
    return common_abs != per_tree_abs


def is_git_work_tree(project_root: Path) -> bool:
    """project_root 가 git work tree 내부인지 판정하는 public git-introspection seam.

    git 미설치(FileNotFoundError) 를 포함한 모든 실패는 False — 호출자(migrate
    등)가 non-git 레이아웃으로 안전하게 폴백할 수 있도록 한다. 과거 migrate.py
    가 같은 판정을 자체 구현(`--show-toplevel`)했는데 이 seam 으로 통합한다."""
    try:
        res = _git(project_root, "rev-parse", "--is-inside-work-tree")
    except (OSError, FileNotFoundError):
        return False
    return res.returncode == 0 and res.stdout.strip() == "true"


def _branch_exists(project_root: Path, branch: str) -> bool:
    res = _git(project_root, "rev-parse", "--verify", "--quiet", f"refs/heads/{branch}")
    return res.returncode == 0


def _branch_checkout_path(project_root: Path, branch: str) -> str:
    """The worktree that currently has *branch* checked out, or "" when none does.

    `git branch -D` refuses to delete a checked-out branch, so an existing-branch
    error that only recommends `-D` is unfollowable whenever the branch is the
    main checkout's HEAD. Callers use this to name the extra step instead.
    """
    res = _git(project_root, "worktree", "list", "--porcelain")
    if res.returncode != 0:
        return ""
    current = ""
    for line in res.stdout.splitlines():
        if line.startswith("worktree "):
            current = line[len("worktree "):].strip()
        elif line.strip() == f"branch refs/heads/{branch}":
            return current
    return ""


def _branch_exists_message(project_root: Path, branch: str, kind: str) -> str:
    """Existing-branch error text whose remedy is actually runnable."""
    checkout = _branch_checkout_path(project_root, branch)
    if checkout:
        return (
            f"{kind} worktree branch already exists: {branch}, and it is checked "
            f"out at {checkout}. `git branch -D {branch}` will refuse while it is "
            f"checked out — switch that checkout to another branch first "
            f"(`git -C {checkout} switch <other-branch>`), then delete it, or "
            "choose a different work-category before retrying."
        )
    return (
        f"{kind} worktree branch already exists: {branch}. "
        f"Delete it (`git branch -D {branch}`) or choose a different "
        "work-category before retrying."
    )


def _head_sha(cwd: Path) -> str:
    res = _git(cwd, "rev-parse", "HEAD")
    if res.returncode != 0:
        return ""
    return res.stdout.strip()


def _resolve_commit_sha(cwd: Path, ref: str) -> str:
    """Resolve a user-supplied ref (branch, tag, short/long SHA) to a full
    commit SHA in `cwd`. Returns empty string when the ref is not resolvable
    so the caller can raise a contextual error.
    """
    res = _git(cwd, "rev-parse", "--verify", "--quiet", f"{ref}^{{commit}}")
    if res.returncode != 0:
        return ""
    return res.stdout.strip()


def main_worktree_path(project_root: Path) -> Path:
    """Locate the repository's MAIN worktree (the original checkout).

    `git worktree list --porcelain` lists worktrees in a stable order
    where the first `worktree <path>` block is the main checkout.
    Falls back to `project_root` if parsing fails — caller still gets
    a working path, sync-dir links just point at the caller's tree
    (the prior behaviour).
    """
    res = _git(project_root, "worktree", "list", "--porcelain")
    if res.returncode != 0:
        return project_root
    for line in res.stdout.splitlines():
        if line.startswith("worktree "):
            return Path(line[len("worktree "):].strip())
    return project_root


def _read_project_json_field(project_root: Path, field: str) -> Optional[tuple[str, ...]]:
    """Read a string-array field from the project's okstra project.json.

    Returns None if the field is absent or the file cannot be parsed (so
    the caller falls back to defaults). Returns an empty tuple if the
    field is explicitly an empty array (caller treats this as "disable").
    A non-list value is treated as missing — we do not raise here because
    field resolution must never block worktree provisioning.
    """
    target = project_json_path(project_root)
    if not target.is_file():
        return None
    try:
        data = load_owned_object(target, artifact="project config")
    except (OSError, JsonBoundaryError):
        return None
    if not isinstance(data, dict):
        return None
    value = data.get(field)
    if not isinstance(value, list):
        return None
    cleaned = tuple(
        item.strip() for item in value
        if isinstance(item, str) and item.strip()
    )
    return cleaned


def _resolve_entries(
    *,
    env_var: str,
    project_field: str,
    default: tuple[str, ...],
    project_root: Optional[Path],
) -> tuple[str, ...]:
    """Generic resolver shared by sync-dirs / sync-files / snapshot-files.

    Precedence: env var (colon-separated, REPLACES) → project.json field →
    built-in default. An empty env value or empty JSON array disables the
    feature (returns `()`).
    """
    raw = os.environ.get(env_var)
    if raw is not None:
        raw = raw.strip()
        if not raw:
            return ()
        return tuple(part for part in (p.strip() for p in raw.split(":")) if part)
    if project_root is not None:
        from_project = _read_project_json_field(project_root, project_field)
        if from_project is not None:
            return from_project
    return default


def _resolve_sync_dirs(project_root: Optional[Path] = None) -> tuple[str, ...]:
    """Return the list of project-root-relative dirs to symlink into the
    new worktree. Precedence: env var → project.json → built-in default.
    See the comment above `DEFAULT_WORKTREE_SYNC_DIRS` for full semantics.
    """
    return _resolve_entries(
        env_var="OKSTRA_WORKTREE_SYNC_DIRS",
        project_field="worktreeSyncDirs",
        default=DEFAULT_WORKTREE_SYNC_DIRS,
        project_root=project_root,
    )


def _resolve_sync_files(project_root: Optional[Path] = None) -> tuple[str, ...]:
    """File-level counterpart to `_resolve_sync_dirs` (FU-V2)."""
    return _resolve_entries(
        env_var="OKSTRA_WORKTREE_SYNC_FILES",
        project_field="worktreeSyncFiles",
        default=DEFAULT_WORKTREE_SYNC_FILES,
        project_root=project_root,
    )


def _resolve_snapshot_files(project_root: Optional[Path] = None) -> tuple[str, ...]:
    """Read-only snapshot file list (FU-V3)."""
    return _resolve_entries(
        env_var="OKSTRA_WORKTREE_SNAPSHOT_FILES",
        project_field="worktreeSnapshotFiles",
        default=DEFAULT_WORKTREE_SNAPSHOT_FILES,
        project_root=project_root,
    )


def okstra_clean_gate_excludes(project_root: Optional[Path] = None) -> tuple[str, ...]:
    """Project-relative paths okstra owns and source clean gates should ignore."""
    out: list[str] = []
    seen: set[str] = set()
    for rel in (
        ".okstra",
        *_resolve_sync_dirs(project_root),
        *_resolve_sync_files(project_root),
        *_resolve_snapshot_files(project_root),
    ):
        cleaned = rel.strip().removeprefix("./").rstrip("/")
        if not cleaned or cleaned == "." or cleaned in seen:
            continue
        seen.add(cleaned)
        out.append(cleaned)
    return tuple(out)


def nested_worktree_excludes(worktree_path) -> tuple[str, ...]:
    """Relative paths of git worktrees nested under `worktree_path`.

    implementation stage worktrees live inside the task worktree
    (`<task-id>/stage-<N>/`), so the parent's `git status --short` reports
    each as an untracked `?? stage-N/`. They are git-registered worktrees,
    not user source changes, so clean gates must ignore them. Derived from
    `git worktree list` rather than a hardcoded pattern so it tracks exactly
    what git registered.
    """
    root = Path(worktree_path).resolve()
    r = _git(root, "worktree", "list", "--porcelain")
    if r.returncode != 0:
        return ()
    out: list[str] = []
    for line in r.stdout.splitlines():
        if not line.startswith("worktree "):
            continue
        wt = Path(line[len("worktree "):].strip()).resolve()
        if wt == root:
            continue
        try:
            rel = wt.relative_to(root)
        except ValueError:
            continue
        out.append(str(rel))
    return tuple(out)


def is_ancestor(cwd, commit: str, head: str) -> bool:
    """True iff `commit` is an ancestor of `head` (both non-empty)."""
    if not commit or not head:
        return False
    return _git(Path(cwd), "merge-base", "--is-ancestor", commit, head).returncode == 0


def dirty_entries_excluding_okstra(cwd) -> list[str]:
    """`git status --short` rows for changes outside okstra-owned paths.

    okstra-owned = `okstra_clean_gate_excludes` (e.g. `.okstra`, synced dirs)
    plus `nested_worktree_excludes` (stage worktrees nested under `cwd`).

    This is the clean-worktree question every okstra gate asks, and the one a
    plan step must ask through `okstra worktree-status` rather than through a
    bare `git status --porcelain`: okstra provisions `.okstra` plus the synced
    entries into every task worktree, so a bare status is never empty there and
    an assertion built on it fails on okstra's own scaffolding.
    """
    owned = (*okstra_clean_gate_excludes(Path(cwd)), *nested_worktree_excludes(cwd))
    excludes = [f":(exclude){p}" for p in owned]
    out = _git(Path(cwd), "status", "--short", "--", ".", *excludes).stdout
    return [line for line in out.splitlines() if line.strip()]


def is_dirty_excluding_okstra(cwd) -> bool:
    """True iff the worktree has changes outside okstra-owned paths."""
    return bool(dirty_entries_excluding_okstra(cwd))


class MergeError(RuntimeError):
    """`git merge` 가 내용 충돌이 아닌 사유로 실패했을 때(추적되지 않은 파일
    덮어쓰기, 진행 중인 머지 잔존, 없는 브랜치 등) 던진다. 충돌 파일이 0개인
    실패를 '충돌'로 둔갑시키지 않고 git stderr 를 그대로 노출해 호출자가 올바른
    복구 경로로 안내하도록 한다."""


def merge_branch(cwd, branch: str, *, no_ff: bool) -> list[str] | None:
    """`branch` 를 cwd 의 현재 HEAD 에 머지한다. 성공이면 None, 내용 충돌이면
    abort 후 충돌 파일 목록을 반환한다(최소 1개). 충돌이 아닌 머지 실패는
    MergeError 로 던진다 — 충돌 0개를 충돌로 보고하지 않기 위함. 호출자가 충돌
    목록을 HandoffConflict/IntegrateError 등 적절한 예외로 변환한다 — 머지·
    충돌감지·abort 시퀀스의 단일 지점."""
    args = ["merge", "--no-ff", "--no-edit", branch] if no_ff \
        else ["merge", "--no-edit", branch]
    res = _git(Path(cwd), *args)
    if res.returncode == 0:
        return None
    conflicts = _git(Path(cwd), "diff", "--name-only",
                     "--diff-filter=U").stdout.split()
    _git(Path(cwd), "merge", "--abort")
    if not conflicts:
        raise MergeError((res.stderr or res.stdout).strip()
                         or f"git merge {branch} failed (exit={res.returncode})")
    return conflicts


def remove_worktree_force(cwd, worktree_path) -> subprocess.CompletedProcess:
    """워크트리를 강제 제거한다. provision_*_worktree 가 워크트리에 .okstra·
    synced 디렉터리/파일 symlink(untracked) 를 깔기 때문에 plain remove 는 그
    잔여물 때문에 git 이 거부한다 — dirty 게이트로 실사용자 변경을 막은 뒤
    --force 로 okstra 잔여물만 정리하는 단일 지점."""
    return _git(Path(cwd), "worktree", "remove", "--force", str(worktree_path))


def _link_sync_dirs(source_root: Path, worktree_path: Path) -> list[str]:
    """Symlink each configured dir from `source_root` (the MAIN
    worktree) into the new worktree.

    Skip rules:
      - Source missing in `source_root` → silently skipped.
      - Target path already exists in worktree (e.g. tracked content
        checked out by `git worktree add`) → skipped to avoid clobbering
        version-controlled files.
      - Parent directories are created as needed for nested entries.

    Entries listed in `CHILD_LINKED_SYNC_DIRS` are materialised as a real
    directory of per-child symlinks instead (see that constant).

    Returns a list of human-readable notes (one per linked entry) so the
    caller can include them in the provisioning note.
    """
    notes: list[str] = []
    for rel in _resolve_sync_dirs(source_root):
        src = (source_root / rel).resolve()
        if not src.exists():
            continue
        dst = worktree_path / rel
        if dst.exists() or dst.is_symlink():
            continue
        dst.parent.mkdir(parents=True, exist_ok=True)
        if rel in CHILD_LINKED_SYNC_DIRS and src.is_dir():
            _link_dir_children(src, dst)
            notes.append(rel)
            continue
        try:
            os.symlink(src, dst)
        except FileExistsError:
            continue
        notes.append(rel)
    return notes


def _link_dir_children(src: Path, dst: Path) -> None:
    """Create `dst` as a real directory whose entries symlink to `src`'s
    children, so git sees the same directory shape it sees in the main
    checkout (`CHILD_LINKED_SYNC_DIRS`).

    Each link points at the MAIN checkout's child rather than its resolved
    target, so a child that is itself a symlink (e.g. `.claude/settings.local.json`
    → `~/.okstra/templates/settings.local.json`) keeps following whatever the
    main checkout currently points at. An unreadable source or a child that
    cannot be linked degrades to "that child is absent in this worktree" —
    provisioning is not worth failing over host config.
    """
    dst.mkdir(parents=True, exist_ok=True)
    try:
        children = sorted(src.iterdir())
    except OSError:
        return
    for child in children:
        link = dst / child.name
        if link.exists() or link.is_symlink():
            continue
        try:
            os.symlink(child, link)
        except OSError:
            continue


def _link_sync_files(source_root: Path, worktree_path: Path) -> list[str]:
    """File-level counterpart to `_link_sync_dirs` (FU-V2).

    Same skip rules: missing source → skipped silently; pre-existing dst
    (tracked content checked out by `git worktree add`) → skipped to avoid
    clobbering. Symlinks (not copies) keep secrets out of duplicate files —
    the link points back to the MAIN worktree's `.env` so rotating the file
    there is reflected in every active task without re-provisioning.
    """
    notes: list[str] = []
    for rel in _resolve_sync_files(source_root):
        src = (source_root / rel).resolve()
        if not src.exists() or not src.is_file():
            continue
        dst = worktree_path / rel
        if dst.exists() or dst.is_symlink():
            continue
        dst.parent.mkdir(parents=True, exist_ok=True)
        try:
            os.symlink(src, dst)
        except FileExistsError:
            continue
        notes.append(rel)
    return notes


def _seed_worktree_settings_symlink(worktree_path: Path) -> None:
    """Seed `.claude/settings.local.json` in the worker worktree so dispatched
    Claude / codex / antigravity sessions inherit the okstra read-only / write
    allowlist. Mirrors the main-project seeding done in `run.py` — needed
    because `_link_sync_dirs` skips `.claude/` whenever `git worktree add`
    already materialised the directory (e.g. tracked `.claude/handoff-*.md`).
    Failures degrade to stderr warning so worktree provisioning still
    succeeds.
    """
    try:
        link = ensure_project_settings_symlink(project_root=worktree_path)
    except SettingsLinkError as exc:
        print(
            f"okstra-settings: failed to seed worker worktree symlink at "
            f"{worktree_path / '.claude/settings.local.json'} — worker dispatch "
            f"may be blocked by Claude Code permissions. ({exc})",
            file=__import__("sys").stderr,
        )
        return
    if link is None:
        print(
            "okstra-settings: ~/.okstra/templates/settings.local.json missing — "
            "re-run 'npx okstra@latest install' (0.14.0+) to provision the "
            "symlink target.",
            file=__import__("sys").stderr,
        )


def _copy_snapshot_files(source_root: Path, worktree_path: Path) -> list[str]:
    """Copy fixture files from MAIN → task worktree as read-only snapshots
    (FU-V3).

    Unlike `_link_sync_files`, this materialises a fresh on-disk copy and
    chmods it to `0o444` so the verifier can read the same rows the
    executor saw without sharing a writable handle that would corrupt the
    main worktree's copy. Skip rules mirror the symlink helpers (missing
    source → skipped silently; pre-existing dst → skipped to avoid
    clobbering tracked content).
    """
    notes: list[str] = []
    for rel in _resolve_snapshot_files(source_root):
        src = (source_root / rel).resolve()
        if not src.exists() or not src.is_file():
            continue
        dst = worktree_path / rel
        if dst.exists() or dst.is_symlink():
            continue
        dst.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(src, dst)
        try:
            os.chmod(dst, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
        except OSError:
            # chmod failure (exotic filesystems, root-squashed mounts) does
            # not invalidate the snapshot itself — it just means the read-
            # only contract is best-effort. Surface in notes so the operator
            # can audit if a verifier later mutates the file.
            notes.append(f"{rel} (rw-fallback)")
            continue
        notes.append(rel)
    return notes


def compute_worktree_path(
    *,
    project_id: str,
    task_group_segment: str,
    task_id_segment: str,
    stage_number: Optional[int] = None,
    group_id: Optional[str] = None,
) -> Path:
    """Pure path computation. One worktree dir per task-key, or per
    `<task-key>/stage-<N>` when stage_number is given (implementation
    stage isolation). Uses `OKSTRA_HOME` when set (test hook), else
    `~/.okstra`."""
    if stage_number is not None and group_id is not None:
        raise ValueError("stage_number and group_id are mutually exclusive")
    base = okstra_home()
    path = (
        base / "worktrees"
        / _safe_segment(project_id)
        / _safe_segment(task_group_segment)
        / _safe_segment(task_id_segment)
    )
    if stage_number is not None:
        path = path / f"stage-{stage_number}"
    if group_id is not None:
        path = path / f"group-{group_id}"
    return path


def compute_branch_name(
    *,
    work_category: str,
    task_id_segment: str,
    stage_number: Optional[int] = None,
    group_id: Optional[str] = None,
) -> str:
    """One branch per task-key as `<namespace>/<task-id>`, or
    `<namespace>/<task-id>-s<N>` for an implementation stage worktree. The
    namespace is a controlled constant so its slash is preserved; only the
    task-id segment is sanitised."""
    if stage_number is not None and group_id is not None:
        raise ValueError("stage_number and group_id are mutually exclusive")
    name = f"{_work_category_namespace(work_category)}/{_safe_segment(task_id_segment)}"
    if stage_number is not None:
        name = f"{name}-s{stage_number}"
    if group_id is not None:
        name = f"{name}-{group_id}"
    return name


def provision_task_worktree(
    *,
    task_type: str,
    project_root: Path,
    project_id: str,
    task_group_segment: str,
    task_id_segment: str,
    work_category: str,
    base_ref: str = "",
    require_base_ref: bool = False,
) -> WorktreeProvision:
    """Materialise (or reuse) the task worktree for this run.

    First phase of a task-key creates the worktree on a new branch.
    Subsequent phases of the same task-key look up the registry and
    return the existing path + branch unchanged.

    ``base_ref`` is the ref (branch name, tag, or commit SHA) to branch
    the new worktree from on first phase. When empty, the main worktree's
    current ``HEAD`` is used (legacy default; the CLI enforces a
    non-empty value on first phase so callers go through the
    AskUserQuestion menu in the okstra-run skill). Subsequent phases
    ignore ``base_ref`` — the registered entry's base is reused.

    Concurrency: callers must hold ``locks.worktree_provision_mutex`` for
    this task-key (run.py's prepare flow does). The exists/branch pre-checks
    and ``git worktree add`` here are not internally locked — only the
    registry reserve row is — so unlocked concurrent calls race (TOCTOU).
    flock is non-reentrant, hence the lock lives at the caller.

    Raises:
        RuntimeError when worktree creation fails (path clash on disk
        that the registry does not know about, branch clash with a
        different task-key, `git worktree add` non-zero). The caller
        (`run.py`) catches and re-raises as PrepareError to keep a
        single error surface.
    """
    decision = preview_worktree_decision(
        project_root=project_root, project_id=project_id,
        task_group_segment=task_group_segment, task_id_segment=task_id_segment,
        work_category=work_category, base_ref=base_ref,
    )

    if decision.status == "skipped-not-git":
        return WorktreeProvision(
            status="skipped-not-git",
            path=decision.path,
            note=(
                "worktree provisioning skipped: project_root is not inside a git "
                "repository; task will operate directly on project_root"
            ),
        )

    if decision.status == "skipped-in-worktree":
        return WorktreeProvision(
            status="skipped-in-worktree",
            path=decision.path,
            note=(
                "worktree provisioning skipped: project_root is already inside a "
                "non-main git worktree; task reuses the caller's worktree"
            ),
        )

    safe_project = _safe_segment(project_id)
    safe_group = _safe_segment(task_group_segment)
    safe_task = _safe_segment(task_id_segment)

    if decision.status == "reused":
        worktree_registry.touch_phase(safe_project, safe_group, safe_task, task_type)
        _seed_worktree_settings_symlink(Path(decision.path))
        return WorktreeProvision(
            status="reused",
            path=decision.path,
            branch=decision.branch,
            base_ref=decision.base_ref,
            note=(
                f"task worktree reused at {decision.path} on branch "
                f"{decision.branch} (base {decision.base_ref[:12]}); phase {task_type}"
            ),
        )

    # decision.status == "new" — proceed with creation
    worktree_path = Path(decision.path)
    branch = decision.branch

    if worktree_path.exists():
        raise RuntimeError(
            f"task worktree path already exists but is not in the registry: "
            f"{worktree_path}. Remove it with `git worktree remove <path>` "
            "(or `rm -rf` if it is not a registered worktree) before retrying."
        )
    if _branch_exists(project_root, branch):
        raise RuntimeError(_branch_exists_message(project_root, branch, "task"))

    main_root = main_worktree_path(project_root)
    requested_base = (base_ref or "").strip()
    if not requested_base and require_base_ref:
        raise RuntimeError(
            "first-phase task worktree requires an explicit base ref; "
            "pass `--base-ref <branch|tag|sha>` (or invoke through the "
            "okstra-run skill which collects this interactively)"
        )
    if requested_base:
        resolved_sha = _resolve_commit_sha(main_root, requested_base)
        if not resolved_sha:
            raise RuntimeError(
                f"could not resolve base ref `{requested_base}` in main worktree "
                f"({main_root}); ensure the branch/tag/SHA exists locally"
            )
        resolved_base_ref = resolved_sha
        base_origin = requested_base
    else:
        resolved_base_ref = _head_sha(main_root)
        if not resolved_base_ref:
            raise RuntimeError(
                "could not resolve HEAD sha in main worktree; cannot create task worktree"
            )
        base_origin = "HEAD"

    worktree_path.parent.mkdir(parents=True, exist_ok=True)
    res = _git(
        main_root,
        "worktree", "add", "-b", branch, str(worktree_path), resolved_base_ref,
    )
    if res.returncode != 0:
        raise RuntimeError(
            f"`git worktree add` failed (exit={res.returncode}): "
            f"{(res.stderr or res.stdout).strip()}"
        )

    # Sync dirs sourced from the MAIN worktree so every task sees the
    # same shared state regardless of which checkout invoked okstra.
    linked = _link_sync_dirs(main_root, worktree_path)
    linked_files = _link_sync_files(main_root, worktree_path)
    snapshot_files = _copy_snapshot_files(main_root, worktree_path)
    linked_parts: list[str] = []
    if linked:
        linked_parts.append(f"linked {', '.join(linked)}")
    if linked_files:
        linked_parts.append(f"linked-files {', '.join(linked_files)}")
    if snapshot_files:
        linked_parts.append(f"snapshot {', '.join(snapshot_files)}")
    linked_suffix = ("; " + "; ".join(linked_parts)) if linked_parts else ""

    try:
        worktree_registry.reserve(
            project_id=safe_project,
            task_group=safe_group,
            task_id=safe_task,
            worktree_path=str(worktree_path),
            branch=branch,
            base_ref=resolved_base_ref,
            phase=task_type,
        )
    except RuntimeError:
        # Roll back the on-disk worktree so the next attempt is not
        # blocked by the lingering directory / branch.
        remove_worktree_force(main_root, worktree_path)
        _git(main_root, "branch", "-D", branch)
        raise

    _seed_worktree_settings_symlink(worktree_path)

    base_label = (
        f"{base_origin} @ {resolved_base_ref[:12]}"
        if base_origin != "HEAD"
        else f"HEAD @ {resolved_base_ref[:12]}"
    )
    return WorktreeProvision(
        status="created",
        path=str(worktree_path),
        branch=branch,
        base_ref=resolved_base_ref,
        note=(
            f"task worktree created at {worktree_path} on branch {branch} "
            f"(base {base_label}; phase {task_type}){linked_suffix}"
        ),
    )


def provision_stage_worktree(
    *,
    project_root: Path,
    project_id: str,
    task_group_segment: str,
    task_id_segment: str,
    work_category: str,
    stage_number: int,
    base_commit: str,
) -> WorktreeProvision:
    """Materialise an isolated worktree for one implementation stage.

    Unlike `provision_task_worktree` (one worktree per task-key shared
    across phases), this provisions a per-stage worktree branched from
    `base_commit` at `<task-key>/stage-<N>/` on branch `<prefix>-<task>-s<N>`.
    The stage-key (`<task-key>#stage-<N>`) is reserved atomically through
    `worktree_registry`; re-entry of the same stage-key returns the
    existing entry. Branch / on-disk conflicts roll back the worktree
    before re-raising so a retry is not blocked.

    Concurrency: callers must hold ``locks.worktree_provision_mutex`` for
    the task-key, acquired BEFORE the Stage Run Claim reads the registry
    (run.py does) — otherwise two `--stage auto` runs can select the same
    stage and the loser silently enters the winner's worktree via the
    "reused" path. flock is non-reentrant, hence the lock lives at the
    caller.
    """
    if not base_commit:
        raise RuntimeError("provision_stage_worktree requires a base_commit")

    decision = resolve_stage_worktree_decision(
        project_id=project_id,
        task_group_segment=task_group_segment,
        task_id_segment=task_id_segment,
        work_category=work_category,
        stage_number=stage_number,
    )
    if decision.status == "reused":
        return WorktreeProvision(
            status="reused",
            path=decision.path,
            branch=decision.branch,
            base_ref=decision.base_ref,
            note=(
                f"stage {stage_number} worktree reused at "
                f"{decision.path} on branch {decision.branch} "
                f"(base {decision.base_ref[:12]})"
            ),
        )

    safe_project = _safe_segment(project_id)
    safe_group = _safe_segment(task_group_segment)
    safe_task = _safe_segment(task_id_segment)
    worktree_path = Path(decision.path)
    branch = decision.branch

    if worktree_path.exists():
        raise RuntimeError(
            f"stage worktree path already exists but is not in the registry: "
            f"{worktree_path}. Remove it before retrying."
        )
    if _branch_exists(project_root, branch):
        raise RuntimeError(_branch_exists_message(project_root, branch, "stage"))

    main_root = main_worktree_path(project_root)
    resolved_sha = _resolve_commit_sha(main_root, base_commit)
    if not resolved_sha:
        raise RuntimeError(
            f"could not resolve base_commit `{base_commit}` in main worktree "
            f"({main_root}); ensure the commit exists locally"
        )

    worktree_path.parent.mkdir(parents=True, exist_ok=True)
    res = _git(
        main_root,
        "worktree", "add", "-b", branch, str(worktree_path), resolved_sha,
    )
    if res.returncode != 0:
        raise RuntimeError(
            f"`git worktree add` failed (exit={res.returncode}): "
            f"{(res.stderr or res.stdout).strip()}"
        )

    _link_sync_dirs(main_root, worktree_path)
    _link_sync_files(main_root, worktree_path)
    _copy_snapshot_files(main_root, worktree_path)

    try:
        worktree_registry.reserve(
            project_id=safe_project,
            task_group=safe_group,
            task_id=safe_task,
            worktree_path=str(worktree_path),
            branch=branch,
            base_ref=resolved_sha,
            phase="implementation",
            stage_number=stage_number,
        )
    except RuntimeError:
        remove_worktree_force(main_root, worktree_path)
        _git(main_root, "branch", "-D", branch)
        raise

    _seed_worktree_settings_symlink(worktree_path)

    return WorktreeProvision(
        status="created", path=str(worktree_path),
        branch=branch, base_ref=resolved_sha,
        note=(
            f"stage {stage_number} worktree created at {worktree_path} "
            f"on branch {branch} (base {resolved_sha[:12]})"
        ),
    )