#!/usr/bin/env python3
# Copyright 2026 Mohsen Seyedkazemi Ardebili
# SPDX-License-Identifier: Apache-2.0
# Part of agent-fleet — https://github.com/MSKazemi/agent-fleet
"""fleet queue engine — a per-project task queue for a fleet of parallel agents.

Runs in ANY project. The queue lives next to the work, not next to this script:

    $AGENT_ROOT               if set, wins outright
    nearest ancestor with .agents/     the usual case, created by `fleet init`
                              ($AGENT_DIR renames that directory; default ".agents")
    git repository root       first run in a repo
    $PWD                      last resort

so `fleet add` in ~/work/foo and in ~/work/bar feed two independent queues, and
several fleets can run at once in different projects without touching each other.

Runtime state lives in <root>/.agents/ — tasks are one JSON file each, in a
directory that IS their state:

    .agents/queue/inbox/    waiting to be claimed
    .agents/queue/active/   claimed by an agent, in progress
    .agents/queue/blocked/  needs a human (login, captcha, 2FA) — agent moved on
    .agents/queue/done/     finished, with evidence
    .agents/queue/failed/   gave up after retries

Claiming is a single os.rename() out of queue/inbox/. rename(2) is atomic, so
when several agents race for the same task exactly one wins and the losers get
FileNotFoundError and try the next file. No lock files, no daemon.

Human commands:  init, where, doctor, add, add-batch, from-note, to-note,
                 plan, list, status, watch, blocked, unblock, reset
Agent commands:  claim, done, block, fail, show
"""

from __future__ import annotations

import argparse
import contextlib
import fnmatch
import json
import os
import re
import shutil
import subprocess
import sys
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path

STATES = ("inbox", "active", "blocked", "done", "failed")
DEFAULT_CONFIG = {
    "max_workers": 4,        # ceiling on browser workers, i.e. on terminals
    "max_subagents": 6,      # ceiling on in-session subagents for the agent lane
    "default_kind": "task",  # free-form label; yours to define
    "worker_command": "claude --dangerously-skip-permissions",
    # Kinds that always need a browser tab even without a URL in the task.
    "browser_kinds": ["form-fill", "browse", "scrape"],
    # Where workers start. Claude resolves skills from the working directory, so
    # point this at a repo holding the skills a task kind needs. "" = the project.
    # Per-kind overrides win: {"apply": "~/work/profile-repo"}.
    "workers_start_in": "",
    "workers_start_in_by_kind": {},
    # Checklist file read by `fleet from-note` (default: TASKS.md / BACKLOG.md / TODO.md).
    "checklist": "",
    # If the checklist has a generated header, the line that ends it.
    "checklist_user_marker": "",

    # --- what this machine can actually take -------------------------------
    # null means "work it out from this machine, now" — nothing here is a fixed
    # assumption about your hardware. Set a number to pin it instead.
    "mem_per_browser_worker_mb": None,   # measured from real sessions, else derived
    "mem_per_agent_task_mb": None,       # measured from real sessions, else derived
    "reserve_mem_mb": None,              # derived: ~10% of RAM, at least 1 GB
    "reserve_cores": None,               # derived: ~1 core per 8, at least 1
    "max_load_per_core": 0.85,           # treat the machine as full above this
    "respect_machine": True,             # False = trust max_workers alone
}

# How a task gets executed. Only the browser lane costs a terminal.
EXEC_MODES = ("browser", "agent", "serial")
STALE_AFTER = 1800          # seconds in active/ before a task looks abandoned
LANE_HELP = {
    "browser": "own Chrome tab + own terminal (parallel, one per site)",
    "agent":   "in-session subagent, no terminal (parallel)",
    "serial":  "one at a time, in this session (no terminal)",
}


def infer_exec(url: str, kind: str, browser_kinds: list[str] | None = None) -> str:
    """A task needs a browser tab only if it acts on a live page.

    A URL is the strong signal; `browser_kinds` in .agents/config.json covers
    kinds that need a browser even when the task is written as a sentence.
    """
    if kind in (browser_kinds if browser_kinds is not None else DEFAULT_CONFIG["browser_kinds"]):
        return "browser"
    return "browser" if url.startswith(("http://", "https://")) else "agent"


# --------------------------------------------------------------------------- #
# where are we
# --------------------------------------------------------------------------- #


def queue_dirname() -> str:
    """The queue directory's NAME under the project root: ``$AGENT_DIR``, else ``.agents``.

    ⛔ ONE place decides it. Read it here and nowhere else — a second hardcoded ".agents"
    (in `init`, in `doctor`, in the ignore rule) is how `init` builds one directory while
    `add` writes another, leaving a phantom queue nothing reads. `bin/fleet` reads the same
    variable, so the two halves of the command cannot disagree about where the queue is.
    """
    return os.environ.get("AGENT_DIR") or ".agents"


def resolve_root() -> Path:
    env = os.environ.get("AGENT_ROOT")
    if env:
        return Path(env).expanduser().resolve()

    cwd = Path.cwd().resolve()
    for candidate in (cwd, *cwd.parents):
        if (candidate / queue_dirname()).is_dir():
            return candidate

    try:
        out = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True, text=True, timeout=5, cwd=cwd,
        )
        if out.returncode == 0 and out.stdout.strip():
            return Path(out.stdout.strip())
    except Exception:
        pass

    return cwd


ROOT = resolve_root()
AGENTS_DIR = ROOT / queue_dirname()
QUEUE = AGENTS_DIR / "queue"
LOGS = AGENTS_DIR / "logs"
CONFIG_PATH = AGENTS_DIR / "config.json"


def config() -> dict:
    cfg = dict(DEFAULT_CONFIG)
    if CONFIG_PATH.exists():
        try:
            cfg.update(json.loads(CONFIG_PATH.read_text()))
        except json.JSONDecodeError as exc:
            print(f"warning: {CONFIG_PATH} is not valid JSON ({exc}); using defaults",
                  file=sys.stderr)
    return cfg


def now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def ensure_dirs() -> None:
    """Create the queue on demand — and protect it the moment it appears.

    Any command may be the first to touch a project, so the gitignore/exclude
    rule cannot live only in `init`: a stray `fleet add` in an uninitialised repo
    would otherwise create .agents/ with personal data and no ignore rule.
    """
    fresh = not AGENTS_DIR.exists()
    for state in STATES:
        (QUEUE / state).mkdir(parents=True, exist_ok=True)
    LOGS.mkdir(parents=True, exist_ok=True)
    if fresh:
        try:
            hide_artifacts(ROOT, quiet=True)
        except Exception as exc:                      # never block work on this
            print(f"warning: could not add the ignore rule ({exc}); "
                  f"run 'fleet doctor --fix'", file=sys.stderr)


def task_files(state: str) -> list[Path]:
    return sorted((QUEUE / state).glob("*.json"))


def find_task(task_id: str) -> tuple[Path, str] | tuple[None, None]:
    """Locate a task by id (or unique id prefix) across every state."""
    for state in STATES:
        for path in task_files(state):
            if path.stem.split("-")[-1].startswith(task_id) or path.stem == task_id:
                return path, state
    return None, None


def read_task(path: Path) -> dict:
    return json.loads(path.read_text())


def write_task(path: Path, task: dict) -> None:
    tmp = path.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(task, indent=2) + "\n")
    tmp.replace(path)


def log(task: dict, event: str, **fields) -> None:
    task.setdefault("history", []).append({"at": now(), "event": event, **fields})


def notify(title: str, body: str) -> None:
    """Best-effort desktop notification; never fatal.

    FLEET_NO_NOTIFY silences it. A blocked task is worth interrupting a human
    for, which is exactly why the test suite and CI must not raise one: a run
    that cries wolf teaches you to ignore the alert that matters.
    """
    if os.environ.get("FLEET_NO_NOTIFY"):
        return
    try:
        subprocess.run(
            ["notify-send", "--app-name=agent-fleet", "--urgency=normal", title, body],
            check=False, timeout=5,
        )
    except Exception:
        pass


def move(path: Path, state: str) -> Path:
    dest = QUEUE / state / path.name
    shutil.move(str(path), str(dest))
    return dest


def site_of(url: str) -> str:
    return url.split("/")[2].removeprefix("www.") if "//" in url else ""


# --------------------------------------------------------------------------- #
# setup
# --------------------------------------------------------------------------- #


TASKS_JSON = ".vscode/tasks.json"


def artifacts(root: Path) -> tuple[str, ...]:
    """The ignore patterns for this project's fleet artifacts.

    The default queue keeps the unanchored ``.agents/`` it has always had. A RENAMED queue is
    ignored by its anchored path instead: a bare ``fleet/`` would hide every directory of that
    name anywhere in the tree, and the project's own ``fleet/`` source is exactly the kind of
    thing that shares it.
    """
    name = queue_dirname()
    if name == ".agents":
        return (".agents/", TASKS_JSON)
    queue_dir = (root / name).resolve()
    top = git(root, "rev-parse", "--show-toplevel").stdout.strip() if in_git_repo(root) else ""
    try:
        rel = queue_dir.relative_to(Path(top).resolve()).as_posix() if top else name
    except ValueError:
        rel = name
    return (f"/{rel}/", TASKS_JSON)
EXCLUDE_HEADER = "# --- agent-fleet: local runtime, never commit ---"


def git(root: Path, *args: str) -> subprocess.CompletedProcess:
    return subprocess.run(["git", "-C", str(root), *args],
                          capture_output=True, text=True, timeout=10)


def in_git_repo(root: Path) -> bool:
    out = git(root, "rev-parse", "--is-inside-work-tree")
    return out.returncode == 0 and out.stdout.strip() == "true"


def tasks_json_tracked(root: Path) -> bool:
    return bool(git(root, "ls-files", "--", ".vscode/tasks.json").stdout.strip())


def hide_artifacts(root: Path, quiet: bool = False) -> None:
    """Keep fleet artifacts out of the project's GitHub.

    Preferred target is .git/info/exclude: it ignores paths exactly like
    .gitignore but is per-clone and is itself never committed, so a project's
    tracked files are not touched at all. Only when this is not a git repo do we
    fall back to .gitignore, so the rule is already in place if it becomes one.
    """
    if in_git_repo(root):
        gitdir = git(root, "rev-parse", "--git-dir").stdout.strip()
        exclude = (root / gitdir / "info" / "exclude").resolve()
        exclude.parent.mkdir(parents=True, exist_ok=True)
        current = exclude.read_text() if exclude.exists() else ""
        # A tasks.json the project commits on purpose is its file, not ours —
        # excluding it would be a lie (git ignores nothing already tracked).
        wanted = [a for a in artifacts(root)
                  if a != TASKS_JSON or not tasks_json_tracked(root)]
        missing = [a for a in wanted if a not in current.split()]
        if missing:
            with open(exclude, "a") as fh:
                if EXCLUDE_HEADER not in current:
                    fh.write(f"\n{EXCLUDE_HEADER}\n")
                fh.write("".join(f"{a}\n" for a in missing))
            if not quiet:
                print(f"excluded {', '.join(missing)} via {exclude}  "
                      f"(local only, never committed)")
        elif not quiet:
            print(f"already excluded via {exclude}")
    else:
        gitignore = root / ".gitignore"
        current = gitignore.read_text() if gitignore.exists() else ""
        missing = [a for a in artifacts(root) if a not in current.split()]
        if missing:
            with open(gitignore, "a") as fh:
                fh.write(f"\n{EXCLUDE_HEADER}\n" + "".join(f"{a}\n" for a in missing))
            if not quiet:
                print(f"added {', '.join(missing)} to {gitignore}  (not a git repo yet)")


def cmd_init(args) -> int:
    """Make the current project a fleet project.

    Falls back to ROOT, not Path.cwd(): every other command resolves its project
    through resolve_root(), so an init that read only the cwd made `AGENT_ROOT=/x
    fleet init` silently set up whatever directory you happened to be standing
    in — and then `fleet add` in the same shell wrote somewhere else entirely.
    An explicit path argument still wins over both.
    """
    root = Path(args.path).expanduser().resolve() if args.path else ROOT.resolve()
    agents_dir = root / queue_dirname()
    for state in STATES:
        (agents_dir / "queue" / state).mkdir(parents=True, exist_ok=True)
    (agents_dir / "logs").mkdir(parents=True, exist_ok=True)

    cfg_path = agents_dir / "config.json"
    if not cfg_path.exists():
        cfg = dict(DEFAULT_CONFIG)
        if args.max_workers:
            cfg["max_workers"] = args.max_workers
        cfg_path.write_text(json.dumps(cfg, indent=2) + "\n")

    hide_artifacts(root)

    # A checklist the project already keeps always wins — including one named by
    # "checklist" in a config that survived an earlier init. Only a project with
    # no checklist at all gets the starter file, and it is never overwritten:
    # a stranger's TODO.md is their work, not ours to reformat.
    checklist_note = ""
    if not args.no_checklist:
        cfg_now = json.loads(cfg_path.read_text())
        candidates = [n for n in (cfg_now.get("checklist"), *NOTE_NAMES) if n]
        found = next((n for n in candidates if (root / n).is_file()), None)
        if found:
            checklist_note = f"  checklist: {found} (yours — left untouched)"
        else:
            (root / NOTE_NAMES[0]).write_text(CHECKLIST_TEMPLATE)
            checklist_note = f"  checklist: {NOTE_NAMES[0]} (created — write tasks there)"

    print(f"fleet project ready: {agents_dir}")
    print(f"  config:   {cfg_path}")
    if checklist_note:
        print(checklist_note)
    print(f"  add work: fleet add <url> --title ...")
    print(f"  run it:   fleet run")
    return 0


def _expand(path: str) -> Path:
    return Path(path).expanduser()


def worker_cwd(kind: str | None = None) -> Path:
    """Where a worker should start.

    Claude resolves skills from the working directory, so a task kind whose
    skills live in another repo must start there — while $AGENT_ROOT keeps the
    worker on this project's queue. Configured, never guessed:

        "workers_start_in": "~/work/toolbox",
        "workers_start_in_by_kind": {"apply": "~/work/profile-repo"}
    """
    cfg = config()
    if kind is None:
        kinds = {t.get("kind", "") for state in ("inbox", "active")
                 for t in (read_task(p) for p in task_files(state))}
        for candidate, path in cfg.get("workers_start_in_by_kind", {}).items():
            if candidate in kinds and _expand(path).is_dir():
                return _expand(path)
    else:
        path = cfg.get("workers_start_in_by_kind", {}).get(kind)
        if path and _expand(path).is_dir():
            return _expand(path)
    default = cfg.get("workers_start_in") or ""
    return _expand(default) if default and _expand(default).is_dir() else ROOT


def check_worker_cwd() -> str:
    """Report only what the launcher cannot resolve by itself."""
    cfg = config()
    configured = {k: v for k, v in cfg.get("workers_start_in_by_kind", {}).items()}
    if cfg.get("workers_start_in"):
        configured["*"] = cfg["workers_start_in"]
    missing = {k: v for k, v in configured.items() if not _expand(v).is_dir()}

    kinds = {t.get("kind", "") for state in ("inbox", "active")
             for t in (read_task(p) for p in task_files(state))}
    relevant = {k: v for k, v in missing.items() if k == "*" or k in kinds}
    if relevant:
        listed = ", ".join(f"{k} → {v}" for k, v in relevant.items())
        return (f"PROBLEM — workers_start_in points at a directory that does not "
                f"exist: {listed}\n"
                f"           fix it in {CONFIG_PATH} or drop the setting")
    resolved = worker_cwd()
    if resolved != ROOT:
        return f"OK — workers will start in {resolved}, serving this queue"
    return ""


BROWSER_PROBE = AGENTS_DIR / "browser-probe.json"


def read_browser_probe() -> dict | None:
    """The last recorded answer to 'can a worker actually open a tab?'.

    Written by `fleet probe-browser`. Nothing here guesses: with no probe on
    file the answer is unknown, and doctor says so rather than assuming yes.
    """
    try:
        return json.loads(BROWSER_PROBE.read_text())
    except (OSError, ValueError):
        return None


def pending_browser_tasks() -> int:
    return sum(1 for state in ("inbox", "active")
               for t in (read_task(p) for p in task_files(state))
               if t.get("exec", "browser") == "browser")


def check_browser() -> tuple[list[str], str]:
    """(lines, problem) — whether the browser lane can actually run from here.

    A worker's Chrome access is not inherited from the session you type in, so
    'the queue is fine' and 'this work is runnable' are two different questions.
    """
    waiting = pending_browser_tasks()
    probe = read_browser_probe()

    if probe is None:
        if not waiting:
            return [], ""
        return ([f"{waiting} task(s) need a Chrome tab, and whether a worker has one",
                 "         is unknown — a worker does not inherit this session's browser.",
                 "         Measure it: fleet probe-browser"], "")

    when = str(probe.get("checked_at", "?"))[:10]
    if probe.get("browser"):
        return [f"a worker could open a tab ✓  (probed {when})"], ""

    # A NO from the default probe is about PRINT mode (it asks with -p). Terminal
    # workers run the same binary interactively, are offered the Chrome toolset,
    # and each get their own per-session tab group — so a print-mode NO must not
    # be reported as "the browser lane is broken here".
    if probe.get("mode", "print") == "print":
        lines = [f"a print-mode session could NOT open a tab ✗  (probed {when})",
                 "         that is the answer for --headless; terminal workers are not",
                 "         covered by it and each get their own Chrome tab group"]
        if waiting:
            lines.append(f"         run the {waiting} queued browser task(s) with: fleet run --terminals")
        return lines, ""

    lines = [f"a worker could NOT open a tab ✗  (probed {when})"]
    if waiting:
        lines.append(f"         {waiting} browser task(s) would come back blocked, not done")
        return lines, "workers have no browser, and browser tasks are queued"
    lines.append("         nothing browser-shaped is queued, so nothing is stuck on it")
    return lines, ""


def cmd_doctor(args) -> int:
    """Prove no fleet artifact can reach this project's remote, and that the
    queued work can actually be done from here."""
    root = ROOT
    problems: list[str] = []
    fixes: list[list[str]] = []

    print(f"project: {root}")

    if AGENTS_DIR.is_dir():
        scope = check_worker_cwd()
        print(f"scope:   {scope if scope else 'queued work is runnable from here ✓'}")
        if scope.startswith("PROBLEM"):
            problems.append("workers_start_in points at a missing directory")

        browser_lines, browser_problem = check_browser()
        if browser_lines:
            print("browser: " + "\n".join(browser_lines))
        if browser_problem:
            problems.append(browser_problem)

    if not in_git_repo(root):
        print("git:     not a git repository — nothing can be pushed from here")
        if problems:
            print("\nproblems:")
            for problem in problems:
                print(f"  - {problem}")
            return 1
        return 0

    remote = git(root, "remote", "-v").stdout.strip().splitlines()
    print(f"remote:  {remote[0] if remote else '(none)'}")

    # A tasks.json the project commits is the project's own file; the fleet only
    # merges into it and restores it. It is not a fleet artifact to hide.
    tasks_tracked = tasks_json_tracked(root)
    ours = [a for a in artifacts(root) if a != TASKS_JSON or not tasks_tracked]

    # 1. are our artifacts ignored at all?
    for artifact in ours:
        path = root / artifact.rstrip("/")
        if not path.exists():
            print(f"ignore:  {artifact:<20} not present")
            continue
        ignored = git(root, "check-ignore", "-q", str(path)).returncode == 0
        print(f"ignore:  {artifact:<20} {'ignored ✓' if ignored else 'NOT IGNORED ✗'}")
        if not ignored:
            problems.append(f"{artifact} is not ignored")

    # 2. is anything already tracked? that is what actually reaches GitHub.
    tracked_queue = git(root, "ls-files", "--", queue_dirname()).stdout.split()
    if tracked_queue:
        print(f"tracked: {len(tracked_queue)} file(s) under {queue_dirname()}/ ARE tracked ✗")
        for path in tracked_queue[:10]:
            print(f"           {path}")
        problems.append(f"{len(tracked_queue)} queue file(s) tracked by git")
        fixes.append(["rm", "-r", "--cached", "--quiet", queue_dirname()])
    else:
        print("tracked: no queue file is tracked ✓")

    # .vscode/tasks.json is different: plenty of projects commit their own on
    # purpose. Untracking it would delete it from their GitHub, so never do that
    # automatically — just say what will show up in `git status`.
    if tasks_tracked:
        print("tasks:   this project commits .vscode/tasks.json — the fleet merges into it,\n"
              "           so a run shows as a local modification. 'agents clean' restores the\n"
              "           original. To hide the edit entirely:\n"
              "             git update-index --skip-worktree .vscode/tasks.json")

    # 3. did any of it ever get committed? that needs history surgery, not a fix.
    hist_paths = [a.rstrip("/") for a in ours]
    hist = git(root, "log", "--oneline", "-1", "--", *hist_paths).stdout.strip()
    if hist:
        print(f"history: fleet artifacts appear in history — {hist}")
        problems.append("fleet artifacts exist in committed history")

    if not problems:
        print("\nclean — nothing fleet-related can reach the remote")
        return 0

    print("\nproblems:")
    for problem in problems:
        print(f"  - {problem}")

    # --fix is about git hygiene only. A missing browser is not a git problem and
    # saying "run --fix" would be a lie about what that flag can do.
    git_problems = [p for p in problems if "browser" not in p]
    if args.fix:
        hide_artifacts(root)
        for fix in fixes:
            out = git(root, *fix)
            print(f"  git {' '.join(fix)} → {'ok' if out.returncode == 0 else out.stderr.strip()}")
        print("\nre-run 'fleet doctor' to confirm")
    elif git_problems:
        print("\nrun 'fleet doctor --fix' to exclude and untrack them")

    if hist:
        print("\nNOTE: files already in committed history stay there until the history is\n"
              "rewritten (git filter-repo) and force-pushed. That is destructive and\n"
              "outward-facing — decide it deliberately, per repo.")
    return 1


def cmd_where(args) -> int:
    cfg = config()
    print(f"root:        {ROOT}")
    print(f"queue:       {QUEUE}")
    print(f"logs:        {LOGS}")
    print(f"config:      {CONFIG_PATH}{'' if CONFIG_PATH.exists() else '  (defaults)'}")
    print(f"max_workers: {cfg['max_workers']}")
    if not AGENTS_DIR.is_dir():
        print("\nnot initialised here yet — run: fleet init")
    return 0


# --------------------------------------------------------------------------- #
# checklist bridge — read tasks from the markdown file you already keep
# --------------------------------------------------------------------------- #

NOTE_NAMES = ("TASKS.md", "BACKLOG.md", "TODO.md")

# Written by `fleet init` only when the project has no checklist at all. The
# example items are INDENTED on purpose: only a top-level "- [ ]" is a task, so
# a fresh project starts with an empty queue instead of six phantom ones.
CHECKLIST_TEMPLATE = """\
# Tasks

Your checklist and the fleet's queue, in one file.
`fleet from-note` promotes items into the queue; `fleet to-note` ticks them back.

## INBOX

<!-- Dump raw ideas here — any format, any language, half a sentence is fine.
     from-note IGNORES this section on purpose: an idea nobody has thought
     through yet should not become a task. Move it to READY when it is real. -->

## READY

<!-- from-note reads ONLY this section (or any heading naming NOW or TODO).

     One task per top-level "- [ ]". Lines indented beneath it travel with it
     as notes, so a task carries its own why / steps / done-when.

     Markers:
       !    raise priority          !!  do this first
       ~    lower priority          ⇒   run serially, never in parallel
       a URL in the line   → browser lane, at most one worker per host
       ? at the line start → your decision to make; never queued
       🔒                  → never touched

     Ticked items, and anything already queued, are skipped — so re-running
     from-note only adds what is new. That makes it safe to edit this file
     while workers are running.

     Examples (indented, so they are notes rather than tasks):
       - [ ] check https://example.com/pricing still lists the free tier
       - [ ] ⇒ migrate the config parser, then run the suite
         why: three other tasks are waiting on the rewrite
         done-when: ./run-tests exits 0
-->

## 🧾 LOG
"""
# Some checklist formats separate a generated header from the user's own region.
# Set "checklist_user_marker" in config to that line and only the region after it
# is read; unset (the default) reads the whole file.
# Only a top-level "- [ ]" is a task. Indented lines under it are its sub-steps
# and travel with it as notes — promoting them separately would shred the item.
ITEM = re.compile(r"^-\s*\[(?P<done>[ xX])\]\s*(?P<body>.+?)\s*$")
URL_IN = re.compile(r"https?://\S+")


def find_note(root: Path) -> Path | None:
    """The checklist this project keeps. `"checklist": "NOTES.md"` in
    .agents/config.json names a different one."""
    configured = config().get("checklist")
    names = (configured, *NOTE_NAMES) if configured else NOTE_NAMES
    for name in names:
        if (root / name).is_file():
            return root / name
    return None


def note_sections(text: str) -> dict[str, list[str]]:
    """Section title → its lines, restricted to the user region when marked."""
    marker = config().get("checklist_user_marker") or ""
    body = text.split(marker, 1)[-1] if marker else text
    sections: dict[str, list[str]] = {}
    current = ""
    for line in body.splitlines():
        if line.startswith("## "):
            current = line[3:].strip()
            sections[current] = []
        elif current:
            sections[current].append(line)
    return sections


#: Everything between the checkbox and the title is where MARKERS live. Outside it, `?` is
#: punctuation, `!` is emphasis and `🔒` is a word about locking.
NOTE_MARKER_ZONE = re.compile(r"^(?:`[^`]{1,12}`|[!~?⇉⇒↻🔒]{1,3}|‼\ufe0f?|❓|🔴|⛔|⭐|·|\s)+")

#: The paths an item declares it will WRITE, from its `touches:` line. Same field `notectl.py`
#: reads; see the mska-note skill.
NOTE_TOUCHES = re.compile(r"^[ \t]*[-*]?[ \t]*touches:[ \t]*(.+?)[ \t]*$", re.IGNORECASE)
NOTE_READ_ONLY = frozenset({
    "(read-only)", "read-only", "readonly", "read only", "(readonly)", "(read only)",
    "none", "(none)", "nothing", "-", "—", "n/a",
})


def note_markers(body: str) -> str:
    """The marker prefix of a note item — the only place a marker counts.

    ⛔ Scanned over the WHOLE line, these are not markers. `"🔒" in body` dropped
    kubeintellect `T203`, whose title is followed by the prose "🔒 the DRAFT is ready"; `"!!" in
    body` invented a priority from an exclamation mark in a sentence; `"⇒" in body` serialised
    any item whose text happened to contain an arrow. A marker is a marker because of WHERE it
    is written. (Same defect, same fix, as `_marker_zone` in notectl.py.)
    """
    m = NOTE_MARKER_ZONE.match(body)
    return (m.group(0) if m else "").replace("❓", "?")


def note_touches(detail: list[str] | None) -> list[str] | None:
    """Paths the item says it will write. ``None`` = undeclared. ``[]`` = declared read-only."""
    if not detail:
        return None
    values = [m.group(1) for line in detail if (m := NOTE_TOUCHES.match(line))]
    if not values:
        return None
    paths: list[str] = []
    for raw in values:
        raw = raw.strip().strip("`").strip()
        if raw.lower() in NOTE_READ_ONLY:
            continue
        for part in re.split(r"[,;]", raw):
            part = part.strip().strip("`").strip().lstrip("./")
            if part and part not in paths:
                paths.append(part)
    return paths


CLAIM_LOCK = "claim.lock"
CLAIM_LOCK_STALE_S = 30       # a claim is a rename; anything older than this is a dead worker


@contextlib.contextmanager
def claim_gate():
    """Serialise the check-then-rename of a claim. Brief by construction.

    ⛔ Breaking a stale gate is a RENAME, never an unlink: `unlink` removes whatever sits at the
    path, which may be a lock another worker created a microsecond ago, and then two workers
    hold it. (Same defect, same fix, as the note lock in notectl.py.)
    """
    lock = QUEUE / CLAIM_LOCK
    for _ in range(200):                    # ~2 s of patience, then break a dead gate
        try:
            fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
            os.write(fd, str(os.getpid()).encode())
            os.close(fd)
            break
        except FileExistsError:
            try:
                if time.time() - lock.stat().st_mtime > CLAIM_LOCK_STALE_S:
                    os.rename(lock, lock.with_suffix(f".broken-{os.getpid()}"))
                    (lock.with_suffix(f".broken-{os.getpid()}")).unlink(missing_ok=True)
                    continue
            except (FileNotFoundError, OSError):
                continue
            time.sleep(0.01)
    try:
        yield
    finally:
        with contextlib.suppress(FileNotFoundError):
            lock.unlink()


def paths_collide(a: str, b: str) -> bool:
    """Do two declared paths write the same thing? Directories and globs count.

    Every ambiguity resolves toward COLLIDE: a false collision costs one serial run, a missed
    one costs a lost edit that is invisible in the run that caused it. Mirrors `_paths_collide`
    in notectl.py so the two readers agree about the same note.
    """
    x, y = a.lower().rstrip("/"), b.lower().rstrip("/")
    if x == y:
        return True
    if ("/" not in x or "/" not in y) and x.rsplit("/", 1)[-1] == y.rsplit("/", 1)[-1]:
        return True
    for m, n in ((x, y), (y, x)):
        if n.startswith(m + "/") or fnmatch.fnmatchcase(n, m):
            return True
    return False


def active_touches() -> list[str]:
    """Every path an in-flight task said it would write."""
    out: list[str] = []
    for path in task_files("active"):
        try:
            out.extend(read_task(path).get("touches") or [])
        except (FileNotFoundError, json.JSONDecodeError):
            continue
    return out


def classify(body: str, detail: list[str] | None = None) -> dict | None:
    """Map one note item onto a queue task. None = deliberately not ours."""
    zone = note_markers(body)
    # Markers that mean "not for an agent": yours to decide, or hands off.
    if "🔒" in zone or "?" in zone or body.lstrip().startswith("-x"):
        return None

    priority = 1 if "!!" in zone else 2 if "!" in zone else 8 if "~" in zone else 5

    # ⛔ THE PARALLEL LANE IS OPT-IN, AND IT DID NOT USED TO BE. This returned `agent` — the
    # concurrent lane — for everything that was not `⇒`-marked and had no URL, so two items
    # writing the SAME FILE were handed to two workers at once. Nothing recorded which files an
    # item writes, so nothing could tell. An item now earns the parallel lane only by declaring
    # `touches:`; undeclared is unknown, and unknown runs serial. That is the same rule
    # `notectl.py plan` applies, so the note means one thing to both readers.
    touches = note_touches(detail)
    url_match = URL_IN.search(body)
    if "⇒" in zone:
        mode = "serial"
    elif url_match:
        mode = "browser"                    # a browser task writes no repo files
    elif touches is None:
        mode = "serial"
    else:
        mode = "agent"

    title = re.sub(r"[`*]", "", body)
    title = re.sub(r"^\s*[!~]+\s*", "", title)
    title = re.sub(r"\s*[⇉⇒·]\s*", " ", title).strip()

    return {
        "url": url_match.group(0) if url_match else title,
        "exec": mode,
        "priority": priority,
        "title": title[:120],
        "touches": touches,
    }


def cmd_from_note(args) -> int:
    """Promote note items into the queue. The note stays the source of truth."""
    ensure_dirs()
    note = Path(args.note).expanduser() if args.note else find_note(ROOT)
    if note is None:
        print(f"no checklist file in {ROOT} (looked for {', '.join(NOTE_NAMES)}); "
              f"pass one with --note", file=sys.stderr)
        return 1

    sections = note_sections(note.read_text())
    wanted = [args.section] if args.section else [
        s for s in sections if any(k in s.upper() for k in ("NOW", "READY", "TODO"))]
    if not wanted:
        print(f"no NOW / READY / TODO section in {note.name}; "
              f"name one with --section", file=sys.stderr)
        return 1

    # Never queue the same note line twice, whatever state it reached.
    seen = {read_task(p).get("note_text") for state in STATES for p in task_files(state)}

    added = skipped = 0
    for title in wanted:
        lines = sections.get(title, [])
        for index, line in enumerate(lines):
            match = ITEM.match(line)
            if not match or match.group("done").lower() == "x":
                continue
            body = match.group("body")

            # Everything indented under the item is its detail: sub-steps, why:,
            # done-when:. The worker needs all of it.
            detail = []
            for follower in lines[index + 1:]:
                if not follower.strip():
                    break
                if not follower[:1].isspace():
                    break
                detail.append(follower.strip())

            if body in seen:
                skipped += 1
                continue
            spec = classify(body, detail)
            if spec is None:
                skipped += 1
                continue

            task_id = uuid.uuid4().hex[:8]
            task = {
                "id": task_id, "created": now(),
                "url": spec["url"], "site": site_of(spec["url"]),
                "kind": args.kind or config()["default_kind"],
                "exec": spec["exec"], "title": spec["title"],
                "touches": spec.get("touches"),
                "notes": "\n".join(detail),
                "priority": spec["priority"], "attempts": 0, "agent": None,
                "note_file": str(note), "note_text": body,
                "history": [{"at": now(), "event": "from-note", "section": title}],
            }
            name = f"p{spec['priority']}-{int(time.time())}-{task_id}.json"
            write_task(QUEUE / "inbox" / name, task)
            print(f"{task_id}  {spec['exec']:<7} p{spec['priority']}  {spec['title'][:70]}")
            added += 1

    print(f"\nadded {added}, skipped {skipped} (already queued, 🔒, ? or -x)")
    if added:
        print("plan it:  fleet plan")
    return 0


def cmd_to_note(args) -> int:
    """Write finished work back: tick the item, append one LOG line. Never deletes."""
    ensure_dirs()
    done = [(p, read_task(p)) for p in task_files("done")]
    pending = [(p, t) for p, t in done if t.get("note_text") and not t.get("note_synced")]
    if not pending:
        print("nothing to write back")
        return 0

    by_note: dict[str, list[tuple[Path, dict]]] = {}
    for path, task in pending:
        by_note.setdefault(task["note_file"], []).append((path, task))

    for note_path, items in by_note.items():
        note = Path(note_path)
        if not note.is_file():
            print(f"note gone: {note}", file=sys.stderr)
            continue
        lines = note.read_text().splitlines()
        ticked, log_lines = 0, []

        for path, task in items:
            for i, line in enumerate(lines):
                match = ITEM.match(line)
                if match and match.group("body") == task["note_text"] \
                        and match.group("done") == " ":
                    lines[i] = line.replace("- [ ]", "- [x]", 1)
                    ticked += 1
                    break
            log_lines.append(f"- {now()[:10]} · fleet · {task.get('title') or task['url']} "
                             f"— {task.get('result', '')[:100]}")
            task["note_synced"] = True
            log(task, "written back to note")
            write_task(path, task)

        # LOG is the one section the fleet appends to, per the note's own protocol.
        try:
            anchor = next(i for i, l in enumerate(lines) if l.startswith("## 🧾 LOG"))
            lines[anchor + 1:anchor + 1] = ["", *log_lines]
        except StopIteration:
            lines += ["", "## 🧾 LOG", "", *log_lines]

        note.write_text("\n".join(lines) + "\n")
        print(f"{note.name}: ticked {ticked}, logged {len(log_lines)}")
    return 0


# --------------------------------------------------------------------------- #
# human commands
# --------------------------------------------------------------------------- #


def cmd_add(args) -> int:
    ensure_dirs()
    task_id = uuid.uuid4().hex[:8]
    kind = args.kind or config()["default_kind"]
    mode = getattr(args, "exec", "") or ("serial" if getattr(args, "serial", False)
                                         else infer_exec(args.url, kind,
                                                         config().get("browser_kinds")))
    task = {
        "id": task_id,
        "created": now(),
        "url": args.url,
        "site": args.site or site_of(args.url),
        "kind": kind,
        "exec": mode,
        "title": args.title or "",
        "notes": args.note or "",
        "priority": args.priority,
        "attempts": 0,
        "agent": None,
        "history": [{"at": now(), "event": "added"}],
    }
    name = f"p{args.priority}-{int(time.time())}-{task_id}.json"
    write_task(QUEUE / "inbox" / name, task)
    print(f"{task_id}  {mode:<7} {kind}  {args.url}")
    return 0


def cmd_add_batch(args) -> int:
    """Read tasks from a file or stdin: one per line, 'URL | title | note'."""
    ensure_dirs()
    stream = sys.stdin if args.file == "-" else open(args.file)
    count = 0
    with stream:
        for line in stream:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            parts = [p.strip() for p in line.split("|")]
            cmd_add(argparse.Namespace(
                url=parts[0],
                title=parts[1] if len(parts) > 1 else "",
                note=parts[2] if len(parts) > 2 else "",
                site="", kind=args.kind, priority=args.priority,
                **{"exec": args.exec, "serial": False},
            ))
            count += 1
    print(f"added {count} task(s)")
    return 0


def summarize(path: Path, state: str) -> str:
    task = read_task(path)
    flag = ""
    if state == "blocked":
        flag = f"  ⟵ {task.get('blocked_reason', 'needs human')}"
    elif state == "failed":
        flag = f"  ⟵ {task.get('failed_reason', 'failed')}"
    elif state == "active":
        flag = f"  ⟵ {task.get('agent', '?')}"
    elif state == "done":
        flag = f"  ⟵ {task.get('result', '')[:60]}"
    return (f"  {task['id']}  p{task['priority']}  {task.get('exec', 'browser'):<7} "
            f"{task['kind']:<10} {task.get('title') or task['url']}{flag}")


def cmd_list(args) -> int:
    ensure_dirs()
    states = [args.state] if args.state else list(STATES)
    for state in states:
        files = task_files(state)
        if not files and args.state is None:
            continue
        print(f"{state} ({len(files)})")
        for path in files:
            print(summarize(path, state))
    return 0


def cmd_status(args) -> int:
    ensure_dirs()
    counts = {s: len(task_files(s)) for s in STATES}
    print("  ".join(f"{s}={counts[s]}" for s in STATES))
    if counts["blocked"]:
        print(f"\n{counts['blocked']} task(s) need you — run: fleet blocked")

    # An agent that died leaves its task in active/ forever; say so rather than
    # letting the queue look busy while nothing is running.
    stale = [p for p in task_files("active")
             if time.time() - p.stat().st_mtime > STALE_AFTER]
    if stale:
        print(f"\n{len(stale)} task(s) have been active for over "
              f"{STALE_AFTER // 60} min with no update — the agent may be gone.")
        print("  recover them with: fleet reset")
    return 0


def _meminfo() -> dict[str, int]:
    """Total and available memory in MB, from the kernel rather than a guess."""
    values = {}
    try:
        for line in Path("/proc/meminfo").read_text().splitlines():
            key, _, rest = line.partition(":")
            if key in ("MemTotal", "MemAvailable"):
                values[key] = int(rest.strip().split()[0]) // 1024
    except OSError:
        pass
    return values


def _cgroup_limit_mb() -> int | None:
    """Honour a container memory limit — the host's RAM is not ours to use."""
    for path, transform in (
        ("/sys/fs/cgroup/memory.max", lambda v: None if v.strip() == "max" else int(v)),
        ("/sys/fs/cgroup/memory/memory.limit_in_bytes", int),
    ):
        try:
            raw = Path(path).read_text().strip()
            limit = transform(raw)
            if limit and limit < (1 << 62):
                return limit // (1024 * 1024)
        except (OSError, ValueError):
            continue
    return None


def _rss_mb(pid: str) -> int | None:
    try:
        for line in Path(f"/proc/{pid}/status").read_text().splitlines():
            if line.startswith("VmRSS:"):
                return int(line.split()[1]) // 1024
    except (OSError, ValueError, IndexError):
        return None
    return None


def _cmdline(pid: str) -> str:
    try:
        return Path(f"/proc/{pid}/cmdline").read_bytes().decode(errors="replace").replace("\0", " ")
    except OSError:
        return ""


def _p75(values: list[int]) -> int:
    """Plan for a heavy worker, not an average one."""
    ordered = sorted(values)
    return ordered[min(len(ordered) - 1, int(len(ordered) * 0.75))]


def measure_worker_cost() -> dict:
    """What a worker actually costs *on this machine*, from live processes.

    An agent session and a browser tab are separate processes, so both are
    sampled: agent sessions by their command line, browser tabs by the renderer
    processes a Chromium-family browser spawns per tab. Nothing is assumed about
    your hardware — with no samples yet we fall back to a deliberately
    conservative figure and say so.
    """
    agents, renderers = [], []
    try:
        pids = [entry for entry in os.listdir("/proc") if entry.isdigit()]
    except OSError:
        pids = []
    for pid in pids:
        mem = _rss_mb(pid)
        if not mem:
            continue
        cmd = _cmdline(pid)
        if "claude" in cmd and "--dangerously-skip-permissions" in cmd:
            agents.append(mem)
        elif "--type=renderer" in cmd and ("chrom" in cmd.lower() or "brave" in cmd.lower()):
            renderers.append(mem)

    FALLBACK_AGENT, FALLBACK_TAB = 500, 300
    agent_mb = _p75(agents) if agents else FALLBACK_AGENT
    tab_mb = _p75(renderers) if renderers else FALLBACK_TAB
    return {
        "agent_mb": agent_mb,
        "tab_mb": tab_mb,
        "browser_worker_mb": agent_mb + tab_mb,
        "agent_samples": len(agents),
        "tab_samples": len(renderers),
        "source": "measured" if agents or renderers else "fallback",
    }


def remember_cost(sample: dict) -> dict:
    """Blend today's measurement into what this project has seen before.

    A rolling estimate rides out a moment when nothing was running, and it lives
    beside the queue, so every machine learns its own numbers.
    """
    store = AGENTS_DIR / "machine.json"
    previous = {}
    if store.exists():
        try:
            previous = json.loads(store.read_text())
        except json.JSONDecodeError:
            previous = {}

    if sample["source"] == "fallback" and previous.get("browser_worker_mb"):
        return {**previous, "source": "remembered"}

    blended = dict(sample)
    for key in ("agent_mb", "tab_mb", "browser_worker_mb"):
        if previous.get(key):
            blended[key] = int(round(0.7 * previous[key] + 0.3 * sample[key]))
    blended["updated"] = now()
    try:
        AGENTS_DIR.mkdir(parents=True, exist_ok=True)
        store.write_text(json.dumps(blended, indent=2) + "\n")
    except OSError:
        pass
    return blended


def probe_machine() -> dict:
    """What this computer can spare right now. Nothing here is hard-coded to a
    particular machine: memory and load are read live, reserves scale with the
    hardware, and worker cost is measured from real processes."""
    cfg = config()
    cores = os.cpu_count() or 1
    try:
        load1 = os.getloadavg()[0]
    except OSError:
        load1 = 0.0

    mem = _meminfo()
    total_mb = mem.get("MemTotal", 0)
    available_mb = mem.get("MemAvailable", total_mb)
    limit_mb = _cgroup_limit_mb()
    if limit_mb:
        total_mb = min(total_mb or limit_mb, limit_mb)
        available_mb = min(available_mb or limit_mb, limit_mb)

    cost = remember_cost(measure_worker_cost())
    provenance = {}

    def setting(key: str, derived: int, how: str) -> int:
        configured = cfg.get(key)
        provenance[key] = "configured" if configured else how
        return int(configured) if configured else derived

    # Reserves scale with the machine: a 4 GB laptop must not hold back 2 GB,
    # and a 256 GB server should hold back more than that.
    reserve_mem = setting("reserve_mem_mb",
                          max(1024, int(total_mb * 0.10)) if total_mb else 1024,
                          "derived from total RAM")
    reserve_cores = setting("reserve_cores", max(1, cores // 8), "derived from core count")
    per_worker = setting("mem_per_browser_worker_mb", cost["browser_worker_mb"], cost["source"])
    per_agent = setting("mem_per_agent_task_mb", cost["agent_mb"], cost["source"])

    idle_cores = cores * cfg["max_load_per_core"] - load1
    cpu_capacity = int(max(0, min(cores - reserve_cores, idle_cores)))

    spare_mb = max(0, available_mb - reserve_mem)
    mem_capacity = int(spare_mb // max(per_worker, 1))

    capacity = min(cpu_capacity, mem_capacity)
    if capacity <= 0:
        bound, capacity = "machine is nearly full — running serially", 1
    elif mem_capacity < cpu_capacity:
        bound = "memory"
    elif cpu_capacity < mem_capacity:
        bound = "cpu/load"
    else:
        bound = "cpu and memory"

    return {
        "cores": cores, "load1": round(load1, 2),
        "total_mb": total_mb, "available_mb": available_mb,
        "containerised": bool(limit_mb),
        "reserve_mem_mb": reserve_mem, "reserve_cores": reserve_cores,
        "mem_per_browser_worker_mb": per_worker, "mem_per_agent_task_mb": per_agent,
        "cost_samples": {"agent": cost.get("agent_samples", 0),
                         "tab": cost.get("tab_samples", 0)},
        "provenance": provenance,
        "cpu_capacity": cpu_capacity, "mem_capacity": mem_capacity,
        "capacity": capacity, "bound_by": bound,
        "subagent_capacity": max(1, int(spare_mb // max(per_agent, 1))),
    }


def cmd_machine(args) -> int:
    """Show what was measured, what was derived, and what you pinned."""
    ensure_dirs()
    machine = probe_machine()
    if args.json:
        print(json.dumps(machine, indent=2))
        return 0

    prov = machine["provenance"]
    print(f"cores            {machine['cores']}   load1 {machine['load1']}")
    print(f"memory           {machine['available_mb'] / 1024:.1f} GB free "
          f"of {machine['total_mb'] / 1024:.1f} GB"
          f"{'   (container limit applies)' if machine['containerised'] else ''}")
    print()
    print(f"reserve_mem      {machine['reserve_mem_mb']:>6} MB   "
          f"[{prov['reserve_mem_mb']}]")
    print(f"reserve_cores    {machine['reserve_cores']:>6}      "
          f"[{prov['reserve_cores']}]")
    print(f"per browser wkr  {machine['mem_per_browser_worker_mb']:>6} MB   "
          f"[{prov['mem_per_browser_worker_mb']}] "
          f"from {machine['cost_samples']['agent']} session(s) + "
          f"{machine['cost_samples']['tab']} tab(s) seen")
    print(f"per subagent     {machine['mem_per_agent_task_mb']:>6} MB   "
          f"[{prov['mem_per_agent_task_mb']}]")
    print()
    print(f"→ room for {machine['capacity']} browser worker(s) "
          f"and {machine['subagent_capacity']} subagent(s), bound by {machine['bound_by']}")
    print(f"  (cpu allows {machine['cpu_capacity']}, memory allows {machine['mem_capacity']})")
    return 0


def build_plan(max_workers: int | None = None, force: bool = False) -> dict:
    """Schedule the pending queue.

    Concurrency is the *smallest* of four ceilings, so the tightest constraint
    always wins:

      tasks        never more workers than there is work
      max_workers  your configured preference
      hosts        one worker per site, so no host is hit in parallel
      machine      what this computer can actually spare right now

    Workers are a pool, not a partition: 100 tasks and 3 workers means each
    worker keeps claiming until the lane is empty. The queue length never
    decides how many browsers open.
    """
    cfg = config()
    cap = max_workers or cfg["max_workers"]
    sub_cap = cfg["max_subagents"]

    pending = [read_task(p) for p in task_files("inbox")]
    lanes = {mode: [t for t in pending if t.get("exec", "browser") == mode]
             for mode in EXEC_MODES}

    sites = {t.get("site", "") for t in lanes["browser"]}
    sites.discard("")

    machine = probe_machine() if cfg.get("respect_machine", True) else None

    ceilings = {
        "tasks": len(lanes["browser"]),
        "max_workers": cap,
        "hosts": max(len(sites), 1),
    }
    if machine and not force:
        ceilings["machine"] = machine["capacity"]

    terminals = min(ceilings.values()) if lanes["browser"] else 0
    bound_by = min(ceilings, key=lambda k: ceilings[k]) if lanes["browser"] else ""

    subagents = min(len(lanes["agent"]), sub_cap)
    if machine and not force:
        subagents = min(subagents, machine["subagent_capacity"])

    return {
        "pending": len(pending),
        "worker_cwd": str(worker_cwd()),
        "kinds": sorted({t.get("kind", "") for t in pending if t.get("kind")}),
        "terminals": terminals,
        "subagents": subagents,
        "sites": sorted(sites),
        "max_workers": cap,
        "max_subagents": sub_cap,
        "ceilings": ceilings,
        "bound_by": bound_by,
        "forced": force,
        "machine": machine,
        "lanes": {mode: len(tasks) for mode, tasks in lanes.items()},
    }


def cmd_plan(args) -> int:
    """How to run this queue. `--count` prints only the number of terminals."""
    ensure_dirs()
    plan = build_plan(args.max, force=getattr(args, "force", False))
    machine = plan["machine"]

    if args.count:
        print(plan["terminals"])
        return 0
    if args.json:
        print(json.dumps(plan, indent=2))
        return 0

    if machine:
        shape = (f"{machine['cores']} cores, load {machine['load1']}, "
                 f"{machine['available_mb'] / 1024:.1f} GB free")
        if machine["containerised"]:
            shape += " (container limit honoured)"
        print(f"machine: {shape}")
        print(f"         → room for {machine['capacity']} browser worker(s) "
              f"and {machine['subagent_capacity']} subagent(s), "
              f"bound by {machine['bound_by']}\n")

    if not plan["pending"]:
        print("nothing queued — add work first")
        return 0

    print(f"pending: {plan['pending']} task(s)\n")
    for mode in EXEC_MODES:
        count = plan["lanes"][mode]
        if not count:
            continue
        if mode == "browser":
            shown = plan["sites"][:4]
            more = len(plan["sites"]) - len(shown)
            hosts = ", ".join(shown) + (f" +{more} more" if more > 0 else "")
            extra = f"→ {plan['terminals']} terminal(s); hosts: {hosts or '—'}"
        elif mode == "agent":
            extra = f"→ {plan['subagents']} subagent(s) in parallel, no terminal"
        else:
            extra = "→ no terminal, run them one at a time"
        print(f"  {mode:<8} {count:>3} task(s)   {LANE_HELP[mode]}")
        print(f"  {'':<8} {'':>3}            {extra}")

    print(f"\nterminals needed: {plan['terminals']}")
    if plan["terminals"] == 1 and plan["lanes"]["browser"] > 1:
        print("  one at a time — this machine has no room for more right now")
    if plan["bound_by"]:
        explain = {
            "tasks": "there is no more browser work than that",
            "max_workers": f"max_workers is {plan['max_workers']} (config or --max)",
            "hosts": "one worker per host, so no site is hit in parallel",
            "machine": f"this machine can spare {plan['ceilings'].get('machine')} "
                       f"({machine['bound_by'] if machine else ''})",
        }
        print(f"  limited by {plan['bound_by']}: {explain.get(plan['bound_by'], '')}")
    if plan["lanes"]["browser"] > plan["terminals"] > 0:
        print(f"  the other {plan['lanes']['browser'] - plan['terminals']} browser task(s) "
              f"wait in the queue — each worker claims the next one as it finishes")
    if plan["forced"]:
        print("  --force: machine limits ignored for this run")
    if plan["lanes"]["agent"] or plan["lanes"]["serial"]:
        print("  the agent and serial lanes cost nothing — run them in this session:")
        print("    fleet claim orchestrator --exec agent")
    return 0


def cmd_blocked(args) -> int:
    """What a human has to clear, grouped by reason."""
    ensure_dirs()
    by_reason: dict[str, list[dict]] = {}
    for path in task_files("blocked"):
        task = read_task(path)
        by_reason.setdefault(task.get("blocked_reason", "unknown"), []).append(task)
    if not by_reason:
        print("nothing blocked")
        return 0
    for reason, tasks in by_reason.items():
        print(f"\n{reason}  ({len(tasks)})")
        for task in tasks:
            who = f"   (was {task['agent']})" if task.get("agent") else ""
            print(f"  {task['id']}  {task.get('title') or task['url']}{who}")
            print(f"          {task['url']}")
            if reason.startswith("needs answer"):
                print(f"          answer it:  fleet unblock {task['id']} --answer \"...\"")
    print("\nafter clearing these:  fleet unblock --all   (or one task: fleet unblock ID [--answer ...])")
    return 0


def cmd_unblock(args) -> int:
    ensure_dirs()
    if args.answer and args.all:
        print("--answer belongs to one task; give its id instead of --all", file=sys.stderr)
        return 2
    if args.all:
        targets = task_files("blocked")
    else:
        if not args.id:
            print("give a task id, or --all to release every blocked task",
                  file=sys.stderr)
            return 2
        path, _ = find_task(args.id)
        if path is None:
            print(f"no such task: {args.id}", file=sys.stderr)
            return 1
        targets = [path]
    for path in targets:
        task = read_task(path)
        task["agent"] = None
        task.pop("blocked_reason", None)
        # A wall a human has cleared is not a failed attempt. Without this, a task that met a
        # login wall and then a captcha reached the worker's "attempts > 2 → fail" rule having
        # done nothing wrong.
        task["attempts"] = max(0, task.get("attempts", 0) - 1)
        if args.answer:
            # The worker reads `notes` on every claim, so this is how an answer reaches it. Kept
            # as an appended line, never a replacement: the original brief must survive.
            stamp = time.strftime("%Y-%m-%d %H:%M")
            task["notes"] = (task.get("notes") or "") + f"\n[human answer {stamp}] {args.answer}"
        log(task, "unblocked", by="human", **({"answer": args.answer} if args.answer else {}))
        write_task(path, task)
        move(path, "inbox")
        print(f"{task['id']} → inbox")
    return 0


def cmd_reset(args) -> int:
    """Return stale active tasks to the inbox (e.g. after an agent died)."""
    ensure_dirs()
    moved = 0
    for path in task_files("active"):
        task = read_task(path)
        age = time.time() - path.stat().st_mtime
        if age < args.older_than:
            continue
        task["agent"] = None
        log(task, "reset", reason=f"stale for {int(age)}s")
        write_task(path, task)
        move(path, "inbox")
        print(f"{task['id']} → inbox (stale {int(age)}s)")
        moved += 1
    print(f"reset {moved} task(s)")
    return 0


def cmd_watch(args) -> int:
    ensure_dirs()
    try:
        while True:
            counts = {s: len(task_files(s)) for s in STATES}
            line = "  ".join(f"{s}={counts[s]}" for s in STATES)
            print(f"\r{datetime.now().strftime('%H:%M:%S')}  {line}   ", end="", flush=True)
            time.sleep(args.interval)
    except KeyboardInterrupt:
        print()
    return 0


# --------------------------------------------------------------------------- #
# agent commands
# --------------------------------------------------------------------------- #


def cmd_claim(args) -> int:
    """Atomically take the highest-priority task. Exit 3 when the inbox is empty.

    --exec limits the claim to one lane, so a terminal worker never picks up work
    meant for an in-session subagent (and vice versa).
    """
    ensure_dirs()
    with claim_gate():
        # ⛔ THE COLLISION CHECK AND THE RENAME MUST BE ONE CRITICAL SECTION. Checked outside a
        # gate, two workers both read "nothing in flight touches my files" and both claimed —
        # the exact time-of-check/time-of-use hole the atomic rename closes for task OWNERSHIP
        # but cannot close for the FILES a task will write. Claims are a rename, so the section
        # is microseconds; the tasks it protects run for minutes.
        busy = active_touches()
        for path in task_files("inbox"):
            try:
                peek = read_task(path)
            except (FileNotFoundError, json.JSONDecodeError):
                continue
            if args.exec and peek.get("exec", "browser") != args.exec:
                continue
            # Only the concurrent lane can collide; `serial` runs alone and `browser` writes
            # no repo files.
            if peek.get("exec") == "agent":
                mine = peek.get("touches") or []
                clash = next((f"{m} ↔ {o}" for m in mine for o in busy
                              if paths_collide(m, o)), None)
                if clash:
                    continue                # in flight elsewhere; leave it for a later claim
            dest = QUEUE / "active" / path.name
            try:
                os.rename(path, dest)      # atomic: exactly one agent wins
            except (FileNotFoundError, OSError):
                continue                    # lost the race, try the next task
            task = read_task(dest)
            task["agent"] = args.agent
            task["attempts"] = task.get("attempts", 0) + 1
            log(task, "claimed", agent=args.agent)
            write_task(dest, task)
            print(json.dumps(task, indent=2))
            return 0
    print(f"inbox empty{f' for lane {args.exec}' if args.exec else ''}", file=sys.stderr)
    return 3


def _finish(task_id: str, state: str, field: str, value: str) -> int:
    path, _ = find_task(task_id)
    if path is None:
        print(f"no such task: {task_id}", file=sys.stderr)
        return 1
    task = read_task(path)
    task[field] = value
    log(task, state, **{field: value})
    write_task(path, task)
    move(path, state)
    print(f"{task['id']} → {state}")
    return 0


def cmd_done(args) -> int:
    return _finish(args.id, "done", "result", args.result)


def cmd_fail(args) -> int:
    return _finish(args.id, "failed", "failed_reason", args.reason)


def cmd_block(args) -> int:
    path, _ = find_task(args.id)
    if path is None:
        print(f"no such task: {args.id}", file=sys.stderr)
        return 1
    task = read_task(path)
    notify(f"Agent blocked: {args.reason}",
           f"{task.get('title') or task['url']}\n{task['url']}")
    return _finish(args.id, "blocked", "blocked_reason", args.reason)


def cmd_show(args) -> int:
    path, state = find_task(args.id)
    if path is None:
        print(f"no such task: {args.id}", file=sys.stderr)
        return 1
    task = read_task(path)
    task["_state"] = state
    print(json.dumps(task, indent=2))
    return 0


# --------------------------------------------------------------------------- #


def main() -> int:
    parser = argparse.ArgumentParser(prog="fleet", description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = parser.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("init", help="make this project a fleet project")
    p.add_argument("path", nargs="?")
    p.add_argument("--max-workers", type=int)
    p.add_argument("--no-checklist", action="store_true",
                   help="do not create a starter TASKS.md")
    p.set_defaults(func=cmd_init)

    p = sub.add_parser("machine", help="what this computer can spare, and where each number came from")
    p.add_argument("--json", action="store_true")
    p.set_defaults(func=cmd_machine)

    p = sub.add_parser("where", help="show which project's queue is in scope")
    p.set_defaults(func=cmd_where)

    p = sub.add_parser(
        "from-note", help="promote checklist items into the queue",
        epilog="Config: \"checklist\" names a different file; "
               "\"checklist_user_marker\" is a line whose text ends a generated "
               "header, so only the region below it is read. Items are top-level "
               "'- [ ]' lines; indented lines below one travel with it as notes.")
    p.add_argument("--note", default="",
                   help="checklist file (default: TASKS.md / BACKLOG.md / TODO.md, "
                        "or \"checklist\" in .agents/config.json)")
    p.add_argument("--section", default="",
                   help="only this heading (default: any NOW / READY / TODO section)")
    p.add_argument("--kind", default="")
    p.set_defaults(func=cmd_from_note)

    p = sub.add_parser("to-note", help="tick finished items in the note and log them")
    p.set_defaults(func=cmd_to_note)

    p = sub.add_parser("doctor", help="prove no fleet artifact can reach the remote")
    p.add_argument("--fix", action="store_true", help="exclude and untrack what it finds")
    p.set_defaults(func=cmd_doctor)

    p = sub.add_parser("add", help="add one task (a URL, or a sentence of work)")
    p.add_argument("url", metavar="URL|TEXT",
                   help="a page to act on, or the task itself in words")
    p.add_argument("--title", default="", help="short name you will recognise later")
    p.add_argument("--note", default="", help="instructions or context for the worker")
    p.add_argument("--site", default="", help="override the host used for lane capping")
    p.add_argument("--kind", default="",
                   help="your label for the work: form-fill, research, migrate, … "
                        "(default: default_kind from .agents/config.json)")
    p.add_argument("--exec", default="", choices=("", *EXEC_MODES),
                   help="browser (needs a tab+terminal), agent (subagent), serial")
    p.add_argument("--serial", action="store_true", help="shorthand for --exec serial")
    p.add_argument("--priority", type=int, default=5,
                   help="1 runs first, 5 default, 9 last")
    p.set_defaults(func=cmd_add)

    p = sub.add_parser("add-batch", help="add tasks from a file or stdin ('URL | title | note')")
    p.add_argument("file", nargs="?", default="-")
    p.add_argument("--kind", default="")
    p.add_argument("--exec", default="", choices=("", *EXEC_MODES))
    p.add_argument("--priority", type=int, default=5)
    p.set_defaults(func=cmd_add_batch)

    p = sub.add_parser("plan", help="how to run this queue: lanes and terminals")
    p.add_argument("--max", type=int)
    p.add_argument("--count", action="store_true", help="print only the terminal count")
    p.add_argument("--json", action="store_true", help="machine-readable plan")
    p.add_argument("--force", action="store_true",
                   help="ignore what the machine can spare (can overload it)")
    p.set_defaults(func=cmd_plan)

    p = sub.add_parser("list", help="list tasks")
    p.add_argument("state", nargs="?", choices=STATES)
    p.set_defaults(func=cmd_list)

    p = sub.add_parser("status", help="one-line queue counts")
    p.set_defaults(func=cmd_status)

    p = sub.add_parser("blocked", help="what needs a human, grouped by reason")
    p.set_defaults(func=cmd_blocked)

    p = sub.add_parser("unblock", help="return blocked task(s) to the inbox")
    p.add_argument("id", nargs="?")
    p.add_argument("--all", action="store_true")
    p.add_argument("--answer", default="",
                   help="the answer to a 'needs answer: ...' block; appended to the task's notes")
    p.set_defaults(func=cmd_unblock)

    p = sub.add_parser("reset", help="return stale active tasks to the inbox")
    p.add_argument("--older-than", type=int, default=STALE_AFTER, metavar="SEC")
    p.set_defaults(func=cmd_reset)

    p = sub.add_parser("watch", help="live queue counters")
    p.add_argument("--interval", type=float, default=2.0)
    p.set_defaults(func=cmd_watch)

    p = sub.add_parser("claim", help="atomically take the next task (exit 3 = empty)")
    p.add_argument("agent")
    p.add_argument("--exec", default="", choices=("", *EXEC_MODES),
                   help="claim only from this lane")
    p.set_defaults(func=cmd_claim)

    p = sub.add_parser("done", help="mark finished, with evidence")
    p.add_argument("id")
    p.add_argument("--result", required=True, help="what was submitted + confirmation seen")
    p.set_defaults(func=cmd_done)

    p = sub.add_parser("block", help="needs a human; notifies and frees the agent")
    p.add_argument("id")
    p.add_argument("--reason", required=True, help="e.g. 'login required: example.com'")
    p.set_defaults(func=cmd_block)

    p = sub.add_parser("fail", help="give up on a task")
    p.add_argument("id")
    p.add_argument("--reason", required=True)
    p.set_defaults(func=cmd_fail)

    p = sub.add_parser("show", help="print one task as JSON")
    p.add_argument("id")
    p.set_defaults(func=cmd_show)

    args = parser.parse_args()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())
