"""Simbiot Harness — command side (CQRS): валидация + события.

Задачи подаются через CLI/Telegram/файл-инбокс; каждая команда
валидируется и только потом пишет событие.
"""
import os
import re
import secrets
from datetime import datetime, timezone

from . import es
from .levels import LEVELS, size_class

# лестница сверху вниз: min → hour → day (sec эскалации не шлёт — рефлекс без LLM)
_ORDER = ["sec", "min", "hour", "day"]


def level_above(level: str):
    """Уровень над данным или None (day эскалирует некуда — это уже вершина)."""
    if level not in _ORDER:
        return None
    i = _ORDER.index(level)
    return _ORDER[i + 1] if i + 1 < len(_ORDER) else None


def raise_escalation(from_level: str, reason: str, context: str = "", severity: str = "info",
                     task_id: str = None, source: str = "cli"):
    """EscalationRaised: важное с нижнего уровня поднимается наверх.

    Проекция кладёт эскалацию в model['escalations'], scheduler inject'ит
    открытые в контекст верхних уровней (build_context). Идемпотентно по id.
    """
    if from_level not in _ORDER:
        raise ValueError(f"неизвестный уровень {from_level}")
    to_level = level_above(from_level)
    if to_level is None:
        raise ValueError(f"у {from_level} нет уровня выше — эскалация некуда")
    if severity not in ("info", "warn", "critical"):
        raise ValueError(f"неизвестный severity {severity} (info|warn|critical)")
    esc_id = f"e-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{secrets.token_hex(3)}"
    es.append_event("harness", esc_id, "EscalationRaised", {
        "id": esc_id, "from_level": from_level, "to_level": to_level,
        "reason": reason, "context": context[:4000], "severity": severity,
        "task_id": task_id, "source": source,
    })
    return esc_id, to_level


def acknowledge_escalation(esc_id: str, level: str = "hour", note: str = ""):
    """Закрыть эскалацию (верхний уровень принял к сведению/отработал)."""
    model = es.run_projections()
    e = (model.get("escalations") or {}).get(esc_id)
    if e is None:
        raise ValueError(f"эскалация {esc_id} не найдена")
    if e.get("status") == "acked":
        raise ValueError(f"эскалация {esc_id} уже закрыта")
    es.append_event("harness", esc_id, "EscalationAcknowledged", {
        "id": esc_id, "level": level, "note": note[:1000],
    })
    return esc_id


def list_escalations(only_open: bool = False):
    model = es.run_projections()
    out = [e for e in (model.get("escalations") or {}).values()
           if not only_open or e.get("status") == "open"]
    out.sort(key=lambda e: e.get("ts", ""), reverse=True)
    return out


# мусорные заголовки: LLM-тик нижнего уровня звал `cli.py task list` / `cli.py task <id>`,
# а argparse принял субкомандо-подобное слово / id как title → мусорные задачи 'list'
# (e-261001082241, e-261003094628). Ловим на входе command side.
_SUSPECT_TITLES = {"list", "tasks", "status", "events", "help", "complete", "promote", "requeue"}
_TASK_ID_RE = re.compile(r"^t-\d{12}-[0-9a-f]{6}$")


def _validate_task_title(title: str, details: str):
    """Guard: заголовок-субкоманда CLI или чужой id задачи без деталей — это просмотр, не подача."""
    if not title or not title.strip():
        raise ValueError("пустой заголовок задачи")
    t = title.strip()
    if not details.strip() and (t.lower() in _SUSPECT_TITLES or _TASK_ID_RE.match(t)):
        raise ValueError(
            f"подозрительный заголовок {t!r} без деталей: похоже на субкоманду CLI или id задачи "
            f"(cli.py task list / cli.py task <id> — это просмотр, не создание). "
            "Подача: cli.py task \"заголовок\" -d \"детали\"; доска: cli.py tasks"
        )


def submit_task(title: str, details: str = "", minutes: int = 5, source: str = "cli"):
    """SubmitTask: класс по оценке длительности. Идемпотентно по id."""
    _validate_task_title(title, details)
    task_id = f"t-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{secrets.token_hex(3)}"
    size = size_class(minutes)
    data = {"id": task_id, "title": title, "details": details,
            "size": size, "est_minutes": minutes, "source": source}
    # круг доверия (harness/trust.py): задача из LLM-тика наследует круг тика —
    # просьба правления потом не исполняется часом с правами владельца
    tick_circle = os.environ.get("SIMBIOT_TICK_CIRCLE", "").strip()
    if tick_circle.isdigit():
        data["circle"] = int(tick_circle)
    es.append_event("task", task_id, "TaskSubmitted", data)
    return task_id, size


def promote_task(task_id: str, reason: str = "manual promote"):
    """Промоушен ready-задачи на уровень выше (TaskPromoted → ready, size выше)."""
    order = ["S", "M", "L"]
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if t is None:
        raise ValueError(f"задача {task_id} не найдена")
    if t["status"] != "ready":
        raise ValueError(f"задача {task_id} в статусе {t['status']} — promote только из ready")
    i = order.index(t["size"]) if t["size"] in order else 0
    if i >= len(order) - 1:
        raise ValueError(f"задача {task_id} уже размера {t['size']} — выше некуда")
    new_size = order[i + 1]
    es.append_event("task", task_id, "TaskPromoted",
                    {"to": new_size, "from": t["size"], "reason": reason})
    return task_id, new_size


def close_task(task_id: str, reason: str = "manual close") -> str:
    """Закрыть failed-задачу без работы: TaskClosed → статус closed.

    Авто-схлопывание старых фейлов (просьба Alex 03.10, e-261003094432):
    дубли и прах не должны висеть красной группой на доске. closed —
    терминальный, НО обратимый: requeue принимает closed → ready.
    Идемпотентно: уже closed/done — просто возвращаем задачу (no-op).
    """
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if t is None:
        raise ValueError(f"задача {task_id} не найдена")
    if t["status"] in ("closed", "done"):
        return task_id  # уже закрыта — no-op (идемпотентность)
    if t["status"] != "failed":
        raise ValueError(
            f"задача {task_id} в статусе {t['status']} — close только для failed; "
            f"для живой работы используй requeue, для завершения — complete")
    es.append_event("task", task_id, "TaskClosed", {"by": "manual", "reason": reason[:300]})
    return task_id


def archive_stale_failed(max_age_hours: float = 24.0, now=None) -> list:
    """Демон-архиватор (e-261003094432): failed старше max_age_hours → closed.

    Зовёт scheduler-демон каждые DELIVERY_POLL: красная группа фейлов
    не растёт — старше суток уходят в «архив» (статус closed), где их
    видно при желании, но они не забивают доску. requeue возвращает
    закрытую в очередь, если она всё-таки нужна.
    now — для тестов (симуляция «прошло 24ч» без подделки событий).
    """
    from datetime import datetime, timezone
    model = es.run_projections()
    now = now or datetime.now(timezone.utc)
    closed = []
    for tid, t in (model.get("tasks") or {}).items():
        if t.get("status") != "failed":
            continue
        fin = t.get("finished") or ""
        try:
            age = (now - datetime.fromisoformat(fin)) if fin else None
        except ValueError:
            age = None
        if age is None or age.total_seconds() < max_age_hours * 3600:
            continue
        es.append_event("task", tid, "TaskClosed", {
            "by": "archive", "reason": f"auto: failed > {max_age_hours:g}ч без работы"})
        closed.append(tid)
    return closed


def block_task(task_id: str, reason: str = "ожидает уточнений") -> str:
    """Заблокировать живую задачу: TaskBlocked → статус blocked (t-261004122019).

    Задача есть, но не готова к работе — ждёт уточнений от Alex или внешнего
    события. scheduler берёт только ready → blocked не попадает в батчи.
    Обратно: requeue (blocked → ready). Идемпотентно: уже blocked — no-op.
    """
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if t is None:
        raise ValueError(f"задача {task_id} не найдена")
    if t["status"] == "blocked":
        return task_id  # уже заблокирована — no-op
    if t["status"] in ("done", "closed"):
        raise ValueError(
            f"задача {task_id} в статусе {t['status']} — blocked только для живых; "
            f"оживить можно requeue")
    es.append_event("task", task_id, "TaskBlocked",
                    {"by": "manual", "reason": reason[:300]})
    return task_id


def requeue_task(task_id: str, reason: str = "manual requeue"):
    """Вернуть failed/running/прерванную задачу в очередь (TaskPromoted → ready).

    running тоже возвращаем: если тик-владелец умер (рестарт демона, таймаут),
    TaskFailed/TaskCompleted не будет — иначе задача зависает в running навсегда.
    """
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if t is None:
        raise ValueError(f"задача {task_id} не найдена")
    # closed тоже возвращаем (авто-схлопывание старых фейлов, e-261003094432):
    # закрытая в архив задача может понадобиться снова — осознанная доработка.
    if t["status"] not in ("failed", "closed", "done", "running", "blocked"):
        raise ValueError(f"задача {task_id} в статусе {t['status']} — requeue не нужен")
    # идемпотентная доставка (t-260930083937): requeue = новая работа над
    # задачей, её результат должен уйти в чат заново → сбрасываем факт
    # доставки и захват. Старая доставка остаётся в истории событий.
    es.forget_delivery(task_id)
    es.append_event("task", task_id, "TaskPromoted",
                    {"to": t["size"], "from": t["size"], "reason": reason})
    return task_id


def mark_awaiting_delivery(task_id: str, reason: str = "", reply_to: str = None):
    """TaskAwaitDelivery: задачу заказали из TG — результат уйдёт в чат сам.

    Автодоставка артефактов (t-260929205927): когда задача, заказанная из
    TG, завершается (TaskCompleted), результат (answer/результат_ref)
    автоматически отправляется в чат — Alex не должен просить дважды.
    Доставка идемпотентна: после DeliverySent повторная не отправляется.
    Флаг живёт через requeue — доработка после неудачи доставляется тоже.
    """
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if t is None:
        raise ValueError(f"задача {task_id} не найдена")
    if t.get("await_delivery"):
        return task_id  # уже помечена — идемпотентно
    es.append_event("task", task_id, "TaskAwaitDelivery", {
        "reply_to": reply_to, "reason": (reason or "заказано из TG")[:300],
    })
    return task_id


import re as _re

# Guard (e-261001074527-f0c1b4): ночной тик 20727 пачкой закрыл 5 L-задач
# шаблонной отпиской воркера «бюджет исчерпан» — это стоп-фраза тика, а не
# результат. Такие «результаты» закрывать задачу не могут.
_BUDGET_STUB_RE = _re.compile(r"бюджет\s+(?:шагов\s+)?исчерпан", _re.IGNORECASE)
_BUDGET_STUB_MARKERS_RE = _re.compile(r"тик\s+завершён|tool-вызов|шагов", _re.IGNORECASE)


def looks_like_budget_stub(answer: str) -> bool:
    """True, если answer — шаблонная отписка воркера о бюджете шагов.

    Требование маркера («тик завершён», «tool-вызов», «шагов») отсекает
    ложные срабатывания: честный результат работы, где по делу упомянут
    исчерпанный бюджет («проверил, бюджет месяца исчерпан»), под гвард
    не попадает — а в стоп-фразе воркера маркер всегда есть.
    """
    s = answer or ""
    return bool(_BUDGET_STUB_RE.search(s) and _BUDGET_STUB_MARKERS_RE.search(s))


MIN_ANSWER = 10  # символов в итоге задачи: короче — не результат («x», «готово»)

# ────────────── гейт кода: правка харнеса = коммит + тест + зелёный тест ──────────────
# Кейс 2026-10-18 (t-261018103940): Коди сам доработал tg_inbox, но закрыл
# задачу итогом «x» ДО конца правки, без теста и без коммита. Проверка — в
# закрытии задачи (не просьба в промпте): изменения кода, сделанные после
# клейма задачи, обязаны быть закоммичены, код — с тестом, тест — зелёным.
CODE_PATHS = ("harness", "cli.py", "tests", "scripts")
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def _git(repo: str, *args) -> str:
    import subprocess
    r = subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True, timeout=30)
    return r.stdout if r.returncode == 0 else ""


def _run_test(repo: str, path: str) -> tuple:
    """(ok, хвост вывода) — тест-скрипт харнеса; FAIL-строки или падение = не ок."""
    import subprocess
    import sys as _sys
    py = os.path.join(repo, "venv", "bin", "python")
    try:
        r = subprocess.run([py if os.path.exists(py) else _sys.executable, path], cwd=repo,
                           capture_output=True, text=True, timeout=240)
    except subprocess.TimeoutExpired:
        return False, "таймаут 240 с"
    out = (r.stdout or "") + (r.stderr or "")
    fails = [ln for ln in out.splitlines() if ln.startswith("FAIL ")]
    bad = bool(fails) or "Traceback (most recent call last)" in out or r.returncode != 0
    return (not bad), "\n".join((fails or out.splitlines())[-3:])[:300]


_WRITE_RE = re.compile(r"sed\s+(?:-\S+\s+)*-i|perl\s+-\S*i|>>?|\btee\b|\b(?:cp|mv|rm|install|touch)\s|"
                       r"\bpatch\b|git\s+(?:apply|checkout|restore|am)\b|open\([^)]*['\"][wa]")
_NOISE_RE = re.compile(r"\d?>&\d|\d?>>?\s*/dev/null")


def _touched_by_trace(task: dict):
    """Правил ли тик задачи файл: предикат path → bool или None (трассы нет).

    edit/write — по пути во входе; shell — путь в команде И признак записи в ней
    (sed -i, >, tee, cp/mv/rm, patch, git apply, open(…,'w')). Чтение (grep, cat,
    sed -n) и запуск (python cli.py …) — не правка: кейс 18.10 — «cli.py» есть
    в каждой команде тика, а чужая правка cli.py блокировала бы закрытие."""
    sess = task.get("session_id") or ""
    if not sess.startswith("s-"):
        return None
    try:
        db = es._conn()
        try:
            rows = db.execute("SELECT data FROM events WHERE event_type='ToolCalled' "
                              "AND json_extract(data,'$.trace_id')=?", (sess[2:],)).fetchall()
        finally:
            db.close()
    except Exception:
        return None
    if not rows:
        return None
    import json as _json
    edits, writes = [], []
    for (d,) in rows:
        d = _json.loads(d)
        inp = d.get("input") or ""
        try:
            obj = _json.loads(inp)
        except ValueError:
            obj = {}
        if d.get("tool") in ("edit", "write"):
            edits.append(str(obj.get("path") or obj.get("file_path") or inp))
        elif d.get("tool") == "shell":
            cmd = str(obj.get("command") or inp)
            if _WRITE_RE.search(_NOISE_RE.sub(" ", cmd)):
                writes.append(cmd)

    def touched(path: str) -> bool:
        base = os.path.basename(path)
        if any(e.endswith(path) or e.endswith("/" + base) or e == base for e in edits):
            return True
        return any(path in c or re.search(rf"(?<![\w./-]){re.escape(base)}\b", c) for c in writes)
    return touched


def code_change_check(task: dict, no_test_reason: str = None):
    """(ok, причина отказа, сводка для итога) по правкам кода во время задачи."""
    repo = os.environ.get("SIMBIOT_CODE_REPO", "").strip() or _REPO_ROOT
    since = task.get("claimed_at")
    if (os.environ.get("SIMBIOT_CODE_GATE", "on").lower() == "off" or not since
            or not os.path.isdir(os.path.join(repo, ".git"))):
        return True, "", ""
    since_ts = datetime.fromisoformat(since).timestamp()
    # что тик задачи реально трогал: входы его инструментов (trace = session без «s-»).
    # Без этого чужая незакоммиченная правка того же времени (кейс 18.10:
    # scripts/deploy-claude-code.sh) блокировала бы закрытие навсегда.
    touched = _touched_by_trace(task)
    dirty = []
    for ln in _git(repo, "status", "--porcelain", "--", *CODE_PATHS).splitlines():
        path = ln[3:].strip().split(" -> ")[-1].strip('"')
        full = os.path.join(repo, path)
        if touched is not None and not touched(path):
            continue  # тик этот файл не правил — чужая правка
        try:
            if os.path.getmtime(full) >= since_ts - 5:
                dirty.append(path)
        except OSError:
            dirty.append(path)  # удалён после клейма — тоже правка
    if dirty:
        return False, ("в коде есть незакоммиченные изменения, сделанные в этой задаче: " + ", ".join(dirty[:8])
                       + " — закоммить их (git add <файлы> && git commit) вместе с тестом или "
                         "откати; потом закрывай задачу"), ""
    # --since у git — с точностью до секунды: коммит той же секунды, но ДО клейма
    # попадал в задачу; точный фильтр — по времени коммита (%ct)
    log = _git(repo, "log", f"--since={since}", "--format=@%h %ct", "--name-only", "--", *CODE_PATHS)
    commits, files, take = [], set(), False
    for ln in log.splitlines():
        if ln.startswith("@"):
            h, ct = ln[1:].split()
            take = int(ct) > int(since_ts)  # секунда клейма — ещё не работа задачи
            if take:
                commits.append(h)
        elif ln.strip() and take:
            files.add(ln.strip())
    if not commits:
        return True, "", ""
    code = sorted(f for f in files if f.endswith(".py") and (f.startswith("harness/") or f == "cli.py"))
    tests = sorted(f for f in files if re.match(r"tests/test_\w+\.py$", f)
                   and os.path.exists(os.path.join(repo, f)))
    if code and not tests and not no_test_reason:
        return False, ("изменён код (" + ", ".join(code[:6]) + ") без теста: добавь/обнови "
                       "tests/test_*.py, закоммить и закрывай снова; если тест невозможен — "
                       "complete --no-test \"причина\" (причина уйдёт Alex)"), ""
    results = []
    for tf in tests:
        ok, tail = _run_test(repo, tf)
        if not ok:
            return False, f"тест {tf} не проходит: {tail} — почини и закрывай снова", ""
        results.append(tf)
    note = "Код: " + ", ".join(commits[:6])
    if results:
        note += " · тесты PASS: " + ", ".join(results[:4])
    if no_test_reason and code and not tests:
        note += f" · без теста: {no_test_reason[:200]}"
    return True, "", note


def complete_task_with_delivery(task_id: str, answer: str, level: str,
                                 result_ref: str = None, no_test_reason: str = None):
    """TaskCompleted + немедленная доставка результата в TG-чат.

    Используется исполнителем (часовой/дневной тик) как атомарная замена
    голому TaskCompleted для задач с await_delivery: завершение и отправка
    идут парой, «сделал и промолчал» больше не случается. Для задач без
    флага доставки — только TaskCompleted (поведение прежнее).

    Возвращает (task_id, delivered: bool).
    """
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if t is None:
        raise ValueError(f"задача {task_id} не найдена")
    if t["status"] != "running":
        raise ValueError(f"задача {task_id} в статусе {t['status']} — завершать можно только running")
    if (answer or "").strip() == "":
        raise ValueError("пустой результат — доставлять нечего, заверши как failed/requeue")
    # мусорный итог (2026-10-18: задача закрыта с answer «x» — Alex получил в чат
    # «✅ Готово … x»; раньше — «2», «готово»): результат должен говорить, что сделано
    if len((answer or "").strip()) < MIN_ANSWER:
        raise ValueError(
            f"результат «{answer.strip()}» слишком короткий (< {MIN_ANSWER} симв.): опиши, что "
            "сделано и где результат (файл/ссылка/вывод) — это уйдёт Alex в чат")
    if looks_like_budget_stub(answer):
        requeue_task(task_id, reason="guard: 'бюджет исчерпан' — не результат, задача в ready (инцидент e-261001074527)")
        raise ValueError(
            f"задача {task_id} ПЕРЕВОДЖЕНА в ready: отпиской «бюджет исчерпан» "
            f"задачу закрывать нельзя. Либо доделай и завершай с реальным "
            f"результатом, либо явно requeue/failed.")
    ok, why, note = code_change_check(t, no_test_reason)
    if not ok:
        raise ValueError(f"задача {task_id} не закрыта: {why}")
    if note:
        answer = f"{answer.rstrip()}\n\n{note}"
    # TaskCompleted — независимо от доставки (событие-истина о завершении)
    es.append_event("task", task_id, "TaskCompleted", {
        "result_ref": result_ref, "answer": answer[:2000],
    })
    delivered = deliver_result(task_id, answer, owner=f"complete:{level}")
    return task_id, delivered


def deliver_result(task_id: str, answer: str, owner: str = "tick") -> bool:
    """Доставка результата готовой задачи в TG-чат (единственный путь).

    Идемпотентность (t-260930083937) — ТРИ барьера, гонка уровней закрыта:
    1) delivery_recorded(task_id) — артефакт уже доставлен → SKIP (событие
       DeliverySkipped с причиной, без отправки);
    2) delivery_claim(task_id) — атомарный захват ДО отправки: второй
       отправитель (параллельный тик/демон) получает False и пропускает;
    3) record_delivery(task_id) — ПОСЛЕ успешной отправки; INSERT OR IGNORE
       гарантирует, что даже два параллельных успеха зафиксируются один раз,
       а второй sender увидит recorded и не отправит повторно.

    Порядок барьеров: (ledger) → (claim) → отправка → (ledger-write) →
    событие DeliverySent. Между claim и record_delivery дубль невозможен;
    сбой отправки → release лока, демон досыплет на след. проходе.

    Три сценария использования, все через эту функцию:
    1) push в тике сразу после TaskCompleted (scheduler.run_level_tick);
    2) complete_task_with_delivery — то же для исполнителя вручную;
    3) pull-страховка deliver_pending (демон): тик умер между
       завершением и отправкой — досылает по read model.
    Пустой answer доставляется как краткое уведомление о готовности.
    Возвращает True, только если отправка выполнена в ЭТОМ вызове.
    """
    model = es.run_projections()
    t = model["tasks"].get(task_id)
    if not t or not t.get("await_delivery"):
        return False
    if t.get("status") != "done":
        return False  # доставка только готового результата
    # барьер 1: уже доставлено (ledger — истина поверх stale read model:
    # модель могла не догнать события, а ledger атомарен). Пропуск
    # логируем событием — «уже доставлено» видно в истории задачи.
    if es.delivery_recorded(task_id):
        es.append_event("task", task_id, "DeliverySkipped", {
            "task_id": task_id, "reason": "already-delivered",
        })
        return False
    if t.get("delivery_status") == "sent":
        # модель говорит «отправлено», но ledger пуст — доставка из ДО
        # фикса (легаси-задачи): молча пропускаем, без события (демон с
        # pre-filter'ом сюда не попадает, повторов нет)
        return False
    # барьер 2: атомарный захват права отправки ДО отправки —
    # параллельный отправитель (мин/час/демон) здесь не пройдёт
    if not es.delivery_claim(task_id, owner=f"{owner}:{os.getpid()}"):
        # кто-то уже отправляет: не мешаем и не считаем дублем — просто ждём
        return False
    try:
        from . import tg_inbox
        text = _delivery_text(t, answer)
        try:
            tg_ok = tg_inbox.notify_chat(text)
        except Exception:
            tg_ok = None  # TG недоступен — событие не пишем, pull повторит
        if tg_ok is None:
            # TG недоступен: освобождаем захват — демон досыплет на след.
            # проходе (15с), а не будет ждать истечения TTL лока
            es.delivery_release(task_id)
            return False
        # барьер 3: фиксируем факт доставки (первый раз — True) и событие
        es.record_delivery(task_id, kind="task")
        es.append_event("task", task_id, "DeliverySent", {
            "task_id": task_id, "text": text[:2000], "tg_ok": bool(tg_ok),
        })
        return True
    except Exception:
        # сбой отправки: освобождаем захват — демон досыплет на след. проходе
        es.delivery_release(task_id)
        raise


def _delivery_text(task: dict, answer: str) -> str:
    """Формат уведомления о готовности: что готово + выжимка результата."""
    title = (task.get("title") or "задача").strip()
    a = (answer or "").strip()
    if not a:
        return f"✅ Готово: «{title[:80]}» (завершено; детали — на дашборде)"
    if len(a) > 1800:
        a = a[:1800] + "…"
    return f"✅ Готово: «{title[:80]}»\n\n{a}"


def deliver_artifact(artifact_id: str, kind: str, send, owner: str = "manual",
                     caption: str = "", force: bool = False, msg: str = None) -> bool:
    """Идемпотентная доставка ПРОИЗВОЛЬНОГО артефакта в чат (t-260930083937).

    Рут-кейс дублей: мин-тик шёл файлом «скинь сюда» curl'ом sendDocument —
    без метки delivered каждый следующий тик видел просьбу в истории чата
    и слал заново (Alex получил презентацию трижды). Здесь тот же барьер,
    что и у задач: ledger по artifact_id → атомарный claim → отправка →
    запись. Параллельные тики (мин/час) и повторные попытки не дублируют.

    artifact_id — стабильный идентификатор вещи (ключ dedup): путь файла,
    «task:<id>», «file:<path>», «page:<url>» и т.п. force=True — повторная
    доставка (игнорирует ledger, но всё равно атомарна: параллель не
    отправит дважды). send — callable без аргументов, возвращает True
    при успехе (уже знает получателя/текст — tg_inbox.* или мок в тестах).
    Возвращает True, только если отправка выполнена в ЭТОМ вызове.
    """
    if not force and es.delivery_recorded(artifact_id):
        es.append_event("harness", artifact_id, "DeliverySkipped", {
            "artifact_id": artifact_id, "kind": kind,
            "reason": "already-delivered",
        })
        return False
    # Анти-гонка уровней (t-261009160906): артефакт привязан к входящему
    # m-NNN, на которое уже отвечено ДРУГИМ артефактом → SKIP. Два уровня
    # (min/hour) на один запрос не должны слать два разных файла; dedup по
    # artifact_id=file:<путь> их не различает (разные пути = разные ключи).
    if msg and not force:
        model = es.run_projections()
        m = (model.get("inbox") or {}).get(msg)
        if (m is not None and m.get("replied")
                and m.get("reply_artifact") not in (None, "", artifact_id)):
            es.append_event("harness", artifact_id, "DeliverySkipped", {
                "artifact_id": artifact_id, "kind": kind,
                "reason": "msg-already-answered-by-other-artifact",
                "msg": msg, "reply_artifact": m.get("reply_artifact"),
            })
            return False
    if force:
        # ручной повтор — явное решение человека: лок недавней доставки
        # не должен блокировать его на TTL (120с). Снимаем свой же захват
        # и берём заново. Параллельная гонка force vs force всё равно
        # атомарна: один claim заберёт право, второй увидит False.
        es.delivery_release(artifact_id)
    if not es.delivery_claim(artifact_id, owner=f"{owner}:{os.getpid()}"):
        return False  # параллельный отправитель уже в полёте
    try:
        ok = bool(send())
    except Exception:
        es.delivery_release(artifact_id)
        raise
    if not ok:
        # отправка не удалась: освобождаем захват под ретрай
        es.delivery_release(artifact_id)
        return False
    # force не перезатирает чужую запись: INSERT OR IGNORE — первая доставка
    # в ledger остаётся фактом; повторный force просто отправит ещё раз,
    # но ledger при этом не «молодеет» (kind/ts не переписываются)
    es.record_delivery(artifact_id, kind=kind)
    ev = {
        "artifact_id": artifact_id, "kind": kind, "tg_ok": True,
        "caption": (caption or "")[:500], "owner": owner,
    }
    if msg:
        ev["msg"] = msg  # анти-гонка уровней (t-261009160906)
    es.append_event("harness", artifact_id, "ArtifactDelivered", ev)
    return True


def set_policy(identity: str = None, cpu_alert_threshold: int = None,
               events_context: str = None):
    """PolicyChanged: идентичность + пороги. Мержим поверх текущей политики.

    events_context (t-261017110430): 'on' — в контекст тика добавляется секция
    свежих содержательных событий EventStore; 'off' — секции нет. Это и есть
    «переключатель подачи событий EventStore в контекст», одобренный Alex.
    """
    model = es.run_projections()
    current = dict(model.get("policy") or {})
    patch = {}
    if identity is not None:
        patch["identity"] = identity
    if cpu_alert_threshold is not None:
        patch["cpu_alert_threshold"] = cpu_alert_threshold
    if events_context is not None:
        patch["events_context"] = events_context
    current.update(patch)
    es.append_event("harness", "policy", "PolicyChanged", current)
    return current


def list_tasks():
    model = es.run_projections()
    out = []
    for t in model["tasks"].values():
        out.append(t)
    return out


def status():
    model = es.run_projections()
    open_escalations = sum(1 for e in (model.get("escalations") or {}).values()
                           if e.get("status") == "open")
    return {
        "cycle_state": model["cycle_state"],
        "tasks_ready": sum(1 for t in model["tasks"].values() if t["status"] == "ready"),
        "tasks_done": sum(1 for t in model["tasks"].values() if t["status"] == "done"),
        "digests": len(model["digests"]),
        "escalations_open": open_escalations,
        "spend": model.get("spend", [])[-8:],
    }