"""Shared engine for the /review command — every surface calls this. /review spawns an independent, full-privilege background subagent (the same async delegation rail as ``delegate_task(background=true)``) whose job is to thoroughly review whatever the recent conversation presented: a PR, a diff, code, documentation, or any other work product. The reviewer's result re-enters the spawning session as a normal async-delegation completion, so the primary agent sees the review and can act on it. Model routing: the reviewer runs on ``auxiliary.review`` (provider/model/ base_url/api_key/api_mode in config.yaml) when configured; otherwise it inherits the parent agent's credentials — main-model-first, same convention as every other auxiliary task. Resolution reuses the delegation credential resolver (``tools.delegate_tool._resolve_delegation_credentials``) via the internal ``credentials_cfg`` parameter of ``delegate_task`` so native-SDK providers, api_mode detection, and credential pools all behave identically to ``delegation.provider`` pins. Surfaces (CLI ``/review``, gateway ``/review``, TUI/Desktop live dispatch) are thin adapters: snapshot the conversation, call :func:`start_review`, print the dispatch note. """ from __future__ import annotations import json import logging import re from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) # How many recent chat messages (user + assistant turns) the reviewer gets. DEFAULT_CONTEXT_MESSAGES = 10 # Per-message excerpt cap. Generous — a PR summary or diff excerpt the primary # agent just printed is exactly what the reviewer needs — but bounded so a # pathological turn can't blow up the child's opening context. _MESSAGE_CHAR_CAP = 12_000 def _message_text(message: Dict[str, Any]) -> str: """Extract display text from a conversation message dict. Handles both plain-string content and OpenAI-style multimodal content lists (text parts joined; non-text parts noted). """ content = message.get("content") if isinstance(content, str): return content if isinstance(content, list): parts: List[str] = [] for part in content: if isinstance(part, dict): if part.get("type") == "text": parts.append(str(part.get("text") or "")) else: parts.append(f"[{part.get('type', 'attachment')}]") return "\n".join(p for p in parts if p) return "" def snapshot_recent_messages( messages: List[Dict[str, Any]], limit: int = DEFAULT_CONTEXT_MESSAGES, ) -> List[Dict[str, str]]: """Return the last ``limit`` user/assistant messages as {role, text} dicts. System messages and tool results are excluded — the chat turns are what the user and their primary agent actually said (the PR link, the summary, the diff excerpt). Empty-text messages (pure tool-call assistant stubs) are skipped. """ out: List[Dict[str, str]] = [] for message in reversed(list(messages or [])): if not isinstance(message, dict): continue role = str(message.get("role") or "") if role not in ("user", "assistant"): continue text = _message_text(message).strip() if not text: continue if len(text) > _MESSAGE_CHAR_CAP: text = text[:_MESSAGE_CHAR_CAP] + "\n[... truncated ...]" out.append({"role": role, "text": text}) if len(out) >= limit: break out.reverse() return out def collect_parent_loaded_skills( parent_agent, messages: List[Dict[str, Any]], limit: int = 8, ) -> List[str]: """Names of skills the parent agent was operating under. Two sources, both surface-independent: * Launch-preloaded skills (``hermes -s``, kanban lanes, TUI skills env): their activation notes are embedded in the parent's ``ephemeral_system_prompt`` with a stable marker (see ``agent.skill_commands.build_preloaded_skills_prompt``). * Mid-session loads: ``skill_view`` tool calls in the parent's conversation history (assistant ``tool_calls`` entries). Order: preloaded first, then history loads, deduped, capped at ``limit`` (a reviewer told to load 30 skills would burn its budget before working). """ names: List[str] = [] seen: set = set() def _add(name: str) -> None: cleaned = (name or "").strip() if cleaned and cleaned not in seen: seen.add(cleaned) names.append(cleaned) prompt = str(getattr(parent_agent, "ephemeral_system_prompt", "") or "") for match in re.finditer(r'with the "([^"]+)" skill\s+preloaded', prompt): _add(match.group(1)) for message in messages or []: if not isinstance(message, dict) or message.get("role") != "assistant": continue for tool_call in message.get("tool_calls") or []: if not isinstance(tool_call, dict): continue fn = tool_call.get("function") or {} if fn.get("name") != "skill_view": continue try: args = json.loads(fn.get("arguments") or "{}") except Exception: continue # Only whole-skill loads seed the reviewer; a reference-file read # (file_path=...) is a detail of the parent's task, and the # reviewer loading the main SKILL.md covers it. if isinstance(args, dict) and not args.get("file_path"): _add(str(args.get("name") or "")) return names[:limit] def build_review_task( snapshot: List[Dict[str, str]], user_prompt: str = "", loaded_skills: Optional[List[str]] = None, ) -> tuple: """Compose the reviewer subagent's (goal, context) pair.""" goal = ( "Act as an independent senior reviewer. Thoroughly review the work " "presented in the conversation excerpt provided in your context: " "investigate any code, pull request, branch, commit, documentation, " "design, or other artifact it references (open the PR, read the " "diff, run the code or tests where feasible) rather than judging " "from the excerpt alone. Produce a full, structured review: what " "the work does, whether it is correct and complete, concrete " "defects or risks found (with file/line references where possible), " "what was verified vs. only read, and a clear final verdict with " "recommended next steps." ) lines = [ "You were spawned by the /review command. The following is an " "excerpt of the most recent conversation between the user and " "their primary agent. It is your starting evidence — the work to " "review is referenced in it.", "", "--- Recent conversation (oldest first) ---", ] for message in snapshot: label = "USER" if message["role"] == "user" else "PRIMARY AGENT" lines.append(f"[{label}]") lines.append(message["text"]) lines.append("") lines.append("--- End of conversation excerpt ---") if loaded_skills: skill_list = ", ".join(loaded_skills) lines.append("") lines.append( "The primary agent was operating under these loaded skills: " f"{skill_list}. Before reviewing, load each with " "skill_view(name=...) and treat their conventions, invariants, " "and review standards as binding for your assessment — the work " "was produced under them and must be judged against them." ) if user_prompt.strip(): lines.append("") lines.append("Additional review instructions from the user:") lines.append(user_prompt.strip()) lines.append("") lines.append( "Your review is delivered back into that conversation, addressed to " "the primary agent and its user. Be direct and specific; do not " "soften findings." ) return goal, "\n".join(lines) def _load_review_credentials_cfg() -> Optional[Dict[str, Any]]: """Read ``auxiliary.review`` into a delegation-credentials-shaped dict. Returns None when the user configured nothing (provider=auto/empty and no model/base_url), which makes the reviewer inherit the parent agent's credentials — the main-model-first default. """ try: from hermes_cli.config import load_config_readonly full = load_config_readonly() aux = full.get("auxiliary") or {} review = aux.get("review") or {} if not isinstance(review, dict): return None except Exception: return None provider = str(review.get("provider") or "").strip() if provider.lower() == "auto": provider = "" model = str(review.get("model") or "").strip() base_url = str(review.get("base_url") or "").strip() if not (provider or model or base_url): return None return { "provider": provider, "model": model, "base_url": base_url, "api_key": str(review.get("api_key") or "").strip(), "api_mode": str(review.get("api_mode") or "").strip(), } def start_review( parent_agent, messages: List[Dict[str, Any]], user_prompt: str = "", ) -> Dict[str, Any]: """Dispatch the reviewer subagent in the background. Returns the parsed ``delegate_task`` dispatch dict (``status: "dispatched"`` with a ``delegation_id`` on success, or the synchronous result dict on channels that cannot route async completions). Raises ValueError when there is nothing to review or the dispatch is rejected/errored. """ if parent_agent is None: raise ValueError("No active agent — send a message first.") snapshot = snapshot_recent_messages(messages) if not snapshot: raise ValueError("Nothing to review yet — the conversation is empty.") loaded_skills = collect_parent_loaded_skills(parent_agent, messages) goal, context = build_review_task(snapshot, user_prompt, loaded_skills) credentials_cfg = _load_review_credentials_cfg() from tools.delegate_tool import delegate_task raw = delegate_task( goal=goal, context=context, background=True, parent_agent=parent_agent, credentials_cfg=credentials_cfg, ) try: result = json.loads(raw) except Exception: raise ValueError(f"Review dispatch failed: {raw!r}") if isinstance(result, dict) and result.get("error"): raise ValueError(str(result["error"])) if not isinstance(result, dict): raise ValueError(f"Review dispatch failed: {raw!r}") result.setdefault("review_model", (credentials_cfg or {}).get("model") or "") return result def format_dispatch_note(result: Dict[str, Any], user_prompt: str = "") -> str: """Human-facing one-liner for a successful dispatch. Shared by surfaces.""" model = str(result.get("review_model") or "").strip() model_note = f" on {model}" if model else "" focus_note = f" (focus: {user_prompt.strip()})" if user_prompt.strip() else "" if result.get("status") == "dispatched": return ( f"⚖ Review subagent dispatched{model_note}{focus_note} — it is " f"investigating the last {DEFAULT_CONTEXT_MESSAGES} messages in " f"the background and its full review will re-enter this " f"conversation when it finishes." ) # Synchronous fallback (channels that cannot route async completions). return ( f"⚖ Review completed synchronously{model_note}{focus_note} — " f"results:\n{json.dumps(result.get('results', result), ensure_ascii=False)[:4000]}" )