diff --git a/scripts/validation/validate-single-class-per-file.py b/scripts/validation/validate-single-class-per-file.py index d4cc27d8db..af3a5f34f3 100755 --- a/scripts/validation/validate-single-class-per-file.py +++ b/scripts/validation/validate-single-class-per-file.py @@ -186,6 +186,12 @@ def should_exclude_file(filepath: Path) -> bool: "model_database_secure_config.py", # Database config with nested models "model_service_registry_config.py", # Service config models "model_epic_state.py", # 3 tightly-coupled epic state models; split tracked in OMN-4401 + # Delegation wire DTO files (graduated from omnibase_compat, OMN-12126). + # These files group related intent/config DTOs by domain. Split tracked separately. + "model_orchestrator_intents.py", # 6 related orchestrator intent DTOs + "model_routing_config.py", # ModelTierModel + ModelRoutingTier + ModelDelegationConfig (hierarchy) + "model_quality_gate.py", # ModelQualityGateInput + ModelQualityGateResult (paired gate DTOs) + "model_bifrost_delegation_config.py", # 7 tightly-coupled Bifrost gateway config DTOs } # Exclude validator files with small helper classes diff --git a/src/omnibase_core/enums/enum_budget_action.py b/src/omnibase_core/enums/enum_budget_action.py new file mode 100644 index 0000000000..afad8c3fa0 --- /dev/null +++ b/src/omnibase_core/enums/enum_budget_action.py @@ -0,0 +1,20 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Budget policy outcome enum for the delegation compliance loop.""" + +from __future__ import annotations + +from enum import StrEnum + + +class EnumBudgetAction(StrEnum): + """Budget policy outcome used by the compliance loop.""" + + CONTINUE = "CONTINUE" + WARN = "WARN" + THROTTLE = "THROTTLE" + ABORT = "ABORT" + + +__all__: list[str] = ["EnumBudgetAction"] diff --git a/src/omnibase_core/models/delegation/wire/__init__.py b/src/omnibase_core/models/delegation/wire/__init__.py new file mode 100644 index 0000000000..f85a14c7bd --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/__init__.py @@ -0,0 +1,86 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Canonical delegation wire DTOs (graduated from omnibase_compat, OMN-12126).""" + +from omnibase_core.models.delegation.wire.model_bifrost_delegation_config import ( + ModelBifrostDelegationConfig, + ModelDelegationBackendConfig, + ModelDelegationCircuitBreakerConfig, + ModelDelegationFailoverConfig, + ModelDelegationFallbackPolicy, + ModelDelegationRoutingRule, + ModelDelegationShadowConfig, +) +from omnibase_core.models.delegation.wire.model_budget import ( + EnumBudgetAction, + ModelBudgetLimits, +) +from omnibase_core.models.delegation.wire.model_delegation_result import ( + ModelDelegationResult, +) +from omnibase_core.models.delegation.wire.model_delegation_wire_envelope import ( + ModelDelegationEventEnvelope, +) +from omnibase_core.models.delegation.wire.model_delegation_wire_request import ( + MAX_WORDS_PER_SENTENCE_RE, + SUPPORTED_ACCEPTANCE_CRITERIA, + EnumQualityContractMode, + ModelDelegationRequest, + validate_acceptance_criteria, +) +from omnibase_core.models.delegation.wire.model_orchestrator_intents import ( + ModelBaselineIntent, + ModelComplianceLoopResult, + ModelInferenceIntent, + ModelInferenceResponseData, + ModelQualityGateIntent, + ModelRoutingIntent, +) +from omnibase_core.models.delegation.wire.model_quality_gate import ( + EnumQualityGateCategory, + ModelQualityGateInput, + ModelQualityGateResult, +) +from omnibase_core.models.delegation.wire.model_routing_config import ( + ModelDelegationConfig, + ModelRoutingTier, + ModelTierModel, +) +from omnibase_core.models.delegation.wire.model_task_delegated_event import ( + TASK_DELEGATED_TOPIC_V1, + ModelTaskDelegatedEvent, +) + +__all__: list[str] = [ + "MAX_WORDS_PER_SENTENCE_RE", + "SUPPORTED_ACCEPTANCE_CRITERIA", + "TASK_DELEGATED_TOPIC_V1", + "EnumBudgetAction", + "EnumQualityContractMode", + "EnumQualityGateCategory", + "ModelBaselineIntent", + "ModelBifrostDelegationConfig", + "ModelBudgetLimits", + "ModelComplianceLoopResult", + "ModelDelegationBackendConfig", + "ModelDelegationCircuitBreakerConfig", + "ModelDelegationConfig", + "ModelDelegationEventEnvelope", + "ModelDelegationFailoverConfig", + "ModelDelegationFallbackPolicy", + "ModelDelegationRequest", + "ModelDelegationResult", + "ModelDelegationRoutingRule", + "ModelDelegationShadowConfig", + "ModelInferenceIntent", + "ModelInferenceResponseData", + "ModelQualityGateInput", + "ModelQualityGateIntent", + "ModelQualityGateResult", + "ModelRoutingIntent", + "ModelRoutingTier", + "ModelTaskDelegatedEvent", + "ModelTierModel", + "validate_acceptance_criteria", +] diff --git a/src/omnibase_core/models/delegation/wire/model_bifrost_delegation_config.py b/src/omnibase_core/models/delegation/wire/model_bifrost_delegation_config.py new file mode 100644 index 0000000000..3637bf11b1 --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_bifrost_delegation_config.py @@ -0,0 +1,259 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Bifrost delegation config wire DTOs.""" + +from __future__ import annotations + +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + + +class ModelDelegationShadowConfig(BaseModel): + """Shadow routing comparison settings.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + enabled: bool = Field(default=False, description="Whether shadow mode is active.") + policy_version: str = Field( + default="unknown", + max_length=128, + description="Human-readable version of the loaded shadow policy checkpoint.", + ) + log_sample_rate: float = Field( + default=1.0, + ge=0.0, + le=1.0, + description="Fraction of requests to log shadow decisions for.", + ) + comparison_logging_enabled: bool = Field( + default=True, + description="Whether to emit comparison events for shadow decisions.", + ) + max_shadow_latency_ms: float = Field( + default=5.0, + ge=0.1, + le=100.0, + description="Maximum allowed latency for shadow policy evaluation (ms).", + ) + shadow_label: Literal["SHADOW"] = Field( + default="SHADOW", + description="Label applied to all shadow comparison events.", + ) + + +class ModelDelegationFallbackPolicy(BaseModel): + """Per-rule fallback behavior when backend attempts fail.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + action: Literal["escalate_to_next_tier", "return_error"] = Field( + ..., + description="Action on backend failure: 'escalate_to_next_tier' or 'return_error'.", + ) + max_retries: int = Field( + default=1, + ge=0, + le=10, + description="Maximum retry attempts across backends in this rule.", + ) + on_exhaust: Literal["return_error", "escalate_to_next_tier"] = Field( + default="return_error", + description="Behavior when all retries are exhausted.", + ) + + +class ModelDelegationRoutingRule(BaseModel): + """Routing rule from task class constraints to ordered backend IDs.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + rule_id: UUID = Field( + ..., description="Stable UUID for audit logging and provenance." + ) + priority: int = Field( + default=100, + ge=0, + description="Evaluation order - lower values evaluated first.", + ) + task_class: str = Field(..., description="Task class this rule targets.") + # string-version-ok: external contract version string from YAML config, not a ModelSemVer object + task_class_contract_version: str = Field( + ..., + description="Version of the task class contract this rule was authored against.", + ) + # string-version-ok: external policy version string from YAML config, not a ModelSemVer object + backend_policy_version: str = Field( + ..., + description="Version of the backend policy applied by this rule.", + ) + match_operation_types: tuple[str, ...] = Field( + default_factory=tuple, + description="Operation types this rule matches (empty = any).", + ) + match_capabilities: tuple[str, ...] = Field( + default_factory=tuple, + description="Capabilities the request must declare (empty = any).", + ) + latency_sla_ms: int | None = Field( + default=None, + ge=1, + description="SLA latency constraint.", + ) + cost_ceiling_usd_per_1k_tokens: float | None = Field( + default=None, + ge=0.0, + description="Maximum allowed cost per 1K tokens for this rule's backends.", + ) + backend_ids: tuple[str, ...] = Field( + ..., + min_length=1, + description="Ordered backend IDs to try when this rule matches.", + ) + fallback_policy: ModelDelegationFallbackPolicy = Field( + ..., + description="Failover behavior when backends are exhausted.", + ) + shadow_policy_id: UUID = Field( + ..., description="Shadow policy UUID for A/B evaluation." + ) + + +class ModelDelegationBackendConfig(BaseModel): + """Backend definition for the Bifrost delegation gateway.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + # string-id-ok: backend_id is a human-readable slug (e.g. "local_qwen3"), not a UUID + backend_id: str = Field( + ..., min_length=1, description="Stable human-readable slug." + ) + base_url_env: str | None = Field( + default=None, + description="Env var name holding the backend base URL (local backends).", + ) + endpoint_url: str | None = Field( + default=None, + description=( + "Endpoint URL populated by deploy-time overlay. Null for local " + "backends until overlay is applied." + ), + ) + model_name: str | None = Field( + default=None, + description=( + "Model identifier sent in outbound requests. Null for local " + "backends resolved at deploy time." + ), + ) + api_key_env: str | None = Field( + default=None, + description="Optional environment variable name holding the backend API key.", + ) + extra_headers: dict[str, str] | None = Field( + default=None, + description="Optional static HTTP headers required by the backend provider.", + ) + tier: Literal["local", "frontier_api"] = Field( + ..., description="Routing tier: 'local' or 'frontier_api'." + ) + timeout_ms: int = Field( + default=30000, + ge=100, + le=600000, + description="Per-backend HTTP timeout in milliseconds.", + ) + capabilities: tuple[str, ...] = Field( + default_factory=tuple, + description="Capabilities this backend supports.", + ) + + +class ModelDelegationCircuitBreakerConfig(BaseModel): + """Circuit breaker settings applying to all Bifrost backends.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + failure_threshold: int = Field( + default=5, + ge=1, + le=100, + description="Consecutive failures that open the circuit.", + ) + window_seconds: int = Field( + default=30, + ge=1, + le=3600, + description="Cooldown duration after circuit opens, in seconds.", + ) + + +class ModelDelegationFailoverConfig(BaseModel): + """Gateway-level failover settings.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + max_attempts: int = Field( + default=3, + ge=1, + le=10, + description="Maximum backend attempts per request.", + ) + backoff_base_ms: int = Field( + default=500, + ge=0, + le=10000, + description="Base exponential backoff delay in milliseconds.", + ) + + +class ModelBifrostDelegationConfig(BaseModel): + """Bifrost delegation gateway configuration.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + # string-version-ok: config_version is a semver string from YAML, not a ModelSemVer object + config_version: str = Field(..., description="Semver config version.") + # string-version-ok: schema_version is a format identifier string, not a ModelSemVer object + schema_version: str = Field( + ..., description="Schema identifier for this config format." + ) + backends: tuple[ModelDelegationBackendConfig, ...] = Field( + ..., + min_length=1, + description="Backend definitions with deployed endpoint URLs.", + ) + routing_rules: tuple[ModelDelegationRoutingRule, ...] = Field( + ..., + min_length=1, + description="Routing rules evaluated in ascending priority order.", + ) + default_backends: tuple[str, ...] = Field( + default_factory=tuple, + description="Fallback backend IDs when no routing rule matches.", + ) + circuit_breaker: ModelDelegationCircuitBreakerConfig = Field( + default_factory=ModelDelegationCircuitBreakerConfig, + description="Circuit breaker settings applying to all backends.", + ) + failover: ModelDelegationFailoverConfig = Field( + default_factory=ModelDelegationFailoverConfig, + description="Gateway-level failover settings.", + ) + shadow_mode: ModelDelegationShadowConfig = Field( + default_factory=ModelDelegationShadowConfig, + description="Shadow mode configuration for delegation A/B testing.", + ) + + +__all__: list[str] = [ + "ModelBifrostDelegationConfig", + "ModelDelegationBackendConfig", + "ModelDelegationCircuitBreakerConfig", + "ModelDelegationFailoverConfig", + "ModelDelegationFallbackPolicy", + "ModelDelegationRoutingRule", + "ModelDelegationShadowConfig", +] diff --git a/src/omnibase_core/models/delegation/wire/model_budget.py b/src/omnibase_core/models/delegation/wire/model_budget.py new file mode 100644 index 0000000000..fe59311a9d --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_budget.py @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Budget limits wire DTO for delegation requests.""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.enums.enum_budget_action import EnumBudgetAction + +__all__: list[str] = ["EnumBudgetAction", "ModelBudgetLimits"] + + +class ModelBudgetLimits(BaseModel): + """Declared budget ceilings for a delegated task execution.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + max_tokens: int = Field(gt=0) + max_cost_usd: float = Field(gt=0.0) + max_time_s: float = Field(gt=0.0) diff --git a/src/omnibase_core/models/delegation/wire/model_delegation_result.py b/src/omnibase_core/models/delegation/wire/model_delegation_result.py new file mode 100644 index 0000000000..128247b0f3 --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_delegation_result.py @@ -0,0 +1,129 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Delegation result wire DTO.""" + +from __future__ import annotations + +from typing import Self +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +class ModelDelegationResult(BaseModel): + """Delegation outcome: content, quality status, model info, and metrics.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + correlation_id: UUID = Field( + ..., + description="Tracks this result back to the original request.", + ) + task_type: str = Field( + ..., description="The task classification from the original request." + ) + model_used: str = Field( + ..., + description="Name of the LLM model that produced the response.", + ) + endpoint_url: str = Field(..., description="URL of the LLM endpoint used.") + content: str = Field(..., description="The LLM-generated response content.") + quality_passed: bool = Field( + ..., + description="Whether the quality gate accepted the response.", + ) + quality_score: float = Field( + ..., + ge=0.0, + le=1.0, + description="Quality score from 0.0 to 1.0.", + ) + latency_ms: int = Field( + ..., ge=0, description="End-to-end latency in milliseconds." + ) + prompt_tokens: int = Field( + default=0, ge=0, description="Number of tokens in the prompt." + ) + completion_tokens: int = Field( + default=0, ge=0, description="Number of tokens in the completion." + ) + total_tokens: int = Field( + default=0, ge=0, description="Total tokens used (prompt + completion)." + ) + fallback_to_claude: bool = Field( + ..., + description="Whether fallback to Claude was triggered.", + ) + failure_reason: str = Field( + default="", + description="Reason for failure, empty string if successful.", + ) + tokens_to_compliance: int = Field( + default=0, + ge=0, + description="Total tokens across all compliance attempts.", + ) + compliance_attempts: int = Field( + default=1, + ge=1, + description="Number of LLM invocations to reach compliance.", + ) + escalation_count: int = Field( + default=0, + ge=0, + description="Number of tier-escalation attempts that occurred.", + ) + escalation_history: tuple[dict[str, object], ...] = Field( + default=(), + description="Serialized per-tier escalation attempt records.", + ) + terminal_failure_reason: str | None = Field( + default=None, + description="Terminal failure reason when delegation fails after escalation.", + ) + routing_tiers_hash: str | None = Field( + default=None, + description="SHA-256 of serialized routing_tiers.yaml at execution time.", + ) + escalation_config_hash: str | None = Field( + default=None, + description="SHA-256 of the escalation contract section at execution time.", + ) + attempts_count: int = Field( + default=1, + ge=1, + description="Total delegation attempts including the initial attempt.", + ) + cumulative_attempt_cost: float = Field( + default=0.0, + ge=0.0, + description="Total estimated cost across all attempts.", + ) + cumulative_input_tokens: int = Field( + default=0, + ge=0, + description="Total input tokens across all attempts.", + ) + cumulative_output_tokens: int = Field( + default=0, + ge=0, + description="Total output tokens across all attempts.", + ) + final_attempt_cost: float = Field( + default=0.0, + ge=0.0, + description="Estimated cost of the final attempt.", + ) + + @model_validator(mode="after") + def validate_total_tokens(self) -> Self: + """Keep token accounting internally consistent at the wire boundary.""" + expected_total = self.prompt_tokens + self.completion_tokens + if self.total_tokens != expected_total: + msg = "total_tokens must equal prompt_tokens + completion_tokens" + raise ValueError(msg) + return self + + +__all__: list[str] = ["ModelDelegationResult"] diff --git a/src/omnibase_core/models/delegation/wire/model_delegation_wire_envelope.py b/src/omnibase_core/models/delegation/wire/model_delegation_wire_envelope.py new file mode 100644 index 0000000000..812fb5b9de --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_delegation_wire_envelope.py @@ -0,0 +1,25 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Delegation event envelope wire DTO.""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict + +from omnibase_core.models.delegation.wire.model_delegation_result import ( + ModelDelegationResult, +) +from omnibase_core.topics import TopicBase + + +class ModelDelegationEventEnvelope(BaseModel): + """Topic plus delegation result payload envelope.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + topic: TopicBase + payload: ModelDelegationResult + + +__all__: list[str] = ["ModelDelegationEventEnvelope"] diff --git a/src/omnibase_core/models/delegation/wire/model_delegation_wire_request.py b/src/omnibase_core/models/delegation/wire/model_delegation_wire_request.py new file mode 100644 index 0000000000..93f75912f7 --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_delegation_wire_request.py @@ -0,0 +1,120 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Delegation request wire DTO (graduated from omnibase_compat, OMN-12126).""" + +from __future__ import annotations + +import re +from datetime import datetime +from typing import Literal, Self +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from omnibase_core.models.delegation.wire.model_budget import ModelBudgetLimits + +EnumQualityContractMode = Literal["extend_task_class", "replace_task_class"] + +SUPPORTED_ACCEPTANCE_CRITERIA = frozenset( + { + "exactly_two_sentences", + "plain_text_only", + "response_non_empty", + } +) +MAX_WORDS_PER_SENTENCE_RE = re.compile(r"^max_words_per_sentence_([1-9]\d*)$") + + +def validate_acceptance_criteria(criteria: tuple[str, ...]) -> tuple[str, ...]: + """Validate request-level quality criteria before they enter dispatch.""" + unsupported = [ + item + for item in criteria + if item not in SUPPORTED_ACCEPTANCE_CRITERIA + and not MAX_WORDS_PER_SENTENCE_RE.match(item) + ] + if unsupported: + joined = ", ".join(sorted(unsupported)) + # error-ok: wire DTO boundary check; pydantic model_validator surface, not an OnexError call site + raise ValueError(f"unsupported acceptance criteria: {joined}") + return criteria + + +class ModelDelegationRequest(BaseModel): + """Delegation command: prompt, task type, source context, and quality contract.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + prompt: str = Field( + ..., description="The user prompt to delegate to the local LLM." + ) + task_type: Literal["test", "document", "research"] = Field( + ..., + description="Classification of the delegation task.", + ) + # string-id-ok: Claude Code session IDs are opaque strings, not UUIDs + source_session_id: str | None = Field( + default=None, + description="Session that originated the delegation request.", + ) + source_file_path: str | None = Field( + default=None, + description="File context for the delegation, if any.", + ) + correlation_id: UUID = Field( + ..., + description="Unique identifier for tracking through the pipeline.", + ) + max_tokens: int = Field( + default=2048, + description="Maximum tokens for the LLM response.", + ) + emitted_at: datetime = Field( + ..., + description="Timestamp when the request was created.", + ) + output_schema_key: str | None = Field( + default=None, + description=( + "When set, the orchestrator runs the schema-compliance loop: it validates each " + "inference response against the registry-resolved schema and emits repair prompts " + "on validation failure. None = legacy single-attempt path." + ), + ) + compliance_budget: ModelBudgetLimits | None = Field( + default=None, + description=( + "Budget ceilings (tokens, cost, elapsed time) the compliance loop enforces between " + "repair attempts. Required when ``output_schema_key`` is set." + ), + ) + quality_contract_mode: EnumQualityContractMode = Field( + default="extend_task_class", + description="How request-level acceptance criteria interact with task-class DoD.", + ) + acceptance_criteria: tuple[str, ...] = Field( + default=(), + description="Request-level quality checks enforced by the quality gate.", + ) + + @model_validator(mode="after") + def _validate_compliance_loop_config(self) -> Self: + if self.output_schema_key is not None and self.compliance_budget is None: + msg = ( + "compliance_budget is required when output_schema_key is set " + "(the compliance loop has nothing to evaluate against without " + "token / cost / time ceilings)" + ) + raise ValueError(msg) + validate_acceptance_criteria(self.acceptance_criteria) + return self + + +__all__: list[str] = [ + "MAX_WORDS_PER_SENTENCE_RE", + "SUPPORTED_ACCEPTANCE_CRITERIA", + "EnumQualityContractMode", + "ModelDelegationRequest", + "validate_acceptance_criteria", +] diff --git a/src/omnibase_core/models/delegation/wire/model_orchestrator_intents.py b/src/omnibase_core/models/delegation/wire/model_orchestrator_intents.py new file mode 100644 index 0000000000..bc6dd376ff --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_orchestrator_intents.py @@ -0,0 +1,188 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Delegation orchestrator intent and response wire DTOs.""" + +from __future__ import annotations + +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.enums.enum_budget_action import EnumBudgetAction +from omnibase_core.models.delegation.wire.model_delegation_wire_request import ( + ModelDelegationRequest, +) +from omnibase_core.models.delegation.wire.model_quality_gate import ( + ModelQualityGateInput, +) + + +class ModelRoutingIntent(BaseModel): + """Intent sent from the orchestrator to the delegation routing reducer.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + intent: Literal["routing_reducer"] = Field(default="routing_reducer") + payload: ModelDelegationRequest + min_tier_name: str | None = Field( + default=None, + description=( + "When set, the routing reducer skips tiers appearing before this " + "tier in routing_tiers.yaml. Used by the escalation path to force " + "routing to a higher-cost tier after quality gate failure. " + "None means normal tier selection from the lowest eligible tier." + ), + ) + + +class ModelInferenceIntent(BaseModel): + """Intent sent from the orchestrator to the LLM inference effect.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + intent: Literal["llm_inference"] = Field(default="llm_inference") + base_url: str + model: str + system_prompt: str + prompt: str + max_tokens: int + temperature: float = Field(default=0.3, ge=0.0, le=2.0) + timeout_seconds: float = Field( + default=30.0, + ge=1.0, + le=600.0, + description="Backend-owned timeout for the downstream inference call.", + ) + correlation_id: UUID + api_key: str | None = Field( + default=None, + description="Resolved API key for authenticated model backends.", + ) + extra_headers: dict[str, str] | None = Field( + default=None, + description="Additional HTTP headers required by the selected backend.", + ) + + +class ModelQualityGateIntent(BaseModel): + """Intent sent from the orchestrator to the quality gate reducer.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + intent: Literal["quality_gate"] = Field(default="quality_gate") + payload: ModelQualityGateInput + + +class ModelBaselineIntent(BaseModel): + """Intent sent from the orchestrator for baseline cost comparison.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + intent: Literal["baseline_comparison"] = Field(default="baseline_comparison") + correlation_id: UUID = Field(..., description="Delegation correlation ID.") + task_type: str = Field(..., description="Task classification.") + baseline_cost_usd: float = Field( + ..., description="Estimated Claude cost for this task." + ) + candidate_cost_usd: float = Field( + default=0.0, + description="Actual local LLM cost (near-zero for self-hosted).", + ) + prompt_tokens: int = Field(default=0, ge=0, description="Prompt token count.") + completion_tokens: int = Field( + default=0, ge=0, description="Completion token count." + ) + total_tokens: int = Field(default=0, ge=0, description="Total token count.") + + +class ModelInferenceResponseData(BaseModel): + """Response data returned by the LLM inference effect.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + correlation_id: UUID = Field(..., description="Workflow correlation ID.") + content: str = Field(..., description="Generated text from the LLM.") + model_used: str = Field( + ..., description="Model identifier that produced the response." + ) + # string-id-ok: LLM call ID is an opaque upstream string (e.g. OpenAI "chatcmpl-..." format) + llm_call_id: str = Field( + default="", + description="Upstream LLM call ID for cost reconciliation (e.g. OpenAI id field).", + ) + latency_ms: int = Field( + default=0, ge=0, description="Inference latency in milliseconds." + ) + prompt_tokens: int = Field(default=0, ge=0, description="Prompt token count.") + completion_tokens: int = Field( + default=0, ge=0, description="Completion token count." + ) + total_tokens: int = Field(default=0, ge=0, description="Total token count.") + error_message: str = Field( + default="", + description="Failure reason when inference could not produce content.", + ) + + +class ModelComplianceLoopResult(BaseModel): + """One-shot outcome of evaluating a single LLM attempt for schema compliance.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + compliant: bool = Field( + ..., + description="True if the candidate output validated against the target schema.", + ) + validated_output: str = Field( + default="", + description=( + "The candidate output, exactly as supplied. Empty when the schema lookup failed " + "before validation could run." + ), + ) + tokens_to_compliance: int = Field( + default=0, + ge=0, + description="Total tokens consumed across all attempts so far (including this one).", + ) + compliance_attempts: int = Field( + default=1, + ge=1, + description=( + "Number of LLM attempts evaluated so far (including this one). The first call to " + "the loop is always attempt 1." + ), + ) + repair_prompt: str = Field( + default="", + description=( + "Prompt to feed back to the LLM for the next attempt when ``compliant`` is False " + "and ``budget_action`` is CONTINUE. Empty when the loop terminates." + ), + ) + budget_action: EnumBudgetAction = Field( + default=EnumBudgetAction.CONTINUE, + description=( + "Result of the budget-policy check after this attempt. CONTINUE means the " + "orchestrator may issue another repair attempt; ABORT means it must stop." + ), + ) + abort_reason: str = Field( + default="", + description=( + "Human-readable explanation when ``budget_action`` is ABORT or when the loop " + "terminated without compliance for another reason." + ), + ) + + +__all__: list[str] = [ + "ModelBaselineIntent", + "ModelComplianceLoopResult", + "ModelInferenceIntent", + "ModelInferenceResponseData", + "ModelQualityGateIntent", + "ModelRoutingIntent", +] diff --git a/src/omnibase_core/models/delegation/wire/model_quality_gate.py b/src/omnibase_core/models/delegation/wire/model_quality_gate.py new file mode 100644 index 0000000000..f1e9bf2d24 --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_quality_gate.py @@ -0,0 +1,109 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Delegation quality gate wire DTOs.""" + +from __future__ import annotations + +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.enums.enum_quality_gate_result import EnumQualityGateResult +from omnibase_core.models.delegation.wire.model_delegation_wire_request import ( + EnumQualityContractMode, +) + +EnumQualityGateCategory = EnumQualityGateResult + + +class ModelQualityGateInput(BaseModel): + """Gate input: LLM response content and expected quality markers.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + correlation_id: UUID = Field( + ..., + description="Tracks this input back to the original request.", + ) + task_type: str = Field( + ..., + description="The task classification for type-specific checks.", + ) + llm_response_content: str = Field( + ..., + description="The raw LLM response to evaluate.", + ) + expected_markers: tuple[str, ...] = Field( + default=(), + description="Strings expected in the response for the task type.", + ) + min_response_length: int = Field( + default=60, + ge=0, + description="Minimum acceptable response length in characters.", + ) + dod_deterministic: tuple[str, ...] = Field( + default=(), + description=( + "Deterministic DoD check names from the task-class contract (OMN-10614). " + "These checks BLOCK delegation result injection on failure." + ), + ) + dod_heuristic: tuple[str, ...] = Field( + default=(), + description=( + "Heuristic DoD check names from the task-class contract (OMN-10614). " + "These checks escalate per contract policy on failure." + ), + ) + quality_contract_mode: EnumQualityContractMode = Field( + default="extend_task_class", + description="How request-level acceptance criteria interact with task-class DoD.", + ) + acceptance_criteria: tuple[str, ...] = Field( + default=(), + description="Request-level quality checks enforced by the quality gate.", + ) + + +class ModelQualityGateResult(BaseModel): + """Gate output: pass/fail verdict, score, failure reasons, and fallback flag.""" + + model_config = ConfigDict(frozen=True, extra="forbid", from_attributes=True) + + correlation_id: UUID = Field( + ..., + description="Tracks this result back to the original request.", + ) + passed: bool = Field( + ..., description="Whether the LLM response passed the quality gate." + ) + fail_category: EnumQualityGateCategory = Field( + default=EnumQualityGateResult.PASS, + description=( + "Structured outcome: 'pass', 'fail_deterministic' (hard block), " + "or 'fail_heuristic' (escalate per contract policy)." + ), + ) + quality_score: float = Field( + ..., + ge=0.0, + le=1.0, + description="Quality score from 0.0 to 1.0.", + ) + failure_reasons: tuple[str, ...] = Field( + default=(), + description="Tuple of human-readable failure reason strings.", + ) + fallback_recommended: bool = Field( + default=False, + description="Whether fallback to Claude is recommended.", + ) + + +__all__: list[str] = [ + "EnumQualityGateCategory", + "ModelQualityGateInput", + "ModelQualityGateResult", +] diff --git a/src/omnibase_core/models/delegation/wire/model_routing_config.py b/src/omnibase_core/models/delegation/wire/model_routing_config.py new file mode 100644 index 0000000000..5f7b7ed51f --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_routing_config.py @@ -0,0 +1,79 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Delegation routing configuration wire DTOs. + +OMN-8596 owns ModelRoutingDecision; this module only carries routing config. +""" + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + + +class ModelTierModel(BaseModel): + """Model candidate inside a delegation routing tier.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + id: str = Field(..., description="Model identifier.") + backend_ref: str = Field( + ..., + description="Backend key in the deployed bifrost contract (bifrost_delegation.yaml).", + ) + max_context_tokens: int = Field( + ..., ge=1, description="Max context window in tokens." + ) + use_for: tuple[str, ...] = Field( + default_factory=tuple, + description="Task types this model handles.", + ) + fast_path_threshold_tokens: int | None = Field( + default=None, + ge=0, + description="If set, prefer this model when prompt tokens <= threshold.", + ) + + +class ModelRoutingTier(BaseModel): + """Ordered routing tier configuration.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str = Field(..., description="Tier name: local, cheap_cloud, or claude.") + models: tuple[ModelTierModel, ...] = Field( + default_factory=tuple, + description="Ordered list of candidate models in this tier.", + ) + eval_before_accept: bool = Field( + default=False, + description="Whether to run eval before accepting result from this tier.", + ) + eval_model: str | None = Field( + default=None, + description="Model to use for eval scoring (tier name required).", + ) + cost_per_1k_tokens: float = Field( + default=0.0, + ge=0.0, + description="Estimated tier cost per 1,000 tokens.", + ) + max_retries: int = Field( + default=0, + ge=0, + description="Max retry attempts within this tier before escalating.", + ) + + +class ModelDelegationConfig(BaseModel): + """Delegation routing policy configuration.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + tiers: tuple[ModelRoutingTier, ...] = Field( + default_factory=tuple, + description="Ordered escalation tiers.", + ) + + +__all__: list[str] = ["ModelDelegationConfig", "ModelRoutingTier", "ModelTierModel"] diff --git a/src/omnibase_core/models/delegation/wire/model_task_delegated_event.py b/src/omnibase_core/models/delegation/wire/model_task_delegated_event.py new file mode 100644 index 0000000000..5f1d6a2af7 --- /dev/null +++ b/src/omnibase_core/models/delegation/wire/model_task_delegated_event.py @@ -0,0 +1,123 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Task delegated event wire DTO.""" + +from __future__ import annotations + +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field + +from omnibase_core.topics import TopicBase + +TASK_DELEGATED_TOPIC_V1 = TopicBase.TASK_DELEGATED + + +class ModelTaskDelegatedEvent(BaseModel): + """Durable event emitted after a task is delegated to a model endpoint.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + topic: TopicBase = Field( + default=TASK_DELEGATED_TOPIC_V1, + description="Kafka topic this event should be routed to.", + ) + timestamp: str = Field(..., description="ISO-8601 timestamp.") + correlation_id: UUID = Field(..., description="Delegation correlation ID.") + session_id: UUID | None = Field(default=None, description="Source session ID.") + task_type: str = Field(..., description="Task classification.") + delegated_to: str = Field(..., description="Model/endpoint that handled the task.") + model_name: str = Field( + default="", + description="LLM model name from the routing decision (e.g. Qwen3-Coder-14B).", + ) + delegated_by: str = Field( + default="delegation-pipeline", + description="Source of the delegation.", + ) + quality_gate_passed: bool = Field( + ..., + description="Whether the quality gate accepted the response.", + ) + quality_gates_checked: list[str] = Field( + default_factory=lambda: ["length", "refusal", "markers"], + description="Names of quality gates checked.", + ) + quality_gates_failed: list[str] = Field( + default_factory=list, + description="Names of quality gates that failed.", + ) + cost_usd: float = Field( + default=0.0, + ge=0.0, + description="Estimated cost of local LLM inference (near-zero).", + ) + cost_savings_usd: float = Field( + default=0.0, + ge=0.0, + description="Estimated savings vs Claude.", + ) + delegation_latency_ms: int = Field( + default=0, + ge=0, + description="End-to-end delegation latency in ms.", + ) + is_shadow: bool = Field(default=False, description="Whether this was a shadow run.") + # string-id-ok: LLM call ID is an opaque upstream string (e.g. OpenAI "chatcmpl-..." format) + llm_call_id: str = Field( + default="", + description="Upstream LLM call ID for JOIN with llm_cost_aggregates.", + ) + tokens_to_compliance: int = Field( + default=0, + ge=0, + description="Total tokens across all compliance attempts.", + ) + compliance_attempts: int = Field( + default=1, + ge=1, + description="Number of LLM invocations to reach compliance.", + ) + prompt_text: str | None = Field( + default=None, description="Raw prompt sent to the delegated model." + ) + response_text: str | None = Field( + default=None, + description="Raw response received from the delegated model.", + ) + pricing_manifest_version: int = Field( + default=0, + ge=0, + description="Version of the pricing manifest used to compute cost_savings_usd.", + ) + escalation_count: int = Field( + default=0, + ge=0, + description="Number of tier-escalation attempts that occurred.", + ) + escalation_history: tuple[dict[str, object], ...] = Field( + default=(), + description="Serialized per-tier escalation attempt records.", + ) + routing_tiers_hash: str | None = Field( + default=None, + description="SHA-256 of serialized routing_tiers.yaml at execution time.", + ) + escalation_config_hash: str | None = Field( + default=None, + description="SHA-256 of the escalation contract section at execution time.", + ) + attempts_count: int = Field( + default=1, + ge=1, + description="Total delegation attempts including the initial attempt.", + ) + cumulative_attempt_cost: float = Field( + default=0.0, + ge=0.0, + description="Total estimated cost across all attempts.", + ) + + +__all__: list[str] = ["TASK_DELEGATED_TOPIC_V1", "ModelTaskDelegatedEvent"] diff --git a/tests/unit/models/delegation/wire/__init__.py b/tests/unit/models/delegation/wire/__init__.py new file mode 100644 index 0000000000..327f1172ed --- /dev/null +++ b/tests/unit/models/delegation/wire/__init__.py @@ -0,0 +1,2 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT diff --git a/tests/unit/models/delegation/wire/test_delegation_wire_models.py b/tests/unit/models/delegation/wire/test_delegation_wire_models.py new file mode 100644 index 0000000000..cb7c40cff5 --- /dev/null +++ b/tests/unit/models/delegation/wire/test_delegation_wire_models.py @@ -0,0 +1,522 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT + +"""Unit tests for graduated delegation wire DTOs (OMN-12126).""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from omnibase_core.models.delegation.wire import ( + TASK_DELEGATED_TOPIC_V1, + EnumBudgetAction, + ModelBaselineIntent, + ModelBifrostDelegationConfig, + ModelBudgetLimits, + ModelComplianceLoopResult, + ModelDelegationBackendConfig, + ModelDelegationConfig, + ModelDelegationEventEnvelope, + ModelDelegationFallbackPolicy, + ModelDelegationRequest, + ModelDelegationResult, + ModelDelegationRoutingRule, + ModelDelegationShadowConfig, + ModelInferenceIntent, + ModelInferenceResponseData, + ModelQualityGateInput, + ModelQualityGateIntent, + ModelQualityGateResult, + ModelRoutingIntent, + ModelRoutingTier, + ModelTaskDelegatedEvent, + ModelTierModel, + validate_acceptance_criteria, +) + + +@pytest.mark.unit +class TestModelBudget: + def test_enum_budget_action_values(self) -> None: + assert EnumBudgetAction.CONTINUE == "CONTINUE" + assert EnumBudgetAction.ABORT == "ABORT" + + def test_model_budget_limits_valid(self) -> None: + b = ModelBudgetLimits(max_tokens=100, max_cost_usd=0.01, max_time_s=30.0) + assert b.max_tokens == 100 + + def test_model_budget_limits_frozen(self) -> None: + b = ModelBudgetLimits(max_tokens=100, max_cost_usd=0.01, max_time_s=30.0) + with pytest.raises(Exception): + b.max_tokens = 200 # type: ignore[misc] + + +@pytest.mark.unit +class TestModelDelegationRequest: + def _make(self, **kwargs: object) -> ModelDelegationRequest: + defaults = { + "prompt": "Write a test", + "task_type": "test", + "correlation_id": uuid.uuid4(), + "emitted_at": datetime.now(tz=UTC), + } + defaults.update(kwargs) + return ModelDelegationRequest(**defaults) # type: ignore[arg-type] + + def test_basic_construction(self) -> None: + r = self._make() + assert r.task_type == "test" + + def test_compliance_budget_required_when_schema_key_set(self) -> None: + with pytest.raises(ValueError, match="compliance_budget is required"): + self._make(output_schema_key="some_schema") + + def test_compliance_budget_accepted_with_schema_key(self) -> None: + budget = ModelBudgetLimits(max_tokens=500, max_cost_usd=0.05, max_time_s=60.0) + r = self._make(output_schema_key="some_schema", compliance_budget=budget) + assert r.output_schema_key == "some_schema" + + def test_invalid_acceptance_criteria(self) -> None: + with pytest.raises(ValueError, match="unsupported acceptance criteria"): + self._make(acceptance_criteria=("not_a_real_criterion",)) + + def test_valid_acceptance_criteria(self) -> None: + r = self._make(acceptance_criteria=("response_non_empty",)) + assert "response_non_empty" in r.acceptance_criteria + + +@pytest.mark.unit +class TestValidateAcceptanceCriteria: + def test_valid(self) -> None: + result = validate_acceptance_criteria(("response_non_empty", "plain_text_only")) + assert "response_non_empty" in result + + def test_max_words_pattern(self) -> None: + result = validate_acceptance_criteria(("max_words_per_sentence_10",)) + assert result == ("max_words_per_sentence_10",) + + def test_invalid_raises(self) -> None: + with pytest.raises(ValueError, match="unsupported"): + validate_acceptance_criteria(("bad_criterion",)) + + +@pytest.mark.unit +class TestModelDelegationResult: + def test_construction(self) -> None: + r = ModelDelegationResult( + correlation_id=uuid.uuid4(), + task_type="test", + model_used="qwen3", + endpoint_url="http://localhost:8000", + content="result", + quality_passed=True, + quality_score=0.9, + latency_ms=100, + fallback_to_claude=False, + ) + assert r.quality_passed is True + assert r.escalation_count == 0 + + def test_escalation_fields_default(self) -> None: + r = ModelDelegationResult( + correlation_id=uuid.uuid4(), + task_type="test", + model_used="qwen3", + endpoint_url="http://localhost:8000", + content="result", + quality_passed=True, + quality_score=0.9, + latency_ms=100, + fallback_to_claude=False, + ) + assert r.escalation_history == () + assert r.terminal_failure_reason is None + assert r.attempts_count == 1 + + def test_rejects_invalid_metric_ranges(self) -> None: + with pytest.raises(ValidationError): + ModelDelegationResult( + correlation_id=uuid.uuid4(), + task_type="test", + model_used="qwen3", + endpoint_url="http://localhost:8000", + content="result", + quality_passed=True, + quality_score=1.1, + latency_ms=-1, + fallback_to_claude=False, + ) + + def test_rejects_inconsistent_total_tokens(self) -> None: + with pytest.raises(ValueError, match="total_tokens must equal"): + ModelDelegationResult( + correlation_id=uuid.uuid4(), + task_type="test", + model_used="qwen3", + endpoint_url="http://localhost:8000", + content="result", + quality_passed=True, + quality_score=0.9, + latency_ms=1, + prompt_tokens=2, + completion_tokens=3, + total_tokens=6, + fallback_to_claude=False, + ) + + +@pytest.mark.unit +class TestModelRoutingIntent: + def test_construction(self) -> None: + corr_id = uuid.uuid4() + req = ModelDelegationRequest( + prompt="test", + task_type="test", + correlation_id=corr_id, + emitted_at=datetime.now(tz=UTC), + ) + intent = ModelRoutingIntent(payload=req) + assert intent.intent == "routing_reducer" + assert intent.min_tier_name is None + + def test_min_tier_name(self) -> None: + corr_id = uuid.uuid4() + req = ModelDelegationRequest( + prompt="test", + task_type="test", + correlation_id=corr_id, + emitted_at=datetime.now(tz=UTC), + ) + intent = ModelRoutingIntent(payload=req, min_tier_name="cheap_cloud") + assert intent.min_tier_name == "cheap_cloud" + + def test_rejects_invalid_intent_literal(self) -> None: + corr_id = uuid.uuid4() + req = ModelDelegationRequest( + prompt="test", + task_type="test", + correlation_id=corr_id, + emitted_at=datetime.now(tz=UTC), + ) + with pytest.raises(ValidationError): + ModelRoutingIntent(intent="wrong", payload=req) + + +@pytest.mark.unit +class TestModelInferenceIntent: + def test_construction(self) -> None: + intent = ModelInferenceIntent( + base_url="http://localhost:8000", + model="qwen3", + system_prompt="You are helpful.", + prompt="Write a test", + max_tokens=512, + correlation_id=uuid.uuid4(), + ) + assert intent.intent == "llm_inference" + assert intent.timeout_seconds == 30.0 + assert intent.api_key is None + + def test_rejects_invalid_intent_literal(self) -> None: + with pytest.raises(ValidationError): + ModelInferenceIntent( + intent="wrong", + base_url="http://localhost:8000", + model="qwen3", + system_prompt="You are helpful.", + prompt="Write a test", + max_tokens=512, + correlation_id=uuid.uuid4(), + ) + + +@pytest.mark.unit +class TestModelRoutingTier: + def test_construction(self) -> None: + tier_model = ModelTierModel( + id="qwen3-14b", + backend_ref="local_qwen3", + max_context_tokens=32768, + use_for=("test",), + ) + tier = ModelRoutingTier(name="local", models=(tier_model,)) + assert tier.cost_per_1k_tokens == 0.0 + assert tier.max_retries == 0 + + def test_rejects_negative_routing_limits(self) -> None: + with pytest.raises(ValidationError): + ModelTierModel( + id="qwen3-14b", + backend_ref="local_qwen3", + max_context_tokens=0, + ) + + with pytest.raises(ValidationError): + ModelTierModel( + id="qwen3-14b", + backend_ref="local_qwen3", + max_context_tokens=32768, + fast_path_threshold_tokens=-1, + ) + + with pytest.raises(ValidationError): + ModelRoutingTier(name="local", max_retries=-1) + + +@pytest.mark.unit +class TestModelDelegationConfig: + def test_empty_tiers(self) -> None: + cfg = ModelDelegationConfig() + assert cfg.tiers == () + + +@pytest.mark.unit +class TestModelQualityGate: + def test_gate_input_construction(self) -> None: + corr_id = uuid.uuid4() + inp = ModelQualityGateInput( + correlation_id=corr_id, + task_type="test", + llm_response_content="This is the response.", + ) + assert inp.min_response_length == 60 + + def test_gate_input_rejects_negative_min_response_length(self) -> None: + with pytest.raises(ValidationError): + ModelQualityGateInput( + correlation_id=uuid.uuid4(), + task_type="test", + llm_response_content="This is the response.", + min_response_length=-1, + ) + + def test_gate_result_pass(self) -> None: + result = ModelQualityGateResult( + correlation_id=uuid.uuid4(), + passed=True, + quality_score=0.95, + ) + assert result.fail_category == "pass" + assert result.fallback_recommended is False + + def test_rejects_quality_score_outside_unit_interval(self) -> None: + with pytest.raises(ValidationError): + ModelQualityGateResult( + correlation_id=uuid.uuid4(), + passed=True, + quality_score=-0.1, + ) + + +@pytest.mark.unit +class TestModelComplianceLoopResult: + def test_construction(self) -> None: + r = ModelComplianceLoopResult(compliant=True) + assert r.budget_action == EnumBudgetAction.CONTINUE + assert r.compliance_attempts == 1 + + +@pytest.mark.unit +class TestModelBifrostDelegationConfig: + def _make_backend(self) -> ModelDelegationBackendConfig: + return ModelDelegationBackendConfig( + backend_id="local_qwen3", + tier="local", + ) + + def _make_rule(self) -> ModelDelegationRoutingRule: + return ModelDelegationRoutingRule( + rule_id=uuid.uuid4(), + task_class="test", + task_class_contract_version="1.0.0", + backend_policy_version="1.0.0", + backend_ids=("local_qwen3",), + fallback_policy=ModelDelegationFallbackPolicy(action="return_error"), + shadow_policy_id=uuid.uuid4(), + ) + + def test_construction(self) -> None: + cfg = ModelBifrostDelegationConfig( + config_version="1.0.0", + schema_version="bifrost/v1", + backends=(self._make_backend(),), + routing_rules=(self._make_rule(),), + ) + assert cfg.circuit_breaker.failure_threshold == 5 + assert cfg.failover.max_attempts == 3 + assert cfg.shadow_mode.enabled is False + + def test_shadow_config_defaults(self) -> None: + shadow = ModelDelegationShadowConfig() + assert shadow.shadow_label == "SHADOW" + assert shadow.log_sample_rate == 1.0 + + def test_rejects_invalid_bifrost_wire_literals(self) -> None: + with pytest.raises(ValidationError): + ModelDelegationFallbackPolicy(action="retry_forever") + + with pytest.raises(ValidationError): + ModelDelegationFallbackPolicy( + action="return_error", + on_exhaust="retry_forever", + ) + + with pytest.raises(ValidationError): + ModelDelegationBackendConfig(backend_id="local_qwen3", tier="unknown") + + +@pytest.mark.unit +class TestModelTaskDelegatedEvent: + def test_default_topic(self) -> None: + event = ModelTaskDelegatedEvent( + timestamp="2026-05-26T00:00:00Z", + correlation_id=uuid.uuid4(), + task_type="test", + delegated_to="local_qwen3", + quality_gate_passed=True, + ) + assert event.topic == TASK_DELEGATED_TOPIC_V1 + assert event.escalation_count == 0 + + def test_rejects_invalid_topic_and_negative_metrics(self) -> None: + with pytest.raises(ValidationError): + ModelTaskDelegatedEvent( + topic="not.a.registered.topic", + timestamp="2026-05-26T00:00:00Z", + correlation_id=uuid.uuid4(), + task_type="test", + delegated_to="local_qwen3", + quality_gate_passed=True, + ) + + with pytest.raises(ValidationError): + ModelTaskDelegatedEvent( + timestamp="2026-05-26T00:00:00Z", + correlation_id=uuid.uuid4(), + task_type="test", + delegated_to="local_qwen3", + quality_gate_passed=True, + cost_usd=-0.01, + ) + + def test_topic_constant(self) -> None: + assert TASK_DELEGATED_TOPIC_V1 == "onex.evt.omniclaude.task-delegated.v1" + + +@pytest.mark.unit +class TestModelDelegationEventEnvelope: + def test_construction(self) -> None: + result = ModelDelegationResult( + correlation_id=uuid.uuid4(), + task_type="test", + model_used="qwen3", + endpoint_url="http://localhost:8000", + content="ok", + quality_passed=True, + quality_score=0.9, + latency_ms=50, + fallback_to_claude=False, + ) + envelope = ModelDelegationEventEnvelope( + topic="onex.evt.omniclaude.task-delegated.v1", + payload=result, + ) + assert envelope.topic == "onex.evt.omniclaude.task-delegated.v1" + + def test_rejects_invalid_topic(self) -> None: + result = ModelDelegationResult( + correlation_id=uuid.uuid4(), + task_type="test", + model_used="qwen3", + endpoint_url="http://localhost:8000", + content="ok", + quality_passed=True, + quality_score=0.9, + latency_ms=50, + fallback_to_claude=False, + ) + with pytest.raises(ValidationError): + ModelDelegationEventEnvelope(topic="not.a.topic", payload=result) + + +@pytest.mark.unit +class TestModelInferenceResponseData: + def test_construction(self) -> None: + resp = ModelInferenceResponseData( + correlation_id=uuid.uuid4(), + content="Generated response.", + model_used="qwen3-14b", + ) + assert resp.error_message == "" + assert resp.latency_ms == 0 + + def test_rejects_negative_metrics(self) -> None: + with pytest.raises(ValidationError): + ModelInferenceResponseData( + correlation_id=uuid.uuid4(), + content="Generated response.", + model_used="qwen3-14b", + latency_ms=-1, + ) + + with pytest.raises(ValidationError): + ModelInferenceResponseData( + correlation_id=uuid.uuid4(), + content="Generated response.", + model_used="qwen3-14b", + prompt_tokens=-1, + ) + + +@pytest.mark.unit +class TestModelBaselineIntent: + def test_construction(self) -> None: + intent = ModelBaselineIntent( + correlation_id=uuid.uuid4(), + task_type="test", + baseline_cost_usd=0.01, + ) + assert intent.intent == "baseline_comparison" + assert intent.candidate_cost_usd == 0.0 + + def test_rejects_invalid_intent_and_negative_tokens(self) -> None: + with pytest.raises(ValidationError): + ModelBaselineIntent( + intent="wrong", + correlation_id=uuid.uuid4(), + task_type="test", + baseline_cost_usd=0.01, + ) + + with pytest.raises(ValidationError): + ModelBaselineIntent( + correlation_id=uuid.uuid4(), + task_type="test", + baseline_cost_usd=0.01, + prompt_tokens=-1, + ) + + +@pytest.mark.unit +class TestModelQualityGateIntent: + def test_construction(self) -> None: + gate_input = ModelQualityGateInput( + correlation_id=uuid.uuid4(), + task_type="test", + llm_response_content="This is a response.", + ) + intent = ModelQualityGateIntent(payload=gate_input) + assert intent.intent == "quality_gate" + + def test_rejects_invalid_intent_literal(self) -> None: + gate_input = ModelQualityGateInput( + correlation_id=uuid.uuid4(), + task_type="test", + llm_response_content="This is a response.", + ) + with pytest.raises(ValidationError): + ModelQualityGateIntent(intent="wrong", payload=gate_input)