This repository has been archived on 2026-04-16. You can view files and clone it, but cannot push or open issues or pull requests.
llm-principal-broker/llm_broker/models.py
Tyler King 944b3fde19 feat: LLM Principal Broker MVP
FastAPI companion service to the GSAP broker for AI agent
identity delegation in governed shell sessions.

Implements GCAP-SPEC-LLM-PRINCIPAL-BROKER-0001:
  POST /delegate — request delegation (human → AI agent)
  POST /delegate/{id}/revoke — revoke delegation
  GET  /delegate/{id} — delegation status
  GET  /agents — list active delegations
  GET  /health — health check

Delegation lifecycle:
  REQUESTED → ACTIVE → EXPIRED | REVOKED
  Cascading revocation on delegator AC revocation
  Background cleanup of expired delegations (30s interval)

Keycloak integration:
  Registers ephemeral agent clients per delegation
  Deletes clients on revocation/expiry
  Dev mode: stubs when no client_secret configured

GSAP broker integration:
  Requests delegated ACs via on_behalf_of pattern
  Scope narrowing: agent ceiling ≤ delegator ceiling
  Dev mode: stubs when no bearer_token configured

Chronicle integration:
  DELEGATION_CREATED (0x3001)
  DELEGATION_REVOKED (0x3003)
  DELEGATION_EXPIRED (0x3004)

All 7 smoke tests pass (health, create, list, query, revoke, verify, empty).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-04 16:28:37 -04:00

100 lines
2.3 KiB
Python

"""Pydantic models — GCAP-SPEC-LLM-PRINCIPAL-BROKER-0001 §3, §8."""
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
from enum import Enum
from uuid import UUID
class DelegationStatus(str, Enum):
REQUESTED = "requested"
ACTIVE = "active"
EXPIRED = "expired"
REVOKED = "revoked"
class DelegationScope(BaseModel):
inherit_corpus: bool = True
inherit_contexts: bool = True
capability_ceiling: str = "CAP_MUTATE"
ceremony_required_for: list[str] = Field(
default_factory=lambda: ["delete", "destroy", "drop"]
)
prohibited_commands: list[str] = Field(default_factory=list)
max_ttl_minutes: int = 60
max_commands: int = 500
class DelegationRequest(BaseModel):
"""POST /delegate request body — §8.1."""
delegator_ac_id: str
agent_type: str = "claude-code"
agent_model: Optional[str] = None
scope: Optional[DelegationScope] = None
accord_template: str = "ai-delegation-standard"
class AgentPrincipal(BaseModel):
did: str
keycloak_client_id: str
display_name: str
class DelegatedAC(BaseModel):
context_id: str
delegator_did: str
agent_did: str
capability_ceiling: str
expires_at: str
delegation_chain: list[dict]
class DelegationResponse(BaseModel):
"""POST /delegate response — §3.2."""
delegation_id: str
agent_principal: AgentPrincipal
delegated_ac: dict
agent_token: str
expires_at: str
max_commands: int
chronicle_cid: Optional[str] = None
class DelegationInfo(BaseModel):
"""GET /delegate/{id} response — §8.3."""
delegation_id: str
status: DelegationStatus
agent_did: str
agent_type: str
delegator_did: str
commands_executed: int
commands_remaining: int
ttl_remaining_seconds: int
created_at: str
expires_at: str
class RevokeRequest(BaseModel):
reason: str = "manual_revocation"
class RevokeResponse(BaseModel):
delegation_id: str
status: str = "revoked"
reason: str
chronicle_cid: Optional[str] = None
class ActiveDelegation(BaseModel):
delegation_id: str
agent_type: str
delegator: str
commands_executed: int
ttl_remaining_seconds: int
status: str
class AgentListResponse(BaseModel):
active_delegations: list[ActiveDelegation]
total_active: int