feat(F): F-PERM permissions, F-APPR approval, F-AIUSE metadata, F-TRANS transparency, F-DATA-POL data policy, F-OVERSIGHT decision record, F-DRY dry-run, F-AUDIT audit log
Check Cross-Plugin Imports / check (push) Has been cancelled
Check Cross-Plugin Imports / check (push) Has been cancelled
- F-PERM: app/ai/agent_permissions.py (230 lines) — AgentPermissionContext, resolve_agent_permissions(), filter_visible_agents(), check_agent_execute_permission(), optimistic locking - F-APPR: app/core/approval.py (160 lines) + app/routes/approvals.py (305 lines) + migration 0123 — ApprovalRequest model, CRUD API, approve/reject/expire - F-AIUSE: app/ai/ai_use_case.py (156 lines) — AIUseCaseMetadata Pydantic model, validate_ai_use_case() - F-TRANS: app/ai/transparency.py (60 lines) — mark_as_ai_generated(), is_ai_participant() - F-DATA-POL: app/ai/data_policy.py (210 lines) — enforce_data_policy() with SENSITIVE_FIELDS + provider compliance - F-OVERSIGHT: app/ai/oversight.py (108 lines) — DecisionRecord, create_decision_record() - F-DRY: agent_loop.py updated with dry_run parameter - F-AUDIT: agent_loop.py updated with audit log for tool calls - agent_routes.py: AI use case metadata endpoints added - main.py: approval routes registered - All Python compile checks pass
This commit is contained in:
Agent Zero
@@ -0,0 +1,156 @@
|
||||
"""AI use-case metadata and validation.
|
||||
|
||||
Defines the structured metadata that describes *why* and *how* an AI agent
|
||||
may process data. This is the governance contract for an agent definition:
|
||||
which data categories it may touch, which providers/models/actions are
|
||||
allowed, and whether human oversight is required.
|
||||
|
||||
Used by:
|
||||
- ``app/ai/data_policy.py`` — runtime enforcement of allowed data categories
|
||||
- ``app/ai/oversight.py`` — human-review policy (``oversight_policy``)
|
||||
- ``app/plugins/builtins/automation/agent_routes.py`` — PATCH/GET endpoints
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
# Constants
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
# Known data categories an agent may declare it processes.
|
||||
KNOWN_DATA_CATEGORIES = (
|
||||
"contact_data",
|
||||
"email_content",
|
||||
"calendar",
|
||||
"tasks",
|
||||
"dms",
|
||||
"communication",
|
||||
"financial",
|
||||
"public",
|
||||
)
|
||||
|
||||
# Valid oversight policies.
|
||||
OVERSIGHT_POLICIES = ("always_required", "on_high_risk", "never")
|
||||
|
||||
# Valid risk classes.
|
||||
RISK_CLASSES = ("low", "medium", "high")
|
||||
|
||||
# Valid allowed actions.
|
||||
KNOWN_ACTIONS = ("read", "summarize", "draft", "send", "create", "update", "delete")
|
||||
|
||||
|
||||
class AIUseCaseMetadata(BaseModel):
|
||||
"""Structured metadata describing an AI agent's intended use case.
|
||||
|
||||
Attributes:
|
||||
intended_purpose: Human-readable description of the use case.
|
||||
owner: User ID or email responsible for the use case.
|
||||
data_categories: Data categories the agent may process.
|
||||
allowed_providers: Provider IDs the agent may use (empty = any).
|
||||
allowed_models: Model names the agent may use (empty = any).
|
||||
allowed_actions: Actions the agent may perform (empty = any).
|
||||
oversight_policy: When human review is required.
|
||||
risk_class: Risk classification of the use case.
|
||||
human_review_required: Whether a human must review outputs.
|
||||
"""
|
||||
|
||||
intended_purpose: str = Field(default="", max_length=1000)
|
||||
owner: str = Field(default="", max_length=255)
|
||||
data_categories: list[str] = Field(default_factory=list)
|
||||
allowed_providers: list[str] = Field(default_factory=list)
|
||||
allowed_models: list[str] = Field(default_factory=list)
|
||||
allowed_actions: list[str] = Field(default_factory=list)
|
||||
oversight_policy: str = Field(default="never")
|
||||
risk_class: str = Field(default="low")
|
||||
human_review_required: bool = False
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any] | None) -> "AIUseCaseMetadata":
|
||||
"""Build metadata from a raw dict (e.g. the agent's JSONB column)."""
|
||||
if not data:
|
||||
return cls()
|
||||
# Only pass known fields so unknown keys don't break validation.
|
||||
known = {k: v for k, v in data.items() if k in cls.model_fields}
|
||||
return cls(**known)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialize to a plain dict for JSONB storage."""
|
||||
return self.model_dump()
|
||||
|
||||
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
# Validation
|
||||
# ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def validate_ai_use_case(metadata: AIUseCaseMetadata, agent_definition: Any) -> list[str]:
|
||||
"""Validate metadata against an agent configuration.
|
||||
|
||||
Returns a list of human-readable warnings. An empty list means the
|
||||
metadata is consistent with the agent definition.
|
||||
|
||||
Checks performed:
|
||||
- ``intended_purpose`` and ``owner`` are set.
|
||||
- ``data_categories`` are known values.
|
||||
- ``oversight_policy`` and ``risk_class`` are valid.
|
||||
- ``allowed_models`` (if non-empty) include the agent's configured model.
|
||||
- ``allowed_providers`` (if non-empty) include the agent's provider.
|
||||
- ``human_review_required`` is consistent with ``oversight_policy``.
|
||||
"""
|
||||
warnings: list[str] = []
|
||||
|
||||
if not metadata.intended_purpose.strip():
|
||||
warnings.append("intended_purpose is empty — describe the AI use case")
|
||||
|
||||
if not metadata.owner.strip():
|
||||
warnings.append("owner is empty — set a responsible user or email")
|
||||
|
||||
for cat in metadata.data_categories:
|
||||
if cat not in KNOWN_DATA_CATEGORIES:
|
||||
warnings.append(f"data_category '{cat}' is not a known category")
|
||||
|
||||
if metadata.oversight_policy not in OVERSIGHT_POLICIES:
|
||||
warnings.append(
|
||||
f"oversight_policy '{metadata.oversight_policy}' is invalid "
|
||||
f"(expected one of {OVERSIGHT_POLICIES})"
|
||||
)
|
||||
|
||||
if metadata.risk_class not in RISK_CLASSES:
|
||||
warnings.append(
|
||||
f"risk_class '{metadata.risk_class}' is invalid "
|
||||
f"(expected one of {RISK_CLASSES})"
|
||||
)
|
||||
|
||||
# Model / provider consistency (only if the agent pins allowed values).
|
||||
agent_model = getattr(agent_definition, "llm_model", None)
|
||||
if metadata.allowed_models and agent_model:
|
||||
# Strip provider prefix for comparison (e.g. "openai/gpt-4o" -> "gpt-4o").
|
||||
bare_model = agent_model.split("/", 1)[-1]
|
||||
if agent_model not in metadata.allowed_models and bare_model not in metadata.allowed_models:
|
||||
warnings.append(
|
||||
f"agent model '{agent_model}' is not in allowed_models {metadata.allowed_models}"
|
||||
)
|
||||
|
||||
agent_provider = getattr(agent_definition, "provider", None)
|
||||
if metadata.allowed_providers and agent_provider:
|
||||
if agent_provider not in metadata.allowed_providers:
|
||||
warnings.append(
|
||||
f"agent provider '{agent_provider}' is not in allowed_providers "
|
||||
f"{metadata.allowed_providers}"
|
||||
)
|
||||
|
||||
# Oversight consistency.
|
||||
if metadata.oversight_policy == "always_required" and not metadata.human_review_required:
|
||||
warnings.append(
|
||||
"oversight_policy is 'always_required' but human_review_required is False"
|
||||
)
|
||||
if metadata.oversight_policy == "never" and metadata.human_review_required:
|
||||
warnings.append(
|
||||
"oversight_policy is 'never' but human_review_required is True"
|
||||
)
|
||||
|
||||
return warnings
|
||||
Reference in New Issue
Block a user