feat: add free-first model fallback policy

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.

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",
 ]

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

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)