#!/usr/bin/env python3
"""Harness change resolution, leases, ports, and integration lock.

Subcommands:
  list              — list active changes
  resolve           — resolve explicit or sole active change-id
  migrate           — backfill change metadata without touching business files
  claim / release   — per-change phase lease
  lease-port        — assign an unused port from a configured range
  integration-lock  — acquire|release global main-branch integration lock

When run from a git worktree, state is resolved via git common dir to the main
project root (.harness/changes lives there). Python 3.10+, stdlib only.
"""

from __future__ import annotations

import argparse
import contextlib
import datetime as dt
import hashlib
import json
import os
import shutil
import subprocess
import sys
import time
import uuid
from pathlib import Path
from typing import Any, Callable, Mapping

SCRIPTS_DIR = Path(__file__).resolve().parent
if str(SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPTS_DIR))

import harness_paths  # noqa: E402

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")
if hasattr(sys.stderr, "reconfigure"):
    sys.stderr.reconfigure(encoding="utf-8")


RUNTIME_REL = Path(".harness") / "runtime"
LEASES_REL = RUNTIME_REL / "leases"
PORTS_REL = RUNTIME_REL / "ports"
INTEGRATION_LOCK_REL = RUNTIME_REL / "integration-lock.json"
CHANGE_CONTEXT_REL = Path("meta") / "change-context.json"
WORKTREE_META_REL = Path("meta") / "worktree.json"
CHECKPOINTS_REL = Path("meta") / "implementation-checkpoints.json"


def now_iso() -> str:
    return dt.datetime.now().astimezone().isoformat(timespec="milliseconds")


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def emit(payload: dict[str, Any], *, as_json: bool) -> None:
    text = json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
    sys.stdout.write(text if as_json else _human_line(payload))


def _human_line(payload: dict[str, Any]) -> str:
    if payload.get("ok"):
        code = payload.get("code", "OK")
        change = payload.get("changeId") or payload.get("change")
        if change:
            return f"ok code={code} change={change}\n"
        return f"ok code={code}\n"
    return f"error code={payload.get('code', 'ERROR')} message={payload.get('message')}\n"


def emit_error(
    code: str,
    message: str,
    *,
    as_json: bool,
    extra: dict[str, Any] | None = None,
    exit_code: int = 1,
) -> int:
    payload: dict[str, Any] = {"ok": False, "code": code, "message": message}
    if extra:
        payload.update(extra)
    if as_json:
        sys.stderr.write(json.dumps(payload, ensure_ascii=False) + "\n")
    else:
        sys.stderr.write(f"error: {message} ({code})\n")
    return exit_code


def _git_text(cwd: Path, *args: str) -> str | None:
    proc = subprocess.run(
        ["git", *args],
        cwd=str(cwd),
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    if proc.returncode != 0:
        return None
    return proc.stdout.strip()


def resolve_main_project_root(cwd: Path | None = None) -> Path:
    """Locate main project root from a worktree or main checkout."""
    start = (cwd or Path.cwd()).resolve()
    common_raw = _git_text(start, "rev-parse", "--git-common-dir")
    if not common_raw:
        return start
    common = Path(common_raw)
    if not common.is_absolute():
        common = (start / common).resolve()
    if common.name == ".git":
        return common.parent
    return start


def changes_dir(project_root: Path) -> Path:
    return project_root / ".harness" / "changes"


def runtime_dir(project_root: Path) -> Path:
    return project_root / RUNTIME_REL


def _read_json(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8-sig"))


def _write_json(path: Path, data: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(data, ensure_ascii=False, indent=2) + "\n"
    tmp = path.with_name(f".{path.name}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
    try:
        tmp.write_text(text, encoding="utf-8", newline="\n")
        os.replace(tmp, path)
    except BaseException:
        tmp.unlink(missing_ok=True)
        raise


@contextlib.contextmanager
def _exclusive_file_lock(path: Path, wait_seconds: float = 5.0):
    path.parent.mkdir(parents=True, exist_ok=True)
    deadline = time.monotonic() + wait_seconds
    descriptor: int | None = None
    while descriptor is None:
        try:
            descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        except (FileExistsError, PermissionError):
            if time.monotonic() >= deadline:
                raise TimeoutError(f"lock unavailable: {path}")
            time.sleep(0.01)
    try:
        os.write(descriptor, f"{os.getpid()}\n".encode("ascii"))
        yield
    finally:
        os.close(descriptor)
        path.unlink(missing_ok=True)


def list_active_changes(project_root: Path) -> list[dict[str, Any]]:
    root = changes_dir(project_root)
    if not root.is_dir():
        return []
    active: list[dict[str, Any]] = []
    for entry in sorted(root.iterdir()):
        if not entry.is_dir():
            continue
        meta_archived = entry / "meta" / "archived.json"
        if meta_archived.is_file():
            try:
                archived = _read_json(meta_archived)
            except (OSError, json.JSONDecodeError):
                archived = {}
            if isinstance(archived, dict) and archived.get("status") == "archived":
                continue
        has_plan = any((entry / "plans").glob("*-plan.md"))
        has_context = (entry / CHANGE_CONTEXT_REL).is_file()
        has_checkpoints = (entry / CHECKPOINTS_REL).is_file()
        worktree_path = entry / WORKTREE_META_REL
        has_active_worktree = False
        if worktree_path.is_file():
            try:
                worktree = _read_json(worktree_path)
                has_active_worktree = isinstance(worktree, dict) and (
                    worktree.get("created") is True or
                    worktree.get("requested") is True
                )
            except (OSError, json.JSONDecodeError):
                has_active_worktree = False
        # Runtime notes, event logs and requested=false worktree metadata can
        # survive archive/submit cleanup. They are residues, not active changes.
        if not (has_plan or has_context or has_checkpoints or has_active_worktree):
            continue
        active.append(
            {
                "changeId": entry.name,
                "path": str(entry.resolve()),
                "hasPlan": has_plan,
                "hasWorktreeMeta": worktree_path.is_file(),
            }
        )
    return active


def _verified_archive_receipts(project_root: Path) -> dict[str, dict[str, Any]]:
    results: dict[str, dict[str, Any]] = {}
    archive_root = project_root / ".harness" / "archive"
    if not archive_root.is_dir():
        return results
    for archive_dir in sorted(archive_root.iterdir()):
        if not archive_dir.is_dir():
            continue
        receipt_path = archive_dir / "meta" / "archive-receipt.json"
        summary_path = archive_dir / "reports" / "final" / "summary-data.json"
        if not receipt_path.is_file() or not summary_path.is_file():
            continue
        try:
            receipt = _read_json(receipt_path)
            summary = _read_json(summary_path)
        except (OSError, json.JSONDecodeError):
            continue
        if not isinstance(receipt, dict) or not isinstance(summary, dict):
            continue
        change_name = str(
            receipt.get("changeName") or summary.get("changeName") or ""
        ).strip()
        if not change_name or receipt.get("status") != "archived":
            continue
        expected = str(receipt.get("summarySha256") or "").removeprefix("sha256:")
        actual = sha256_file(summary_path)
        results[change_name] = {
            "archivePath": str(archive_dir.resolve()),
            "receiptPath": str(receipt_path.resolve()),
            "summaryPath": str(summary_path.resolve()),
            "expectedSha256": expected,
            "actualSha256": actual,
            "verified": bool(expected) and expected == actual,
        }
    return results


def classify_changes(project_root: Path) -> list[dict[str, Any]]:
    """Classify every change directory without deleting or rewriting anything."""
    root = changes_dir(project_root)
    if not root.is_dir():
        return []
    active_ids = {
        item["changeId"] for item in list_active_changes(project_root)
    }
    archive_receipts = _verified_archive_receipts(project_root)
    results: list[dict[str, Any]] = []
    for entry in sorted(root.iterdir()):
        if not entry.is_dir():
            continue
        reasons: list[str] = []
        evidence: list[str] = []
        archive = archive_receipts.get(entry.name)
        if archive is not None:
            evidence.extend([archive["receiptPath"], archive["summaryPath"]])
            if archive["verified"]:
                status = "ARCHIVED_LEFTOVER"
                reasons.append("ARCHIVE_RECEIPT_VERIFIED")
            else:
                status = "INVALID"
                reasons.append("ARCHIVE_HASH_MISMATCH")
        elif entry.name in active_ids:
            status = "ACTIVE"
            reasons.append("ACTIVE_CHANGE_EVIDENCE")
        else:
            archived_meta = entry / "meta" / "archived.json"
            if archived_meta.is_file():
                try:
                    archived = _read_json(archived_meta)
                except (OSError, json.JSONDecodeError):
                    archived = None
                if not isinstance(archived, dict):
                    status = "INVALID"
                    reasons.append("ARCHIVED_METADATA_INVALID")
                elif archived.get("status") == "archived":
                    status = "ARCHIVED_LEFTOVER"
                    reasons.append("LOCAL_ARCHIVED_MARKER")
                    evidence.append(str(archived_meta.resolve()))
                else:
                    status = "INVALID"
                    reasons.append("ARCHIVED_METADATA_CONTRADICTORY")
            elif (
                (entry / "events.ndjson").is_file()
                or (entry / "runtime").is_dir()
                or (entry / "evidence").is_dir()
            ):
                status = "RECOVERABLE"
                reasons.append("INTERRUPTED_CHANGE_EVIDENCE")
            else:
                status = "ORPHAN"
                reasons.append("NO_ACTIVITY_OR_ARCHIVE_EVIDENCE")
        results.append({
            "changeId": entry.name,
            "path": str(entry.resolve()),
            "status": status,
            "reasonCodes": reasons,
            "evidence": evidence,
            "safeToCleanup": status == "ARCHIVED_LEFTOVER",
        })
    return results


def cleanup_changes(project_root: Path, *, apply: bool = False) -> dict[str, Any]:
    """Safely quarantine only hash-verified archived leftovers."""
    statuses = classify_changes(project_root)
    eligible = sorted(
        item["changeId"]
        for item in statuses
        if item["status"] == "ARCHIVED_LEFTOVER"
        and "ARCHIVE_RECEIPT_VERIFIED" in item["reasonCodes"]
    )
    moved: list[dict[str, str]] = []
    if apply and eligible:
        stamp = now_iso().replace(":", "-").replace(".", "-")
        quarantine = (
            project_root / ".harness" / "runtime"
            / "change-cleanup" / stamp
        )
        quarantine.mkdir(parents=True, exist_ok=True)
        for change_id in eligible:
            source = changes_dir(project_root) / change_id
            target = quarantine / change_id
            if not source.is_dir() or target.exists():
                continue
            shutil.move(str(source), str(target))
            moved.append({
                "changeId": change_id,
                "from": str(source),
                "quarantine": str(target),
            })
    return {
        "ok": True,
        "dryRun": not apply,
        "eligible": eligible,
        "moved": moved,
        "recoverable": True,
    }


def change_dir_for_id(project_root: Path, change_id: str) -> Path | None:
    """按 change id 或其路径写法解析 change 目录。

    2026-09 dogfood: 多个 SKILL.md 教 --change-dir ".harness/changes/<cn>",
    而这里此前只接受裸 id, 按文档路径传参会拼成
    .harness/changes/.harness/changes/<cn> 报 CHANGE_NOT_FOUND (agent 照文档
    执行必卡)。兼容两种写法: 裸 id 或 .harness/changes/<cn> 前缀, 取末段解析。
    """
    normalized = (change_id or "").strip().replace("\\", "/").strip("/")
    if normalized.startswith(".harness/changes/"):
        normalized = normalized[len(".harness/changes/"):]
    candidate = changes_dir(project_root) / normalized
    if normalized and candidate.is_dir():
        return candidate.resolve()
    return None


def _layout_fields(project_root: Path, change_id: str) -> dict[str, Any]:
    """Layout enrichment for resolve payloads; empty on any resolution issue."""
    try:
        layout = harness_paths.resolve_change_layout(project_root, change_id)
    except (FileNotFoundError, ValueError, OSError):
        return {}
    return {
        "contractRoot": layout["contractRoot"],
        "stateRoot": layout["stateRoot"],
        "layout": layout["layout"],
        "repositoryId": layout["repositoryId"],
    }


def read_concurrency_mode(project_root: Path) -> str:
    """Return the configured concurrency mode (retro §5.2).

    Defaults to ``single-active`` when no config declares a mode. Supported
    values: ``single-active`` (default, blocks a second active Change),
    ``isolated-multi-active`` (allows multiple active Changes but all
    Change-scoped commands require ``--change``).
    """
    configs = [
        project_root / ".harness" / "config" / "harness.json",
        project_root / ".harness" / "config.json",
    ]
    for cfg in configs:
        if not cfg.is_file():
            continue
        try:
            data = json.loads(cfg.read_text(encoding="utf-8"))
            if isinstance(data, dict):
                mode = data.get("concurrencyMode")
                if isinstance(mode, str) and mode in {
                    "single-active",
                    "isolated-multi-active",
                }:
                    return mode
        except (OSError, json.JSONDecodeError):
            continue
    return "single-active"


def check_concurrency_block(
    project_root: Path, change_id: str
) -> dict[str, Any] | None:
    """Return a blocking payload when a second active Change is forbidden.

    Returns ``None`` when the begin may proceed. In ``single-active`` mode a
    second active Change (any other than ``change_id``) blocks begin; in
    ``isolated-multi-active`` mode multiple active Changes are allowed.
    """
    mode = read_concurrency_mode(project_root)
    if mode == "isolated-multi-active":
        return None
    active = list_active_changes(project_root)
    others = [entry for entry in active if entry.get("changeId") != change_id]
    if not others:
        return None
    return {
        "ok": False,
        "code": "SINGLE_ACTIVE_BLOCKED",
        "message": (
            "single-active concurrency mode: another active Change exists; "
            "use portfolio/decompose to sequence multiple Changes, or switch "
            "to isolated-multi-active after ensuring full Change-scoped isolation"
        ),
        "concurrencyMode": mode,
        "activeChanges": active,
        "blockingChanges": others,
    }


def resolve_change(
    project_root: Path,
    change_id: str | None,
) -> dict[str, Any]:
    active = list_active_changes(project_root)
    if change_id:
        resolved = change_dir_for_id(project_root, change_id)
        if resolved is None:
            return {
                "ok": False,
                "code": "CHANGE_NOT_FOUND",
                "message": f"change not found: {change_id}",
                "changeId": change_id,
            }
        return {
            "ok": True,
            "code": "RESOLVED",
            "changeId": change_id,
            "changeDir": str(resolved),
            "projectRoot": str(project_root.resolve()),
            "activeCount": len(active),
            **_layout_fields(project_root, change_id),
        }
    if not active:
        return {
            "ok": False,
            "code": "NO_ACTIVE_CHANGE",
            "message": "no active change under .harness/changes",
            "activeChanges": [],
        }
    if len(active) == 1:
        only = active[0]
        return {
            "ok": True,
            "code": "RESOLVED",
            "changeId": only["changeId"],
            "changeDir": only["path"],
            "projectRoot": str(project_root.resolve()),
            "activeCount": 1,
            "autoSelected": True,
            **_layout_fields(project_root, only["changeId"]),
        }
    return {
        "ok": False,
        "code": "CHANGE_SELECTION_REQUIRED",
        "message": "multiple active changes; pass --change <id>",
        "activeChanges": active,
        "activeCount": len(active),
    }


def migrate_change(project_root: Path, change_id: str) -> dict[str, Any]:
    resolved = resolve_change(project_root, change_id)
    if not resolved.get("ok"):
        return resolved
    change_dir = Path(resolved["changeDir"])
    created: list[str] = []

    context_path = change_dir / CHANGE_CONTEXT_REL
    if not context_path.is_file():
        worktree_meta = change_dir / WORKTREE_META_REL
        worktree_root = project_root.resolve()
        branch = _git_text(project_root, "rev-parse", "--abbrev-ref", "HEAD") or ""
        if worktree_meta.is_file():
            try:
                wt = _read_json(worktree_meta)
                if isinstance(wt, dict):
                    if isinstance(wt.get("path"), str) and wt["path"].strip():
                        worktree_root = Path(wt["path"]).expanduser().resolve()
                    elif isinstance(wt.get("worktreeRoot"), str) and wt["worktreeRoot"].strip():
                        worktree_root = Path(wt["worktreeRoot"]).expanduser().resolve()
                    if isinstance(wt.get("branch"), str) and wt["branch"].strip():
                        branch = wt["branch"].strip()
            except (OSError, json.JSONDecodeError):
                pass
        context = {
            "schemaVersion": 1,
            "changeId": change_id,
            "mainProjectRoot": str(project_root.resolve()),
            "worktreeRoot": str(worktree_root),
            "stateDir": str(change_dir.resolve()),
            "branch": branch,
            "migratedAt": now_iso(),
        }
        _write_json(context_path, context)
        created.append(str(context_path.relative_to(change_dir)))

    checkpoints_path = change_dir / CHECKPOINTS_REL
    if not checkpoints_path.is_file():
        checkpoints = {
            "schemaVersion": 1,
            "changeName": change_id,
            "checkpoints": [
                {
                    "id": "foundation-gate",
                    "afterTasks": [1, 2, 3, 4],
                    "beforeTasks": [6, 7, 8, 9, 10],
                    "reviewerTool": "codex",
                    "status": "pending",
                    "blocking": True,
                    "requiredReport": "reports/review/foundation-gate-review.md",
                    "purpose": "Block tasks 6+ until foundation interfaces are reviewed.",
                }
            ],
        }
        _write_json(checkpoints_path, checkpoints)
        created.append(str(checkpoints_path.relative_to(change_dir)))

    return {
        "ok": True,
        "code": "MIGRATED",
        "changeId": change_id,
        "changeDir": str(change_dir),
        "created": created,
    }


def _lease_path(project_root: Path, change_id: str) -> Path:
    return project_root / LEASES_REL / f"{change_id}.json"


def _lease_expired(lease: dict[str, Any]) -> bool:
    expires = lease.get("expiresAt")
    if not isinstance(expires, str) or not expires.strip():
        return True
    try:
        exp_dt = dt.datetime.fromisoformat(expires.replace("Z", "+00:00"))
    except ValueError:
        return True
    return dt.datetime.now().astimezone() >= exp_dt


def inspect_lease_state(project_root: Path, change_id: str) -> dict[str, Any]:
    """Read the lease without mutating it, keeping the three failure modes apart.

    ``inspect_lease`` collapses absent / expired / corrupt into ``None``, so a
    caller cannot tell "nobody ever held this" from "the phase outlived its
    TTL". Those need different answers: an expired lease still carries the
    runId that proves who owns the phase, a corrupt one proves nothing, and an
    absent one may just mean the phase already closed.

    state is one of ``active`` | ``expired`` | ``absent`` | ``corrupt``. The
    lease dict is returned for ``active`` and ``expired``; ``None`` otherwise.
    """
    path = _lease_path(project_root, change_id)
    if not path.is_file():
        return {"state": "absent", "lease": None}
    try:
        lease = _read_json(path)
    except (OSError, json.JSONDecodeError):
        return {"state": "corrupt", "lease": None}
    if not isinstance(lease, dict):
        return {"state": "corrupt", "lease": None}
    if _lease_expired(lease):
        return {"state": "expired", "lease": lease}
    return {"state": "active", "lease": lease}


def inspect_lease(project_root: Path, change_id: str) -> dict[str, Any] | None:
    """Return the current non-expired lease without mutating it."""
    state = inspect_lease_state(project_root, change_id)
    return state["lease"] if state["state"] == "active" else None


def _claim_lease_locked(
    project_root: Path,
    *,
    change_id: str,
    phase: str,
    run_id: str,
    ttl_seconds: int,
    steal: bool = False,
    expected_generation: int | None = None,
) -> dict[str, Any]:
    path = _lease_path(project_root, change_id)
    path.parent.mkdir(parents=True, exist_ok=True)
    now = dt.datetime.now().astimezone()
    expires = now + dt.timedelta(seconds=max(1, ttl_seconds))
    new_lease = {
        "leaseId": str(uuid.uuid4()),
        "changeId": change_id,
        "phase": phase,
        "runId": run_id,
        "pid": os.getpid(),
        "acquiredAt": now.isoformat(timespec="milliseconds"),
        "expiresAt": expires.isoformat(timespec="milliseconds"),
        "ttlSeconds": ttl_seconds,
        "generation": 1,
    }
    if path.is_file():
        try:
            existing = _read_json(path)
        except (OSError, json.JSONDecodeError):
            existing = None
        if isinstance(existing, dict) and not _lease_expired(existing):
            if expected_generation is not None and existing.get("generation") != expected_generation:
                return {
                    "ok": False,
                    "code": "LEASE_GENERATION_CONFLICT",
                    "holder": existing,
                }
            same_owner = str(existing.get("runId")) == run_id
            if same_owner:
                existing.update(
                    {
                        "phase": phase,
                        "pid": os.getpid(),
                        "expiresAt": expires.isoformat(timespec="milliseconds"),
                        "refreshedAt": now.isoformat(timespec="milliseconds"),
                        "generation": int(existing.get("generation") or 1) + 1,
                    }
                )
                _write_json(path, existing)
                return {"ok": True, "code": "LEASE_REFRESHED", "lease": existing}
            if not steal:
                return {
                    "ok": False,
                    "code": "LEASE_CONFLICT",
                    "message": "change lease held by another run",
                    "holder": existing,
                }
    _write_json(path, new_lease)
    return {"ok": True, "code": "LEASE_CLAIMED", "lease": new_lease}


def claim_lease(
    project_root: Path,
    *,
    change_id: str,
    phase: str,
    run_id: str,
    ttl_seconds: int,
    steal: bool = False,
    expected_generation: int | None = None,
) -> dict[str, Any]:
    path = _lease_path(project_root, change_id)
    with _exclusive_file_lock(path.with_suffix(".lock")):
        return _claim_lease_locked(
            project_root,
            change_id=change_id,
            phase=phase,
            run_id=run_id,
            ttl_seconds=ttl_seconds,
            steal=steal,
            expected_generation=expected_generation,
        )


def _release_lease_locked(
    project_root: Path,
    *,
    change_id: str,
    phase: str,
    run_id: str,
    lease_id: str | None = None,
    generation: int | None = None,
) -> dict[str, Any]:
    path = _lease_path(project_root, change_id)
    if not path.is_file():
        return {
            "ok": True,
            "code": "LEASE_ABSENT",
            "message": "no lease file present",
        }
    try:
        existing = _read_json(path)
    except (OSError, json.JSONDecodeError) as exc:
        return {
            "ok": False,
            "code": "LEASE_INVALID",
            "message": str(exc),
        }
    if not isinstance(existing, dict):
        return {"ok": False, "code": "LEASE_INVALID", "message": "lease is not an object"}
    if str(existing.get("runId")) != run_id:
        return {
            "ok": False,
            "code": "LEASE_OWNER_MISMATCH",
            "message": "run id does not match lease owner",
            "holder": existing,
        }
    if str(existing.get("phase")) != phase:
        return {
            "ok": False,
            "code": "LEASE_PHASE_MISMATCH",
            "message": f"lease phase is {existing.get('phase')}, not {phase}",
            "holder": existing,
        }
    if lease_id is not None and str(existing.get("leaseId")) != lease_id:
        return {
            "ok": False,
            "code": "LEASE_ID_MISMATCH",
            "holder": existing,
        }
    if generation is not None and existing.get("generation") != generation:
        return {
            "ok": False,
            "code": "LEASE_GENERATION_CONFLICT",
            "holder": existing,
        }
    path.unlink(missing_ok=True)
    return {"ok": True, "code": "LEASE_RELEASED", "changeId": change_id, "phase": phase}


def release_lease(
    project_root: Path,
    *,
    change_id: str,
    phase: str,
    run_id: str,
    lease_id: str | None = None,
    generation: int | None = None,
) -> dict[str, Any]:
    path = _lease_path(project_root, change_id)
    with _exclusive_file_lock(path.with_suffix(".lock")):
        return _release_lease_locked(
            project_root,
            change_id=change_id,
            phase=phase,
            run_id=run_id,
            lease_id=lease_id,
            generation=generation,
        )


def lease_port(
    project_root: Path,
    *,
    change_id: str,
    run_id: str,
    port_range: tuple[int, int],
    generation: int = 1,
    listener_identity: Mapping[str, Any] | None = None,
) -> dict[str, Any]:
    start, end = port_range
    if start > end:
        return {
            "ok": False,
            "code": "INVALID_PORT_RANGE",
            "message": f"invalid range {start}-{end}",
        }
    ports_root = project_root / PORTS_REL
    ports_root.mkdir(parents=True, exist_ok=True)
    registry_path = ports_root / "registry.json"
    with _exclusive_file_lock(registry_path.with_suffix(".lock")):
        registry: dict[str, Any] = {"leases": []}
        if registry_path.is_file():
            try:
                loaded = _read_json(registry_path)
                if isinstance(loaded, dict) and isinstance(loaded.get("leases"), list):
                    registry = loaded
            except (OSError, json.JSONDecodeError):
                registry = {"leases": []}

        leases = [
            item for item in registry["leases"]
            if isinstance(item, dict) and not _lease_expired(item)
        ]
        used = {
            int(item["port"])
            for item in leases
            if isinstance(item.get("port"), int)
        }
        for port in range(start, end + 1):
            if port not in used:
                lease_id = str(uuid.uuid4())
                entry = {
                    "leaseId": lease_id,
                    "changeId": change_id,
                    "runId": run_id,
                    "generation": generation,
                    "pid": os.getpid(),
                    "port": port,
                    "acquiredAt": now_iso(),
                    "expiresAt": (
                        dt.datetime.now().astimezone() + dt.timedelta(hours=4)
                    ).isoformat(timespec="milliseconds"),
                }
                if isinstance(listener_identity, Mapping):
                    entry["listenerIdentity"] = dict(listener_identity)
                    entry["listenerProofRequired"] = True
                leases.append(entry)
                registry["leases"] = leases
                _write_json(registry_path, registry)
                return {"ok": True, "code": "PORT_LEASED", "port": port, "leaseId": lease_id, "lease": entry}
    return {
        "ok": False,
        "code": "PORT_RANGE_EXHAUSTED",
        "message": f"no free port in {start}-{end}",
    }


def release_port(
    project_root: Path,
    *,
    change_id: str,
    run_id: str,
    port: int | None = None,
    lease_id: str | None = None,
    generation: int | None = None,
    listener_identity: Mapping[str, Any] | None = None,
    listener_observer: Callable[[int], Mapping[str, Any] | None] | None = None,
) -> dict[str, Any]:
    """Release port leases by subset (retro §5.16).

    - `--lease-id`: release only the matching leaseId.
    - `--port`: release only the matching port (must match changeId+runId).
    - Neither: release all leases matching (changeId, runId) — does NOT require
      all leases under changeId to have the same runId.
    """
    registry_path = project_root / PORTS_REL / "registry.json"
    if not registry_path.is_file():
        return {"ok": True, "code": "PORT_LEASE_ABSENT", "changeId": change_id}
    with _exclusive_file_lock(registry_path.with_suffix(".lock")):
        try:
            registry = _read_json(registry_path)
        except (OSError, json.JSONDecodeError) as exc:
            return {"ok": False, "code": "PORT_REGISTRY_INVALID", "message": str(exc)}
        leases = registry.get("leases") if isinstance(registry, dict) else None
        if not isinstance(leases, list):
            return {"ok": False, "code": "PORT_REGISTRY_INVALID", "message": "leases is not a list"}

        # Select the subset to release based on lease_id, port, or (changeId, runId)
        if lease_id is not None:
            to_release = [
                item for item in leases
                if isinstance(item, dict) and str(item.get("leaseId")) == lease_id
            ]
            # Validate that the lease belongs to this change
            for item in to_release:
                if generation is not None and item.get("generation") != generation:
                    return {
                        "ok": False,
                        "code": "PORT_LEASE_GENERATION_CONFLICT",
                        "holder": item,
                    }
                if str(item.get("changeId")) != change_id:
                    return {
                        "ok": False,
                        "code": "PORT_LEASE_OWNER_MISMATCH",
                        "message": "leaseId does not belong to this change",
                        "holder": item,
                    }
        elif port is not None:
            to_release = [
                item for item in leases
                if isinstance(item, dict)
                and item.get("port") == port
                and str(item.get("changeId")) == change_id
            ]
            # Validate runId matches
            for item in to_release:
                if generation is not None and item.get("generation") != generation:
                    return {
                        "ok": False,
                        "code": "PORT_LEASE_GENERATION_CONFLICT",
                        "holder": item,
                    }
                if str(item.get("runId")) != run_id:
                    return {
                        "ok": False,
                        "code": "PORT_LEASE_OWNER_MISMATCH",
                        "message": f"port {port} owned by different runId",
                        "holder": item,
                        "conflictingOwners": [
                            {"runId": str(i.get("runId")), "port": i.get("port")}
                            for i in leases
                            if isinstance(i, dict)
                            and i.get("port") == port
                            and str(i.get("changeId")) == change_id
                        ],
                    }
        else:
            # Release all matching (changeId, runId) — subset release
            to_release = [
                item for item in leases
                if isinstance(item, dict)
                and str(item.get("changeId")) == change_id
                and str(item.get("runId")) == run_id
            ]
            if generation is not None:
                mismatched = [
                    item for item in to_release if item.get("generation") != generation
                ]
                if mismatched:
                    return {
                        "ok": False,
                        "code": "PORT_LEASE_GENERATION_CONFLICT",
                        "holder": mismatched[0],
                    }

        for item in to_release:
            expected_listener = item.get("listenerIdentity")
            if not isinstance(expected_listener, Mapping):
                continue
            observed_listener = (
                listener_observer(int(item["port"]))
                if listener_observer is not None and isinstance(item.get("port"), int)
                else listener_identity
            )
            if not isinstance(observed_listener, Mapping):
                return {
                    "ok": False,
                    "code": "LISTENER_IDENTITY_UNVERIFIABLE",
                    "message": "listener identity could not be independently observed",
                    "holder": item,
                }
            try:
                from harness_process import verify_process_identity

                decision = verify_process_identity(expected_listener, observed_listener)
            except (ImportError, TypeError, ValueError):
                decision = {"ok": False, "reasonCode": "IDENTITY_UNVERIFIABLE"}
            if decision.get("ok") is not True:
                return {
                    "ok": False,
                    "code": "LISTENER_IDENTITY_UNVERIFIABLE",
                    "message": "listener identity did not match the leased service",
                    "details": decision,
                    "holder": item,
                }

        if not to_release:
            # Check if there are other leases under this changeId (different runId)
            other_owned = [
                item for item in leases
                if isinstance(item, dict)
                and str(item.get("changeId")) == change_id
                and str(item.get("runId")) != run_id
            ]
            if other_owned:
                return {
                    "ok": False,
                    "code": "PORT_LEASE_OWNER_MISMATCH",
                    "message": "no matching leases for this runId; other runIds exist under this changeId",
                    "conflictingOwners": [
                        {"runId": str(i.get("runId")), "port": i.get("port"), "leaseId": i.get("leaseId")}
                        for i in other_owned
                    ],
                }
            return {"ok": True, "code": "PORT_LEASE_ABSENT", "changeId": change_id}

        released_ports = [item.get("port") for item in to_release]
        released_ids = [item.get("leaseId") for item in to_release]
        registry["leases"] = [item for item in leases if item not in to_release]
        _write_json(registry_path, registry)
        return {
            "ok": True,
            "code": "PORT_LEASE_RELEASED",
            "changeId": change_id,
            "ports": released_ports,
            "leaseIds": released_ids,
        }


def _integration_lock_acquire_locked(
    project_root: Path,
    *,
    run_id: str,
    ttl_seconds: int = 3600,
) -> dict[str, Any]:
    path = project_root / INTEGRATION_LOCK_REL
    path.parent.mkdir(parents=True, exist_ok=True)
    now = dt.datetime.now().astimezone()
    payload = {
        "runId": run_id,
        "pid": os.getpid(),
        "acquiredAt": now.isoformat(timespec="milliseconds"),
        "expiresAt": (now + dt.timedelta(seconds=ttl_seconds)).isoformat(
            timespec="milliseconds"
        ),
    }
    if path.is_file():
        try:
            existing = _read_json(path)
        except (OSError, json.JSONDecodeError):
            existing = None
        if isinstance(existing, dict) and not _lease_expired(existing):
            if str(existing.get("runId")) == run_id:
                _write_json(path, payload)
                return {"ok": True, "code": "INTEGRATION_LOCK_REFRESHED", "lock": payload}
            return {
                "ok": False,
                "code": "INTEGRATION_LOCK_HELD",
                "message": "integration lock held by another run",
                "holder": existing,
            }
    _write_json(path, payload)
    return {"ok": True, "code": "INTEGRATION_LOCK_ACQUIRED", "lock": payload}


def integration_lock_acquire(
    project_root: Path,
    *,
    run_id: str,
    ttl_seconds: int = 3600,
) -> dict[str, Any]:
    path = project_root / INTEGRATION_LOCK_REL
    with _exclusive_file_lock(path.with_suffix(".lock")):
        return _integration_lock_acquire_locked(
            project_root, run_id=run_id, ttl_seconds=ttl_seconds
        )


def _integration_lock_release_locked(project_root: Path, *, run_id: str) -> dict[str, Any]:
    path = project_root / INTEGRATION_LOCK_REL
    if not path.is_file():
        return {"ok": True, "code": "INTEGRATION_LOCK_ABSENT"}
    try:
        existing = _read_json(path)
    except (OSError, json.JSONDecodeError) as exc:
        return {"ok": False, "code": "INTEGRATION_LOCK_INVALID", "message": str(exc)}
    if not isinstance(existing, dict):
        return {"ok": False, "code": "INTEGRATION_LOCK_INVALID", "message": "not an object"}
    if str(existing.get("runId")) != run_id:
        return {
            "ok": False,
            "code": "INTEGRATION_LOCK_OWNER_MISMATCH",
            "message": "run id does not match lock owner",
            "holder": existing,
        }
    path.unlink(missing_ok=True)
    return {"ok": True, "code": "INTEGRATION_LOCK_RELEASED"}


def integration_lock_release(project_root: Path, *, run_id: str) -> dict[str, Any]:
    path = project_root / INTEGRATION_LOCK_REL
    with _exclusive_file_lock(path.with_suffix(".lock")):
        return _integration_lock_release_locked(project_root, run_id=run_id)


def parse_port_range(raw: str) -> tuple[int, int] | None:
    if "-" not in raw:
        return None
    left, right = raw.split("-", 1)
    try:
        return int(left.strip()), int(right.strip())
    except ValueError:
        return None


def cmd_list(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    payload = {
        "ok": True,
        "code": "LISTED",
        "projectRoot": str(project),
        "activeChanges": list_active_changes(project),
    }
    emit(payload, as_json=bool(args.json))
    return 0


def cmd_resolve(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    payload = resolve_change(project, args.change)
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "RESOLVE_FAILED")),
        str(payload.get("message", "resolve failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def allow_local_release(project_root: Path, change_id: str) -> dict[str, Any]:
    """在 gate-policy.json 写入 candidateVerification.allowLocalRelease=true。

    archive 的 PROJECT_RELEASE_POLICY_BLOCKED 要求这个策略位，此前只能手工
    编辑 meta/gate-policy.json（2026-08-30 sales-insight-agent archive 实测）。
    这里给出正规入口：保留既有策略内容，只翻转一个布尔位；幂等。
    """
    resolved = resolve_change(project_root, change_id)
    if not resolved.get("ok"):
        return resolved
    change_dir = Path(resolved["changeDir"])
    policy_path = change_dir / "meta" / "gate-policy.json"
    policy: dict[str, Any] = {"schemaVersion": 1}
    if policy_path.is_file():
        try:
            data = _read_json(policy_path)
        except (OSError, json.JSONDecodeError) as exc:
            return {
                "ok": False,
                "code": "GATE_POLICY_INVALID",
                "message": str(exc),
                "path": str(policy_path),
            }
        if not isinstance(data, dict) or data.get("schemaVersion") != 1:
            return {
                "ok": False,
                "code": "GATE_POLICY_INVALID",
                "message": "gate-policy.json 必须是 schemaVersion=1 的对象",
                "path": str(policy_path),
            }
        policy = data
    candidate = policy.get("candidateVerification")
    candidate = dict(candidate) if isinstance(candidate, dict) else {}
    if candidate.get("allowLocalRelease") is True:
        return {
            "ok": True,
            "code": "LOCAL_RELEASE_ALREADY_ALLOWED",
            "changeId": resolved["changeId"],
            "path": str(policy_path),
            "candidateVerification": candidate,
        }
    candidate["allowLocalRelease"] = True
    policy["candidateVerification"] = candidate
    _write_json(policy_path, policy)
    return {
        "ok": True,
        "code": "LOCAL_RELEASE_ALLOWED",
        "changeId": resolved["changeId"],
        "path": str(policy_path),
        "candidateVerification": candidate,
    }


def declare_product_ownership(
    project_root: Path, change_id: str, *, product_paths: list[str]
) -> dict[str, Any]:
    """把 `ownership.productPaths` 写进 change 契约。

    这个字段此前没有任何写入方：plan 的 validate_product_ownership 只校验、缺失时
    软放行，而归档的 compute_ownership_diff 会把全部改动判成 foreignPaths，
    filesChanged=0 直接触发 DIFF_ZERO_WITH_NONEMPTY_COMMIT。两端口径不一致，中间
    没工具能补——只能手改契约。这里给出正规入口。

    规则与 plan 校验一致：只收精确文件或目录前缀，不接受通配。
    """
    resolved = resolve_change(project_root, change_id)
    if not resolved.get("ok"):
        return resolved
    change_dir = Path(resolved["changeDir"])
    normalized = sorted({
        str(item).strip().replace("\\", "/").removeprefix("./")
        for item in product_paths
        if isinstance(item, str) and item.strip()
    })
    if not normalized:
        return {
            "ok": False,
            "code": "PLAN_PRODUCT_PATHS_REQUIRED",
            "message": "至少声明一条 productPaths（精确文件或目录前缀）",
        }
    unsupported = sorted(p for p in normalized if any(ch in p for ch in "*?[]"))
    if unsupported:
        return {
            "ok": False,
            "code": "PLAN_PRODUCT_PATHS_GLOB_UNSUPPORTED",
            "message": "productPaths 只接受精确文件或目录前缀，不支持通配：" + ", ".join(unsupported),
            "unsupportedPaths": unsupported,
        }

    context_path = change_dir / CHANGE_CONTEXT_REL
    try:
        context = _read_json(context_path)
    except (OSError, json.JSONDecodeError) as exc:
        return {"ok": False, "code": "CHANGE_CONTEXT_INVALID", "message": str(exc)}
    if not isinstance(context, dict):
        return {"ok": False, "code": "CHANGE_CONTEXT_INVALID", "message": "change-context.json 不是对象"}

    ownership = context.get("ownership")
    ownership = dict(ownership) if isinstance(ownership, dict) else {}
    if ownership.get("productPaths") == normalized:
        return {
            "ok": True,
            "code": "PRODUCT_OWNERSHIP_DECLARED",
            "idempotent": True,
            "changeId": change_id,
            "productPaths": normalized,
            "path": str(context_path),
        }
    ownership["productPaths"] = normalized
    context["ownership"] = ownership
    _write_json(context_path, context)
    return {
        "ok": True,
        "code": "PRODUCT_OWNERSHIP_DECLARED",
        "idempotent": False,
        "changeId": change_id,
        "productPaths": normalized,
        "path": str(context_path),
    }


def cmd_allow_local_release(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    payload = allow_local_release(project, args.change)
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "ALLOW_LOCAL_RELEASE_FAILED")),
        str(payload.get("message", "allow local release failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_declare_ownership(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    payload = declare_product_ownership(
        project, args.change, product_paths=list(args.product_path)
    )
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "PRODUCT_OWNERSHIP_FAILED")),
        str(payload.get("message", "declare ownership failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_migrate(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    if not args.change:
        return emit_error("CHANGE_REQUIRED", "--change is required", as_json=bool(args.json))
    payload = migrate_change(project, args.change)
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "MIGRATE_FAILED")),
        str(payload.get("message", "migrate failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_claim(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    payload = claim_lease(
        project,
        change_id=args.change,
        phase=args.phase,
        run_id=args.run_id,
        ttl_seconds=int(args.ttl_seconds),
        steal=bool(args.steal),
        expected_generation=getattr(args, "expected_generation", None),
    )
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "CLAIM_FAILED")),
        str(payload.get("message", "claim failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_release(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    payload = release_lease(
        project,
        change_id=args.change,
        phase=args.phase,
        run_id=args.run_id,
        lease_id=getattr(args, "lease_id", None),
        generation=getattr(args, "generation", None),
    )
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "RELEASE_FAILED")),
        str(payload.get("message", "release failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_lease_port(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    parsed = parse_port_range(args.range)
    if parsed is None:
        return emit_error(
            "INVALID_PORT_RANGE",
            f"expected --range <start-end>, got {args.range!r}",
            as_json=bool(args.json),
        )
    listener_identity = None
    if getattr(args, "listener_identity_json", None):
        try:
            parsed_listener = json.loads(args.listener_identity_json)
            if not isinstance(parsed_listener, dict):
                raise ValueError("listener identity must be an object")
            listener_identity = parsed_listener
        except (TypeError, ValueError, json.JSONDecodeError) as exc:
            return emit_error(
                "LISTENER_IDENTITY_UNVERIFIABLE",
                str(exc),
                as_json=bool(args.json),
            )
    payload = lease_port(
        project,
        change_id=args.change,
        run_id=args.run_id,
        port_range=parsed,
        generation=int(getattr(args, "generation", 1)),
        listener_identity=listener_identity,
    )
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "PORT_LEASE_FAILED")),
        str(payload.get("message", "port lease failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_release_port(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    listener_identity = None
    if getattr(args, "listener_identity_json", None):
        try:
            parsed_listener = json.loads(args.listener_identity_json)
            if not isinstance(parsed_listener, dict):
                raise ValueError("listener identity must be an object")
            listener_identity = parsed_listener
        except (TypeError, ValueError, json.JSONDecodeError) as exc:
            return emit_error(
                "LISTENER_IDENTITY_UNVERIFIABLE",
                str(exc),
                as_json=bool(args.json),
            )
    payload = release_port(
        project,
        change_id=args.change,
        run_id=args.run_id,
        port=getattr(args, "port", None),
        lease_id=getattr(args, "lease_id", None),
        generation=getattr(args, "generation", None),
        listener_identity=listener_identity,
    )
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "PORT_RELEASE_FAILED")),
        str(payload.get("message", "port release failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def cmd_integration_lock(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    if args.integration_action == "acquire":
        payload = integration_lock_acquire(
            project,
            run_id=args.run_id,
            ttl_seconds=int(args.ttl_seconds),
        )
    else:
        payload = integration_lock_release(project, run_id=args.run_id)
    if payload.get("ok"):
        emit(payload, as_json=bool(args.json))
        return 0
    return emit_error(
        str(payload.get("code", "INTEGRATION_LOCK_FAILED")),
        str(payload.get("message", "integration lock failed")),
        as_json=bool(args.json),
        extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
    )


def _ensure_change_identity(change_dir: Path) -> dict[str, str]:
    """Ensure meta/change-identity.json exists with a stable UUID (retro §5.5).

    Returns the identity dict with changeUuid and changeName.
    """
    identity_path = change_dir / "meta" / "change-identity.json"
    if identity_path.is_file():
        try:
            data = json.loads(identity_path.read_text(encoding="utf-8-sig"))
            if isinstance(data, dict) and data.get("changeUuid"):
                return data
        except (OSError, json.JSONDecodeError):
            pass
    # Generate new identity
    identity = {
        "schemaVersion": 1,
        "changeUuid": str(uuid.uuid4()),
        "changeName": change_dir.name,
        "createdAt": now_iso(),
    }
    identity_path.parent.mkdir(parents=True, exist_ok=True)
    identity_path.write_text(
        json.dumps(identity, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
        newline="\n",
    )
    return identity


def cmd_rename(args: argparse.Namespace) -> int:
    """Rename a change atomically: directory, pointers, worktree, identity (retro §5.5).

    Appends a change.rename event; does not rewrite history.
    """
    project = resolve_main_project_root()
    changes_root = project / ".harness" / "changes"
    old_dir = changes_root / args.change
    new_dir = changes_root / args.to

    if not old_dir.is_dir():
        return emit_error(
            "CHANGE_NOT_FOUND",
            f"change directory not found: {old_dir}",
            as_json=bool(args.json),
        )
    if new_dir.exists():
        return emit_error(
            "CHANGE_ALREADY_EXISTS",
            f"target change directory already exists: {new_dir}",
            as_json=bool(args.json),
        )

    # Ensure identity exists before rename
    identity = _ensure_change_identity(old_dir)
    old_uuid = identity.get("changeUuid", "")

    # Rename directory
    import shutil
    shutil.move(str(old_dir), str(new_dir))

    # Update knowledge-context.json.changeId if present
    kc_path = new_dir / "meta" / "knowledge-context.json"
    if kc_path.is_file():
        try:
            kc = json.loads(kc_path.read_text(encoding="utf-8-sig"))
            if isinstance(kc, dict):
                kc["changeId"] = args.to
                kc["changeUuid"] = old_uuid
                kc_path.write_text(
                    json.dumps(kc, ensure_ascii=False, indent=2) + "\n",
                    encoding="utf-8",
                    newline="\n",
                )
        except (OSError, json.JSONDecodeError):
            pass

    # Update worktree.json path/branch if present
    wt_path = new_dir / "meta" / "worktree.json"
    if wt_path.is_file():
        try:
            wt = json.loads(wt_path.read_text(encoding="utf-8-sig"))
            if isinstance(wt, dict):
                old_name = args.change
                new_name = args.to
                # Update path and branch if they contain old name
                if isinstance(wt.get("path"), str) and old_name in wt["path"]:
                    wt["path"] = wt["path"].replace(old_name, new_name)
                if isinstance(wt.get("branch"), str) and old_name in wt["branch"]:
                    wt["branch"] = wt["branch"].replace(old_name, new_name)
                wt_path.write_text(
                    json.dumps(wt, ensure_ascii=False, indent=2) + "\n",
                    encoding="utf-8",
                    newline="\n",
                )
        except (OSError, json.JSONDecodeError):
            pass

    # Update change-identity.json.changeName
    identity_path = new_dir / "meta" / "change-identity.json"
    if identity_path.is_file():
        try:
            ident = json.loads(identity_path.read_text(encoding="utf-8-sig"))
            if isinstance(ident, dict):
                ident["changeName"] = args.to
                ident["renamedFrom"] = args.change
                identity_path.write_text(
                    json.dumps(ident, ensure_ascii=False, indent=2) + "\n",
                    encoding="utf-8",
                    newline="\n",
                )
        except (OSError, json.JSONDecodeError):
            pass

    # Append change.rename event (does not rewrite history)
    try:
        import harness_events
        harness_events.append_event(
            new_dir,
            phase="plan",
            type_="change.rename",
            renamed_from=args.change,
            renamed_to=args.to,
            change_uuid=old_uuid,
            note=f"change renamed from {args.change} to {args.to}",
        )
    except Exception:
        pass  # event append failure should not block rename

    payload = {
        "ok": True,
        "code": "RENAMED",
        "changeUuid": old_uuid,
        "renamedFrom": args.change,
        "renamedTo": args.to,
        "changeDir": str(new_dir),
    }
    emit(payload, as_json=bool(args.json))
    return 0


def cmd_ensure_identity(args: argparse.Namespace) -> int:
    """Ensure meta/change-identity.json exists with a stable UUID (retro §5.5)."""
    project = resolve_main_project_root()
    change_dir = project / ".harness" / "changes" / args.change
    if not change_dir.is_dir():
        return emit_error(
            "CHANGE_NOT_FOUND",
            f"change directory not found: {change_dir}",
            as_json=bool(args.json),
        )
    identity = _ensure_change_identity(change_dir)
    emit(identity, as_json=bool(args.json))
    return 0


def cmd_status(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    change_id = str(getattr(args, "change", None) or "").strip()
    if change_id:
        payload = change_recovery_view(project, change_id)
        if payload.get("ok"):
            emit(payload, as_json=bool(args.json))
            return 0
        return emit_error(
            str(payload.get("code", "STATUS_FAILED")),
            str(payload.get("message", "status failed")),
            as_json=bool(args.json),
            extra={k: v for k, v in payload.items() if k not in {"ok", "message"}},
        )
    items = classify_changes(project)
    payload = {
        "ok": True,
        "code": "CHANGE_STATUS",
        "project": str(project),
        "items": items,
        "summary": {
            status: sum(1 for item in items if item["status"] == status)
            for status in (
                "ACTIVE",
                "ARCHIVED_LEFTOVER",
                "RECOVERABLE",
                "ORPHAN",
                "INVALID",
            )
        },
    }
    emit(payload, as_json=bool(args.json))
    return 0


# ---------------------------------------------------------------------------
# status --change <cn>：统一只读恢复视图（批次 2 WI-3）
#
# 故障路径此前要法证式读 gate-policy/events/ledger/state-snapshot 双层目录
# 才能拼出「现在在哪、下一步做什么」。本视图只读派生单一权威状态，不新建
# 任何可写状态（提案边界：不保留两套可写权威状态）。
# ---------------------------------------------------------------------------

def _status_read_json(path: Path) -> dict[str, Any] | None:
    if not path.is_file():
        return None
    try:
        data = _read_json(path)
    except (OSError, json.JSONDecodeError):
        return None
    return data if isinstance(data, dict) else None


def _status_read_events(change_dir: Path) -> list[dict[str, Any]]:
    """读事件（state 目录权威位置 + 契约目录 legacy 位置，去重保序）。"""
    import harness_events as he

    events_file = he.events_path(change_dir)
    events = list(he.load_events(events_file))
    legacy = change_dir / "events.ndjson"
    if legacy.is_file() and legacy.resolve() != events_file.resolve():
        seen = {id(e) for e in events}
        for item in he.load_events(legacy):
            if not any(
                item.get("id") == e.get("id") and item.get("id") for e in events
            ):
                events.append(item)
        del seen
    return events


def _status_open_phase(events: list[dict[str, Any]]) -> dict[str, Any] | None:
    """最近的未关门 phase.start（run_id 无对应 phase.end / phase.auto_sealed）。

    覆盖 task 阶段（轻任务）与 plan/execute/…（完整流程）——两者都是
    phase.start/phase.end 生命周期，只是阶段名不同。
    """
    started: list[dict[str, Any]] = []
    closed: set[str] = set()
    for event in events:
        if event.get("type") == "phase.start":
            started.append(event)
        elif event.get("type") in {"phase.end", "phase.auto_sealed"}:
            run_id = str(event.get("run_id") or "")
            if run_id:
                closed.add(run_id)
    for event in reversed(started):
        run_id = str(event.get("run_id") or "")
        if run_id and run_id in closed:
            continue
        return event
    return None


def _status_read_transitions(state_dir: Path) -> list[dict[str, Any]]:
    """读 context 转换收据（runtime/transitions.ndjson，只读）。

    v2 plan finalize 只写发布 journal + 转换收据、不写 phase.end 事件
    （harness_context._ensure_phase_end_event 文档化的结构性缺口），
    因此「阶段已关门」的证据 = phase.end 事件 ∪ fromPhase 收据。
    """
    transitions_file = state_dir / "runtime" / "transitions.ndjson"
    if not transitions_file.is_file():
        return []
    receipts: list[dict[str, Any]] = []
    try:
        text = transitions_file.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            item = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(item, dict):
            receipts.append(item)
    return receipts


def _status_phase_progress(
    events: list[dict[str, Any]],
    planned: list[str] | None,
    transitions: list[dict[str, Any]] | None = None,
    committed_journal: Path | None = None,
    current: str | None = None,
) -> dict[str, Any] | None:
    """plannedPhases 进度：已完成 / 当前 / 待办。

    完成证据取并集：phase.end 事件（gate close 或自动配对）、转换收据
    fromPhase（v2 plan finalize 路径只写收据）、committed 发布 journal
    （finalize 已提交但交接未补录的 T6 中断窗口）。current 由调用方传入
    权威阶段（_status_phase_identity 的推导），不从 open start 自行推导
    ——v2 plan start 永远 open，会压过收据/journal 的完成事实。
    """
    if not planned:
        return None
    ended: set[str] = set()
    for event in events:
        if event.get("type") != "phase.end":
            continue
        phase = harness_paths.resolve_phase_name(event.get("phase"))
        if phase:
            ended.add(phase)
    for receipt in transitions or []:
        phase = harness_paths.resolve_phase_name(receipt.get("fromPhase"))
        if phase:
            ended.add(phase)
    if committed_journal is not None:
        ended.add("plan")
    completed = [p for p in planned if p in ended and p != current]
    pending = [p for p in planned if p not in ended and p != current]
    return {
        "plannedPhases": list(planned),
        "completed": completed,
        "current": current,
        "pending": pending,
    }


def _status_ledger_verifications(change_dir: Path) -> dict[str, Any] | None:
    """各 verification kind 的最新记录与 status（ledger 只读）。"""
    import harness_ledger as hl

    try:
        ledger, path = hl.load_ledger(change_dir)
    except (OSError, ValueError, json.JSONDecodeError):
        return None
    if ledger is None:
        return None
    validations = ledger.get("validations")
    if not isinstance(validations, dict):
        return {"ledgerPath": str(path) if path else None, "verifications": {}}
    verifications: dict[str, Any] = {}
    for key, value in validations.items():
        if not isinstance(value, dict):
            continue
        # v3 结构：kind -> {attempts: [{status,...}]}；取最新 attempt
        attempts = value.get("attempts")
        if isinstance(attempts, list) and attempts:
            latest = attempts[-1]
            if isinstance(latest, dict):
                verifications[str(key)] = {
                    "status": latest.get("status"),
                    "recordedAt": latest.get("recordedAt")
                    or latest.get("timestamp"),
                }
                continue
        # legacy 结构：kind -> {status,...} 直接平铺
        if value.get("status") is not None:
            verifications[str(key)] = {
                "status": value.get("status"),
                "recordedAt": value.get("recordedAt") or value.get("timestamp"),
            }
    return {
        "ledgerPath": str(path) if path else None,
        "verifications": verifications,
    }


def _status_dirty_tree(project: Path) -> list[str]:
    """git status --porcelain 路径（重命名拆两侧，排除 .harness/**）。"""
    proc = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=project,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        check=False,
    )
    paths: list[str] = []
    for line in proc.stdout.splitlines():
        if not line.strip():
            continue
        raw = line[3:].strip().strip('"')
        if " -> " in raw:
            old, new = raw.split(" -> ", 1)
            paths.extend((old.strip().strip('"'), new.strip().strip('"')))
        else:
            paths.append(raw)
    return [
        p.replace("\\", "/") for p in paths if not p.replace("\\", "/").startswith(".harness/")
    ]


def _status_committed_journal(change_dir: Path) -> Path | None:
    """已 committed 的 v2 发布 journal——plan 完成的机器证据（只读）。

    与 harness_context._committed_publication_journal 同一判据；不 import
    是为了避免 harness_change ↔ harness_context 的模块级循环依赖
    （harness_context 顶部已 import harness_ledger，且在函数内延迟
    import harness_change）。
    """
    journal_dir = change_dir / "meta" / "publication-journals"
    if not journal_dir.is_dir():
        return None
    for path in sorted(journal_dir.glob("*.json")):
        payload = _status_read_json(path)
        if payload is not None and payload.get("state") == "committed":
            return path
    return None


def _status_phase_identity(
    *,
    events: list[dict[str, Any]],
    transitions: list[dict[str, Any]],
    committed_journal: Path | None,
    planned: list[str] | None,
    entry: str,
) -> dict[str, Any]:
    """推导当前阶段身份（phase/runId/attempt/handoffPending）。

    权威顺序与 context_view/gate close 的写入顺序一致：
    1. 有转换收据 → 最新收据 toPhase 即当前阶段（v2 plan finalize 不写
       plan 的 phase.end，plan start 永远 open，不能压过收据）；
       但 toPhase 阶段自身已有 phase.end 且没有以它为 fromPhase 的收据
       → gate close 在该阶段中断（end 已写、handoff 未落盘），
       handoffPending=true；
    2. 无收据但最新 phase.end 已写 → gate close 中断窗口（phase.end 先于
       handoff 落盘）：阶段仍按 end 的阶段，handoffPending=true；
    3. committed 发布 journal（plan 完成的机器证据）→ T6 式中断窗口：
       下一阶段按 plannedPhases 推导，等 bootstrap-execute/gate begin；
    4. 否则最近未关门 phase.start → 该阶段进行中。
    """
    phase = None
    run_id = None
    attempt = None
    handoff_pending = False

    open_phase = _status_open_phase(events)
    open_name = (
        harness_paths.resolve_phase_name(open_phase.get("phase")) if open_phase else None
    )

    if transitions:
        phase = harness_paths.resolve_phase_name(transitions[-1].get("toPhase"))
        # toPhase 阶段已关门（phase.end）但没有推进出它的收据 →
        # gate close 在 end 与 handoff 之间中断
        ended_phases = {
            harness_paths.resolve_phase_name(event.get("phase"))
            for event in events
            if event.get("type") == "phase.end"
        }
        advanced_from = {
            harness_paths.resolve_phase_name(receipt.get("fromPhase"))
            for receipt in transitions
        }
        if phase in ended_phases and phase not in advanced_from:
            handoff_pending = True
            return {
                "phase": phase,
                "runId": None,
                "attempt": None,
                "handoffPending": handoff_pending,
            }
        if open_phase is not None and open_name != phase:
            # 收据推进后的旧 start（典型：v2 plan）不再提供 run 身份
            run_id = None
            attempt = None
        elif open_phase is not None:
            run_id = str(open_phase.get("run_id") or "") or None
            raw_attempt = open_phase.get("attempt")
            attempt = (
                raw_attempt if isinstance(raw_attempt, int) and raw_attempt > 0 else None
            )
        return {
            "phase": phase,
            "runId": run_id,
            "attempt": attempt,
            "handoffPending": handoff_pending,
        }

    # 无收据：找最新 phase.end（gate close 先写 end 后写收据）
    latest_end = None
    for event in events:
        if event.get("type") == "phase.end":
            latest_end = event
    if latest_end is not None:
        phase = harness_paths.resolve_phase_name(latest_end.get("phase"))
        handoff_pending = True
        return {
            "phase": phase,
            "runId": None,
            "attempt": None,
            "handoffPending": handoff_pending,
        }

    if committed_journal is not None and entry == "full-flow":
        # plan finalize 已 committed、交接未补录：下一阶段等 begin
        after_plan = None
        if isinstance(planned, list):
            for index, name in enumerate(planned):
                if (
                    harness_paths.resolve_phase_name(name) == "plan"
                    and index + 1 < len(planned)
                ):
                    after_plan = harness_paths.resolve_phase_name(planned[index + 1])
                    break
        return {
            "phase": after_plan,
            "runId": None,
            "attempt": None,
            "handoffPending": False,
        }

    if open_phase is not None:
        phase = open_name or str(open_phase.get("phase"))
        run_id = str(open_phase.get("run_id") or "") or None
        raw_attempt = open_phase.get("attempt")
        attempt = (
            raw_attempt if isinstance(raw_attempt, int) and raw_attempt > 0 else None
        )
    return {"phase": phase, "runId": run_id, "attempt": attempt, "handoffPending": False}


def _status_next_action(
    *,
    entry: str,
    task: dict[str, Any] | None,
    identity: dict[str, Any],
    progress: dict[str, Any] | None,
    transitions: list[dict[str, Any]] | None = None,
) -> str:
    """按状态机推导下一步动作，复用各命令既有文案，不新造话术。"""
    phase = identity.get("phase")
    if entry == "light-task":
        status = str((task or {}).get("status") or "open")
        if status != "open":
            return "任务已终态；归档目录见 archiveDir 或 .harness/archive/"
        change = str((task or {}).get("changeId") or "")
        return (
            "继续编辑/测试，然后 harness_task.py finish --project . "
            f"--change {change} --json"
        )
    # 完整流程
    if identity.get("handoffPending"):
        # gate close 中断窗口：phase.end 已写、交接收据未落盘——
        # 复用 gate 既有 recoveryAction 语义（幂等续跑，不需重取租约）
        return (
            "本地关门已完成（phase.end 已写），交接未落盘：用原 close 命令"
            "补 --to-phase 重跑即幂等续跑（harness_gate.py close --phase "
            f"{phase} --change <cn> --status <OK|WARN> --to-phase <后继> --json）"
        )
    if phase is None:
        return (
            "无进行中阶段：新变更从 bootstrap-plan 开始"
            "（harness_context.py bootstrap-plan --project . --change <cn> "
            "--executor <tool> --json）"
        )
    if phase == "plan":
        return (
            "完成计划产物后运行 plan finalize（发布 journal committed 后 "
            "bootstrap-execute 会自动补录交接凭证）"
        )
    if phase == "execute":
        if identity.get("runId") is None:
            return (
                "阶段 execute 已交接但未开始：运行 "
                "harness_context.py bootstrap-execute --project . "
                "--change <cn> --executor <tool> --json"
            )
        return (
            "TDD 编码与验证；完成后运行 "
            f"harness_gate.py close --phase execute --status <OK|WARN> --json"
        )
    if phase == "review":
        return "完成评审产出后运行 harness_gate.py close --phase review --json"
    if phase == "submit":
        return "完成提交准备后运行 harness_gate.py close --phase submit --json"
    if identity.get("runId") is None and transitions:
        return f"阶段 {phase} 已交接但未开始：运行 gate begin 进入该阶段"
    pending = (progress or {}).get("pending") or []
    if pending:
        return f"进入下一阶段 {pending[0]}（bootstrap-execute 或 gate begin）"
    return "全部计划阶段已完成；归档入口 harness_archive.py archive --json"


def change_recovery_view(project_root: Path, change_id: str) -> dict[str, Any]:
    """单 change 统一只读恢复视图：轻任务与完整流程同一 resolver。"""
    resolved = resolve_change(project_root, change_id)
    if not resolved.get("ok"):
        return resolved
    change_dir = Path(resolved["changeDir"])
    state_dir = harness_paths.resolve_state_dir_for_contract(change_dir)

    # 入口代际：task.json 存在 = 轻任务；否则完整流程
    task = _status_read_json(change_dir / "meta" / "task.json")
    entry = "light-task" if task is not None else "full-flow"

    events = _status_read_events(change_dir)
    transitions = _status_read_transitions(state_dir)

    # 档位及来源：轻任务区分声明 floor 与 finish 裁决；完整流程读 gate-policy
    tier: str | None = None
    tier_source: str | None = None
    declared_tier: str | None = None
    if entry == "light-task":
        declared_tier = task.get("declaredTier") if isinstance(task, dict) else None
        tier = task.get("tier") if isinstance(task, dict) else None
        tier_source = (
            "finish-adjudicated" if tier else None
        )
    else:
        policy_loaded = harness_paths.load_change_gate_policy(change_dir)
        policy = policy_loaded.get("policy") if isinstance(policy_loaded, dict) else None
        if isinstance(policy, dict):
            tier = policy.get("tier")
            tier_source = str(policy_loaded.get("source") or "gate-policy-json")

    # plannedPhases（完整流程；轻任务无阶段计划）
    committed_journal = _status_committed_journal(change_dir)
    planned = None
    if entry == "full-flow":
        policy_loaded = harness_paths.load_change_gate_policy(change_dir)
        policy = policy_loaded.get("policy") if isinstance(policy_loaded, dict) else None
        planned = policy.get("plannedPhases") if isinstance(policy, dict) else None
        if not planned and isinstance(policy, dict):
            planned = policy.get("defaultPhases")

    identity = _status_phase_identity(
        events=events,
        transitions=transitions,
        committed_journal=committed_journal,
        planned=planned,
        entry=entry,
    )

    progress = None
    if entry == "full-flow":
        progress = _status_phase_progress(
            events,
            planned,
            transitions,
            committed_journal,
            current=identity.get("phase"),
        )

    ledger_view = _status_ledger_verifications(change_dir)
    dirty = _status_dirty_tree(project_root)

    # 外来脏路径：仅轻任务有 begin 基线可比（full-flow 的 foreign 判定在
    # classify/archive 侧，视图不重复推导，避免第二套语义）
    foreign_paths: list[str] = []
    if entry == "light-task" and isinstance(task, dict):
        baseline = task.get("dirtyBaseline")
        if isinstance(baseline, dict):
            import harness_task as ht

            foreign_paths = ht.detect_foreign_dirt(project_root, baseline)

    lease_state = inspect_lease_state(project_root, change_id)

    next_action = _status_next_action(
        entry=entry,
        task=task,
        identity=identity,
        progress=progress,
        transitions=transitions,
    )

    return {
        "ok": True,
        "code": "CHANGE_RECOVERY_VIEW",
        "changeId": resolved["changeId"],
        "changeDir": str(change_dir),
        "stateDir": str(state_dir),
        "entryGeneration": entry,
        "currentPhase": identity.get("phase"),
        "runId": identity.get("runId"),
        "attempt": identity.get("attempt"),
        "handoffPending": identity.get("handoffPending"),
        "tier": tier,
        "tierSource": tier_source,
        "declaredTier": declared_tier,
        "phaseProgress": progress,
        "verifications": (ledger_view or {}).get("verifications"),
        "ledgerPath": (ledger_view or {}).get("ledgerPath"),
        "uncommittedPaths": dirty,
        "foreignPaths": foreign_paths,
        "lease": {
            "state": lease_state.get("state"),
            "phase": (lease_state.get("lease") or {}).get("phase")
            if isinstance(lease_state.get("lease"), dict)
            else None,
            "runId": (lease_state.get("lease") or {}).get("runId")
            if isinstance(lease_state.get("lease"), dict)
            else None,
            "expiresAt": (lease_state.get("lease") or {}).get("expiresAt")
            if isinstance(lease_state.get("lease"), dict)
            else None,
        },
        "taskStatus": (task or {}).get("status") if isinstance(task, dict) else None,
        "nextAction": next_action,
    }


def cmd_cleanup_changes(args: argparse.Namespace) -> int:
    project = resolve_main_project_root()
    result = cleanup_changes(project, apply=bool(args.apply))
    emit(result, as_json=bool(args.json))
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="harness_change.py")
    parser.add_argument("--json", action="store_true")
    sub = parser.add_subparsers(dest="command_name", required=True)

    shared = argparse.ArgumentParser(add_help=False)
    shared.add_argument("--json", action="store_true", default=argparse.SUPPRESS)

    p_list = sub.add_parser("list", parents=[shared])
    p_list.set_defaults(func=cmd_list)

    p_status = sub.add_parser("status", parents=[shared])
    p_status.add_argument("--all", action="store_true", dest="show_all")
    p_status.add_argument(
        "--change",
        default=None,
        help="单 change 只读恢复视图（轻任务/完整流程统一）；缺省列出全部 change 分类",
    )
    p_status.set_defaults(func=cmd_status)

    p_cleanup = sub.add_parser("cleanup", parents=[shared])
    cleanup_mode = p_cleanup.add_mutually_exclusive_group()
    cleanup_mode.add_argument("--dry-run", action="store_true")
    cleanup_mode.add_argument("--apply", action="store_true")
    p_cleanup.set_defaults(func=cmd_cleanup_changes)

    p_resolve = sub.add_parser("resolve", parents=[shared])
    p_resolve.add_argument("--change", default=None)
    p_resolve.set_defaults(func=cmd_resolve)

    p_migrate = sub.add_parser("migrate", parents=[shared])
    p_migrate.add_argument("--change", required=True)
    p_migrate.set_defaults(func=cmd_migrate)

    p_ownership = sub.add_parser(
        "declare-ownership", parents=[shared],
        help="declare ownership.productPaths so archive can project the real diff",
    )
    p_ownership.add_argument("--change", required=True)
    p_ownership.add_argument(
        "--product-path", required=True, action="append",
        help="exact file or directory prefix (no globs); repeatable",
    )
    p_ownership.set_defaults(func=cmd_declare_ownership)

    p_local_release = sub.add_parser(
        "allow-local-release", parents=[shared],
        help="set candidateVerification.allowLocalRelease=true in gate-policy.json",
    )
    p_local_release.add_argument("--change", required=True)
    p_local_release.set_defaults(func=cmd_allow_local_release)

    p_claim = sub.add_parser("claim", parents=[shared])
    p_claim.add_argument("--change", required=True)
    p_claim.add_argument("--phase", required=True)
    p_claim.add_argument("--run-id", required=True)
    p_claim.add_argument("--ttl-seconds", type=int, default=3600)
    p_claim.add_argument("--steal", action="store_true")
    p_claim.add_argument("--expected-generation", type=int, default=None)
    p_claim.set_defaults(func=cmd_claim)

    p_release = sub.add_parser("release", parents=[shared])
    p_release.add_argument("--change", required=True)
    p_release.add_argument("--phase", required=True)
    p_release.add_argument("--run-id", required=True)
    p_release.add_argument("--lease-id", default=None)
    p_release.add_argument("--generation", type=int, default=None)
    p_release.set_defaults(func=cmd_release)

    p_port = sub.add_parser("lease-port", parents=[shared])
    p_port.add_argument("--change", required=True)
    p_port.add_argument("--run-id", required=True)
    p_port.add_argument("--range", required=True)
    p_port.add_argument("--generation", type=int, default=1)
    p_port.add_argument("--listener-identity-json", default=None)
    p_port.set_defaults(func=cmd_lease_port)

    p_port_release = sub.add_parser("release-port", parents=[shared])
    p_port_release.add_argument("--change", required=True)
    p_port_release.add_argument("--run-id", required=True)
    p_port_release.add_argument("--port", type=int, default=None)
    p_port_release.add_argument("--lease-id", default=None)
    p_port_release.add_argument("--generation", type=int, default=None)
    p_port_release.add_argument("--listener-identity-json", default=None)
    p_port_release.set_defaults(func=cmd_release_port)

    p_lock = sub.add_parser("integration-lock", parents=[shared])
    p_lock.add_argument("integration_action", choices=["acquire", "release"])
    p_lock.add_argument("--run-id", required=True)
    p_lock.add_argument("--ttl-seconds", type=int, default=3600)
    p_lock.set_defaults(func=cmd_integration_lock)

    p_rename = sub.add_parser("rename", parents=[shared])
    p_rename.add_argument("--change", required=True)
    p_rename.add_argument("--to", required=True)
    p_rename.set_defaults(func=cmd_rename)

    p_identity = sub.add_parser("ensure-identity", parents=[shared])
    p_identity.add_argument("--change", required=True)
    p_identity.set_defaults(func=cmd_ensure_identity)

    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    return int(args.func(args))


if __name__ == "__main__":
    raise SystemExit(main())