dev-puppeteer/my-deepagent-seed/schemas/validate.py

"""Validate all seed assets. Run from schemas/ directory."""

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

import jsonschema
import yaml
from pydantic import BaseModel, Field, ValidationError


class FilesystemPermissionSpec(BaseModel):
    operations: list[Literal["read", "write", "edit", "ls"]]
    paths: list[str]
    mode: Literal["allow", "deny"] = "allow"


class PersonaSubagent(BaseModel):
    name: str
    description: str
    system_prompt: str
    allowed_tools: list[str] = []
    model: str | None = None
    permissions: list[FilesystemPermissionSpec] = []
    interrupt_on: dict[str, Any] = {}


class Persona(BaseModel):
    name: str
    version: int = Field(ge=1)
    description: str | None = None
    backend: Literal["openrouter", "anthropic", "openai", "google", "fake"]
    model: str
    provider_origin: str
    capabilities: list[str]
    max_risk_level: Literal["low", "medium", "high"]
    allowed_roles: list[str] | None = None
    system_prompt: str
    allowed_tools: list[str] | None = None
    subagents: list[PersonaSubagent] = []
    permissions: list[FilesystemPermissionSpec] = []
    interrupt_on: dict[str, Any] | None = None
    model_params: dict[str, Any] = {}
    deepagents_backend: Literal[
        "state", "local_shell", "filesystem", "composite", "langsmith"
    ] = "local_shell"
    fallback_model: str | None = None
    max_cost_per_call_usd: float | None = None


class ExpectedArtifact(BaseModel):
    path: str
    schema_: str = Field(alias="schema")

    model_config = {"populate_by_name": True}


class WorkflowPhase(BaseModel):
    key: str
    title: str
    risk: Literal["low", "medium", "high"]
    role: str
    expected_artifact: ExpectedArtifact | None = None
    gates: list[str] = []
    timeout_seconds: int | None = None
    instructions: str
    max_budget_usd: float | None = None


class WorkflowRole(BaseModel):
    id: str
    required_capabilities: list[str]
    preferred_backends: list[str] = []
    fallback_personas: list[str] = []


class WorkflowTemplate(BaseModel):
    name: str
    version: int = Field(ge=1)
    description: str | None = None
    roles: list[WorkflowRole]
    phases: list[WorkflowPhase]
    default_gates: list[str] = []
    max_total_budget_usd: float | None = None


ROOT = Path(__file__).parent
CAPABILITIES = {
    "spec_write",
    "phase_planning",
    "task_dag_planning",
    "code_edit",
    "test_first_development",
    "code_review",
    "evidence_check",
    "command_execute",
    "backtest_run",
    "metric_extract",
    "failure_mining",
    "objective_eval",
    "final_report_compose",
}

errors: list[str] = []


def validate_personas() -> int:
    count = 0
    for f in sorted((ROOT / "personas").glob("*.yaml")):
        try:
            data = yaml.safe_load(f.read_text())
            p = Persona.model_validate(data)
            unknown_caps = set(p.capabilities) - CAPABILITIES
            if unknown_caps:
                raise ValueError(f"unknown capabilities: {unknown_caps}")
            print(f"  ok personas/{f.name}")
            count += 1
        except (ValidationError, ValueError, yaml.YAMLError) as e:
            errors.append(f"personas/{f.name}: {e}")
            print(f"  FAIL personas/{f.name}: {e}")
    return count


def validate_workflows() -> int:
    count = 0
    for f in sorted((ROOT / "workflows").glob("*.yaml")):
        try:
            data = yaml.safe_load(f.read_text())
            w = WorkflowTemplate.model_validate(data)
            role_ids = {r.id for r in w.roles}
            for ph in w.phases:
                if ph.role not in role_ids:
                    raise ValueError(
                        f"phase '{ph.key}' references unknown role '{ph.role}'"
                    )
            print(f"  ok workflows/{f.name}")
            count += 1
        except (ValidationError, ValueError, yaml.YAMLError) as e:
            errors.append(f"workflows/{f.name}: {e}")
            print(f"  FAIL workflows/{f.name}: {e}")
    return count


def validate_artifact_schemas() -> int:
    count = 0
    for f in sorted((ROOT / "artifacts").rglob("*.json")):
        try:
            schema = json.loads(f.read_text())
            jsonschema.Draft202012Validator.check_schema(schema)
            if "$id" not in schema:
                raise ValueError("missing $id")
            print(f"  ok artifacts/{f.relative_to(ROOT / 'artifacts')}")
            count += 1
        except (
            jsonschema.exceptions.SchemaError,
            ValueError,
            json.JSONDecodeError,
        ) as e:
            errors.append(f"artifacts/{f.relative_to(ROOT / 'artifacts')}: {e}")
            print(f"  FAIL artifacts/{f.relative_to(ROOT / 'artifacts')}: {e}")
    return count


if __name__ == "__main__":
    print("Personas:")
    p_count = validate_personas()
    print("\nWorkflows:")
    w_count = validate_workflows()
    print("\nArtifact schemas:")
    a_count = validate_artifact_schemas()
    print(f"\nTotal: personas={p_count}, workflows={w_count}, artifacts={a_count}")
    if errors:
        print(f"\nERRORS: {len(errors)}")
        for e in errors:
            print(f"  - {e}")
        sys.exit(1)
    print("\nAll seeds validated.")