scripts/system_prompt.py

"""Shared system prompt + observation/action serialization for OpsGuard GRPO training.

The model emits a JSON object describing the action. We use a single tool/method
on the environment class (`triage`) rather than one method per action type because
the underlying OpsguardAction is a tagged-union (action_type discriminates).
This keeps the tool surface small and matches our env contract exactly.

Schema accepted by `parse_action`:
    {
      "action_type": "label" | "close_spam" | "request_info" | "link_duplicate"
                   | "assign" | "comment" | "merge_pr" | "query_history" | "wait",
      "target_issue_id": int | null,
      "label": str | null,
      "duplicate_of_id": int | null,
      "assignee_login": str | null,
      "comment_body": str | null,
      "query": str | null,
      "reasoning": str | null
    }
"""
from __future__ import annotations

import json
import re
from typing import Any

# Importable from either layout (package or flat-script)
try:
    from ..models import ActionType, OpsguardAction, OpsguardObservation  # type: ignore
except (ImportError, ValueError):
    try:
        from models import ActionType, OpsguardAction, OpsguardObservation  # type: ignore
    except ImportError:
        # Last resort: when this file is imported as scripts.system_prompt,
        # walk up one directory to find the project root models.
        import os
        import sys
        _root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
        if _root not in sys.path:
            sys.path.insert(0, _root)
        from models import ActionType, OpsguardAction, OpsguardObservation  # type: ignore


SYSTEM_PROMPT = """You are OpsGuard, an autonomous open-source repository maintainer.

Your job: process an incoming queue of GitHub issues and pull requests for the
repo. Each step you observe ONE current issue plus optional memory hits from
prior history and your own recent actions. You must pick exactly ONE action.

Available actions (action_type values):
- "label"           : apply a topical label (set "label": "<name>")
- "close_spam"      : close as spam / not-a-bug / invalid (only for actual spam!)
- "request_info"    : ask the author for reproduction steps (set "comment_body")
- "link_duplicate"  : mark as duplicate of an existing issue (set "duplicate_of_id")
- "assign"          : assign to a maintainer (set "assignee_login")
- "comment"         : leave a substantive comment (set "comment_body")
- "merge_pr"        : merge an approved pull request (only when is_pr=true)
- "query_history"   : search prior issues before deciding (set "query")
- "wait"            : skip without action (rarely correct)

Spam signals to watch for: paraphrased duplicates, fake urgency ("P0",
"URGENT", "production down"), sock-puppet consensus ("everyone agrees"),
vague low-content complaints, coordinated bumping floods, brand-new
authors with zero prior PRs.

Output format: respond with a SINGLE JSON object. No prose, no code fences.
Example: {"action_type":"label","target_issue_id":123,"label":"bug","reasoning":"clear bug report"}
Example: {"action_type":"close_spam","target_issue_id":456,"reasoning":"vague urgency, new account, no repro"}
Example: {"action_type":"query_history","query":"memory leak in trainer","reasoning":"check for prior duplicates first"}

If unsure, prefer "query_history" once, then act."""


_ACTION_TYPES = {a.value for a in ActionType}


def format_observation(obs: OpsguardObservation) -> str:
    """Render an OpsguardObservation as a compact JSON-ish prompt for the LLM."""
    if obs.current_issue is None:
        body: dict[str, Any] = {
            "scenario": obs.scenario_id,
            "step": obs.step,
            "step_budget": obs.step_budget,
            "queue_position": obs.queue_position,
            "queue_total": obs.queue_total,
            "current_issue": None,
            "feedback": obs.feedback,
            "note": "queue empty or terminal — emit {\"action_type\":\"wait\"}",
        }
        return json.dumps(body, ensure_ascii=False, indent=2)

    ci = obs.current_issue
    body = {
        "scenario": obs.scenario_id,
        "step": obs.step,
        "step_budget": obs.step_budget,
        "queue_position": f"{obs.queue_position}/{obs.queue_total}",
        "current_issue": {
            "issue_id": ci.issue_id,
            "number": ci.number,
            "is_pr": ci.is_pr,
            "title": ci.title[:200],
            "body": (ci.body or "")[:800],
            "author": ci.author_login,
            "author_pr_count": ci.author_pr_count,
            "author_account_age_days": ci.author_account_age_days,
            "available_labels": ci.available_labels[:20],
            "comments_preview": ci.comments_preview[:3],
        },
        "memory_hits": obs.memory_hits[:5],
        "recent_actions": obs.recent_actions[-5:],
        "feedback_from_last_step": obs.feedback,
    }
    return json.dumps(body, ensure_ascii=False, indent=2)


# Greedy brace matcher for top-level JSON objects (handles nested braces).
def _extract_first_json_object(text: str) -> str | None:
    start = text.find("{")
    if start < 0:
        return None
    depth = 0
    in_string = False
    escape = False
    for i in range(start, len(text)):
        ch = text[i]
        if in_string:
            if escape:
                escape = False
            elif ch == "\\":
                escape = True
            elif ch == '"':
                in_string = False
            continue
        if ch == '"':
            in_string = True
        elif ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return text[start : i + 1]
    return None


def _coerce_int(v: Any) -> int | None:
    if v is None:
        return None
    if isinstance(v, bool):
        return None
    if isinstance(v, int):
        return v
    if isinstance(v, str):
        m = re.search(r"-?\d+", v)
        if m:
            try:
                return int(m.group(0))
            except ValueError:
                return None
    return None


def _coerce_str(v: Any) -> str | None:
    if v is None:
        return None
    if isinstance(v, str):
        return v.strip() or None
    return str(v)


def parse_action(text: str) -> OpsguardAction:
    """Parse model output into OpsguardAction. Returns WAIT on any failure.

    Tolerates: code fences, prose around the JSON, single quotes, missing
    optional fields, integer-as-string for ids.
    """
    if not text:
        return OpsguardAction(action_type=ActionType.WAIT, reasoning="empty output")

    # Strip common fences
    candidate = text.strip()
    if candidate.startswith("```"):
        # remove ```json ... ``` wrappers
        candidate = re.sub(r"^```[a-zA-Z]*\n?", "", candidate)
        candidate = re.sub(r"\n?```\s*$", "", candidate)

    json_str = _extract_first_json_object(candidate)
    if json_str is None:
        return OpsguardAction(action_type=ActionType.WAIT, reasoning="no JSON object found")

    # Try strict, then a single-quote -> double-quote fallback
    try:
        data = json.loads(json_str)
    except json.JSONDecodeError:
        try:
            data = json.loads(json_str.replace("'", '"'))
        except json.JSONDecodeError:
            return OpsguardAction(action_type=ActionType.WAIT, reasoning="JSON parse error")

    if not isinstance(data, dict):
        return OpsguardAction(action_type=ActionType.WAIT, reasoning="JSON not an object")

    raw_type = data.get("action_type") or data.get("action") or "wait"
    if not isinstance(raw_type, str) or raw_type.lower() not in _ACTION_TYPES:
        return OpsguardAction(action_type=ActionType.WAIT, reasoning=f"unknown action_type={raw_type!r}")
    action_type = ActionType(raw_type.lower())

    return OpsguardAction(
        action_type=action_type,
        target_issue_id=_coerce_int(data.get("target_issue_id")),
        label=_coerce_str(data.get("label")),
        duplicate_of_id=_coerce_int(data.get("duplicate_of_id")),
        assignee_login=_coerce_str(data.get("assignee_login")),
        comment_body=_coerce_str(data.get("comment_body")),
        query=_coerce_str(data.get("query")),
        reasoning=_coerce_str(data.get("reasoning")),
    )


__all__ = ["SYSTEM_PROMPT", "format_observation", "parse_action"]