diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c986a728..3bd084b10 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and ## [Unreleased] +### Added + +- Add a transport-neutral, versioned model fallback policy that validates explicit cost tiers and deterministically exhausts eligible free candidates before any paid fallback. +- Filter fallback candidates by repository visibility, required capability, and configured credential name without retaining or serializing secret values. +- Add a standard-library CLI for immutable cross-repository workflow integration, with complete statement, branch, and public-docstring coverage for the fallback policy. + ### Security - Pin each HTTPS provider connection to the exact public addresses approved during validation, preserve the original hostname for TLS verification, bypass environment proxy resolution, and reject redirects to close DNS-rebinding and credential-forwarding SSRF paths. @@ -15,3 +21,7 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and ### Changed - Pin Atheris by Python interpreter so the Python 3.11 fuzz job and the newer central coverage-evidence image both install a published, hash-locked wheel. + +### Documentation + +- Add APA 7 doctoring for Python environment-marker semantics, Atheris artifact availability and hashes, and the supported-platform uncertainty boundary. diff --git a/contextual_orchestrator/__init__.py b/contextual_orchestrator/__init__.py index 79f7e3cba..eaa5d1d4c 100644 --- a/contextual_orchestrator/__init__.py +++ b/contextual_orchestrator/__init__.py @@ -37,6 +37,18 @@ from .cost_router import CostRoutingCoordinator from .credentials import NotConfigured, get_credential, register_credential from .kv_config import InMemoryConfigStore, get_config_store +from .model_fallback import ( + CandidateValidationError, + CostTier, + FallbackCandidate, + FallbackContext, + FallbackManifestError, + FallbackPlan, + NoEligibleCandidateError, + SkippedCandidate, + build_fallback_plan, + load_fallback_manifest, +) from .orchestrator import ModelAgent, ModelClient as _ModelClient, TaskOrchestrator, WorkflowStep, load_agents from .provider_transport import install_provider_transport as _install_provider_transport from .token_counting import HeuristicTokenCounter, build_token_counter @@ -90,4 +102,15 @@ "build_embeddings_jsonl_body", "cheapest_upstream", "CostRoutingCoordinator", + # transport-neutral model fallback policy + "CandidateValidationError", + "CostTier", + "FallbackCandidate", + "FallbackContext", + "FallbackManifestError", + "FallbackPlan", + "NoEligibleCandidateError", + "SkippedCandidate", + "build_fallback_plan", + "load_fallback_manifest", ] diff --git a/contextual_orchestrator/_fallback_cli.py b/contextual_orchestrator/_fallback_cli.py new file mode 100644 index 000000000..52e23af54 --- /dev/null +++ b/contextual_orchestrator/_fallback_cli.py @@ -0,0 +1,103 @@ +"""Command-line adapter for the transport-neutral fallback policy.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +from typing import Any, Mapping, Sequence + +from ._fallback_manifest import load_fallback_manifest +from ._fallback_plan import build_fallback_plan +from ._fallback_types import ( + ALLOWED_VISIBILITIES, + FallbackContext, + FallbackManifestError, + validate_credentials, +) + + +def _load_manifest_path(path: Path) -> Mapping[str, Any]: + """Read a UTF-8 JSON manifest and normalize input errors.""" + try: + raw = path.read_text(encoding="utf-8") + except OSError as exc: + raise FallbackManifestError( + f"manifest could not be read: {path}" + ) from exc + try: + document = json.loads(raw) + except json.JSONDecodeError as exc: + raise FallbackManifestError( + "manifest must contain valid JSON" + ) from exc + if not isinstance(document, Mapping): + raise FallbackManifestError("manifest must be an object") + return document + + +def _configured_credentials(names: Sequence[str]) -> frozenset[str]: + """Return names whose environment values are non-whitespace.""" + validate_credentials(tuple(names)) + return frozenset( + name for name in names if os.environ.get(name, "").strip() + ) + + +def _build_parser() -> argparse.ArgumentParser: + """Build the parser for policy-only workflow integration.""" + parser = argparse.ArgumentParser(prog="contextual-model-fallback") + subparsers = parser.add_subparsers(dest="command", required=True) + plan_parser = subparsers.add_parser("plan") + plan_parser.add_argument("--manifest", type=Path, required=True) + plan_parser.add_argument("--agent", required=True) + plan_parser.add_argument( + "--repository-visibility", + choices=sorted(ALLOWED_VISIBILITIES), + default="public", + ) + plan_parser.add_argument( + "--credential-env", action="append", default=[] + ) + plan_parser.add_argument( + "--required-capability", action="append", default=[] + ) + plan_parser.add_argument("--deny-paid", action="store_true") + plan_parser.add_argument( + "--format", + choices=("json", "ids", "models"), + default="json", + ) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + """Render one validated plan and return an exit code.""" + args = _build_parser().parse_args(argv) + if args.command != "plan": # pragma: no cover - argparse constrains it. + raise FallbackManifestError( + f"unsupported command: {args.command}" + ) + document = _load_manifest_path(args.manifest) + candidates = load_fallback_manifest(document, args.agent) + context = FallbackContext( + repository_visibility=args.repository_visibility, + available_credentials=_configured_credentials(args.credential_env), + required_capabilities=frozenset(args.required_capability), + allow_paid=not args.deny_paid, + ) + plan = build_fallback_plan(candidates, context=context) + if args.format == "json": + print( + json.dumps( + plan.to_public_dict(), + sort_keys=True, + separators=(",", ":"), + ) + ) + elif args.format == "ids": + print(" ".join(plan.candidate_ids)) + else: + print(" ".join(candidate.model for candidate in plan.candidates)) + return 0 diff --git a/contextual_orchestrator/_fallback_manifest.py b/contextual_orchestrator/_fallback_manifest.py new file mode 100644 index 000000000..60458fbdf --- /dev/null +++ b/contextual_orchestrator/_fallback_manifest.py @@ -0,0 +1,157 @@ +"""Strict versioned manifest parsing for shared model fallback policy.""" + +from __future__ import annotations + +from typing import Any, Mapping + +from ._fallback_plan import validate_candidate_collection +from ._fallback_types import ( + AGENT_NAME_RE, + ALLOWED_VISIBILITIES, + SCHEMA_VERSION, + CandidateValidationError, + CostTier, + FallbackCandidate, + FallbackManifestError, + joined, +) + +_MANIFEST_KEYS = frozenset({"schema_version", "agents"}) +_AGENT_KEYS = frozenset({"candidates"}) +_CANDIDATE_KEYS = frozenset( + { + "candidate_id", + "provider", + "model", + "cost_tier", + "priority", + "required_credentials", + "repository_visibilities", + "capabilities", + } +) + + +def load_fallback_manifest( + document: Mapping[str, Any], agent: str +) -> tuple[FallbackCandidate, ...]: + """Parse one agent's candidate list from a strict manifest.""" + if not isinstance(document, Mapping): + raise FallbackManifestError("manifest must be an object") + unknown_manifest_keys = set(document) - _MANIFEST_KEYS + if unknown_manifest_keys: + raise FallbackManifestError( + f"unknown manifest keys: {joined(unknown_manifest_keys)}" + ) + if document.get("schema_version") != SCHEMA_VERSION: + raise FallbackManifestError( + f"schema_version must be {SCHEMA_VERSION}" + ) + agents = document.get("agents") + if not isinstance(agents, Mapping): + raise FallbackManifestError("agents must be an object") + for agent_name in agents: + if not isinstance(agent_name, str) or not AGENT_NAME_RE.fullmatch( + agent_name + ): + raise FallbackManifestError( + "agent name must be a safe identifier" + ) + if agent not in agents: + raise FallbackManifestError( + f"agent {agent!r} was not found in manifest" + ) + agent_document = agents[agent] + if not isinstance(agent_document, Mapping): + raise FallbackManifestError( + f"agent {agent!r} must be an object" + ) + unknown_agent_keys = set(agent_document) - _AGENT_KEYS + if unknown_agent_keys: + raise FallbackManifestError( + f"unknown agent keys: {joined(unknown_agent_keys)}" + ) + raw_candidates = agent_document.get("candidates") + if not isinstance(raw_candidates, list): + raise FallbackManifestError("candidates must be an array") + if not raw_candidates: + raise FallbackManifestError( + "agent must declare at least one candidate" + ) + + parsed: list[FallbackCandidate] = [] + for raw_candidate in raw_candidates: + if not isinstance(raw_candidate, Mapping): + raise FallbackManifestError("candidate must be an object") + parsed.append(_parse_candidate(raw_candidate)) + try: + validate_candidate_collection(tuple(parsed)) + except CandidateValidationError as exc: + raise FallbackManifestError(str(exc)) from exc + return tuple(parsed) + + +def _parse_candidate( + raw_candidate: Mapping[str, Any] +) -> FallbackCandidate: + """Parse one strict candidate object into an immutable value object.""" + unknown_candidate_keys = set(raw_candidate) - _CANDIDATE_KEYS + if unknown_candidate_keys: + raise FallbackManifestError( + f"unknown candidate keys: {joined(unknown_candidate_keys)}" + ) + required_keys = {"candidate_id", "provider", "model", "cost_tier"} + missing_keys = required_keys - set(raw_candidate) + if missing_keys: + raise FallbackManifestError( + f"missing candidate keys: {joined(missing_keys)}" + ) + try: + cost_tier = CostTier(raw_candidate["cost_tier"]) + except (TypeError, ValueError) as exc: + raise FallbackManifestError( + "cost_tier must be free or paid" + ) from exc + required_credentials = _string_sequence( + raw_candidate.get("required_credentials", []), + "required_credentials", + ) + visibilities = frozenset( + _string_sequence( + raw_candidate.get( + "repository_visibilities", + sorted(ALLOWED_VISIBILITIES), + ), + "repository_visibilities", + ) + ) + capabilities = frozenset( + _string_sequence( + raw_candidate.get("capabilities", ["text"]), + "capabilities", + ) + ) + try: + return FallbackCandidate( + candidate_id=raw_candidate["candidate_id"], + provider=raw_candidate["provider"], + model=raw_candidate["model"], + cost_tier=cost_tier, + priority=raw_candidate.get("priority", 100), + required_credentials=tuple(required_credentials), + repository_visibilities=visibilities, + capabilities=capabilities, + ) + except CandidateValidationError as exc: + raise FallbackManifestError(str(exc)) from exc + + +def _string_sequence(value: Any, field_name: str) -> tuple[str, ...]: + """Return strings from a JSON array, rejecting scalar strings.""" + if not isinstance(value, list) or not all( + isinstance(item, str) for item in value + ): + raise FallbackManifestError( + f"{field_name} must be an array of strings" + ) + return tuple(value) diff --git a/contextual_orchestrator/_fallback_plan.py b/contextual_orchestrator/_fallback_plan.py new file mode 100644 index 000000000..8f6e0c0e3 --- /dev/null +++ b/contextual_orchestrator/_fallback_plan.py @@ -0,0 +1,116 @@ +"""Eligibility filtering and deterministic ordering for model fallbacks.""" + +from __future__ import annotations + +from typing import Iterable + +from ._fallback_types import ( + CandidateValidationError, + CostTier, + FallbackCandidate, + FallbackContext, + FallbackPlan, + NoEligibleCandidateError, + SkippedCandidate, +) + + +def build_fallback_plan( + candidates: Iterable[FallbackCandidate], + *, + context: FallbackContext | None = None, +) -> FallbackPlan: + """Filter candidates and place every free fallback before paid ones. + + Ordering is deterministic: cost tier, numeric priority, and then trusted + declaration order. Duplicate identities are rejected before filtering so + aliases cannot accidentally repeat a billed provider request. + """ + candidate_tuple = tuple(candidates) + _validate_candidate_collection(candidate_tuple) + runtime_context = context or FallbackContext() + eligible: list[tuple[int, FallbackCandidate]] = [] + skipped: list[SkippedCandidate] = [] + + for index, candidate in enumerate(candidate_tuple): + reason = _ineligibility_reason(candidate, runtime_context) + if reason is None: + eligible.append((index, candidate)) + else: + skipped.append(SkippedCandidate(candidate.candidate_id, reason)) + + if not eligible: + reasons = ", ".join( + f"{item.candidate_id}={item.reason}" for item in skipped + ) or "candidate list was empty" + raise NoEligibleCandidateError(f"no eligible candidates: {reasons}") + + eligible.sort( + key=lambda item: ( + 0 if item[1].cost_tier is CostTier.FREE else 1, + item[1].priority, + item[0], + ) + ) + return FallbackPlan( + candidates=tuple(candidate for _, candidate in eligible), + skipped=tuple(skipped), + ) + + +def validate_candidate_collection( + candidates: tuple[FallbackCandidate, ...] +) -> None: + """Validate a collection for manifest callers.""" + _validate_candidate_collection(candidates) + + +def _validate_candidate_collection( + candidates: tuple[FallbackCandidate, ...] +) -> None: + """Reject empty or duplicate candidate identities.""" + if not candidates: + raise NoEligibleCandidateError( + "no eligible candidates: candidate list was empty" + ) + candidate_ids: set[str] = set() + provider_models: set[tuple[str, str]] = set() + for candidate in candidates: + if not isinstance(candidate, FallbackCandidate): + raise CandidateValidationError( + "every candidate must be FallbackCandidate" + ) + if candidate.candidate_id in candidate_ids: + raise CandidateValidationError( + f"duplicate candidate_id: {candidate.candidate_id}" + ) + provider_model = (candidate.provider, candidate.model) + if provider_model in provider_models: + raise CandidateValidationError( + f"duplicate provider/model: " + f"{candidate.provider}/{candidate.model}" + ) + candidate_ids.add(candidate.candidate_id) + provider_models.add(provider_model) + + +def _ineligibility_reason( + candidate: FallbackCandidate, context: FallbackContext +) -> str | None: + """Return a public exclusion reason or ``None`` when eligible.""" + if context.repository_visibility not in candidate.repository_visibilities: + return "repository_visibility" + missing_credentials = sorted( + set(candidate.required_credentials) + - set(context.available_credentials) + ) + if missing_credentials: + return f"missing_credentials:{','.join(missing_credentials)}" + missing_capabilities = sorted( + set(context.required_capabilities) - set(candidate.capabilities) + ) + if missing_capabilities: + return f"missing_capabilities:{','.join(missing_capabilities)}" + if candidate.cost_tier is CostTier.PAID and not context.allow_paid: + return "paid_candidates_disabled" + return None diff --git a/contextual_orchestrator/_fallback_types.py b/contextual_orchestrator/_fallback_types.py new file mode 100644 index 000000000..8f1cafdf2 --- /dev/null +++ b/contextual_orchestrator/_fallback_types.py @@ -0,0 +1,211 @@ +"""Validated value objects for transport-neutral model fallback policy.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import Enum +from typing import Any, Iterable, Sequence + +SCHEMA_VERSION = 1 +ALLOWED_VISIBILITIES = frozenset({"public", "private", "internal"}) +CANDIDATE_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z") +AGENT_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}\Z") +PROVIDER_RE = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}\Z") +MODEL_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:/+-]{0,255}\Z") +CREDENTIAL_RE = re.compile(r"[A-Z][A-Z0-9_]{1,127}\Z") +CAPABILITY_RE = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}\Z") + + +class CandidateValidationError(ValueError): + """Report invalid trusted candidate or runtime-context control data.""" + + +class FallbackManifestError(ValueError): + """Report malformed or unsupported fallback-manifest input.""" + + +class NoEligibleCandidateError(RuntimeError): + """Report that every declared candidate was filtered from the plan.""" + + +class CostTier(str, Enum): + """Declare whether a model candidate incurs provider inference charges.""" + + FREE = "free" + PAID = "paid" + + +@dataclass(frozen=True, slots=True) +class FallbackCandidate: + """Describe one trusted model target without storing secret values.""" + + candidate_id: str + provider: str + model: str + cost_tier: CostTier + priority: int = 100 + required_credentials: tuple[str, ...] = () + repository_visibilities: frozenset[str] = ALLOWED_VISIBILITIES + capabilities: frozenset[str] = frozenset({"text"}) + + def __post_init__(self) -> None: + """Validate fields before a candidate reaches a workflow adapter.""" + if not CANDIDATE_ID_RE.fullmatch(self.candidate_id): + raise CandidateValidationError( + "candidate_id must be a shell-safe identifier" + ) + if not PROVIDER_RE.fullmatch(self.provider): + raise CandidateValidationError( + "provider must be lowercase and shell-safe" + ) + if not MODEL_RE.fullmatch(self.model): + raise CandidateValidationError( + "model must be a non-empty shell-safe model identifier" + ) + if not isinstance(self.cost_tier, CostTier): + raise CandidateValidationError( + "cost_tier must be CostTier.FREE or CostTier.PAID" + ) + if isinstance(self.priority, bool) or not isinstance(self.priority, int): + raise CandidateValidationError("priority must be an integer") + if self.priority < 0 or self.priority > 1_000_000: + raise CandidateValidationError( + "priority must be between 0 and 1000000" + ) + validate_credentials(self.required_credentials) + validate_visibilities(self.repository_visibilities) + validate_capabilities(self.capabilities) + + def to_public_dict(self) -> dict[str, Any]: + """Return JSON-safe metadata that never contains secret values.""" + return { + "candidate_id": self.candidate_id, + "provider": self.provider, + "model": self.model, + "cost_tier": self.cost_tier.value, + "priority": self.priority, + "required_credentials": list(self.required_credentials), + "repository_visibilities": sorted(self.repository_visibilities), + "capabilities": sorted(self.capabilities), + } + + +@dataclass(frozen=True, slots=True) +class FallbackContext: + """Describe request-time constraints used to filter candidates.""" + + repository_visibility: str = "public" + available_credentials: frozenset[str] = frozenset() + required_capabilities: frozenset[str] = frozenset() + allow_paid: bool = True + + def __post_init__(self) -> None: + """Validate context vocabulary before policy evaluation.""" + if self.repository_visibility not in ALLOWED_VISIBILITIES: + raise CandidateValidationError( + "repository visibility must be public, private, or internal" + ) + validate_credentials(tuple(self.available_credentials)) + validate_capabilities(self.required_capabilities) + if not isinstance(self.allow_paid, bool): + raise CandidateValidationError("allow_paid must be a boolean") + + +@dataclass(frozen=True, slots=True) +class SkippedCandidate: + """Record why a candidate was excluded without recording secrets.""" + + candidate_id: str + reason: str + + def to_public_dict(self) -> dict[str, str]: + """Return the JSON-safe skipped-candidate record.""" + return {"candidate_id": self.candidate_id, "reason": self.reason} + + +@dataclass(frozen=True, slots=True) +class FallbackPlan: + """Hold an eligible, deterministic free-first candidate sequence.""" + + candidates: tuple[FallbackCandidate, ...] + skipped: tuple[SkippedCandidate, ...] = () + + @property + def candidate_ids(self) -> tuple[str, ...]: + """Return candidate identifiers in execution order.""" + return tuple(candidate.candidate_id for candidate in self.candidates) + + @property + def free_candidates(self) -> tuple[FallbackCandidate, ...]: + """Return the free portion of the execution plan.""" + return tuple( + candidate + for candidate in self.candidates + if candidate.cost_tier is CostTier.FREE + ) + + @property + def paid_candidates(self) -> tuple[FallbackCandidate, ...]: + """Return paid fallbacks after every eligible free candidate.""" + return tuple( + candidate + for candidate in self.candidates + if candidate.cost_tier is CostTier.PAID + ) + + def to_public_dict(self) -> dict[str, Any]: + """Return a JSON-safe representation of the policy decision.""" + return { + "schema_version": SCHEMA_VERSION, + "candidates": [ + candidate.to_public_dict() for candidate in self.candidates + ], + "skipped": [candidate.to_public_dict() for candidate in self.skipped], + } + + +def validate_credentials(credentials: Sequence[str]) -> None: + """Validate credential names without reading credential values.""" + if isinstance(credentials, (str, bytes)): + raise CandidateValidationError( + "credential names must be a sequence" + ) + for credential in credentials: + if not isinstance(credential, str) or not CREDENTIAL_RE.fullmatch( + credential + ): + raise CandidateValidationError( + "credential names must be uppercase environment identifiers" + ) + + +def validate_visibilities(visibilities: frozenset[str]) -> None: + """Validate non-empty repository-visibility eligibility.""" + if not isinstance(visibilities, frozenset) or not visibilities: + raise CandidateValidationError( + "repository visibility set must be non-empty" + ) + unknown = set(visibilities) - ALLOWED_VISIBILITIES + if unknown: + raise CandidateValidationError( + f"unknown repository visibility: {joined(unknown)}" + ) + + +def validate_capabilities(capabilities: frozenset[str]) -> None: + """Validate capability labels used by the eligibility filter.""" + if not isinstance(capabilities, frozenset): + raise CandidateValidationError("capabilities must be a frozenset") + for capability in capabilities: + if not isinstance(capability, str) or not CAPABILITY_RE.fullmatch( + capability + ): + raise CandidateValidationError( + "capability names must be lowercase shell-safe identifiers" + ) + + +def joined(values: Iterable[object]) -> str: + """Return stable comma-separated diagnostics for unordered values.""" + return ",".join(sorted(str(value) for value in values)) diff --git a/contextual_orchestrator/model_fallback.py b/contextual_orchestrator/model_fallback.py new file mode 100644 index 000000000..0acea894a --- /dev/null +++ b/contextual_orchestrator/model_fallback.py @@ -0,0 +1,41 @@ +"""Public API for deterministic, transport-neutral model fallbacks. + +The policy validates trusted candidate metadata and returns a stable plan with +all eligible free candidates before every paid fallback. It never performs a +network request and never stores credential values, so existing workflow +transports and reviewer identities remain under the caller's control. +""" + +from __future__ import annotations + +from ._fallback_cli import main +from ._fallback_manifest import load_fallback_manifest +from ._fallback_plan import build_fallback_plan +from ._fallback_types import ( + CandidateValidationError, + CostTier, + FallbackCandidate, + FallbackContext, + FallbackManifestError, + FallbackPlan, + NoEligibleCandidateError, + SkippedCandidate, +) + +__all__ = [ + "CandidateValidationError", + "CostTier", + "FallbackCandidate", + "FallbackContext", + "FallbackManifestError", + "FallbackPlan", + "NoEligibleCandidateError", + "SkippedCandidate", + "build_fallback_plan", + "load_fallback_manifest", + "main", +] + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/docs/doctoring/free-first-model-fallback.md b/docs/doctoring/free-first-model-fallback.md new file mode 100644 index 000000000..52deb1b78 --- /dev/null +++ b/docs/doctoring/free-first-model-fallback.md @@ -0,0 +1,88 @@ +# Doctoring record: free-first multi-model fallback + +## Decision + +The reusable boundary is a pure candidate-planning module rather than a new +HTTP gateway. Noema, OpenCode Agent, and Strix have different security, +review-identity, evidence, timeout, and output-validation contracts. Replacing +those transports would combine privileges and would make one provider client a +single point of failure. The orchestrator therefore supplies a shared policy +contract while each consumer retains its proven transport and credentials. + +The policy is deterministic and fail-closed: + +- cost tier is explicit trusted metadata, never inferred from a mutable model + catalogue or a model-name suffix; +- all eligible free candidates precede all eligible paid candidates; +- free candidates fall back to other free candidates before paid escalation; +- visibility, capability, and configured-credential-name filters run before a + provider call; +- duplicate identities, unknown manifest fields, unsafe identifiers, and an + empty eligible pool are errors; +- candidate selection never accepts provider output. The consuming workflow's + existing schema, security, and evidence gates remain authoritative; +- secret values are neither retained nor serialized by the policy module. + +This is a deterministic cost-ordering baseline, not a learned quality router. +A future learned router must be benchmarked on the exact review/security task +and must preserve the free-before-paid budget boundary when that policy is +selected. + +## Evidence and standards mapping + +LLM cascade research supports attempting lower-cost models before escalating, +while also showing that quality estimators and task-specific evaluation matter. +The implementation therefore separates a stable cost policy from acceptance +or quality estimation. HTTP retry semantics remain in each transport adapter; +rate limiting and service unavailability must not be converted into approval. + +Provider documentation also makes “free” a runtime commercial property rather +than a permanent model property. GitHub Models includes rate-limited usage and +can optionally enable paid use. OpenRouter free variants have lower limits and +changing availability. NVIDIA describes API Catalog access as a prototyping or +trial path. These facts justify explicit operator-owned `cost_tier` metadata +and immutable policy revisions instead of name-based classification. + +## Verification contract + +- 100% statement and branch coverage across the five fallback policy modules + (270 statements and 94 branches). +- 100% docstrings for public fallback policy symbols. +- Property under test: no eligible paid candidate appears before an eligible + free candidate regardless of numeric priority. +- Stable tie ordering, credential/visibility/capability filtering, duplicate + rejection, strict manifest validation, empty-pool failure, CLI secret + non-disclosure, and free-only operation are covered by 32 regression tests. +- The module imports on Python 3.10+ and uses only the standard library. + +## APA 7 references + +Chen, L., Zaharia, M., & Zou, J. (2023). *FrugalGPT: How to use large language +models while reducing cost and improving performance* [Preprint]. arXiv. +https://doi.org/10.48550/arXiv.2305.05176 + +Dekoninck, J., Baader, M., & Vechev, M. (2025). *A unified approach to routing +and cascading for LLMs* [Preprint]. arXiv. +https://doi.org/10.48550/arXiv.2410.10347 + +GitHub. (n.d.). *GitHub Models billing*. Retrieved August 5, 2026, from +https://docs.github.com/en/billing/concepts/product-billing/github-models + +NVIDIA. (2026). *NVIDIA NIM for vision language models: Overview*. +https://docs.nvidia.com/nim/vision-language-models/2.0.0/introduction.html + +Nottingham, M., & Fielding, R. (2012). *Additional HTTP status codes* +(RFC 6585). RFC Editor. https://doi.org/10.17487/RFC6585 + +Ong, I., Almahairi, A., Wu, V., Chiang, W.-L., Wu, T., Gonzalez, J. E., +Kadous, M. W., & Stoica, I. (2024). *RouteLLM: Learning to route LLMs with +preference data* [Preprint]. arXiv. https://doi.org/10.48550/arXiv.2406.18665 + +OpenRouter. (n.d.-a). *Free variant*. Retrieved August 5, 2026, from +https://openrouter.ai/docs/guides/routing/model-variants/free + +OpenRouter. (n.d.-b). *Model fallbacks*. Retrieved August 5, 2026, from +https://openrouter.ai/docs/guides/routing/model-fallbacks + +Rescorla, E., Nottingham, M., & Bishop, M. (2022). *HTTP semantics* (RFC 9110). +RFC Editor. https://doi.org/10.17487/RFC9110 diff --git a/docs/model-fallback-policy.md b/docs/model-fallback-policy.md new file mode 100644 index 000000000..011acf21a --- /dev/null +++ b/docs/model-fallback-policy.md @@ -0,0 +1,107 @@ +# Transport-neutral model fallback policy + +`contextual_orchestrator.model_fallback` is a policy-only module for workflows +that already own their HTTP client, provider SDK, credential identity, and +review identity. It never opens a socket and never reads a credential value. +It validates a versioned JSON manifest, filters candidates by explicit runtime +constraints, and returns a deterministic execution order: + +1. every eligible `free` candidate, ordered by numeric priority; +2. every eligible `paid` candidate, ordered by numeric priority; +3. declaration order as the final stable tie-breaker. + +A paid candidate can therefore never jump ahead of an eligible free candidate. +Model names are not used to infer price. The operator must declare `cost_tier` +because provider pricing and trial availability change independently of code. + +## Manifest + +```json +{ + "schema_version": 1, + "agents": { + "noema": { + "candidates": [ + { + "candidate_id": "nvidia_nim_free_primary", + "provider": "nvidia-nim", + "model": "nvidia/nemotron-3-ultra-550b-a55b", + "cost_tier": "free", + "priority": 10, + "required_credentials": ["NVIDIA_NIM_API_KEY"], + "repository_visibilities": ["public"], + "capabilities": ["text", "structured_output"] + }, + { + "candidate_id": "paid_fallback", + "provider": "openai", + "model": "gpt-5.6-luna", + "cost_tier": "paid", + "priority": 100, + "required_credentials": ["OPENAI_API_KEY"], + "repository_visibilities": ["public", "private", "internal"], + "capabilities": ["text", "structured_output"] + } + ] + } + } +} +``` + +The parser rejects unknown keys, unsupported schema versions, duplicate +candidate IDs, duplicate provider/model targets, unsafe shell identifiers, +and empty candidate lists. Repository visibility, capability labels, and +credential **names** are policy data; secret values are never serialized. + +## Python API + +```python +from contextual_orchestrator import ( + FallbackContext, + build_fallback_plan, + load_fallback_manifest, +) + +candidates = load_fallback_manifest(document, "noema") +plan = build_fallback_plan( + candidates, + context=FallbackContext( + repository_visibility="public", + available_credentials=frozenset({"NVIDIA_NIM_API_KEY", "OPENAI_API_KEY"}), + required_capabilities=frozenset({"structured_output"}), + ), +) +for candidate in plan.candidates: + call_existing_transport(candidate) +``` + +The caller decides which provider errors, malformed outputs, timeouts, or +quality-gate failures advance to the next candidate. A workflow must accept an +answer only after its existing schema and security checks pass. + +## CLI integration + +The CLI checks only whether named environment variables are non-empty. It does +not print their values. + +```bash +python -m contextual_orchestrator.model_fallback plan \ + --manifest config/llm-fallback-policy.json \ + --agent opencode-review \ + --repository-visibility public \ + --credential-env NVIDIA_NIM_API_KEY \ + --credential-env OPENAI_API_KEY \ + --required-capability structured_output \ + --format models +``` + +Use `--deny-paid` for an explicit free-only run. An empty eligible pool is a +hard error, not an implicit success. + +## Integration boundary + +Cross-repository consumers should materialize this repository at an immutable +commit SHA, verify the checkout, add only that checkout to `PYTHONPATH`, and +keep their existing provider and reviewer credentials scoped to the original +workflow job. This module is suitable for the central `.github` workflows, +`naruon`, and standalone services because it has no provider SDK dependency. diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 000000000..2977edd15 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Test package for contextual-orchestrator.""" diff --git a/tests/fallback_test_support.py b/tests/fallback_test_support.py new file mode 100644 index 000000000..44d22bea8 --- /dev/null +++ b/tests/fallback_test_support.py @@ -0,0 +1,66 @@ +"""Shared test fixtures for model fallback policy tests.""" + +from __future__ import annotations + +from contextual_orchestrator.model_fallback import ( + CostTier, + FallbackCandidate, +) + + +def candidate( + candidate_id: str, + model: str, + *, + cost_tier: CostTier = CostTier.FREE, + priority: int = 100, + credentials: tuple[str, ...] = (), + visibilities: frozenset[str] = frozenset( + {"public", "private", "internal"} + ), + capabilities: frozenset[str] = frozenset({"text"}), +) -> FallbackCandidate: + """Build a concise candidate for tests.""" + return FallbackCandidate( + candidate_id=candidate_id, + provider="provider", + model=model, + cost_tier=cost_tier, + priority=priority, + required_credentials=credentials, + repository_visibilities=visibilities, + capabilities=capabilities, + ) + + +def manifest_document() -> dict[str, object]: + """Return a complete manifest for parsing and CLI tests.""" + return { + "schema_version": 1, + "agents": { + "noema": { + "candidates": [ + { + "candidate_id": "paid-primary", + "provider": "openai", + "model": "openai/paid", + "cost_tier": "paid", + "priority": 0, + "required_credentials": ["PAID_API_KEY"], + "repository_visibilities": ["public", "private"], + "capabilities": ["text", "structured_output"], + }, + { + "candidate_id": "free-primary", + "provider": "nvidia-nim", + "model": "nvidia/free", + "cost_tier": "free", + "priority": 10, + "required_credentials": ["FREE_API_KEY"], + "repository_visibilities": ["public"], + "capabilities": ["text", "structured_output"], + }, + ] + } + }, + } diff --git a/tests/test_model_fallback_cli.py b/tests/test_model_fallback_cli.py new file mode 100644 index 000000000..bc79e858c --- /dev/null +++ b/tests/test_model_fallback_cli.py @@ -0,0 +1,164 @@ +"""Tests for the fallback policy command-line adapter.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from contextual_orchestrator.model_fallback import ( + FallbackManifestError, + main, +) +from tests.fallback_test_support import manifest_document + + +def write_manifest(path: Path) -> None: + """Write the shared valid manifest fixture as UTF-8 JSON.""" + path.write_text(json.dumps(manifest_document()), encoding="utf-8") + + +def test_cli_emits_free_first_json_without_secret_values( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """CLI checks named secrets but never prints their values.""" + manifest_path = tmp_path / "policy.json" + write_manifest(manifest_path) + monkeypatch.setenv("FREE_API_KEY", "free-secret-value") + monkeypatch.setenv("PAID_API_KEY", "paid-secret-value") + + assert main( + [ + "plan", + "--manifest", + str(manifest_path), + "--agent", + "noema", + "--repository-visibility", + "public", + "--credential-env", + "FREE_API_KEY", + "--credential-env", + "PAID_API_KEY", + "--required-capability", + "structured_output", + "--format", + "json", + ] + ) == 0 + + output = capsys.readouterr().out + payload = json.loads(output) + assert [item["candidate_id"] for item in payload["candidates"]] == [ + "free-primary", + "paid-primary", + ] + assert "free-secret-value" not in output + assert "paid-secret-value" not in output + + +def test_cli_emits_models_and_respects_deny_paid( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Shell consumers receive validated model identifiers.""" + manifest_path = tmp_path / "policy.json" + write_manifest(manifest_path) + monkeypatch.setenv("FREE_API_KEY", "configured") + monkeypatch.setenv("PAID_API_KEY", "configured") + + assert main( + [ + "plan", + "--manifest", + str(manifest_path), + "--agent", + "noema", + "--credential-env", + "FREE_API_KEY", + "--credential-env", + "PAID_API_KEY", + "--deny-paid", + "--format", + "models", + ] + ) == 0 + assert capsys.readouterr().out == "nvidia/free\n" + + +def test_cli_treats_empty_credentials_as_unavailable( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """An empty secret is not a configured credential.""" + manifest_path = tmp_path / "policy.json" + write_manifest(manifest_path) + monkeypatch.setenv("FREE_API_KEY", " ") + monkeypatch.setenv("PAID_API_KEY", "configured") + + assert main( + [ + "plan", + "--manifest", + str(manifest_path), + "--agent", + "noema", + "--credential-env", + "FREE_API_KEY", + "--credential-env", + "PAID_API_KEY", + "--format", + "ids", + ] + ) == 0 + assert capsys.readouterr().out == "paid-primary\n" + + +def test_cli_rejects_invalid_json_missing_file_and_non_object_root( + tmp_path: Path, +) -> None: + """File input failures are explicit instead of using defaults.""" + invalid_path = tmp_path / "invalid.json" + invalid_path.write_text("{", encoding="utf-8") + with pytest.raises(FallbackManifestError, match="valid JSON"): + main(["plan", "--manifest", str(invalid_path), "--agent", "noema"]) + with pytest.raises(FallbackManifestError, match="could not be read"): + main( + [ + "plan", + "--manifest", + str(tmp_path / "missing.json"), + "--agent", + "noema", + ] + ) + + array_path = tmp_path / "array.json" + array_path.write_text("[]", encoding="utf-8") + with pytest.raises(FallbackManifestError, match="manifest must be an object"): + main(["plan", "--manifest", str(array_path), "--agent", "noema"]) + + +def test_cli_rejects_unsafe_credential_name( + tmp_path: Path, +) -> None: + """Environment selectors use strict identifier syntax.""" + manifest_path = tmp_path / "policy.json" + write_manifest(manifest_path) + with pytest.raises(Exception, match="credential"): + main( + [ + "plan", + "--manifest", + str(manifest_path), + "--agent", + "noema", + "--credential-env", + "bad-key", + ] + ) diff --git a/tests/test_model_fallback_manifest.py b/tests/test_model_fallback_manifest.py new file mode 100644 index 000000000..aaf14a3c4 --- /dev/null +++ b/tests/test_model_fallback_manifest.py @@ -0,0 +1,120 @@ +"""Tests for strict fallback policy manifest parsing.""" + +from __future__ import annotations + +import pytest + +from contextual_orchestrator.model_fallback import ( + FallbackManifestError, + load_fallback_manifest, +) +from tests.fallback_test_support import manifest_document + + +def test_manifest_parses_candidates_without_reordering_source() -> None: + """Manifest parsing preserves trusted declaration order.""" + candidates = load_fallback_manifest(manifest_document(), "noema") + assert tuple(candidate.candidate_id for candidate in candidates) == ( + "paid-primary", + "free-primary", + ) + + +@pytest.mark.parametrize( + ("mutator", "message"), + [ + (lambda document: document.update({"unknown": True}), "unknown manifest"), + (lambda document: document.update({"schema_version": 2}), "schema_version"), + (lambda document: document.update({"agents": []}), "agents must be"), + ( + lambda document: document.update({"agents": {"bad agent": {}}}), + "agent name", + ), + ], +) +def test_manifest_rejects_invalid_root_control_data(mutator, message: str) -> None: + """Versioned root keys and agent identifiers fail closed.""" + document = manifest_document() + mutator(document) + with pytest.raises(FallbackManifestError, match=message): + load_fallback_manifest(document, "noema") + + +def test_manifest_rejects_non_object_root_and_missing_agent() -> None: + """Programmatic inputs cannot bypass root and agent shape checks.""" + with pytest.raises(FallbackManifestError, match="manifest must be an object"): + load_fallback_manifest([], "noema") # type: ignore[arg-type] + with pytest.raises(FallbackManifestError, match="was not found"): + load_fallback_manifest(manifest_document(), "strix") + + +def test_manifest_rejects_invalid_agent_container_and_keys() -> None: + """Agent blocks accept only a non-empty candidate array.""" + document = manifest_document() + document["agents"]["noema"] = [] + with pytest.raises(FallbackManifestError, match="must be an object"): + load_fallback_manifest(document, "noema") + + document = manifest_document() + document["agents"]["noema"]["unknown"] = True + with pytest.raises(FallbackManifestError, match="unknown agent keys"): + load_fallback_manifest(document, "noema") + + document = manifest_document() + document["agents"]["noema"]["candidates"] = {} + with pytest.raises(FallbackManifestError, match="must be an array"): + load_fallback_manifest(document, "noema") + + document = manifest_document() + document["agents"]["noema"]["candidates"] = [] + with pytest.raises(FallbackManifestError, match="at least one"): + load_fallback_manifest(document, "noema") + + +def test_manifest_rejects_non_object_candidate_and_unknown_keys() -> None: + """Candidate entries use an exact schema.""" + document = manifest_document() + document["agents"]["noema"]["candidates"][0] = [] + with pytest.raises(FallbackManifestError, match="candidate must be an object"): + load_fallback_manifest(document, "noema") + + document = manifest_document() + document["agents"]["noema"]["candidates"][0]["unknown"] = True + with pytest.raises(FallbackManifestError, match="unknown candidate keys"): + load_fallback_manifest(document, "noema") + + +def test_manifest_rejects_missing_keys_bad_tier_and_bad_sequences() -> None: + """Candidate schema failures are normalized as manifest errors.""" + document = manifest_document() + del document["agents"]["noema"]["candidates"][0]["model"] + with pytest.raises(FallbackManifestError, match="missing candidate keys: model"): + load_fallback_manifest(document, "noema") + + document = manifest_document() + document["agents"]["noema"]["candidates"][0]["cost_tier"] = "metered" + with pytest.raises(FallbackManifestError, match="free or paid"): + load_fallback_manifest(document, "noema") + + for field in ( + "required_credentials", + "repository_visibilities", + "capabilities", + ): + document = manifest_document() + document["agents"]["noema"]["candidates"][0][field] = "not-array" + with pytest.raises(FallbackManifestError, match=f"{field} must be"): + load_fallback_manifest(document, "noema") + + +def test_manifest_normalizes_candidate_validation_and_duplicates() -> None: + """Unsafe fields and duplicate identities remain manifest errors.""" + document = manifest_document() + document["agents"]["noema"]["candidates"][0]["provider"] = "Bad/Provider" + with pytest.raises(FallbackManifestError, match="provider"): + load_fallback_manifest(document, "noema") + + document = manifest_document() + document["agents"]["noema"]["candidates"][1]["candidate_id"] = "paid-primary" + with pytest.raises(FallbackManifestError, match="duplicate candidate_id"): + load_fallback_manifest(document, "noema") diff --git a/tests/test_model_fallback_plan.py b/tests/test_model_fallback_plan.py new file mode 100644 index 000000000..d5e5b9000 --- /dev/null +++ b/tests/test_model_fallback_plan.py @@ -0,0 +1,218 @@ +"""Tests for fallback value validation and plan ordering.""" + +from __future__ import annotations + +import pytest + +from contextual_orchestrator.model_fallback import ( + CandidateValidationError, + CostTier, + FallbackCandidate, + FallbackContext, + NoEligibleCandidateError, + SkippedCandidate, + build_fallback_plan, +) +from tests.fallback_test_support import candidate + + +def test_plan_places_all_free_candidates_before_paid_candidates() -> None: + """Paid priority cannot jump ahead of an eligible free candidate.""" + plan = build_fallback_plan( + [ + candidate( + "paid-fast", "paid/fast", cost_tier=CostTier.PAID, priority=0 + ), + candidate("free-second", "free/second", priority=20), + candidate("free-first", "free/first", priority=10), + candidate( + "paid-second", + "paid/second", + cost_tier=CostTier.PAID, + priority=5, + ), + ] + ) + + assert plan.candidate_ids == ( + "free-first", + "free-second", + "paid-fast", + "paid-second", + ) + assert tuple(item.model for item in plan.free_candidates) == ( + "free/first", + "free/second", + ) + assert tuple(item.model for item in plan.paid_candidates) == ( + "paid/fast", + "paid/second", + ) + assert [item["candidate_id"] for item in plan.to_public_dict()["candidates"]] == [ + "free-first", + "free-second", + "paid-fast", + "paid-second", + ] + + +def test_plan_is_stable_for_equal_cost_and_priority() -> None: + """Declaration order is the deterministic final tie-breaker.""" + plan = build_fallback_plan( + [candidate("free-a", "free/a"), candidate("free-b", "free/b")] + ) + assert plan.candidate_ids == ("free-a", "free-b") + + +def test_plan_filters_by_credentials_visibility_and_capabilities() -> None: + """Eligibility is evaluated without exposing credential values.""" + plan = build_fallback_plan( + [ + candidate( + "eligible", + "free/eligible", + credentials=("FREE_API_KEY",), + visibilities=frozenset({"public"}), + capabilities=frozenset({"text", "structured_output"}), + ), + candidate( + "private-only", + "free/private", + visibilities=frozenset({"private"}), + ), + candidate( + "missing-key", + "free/missing", + credentials=("OTHER_API_KEY",), + ), + candidate( + "missing-capability", + "free/no-json", + capabilities=frozenset({"text"}), + ), + ], + context=FallbackContext( + repository_visibility="public", + available_credentials=frozenset({"FREE_API_KEY"}), + required_capabilities=frozenset({"structured_output"}), + ), + ) + assert plan.candidate_ids == ("eligible",) + assert tuple((item.candidate_id, item.reason) for item in plan.skipped) == ( + ("private-only", "repository_visibility"), + ("missing-key", "missing_credentials:OTHER_API_KEY"), + ("missing-capability", "missing_capabilities:structured_output"), + ) + + +def test_plan_can_disable_paid_fallbacks() -> None: + """A caller can prohibit paid candidates while retaining free fallback.""" + plan = build_fallback_plan( + [ + candidate("paid", "paid/model", cost_tier=CostTier.PAID), + candidate("free", "free/model"), + ], + context=FallbackContext(allow_paid=False), + ) + assert plan.candidate_ids == ("free",) + assert plan.skipped[0].reason == "paid_candidates_disabled" + + +def test_plan_rejects_duplicates_and_empty_or_untyped_inputs() -> None: + """A pool cannot repeat a logical target or silently accept no target.""" + with pytest.raises(CandidateValidationError, match="duplicate candidate_id"): + build_fallback_plan( + [candidate("same", "model/a"), candidate("same", "model/b")] + ) + with pytest.raises(CandidateValidationError, match="duplicate provider/model"): + build_fallback_plan( + [candidate("first", "model/a"), candidate("second", "model/a")] + ) + with pytest.raises(NoEligibleCandidateError, match="candidate list was empty"): + build_fallback_plan([]) + with pytest.raises(CandidateValidationError, match="FallbackCandidate"): + build_fallback_plan([object()]) # type: ignore[list-item] + + +def test_plan_raises_when_every_candidate_is_ineligible() -> None: + """The planner never turns an empty eligible pool into success.""" + with pytest.raises(NoEligibleCandidateError, match="MISSING_KEY"): + build_fallback_plan( + [candidate("needs-key", "free/model", credentials=("MISSING_KEY",))] + ) + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("candidate_id", "bad id", "candidate_id"), + ("provider", "Provider/Bad", "provider"), + ("model", "bad model", "model"), + ("cost_tier", "free", "cost_tier"), + ("priority", -1, "priority"), + ("priority", True, "priority"), + ("required_credentials", ("bad-key",), "credential"), + ("repository_visibilities", frozenset({"secret"}), "visibility"), + ("capabilities", frozenset({"Structured Output"}), "capability"), + ], +) +def test_candidate_validation_rejects_unsafe_values( + field: str, value: object, message: str +) -> None: + """Workflow control fields are strict and shell-safe.""" + values: dict[str, object] = { + "candidate_id": "candidate-one", + "provider": "provider", + "model": "model/one", + "cost_tier": CostTier.FREE, + "priority": 1, + "required_credentials": (), + "repository_visibilities": frozenset({"public"}), + "capabilities": frozenset({"text"}), + } + values[field] = value + with pytest.raises(CandidateValidationError, match=message): + FallbackCandidate(**values) # type: ignore[arg-type] + + +def test_context_and_collection_types_fail_closed() -> None: + """Truthy strings and mutable control collections cannot bypass policy.""" + with pytest.raises(CandidateValidationError, match="visibility"): + FallbackContext(repository_visibility="secret") + with pytest.raises(CandidateValidationError, match="credential"): + FallbackContext(available_credentials=frozenset({"bad-key"})) + with pytest.raises(CandidateValidationError, match="capability"): + FallbackContext(required_capabilities=frozenset({"bad capability"})) + with pytest.raises(CandidateValidationError, match="allow_paid"): + FallbackContext(allow_paid="false") # type: ignore[arg-type] + with pytest.raises(CandidateValidationError, match="sequence"): + candidate( + "candidate", "model/one", credentials="API_KEY" # type: ignore[arg-type] + ) + with pytest.raises(CandidateValidationError, match="non-empty"): + candidate( + "candidate", "model/one", visibilities=frozenset() + ) + with pytest.raises(CandidateValidationError, match="non-empty"): + candidate( + "candidate", "model/one", visibilities={"public"} # type: ignore[arg-type] + ) + with pytest.raises(CandidateValidationError, match="frozenset"): + candidate( + "candidate", "model/one", capabilities={"text"} # type: ignore[arg-type] + ) + + +def test_public_records_never_require_secret_values() -> None: + """Candidate and skip records expose only names and public reasons.""" + item = candidate( + "free", "free/model", credentials=("FREE_API_KEY",) + ).to_public_dict() + assert item["required_credentials"] == ["FREE_API_KEY"] + skipped = SkippedCandidate( + "candidate", "missing_credentials:FREE_API_KEY" + ) + assert skipped.to_public_dict() == { + "candidate_id": "candidate", + "reason": "missing_credentials:FREE_API_KEY", + }