goal_engine.py

"""Goal Engine — yearly plan (plain English) → loop KPI targets.

Ticked alongside your own heal cadence — no new cron surface required.

Guardrail (non-negotiable): this script can only ever raise/lower a loop's
existing KPI *target* number, clamped to a small delta, and never touches a
hard-stop ($/token budget, spend cap), governance policy, the HITL trigger
list, or approves/denies anything. If a goal would genuinely need a bigger
hard-stop, that's surfaced as a *recommendation* in the log only — a human
raises the real ceiling, this script never does.

Two honest gaps, neither faked:
  1. The founder's yearly plan file (``PLAN_PATH`` below) doesn't exist
     yet — it's the founder's plain-English yearly plan, not an agent
     deliverable. Until it exists with real content (not placeholder
     boilerplate), this reports insufficient_data and changes nothing.
  2. The structuring pass (plain English → context/company_goals.yaml) is
     a Cursor SDK call via ``cursor_bridge.py``. Until CURSOR_API_KEY is
     provisioned, this reports that gap explicitly too instead of guessing
     with a "good enough" regex parser against a founder's actual strategy
     text.

KPI target overrides land in context/loop_kpi_overrides.json — a new,
small, human-readable file a loop's own KPI-target lookup can check before
falling back to its own hardcoded default. Which loops are wired (i.e. can
have their KPI target nudged) is config-driven via ``_LOOP_CEILINGS``
below — add your own loops there, using each loop's own current
``KPI["target"]`` default as the ceiling anchor, not an invented number.

Usage:
    python3 goal_engine.py --dry-run   # shows proposed deltas, no write
    python3 goal_engine.py             # writes + logs
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(os.environ.get("PHANTOM_BOX_ROOT", Path.home() / "phantom-box"))
PLAN_PATH = ROOT / "strategy" / "CEO" / "Yearly-Plan.md"
GOALS_YAML = ROOT / "context" / "company_goals.yaml"
GOAL_LOG = ROOT / "logs" / "dogfood" / "goal_engine.jsonl"
METRICS_PATH = ROOT / "context" / "active_loop_metrics.md"
OVERRIDES_PATH = ROOT / "context" / "loop_kpi_overrides.json"

# A goal's mapped KPI delta is clamped to this range regardless of what a
# structuring pass proposes — protects against a bad completion (or a bad
# heuristic) silently blowing up a loop's target by 100x.
MAX_TARGET_DELTA_PCT = 0.50  # +/- 50% of current target, per tick

PLACEHOLDER_MARKERS = (
    "Placeholder — fill this in",
    "_[Fill]_",
)


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


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


def plan_status() -> dict:
    """Real, honest read of whether the founder has actually written a plan yet."""
    if not PLAN_PATH.is_file():
        return {"exists": False, "has_real_content": False, "reason": f"{PLAN_PATH} does not exist"}
    try:
        text = PLAN_PATH.read_text(encoding="utf-8")
    except OSError as exc:
        return {"exists": True, "has_real_content": False, "reason": f"unreadable: {exc}"}
    if any(marker in text for marker in PLACEHOLDER_MARKERS):
        return {"exists": True, "has_real_content": False, "reason": "still contains placeholder boilerplate"}
    if len(text.strip()) < 80:
        return {"exists": True, "has_real_content": False, "reason": "too short to be a real plan"}
    return {"exists": True, "has_real_content": True, "reason": "", "text": text}


def structure_plan_via_cursor(plan_text: str) -> dict | None:
    """Cursor SDK structuring pass, plain English -> OKR list. None if not configured/failed —
    caller reports that gap explicitly rather than falling back to a guessed parse of a
    founder's actual strategy text."""
    try:
        sys.path.insert(0, str(Path(__file__).resolve().parent))
        from cursor_bridge import CursorBridgeError, cursor_available, cursor_prompt  # noqa: WPS433
    except ImportError:
        return None
    if not cursor_available():
        return None

    known_loops = ", ".join(sorted(_LOOP_CEILINGS)) or "(none wired yet)"
    prompt = (
        "Read this yearly plan (plain English) and output ONLY a JSON array of OKRs, "
        "each item: {\"goal\": str, \"metric\": str, \"target_value\": str, \"target_date\": str, "
        f"\"loop\": str (best-guess department loop name from: {known_loops})}}. "
        "No prose, no markdown fences, JSON array only.\n\n"
        f"PLAN:\n{plan_text[:4000]}"
    )
    try:
        result = cursor_prompt(prompt, caller="goal_engine_tick")
    except CursorBridgeError as exc:
        _append_log({"event": "cursor_structuring_failed", "error": str(exc)[:200]})
        return None
    try:
        okrs = json.loads(result.text.strip())
    except json.JSONDecodeError:
        _append_log({"event": "cursor_structuring_unparseable", "raw_len": len(result.text)})
        return None
    if not isinstance(okrs, list):
        return None
    return {"generated_at": _utc_now(), "source": "cursor_structuring_pass", "okrs": okrs}


def _write_goals_yaml(payload: dict) -> None:
    GOALS_YAML.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        f"# Auto-generated by goal_engine.py — {payload.get('generated_at')}",
        f"# source: {payload.get('source')}",
        "okrs:",
    ]
    for okr in payload.get("okrs", []):
        lines.append(f"  - goal: {json.dumps(okr.get('goal', ''))}")
        lines.append(f"    metric: {json.dumps(okr.get('metric', ''))}")
        lines.append(f"    target_value: {json.dumps(str(okr.get('target_value', '')))}")
        lines.append(f"    target_date: {json.dumps(str(okr.get('target_date', '')))}")
        lines.append(f"    loop: {json.dumps(okr.get('loop', ''))}")
    GOALS_YAML.write_text("\n".join(lines) + "\n", encoding="utf-8")


# loop key -> that loop's own KPI['target'] ceiling anchor. Add your own
# department loops here, one entry each, using that loop's own current
# KPI["target"] default as the ceiling — never an invented number. A goal
# can only move a target up to MAX_TARGET_DELTA_PCT away from this anchor.
# The two entries below are illustrative examples matching the shape used
# by `learning/outcome_tracker.py`'s example config.
_LOOP_CEILINGS: dict[str, int] = {
    "example_content": 3,
    "example_outbound": 3,
}

# Actions that may write a KPI override (library consumers: COO Agent, etc.).
_WRITABLE_ACTIONS = frozenset({"raise_kpi_target", "lower_kpi_target", "set_kpi_target"})


def load_overrides() -> dict:
    """Public library helper — same file Goal Engine / COO Agent write through."""
    if not OVERRIDES_PATH.is_file():
        return {}
    try:
        return json.loads(OVERRIDES_PATH.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}


def _load_overrides() -> dict:
    return load_overrides()


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


def current_kpi_target(loop: str) -> int | None:
    """Effective target: override if present, else wired default, else None."""
    overrides = load_overrides()
    entry = overrides.get(loop)
    if isinstance(entry, dict) and "target" in entry:
        try:
            return int(entry["target"])
        except (TypeError, ValueError):
            pass
    if loop in _LOOP_CEILINGS:
        return int(_LOOP_CEILINGS[loop])
    return None


def clamp_kpi_target(loop: str, proposed: int, *, current: int | None = None) -> dict:
    """Single source of truth for the ±MAX_TARGET_DELTA_PCT clamp.

    Used by Goal Engine and Layer 10 COO Agent — do not duplicate this math.
    """
    if loop not in _LOOP_CEILINGS:
        return {
            "loop": loop,
            "action": "skipped_not_wired",
            "reason": f"goal_engine_tick.py has no KPI-target wiring for loop '{loop}' yet",
            "proposed": proposed,
        }
    base = current if current is not None else current_kpi_target(loop)
    if base is None:
        base = _LOOP_CEILINGS[loop]
    max_delta = max(1, int(base * MAX_TARGET_DELTA_PCT))
    lo = max(1, base - max_delta)
    hi = base + max_delta
    clamped = max(lo, min(int(proposed), hi))
    if clamped > base:
        action = "raise_kpi_target"
    elif clamped < base:
        action = "lower_kpi_target"
    else:
        action = "set_kpi_target"
    return {
        "loop": loop,
        "action": action,
        "from": base,
        "to": clamped,
        "proposed_raw": int(proposed),
        "clamp_max_delta": max_delta,
    }


def propose_kpi_deltas(goals: dict) -> list[dict]:
    """Map each OKR to a clamped KPI-target delta for a loop this engine can act on.

    Read-only computation — callers decide whether to actually write the
    override. Loops not in _LOOP_CEILINGS are reported, not silently
    dropped, so a human can see "goal engine wanted to touch a loop it
    isn't wired for yet."
    """
    proposals: list[dict] = []
    for okr in goals.get("okrs", []):
        loop = str(okr.get("loop") or "").strip()
        if not loop:
            continue
        if loop not in _LOOP_CEILINGS:
            proposals.append(
                {
                    "loop": loop,
                    "goal": okr.get("goal"),
                    "action": "skipped_not_wired",
                    "reason": f"goal_engine_tick.py has no KPI-target wiring for loop '{loop}' yet",
                }
            )
            continue
        current = current_kpi_target(loop) or _LOOP_CEILINGS[loop]
        # Conservative default nudge until real target_value parsing exists —
        # raise by 1 (clamped), never invent a big jump from a fuzzy string.
        clamped = clamp_kpi_target(loop, current + 1, current=current)
        clamped["goal"] = okr.get("goal")
        proposals.append(clamped)
    return proposals


def apply_proposals(
    proposals: list[dict],
    *,
    dry_run: bool,
    set_by: str = "goal_engine_tick",
    log_fn=None,
) -> int:
    """Write clamped KPI overrides. ``log_fn`` lets COO Agent log to its own jsonl."""
    logger = log_fn or (lambda obj: _append_log(obj))
    overrides = load_overrides()
    applied = 0
    for p in proposals:
        action = p.get("action")
        if action not in _WRITABLE_ACTIONS:
            logger({"event": "proposal_skipped", **p})
            continue
        if not dry_run:
            overrides[p["loop"]] = {
                "target": p["to"],
                "reason": p.get("goal") or p.get("reason"),
                "set_by": set_by,
                "set_at": _utc_now(),
            }
            applied += 1
        logger({"event": "kpi_target_proposed" if dry_run else "kpi_target_applied", **p, "set_by": set_by})
    if not dry_run and applied:
        _save_overrides(overrides)
    return applied


def _update_metrics_footer(status_line: str) -> None:
    if not METRICS_PATH.is_file():
        return
    try:
        text = METRICS_PATH.read_text(encoding="utf-8")
    except OSError:
        return
    placeholder = "Overnight OKR digests write here when Phase 3+ loops go live."
    if placeholder in text:
        text = text.replace(placeholder, status_line)
    elif re.search(r"^Goal Engine:.*$", text, flags=re.MULTILINE):
        text = re.sub(r"^Goal Engine:.*$", status_line, text, flags=re.MULTILINE)
    else:
        return
    try:
        METRICS_PATH.write_text(text, encoding="utf-8")
    except OSError:
        pass


def run(*, dry_run: bool = False) -> dict:
    plan = plan_status()
    if not plan.get("has_real_content"):
        report = {
            "status": "insufficient_data",
            "reason": plan.get("reason"),
            "plan_path": str(PLAN_PATH),
        }
        _append_log({"event": "no_plan", **report})
        _update_metrics_footer(
            f"Goal Engine: insufficient_data — {plan.get('reason')} ({PLAN_PATH.name} is the founder's to write)."
        )
        return report

    goals = structure_plan_via_cursor(plan["text"])
    if goals is None:
        report = {
            "status": "blocked_on_cursor",
            "reason": "CURSOR_API_KEY not provisioned or structuring call failed — see logs/dogfood/goal_engine.jsonl",
        }
        _append_log({"event": "structuring_blocked", **report})
        _update_metrics_footer("Goal Engine: plan exists — blocked on Cursor SDK structuring pass (no key yet).")
        return report

    if not dry_run:
        _write_goals_yaml(goals)
    proposals = propose_kpi_deltas(goals)
    applied = apply_proposals(proposals, dry_run=dry_run)
    status_line = (
        f"Goal Engine: {len(goals.get('okrs', []))} OKR(s) parsed, "
        f"{applied} KPI target(s) {'would be updated' if dry_run else 'updated'}."
    )
    _update_metrics_footer(status_line)
    return {
        "status": "ok",
        "okr_count": len(goals.get("okrs", [])),
        "proposals": proposals,
        "applied": applied,
        "dry_run": dry_run,
    }


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Goal Engine tick (yearly plan -> loop KPI targets)")
    ap.add_argument("--dry-run", action="store_true", help="Compute + log proposals, don't write overrides")
    args = ap.parse_args(argv)
    report = run(dry_run=bool(args.dry_run))
    print(json.dumps(report, indent=2, ensure_ascii=False))
    return 0


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