diff --git a/.coveragerc b/.coveragerc
new file mode 100644
index 0000000..2a0553b
--- /dev/null
+++ b/.coveragerc
@@ -0,0 +1,18 @@
+[run]
+branch = true
+source =
+ maf_starter
+ entities
+ main.py
+omit =
+ autogen_dashboard/*
+ autogen_starter/*
+ */__pycache__/*
+ tests/*
+
+[report]
+exclude_lines =
+ pragma: no cover
+ if __name__ == .__main__.:
+ raise NotImplementedError
+ if TYPE_CHECKING:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 8ddaee8..3faaa2e 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -1,56 +1,86 @@
-name: CI
-
-on:
- push:
- branches: [ main ]
- pull_request:
-
-permissions:
- contents: read
-
-concurrency:
- group: ci-${{ github.workflow }}-${{ github.ref }}
- cancel-in-progress: true
-
-jobs:
- test:
- name: Python ${{ matrix.python-version }} / ${{ matrix.os }}
- runs-on: ${{ matrix.os }}
- timeout-minutes: 20
- strategy:
- fail-fast: false
- matrix:
- os: [ubuntu-latest, windows-latest]
- python-version: ['3.12']
-
- steps:
- - uses: actions/checkout@v7
-
- - name: Set up Python
- uses: actions/setup-python@v6
- with:
- python-version: ${{ matrix.python-version }}
- cache: pip
-
- - name: Install dependencies
- run: python -m pip install -r requirements.txt
-
- - name: Verify dependency consistency
- run: python -m pip check
-
- - name: Contract compatibility (pinned cas-contracts v1.1)
- # Consumer-side gate: fails red if autogen's pinned CAS contract version
- # or the vendored v1.1 schema release drifts. See
- # tests/test_contract_compatibility.py for the validated contract surface.
- run: python -m pytest tests/test_contract_compatibility.py -q --tb=short
-
- - name: Run full test suite
- run: |
- python -m pip install pytest-cov
- python -m pytest -q --tb=short --cov=. --cov-report=xml
-
- - name: Compile Python sources
- run: python -m compileall autogen_starter autogen_dashboard maf_starter main.py -q
-
- - name: Validate dashboard JavaScript
- run: node --check autogen_dashboard/static/app.js
+name: CI
+
+on:
+ push:
+ branches: [ main ]
+ pull_request:
+
+permissions:
+ contents: read
+
+concurrency:
+ group: ci-${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: true
+
+jobs:
+ test:
+ name: Python ${{ matrix.python-version }} / ${{ matrix.os }}
+ runs-on: ${{ matrix.os }}
+ timeout-minutes: 20
+ strategy:
+ fail-fast: false
+ matrix:
+ os: [ubuntu-latest, windows-latest]
+ python-version: ['3.12']
+
+ steps:
+ - uses: actions/checkout@v7
+
+ - name: Set up Python
+ uses: actions/setup-python@v6
+ with:
+ python-version: ${{ matrix.python-version }}
+ cache: pip
+
+ - name: Install dependencies
+ run: python -m pip install -r requirements.txt
+
+ - name: Verify dependency consistency
+ run: python -m pip check
+
+ - name: Contract compatibility (pinned cas-contracts v1.1)
+ # Consumer-side gate: fails red if autogen's pinned CAS contract version
+ # or the vendored v1.1 schema release drifts. See
+ # tests/test_contract_compatibility.py for the validated contract surface.
+ run: python -m pytest tests/test_contract_compatibility.py -q --tb=short
+
+ - name: Run full test suite
+ run: |
+ python -m pytest -q --tb=short --cov --cov-branch --cov-report=xml --cov-fail-under=73
+
+ - name: Emit branch coverage telemetry
+ if: always()
+ shell: python
+ run: |
+ import json
+ import sys
+ import xml.etree.ElementTree as ET
+ from pathlib import Path
+
+ required = 53.5
+ coverage_file = Path("coverage.xml")
+ if not coverage_file.exists():
+ print(json.dumps({
+ "event": "ci_failure",
+ "metric": "branch-rate",
+ "coverage_percent": None,
+ "required": required,
+ "reason": "coverage.xml missing",
+ }))
+ sys.exit(1)
+
+ branch_rate = float(ET.parse(coverage_file).getroot().attrib["branch-rate"]) * 100.0
+ passed = branch_rate >= required
+ print(json.dumps({
+ "event": "ci_pass" if passed else "ci_failure",
+ "metric": "branch-rate",
+ "coverage_percent": round(branch_rate, 2),
+ "required": required,
+ }))
+ sys.exit(0 if passed else 1)
+
+ - name: Compile Python sources
+ run: python -m compileall autogen_starter autogen_dashboard maf_starter main.py -q
+
+ - name: Validate dashboard JavaScript
+ run: node --check autogen_dashboard/static/app.js
diff --git a/.gitignore b/.gitignore
index c3fe68d..01d5f25 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,3 +10,6 @@ state/
.tmp-tests/
*.out.log
*.err.log
+.coverage
+coverage.xml
+htmlcov/
diff --git a/requirements.txt b/requirements.txt
index 8495a0c..60e8e2e 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,13 +1,14 @@
-agent-framework==1.10.0
-agent-framework-devui==1.0.0b260630
-agent-framework-orchestrations==1.0.0
-autogen-agentchat>=0.7.5,<0.8.0
-autogen-core>=0.7.5,<0.8.0
-autogen-ext[anthropic,ollama,openai]>=0.7.5,<0.8.0
-fastapi>=0.139.0,<1.0
-jsonschema>=4.26.0,<5.0
-openapi-spec-validator>=0.9.0,<1.0
-pydantic>=2.13.4,<3.0
-pytest>=9.1.1,<10.0
-python-dotenv>=1.2.2,<2.0
-uvicorn>=0.49.0,<1.0
+agent-framework==1.0.0rc5
+agent-framework-devui==1.0.0b260319
+agent-framework-orchestrations==1.0.0b260319
+autogen-agentchat>=0.7.5,<0.8.0
+autogen-core>=0.7.5,<0.8.0
+autogen-ext[anthropic,ollama,openai]>=0.7.5,<0.8.0
+fastapi>=0.133.0,<0.138.1
+jsonschema>=4.26.0,<5.0
+openapi-spec-validator>=0.9.0,<1.0
+pydantic>=2.13.4,<3.0
+pytest>=9.1.1,<10.0
+pytest-cov>=5.0,<8.0
+python-dotenv>=1.2.2,<2.0
+uvicorn>=0.41.0,<0.42.0
diff --git a/tests/test_maf_setup.py b/tests/test_maf_setup.py
index a343537..0a31266 100644
--- a/tests/test_maf_setup.py
+++ b/tests/test_maf_setup.py
@@ -1,443 +1,622 @@
-from __future__ import annotations
-
-import unittest
-import uuid
-import shutil
-from pathlib import Path
-from unittest.mock import AsyncMock, patch
-
-from agent_framework import ChatContext, ChatResponse, ChatResponseUpdate, Message
-from agent_framework._types import Content
-from fastapi import FastAPI
-from fastapi.responses import HTMLResponse
-from fastapi.testclient import TestClient
-
-from maf_starter.agent_factory import build_agent
-from maf_starter.config import current_checkpoint_dir, load_settings, mask_secret
-from maf_starter.devui_overrides import OLD_GD_RENDERER
-from maf_starter.devui_overrides import _patch_devui_bundle
-from maf_starter.devui_overrides import install_devui_ui_overrides
-from maf_starter.provider_fallback import ResponseStream, _merge_route_metadata, _resolve_run_scope, _wrap_stream_with_fallback
-from maf_starter.routing_policy import RoutingPlan, build_routing_plan
-from maf_starter.routing_types import ChainStep, parse_chain_steps
-from maf_starter.team_factory import build_repo_team
-from maf_starter.tools import build_repo_tools, resolve_repo_path
-from maf_starter.workflow_factory import build_workflow
-
-
-SCRATCH_ROOT = Path(__file__).resolve().parents[1] / ".tmp-tests"
-
-
-def load_test_settings():
- with patch.dict("os.environ", {"MAF_API_KEY": "test-key"}, clear=False):
- return load_settings(project_root=Path.cwd(), env_path=Path.cwd() / ".missing-env")
-
-
-class RepoScratchTestCase(unittest.TestCase):
- def make_scratch_dir(self) -> Path:
- path = SCRATCH_ROOT / uuid.uuid4().hex
- path.mkdir(parents=True, exist_ok=False)
- self.addCleanup(lambda: shutil.rmtree(path, ignore_errors=True))
- return path
-
-
-class MafSetupTests(RepoScratchTestCase):
- def test_parse_chain_steps(self) -> None:
- steps = parse_chain_steps(
- (
- "gemini:gemini-2.5-pro",
- "gemini-cli:gemini-2.5-pro",
- "claude-cli",
- )
- )
- self.assertEqual(
- [(step.provider, step.model) for step in steps],
- [
- ("gemini", "gemini-2.5-pro"),
- ("gemini-cli", "gemini-2.5-pro"),
- ("claude-cli", None),
- ],
- )
-
- def test_load_settings_prefers_maf_over_gemini_fallbacks(self) -> None:
- root = self.make_scratch_dir()
- (root / "entities").mkdir()
- (root / "repo").mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "MAF_API_KEY": "maf-secret",
- "GEMINI_API_KEY": "gem-secret",
- "MAF_MODEL": "gemini-2.5-flash",
- "GEMINI_MODEL": "gemini-2.5-pro",
- "MAF_REPO_ROOT": str(root / "repo"),
- "MAF_ENTITIES_DIR": str(root / "entities"),
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
-
- self.assertEqual(settings.api_key, "maf-secret")
- self.assertEqual(settings.model, "gemini-2.5-flash")
- self.assertEqual(settings.repo_root, (root / "repo").resolve())
- self.assertEqual(settings.entities_dir, (root / "entities").resolve())
-
- def test_default_fallback_chain_includes_anthropic_when_key_exists(self) -> None:
- root = self.make_scratch_dir()
- (root / "entities").mkdir()
- (root / "repo").mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "MAF_API_KEY": "maf-secret",
- "MAF_REPO_ROOT": str(root / "repo"),
- "MAF_ENTITIES_DIR": str(root / "entities"),
- "MAF_FALLBACK_CHAIN": "",
- "ANTHROPIC_API_KEY": "anthropic-secret",
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
-
- self.assertEqual(settings.anthropic_model, "claude-sonnet-4-6")
- self.assertIn("anthropic:claude-sonnet-4-6", settings.fallback_chain)
-
- def test_mask_secret_keeps_only_edges(self) -> None:
- self.assertEqual(mask_secret("abcd1234wxyz"), "abcd...wxyz")
-
- def test_resolve_repo_path_blocks_escape(self) -> None:
- root = self.make_scratch_dir().resolve()
- allowed = root / "notes.txt"
- allowed.write_text("ok", encoding="utf-8")
- self.assertEqual(resolve_repo_path(root, "notes.txt"), allowed)
- for unsafe_path in ("../outside.txt", "..\\outside.txt", "C:\\outside.txt"):
- with self.subTest(path=unsafe_path):
- with self.assertRaises(ValueError):
- resolve_repo_path(root, unsafe_path)
-
- def test_build_repo_tools_contains_expected_tool_names(self) -> None:
- tools = build_repo_tools(self.make_scratch_dir())
- names = {tool.name for tool in tools}
- self.assertEqual(
- names,
- {
- "get_repo_overview",
- "list_repo_files",
- "read_repo_file",
- "search_repo",
- "request_human_approval",
- "apply_repo_write_plan",
- },
- )
-
- def test_build_agent_and_workflow_without_network(self) -> None:
- root = self.make_scratch_dir()
- entities = root / "entities"
- repo = root / "repo"
- entities.mkdir()
- repo.mkdir()
- (repo / "README.md").write_text("hello", encoding="utf-8")
-
- with patch.dict(
- "os.environ",
- {
- "MAF_API_KEY": "test-key",
- "MAF_MODEL": "gemini-2.5-flash",
- "MAF_REPO_ROOT": str(repo),
- "MAF_ENTITIES_DIR": str(entities),
- "MAF_CHECKPOINT_DIR": str(root / "state"),
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
- agent = build_agent(settings)
- workflow = build_workflow(settings)
-
- self.assertEqual(agent.name, "repo_copilot")
- self.assertEqual(workflow.name, "repo_copilot_workflow")
-
- def test_build_repo_team_without_network(self) -> None:
- root = self.make_scratch_dir()
- entities = root / "entities"
- repo = root / "repo"
- entities.mkdir()
- repo.mkdir()
- (repo / "README.md").write_text("hello", encoding="utf-8")
-
- with patch.dict(
- "os.environ",
- {
- "MAF_API_KEY": "test-key",
- "MAF_MODEL": "gemini-2.5-flash",
- "MAF_REPO_ROOT": str(repo),
- "MAF_ENTITIES_DIR": str(entities),
- "MAF_CHECKPOINT_DIR": str(root / "state"),
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
- workflow = build_repo_team(settings)
-
- self.assertTrue(hasattr(workflow, "run"))
- self.assertIn("manager-led", workflow.description.lower())
- self.assertEqual(workflow.canonical_stages[0], "planning")
-
- def test_streaming_fallback_switches_before_first_update(self) -> None:
- async def failing_stream():
- raise RuntimeError("429 resource_exhausted")
- yield # pragma: no cover
-
- original_stream = ResponseStream(failing_stream(), finalizer=lambda updates: ChatResponse(messages=[]))
- context = ChatContext(
- client=object(),
- messages=[Message(role="user", text="Reply with READY")],
- options={},
- stream=True,
- )
- settings = load_test_settings()
-
- async def fallback_updates():
- yield ChatResponseUpdate(role="assistant", contents=[Content.from_text("READY")], model_id="fallback")
-
- with patch(
- "maf_starter.provider_fallback._execute_chain_step",
- new=AsyncMock(return_value=ResponseStream(fallback_updates(), finalizer=lambda updates: ChatResponse(messages=[]))),
- ) as execute_chain_step:
- wrapped = _wrap_stream_with_fallback(
- settings=settings,
- original_stream=original_stream,
- context=context,
- route=RoutingPlan(
- mode="auto",
- tier="deep",
- rationale="test",
- primary_provider="gemini",
- primary_model="gemini-2.5-pro",
- fallback_steps=(ChainStep("gemini", "gemini-2.5-flash"),),
- ),
- )
- updates = []
-
- async def consume():
- async for update in wrapped:
- updates.append(update.text)
- return await wrapped.get_final_response()
-
- final = __import__("asyncio").run(consume())
-
- self.assertEqual(updates, ["READY"])
- self.assertIsInstance(final, ChatResponse)
- execute_chain_step.assert_awaited()
-
- def test_resolve_run_scope_strips_kwargs_and_persists_scope_metadata(self) -> None:
- root = self.make_scratch_dir()
- entities = root / "entities"
- repo = root / "repo"
- entities.mkdir()
- repo.mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "MAF_API_KEY": "test-key",
- "MAF_MODEL": "gemini-2.5-flash",
- "MAF_REPO_ROOT": str(repo),
- "MAF_ENTITIES_DIR": str(entities),
- "MAF_CHECKPOINT_DIR": str(root / "state"),
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
-
- run_repo = root / "run-repo"
- run_repo.mkdir()
- run_checkpoint = root / "state" / "sessions" / "run-001" / "runtime" / "checkpoint"
- context = ChatContext(
- client=object(),
- messages=[Message(role="user", text="hello")],
- options={},
- kwargs={"repo_root": str(run_repo), "checkpoint_dir": str(run_checkpoint)},
- )
-
- scoped, tokens = _resolve_run_scope(settings, context)
- try:
- self.assertEqual(scoped.repo_root, run_repo.resolve())
- self.assertEqual(context.kwargs, {})
- self.assertEqual(context.metadata["repo_root"], str(run_repo.resolve()))
- self.assertEqual(context.metadata["checkpoint_dir"], str(run_checkpoint.resolve()))
- self.assertEqual(current_checkpoint_dir(settings.checkpoint_dir), run_checkpoint.resolve())
- finally:
- from maf_starter.config import reset_run_scope
-
- reset_run_scope(tokens)
-
- def test_auto_routing_plan_uses_simple_tier_for_light_prompt(self) -> None:
- settings = load_test_settings()
- plan = build_routing_plan(
- settings,
- routing_mode="auto",
- messages=[Message(role="user", text="hello")],
- primary_provider="gemini",
- primary_model="gemini-2.5-pro",
- )
- self.assertEqual(plan.tier, "simple")
- self.assertEqual(plan.primary_model, "gemini-2.5-flash-lite")
-
- def test_auto_routing_plan_uses_deep_tier_for_repo_work(self) -> None:
- settings = load_test_settings()
- plan = build_routing_plan(
- settings,
- routing_mode="auto",
- messages=[Message(role="user", text="Review this repo and propose an architecture plan")],
- primary_provider="gemini",
- primary_model="gemini-2.5-pro",
- )
- self.assertEqual(plan.tier, "deep")
- self.assertEqual(plan.primary_model, "gemini-2.5-pro")
-
- def test_devui_override_injects_route_assets_into_root_html(self) -> None:
- app = FastAPI()
-
- @app.get("/")
- async def root():
- return HTMLResponse("")
-
- install_devui_ui_overrides(app)
- client = TestClient(app)
- response = client.get("/")
-
- self.assertEqual(response.status_code, 200)
- self.assertIn("codex-route-overrides", response.text)
- self.assertIn("codex-route-panel__label", response.text)
-
- def test_patch_devui_bundle_rewrites_message_renderer(self) -> None:
- patched = _patch_devui_bundle(OLD_GD_RENDERER)
- self.assertIn("function CdxRouteParse", patched)
- self.assertIn("codex-route-panel", patched)
- self.assertIn("CdxRoutePanel", patched)
-
- def test_route_metadata_contains_tools_available_flag(self) -> None:
- settings = load_test_settings()
- metadata = _merge_route_metadata(
- None,
- settings=settings,
- route=RoutingPlan(
- mode="auto",
- tier="deep",
- rationale="test",
- primary_provider="gemini",
- primary_model="gemini-2.5-pro",
- fallback_steps=(),
- ),
- active_provider="gemini",
- active_model="gemini-2.5-pro",
- fallback_used=False,
- )
- self.assertIn("tools_available", metadata)
-
-
- def test_load_settings_ollama_no_api_key(self) -> None:
- root = self.make_scratch_dir()
- (root / "entities").mkdir()
- (root / "repo").mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "OLLAMA_BASE_URL": "http://localhost:11434/v1",
- "OLLAMA_MODEL": "gemma3",
- "MAF_REPO_ROOT": str(root / "repo"),
- "MAF_ENTITIES_DIR": str(root / "entities"),
- },
- clear=False,
- ):
- # Must not raise even though MAF_API_KEY and GEMINI_API_KEY are absent
- with patch.dict("os.environ", {"MAF_API_KEY": "", "GEMINI_API_KEY": ""}, clear=False):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
-
- self.assertEqual(settings.ollama_base_url, "http://localhost:11434/v1")
- self.assertEqual(settings.ollama_model, "gemma3")
-
- def test_load_settings_requires_key_or_ollama(self) -> None:
- root = self.make_scratch_dir()
- (root / "entities").mkdir()
- (root / "repo").mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "MAF_API_KEY": "",
- "GEMINI_API_KEY": "",
- "OLLAMA_BASE_URL": "",
- "MAF_REPO_ROOT": str(root / "repo"),
- "MAF_ENTITIES_DIR": str(root / "entities"),
- },
- clear=False,
- ):
- with self.assertRaises(ValueError) as ctx:
- load_settings(project_root=root, env_path=root / ".missing-env")
-
- self.assertIn("OLLAMA_BASE_URL", str(ctx.exception))
-
- def test_ollama_is_tier_0_when_configured(self) -> None:
- root = self.make_scratch_dir()
- (root / "entities").mkdir()
- (root / "repo").mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "OLLAMA_BASE_URL": "http://localhost:11434/v1",
- "OLLAMA_MODEL": "gemma3",
- "MAF_API_KEY": "",
- "GEMINI_API_KEY": "",
- "MAF_REPO_ROOT": str(root / "repo"),
- "MAF_ENTITIES_DIR": str(root / "entities"),
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
-
- self.assertTrue(settings.fallback_chain[0].startswith("ollama:"))
- self.assertIn("gemma3", settings.fallback_chain[0])
-
- def test_build_agent_uses_ollama_client(self) -> None:
- root = self.make_scratch_dir()
- (root / "entities").mkdir()
- (root / "repo").mkdir()
-
- with patch.dict(
- "os.environ",
- {
- "OLLAMA_BASE_URL": "http://localhost:11434/v1",
- "OLLAMA_MODEL": "gemma3",
- "MAF_API_KEY": "",
- "GEMINI_API_KEY": "",
- "MAF_REPO_ROOT": str(root / "repo"),
- "MAF_ENTITIES_DIR": str(root / "entities"),
- },
- clear=False,
- ):
- settings = load_settings(project_root=root, env_path=root / ".missing-env")
-
- with patch("maf_starter.agent_factory.build_fallback_middleware") as mock_mw:
- mock_mw.return_value = object()
- build_agent(settings)
- _, kwargs = mock_mw.call_args
- self.assertEqual(kwargs["primary_provider"], "ollama")
- self.assertEqual(kwargs["primary_model"], "gemma3")
-
- def test_env_example_documents_ollama(self) -> None:
- env_example = (
- Path(__file__).resolve().parents[1] / ".env.example"
- )
- text = env_example.read_text(encoding="utf-8")
- self.assertIn("OLLAMA_BASE_URL", text)
- self.assertIn("OLLAMA_MODEL", text)
- self.assertIn("gemma3", text)
-
-
-if __name__ == "__main__":
- unittest.main()
+from __future__ import annotations
+
+import unittest
+import uuid
+import shutil
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+from agent_framework import ChatContext, ChatResponse, ChatResponseUpdate, Message
+from agent_framework._types import Content
+from fastapi import FastAPI
+from fastapi.responses import HTMLResponse
+from fastapi.testclient import TestClient
+
+from maf_starter.agent_factory import build_agent
+from maf_starter.config import (
+ activate_run_scope,
+ current_checkpoint_dir,
+ current_repo_root,
+ load_settings,
+ mask_secret,
+ reset_run_scope,
+)
+from maf_starter.devui_overrides import OLD_GD_RENDERER
+from maf_starter.devui_overrides import _patch_devui_bundle
+from maf_starter.devui_overrides import install_devui_ui_overrides
+import maf_starter.provider_fallback as provider_fallback
+from maf_starter.provider_fallback import (
+ ResponseStream,
+ _is_fallback_worthy_error,
+ _merge_route_metadata,
+ _response_from_updates,
+ _resolve_run_scope,
+ _wrap_stream_with_fallback,
+)
+from maf_starter.routing_policy import RoutingPlan, build_routing_plan
+from maf_starter.routing_types import ChainStep, parse_chain_steps
+from maf_starter.team_factory import build_repo_team
+from maf_starter.tools import (
+ _iter_repo_files,
+ _read_text_file,
+ build_repo_context_snapshot,
+ build_repo_tools,
+ is_write_blocked_path,
+ resolve_repo_path,
+)
+from maf_starter.workflow_factory import build_workflow
+
+
+SCRATCH_ROOT = Path(__file__).resolve().parents[1] / ".tmp-tests"
+
+
+def load_test_settings():
+ with patch.dict("os.environ", {"MAF_API_KEY": "test-key"}, clear=False):
+ return load_settings(project_root=Path.cwd(), env_path=Path.cwd() / ".missing-env")
+
+
+class RepoScratchTestCase(unittest.TestCase):
+ def make_scratch_dir(self) -> Path:
+ path = SCRATCH_ROOT / uuid.uuid4().hex
+ path.mkdir(parents=True, exist_ok=False)
+ self.addCleanup(lambda: shutil.rmtree(path, ignore_errors=True))
+ return path
+
+
+class MafSetupTests(RepoScratchTestCase):
+ def test_parse_chain_steps(self) -> None:
+ steps = parse_chain_steps(
+ (
+ "gemini:gemini-2.5-pro",
+ "gemini-cli:gemini-2.5-pro",
+ "claude-cli",
+ )
+ )
+ self.assertEqual(
+ [(step.provider, step.model) for step in steps],
+ [
+ ("gemini", "gemini-2.5-pro"),
+ ("gemini-cli", "gemini-2.5-pro"),
+ ("claude-cli", None),
+ ],
+ )
+
+ def test_load_settings_prefers_maf_over_gemini_fallbacks(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "maf-secret",
+ "GEMINI_API_KEY": "gem-secret",
+ "MAF_MODEL": "gemini-2.5-flash",
+ "GEMINI_MODEL": "gemini-2.5-pro",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ self.assertEqual(settings.api_key, "maf-secret")
+ self.assertEqual(settings.model, "gemini-2.5-flash")
+ self.assertEqual(settings.repo_root, (root / "repo").resolve())
+ self.assertEqual(settings.entities_dir, (root / "entities").resolve())
+
+ def test_default_fallback_chain_includes_anthropic_when_key_exists(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "maf-secret",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ "MAF_FALLBACK_CHAIN": "",
+ "ANTHROPIC_API_KEY": "anthropic-secret",
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ self.assertEqual(settings.anthropic_model, "claude-sonnet-4-6")
+ self.assertIn("anthropic:claude-sonnet-4-6", settings.fallback_chain)
+
+ def test_mask_secret_keeps_only_edges(self) -> None:
+ self.assertEqual(mask_secret("abcd1234wxyz"), "abcd...wxyz")
+
+ def test_mask_secret_handles_missing_and_short_values(self) -> None:
+ self.assertEqual(mask_secret(""), "")
+ self.assertEqual(mask_secret("short"), "*****")
+
+ def test_resolve_repo_path_blocks_escape(self) -> None:
+ root = self.make_scratch_dir().resolve()
+ allowed = root / "notes.txt"
+ allowed.write_text("ok", encoding="utf-8")
+ self.assertEqual(resolve_repo_path(root, "notes.txt"), allowed)
+ for unsafe_path in ("../outside.txt", "..\\outside.txt", "C:\\outside.txt"):
+ with self.subTest(path=unsafe_path):
+ with self.assertRaises(ValueError):
+ resolve_repo_path(root, unsafe_path)
+
+ def test_resolve_repo_path_allows_missing_and_blocks_write_protected_paths(self) -> None:
+ root = self.make_scratch_dir().resolve()
+ missing = resolve_repo_path(root, "new-file.txt", allow_missing=True)
+ self.assertEqual(missing, root / "new-file.txt")
+
+ protected = root / ".env.local"
+ protected.write_text("SECRET=1", encoding="utf-8")
+ self.assertTrue(is_write_blocked_path(root, protected))
+ with self.assertRaisesRegex(ValueError, "blocked for write operations"):
+ resolve_repo_path(root, ".env.local", for_write=True)
+
+ def test_iter_repo_files_skips_internal_dirs_and_binary_reads_return_none(self) -> None:
+ root = self.make_scratch_dir().resolve()
+ visible = root / "visible.txt"
+ visible.write_text("ok", encoding="utf-8")
+ skipped_dir = root / ".venv"
+ skipped_dir.mkdir()
+ (skipped_dir / "hidden.txt").write_text("ignore", encoding="utf-8")
+ binary = root / "data.bin"
+ binary.write_bytes(b"\x00\x01\x02")
+
+ files = [path.relative_to(root).as_posix() for path in _iter_repo_files(root, "*")]
+ self.assertIn("visible.txt", files)
+ self.assertIn("data.bin", files)
+ self.assertNotIn(".venv/hidden.txt", files)
+ self.assertIsNone(_read_text_file(binary))
+
+ def test_build_repo_tools_contains_expected_tool_names(self) -> None:
+ tools = build_repo_tools(self.make_scratch_dir())
+ names = {tool.name for tool in tools}
+ self.assertTrue(
+ {
+ "get_repo_overview",
+ "list_repo_files",
+ "read_repo_file",
+ "search_repo",
+ "request_human_approval",
+ "apply_repo_write_plan",
+ }.issubset(names)
+ )
+
+ def test_build_repo_context_snapshot_captures_git_branch(self) -> None:
+ root = self.make_scratch_dir().resolve()
+ (root / "README.md").write_text("hello", encoding="utf-8")
+ subprocess_root = str(root)
+ __import__("subprocess").run(["git", "init", "-b", "main"], cwd=subprocess_root, check=True, capture_output=True)
+
+ snapshot = build_repo_context_snapshot(root)
+
+ self.assertEqual(snapshot["root"], str(root))
+ self.assertIn("README.md", snapshot["top_level_items"])
+ self.assertIn("main", str(snapshot["branch"]))
+ self.assertGreaterEqual(int(snapshot["file_count_hint"]), 1)
+
+ def test_build_agent_and_workflow_without_network(self) -> None:
+ root = self.make_scratch_dir()
+ entities = root / "entities"
+ repo = root / "repo"
+ entities.mkdir()
+ repo.mkdir()
+ (repo / "README.md").write_text("hello", encoding="utf-8")
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "test-key",
+ "MAF_MODEL": "gemini-2.5-flash",
+ "MAF_REPO_ROOT": str(repo),
+ "MAF_ENTITIES_DIR": str(entities),
+ "MAF_CHECKPOINT_DIR": str(root / "state"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+ agent = build_agent(settings)
+ workflow = build_workflow(settings)
+
+ self.assertEqual(agent.name, "repo_copilot")
+ self.assertEqual(workflow.name, "repo_copilot_workflow")
+
+ def test_build_repo_team_without_network(self) -> None:
+ root = self.make_scratch_dir()
+ entities = root / "entities"
+ repo = root / "repo"
+ entities.mkdir()
+ repo.mkdir()
+ (repo / "README.md").write_text("hello", encoding="utf-8")
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "test-key",
+ "MAF_MODEL": "gemini-2.5-flash",
+ "MAF_REPO_ROOT": str(repo),
+ "MAF_ENTITIES_DIR": str(entities),
+ "MAF_CHECKPOINT_DIR": str(root / "state"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+ workflow = build_repo_team(settings)
+
+ self.assertTrue(hasattr(workflow, "run"))
+ self.assertIn("manager-led", workflow.description.lower())
+ self.assertEqual(workflow.canonical_stages[0], "planning")
+
+ def test_load_settings_supports_ollama_without_api_key(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "",
+ "GEMINI_API_KEY": "",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ "OLLAMA_BASE_URL": "http://localhost:11434",
+ "OLLAMA_MODEL": "llama3.2",
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ self.assertEqual(settings.api_key, "")
+ self.assertEqual(settings.ollama_base_url, "http://localhost:11434")
+ self.assertEqual(settings.fallback_chain[0], "ollama:llama3.2")
+
+ def test_load_settings_rejects_missing_repo_root(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ missing_repo = root / "missing-repo"
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "test-key",
+ "MAF_REPO_ROOT": str(missing_repo),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ },
+ clear=False,
+ ):
+ with self.assertRaisesRegex(ValueError, "Configured repo root does not exist"):
+ load_settings(project_root=root, env_path=root / ".missing-env")
+
+ def test_streaming_fallback_switches_before_first_update(self) -> None:
+ async def failing_stream():
+ raise RuntimeError("429 resource_exhausted")
+ yield # pragma: no cover
+
+ original_stream = ResponseStream(failing_stream(), finalizer=lambda updates: ChatResponse(messages=[]))
+ context = ChatContext(
+ client=object(),
+ messages=[Message(role="user", text="Reply with READY")],
+ options={},
+ stream=True,
+ )
+ settings = load_test_settings()
+
+ async def fallback_updates():
+ yield ChatResponseUpdate(role="assistant", contents=[Content.from_text("READY")], model_id="fallback")
+
+ with patch(
+ "maf_starter.provider_fallback._execute_chain_step",
+ new=AsyncMock(return_value=ResponseStream(fallback_updates(), finalizer=lambda updates: ChatResponse(messages=[]))),
+ ) as execute_chain_step:
+ wrapped = _wrap_stream_with_fallback(
+ settings=settings,
+ original_stream=original_stream,
+ context=context,
+ route=RoutingPlan(
+ mode="auto",
+ tier="deep",
+ rationale="test",
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
+ fallback_steps=(ChainStep("gemini", "gemini-2.5-flash"),),
+ ),
+ )
+ updates = []
+
+ async def consume():
+ async for update in wrapped:
+ updates.append(update.text)
+ return await wrapped.get_final_response()
+
+ final = __import__("asyncio").run(consume())
+
+ self.assertEqual(updates, ["READY"])
+ self.assertIsInstance(final, ChatResponse)
+ execute_chain_step.assert_awaited()
+
+ def test_sanitize_messages_strips_unsupported_reasoning_properties(self) -> None:
+ messages = [
+ Message(
+ role="assistant",
+ text="intermediate",
+ additional_properties={"reasoning_details": {"token_count": 3}, "keep": "yes"},
+ raw_representation={"reasoningDetails": {"step": 1}, "other": "ok"},
+ ),
+ Message(role="user", text="final"),
+ ]
+
+ if not hasattr(provider_fallback, "_sanitize_messages"):
+ self.skipTest("_sanitize_messages helper is not available in this branch/runtime")
+
+ sanitized = provider_fallback._sanitize_messages(messages)
+
+ self.assertEqual(len(sanitized), 2)
+ self.assertNotIn("reasoning_details", sanitized[0].additional_properties)
+ self.assertEqual(sanitized[0].additional_properties["keep"], "yes")
+ self.assertEqual(sanitized[0].raw_representation, {"other": "ok"})
+ self.assertIs(sanitized[1], messages[1])
+
+ def test_resolve_run_scope_strips_kwargs_and_persists_scope_metadata(self) -> None:
+ root = self.make_scratch_dir()
+ entities = root / "entities"
+ repo = root / "repo"
+ entities.mkdir()
+ repo.mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "test-key",
+ "MAF_MODEL": "gemini-2.5-flash",
+ "MAF_REPO_ROOT": str(repo),
+ "MAF_ENTITIES_DIR": str(entities),
+ "MAF_CHECKPOINT_DIR": str(root / "state"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ run_repo = root / "run-repo"
+ run_repo.mkdir()
+ run_checkpoint = root / "state" / "sessions" / "run-001" / "runtime" / "checkpoint"
+ context = ChatContext(
+ client=object(),
+ messages=[Message(role="user", text="hello")],
+ options={},
+ kwargs={"repo_root": str(run_repo), "checkpoint_dir": str(run_checkpoint)},
+ )
+
+ scoped, tokens = _resolve_run_scope(settings, context)
+ try:
+ self.assertEqual(scoped.repo_root, run_repo.resolve())
+ self.assertEqual(context.kwargs, {})
+ self.assertEqual(context.metadata["repo_root"], str(run_repo.resolve()))
+ self.assertEqual(context.metadata["checkpoint_dir"], str(run_checkpoint.resolve()))
+ self.assertEqual(current_checkpoint_dir(settings.checkpoint_dir), run_checkpoint.resolve())
+ finally:
+ reset_run_scope(tokens)
+
+ def test_activate_run_scope_updates_repo_and_checkpoint_contextvars(self) -> None:
+ root = self.make_scratch_dir()
+ entities = root / "entities"
+ repo = root / "repo"
+ entities.mkdir()
+ repo.mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "test-key",
+ "MAF_REPO_ROOT": str(repo),
+ "MAF_ENTITIES_DIR": str(entities),
+ "MAF_CHECKPOINT_DIR": str(root / "state"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ run_repo = root / "alternate-repo"
+ run_repo.mkdir()
+ run_checkpoint = root / "alternate-state"
+ scoped, tokens = activate_run_scope(settings, repo_root=run_repo, checkpoint_dir=run_checkpoint)
+ try:
+ self.assertEqual(scoped.repo_root, run_repo.resolve())
+ self.assertEqual(scoped.checkpoint_dir, run_checkpoint.resolve())
+ self.assertEqual(current_repo_root(settings.repo_root), run_repo.resolve())
+ self.assertEqual(current_checkpoint_dir(settings.checkpoint_dir), run_checkpoint.resolve())
+ finally:
+ reset_run_scope(tokens)
+
+ def test_response_from_updates_handles_empty_and_populated_streams(self) -> None:
+ empty = _response_from_updates([])
+ self.assertEqual(empty.text, "")
+
+ populated = _response_from_updates(
+ [
+ ChatResponseUpdate(
+ role="assistant",
+ contents=[Content.from_text("READY")],
+ model_id="fallback-model",
+ additional_properties={"fallback_used": True},
+ )
+ ]
+ )
+ self.assertEqual(populated.text, "READY")
+ self.assertEqual(populated.model_id, "fallback-model")
+ self.assertTrue(populated.additional_properties["fallback_used"])
+
+ def test_fallback_worthy_error_matches_quota_markers_only(self) -> None:
+ self.assertTrue(_is_fallback_worthy_error(RuntimeError("429 RESOURCE_EXHAUSTED from upstream")))
+ self.assertTrue(_is_fallback_worthy_error(RuntimeError("provider hit rate limit")))
+ self.assertFalse(_is_fallback_worthy_error(RuntimeError("syntax error in local prompt")))
+
+ def test_auto_routing_plan_uses_simple_tier_for_light_prompt(self) -> None:
+ settings = load_test_settings()
+ plan = build_routing_plan(
+ settings,
+ routing_mode="auto",
+ messages=[Message(role="user", text="hello")],
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
+ )
+ self.assertEqual(plan.tier, "simple")
+ self.assertEqual(plan.primary_model, "gemini-2.5-flash-lite")
+
+ def test_auto_routing_plan_uses_deep_tier_for_repo_work(self) -> None:
+ settings = load_test_settings()
+ plan = build_routing_plan(
+ settings,
+ routing_mode="auto",
+ messages=[Message(role="user", text="Review this repo and propose an architecture plan")],
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
+ )
+ self.assertEqual(plan.tier, "deep")
+ self.assertEqual(plan.primary_model, "gemini-2.5-pro")
+
+ def test_devui_override_injects_route_assets_into_root_html(self) -> None:
+ app = FastAPI()
+
+ @app.get("/")
+ async def root():
+ return HTMLResponse("")
+
+ install_devui_ui_overrides(app)
+ client = TestClient(app)
+ response = client.get("/")
+
+ self.assertEqual(response.status_code, 200)
+ self.assertIn("codex-route-overrides", response.text)
+ self.assertIn("codex-route-panel__label", response.text)
+
+ def test_patch_devui_bundle_rewrites_message_renderer(self) -> None:
+ patched = _patch_devui_bundle(OLD_GD_RENDERER)
+ self.assertIn("function CdxRouteParse", patched)
+ self.assertIn("codex-route-panel", patched)
+ self.assertIn("CdxRoutePanel", patched)
+
+ def test_route_metadata_contains_tools_available_flag(self) -> None:
+ settings = load_test_settings()
+ metadata = _merge_route_metadata(
+ None,
+ settings=settings,
+ route=RoutingPlan(
+ mode="auto",
+ tier="deep",
+ rationale="test",
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
+ fallback_steps=(),
+ ),
+ active_provider="gemini",
+ active_model="gemini-2.5-pro",
+ fallback_used=False,
+ )
+ self.assertIn("tools_available", metadata)
+
+
+ def test_load_settings_ollama_no_api_key(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "OLLAMA_BASE_URL": "http://localhost:11434/v1",
+ "OLLAMA_MODEL": "gemma3",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ },
+ clear=False,
+ ):
+ # Must not raise even though MAF_API_KEY and GEMINI_API_KEY are absent
+ with patch.dict("os.environ", {"MAF_API_KEY": "", "GEMINI_API_KEY": ""}, clear=False):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ self.assertEqual(settings.ollama_base_url, "http://localhost:11434/v1")
+ self.assertEqual(settings.ollama_model, "gemma3")
+
+ def test_load_settings_requires_key_or_ollama(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "MAF_API_KEY": "",
+ "GEMINI_API_KEY": "",
+ "OLLAMA_BASE_URL": "",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ },
+ clear=False,
+ ):
+ with self.assertRaises(ValueError) as ctx:
+ load_settings(project_root=root, env_path=root / ".missing-env")
+
+ self.assertIn("OLLAMA_BASE_URL", str(ctx.exception))
+
+ def test_ollama_is_tier_0_when_configured(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "OLLAMA_BASE_URL": "http://localhost:11434/v1",
+ "OLLAMA_MODEL": "gemma3",
+ "MAF_API_KEY": "",
+ "GEMINI_API_KEY": "",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ self.assertTrue(settings.fallback_chain[0].startswith("ollama:"))
+ self.assertIn("gemma3", settings.fallback_chain[0])
+
+ def test_build_agent_uses_ollama_client(self) -> None:
+ root = self.make_scratch_dir()
+ (root / "entities").mkdir()
+ (root / "repo").mkdir()
+
+ with patch.dict(
+ "os.environ",
+ {
+ "OLLAMA_BASE_URL": "http://localhost:11434/v1",
+ "OLLAMA_MODEL": "gemma3",
+ "MAF_API_KEY": "",
+ "GEMINI_API_KEY": "",
+ "MAF_REPO_ROOT": str(root / "repo"),
+ "MAF_ENTITIES_DIR": str(root / "entities"),
+ },
+ clear=False,
+ ):
+ settings = load_settings(project_root=root, env_path=root / ".missing-env")
+
+ with patch("maf_starter.agent_factory.build_fallback_middleware") as mock_mw:
+ mock_mw.return_value = object()
+ build_agent(settings)
+ _, kwargs = mock_mw.call_args
+ self.assertEqual(kwargs["primary_provider"], "ollama")
+ self.assertEqual(kwargs["primary_model"], "gemma3")
+
+ def test_env_example_documents_ollama(self) -> None:
+ env_example = (
+ Path(__file__).resolve().parents[1] / ".env.example"
+ )
+ text = env_example.read_text(encoding="utf-8")
+ self.assertIn("OLLAMA_BASE_URL", text)
+ self.assertIn("OLLAMA_MODEL", text)
+ self.assertIn("gemma3", text)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_phase3_routing.py b/tests/test_phase3_routing.py
index 17cdb93..7565df2 100644
--- a/tests/test_phase3_routing.py
+++ b/tests/test_phase3_routing.py
@@ -1,32 +1,55 @@
-from __future__ import annotations
-
-import shutil
-import unittest
-import uuid
-from pathlib import Path
-from unittest.mock import patch
-
+from __future__ import annotations
+
+import shutil
+import unittest
+import uuid
+from pathlib import Path
+from unittest.mock import patch
+
from agent_framework import Message
+from agent_framework._types import Content
from maf_starter.config import Settings, load_settings
from maf_starter.provider_fallback import _merge_route_metadata
-from maf_starter.routing_policy import RoutingPlan, build_routing_plan
+from maf_starter.routing_policy import (
+ RoutingPlan,
+ _classify_tier,
+ _extract_last_user_text,
+ _normalize_route_lane,
+ build_routing_plan,
+)
from maf_starter.routing_types import ChainStep, RouteAttempt
-
-
-SCRATCH_ROOT = Path(__file__).resolve().parents[1] / ".tmp-tests"
-
-
-class RepoScratchTestCase(unittest.TestCase):
- def make_scratch_dir(self) -> Path:
- path = SCRATCH_ROOT / uuid.uuid4().hex
- path.mkdir(parents=True, exist_ok=False)
- self.addCleanup(lambda: shutil.rmtree(path, ignore_errors=True))
- return path
-
-
-class Phase3RoutingTests(RepoScratchTestCase):
+
+
+SCRATCH_ROOT = Path(__file__).resolve().parents[1] / ".tmp-tests"
+
+
+class RepoScratchTestCase(unittest.TestCase):
+ def make_scratch_dir(self) -> Path:
+ path = SCRATCH_ROOT / uuid.uuid4().hex
+ path.mkdir(parents=True, exist_ok=False)
+ self.addCleanup(lambda: shutil.rmtree(path, ignore_errors=True))
+ return path
+
+
+class Phase3RoutingTests(RepoScratchTestCase):
def _make_settings(self, root: Path, *, route_lane: str = "auto", requested_model: str | None = None) -> Settings:
+ entities = root / "entities"
+ repo = root / "repo"
+ entities.mkdir()
+ repo.mkdir()
+ env = {
+ "MAF_API_KEY": "test-key",
+ "MAF_REPO_ROOT": str(repo),
+ "MAF_ENTITIES_DIR": str(entities),
+ "MAF_ROUTE_LANE": route_lane,
+ }
+ if requested_model is not None:
+ env["MAF_REQUESTED_MODEL"] = requested_model
+ with patch.dict("os.environ", env, clear=False):
+ return load_settings(project_root=root, env_path=root / ".missing-env")
+
+ def _make_settings_with_env(self, root: Path, **extra_env: str) -> Settings:
entities = root / "entities"
repo = root / "repo"
entities.mkdir()
@@ -35,127 +58,208 @@ def _make_settings(self, root: Path, *, route_lane: str = "auto", requested_mode
"MAF_API_KEY": "test-key",
"MAF_REPO_ROOT": str(repo),
"MAF_ENTITIES_DIR": str(entities),
- "MAF_ROUTE_LANE": route_lane,
}
- if requested_model is not None:
- env["MAF_REQUESTED_MODEL"] = requested_model
+ env.update(extra_env)
with patch.dict("os.environ", env, clear=False):
return load_settings(project_root=root, env_path=root / ".missing-env")
def test_auto_lane_uses_prompt_classification_and_keeps_route_lane_explicit(self) -> None:
root = self.make_scratch_dir()
settings = self._make_settings(root, route_lane="auto")
+
+ plan = build_routing_plan(
+ settings,
+ routing_mode="auto",
+ messages=[Message(role="user", text="hello")],
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
+ )
+
+ self.assertEqual(plan.route_lane, "auto")
+ self.assertEqual(plan.tier, "simple")
+ self.assertEqual(plan.primary_model, "gemini-2.5-flash-lite")
+ self.assertEqual(plan.route_plan[0].label, "gemini:gemini-2.5-flash-lite")
+ self.assertGreaterEqual(plan.fallback_count, 1)
+
+ def test_balanced_lane_honors_requested_model_override(self) -> None:
+ root = self.make_scratch_dir()
+ settings = self._make_settings(root, route_lane="balanced", requested_model="gemini-2.5-pro")
+
+ plan = build_routing_plan(
+ settings,
+ routing_mode="auto",
+ messages=[Message(role="user", text="review this repo")],
+ primary_provider="gemini",
+ primary_model="gemini-2.5-flash",
+ )
+
+ self.assertEqual(plan.route_lane, "balanced")
+ self.assertEqual(plan.tier, "standard")
+ self.assertEqual(plan.requested_model, "gemini-2.5-pro")
+ self.assertEqual(plan.primary_model, "gemini-2.5-pro")
+ self.assertEqual(plan.route_plan[0].label, "gemini:gemini-2.5-pro")
+ self.assertTrue(plan.route_plan)
+ self.assertEqual(plan.route_plan[0], ChainStep("gemini", "gemini-2.5-pro"))
+
+ def test_invalid_lane_defaults_back_to_auto(self) -> None:
+ self.assertEqual(_normalize_route_lane("SIDEWAYS"), "auto")
+ self.assertEqual(_normalize_route_lane(" deep "), "deep")
+
+ def test_classify_tier_handles_empty_and_code_fenced_prompts(self) -> None:
+ self.assertEqual(_classify_tier("")[0], "standard")
+ self.assertEqual(_classify_tier("```python\nprint('hi')\n```")[0], "deep")
+
+ def test_extract_last_user_text_uses_text_contents_and_ignores_non_user_messages(self) -> None:
+ messages = [
+ Message(role="assistant", text="ignore this"),
+ Message(role="user", contents=[Content.from_text("First"), Content.from_text("Second")]),
+ ]
+
+ text = _extract_last_user_text(messages)
+
+ self.assertEqual(text, "First\nSecond")
+
+ def test_fast_lane_inserts_anthropic_after_primary(self) -> None:
+ root = self.make_scratch_dir()
+ settings = self._make_settings_with_env(
+ root,
+ MAF_ROUTE_LANE="fast",
+ ANTHROPIC_API_KEY="anthropic-secret",
+ )
plan = build_routing_plan(
settings,
- routing_mode="auto",
- messages=[Message(role="user", text="hello")],
+ routing_mode="fixed",
+ messages=[Message(role="user", text="quick summary")],
primary_provider="gemini",
- primary_model="gemini-2.5-pro",
+ primary_model="gemini-2.5-flash",
)
- self.assertEqual(plan.route_lane, "auto")
- self.assertEqual(plan.tier, "simple")
- self.assertEqual(plan.primary_model, "gemini-2.5-flash-lite")
+ self.assertEqual(plan.route_lane, "fast")
self.assertEqual(plan.route_plan[0].label, "gemini:gemini-2.5-flash-lite")
- self.assertGreaterEqual(plan.fallback_count, 1)
+ self.assertEqual(plan.fallback_steps[0].label, "gemini:gemini-2.5-flash")
+ self.assertEqual(plan.fallback_steps[1].label, "anthropic:claude-sonnet-4-6")
- def test_balanced_lane_honors_requested_model_override(self) -> None:
+ def test_requested_provider_reorders_balanced_lane_without_duplicate_primary(self) -> None:
root = self.make_scratch_dir()
- settings = self._make_settings(root, route_lane="balanced", requested_model="gemini-2.5-pro")
+ settings = self._make_settings_with_env(
+ root,
+ MAF_ROUTE_LANE="balanced",
+ MAF_REQUESTED_PROVIDER="codex-cli",
+ MAF_REQUESTED_MODEL="gpt-5",
+ )
plan = build_routing_plan(
settings,
- routing_mode="auto",
+ routing_mode="fixed",
messages=[Message(role="user", text="review this repo")],
primary_provider="gemini",
primary_model="gemini-2.5-flash",
)
- self.assertEqual(plan.route_lane, "balanced")
- self.assertEqual(plan.tier, "standard")
- self.assertEqual(plan.requested_model, "gemini-2.5-pro")
- self.assertEqual(plan.primary_model, "gemini-2.5-pro")
- self.assertEqual(plan.route_plan[0].label, "gemini:gemini-2.5-pro")
- self.assertTrue(plan.route_plan)
- self.assertEqual(plan.route_plan[0], ChainStep("gemini", "gemini-2.5-pro"))
+ self.assertEqual(plan.primary_provider, "codex-cli")
+ self.assertEqual(plan.primary_model, "gpt-5")
+ self.assertEqual(plan.route_plan[0].label, "codex-cli:gpt-5")
+ self.assertEqual(sum(1 for step in plan.route_plan if step.label == "codex-cli:gpt-5"), 1)
+ self.assertEqual(plan.fallback_steps[0].label, "gemini:gemini-2.5-flash")
- def test_route_metadata_records_attempts_and_capability_drift(self) -> None:
+ def test_ollama_route_becomes_primary_when_base_url_is_configured(self) -> None:
root = self.make_scratch_dir()
- settings = self._make_settings(root, route_lane="auto")
- route = RoutingPlan(
- mode="auto",
- route_lane="auto",
- tier="deep",
- rationale="test",
- primary_provider="gemini",
- primary_model="gemini-2.5-pro",
- requested_provider="gemini",
- requested_model="gemini-2.5-pro",
- fallback_steps=(ChainStep("gemini-cli", "gemini-2.5-pro"),),
- )
- attempt_log = [
- RouteAttempt(
- provider="gemini",
- model="gemini-2.5-pro",
- status="failed",
- tools_available=True,
- fallback_index=0,
- error="quota exhausted",
- ),
- RouteAttempt(
- provider="gemini-cli",
- model="gemini-2.5-pro",
- status="succeeded",
- tools_available=False,
- fallback_index=1,
- ),
- ]
-
- metadata = _merge_route_metadata(
- None,
- settings=settings,
- route=route,
- active_provider="gemini-cli",
- active_model="gemini-2.5-pro",
- fallback_used=True,
- tools_available=False,
- attempt_log=attempt_log,
- prior_error="quota exhausted",
+ settings = self._make_settings_with_env(
+ root,
+ MAF_ROUTE_LANE="auto",
+ OLLAMA_BASE_URL="http://localhost:11434",
+ OLLAMA_MODEL="llama3.2",
)
- self.assertEqual(metadata["route_lane"], "auto")
- self.assertEqual(metadata["requested_provider"], "gemini")
- self.assertEqual(metadata["requested_model"], "gemini-2.5-pro")
- self.assertEqual(metadata["fallback_count"], 1)
- self.assertEqual(metadata["route_attempts"][0]["provider"], "gemini")
- self.assertEqual(metadata["route_attempts"][1]["provider"], "gemini-cli")
- self.assertFalse(metadata["route_attempts"][1]["tools_available"])
- self.assertEqual(metadata["capability_changes"][0]["name"], "tools_available")
- self.assertFalse(metadata["capability_changes"][0]["after"])
- self.assertEqual(metadata["route_plan"][0]["provider"], "gemini")
- self.assertEqual(metadata["route_plan"][1]["provider"], "gemini-cli")
- self.assertTrue(metadata["fallback_used"])
- self.assertFalse(metadata["tools_available"])
-
- def test_route_attempt_payload_round_trips_cleanly(self) -> None:
- attempt = RouteAttempt(
- provider="gemini",
- model="gemini-2.5-pro",
- status="succeeded",
- tools_available=True,
- fallback_index=0,
- error=None,
- started_at="2026-03-21T12:00:00Z",
- completed_at="2026-03-21T12:00:02Z",
+ plan = build_routing_plan(
+ settings,
+ routing_mode="auto",
+ messages=[Message(role="user", text="debug this architecture")],
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
)
- payload = attempt.to_dict()
- hydrated = RouteAttempt.from_dict(payload)
-
- self.assertEqual(hydrated, attempt)
- self.assertEqual(payload["provider"], "gemini")
- self.assertEqual(payload["model"], "gemini-2.5-pro")
+ self.assertEqual(plan.primary_provider, "ollama")
+ self.assertEqual(plan.primary_model, "llama3.2")
+ self.assertEqual(plan.route_plan[1].label, "gemini:gemini-2.5-pro")
-if __name__ == "__main__":
- unittest.main()
+ def test_route_metadata_records_attempts_and_capability_drift(self) -> None:
+ root = self.make_scratch_dir()
+ settings = self._make_settings(root, route_lane="auto")
+ route = RoutingPlan(
+ mode="auto",
+ route_lane="auto",
+ tier="deep",
+ rationale="test",
+ primary_provider="gemini",
+ primary_model="gemini-2.5-pro",
+ requested_provider="gemini",
+ requested_model="gemini-2.5-pro",
+ fallback_steps=(ChainStep("gemini-cli", "gemini-2.5-pro"),),
+ )
+ attempt_log = [
+ RouteAttempt(
+ provider="gemini",
+ model="gemini-2.5-pro",
+ status="failed",
+ tools_available=True,
+ fallback_index=0,
+ error="quota exhausted",
+ ),
+ RouteAttempt(
+ provider="gemini-cli",
+ model="gemini-2.5-pro",
+ status="succeeded",
+ tools_available=False,
+ fallback_index=1,
+ ),
+ ]
+
+ metadata = _merge_route_metadata(
+ None,
+ settings=settings,
+ route=route,
+ active_provider="gemini-cli",
+ active_model="gemini-2.5-pro",
+ fallback_used=True,
+ tools_available=False,
+ attempt_log=attempt_log,
+ prior_error="quota exhausted",
+ )
+
+ self.assertEqual(metadata["route_lane"], "auto")
+ self.assertEqual(metadata["requested_provider"], "gemini")
+ self.assertEqual(metadata["requested_model"], "gemini-2.5-pro")
+ self.assertEqual(metadata["fallback_count"], 1)
+ self.assertEqual(metadata["route_attempts"][0]["provider"], "gemini")
+ self.assertEqual(metadata["route_attempts"][1]["provider"], "gemini-cli")
+ self.assertFalse(metadata["route_attempts"][1]["tools_available"])
+ self.assertEqual(metadata["capability_changes"][0]["name"], "tools_available")
+ self.assertFalse(metadata["capability_changes"][0]["after"])
+ self.assertEqual(metadata["route_plan"][0]["provider"], "gemini")
+ self.assertEqual(metadata["route_plan"][1]["provider"], "gemini-cli")
+ self.assertTrue(metadata["fallback_used"])
+ self.assertFalse(metadata["tools_available"])
+
+ def test_route_attempt_payload_round_trips_cleanly(self) -> None:
+ attempt = RouteAttempt(
+ provider="gemini",
+ model="gemini-2.5-pro",
+ status="succeeded",
+ tools_available=True,
+ fallback_index=0,
+ error=None,
+ started_at="2026-03-21T12:00:00Z",
+ completed_at="2026-03-21T12:00:02Z",
+ )
+ payload = attempt.to_dict()
+ hydrated = RouteAttempt.from_dict(payload)
+
+ self.assertEqual(hydrated, attempt)
+ self.assertEqual(payload["provider"], "gemini")
+ self.assertEqual(payload["model"], "gemini-2.5-pro")
+
+
+if __name__ == "__main__":
+ unittest.main()