leadership_common.py

"""Shared helpers for the AI leadership layer (CEO/COO + CTO/CFO/CMO/CRO chiefs).

Factored out of the chief-agent tick pattern so multiple chief scripts don't
duplicate the same utc/log/json/doctrine/cursor-prompt/hands-delegation
plumbing.

Guardrail contract every importer must honor (never a new lever beyond
what's already here):
  - KPI nudges only via `goal_engine.py`'s `clamp_kpi_target()` /
    `apply_proposals()` -- imported as a library, never duplicated or
    bypassed.
  - Never write the founder's real plan file (`REAL_PLAN_PATH` below).
  - Never emit or honor a self-authored `APPROVED:` line -- only the
    founder's hand-written line on the proposed-plan file counts.
  - Sub-agent delegation only via `enqueue_hands_task()` below, onto the
    existing `hands/scratch/hands_queue/pending/` mechanism that
    `hands/agent_runtime.py` already drains -- never a second queue.
  - Every HITL force trigger (money, credentials, kill-switch, policy
    promote, etc. -- define your own force-trigger list) stays
    founder-only regardless of which chief is calling in.
"""

from __future__ import annotations

import json
import os
import re
import uuid
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(os.environ.get("PHANTOM_BOX_ROOT", Path.home() / "phantom-box"))

# Canonical plan paths -- never written by anything in this module.
REAL_PLAN_PATH = ROOT / "strategy" / "CEO" / "Yearly-Plan.md"
PROPOSED_PLAN_PATH = ROOT / "strategy" / "CEO" / "AI-CEO-Proposed-Plan.md"

# Sub-agent delegation target -- the *existing* hands queue, not a new one.
HANDS_QUEUE_PENDING = ROOT / "hands" / "scratch" / "hands_queue" / "pending"

# A chief/director may propose at most this many hands-queue delegations per
# tick -- bounded, well-scoped, never a bulk dispatch.
MAX_HANDS_DELEGATIONS_PER_TICK = 1

APPROVED_RE = re.compile(r"^APPROVED:\s*\d{4}-\d{2}-\d{2}\s*$", re.MULTILINE | re.IGNORECASE)


def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%SZ")


def et_now() -> str:
    try:
        from zoneinfo import ZoneInfo

        return datetime.now(ZoneInfo("America/New_York")).strftime("%Y-%m-%d %H:%M %Z")
    except Exception:
        return utc_now() + " (UTC fallback)"


def append_log(path: Path, obj: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as f:
        f.write(json.dumps({"ts": utc_now(), **obj}, ensure_ascii=False) + "\n")


def load_json(path: Path) -> dict:
    if not path.is_file():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except (OSError, json.JSONDecodeError):
        return {}


def save_json(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")


def read_jsonl(path: Path, limit: int = 2000) -> list[dict]:
    if not path.is_file():
        return []
    rows: list[dict] = []
    try:
        lines = path.read_text(encoding="utf-8").splitlines()[-limit:]
    except OSError:
        return []
    for line in lines:
        line = line.strip()
        if not line:
            continue
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(row, dict):
            rows.append(row)
    return rows


def doctrine_excerpt(paths: tuple[Path, ...], limit: int = 3500) -> str:
    """Same pattern as ceo_agent_tick.py's _doctrine_excerpt -- best-effort,
    silently skips missing/unreadable files, never a hard requirement."""
    chunks: list[str] = []
    remaining = limit
    for path in paths:
        if not path.is_file() or remaining <= 0:
            continue
        try:
            text = path.read_text(encoding="utf-8")[: min(1200, remaining)]
        except OSError:
            continue
        chunks.append(f"### {path.name}\n{text}")
        remaining -= len(text)
    return "\n\n".join(chunks)


def approved_proposed_plan_excerpt(limit: int = 2500) -> str | None:
    """The AI-CEO-Proposed-Plan.md body, but ONLY once the founder's own
    `APPROVED: YYYY-MM-DD` line is present on it.

    Chiefs read the *approved* plan as doctrine context, same pattern used
    for reading any other strategic-priorities doctrine file. Returns None
    while the plan is still PROPOSED so an unapproved AI draft can never
    quietly steer a chief's KPI nudges -- matches the CEO Agent's own
    guardrail that it never self-approves.
    """
    if not PROPOSED_PLAN_PATH.is_file():
        return None
    try:
        text = PROPOSED_PLAN_PATH.read_text(encoding="utf-8")
    except OSError:
        return None
    if not APPROVED_RE.search(text):
        return None
    return text[:limit]


def cursor_prompt_text(prompt: str, *, caller: str, tier: str = "strategic") -> str | None:
    """Cursor SDK shape-4a call, or None if unavailable/failed. Caller decides
    what "blocked_on_cursor" reporting looks like -- this never fabricates.

    `tier` defaults to "strategic" (cursor_bridge.py's best-available/auto
    model). Pass tier="routine" for high-frequency, low-stakes calls -- see
    cursor_bridge.py's MODEL_TIERS.
    """
    import sys

    sys.path.insert(0, str(Path(__file__).resolve().parent))
    try:
        from cursor_bridge import CursorBridgeError, cursor_available, cursor_prompt  # noqa: WPS433
    except ImportError:
        return None
    if not cursor_available():
        return None
    try:
        result = cursor_prompt(prompt, caller=caller, tier=tier)
    except CursorBridgeError:
        return None
    return (result.text or "").strip() or None


def parse_cursor_json(text: str) -> dict | None:
    """Tolerant JSON extraction from a model response: strip accidental
    markdown fences, else grab the first {...} span."""
    text = (text or "").strip()
    if text.startswith("```"):
        text = text.strip("`")
        if text.startswith("json"):
            text = text[4:].strip()
    try:
        data = json.loads(text)
    except json.JSONDecodeError:
        start, end = text.find("{"), text.rfind("}")
        if start >= 0 and end > start:
            try:
                data = json.loads(text[start : end + 1])
            except json.JSONDecodeError:
                return None
        else:
            return None
    return data if isinstance(data, dict) else None


def enqueue_hands_task(
    prompt: str,
    *,
    chief: str,
    tag: str,
    caller: str | None = None,
    model: str = "auto",
    task_class: str = "standard",
) -> dict:
    """Enqueue ONE bounded sub-task onto the existing hands queue.

    This does not build a second sub-agent mechanism -- it drops a task file
    into the same `hands/scratch/hands_queue/pending/` directory that
    `hands/agent_runtime.py`'s already-deployed worker polls and drains.
    That worker should apply the exact same HITL force triggers (money,
    creds, kill-switch, policy-promote, brand-critical publish, etc.) to
    whatever the resulting Agent.prompt() run proposes -- delegation never
    grants a chief a new bypass, it only means the drafting/analysis step
    itself runs as a queued task instead of inline in the chief's own tick.

    `chief` / `tag` are recorded on the task file so agent_runtime.py's log
    + done/failed records carry the authorizing chief through -- see the
    minimal, additive schema extension in hands/agent_runtime.py.

    `task_class` defaults to "standard" -- the base token ceiling. Pass
    task_class="research" for genuinely multi-tool-call work (real
    multi-repo skim/clone, full-site crawl/audit) that legitimately needs
    the higher ceiling agent_runtime.py's TASK_CLASS_TOKEN_CEILINGS
    defines -- this only widens the *token* ceiling checked post-hoc
    against a real completed run; it grants no new tool, no new HITL
    bypass, and no cost-based control (the SDK has no cost field).
    """
    HANDS_QUEUE_PENDING.mkdir(parents=True, exist_ok=True)
    task_id = f"{chief}-{uuid.uuid4().hex[:12]}"
    payload = {
        "id": task_id,
        "prompt": prompt,
        "caller": caller or f"{chief}_agent_tick",
        "authorized_by": chief,
        "tag": tag,
        "model": model,
        "task_class": task_class,
        "created_at": utc_now(),
    }
    (HANDS_QUEUE_PENDING / f"{task_id}.json").write_text(
        json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8"
    )
    return payload