verify.py

#!/usr/bin/env python3.12
"""Verification layer - ground truth checking for ULTRON actions."""

import logging
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path

import requests

log = logging.getLogger(__name__)

AOMS_URL = "http://localhost:9100"


@dataclass
class VerificationResult:
    success: bool
    method: str
    evidence: str
    confidence: float  # 0-1


class Verifier:

    def verify_file_exists(self, path: str, min_size: int = 10) -> VerificationResult:
        p = Path(path).expanduser().resolve()
        if not p.exists():
            return VerificationResult(False, "file_exists", f"File not found: {path}", 1.0)
        size = p.stat().st_size
        if size < min_size:
            return VerificationResult(False, "file_size", f"File too small: {size} bytes", 1.0)
        return VerificationResult(True, "file_exists", f"File exists: {size} bytes", 1.0)

    def verify_file_contains(self, path: str, expected: list[str]) -> VerificationResult:
        p = Path(path).expanduser().resolve()
        if not p.exists():
            return VerificationResult(False, "file_contains", "File not found", 1.0)
        content = p.read_text()
        missing = [s for s in expected if s not in content]
        if missing:
            return VerificationResult(False, "file_contains", f"Missing: {missing}", 1.0)
        return VerificationResult(True, "file_contains", "All expected content found", 1.0)

    def verify_command_succeeds(self, cmd: str, timeout: int = 30) -> VerificationResult:
        try:
            result = subprocess.run(cmd, shell=True, capture_output=True, timeout=timeout, text=True)
            if result.returncode == 0:
                return VerificationResult(True, "command", f"Exit 0: {result.stdout[:200]}", 1.0)
            return VerificationResult(False, "command", f"Exit {result.returncode}: {result.stderr[:200]}", 1.0)
        except subprocess.TimeoutExpired:
            return VerificationResult(False, "command", "Timeout", 1.0)
        except Exception as e:
            return VerificationResult(False, "command", str(e), 1.0)

    def verify_url_accessible(self, url: str, expected_status: int = 200) -> VerificationResult:
        try:
            resp = requests.head(url, timeout=10, allow_redirects=True)
            if resp.status_code == expected_status:
                return VerificationResult(True, "url", f"Status {resp.status_code}", 1.0)
            return VerificationResult(False, "url", f"Status {resp.status_code}, expected {expected_status}", 1.0)
        except Exception as e:
            return VerificationResult(False, "url", str(e), 1.0)

    def verify_tweet_exists(self, tweet_id: str) -> VerificationResult:
        try:
            result = subprocess.run(
                ["xurl", "get", tweet_id],
                capture_output=True, text=True, timeout=15,
            )
            if result.returncode == 0 and "text" in result.stdout:
                return VerificationResult(True, "tweet", "Tweet found", 1.0)
            return VerificationResult(False, "tweet", "Tweet not found or xurl failed", 0.8)
        except Exception as e:
            return VerificationResult(False, "tweet", str(e), 0.8)

    def verify_github_repo(self, repo: str) -> VerificationResult:
        try:
            resp = requests.get(f"https://api.github.com/repos/{repo}", timeout=10)
            if resp.status_code == 200:
                return VerificationResult(True, "github", f"Repo exists: {repo}", 1.0)
            return VerificationResult(False, "github", f"Status {resp.status_code}", 1.0)
        except Exception as e:
            return VerificationResult(False, "github", str(e), 1.0)

    def verify_aoms_stored(self, query: str) -> VerificationResult:
        try:
            resp = requests.post(f"{AOMS_URL}/memory/search",
                json={"query": query, "limit": 1}, timeout=5)
            data = resp.json()
            results = data if isinstance(data, list) else data.get("results", [])
            if results:
                return VerificationResult(True, "aoms", "Found in AOMS", 1.0)
            return VerificationResult(False, "aoms", "Not found in AOMS", 0.8)
        except Exception as e:
            return VerificationResult(False, "aoms", str(e), 1.0)


# --- Extraction helpers ---

def extract_file_paths(text: str) -> list[str]:
    """Extract file paths from Ollama output."""
    patterns = [
        r'(?:outputs/[\w./-]+)',
        r'(?:results/[\w./-]+)',
        r'(?:~/curiosity/[\w./-]+)',
    ]
    paths = []
    for p in patterns:
        paths.extend(re.findall(p, text))
    return paths


def extract_tweet_id(text: str) -> str | None:
    match = re.search(r'(?:status/|tweet_id["\s:=]+)(\d{10,})', text)
    return match.group(1) if match else None


def extract_github_repo(text: str) -> str | None:
    match = re.search(r'(?:github\.com/|repos/)([a-zA-Z0-9_-]+/[a-zA-Z0-9_.-]+)', text)
    return match.group(1) if match else None


# --- Smart verification routing ---

def verify_task_completion(task: dict, result: str, verifier: Verifier | None = None) -> VerificationResult:
    """Choose verification method based on task type."""
    if verifier is None:
        verifier = Verifier()
    task_name = task["name"].lower()

    # File-related tasks
    if any(w in task_name for w in ["write", "create file", "save", "readme"]):
        paths = extract_file_paths(result)
        if paths:
            return verifier.verify_file_exists(paths[0])
        # Check if result file was saved by goal_engine
        from pathlib import Path as P
        result_file = P(f"~/curiosity/results").expanduser() / task.get("_goal_id", "unknown") / f"{task['id']}.md"
        return verifier.verify_file_exists(str(result_file), min_size=200)

    # Tweet-related tasks
    if any(w in task_name for w in ["tweet", "post to twitter", "x post"]):
        tid = extract_tweet_id(result)
        if tid:
            return verifier.verify_tweet_exists(tid)
        return VerificationResult(False, "tweet", "No tweet ID found in result", 0.5)

    # GitHub tasks
    if any(w in task_name for w in ["github", "repo", "repository"]):
        repo = extract_github_repo(result)
        if repo:
            return verifier.verify_github_repo(repo)
        return VerificationResult(False, "github", "No repo found in result", 0.5)

    # Research tasks - verify stored in AOMS
    if any(w in task_name for w in ["research", "analyze", "investigate"]):
        return verifier.verify_aoms_stored(task["name"][:50])

    # Default: content quality check
    if len(result) > 200:
        return VerificationResult(True, "content_length", f"Result: {len(result)} chars", 0.7)
    return VerificationResult(False, "content_length", f"Result too short: {len(result)} chars", 0.5)


def should_ask_human(verification: VerificationResult, task: dict) -> bool:
    """Decide if human input is needed."""
    if verification.confidence < 0.7:
        return True
    high_risk = ["delete", "publish", "send", "money", "payment", "deploy"]
    if any(w in task["name"].lower() for w in high_risk):
        return True
    if task.get("retry_count", 0) >= 2:
        return True
    return False