"""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": "") - "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"]