simbiot-harness/harness/commands.py

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

Задачи подаются через CLI/Telegram/файл-инбокс; каждая команда
валидируется и только потом пишет событие.
"""
import os
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


def submit_task(title: str, details: str = "", minutes: int = 5, source: str = "cli"):
    """SubmitTask: класс по оценке длительности. Идемпотентно по id."""
    task_id = f"t-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{secrets.token_hex(3)}"
    size = size_class(minutes)
    es.append_event("task", task_id, "TaskSubmitted", {
        "id": task_id, "title": title, "details": details,
        "size": size, "est_minutes": minutes, "source": source,
    })
    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 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} не найдена")
    if t["status"] not in ("failed", "done", "running"):
        raise ValueError(f"задача {task_id} в статусе {t['status']} — requeue не нужен")
    es.append_event("task", task_id, "TaskPromoted",
                    {"to": t["size"], "from": t["size"], "reason": reason})
    return task_id


def set_policy(identity: str = None, cpu_alert_threshold: int = None):
    """PolicyChanged: идентичность + пороги. Мержим поверх текущей политики."""
    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
    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:],
    }