CascadingDev/automation/agents.py

#!/usr/bin/env python3
"""
AI-powered agents for extracting structured information from discussions.

All automation paths (runner + workflow) now share the same ModelConfig cascade,
so CLI fallbacks (Claude → Codex → Gemini, etc.) behave consistently and surface
the same progress messages.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any, Mapping

from automation.ai_config import load_ai_settings
from automation.patcher import ModelConfig, _run_ai_command  # type: ignore


_MODEL_CACHE: ModelConfig | None = None


def _get_model() -> ModelConfig:
    global _MODEL_CACHE
    if _MODEL_CACHE is None:
        repo_root = Path.cwd().resolve()
        settings = load_ai_settings(repo_root)
        _MODEL_CACHE = ModelConfig(
            commands=settings.runner.command_chain,
            sentinel=settings.runner.sentinel,
            runner_settings=settings.runner,
        )
    return _MODEL_CACHE


def _invoke_model(prompt: str, hint: str = "fast") -> str | None:
    model = _get_model()
    commands = model.get_commands_for_hint(hint)
    repo_root = Path.cwd().resolve()
    errors: list[str] = []
    sentinel_seen = False

    total = len(commands)
    for idx, command in enumerate(commands, start=1):
        provider_name = command.split()[0]
        sys.stderr.write(f"[agents] provider {idx}/{total} → {provider_name}\n")
        sys.stderr.flush()
        executor, raw_stdout, stderr, returncode = _run_ai_command(command, prompt, repo_root)

        if raw_stdout:
            stripped = raw_stdout.strip()
            if stripped == model.sentinel:
                sentinel_seen = True
                sys.stderr.write(
                    f"[agents] provider {idx}/{total} → {provider_name} returned sentinel (no change)\n"
                )
                sys.stderr.flush()
                continue
            return stripped

        if returncode == 0:
            errors.append(f"{executor!r} produced no output")
            sys.stderr.write(
                f"[agents] provider {idx}/{total} → {provider_name} returned no output\n"
            )
            sys.stderr.flush()
        else:
            errors.append(f"{executor!r} exited with {returncode}: {stderr or 'no stderr'}")
            sys.stderr.write(
                f"[agents] provider {idx}/{total} → {provider_name} exited with {returncode}\n"
            )
            sys.stderr.flush()

    if sentinel_seen:
        return None

    if errors:
        sys.stderr.write("[agents] warning: " + "; ".join(errors) + "\n")
    return None


def _request_json(prompt: str, content: str, hint: str = "fast") -> dict[str, Any] | None:
    full_prompt = (
        "You must answer with valid JSON only. No code fences, no backticks, "
        "no explanations.\n\n"
        f"{prompt}\n\n"
        "<discussion_content>\n"
        f"{content}\n"
        "</discussion_content>\n"
    )
    raw = _invoke_model(full_prompt, hint=hint)
    if not raw:
        return None

    cleaned = raw.strip()
    if cleaned.startswith("```"):
        cleaned = cleaned.strip("`")
    try:
        return json.loads(cleaned)
    except json.JSONDecodeError as exc:
        preview = cleaned[:200].replace("\n", " ")
        sys.stderr.write(f"[agents] warning: failed to parse JSON ({exc}): {preview}\n")
        return None


def normalize_discussion(content: str) -> dict[str, Any] | None:
    prompt = """You are a discussion normalizer. Extract structured information from the discussion content.

Return a JSON object with these fields:
{
  "votes": [
    {"participant": "Name", "vote": "READY|CHANGES|REJECT", "line": "original line text"}
  ],
  "questions": [
    {"participant": "Name", "question": "text", "line": "original line"}
  ],
  "decisions": [
    {"participant": "Name", "decision": "text", "rationale": "why", "line": "original line"}
  ],
  "action_items": [
    {"participant": "Name", "action": "text", "status": "TODO|ASSIGNED|DONE", "assignee": "Name or null", "line": "original line"}
  ],
  "mentions": [
    {"from": "Name", "to": "Name or @all", "context": "surrounding text"}
  ]
}

Only include items that are clearly present. Return empty arrays for missing categories."""
    return _request_json(prompt, content, hint="fast")


def track_questions(content: str, existing_questions: list[dict[str, Any]]) -> dict[str, Any] | None:
    existing_q_json = json.dumps(existing_questions, indent=2)
    prompt = f"""You are a question tracking agent. Process new discussion content and track questions.

Existing tracked questions:
{existing_q_json}

For the new content:
1. Identify new questions
2. Match answers to existing questions (by context/participant names)
3. Mark questions as: OPEN, PARTIAL (some answers), or ANSWERED

Return JSON with "new_questions" and "updated_questions" arrays."""
    return _request_json(prompt, content, hint="fast")


def track_action_items(content: str, existing_items: list[dict[str, Any]]) -> dict[str, Any] | None:
    existing_items_json = json.dumps(existing_items, indent=2)
    prompt = f"""You are an action item tracking agent. Process new discussion content and track action items.

Existing tracked items:
{existing_items_json}

For the new content:
1. Identify new action items
2. Track assignments
3. Track completions

Return JSON with "new_items" and "updated_items" arrays."""
    return _request_json(prompt, content, hint="fast")


def track_decisions(content: str, existing_decisions: list[dict[str, Any]]) -> dict[str, Any] | None:
    existing_decisions_json = json.dumps(existing_decisions, indent=2)
    prompt = f"""You are a decision logging agent. Extract architectural and design decisions from discussions.

Existing decisions:
{existing_decisions_json}

Identify decisions with rationale, supporters, and whether they supersede previous decisions.

Return JSON with "new_decisions" and "updated_decisions" arrays."""
    return _request_json(prompt, content, hint="quality")


def extract_mentions(content: str) -> list[dict[str, str]]:
    """
    Lightweight fallback: detect @mentions without AI (regex-based).
    """
    mentions: list[dict[str, str]] = []
    for line in content.splitlines():
        segments = line.split()
        for segment in segments:
            if segment.startswith("@") and len(segment) > 1:
                mentions.append(
                    {
                        "from": "unknown",
                        "to": segment.lstrip("@").rstrip(":"),
                        "context": line.strip(),
                    }
                )
    return mentions