"""Strict Pydantic schemas for provisions, context entries, triggers, memory, outcomes, and receipts.""" from __future__ import annotations import re from datetime import UTC, datetime from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator SYSTEM_BUILTIN_SCOPE = "system_builtin" PROV_ID_PATTERN = r"^PROV-[A-Z0-9][A-Z0-9-]{1,46}$" PROJECT_TAG_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9_.:/-]{0,63}$" ALLOWED_URI_SCHEMES = ("https://", "tessera://", "mcp://", "urn:") HarnessKind = Literal[ "muse-code", "opencode", "claude-code", "codex", "swarm-worker", "custom-harness", "personal-assistant", ] HarnessScope = Literal["route", "configure"] TriggerTier = Literal["keyword", "semantic", "micromodel"] WireFormat = Literal["chat_completions", "responses", "messages", "dry_run"] ProvisionAction = Literal["shim_context", "respond_directly"] ContextEntryKind = Literal["noun", "verb", "detail"] EntityKind = Literal[ "file_path", "code_symbol", "tessera_uri", "statute_card_id", "error_signature", ] OutcomeKind = Literal["accepted", "retry_error", "overridden"] CallerProfileId = Literal[ "coding_harness", "chat_assistant", "swarm_worker", "ngo_caseworker", "steward_reviewer", "benchmark_auditor", "incident_responder", "public_researcher", ] RankingStrategy = Literal[ "empirical_bayes", "borda_consensus", "ucb1_explore", "wilson_lower_bound", "pareto_efficiency", "urgency_priority", ] class ProvisionURI(BaseModel): """Authoritative URI pointer attached when a provision triggers.""" model_config = ConfigDict(extra="forbid") uri: str = Field(min_length=6, max_length=512) label: str = Field(min_length=2, max_length=140) kind: Literal["doc", "schema", "mcp_resource", "statute_card", "store_record"] = "doc" inline_excerpt: str = Field(default="", max_length=800) @field_validator("uri") @classmethod def _validate_uri_scheme(cls, value: str) -> str: trimmed = value.strip() if not trimmed.startswith(ALLOWED_URI_SCHEMES): allowed = ", ".join(ALLOWED_URI_SCHEMES) raise ValueError(f"URI {trimmed!r} must start with one of: {allowed}") return trimmed class ContextEntry(BaseModel): """Atomic noun, verb, or detail entry stored in Herma's user-configured context database.""" model_config = ConfigDict(extra="forbid") governance: dict[str, Any] | None = None entry_id: str = Field(min_length=2, max_length=160) kind: ContextEntryKind title: str = Field(min_length=2, max_length=160) summary: str = Field(min_length=4, max_length=600) schema_or_body: dict[str, Any] | str = "" source_uri: str | None = None version: int = Field(default=1, ge=1) updated_at: str = Field(default_factory=lambda: datetime.now(UTC).isoformat()) @field_validator("entry_id") @classmethod def _validate_entry_id(cls, value: str) -> str: trimmed = value.strip() if len(trimmed) < 2: raise ValueError("entry_id must contain at least 2 non-whitespace characters.") return trimmed @field_validator("source_uri") @classmethod def _validate_source_uri(cls, value: str | None) -> str | None: if value is None: return None trimmed = value.strip() if not trimmed: return None if not trimmed.startswith(ALLOWED_URI_SCHEMES): allowed = ", ".join(ALLOWED_URI_SCHEMES) raise ValueError(f"source_uri {trimmed!r} must start with one of: {allowed}") return trimmed class KeywordTrigger(BaseModel): """Tier-1 deterministic trigger (~0 ms): keywords, regexes, file globs, and tool names.""" model_config = ConfigDict(extra="forbid") any_keywords: list[str] = Field(default_factory=list) all_keywords: list[str] = Field(default_factory=list) regex_patterns: list[str] = Field(default_factory=list) path_globs: list[str] = Field(default_factory=list) tool_names: list[str] = Field(default_factory=list) case_sensitive: bool = False @field_validator("regex_patterns") @classmethod def _compile_regexes(cls, patterns: list[str]) -> list[str]: for pat in patterns: if len(pat) > 240: raise ValueError("Regex pattern exceeds 240 characters.") re.compile(pat) return patterns @model_validator(mode="after") def _non_empty(self) -> KeywordTrigger: if not ( self.any_keywords or self.all_keywords or self.regex_patterns or self.path_globs or self.tool_names ): raise ValueError( "KeywordTrigger requires at least one keyword, regex, path_glob, or tool_name." ) return self class SemanticTrigger(BaseModel): """Tier-2 semantic similarity trigger (~1-5 ms): exemplar phrases and concept clusters.""" model_config = ConfigDict(extra="forbid") exemplars: list[str] = Field(min_length=1) concepts: list[str] = Field(default_factory=list) threshold: float = Field(default=0.26, ge=0.05, le=0.99) class MicromodelTrigger(BaseModel): """Tier-3 micromodel router trigger: lightweight classifier prompt with offline fallback.""" model_config = ConfigDict(extra="forbid") intent_question: str = Field(min_length=8, max_length=400) positive_labels: list[str] = Field(default_factory=lambda: ["apply", "yes", "true"]) min_confidence: float = Field(default=0.60, ge=0.1, le=1.0) fallback_keywords: list[str] = Field(default_factory=list) class TriggerSpec(BaseModel): """Composable 3-tier trigger specification for a provision.""" model_config = ConfigDict(extra="forbid") mode: Literal["any", "all"] = "any" keyword: KeywordTrigger | None = None semantic: SemanticTrigger | None = None micromodel: MicromodelTrigger | None = None @model_validator(mode="after") def _require_one_tier(self) -> TriggerSpec: if self.keyword is None and self.semantic is None and self.micromodel is None: raise ValueError( "TriggerSpec requires at least one of keyword, semantic, or micromodel." ) return self class Provision(BaseModel): """Versioned rule/provision record that shims context or responds directly before LLM execution.""" model_config = ConfigDict(extra="forbid") prov_id: str = Field(pattern=PROV_ID_PATTERN) title: str = Field(min_length=3, max_length=140) summary: str = Field(min_length=6, max_length=400) priority: int = Field(default=100, ge=1, le=1000) enabled: bool = True action: ProvisionAction = "shim_context" direct_response: str | None = Field(default=None, max_length=12000) context_refs: list[str] = Field(default_factory=list) inject_position: Literal["system_prefix", "system_suffix", "developer_context"] = ( "system_suffix" ) dedupe_mode: Literal["once_per_conversation", "every_turn"] = "once_per_conversation" max_chars: int = Field(default=2400, ge=120, le=16000) inject_text: str = Field(default="", max_length=12000) uris: list[ProvisionURI] = Field(default_factory=list) metadata_tags: dict[str, str] = Field(default_factory=dict) triggers: TriggerSpec @field_validator("context_refs") @classmethod def _validate_context_refs(cls, refs: list[str]) -> list[str]: cleaned: list[str] = [] for raw in refs: ref = raw.strip() if len(ref) < 2 or len(ref) > 160: raise ValueError("Each context_refs entry must be between 2 and 160 characters.") if ref not in cleaned: cleaned.append(ref) return cleaned @model_validator(mode="after") def _validate_action_fields(self) -> Provision: if self.action == "respond_directly": if self.direct_response is None or not self.direct_response.strip(): raise ValueError( "Provision with action='respond_directly' requires a non-empty direct_response." ) elif len(self.inject_text.strip()) < 10: raise ValueError( "Provision with action='shim_context' requires inject_text of at least 10 characters." ) return self class StoredProvision(BaseModel): """Provision wrapped with caller_scope, compare-and-set version, and SHA-256 content hash.""" model_config = ConfigDict(extra="forbid") caller_scope: str = SYSTEM_BUILTIN_SCOPE provision: Provision version: int = Field(ge=1) content_sha256: str = Field(min_length=64, max_length=64) source_store: str = "local" updated_at: float class TriggerMatch(BaseModel): """Detailed explanation of why a provision fired on an outbound request.""" model_config = ConfigDict(extra="forbid") prov_id: str version: int priority: int matched_tiers: list[TriggerTier] reasons: list[str] score: float class HarnessKeyInfo(BaseModel): """Public metadata for an issued configurable-harness credential (never contains the secret).""" model_config = ConfigDict(extra="forbid") key_id: str label: str harness_kind: HarnessKind scopes: list[HarnessScope] created_at: float revoked_at: float | None = None class InjectionReceipt(BaseModel): """Cryptographic audit receipt for a single shimmed or directly answered LLM request.""" model_config = ConfigDict(extra="forbid") receipt_id: str receipt_sha256: str harness_kind: HarnessKind key_id: str project_tag: str = "default" wire_format: WireFormat model: str provisions_fired: list[str] tiers_matched: list[TriggerTier] uris_injected: list[str] injected_chars: int deduped_skipped: list[str] responded_directly: bool = False direct_response_text: str | None = None shimmed_context_refs: list[str] = Field(default_factory=list) entities_tracked: int = 0 caller_profile: CallerProfileId = "coding_harness" ranking_strategy: RankingStrategy = "empirical_bayes" created_at: float ShimReceipt = InjectionReceipt class ExtractedEntity(BaseModel): """Pure deterministically extracted entity from an outbound coding-harness turn.""" model_config = ConfigDict(extra="forbid") kind: EntityKind value: str = Field(min_length=2, max_length=256) class WorkingEntity(BaseModel): """Caller-isolated top-of-mind entity tracked per (caller_scope, project_tag).""" model_config = ConfigDict(extra="forbid") caller_scope: str project_tag: str = "default" entity_kind: EntityKind entity_value: str hit_count: int = Field(ge=1) first_seen_at: float last_seen_at: float last_receipt_id: str class OutcomeReportRequest(BaseModel): """Explicit operator or harness feedback on a shimmed or directly answered turn.""" model_config = ConfigDict(extra="forbid") receipt_id: str = Field(min_length=4, max_length=96) prov_id: str | None = Field(default=None, pattern=PROV_ID_PATTERN) outcome: OutcomeKind project_tag: str | None = Field(default=None, pattern=PROJECT_TAG_PATTERN) tokens_saved_est: int = Field(default=0, ge=0, le=1_000_000) latency_ms: float = Field(default=0.0, ge=0.0, le=600_000.0) error_signature: str | None = Field(default=None, max_length=240) class RuleOutcomeRecord(BaseModel): """Recorded rule outcome row for self-improving telemetry.""" model_config = ConfigDict(extra="forbid") outcome_id: str caller_scope: str project_tag: str receipt_id: str prov_id: str outcome: OutcomeKind tokens_saved_est: int = 0 latency_ms: float = 0.0 error_signature: str | None = None created_at: float class RuleTelemetry(BaseModel): """Aggregated telemetry per provision within a caller scope.""" model_config = ConfigDict(extra="forbid") prov_id: str fires: int accepted_count: int retry_error_count: int tokens_saved_est: int retry_error_rate: float avg_latency_ms: float class InsightsResponse(BaseModel): """Caller-isolated top-of-mind entity memory, rule telemetry, and reviewable candidate rules.""" model_config = ConfigDict(extra="forbid") caller_scope: str project_tag: str | None = None top_entities: list[WorkingEntity] rule_telemetry: list[RuleTelemetry] candidate_provisions: list[Provision] class DryRunRequest(BaseModel): """Harness dry-run request to inspect trigger matches and augmented payload without calling an LLM.""" model_config = ConfigDict(extra="forbid") wire_format: Literal["chat_completions", "responses", "messages"] = "chat_completions" payload: dict[str, Any] workspace_paths: list[str] = Field(default_factory=list) caller_profile: CallerProfileId | None = None class DryRunResponse(BaseModel): """Structured dry-run result showing augmented payload, matches, shimmed refs, and direct response.""" model_config = ConfigDict(extra="forbid") augmented_payload: dict[str, Any] matches: list[TriggerMatch] receipt: InjectionReceipt responded_directly: bool = False direct_response_text: str | None = None shimmed_context_refs: list[str] = Field(default_factory=list) SURFACE_ID_PATTERN = r"^[a-z0-9][a-z0-9_-]{0,47}$" class HandshakeRequest(BaseModel): """Surface Handshake v1.0 negotiation request sent by harnesses or product surfaces.""" model_config = ConfigDict(extra="forbid") protocol_version: str = Field(default="1.0", max_length=16) surface_id: str | None = Field(default=None, pattern=SURFACE_ID_PATTERN) surface_version: str = Field(default="0.1.0", max_length=64) requested_max_chars: int = Field(default=4000, ge=0, le=10_000_000) caller_profile: CallerProfileId | None = None class HandshakeResponse(BaseModel): """Negotiated Surface Handshake v1.0 receipt returned by Herma middleware.""" model_config = ConfigDict(extra="forbid") protocol_version: str = "1.0" backend: str = "herma" backend_version: str surface_id: str harness_kind: HarnessKind caller_profile: CallerProfileId = "coding_harness" ranking_strategy: RankingStrategy = "empirical_bayes" profile_max_chars: int = 6000 clamped_max_chars: int active_provisions: int context_entries: int transactional_db_exempt_from_middleware: bool = True capabilities: list[str] = Field(default_factory=list) handshake_id: str class ProfileCompareRequest(BaseModel): """Zero-LLM side-by-side comparison request across caller profile shimming lenses.""" model_config = ConfigDict(extra="forbid") prompt: str = Field(min_length=1, max_length=16_000) profiles: list[CallerProfileId] = Field(default_factory=list, max_length=8) wire_format: Literal["chat_completions", "responses", "messages"] = "chat_completions" model: str = Field(default="muse-spark-open", max_length=120) workspace_paths: list[str] = Field(default_factory=list, max_length=32) class ProfileShimDiff(BaseModel): """Deterministic shim result for a single CallerProfile lens.""" model_config = ConfigDict(extra="forbid") caller_profile: CallerProfileId ranking_strategy: RankingStrategy max_chars: int suppress_code_dumps: bool profile_guidance_header: str injected_chars: int selected_provisions: list[str] = Field(default_factory=list) provisions_fired: list[str] = Field(default_factory=list) selected_context_refs: list[str] = Field(default_factory=list) uris_injected: list[str] = Field(default_factory=list) responded_directly: bool = False direct_response_text: str | None = None shimmed_preview: str = "" receipt_sha256: str class ProfileCompareResponse(BaseModel): """Side-by-side deterministic shim comparison across CallerProfile lenses (0 LLM calls).""" model_config = ConfigDict(extra="forbid") ok: bool = True persisted: bool = False upstream_llm_calls: int = 0 scope_evaluated: str = SYSTEM_BUILTIN_SCOPE prompt: str wire_format: Literal["chat_completions", "responses", "messages"] = "chat_completions" comparisons: list[ProfileShimDiff] by_profile: dict[str, ProfileShimDiff] = Field(default_factory=dict)