Harness

"""Shared utilities for harness hooks.

Consolidates duplicated logic: payload reading, state root discovery,
JSON I/O, lock primitives, task eligibility, and ISO time helpers.

Ported from Codex harness hooks to Claude Code.
"""
from __future__ import annotations

import datetime as _dt
import hashlib
import json
import os
import shutil
import sys
import time
from pathlib import Path
from typing import Any, Optional

# ---------------------------------------------------------------------------
# Time helpers
# ---------------------------------------------------------------------------

def utc_now() -> _dt.datetime:
    return _dt.datetime.now(tz=_dt.timezone.utc)


def iso_z(dt: _dt.datetime) -> str:
    dt = dt.astimezone(_dt.timezone.utc).replace(microsecond=0)
    return dt.isoformat().replace("+00:00", "Z")


def parse_iso(ts: Any) -> Optional[_dt.datetime]:
    if not isinstance(ts, str) or not ts.strip():
        return None
    s = ts.strip()
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    try:
        dt = _dt.datetime.fromisoformat(s)
    except Exception:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=_dt.timezone.utc)
    return dt.astimezone(_dt.timezone.utc)


# ---------------------------------------------------------------------------
# Hook payload
# ---------------------------------------------------------------------------

def read_hook_payload() -> dict[str, Any]:
    """Read JSON payload from stdin (sent by Claude Code to command hooks)."""
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def maybe_log_hook_event(root: Path, payload: dict[str, Any], hook_script: str) -> None:
    """Optionally append a compact hook execution record to HARNESS_HOOK_LOG.

    This is opt-in debugging: when HARNESS_HOOK_LOG is unset, it is a no-op.
    Call this only after the .harness-active guard passes.
    """
    log_path = os.environ.get("HARNESS_HOOK_LOG")
    if not log_path:
        return

    entry: dict[str, Any] = {
        "ts": iso_z(utc_now()),
        "hook_script": hook_script,
        "hook_event_name": payload.get("hook_event_name"),
        "harness_root": str(root),
    }
    for k in (
        "session_id",
        "cwd",
        "source",
        "reason",
        "teammate_name",
        "team_name",
        "agent_id",
        "agent_type",
        "stop_hook_active",
    ):
        if k in payload:
            entry[k] = payload.get(k)

    try:
        Path(log_path).expanduser().parent.mkdir(parents=True, exist_ok=True)
        with Path(log_path).expanduser().open("a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    except Exception:
        return


# ---------------------------------------------------------------------------
# State root discovery
# ---------------------------------------------------------------------------

def find_harness_root(payload: dict[str, Any]) -> Optional[Path]:
    """Locate the directory containing harness-tasks.json.

    Search order:
    1. HARNESS_STATE_ROOT env var
    2. CLAUDE_PROJECT_DIR env var (+ parents)
    3. payload["cwd"] / os.getcwd() (+ parents)
    """
    env_root = os.environ.get("HARNESS_STATE_ROOT")
    if env_root:
        p = Path(env_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p

    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))
    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))

    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def is_harness_active(root: Path) -> bool:
    """True when .harness-active marker exists (hooks are live)."""
    return (root / ".harness-active").is_file()


# ---------------------------------------------------------------------------
# JSON I/O
# ---------------------------------------------------------------------------

def load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError(f"{path.name} must be a JSON object")
    return data


def atomic_write_json(path: Path, data: dict[str, Any]) -> None:
    """Write JSON atomically: backup -> tmp -> rename."""
    bak = path.with_name(f"{path.name}.bak")
    tmp = path.with_name(f"{path.name}.tmp")
    shutil.copy2(path, bak)
    tmp.write_text(
        json.dumps(data, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    os.replace(tmp, path)


def tail_text(path: Path, max_bytes: int = 200_000) -> str:
    """Read the last max_bytes of a text file."""
    with path.open("rb") as f:
        try:
            f.seek(0, os.SEEK_END)
            size = f.tell()
            f.seek(max(0, size - max_bytes), os.SEEK_SET)
        except Exception:
            f.seek(0, os.SEEK_SET)
        chunk = f.read()
    return chunk.decode("utf-8", errors="replace")


# ---------------------------------------------------------------------------
# Lock primitives (mkdir-based, POSIX-portable)
# ---------------------------------------------------------------------------

def lockdir_for_root(root: Path) -> Path:
    h = hashlib.sha256(str(root).encode("utf-8")).hexdigest()[:16]
    return Path("/tmp") / f"harness-{h}.lock"


def _pid_alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except Exception:
        return False


def _read_pid(lockdir: Path) -> Optional[int]:
    try:
        raw = (lockdir / "pid").read_text("utf-8").strip()
        return int(raw) if raw else None
    except Exception:
        return None


def acquire_lock(lockdir: Path, timeout_seconds: float = 5.0) -> None:
    deadline = time.time() + timeout_seconds
    missing_pid_since: Optional[float] = None
    while True:
        try:
            lockdir.mkdir(mode=0o700)
            (lockdir / "pid").write_text(str(os.getpid()), encoding="utf-8")
            return
        except FileExistsError:
            pid = _read_pid(lockdir)
            if pid is None:
                if missing_pid_since is None:
                    missing_pid_since = time.time()
                if time.time() - missing_pid_since < 1.0:
                    if time.time() >= deadline:
                        raise TimeoutError("lock busy (pid missing)")
                    time.sleep(0.05)
                    continue
            else:
                missing_pid_since = None
                if _pid_alive(pid):
                    if time.time() >= deadline:
                        raise TimeoutError(f"lock busy (pid={pid})")
                    time.sleep(0.05)
                    continue

            stale = lockdir.with_name(
                f"{lockdir.name}.stale.{os.getpid()}.{int(time.time())}"
            )
            try:
                lockdir.rename(stale)
            except Exception:
                if time.time() >= deadline:
                    raise TimeoutError("lock contention")
                time.sleep(0.05)
                continue
            shutil.rmtree(stale, ignore_errors=True)
            missing_pid_since = None
            continue


def release_lock(lockdir: Path) -> None:
    shutil.rmtree(lockdir, ignore_errors=True)


# ---------------------------------------------------------------------------
# Task helpers
# ---------------------------------------------------------------------------

def priority_rank(v: Any) -> int:
    return {"P0": 0, "P1": 1, "P2": 2}.get(str(v or ""), 9)


def task_attempts(t: dict[str, Any]) -> int:
    try:
        return int(t.get("attempts") or 0)
    except Exception:
        return 0


def task_max_attempts(t: dict[str, Any]) -> int:
    try:
        v = t.get("max_attempts")
        return int(v) if v is not None else 3
    except Exception:
        return 3


def deps_completed(t: dict[str, Any], completed_ids: set[str]) -> bool:
    deps = t.get("depends_on") or []
    if not isinstance(deps, list):
        return False
    return all(str(d) in completed_ids for d in deps)


def parse_tasks(state: dict[str, Any]) -> list[dict[str, Any]]:
    """Extract validated task list from state dict."""
    tasks_raw = state.get("tasks") or []
    if not isinstance(tasks_raw, list):
        raise ValueError("tasks must be a list")
    return [t for t in tasks_raw if isinstance(t, dict)]


def completed_ids(tasks: list[dict[str, Any]]) -> set[str]:
    return {
        str(t.get("id", ""))
        for t in tasks
        if str(t.get("status", "")) == "completed"
    }


def eligible_tasks(
    tasks: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """Return (pending_eligible, retryable) sorted by priority then id."""
    done = completed_ids(tasks)

    pending = [
        t for t in tasks
        if str(t.get("status", "")) == "pending" and deps_completed(t, done)
    ]
    retry = [
        t for t in tasks
        if str(t.get("status", "")) == "failed"
        and task_attempts(t) < task_max_attempts(t)
        and deps_completed(t, done)
    ]

    def key(t: dict[str, Any]) -> tuple[int, str]:
        return (priority_rank(t.get("priority")), str(t.get("id", "")))

    pending.sort(key=key)
    retry.sort(key=key)
    return pending, retry


def pick_next(
    pending: list[dict[str, Any]], retry: list[dict[str, Any]]
) -> Optional[dict[str, Any]]:
    return pending[0] if pending else (retry[0] if retry else None)


def status_counts(tasks: list[dict[str, Any]]) -> dict[str, int]:
    counts: dict[str, int] = {}
    for t in tasks:
        s = str(t.get("status") or "pending")
        counts[s] = counts.get(s, 0) + 1
    return counts


def reap_stale_leases(
    tasks: list[dict[str, Any]], now: _dt.datetime
) -> bool:
    """Reset in_progress tasks with expired leases to failed. Returns True if any changed."""
    changed = False
    for t in tasks:
        if str(t.get("status", "")) != "in_progress":
            continue
        exp = parse_iso(t.get("lease_expires_at"))
        if exp is None or exp > now:
            continue

        t["attempts"] = task_attempts(t) + 1
        err = f"[SESSION_TIMEOUT] Lease expired (claimed_by={t.get('claimed_by')})"
        log = t.get("error_log")
        if isinstance(log, list):
            log.append(err)
        else:
            t["error_log"] = [err]

        t["status"] = "failed"
        t.pop("claimed_by", None)
        t.pop("lease_expires_at", None)
        t.pop("claimed_at", None)
        changed = True
    return changed


# ---------------------------------------------------------------------------
# Session config helpers
# ---------------------------------------------------------------------------

def get_session_config(state: dict[str, Any]) -> dict[str, Any]:
    cfg = state.get("session_config") or {}
    return cfg if isinstance(cfg, dict) else {}


def is_concurrent(cfg: dict[str, Any]) -> bool:
    return str(cfg.get("concurrency_mode") or "exclusive") == "concurrent"


# ---------------------------------------------------------------------------
# Hook output helpers
# ---------------------------------------------------------------------------

def emit_block(reason: str) -> None:
    """Print a JSON block decision to stdout and exit 0."""
    print(json.dumps({"decision": "block", "reason": reason}, ensure_ascii=False))


def emit_allow(reason: str = "") -> None:
    """Print a JSON allow decision to stdout and exit 0."""
    out: dict[str, Any] = {"decision": "allow"}
    if reason:
        out["reason"] = reason
    print(json.dumps(out, ensure_ascii=False))


def emit_context(context: str) -> None:
    """Inject additional context via hookSpecificOutput."""
    print(json.dumps(
        {"hookSpecificOutput": {"additionalContext": context}},
        ensure_ascii=False,
    ))


def emit_json(data: dict[str, Any]) -> None:
    """Print arbitrary JSON to stdout."""
    print(json.dumps(data, ensure_ascii=False))

#!/usr/bin/env python3
from __future__ import annotations

import datetime as _dt
import hashlib
import json
import os
import shutil
import socket
import sys
import time
from pathlib import Path
from typing import Any, Optional


def _utc_now() -> _dt.datetime:
    return _dt.datetime.now(tz=_dt.timezone.utc)


def _iso_z(dt: _dt.datetime) -> str:
    dt = dt.astimezone(_dt.timezone.utc).replace(microsecond=0)
    return dt.isoformat().replace("+00:00", "Z")


def _parse_iso(ts: Any) -> Optional[_dt.datetime]:
    if not isinstance(ts, str) or not ts.strip():
        return None
    s = ts.strip()
    if s.endswith("Z"):
        s = s[:-1] + "+00:00"
    try:
        dt = _dt.datetime.fromisoformat(s)
    except Exception:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=_dt.timezone.utc)
    return dt.astimezone(_dt.timezone.utc)


def _read_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _find_state_root(payload: dict[str, Any]) -> Optional[Path]:
    state_root = os.environ.get("HARNESS_STATE_ROOT")
    if state_root:
        p = Path(state_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p

    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))

    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))

    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def _lockdir_for_root(root: Path) -> Path:
    h = hashlib.sha256(str(root).encode("utf-8")).hexdigest()[:16]
    return Path("/tmp") / f"harness-{h}.lock"


def _pid_alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except Exception:
        return False


def _read_pid(lockdir: Path) -> Optional[int]:
    try:
        raw = (lockdir / "pid").read_text("utf-8").strip()
        return int(raw) if raw else None
    except Exception:
        return None


def _acquire_lock(lockdir: Path, timeout_seconds: float) -> None:
    deadline = time.time() + timeout_seconds
    missing_pid_since: Optional[float] = None
    while True:
        try:
            lockdir.mkdir(mode=0o700)
            (lockdir / "pid").write_text(str(os.getpid()), encoding="utf-8")
            return
        except FileExistsError:
            pid = _read_pid(lockdir)
            if pid is None:
                if missing_pid_since is None:
                    missing_pid_since = time.time()
                if time.time() - missing_pid_since < 1.0:
                    if time.time() >= deadline:
                        raise TimeoutError("lock busy (pid missing)")
                    time.sleep(0.05)
                    continue
            else:
                missing_pid_since = None
                if _pid_alive(pid):
                    if time.time() >= deadline:
                        raise TimeoutError(f"lock busy (pid={pid})")
                    time.sleep(0.05)
                    continue

            stale = lockdir.with_name(f"{lockdir.name}.stale.{os.getpid()}.{int(time.time())}")
            try:
                lockdir.rename(stale)
            except Exception:
                if time.time() >= deadline:
                    raise TimeoutError("lock contention")
                time.sleep(0.05)
                continue
            shutil.rmtree(stale, ignore_errors=True)
            missing_pid_since = None
            continue


def _release_lock(lockdir: Path) -> None:
    shutil.rmtree(lockdir, ignore_errors=True)


def _load_state(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError("harness-tasks.json must be an object")
    return data


def _atomic_write_json(path: Path, data: dict[str, Any]) -> None:
    bak = path.with_name(f"{path.name}.bak")
    tmp = path.with_name(f"{path.name}.tmp")
    shutil.copy2(path, bak)
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    os.replace(tmp, path)


def _priority_rank(v: Any) -> int:
    return {"P0": 0, "P1": 1, "P2": 2}.get(str(v or ""), 9)


def _eligible_tasks(tasks: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    completed = {str(t.get("id", "")) for t in tasks if str(t.get("status", "")) == "completed"}

    def deps_ok(t: dict[str, Any]) -> bool:
        deps = t.get("depends_on") or []
        if not isinstance(deps, list):
            return False
        return all(str(d) in completed for d in deps)

    def attempts(t: dict[str, Any]) -> int:
        try:
            return int(t.get("attempts") or 0)
        except Exception:
            return 0

    def max_attempts(t: dict[str, Any]) -> int:
        try:
            v = t.get("max_attempts")
            return int(v) if v is not None else 3
        except Exception:
            return 3

    pending = [t for t in tasks if str(t.get("status", "")) == "pending" and deps_ok(t)]
    retry = [
        t
        for t in tasks
        if str(t.get("status", "")) == "failed"
        and attempts(t) < max_attempts(t)
        and deps_ok(t)
    ]

    def key(t: dict[str, Any]) -> tuple[int, str]:
        return (_priority_rank(t.get("priority")), str(t.get("id", "")))

    pending.sort(key=key)
    retry.sort(key=key)
    return pending, retry


def _reap_stale_leases(tasks: list[dict[str, Any]], now: _dt.datetime) -> bool:
    changed = False
    for t in tasks:
        if str(t.get("status", "")) != "in_progress":
            continue
        exp = _parse_iso(t.get("lease_expires_at"))
        if exp is None or exp > now:
            continue

        try:
            t["attempts"] = int(t.get("attempts") or 0) + 1
        except Exception:
            t["attempts"] = 1

        err = f"[SESSION_TIMEOUT] Lease expired (claimed_by={t.get('claimed_by')})"
        log = t.get("error_log")
        if isinstance(log, list):
            log.append(err)
        else:
            t["error_log"] = [err]

        t["status"] = "failed"
        t.pop("claimed_by", None)
        t.pop("lease_expires_at", None)
        t.pop("claimed_at", None)
        changed = True
    return changed


def main() -> int:
    payload = _read_payload()
    root = _find_state_root(payload)
    if root is None:
        print(json.dumps({"claimed": False, "error": "state root not found"}, ensure_ascii=False))
        return 0

    tasks_path = root / "harness-tasks.json"
    lockdir = _lockdir_for_root(root)

    timeout_s = float(os.environ.get("HARNESS_LOCK_TIMEOUT_SECONDS") or "5")
    _acquire_lock(lockdir, timeout_s)
    try:
        state = _load_state(tasks_path)
        session_config = state.get("session_config") or {}
        if not isinstance(session_config, dict):
            session_config = {}
        concurrency_mode = str(session_config.get("concurrency_mode") or "exclusive")
        is_concurrent = concurrency_mode == "concurrent"
        tasks_raw = state.get("tasks") or []
        if not isinstance(tasks_raw, list):
            raise ValueError("tasks must be a list")
        tasks = [t for t in tasks_raw if isinstance(t, dict)]

        now = _utc_now()
        if _reap_stale_leases(tasks, now):
            state["tasks"] = tasks
            _atomic_write_json(tasks_path, state)

        pending, retry = _eligible_tasks(tasks)
        task = pending[0] if pending else (retry[0] if retry else None)
        if task is None:
            print(json.dumps({"claimed": False}, ensure_ascii=False))
            return 0

        worker_id = os.environ.get("HARNESS_WORKER_ID") or ""
        if is_concurrent and not worker_id:
            print(json.dumps({"claimed": False, "error": "missing HARNESS_WORKER_ID"}, ensure_ascii=False))
            return 0
        if not worker_id:
            worker_id = f"{socket.gethostname()}:{os.getpid()}"
        lease_seconds = int(os.environ.get("HARNESS_LEASE_SECONDS") or "1800")
        exp = now + _dt.timedelta(seconds=lease_seconds)

        task["status"] = "in_progress"
        task["claimed_by"] = worker_id
        task["claimed_at"] = _iso_z(now)
        task["lease_expires_at"] = _iso_z(exp)
        state["tasks"] = tasks
        _atomic_write_json(tasks_path, state)

        out = {
            "claimed": True,
            "worker_id": worker_id,
            "task_id": str(task.get("id") or ""),
            "title": str(task.get("title") or ""),
            "lease_expires_at": task["lease_expires_at"],
        }
        print(json.dumps(out, ensure_ascii=False))
        return 0
    finally:
        _release_lock(lockdir)


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

#!/usr/bin/env python3
from __future__ import annotations

import datetime as _dt
import hashlib
import json
import os
import shutil
import sys
import time
from pathlib import Path
from typing import Any, Optional


def _utc_now() -> _dt.datetime:
    return _dt.datetime.now(tz=_dt.timezone.utc)


def _iso_z(dt: _dt.datetime) -> str:
    dt = dt.astimezone(_dt.timezone.utc).replace(microsecond=0)
    return dt.isoformat().replace("+00:00", "Z")


def _read_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _find_state_root(payload: dict[str, Any]) -> Optional[Path]:
    state_root = os.environ.get("HARNESS_STATE_ROOT")
    if state_root:
        p = Path(state_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p

    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))

    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))

    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def _lockdir_for_root(root: Path) -> Path:
    h = hashlib.sha256(str(root).encode("utf-8")).hexdigest()[:16]
    return Path("/tmp") / f"harness-{h}.lock"


def _pid_alive(pid: int) -> bool:
    try:
        os.kill(pid, 0)
        return True
    except Exception:
        return False


def _read_pid(lockdir: Path) -> Optional[int]:
    try:
        raw = (lockdir / "pid").read_text("utf-8").strip()
        return int(raw) if raw else None
    except Exception:
        return None


def _acquire_lock(lockdir: Path, timeout_seconds: float) -> None:
    deadline = time.time() + timeout_seconds
    missing_pid_since: Optional[float] = None
    while True:
        try:
            lockdir.mkdir(mode=0o700)
            (lockdir / "pid").write_text(str(os.getpid()), encoding="utf-8")
            return
        except FileExistsError:
            pid = _read_pid(lockdir)
            if pid is None:
                if missing_pid_since is None:
                    missing_pid_since = time.time()
                if time.time() - missing_pid_since < 1.0:
                    if time.time() >= deadline:
                        raise TimeoutError("lock busy (pid missing)")
                    time.sleep(0.05)
                    continue
            else:
                missing_pid_since = None
                if _pid_alive(pid):
                    if time.time() >= deadline:
                        raise TimeoutError(f"lock busy (pid={pid})")
                    time.sleep(0.05)
                    continue

            stale = lockdir.with_name(f"{lockdir.name}.stale.{os.getpid()}.{int(time.time())}")
            try:
                lockdir.rename(stale)
            except Exception:
                if time.time() >= deadline:
                    raise TimeoutError("lock contention")
                time.sleep(0.05)
                continue
            shutil.rmtree(stale, ignore_errors=True)
            missing_pid_since = None
            continue


def _release_lock(lockdir: Path) -> None:
    shutil.rmtree(lockdir, ignore_errors=True)


def _load_state(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError("harness-tasks.json must be an object")
    return data


def _atomic_write_json(path: Path, data: dict[str, Any]) -> None:
    bak = path.with_name(f"{path.name}.bak")
    tmp = path.with_name(f"{path.name}.tmp")
    shutil.copy2(path, bak)
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    os.replace(tmp, path)


def main() -> int:
    payload = _read_payload()
    root = _find_state_root(payload)
    if root is None:
        print(json.dumps({"renewed": False, "error": "state root not found"}, ensure_ascii=False))
        return 0

    task_id = os.environ.get("HARNESS_TASK_ID") or str(payload.get("task_id") or "").strip()
    if not task_id:
        print(json.dumps({"renewed": False, "error": "missing task_id"}, ensure_ascii=False))
        return 0

    worker_id = os.environ.get("HARNESS_WORKER_ID") or ""
    if not worker_id:
        print(json.dumps({"renewed": False, "error": "missing HARNESS_WORKER_ID"}, ensure_ascii=False))
        return 0
    lease_seconds = int(os.environ.get("HARNESS_LEASE_SECONDS") or "1800")

    tasks_path = root / "harness-tasks.json"
    lockdir = _lockdir_for_root(root)

    timeout_s = float(os.environ.get("HARNESS_LOCK_TIMEOUT_SECONDS") or "5")
    try:
        _acquire_lock(lockdir, timeout_s)
    except Exception as e:
        print(json.dumps({"renewed": False, "error": str(e)}, ensure_ascii=False))
        return 0

    try:
        state = _load_state(tasks_path)
        tasks_raw = state.get("tasks") or []
        if not isinstance(tasks_raw, list):
            raise ValueError("tasks must be a list")
        tasks = [t for t in tasks_raw if isinstance(t, dict)]

        task = next((t for t in tasks if str(t.get("id") or "") == task_id), None)
        if task is None:
            print(json.dumps({"renewed": False, "error": "task not found"}, ensure_ascii=False))
            return 0

        if str(task.get("status") or "") != "in_progress":
            print(json.dumps({"renewed": False, "error": "task not in_progress"}, ensure_ascii=False))
            return 0

        claimed_by = str(task.get("claimed_by") or "")
        if claimed_by and claimed_by != worker_id:
            print(json.dumps({"renewed": False, "error": "task owned by other worker"}, ensure_ascii=False))
            return 0

        now = _utc_now()
        exp = now + _dt.timedelta(seconds=lease_seconds)
        task["lease_expires_at"] = _iso_z(exp)
        task["claimed_by"] = worker_id
        state["tasks"] = tasks
        _atomic_write_json(tasks_path, state)

        print(json.dumps({"renewed": True, "task_id": task_id, "lease_expires_at": task["lease_expires_at"]}, ensure_ascii=False))
        return 0
    except Exception as e:
        print(json.dumps({"renewed": False, "error": str(e)}, ensure_ascii=False))
        return 0
    finally:
        _release_lock(lockdir)


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

#!/usr/bin/env python3
from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any, Optional


def _read_hook_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {"_invalid_json": True}


def _find_harness_root(payload: dict[str, Any]) -> Optional[Path]:
    state_root = os.environ.get("HARNESS_STATE_ROOT")
    if state_root:
        p = Path(state_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p

    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))

    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))

    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def _load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError(f"{path.name} must be a JSON object")
    return data


def _tail_text(path: Path, max_bytes: int = 8192) -> str:
    with path.open("rb") as f:
        try:
            f.seek(0, os.SEEK_END)
            size = f.tell()
            f.seek(max(0, size - max_bytes), os.SEEK_SET)
        except Exception:
            f.seek(0, os.SEEK_SET)
        chunk = f.read()
    return chunk.decode("utf-8", errors="replace")


def _priority_rank(v: Any) -> int:
    return {"P0": 0, "P1": 1, "P2": 2}.get(str(v or ""), 9)


def _pick_next_eligible(tasks: list[dict[str, Any]]) -> Optional[dict[str, Any]]:
    completed = {str(t.get("id", "")) for t in tasks if str(t.get("status", "")) == "completed"}

    def deps_ok(t: dict[str, Any]) -> bool:
        deps = t.get("depends_on") or []
        if not isinstance(deps, list):
            return False
        return all(str(d) in completed for d in deps)

    def attempts(t: dict[str, Any]) -> int:
        try:
            return int(t.get("attempts") or 0)
        except Exception:
            return 0

    def max_attempts(t: dict[str, Any]) -> int:
        try:
            v = t.get("max_attempts")
            return int(v) if v is not None else 3
        except Exception:
            return 3

    pending = [t for t in tasks if str(t.get("status", "")) == "pending" and deps_ok(t)]
    retry = [
        t
        for t in tasks
        if str(t.get("status", "")) == "failed"
        and attempts(t) < max_attempts(t)
        and deps_ok(t)
    ]

    def key(t: dict[str, Any]) -> tuple[int, str]:
        return (_priority_rank(t.get("priority")), str(t.get("id", "")))

    pending.sort(key=key)
    retry.sort(key=key)
    return pending[0] if pending else (retry[0] if retry else None)


def _is_harness_active(root: Path) -> bool:
    """Check if harness skill is actively running (marker file exists)."""
    return (root / ".harness-active").is_file()


def main() -> int:
    payload = _read_hook_payload()
    root = _find_harness_root(payload)
    if root is None:
        return 0

    # Guard: only active when harness skill is triggered
    if not _is_harness_active(root):
        return 0

    tasks_path = root / "harness-tasks.json"
    progress_path = root / "harness-progress.txt"

    try:
        state = _load_json(tasks_path)
        tasks_raw = state.get("tasks") or []
        if not isinstance(tasks_raw, list):
            raise ValueError("tasks must be a list")
        tasks = [t for t in tasks_raw if isinstance(t, dict)]
    except Exception as e:
        context = f"HARNESS: CONFIG error: cannot read {tasks_path.name}: {e}"
        print(json.dumps({"hookSpecificOutput": {"additionalContext": context}}, ensure_ascii=False))
        return 0

    counts: dict[str, int] = {}
    for t in tasks:
        s = str(t.get("status") or "pending")
        counts[s] = counts.get(s, 0) + 1

    next_task = _pick_next_eligible(tasks)
    next_hint = ""
    if next_task is not None:
        tid = str(next_task.get("id") or "")
        title = str(next_task.get("title") or "").strip()
        next_hint = f" next={tid}{(': ' + title) if title else ''}"

    last_stats = ""
    if progress_path.is_file():
        tail = _tail_text(progress_path)
        lines = [ln.strip() for ln in tail.splitlines() if ln.strip()]
        for ln in reversed(lines[-200:]):
            if " STATS " in f" {ln} " or ln.endswith(" STATS"):
                last_stats = ln
                break
        if not last_stats and lines:
            last_stats = lines[-1]
        if len(last_stats) > 220:
            last_stats = last_stats[:217] + "..."

    summary = (
        "HARNESS: "
        + " ".join(f"{k}={v}" for k, v in sorted(counts.items()))
        + f" total={len(tasks)}"
        + next_hint
    ).strip()
    if last_stats:
        summary += f"\nHARNESS: last_log={last_stats}"

    print(json.dumps({"hookSpecificOutput": {"additionalContext": summary}}, ensure_ascii=False))
    return 0


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

#!/usr/bin/env python3
"""Harness Stop hook — blocks Claude from stopping when eligible tasks remain.

Uses `stop_hook_active` field and a consecutive-block counter to prevent
infinite loops. If the hook blocks N times in a row without any task
completing, it allows the stop with a warning.
"""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any, Optional

MAX_CONSECUTIVE_BLOCKS = 8  # safety valve


def _read_hook_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {"_invalid_json": True}


def _find_harness_root(payload: dict[str, Any]) -> Optional[Path]:
    state_root = os.environ.get("HARNESS_STATE_ROOT")
    if state_root:
        p = Path(state_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p

    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))
    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))

    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def _load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError(f"{path.name} must be a JSON object")
    return data


def _tail_text(path: Path, max_bytes: int = 200_000) -> str:
    with path.open("rb") as f:
        try:
            f.seek(0, os.SEEK_END)
            size = f.tell()
            f.seek(max(0, size - max_bytes), os.SEEK_SET)
        except Exception:
            f.seek(0, os.SEEK_SET)
        chunk = f.read()
    return chunk.decode("utf-8", errors="replace")


def _priority_rank(v: Any) -> int:
    return {"P0": 0, "P1": 1, "P2": 2}.get(str(v or ""), 9)


def _deps_completed(t: dict[str, Any], completed: set[str]) -> bool:
    deps = t.get("depends_on") or []
    if not isinstance(deps, list):
        return False
    return all(str(d) in completed for d in deps)


def _attempts(t: dict[str, Any]) -> int:
    try:
        return int(t.get("attempts") or 0)
    except Exception:
        return 0


def _max_attempts(t: dict[str, Any]) -> int:
    try:
        v = t.get("max_attempts")
        return int(v) if v is not None else 3
    except Exception:
        return 3


def _pick_next(pending: list[dict[str, Any]], retry: list[dict[str, Any]]) -> Optional[dict[str, Any]]:
    def key(t: dict[str, Any]) -> tuple[int, str]:
        return (_priority_rank(t.get("priority")), str(t.get("id", "")))
    pending.sort(key=key)
    retry.sort(key=key)
    return pending[0] if pending else (retry[0] if retry else None)


def _block_counter_path(root: Path) -> Path:
    return root / ".harness-stop-counter"


def _read_block_counter(root: Path) -> tuple[int, int]:
    """Returns (consecutive_blocks, last_completed_count)."""
    p = _block_counter_path(root)
    try:
        raw = p.read_text("utf-8").strip()
        parts = raw.split(",")
        return int(parts[0]), int(parts[1]) if len(parts) > 1 else 0
    except Exception:
        return 0, 0


def _write_block_counter(root: Path, blocks: int, completed: int) -> None:
    p = _block_counter_path(root)
    tmp = p.with_name(f"{p.name}.tmp.{os.getpid()}")
    try:
        tmp.write_text(f"{blocks},{completed}", encoding="utf-8")
        os.replace(tmp, p)
    except Exception:
        try:
            tmp.unlink(missing_ok=True)
        except Exception:
            pass


def _reset_block_counter(root: Path) -> None:
    p = _block_counter_path(root)
    try:
        p.unlink(missing_ok=True)
    except Exception:
        pass


def _is_harness_active(root: Path) -> bool:
    """Check if harness skill is actively running (marker file exists)."""
    return (root / ".harness-active").is_file()


def main() -> int:
    payload = _read_hook_payload()

    # Safety: if stop_hook_active is True, Claude is already continuing
    # from a previous Stop hook block. Check if we should allow stop
    # to prevent infinite loops.
    stop_hook_active = payload.get("stop_hook_active", False)

    root = _find_harness_root(payload)
    if root is None:
        return 0  # no harness project, allow stop

    # Guard: only active when harness skill is triggered
    if not _is_harness_active(root):
        return 0

    tasks_path = root / "harness-tasks.json"
    progress_path = root / "harness-progress.txt"
    try:
        state = _load_json(tasks_path)
        tasks_raw = state.get("tasks") or []
        if not isinstance(tasks_raw, list):
            raise ValueError("tasks must be a list")
        tasks = [t for t in tasks_raw if isinstance(t, dict)]
    except Exception as e:
        if stop_hook_active:
            sys.stderr.write(
                "HARNESS: WARN — harness-tasks.json 无法解析且 stop_hook_active=True，"
                "为避免无限循环，本次允许停止。\n"
            )
            return 0
        reason = (
            "HARNESS: 检测到配置损坏，无法解析 harness-tasks.json。\n"
            f"HARNESS: error={e}\n"
            "按 SKILL.md 的 JSON corruption 恢复：优先用 harness-tasks.json.bak 还原；无法还原则停止并要求人工修复。"
        )
        print(json.dumps({"decision": "block", "reason": reason}, ensure_ascii=False))
        return 0

    session_config = state.get("session_config") or {}
    if not isinstance(session_config, dict):
        session_config = {}

    concurrency_mode = str(session_config.get("concurrency_mode") or "exclusive")
    is_concurrent = concurrency_mode == "concurrent"
    worker_id = os.environ.get("HARNESS_WORKER_ID") or None

    # Check session limits
    try:
        session_count = int(state.get("session_count") or 0)
    except Exception:
        session_count = 0
    try:
        max_sessions = int(session_config.get("max_sessions") or 0)
    except Exception:
        max_sessions = 0
    if max_sessions > 0 and session_count >= max_sessions:
        _reset_block_counter(root)
        return 0  # session limit reached, allow stop

    # Check per-session task limit
    try:
        max_tasks_per_session = int(session_config.get("max_tasks_per_session") or 0)
    except Exception:
        max_tasks_per_session = 0
    if not is_concurrent and max_tasks_per_session > 0 and session_count > 0 and progress_path.is_file():
        tail = _tail_text(progress_path)
        tag = f"[SESSION-{session_count}]"
        finished = 0
        for ln in tail.splitlines():
            if tag not in ln:
                continue
            if " Completed [" in ln or (" ERROR [" in ln and "[task-" in ln):
                finished += 1
        if finished >= max_tasks_per_session:
            _reset_block_counter(root)
            return 0  # per-session limit reached, allow stop

    # Compute eligible tasks
    counts: dict[str, int] = {}
    for t in tasks:
        s = str(t.get("status") or "pending")
        counts[s] = counts.get(s, 0) + 1

    completed_ids = {str(t.get("id", "")) for t in tasks if str(t.get("status", "")) == "completed"}
    completed_count = len(completed_ids)

    pending_eligible = [t for t in tasks if str(t.get("status", "")) == "pending" and _deps_completed(t, completed_ids)]
    retryable = [
        t for t in tasks
        if str(t.get("status", "")) == "failed"
        and _attempts(t) < _max_attempts(t)
        and _deps_completed(t, completed_ids)
    ]
    in_progress_any = [t for t in tasks if str(t.get("status", "")) == "in_progress"]
    if is_concurrent and worker_id:
        in_progress_blocking = [
            t for t in in_progress_any
            if str(t.get("claimed_by") or "") == worker_id or not t.get("claimed_by")
        ]
    else:
        in_progress_blocking = in_progress_any

    # If nothing left to do, allow stop
    if not pending_eligible and not retryable and not in_progress_blocking:
        _reset_block_counter(root)
        # Signal self-reflect hook BEFORE removing active marker
        try:
            (root / ".harness-reflect").touch()
        except Exception:
            pass
        try:
            (root / ".harness-active").unlink(missing_ok=True)
        except Exception:
            pass
        return 0

    # Safety valve: track consecutive blocks without progress
    prev_blocks, prev_completed = _read_block_counter(root)
    if completed_count > prev_completed:
        # Progress was made, reset counter
        prev_blocks = 0
    consecutive = prev_blocks + 1
    _write_block_counter(root, consecutive, completed_count)

    if stop_hook_active and consecutive > MAX_CONSECUTIVE_BLOCKS:
        # Too many consecutive blocks without progress — allow stop to prevent infinite loop
        _reset_block_counter(root)
        sys.stderr.write(
            f"HARNESS: WARN — Stop hook blocked {consecutive} times without progress. "
            "Allowing stop to prevent infinite loop. Check task definitions and validation commands.\n"
        )
        return 0

    # Block the stop — tasks remain
    next_task = _pick_next(pending_eligible, retryable)
    next_hint = ""
    if next_task is not None:
        tid = str(next_task.get("id") or "")
        title = str(next_task.get("title") or "").strip()
        next_hint = f"next={tid}{(': ' + title) if title else ''}"

    summary = (
        "HARNESS: 未满足停止条件，继续执行。\n"
        + "HARNESS: "
        + " ".join(f"{k}={v}" for k, v in sorted(counts.items()))
        + f" total={len(tasks)}"
        + (f" {next_hint}" if next_hint else "")
    ).strip()

    reason = (
        summary
        + "\n"
        + "请按 SKILL.md 的 Task Selection Algorithm 选择下一个 eligible 任务，并完整执行 Task Execution Cycle："
        "Claim → Checkpoint → Validate → Record outcome → STATS（如需）→ Continue。"
    )

    print(json.dumps({"decision": "block", "reason": reason}, ensure_ascii=False))
    return 0


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

#!/usr/bin/env python3
"""Harness SubagentStop hook — blocks subagents from stopping when they
have assigned harness tasks still in progress.

Uses the same decision format as Stop hooks:
  {"decision": "block", "reason": "..."}
"""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any, Optional


def _read_hook_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _find_harness_root(payload: dict[str, Any]) -> Optional[Path]:
    state_root = os.environ.get("HARNESS_STATE_ROOT")
    if state_root:
        p = Path(state_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p
    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))
    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))
    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def _load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError(f"{path.name} must be a JSON object")
    return data


def _is_harness_active(root: Path) -> bool:
    """Check if harness skill is actively running (marker file exists)."""
    return (root / ".harness-active").is_file()


def main() -> int:
    payload = _read_hook_payload()

    # Safety: respect stop_hook_active to prevent infinite loops
    if payload.get("stop_hook_active", False):
        return 0

    root = _find_harness_root(payload)
    if root is None:
        return 0  # no harness project, allow stop

    # Guard: only active when harness skill is triggered
    if not _is_harness_active(root):
        return 0

    tasks_path = root / "harness-tasks.json"
    try:
        state = _load_json(tasks_path)
        session_config = state.get("session_config") or {}
        if not isinstance(session_config, dict):
            session_config = {}
        is_concurrent = str(session_config.get("concurrency_mode") or "exclusive") == "concurrent"
        tasks_raw = state.get("tasks") or []
        if not isinstance(tasks_raw, list):
            return 0
        tasks = [t for t in tasks_raw if isinstance(t, dict)]
    except Exception:
        return 0

    in_progress = [t for t in tasks if str(t.get("status", "")) == "in_progress"]
    worker_id = str(os.environ.get("HARNESS_WORKER_ID") or "").strip()
    agent_id = str(payload.get("agent_id") or "").strip()
    teammate_name = str(payload.get("teammate_name") or "").strip()
    identities = {x for x in (worker_id, agent_id, teammate_name) if x}

    if is_concurrent and in_progress and not identities:
        reason = (
            "HARNESS: concurrent 模式缺少 worker identity（HARNESS_WORKER_ID/agent_id）。"
            "为避免误停导致任务悬空，本次阻止停止。"
        )
        print(json.dumps({"decision": "block", "reason": reason}, ensure_ascii=False))
        return 0

    if is_concurrent:
        owned = [
            t for t in in_progress
            if str(t.get("claimed_by") or "") in identities
        ] if identities else []
    else:
        owned = in_progress

    # Only block when this subagent still owns in-progress work.
    if owned:
        tid = str(owned[0].get("id") or "")
        title = str(owned[0].get("title") or "")
        reason = (
            f"HARNESS: 子代理仍有进行中的任务 [{tid}] {title}。"
            "请完成当前任务的验证和记录后再停止。"
        )
        print(json.dumps({"decision": "block", "reason": reason}, ensure_ascii=False))
        return 0

    return 0  # all done, allow stop


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

#!/usr/bin/env python3
"""Harness TeammateIdle hook — prevents teammates from going idle when
harness tasks remain eligible for execution.

Exit code 2 + stderr message keeps the teammate working.
Exit code 0 allows the teammate to go idle.
"""
from __future__ import annotations

import json
import os
import sys
from pathlib import Path
from typing import Any, Optional


def _read_hook_payload() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        return {}
    try:
        data = json.loads(raw)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _find_harness_root(payload: dict[str, Any]) -> Optional[Path]:
    state_root = os.environ.get("HARNESS_STATE_ROOT")
    if state_root:
        p = Path(state_root)
        if (p / "harness-tasks.json").is_file():
            try:
                return p.resolve()
            except Exception:
                return p
    candidates: list[Path] = []
    env_dir = os.environ.get("CLAUDE_PROJECT_DIR")
    if env_dir:
        candidates.append(Path(env_dir))
    cwd = payload.get("cwd") or os.getcwd()
    candidates.append(Path(cwd))
    seen: set[str] = set()
    for base in candidates:
        try:
            base = base.resolve()
        except Exception:
            continue
        if str(base) in seen:
            continue
        seen.add(str(base))
        for parent in [base, *list(base.parents)[:8]]:
            if (parent / "harness-tasks.json").is_file():
                return parent
    return None


def _load_json(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError(f"{path.name} must be a JSON object")
    return data


def _priority_rank(v: Any) -> int:
    return {"P0": 0, "P1": 1, "P2": 2}.get(str(v or ""), 9)


def _is_harness_active(root: Path) -> bool:
    """Check if harness skill is actively running (marker file exists)."""
    return (root / ".harness-active").is_file()


def main() -> int:
    payload = _read_hook_payload()
    root = _find_harness_root(payload)
    if root is None:
        return 0  # no harness project, allow idle

    # Guard: only active when harness skill is triggered
    if not _is_harness_active(root):
        return 0

    tasks_path = root / "harness-tasks.json"
    try:
        state = _load_json(tasks_path)
        tasks_raw = state.get("tasks") or []
        if not isinstance(tasks_raw, list):
            return 0
        tasks = [t for t in tasks_raw if isinstance(t, dict)]
    except Exception:
        return 0  # can't read state, allow idle

    completed = {str(t.get("id", "")) for t in tasks if str(t.get("status", "")) == "completed"}

    def deps_ok(t: dict[str, Any]) -> bool:
        deps = t.get("depends_on") or []
        if not isinstance(deps, list):
            return False
        return all(str(d) in completed for d in deps)

    def attempts(t: dict[str, Any]) -> int:
        try:
            return int(t.get("attempts") or 0)
        except Exception:
            return 0

    def max_attempts(t: dict[str, Any]) -> int:
        try:
            v = t.get("max_attempts")
            return int(v) if v is not None else 3
        except Exception:
            return 3

    pending = [t for t in tasks if str(t.get("status", "")) == "pending" and deps_ok(t)]
    retryable = [
        t for t in tasks
        if str(t.get("status", "")) == "failed"
        and attempts(t) < max_attempts(t)
        and deps_ok(t)
    ]
    def key(t: dict[str, Any]) -> tuple[int, str]:
        return (_priority_rank(t.get("priority")), str(t.get("id", "")))
    pending.sort(key=key)
    retryable.sort(key=key)
    in_progress = [t for t in tasks if str(t.get("status", "")) == "in_progress"]

    # Check if this teammate owns any in-progress tasks
    worker_id = os.environ.get("HARNESS_WORKER_ID") or ""
    teammate_name = payload.get("teammate_name", "")
    owned = [
        t for t in in_progress
        if str(t.get("claimed_by") or "") in (worker_id, teammate_name)
    ] if (worker_id or teammate_name) else []

    if owned:
        tid = str(owned[0].get("id") or "")
        title = str(owned[0].get("title") or "")
        sys.stderr.write(
            f"HARNESS: 你仍有进行中的任务 [{tid}] {title}。"
            "请继续执行或完成该任务后再停止。\n"
        )
        return 2  # block idle

    if pending or retryable:
        next_t = pending[0] if pending else retryable[0]
        tid = str(next_t.get("id") or "")
        title = str(next_t.get("title") or "")
        sys.stderr.write(
            f"HARNESS: 仍有 {len(pending)} 个待执行 + {len(retryable)} 个可重试任务。"
            f"下一个: [{tid}] {title}。请继续执行。\n"
        )
        return 2  # block idle

    return 0  # all done, allow idle


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