control_plane/schema.py

"""Structured action contract.

The agent's only output is a typed, structured object describing a single action
it wishes to take. That object is validated *locally* — before it ever reaches
the governance gate — so that malformed proposals (missing fields, bad enum
values, wrong types) are rejected with a clear error rather than being parsed
out of free-text. The agent never executes; it only proposes.

Two models live here:

- :class:`ProposedAction` — the action itself (the rows of the action table).
- :class:`AgentDecision` — the enclosing object the agent emits: a summary, a
  risk level, the proposed action, and any prohibited actions it detected while
  reasoning (surfaced, never acted upon).
"""

from __future__ import annotations

from enum import Enum
from typing import Any

from pydantic import BaseModel, ConfigDict, Field, field_validator


class AutonomyTier(str, Enum):
    """The four autonomy tiers."""

    L0_READ_ONLY = "L0_READ_ONLY"
    L1_RECOMMEND_ONLY = "L1_RECOMMEND_ONLY"
    L2_BOUNDED_ACTION = "L2_BOUNDED_ACTION"
    L3_APPROVAL_REQUIRED_ACTION = "L3_APPROVAL_REQUIRED_ACTION"


class Backend(str, Enum):
    """The four execution backends."""

    DIRECT_API = "direct_api"
    FUNCTION_CALL = "function_call"
    MCP_CLIENT = "mcp_client"
    CLI_EXECUTOR = "cli_executor"


class RiskLevel(str, Enum):
    """Agent-assessed risk level for the enclosing decision."""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


# Reject unknown fields and forbid coercion that would silently accept the wrong
# type (e.g. an int where a string is required). `strict=True` makes Pydantic
# refuse to coerce "1" -> 1 and similar, which the "wrong type" rule needs.
_MODEL_CONFIG = ConfigDict(extra="forbid", strict=True)


class ProposedAction(BaseModel):
    """A single action the agent proposes.

    Required string fields must be non-empty; `arguments` is a structured map and
    `evidence` is a list of identifiers. Validation failures raise
    ``pydantic.ValidationError``.
    """

    model_config = _MODEL_CONFIG

    incident_id: str = Field(min_length=1, description="Incident/case identifier.")

    agent_id: str = Field(min_length=1, description="Identity of the proposing agent.")
    # `strict=False` lets the enum accept its string value (the agent emits JSON
    # strings); membership is still enforced, so unknown values are rejected.

    autonomy_tier: AutonomyTier = Field(
        strict=False, description="Tier the action requires."
    )

    backend: Backend = Field(strict=False, description="Target execution path.")

    action_name: str = Field(min_length=1, description="Requested backend action.")

    arguments: dict[str, Any] = Field(
        default_factory=dict, description="Arguments for the action."
    )

    reason: str = Field(min_length=1, description="Why the action is proposed.")
    evidence: list[str] = Field(
        default_factory=list, description="Supporting evidence identifiers."
    )

    @field_validator("incident_id", "agent_id", "action_name", "reason")
    @classmethod
    def _not_blank(cls, v: str) -> str:
        """Reject whitespace-only values that ``min_length`` alone would pass."""
        if not v.strip():
            raise ValueError("must not be empty or whitespace")
        return v

    @field_validator("evidence")
    @classmethod
    def _evidence_ids_non_empty(cls, v: list[str]) -> list[str]:
        """Evidence entries are identifiers; none may be blank."""
        for item in v:
            if not item.strip():
                raise ValueError("evidence identifiers must not be empty")
        return v


class AgentDecision(BaseModel):
    """The enclosing object the agent emits.

    Carries a human-readable summary, an assessed risk level, the proposed
    action, and any prohibited actions the agent detected while reasoning but did
    not act on.
    """

    model_config = _MODEL_CONFIG

    summary: str = Field(min_length=1, description="Human-readable decision summary.")

    risk_level: RiskLevel = Field(strict=False, description="Assessed risk level.")

    proposed_action: ProposedAction = Field(description="The single proposed action.")
    
    prohibited_actions_detected: list[str] = Field(
        default_factory=list,
        description="Prohibited actions the agent detected but did not act on.",
    )

    @field_validator("summary")
    @classmethod
    def _summary_not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("summary must not be empty or whitespace")
        return v