diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0c3bf76..966ff07 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,7 +10,7 @@ name: ci # ruff is a pre-existing 347-error backlog — see step comment). # - macos: proves the install path on macOS-14. Cannot be validated from the # Linux dev box; needs a real macOS runner (see NOTE in the job). -# - wheel-canary: NON-BLOCKING honest red marker — builds + installs the wheel +# - wheel-canary: blocking install check — builds + installs the wheel # and imports the console entrypoints. # - integration: runs `pytest -m integration` so integration-marked tests # can't silently rot. Most skip gracefully when their external dependency @@ -46,13 +46,15 @@ jobs: # Isolate all state under a throwaway dir so tests never touch a real # ~/.cortex or the checkout's own state. CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | mkdir -p "$CORTEX_ROOT_DIR" # synthetic/tests/ depends on [analytics] (xgboost, shap) which we # don't install in smoke — excluded here (matches --collect-only # policy). This runs the FULL test BODIES, not just collection. set +e - python -m pytest -p no:cacheprovider --ignore=synthetic/tests -q \ + python -m pytest -p tests.network_guard -p no:cacheprovider --ignore=synthetic/tests -q --junitxml=/tmp/pytest.xml \ | tee /tmp/pytest.out code=${PIPESTATUS[0]} set -e @@ -62,25 +64,34 @@ jobs: # branch. A floor of 1960 catches accidental mass-skips or a # collection regression that silently drops tests while still # "passing". Bump the floor deliberately as the suite grows. - passed="$(grep -oE '[0-9]+ passed' /tmp/pytest.out | tail -1 | grep -oE '[0-9]+')" - echo "pytest reported ${passed} passed (floor: 1960)" - test -n "$passed" - test "$passed" -ge 1960 + python - <<'PYTEST_COUNTS' + import xml.etree.ElementTree as ET + root = ET.parse("/tmp/pytest.xml").getroot() + suites = [root] if root.tag == "testsuite" else list(root.findall("testsuite")) + passed = sum(int(s.get("tests", 0)) - int(s.get("failures", 0)) + - int(s.get("errors", 0)) - int(s.get("skipped", 0)) for s in suites) + print(f"JUnit reported {passed} passed (floor: 1960)") + assert passed >= 1960, f"Unexpected test count: {passed}" + PYTEST_COUNTS - name: cortex demo must print the FK trail env: CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | output="$(cortex demo)" echo "$output" # Three synthesized prompts must each appear as a linked entry # with score 0.80. If this assertion ever fails, the headline - # 'compounding intelligence' claim is broken on this build. + # synthetic linkage fixture is broken; this does not prove useful learning. echo "$output" | grep -c 'score 0.80' | grep -q '^3$' - name: cortex briefing must exit 2 without ANTHROPIC_API_KEY env: CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | # Pre-flight contract: missing key → exit 2 with actionable msg. # Anything else (silent hang, exit 0, exit 1) regresses Phase 1. @@ -94,6 +105,8 @@ jobs: - name: MCP smoke drill — bridge DOWN (crash-proof headline) env: CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | # No bridge is started. The core-8 in-process MCP tools must still be # green (crash-proof contract). Passthrough tools that need the bridge @@ -133,11 +146,15 @@ jobs: - name: ruff check (release-touched paths) run: | - # Scoped to the Python files this release (Workstream F) touched. - # Both are ruff-clean; widen this list as backlog dirs reach zero. + # Core release paths plus the execution-integrity regression surface. + # Widen this list as other backlog directories reach zero. python -m ruff check \ bridge_intelligence.py \ - tests/contract/test_phase5_invariants.py + tests/contract/test_phase5_invariants.py \ + supervisor/dispatch.py supervisor/collector.py supervisor/router.py \ + store_registry.py tests/test_dispatch_integrity.py \ + tests/test_orchestration_pipeline.py tests/test_model_router.py \ + tests/test_doctor_json_reset.py tests/test_mcp_research_tools.py tests/network_guard.py macos: # macOS-14 install path. NOTE: this job CANNOT be validated from the Linux @@ -166,6 +183,8 @@ jobs: - name: cortex doctor must run without crashing (non-zero exit allowed) env: CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | mkdir -p "$CORTEX_ROOT_DIR" # doctor reports env facts (API key, launchd, bridge) — a red env is @@ -184,6 +203,8 @@ jobs: - name: cortex demo must print the FK trail env: CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | output="$(cortex demo)" echo "$output" @@ -192,17 +213,14 @@ jobs: - name: pytest (full suite, hermetic) — Python 3.12 env: CORTEX_ROOT_DIR: ${{ runner.temp }}/cortex-home + CORTEX_STATE_DIR: ${{ runner.temp }}/cortex-state + CORTEX_HOME: ${{ runner.temp }}/cortex-state run: | - python -m pytest -p no:cacheprovider --ignore=synthetic/tests -q + python -m pytest -p tests.network_guard -p no:cacheprovider --ignore=synthetic/tests -q wheel-canary: - # NON-BLOCKING honest red marker for the git-clone-only beta. Builds the - # wheel, installs it into a clean venv, and imports the console entrypoints. - # Kept continue-on-error per the release plan even though the packaging was - # fixed on this branch and it is expected to pass locally — a green here is - # a bonus, a red here is a signal, neither blocks the release gate. + # The deployed artifact must contain the diagnostics as well as entrypoints. runs-on: ubuntu-latest - continue-on-error: true steps: - uses: actions/checkout@v4 @@ -226,7 +244,7 @@ jobs: # The two console-script entrypoints must import from the installed # wheel (not the checkout) — run from /tmp so cwd isn't the repo root. cd /tmp - /tmp/wheelvenv/bin/python -c "import cli, mcp_server; print('wheel imports OK')" + /tmp/wheelvenv/bin/python -c "import cli, mcp_server, store_registry, recall_confidence; print('wheel imports OK')" /tmp/wheelvenv/bin/cortex --help integration: diff --git a/docs/design/truthful-execution.md b/docs/design/truthful-execution.md new file mode 100644 index 0000000..57848e3 --- /dev/null +++ b/docs/design/truthful-execution.md @@ -0,0 +1,76 @@ +# Execution receipts and unverified outcomes + +This change is a containment step, not a completed autonomous learning system. + +## Contract + +- `DispatchResult.success` and summary `succeeded` describe execution only. +- `execution_status` distinguishes completed, failed, incomplete, timed_out, + unsupported, and dry_run. Dry runs are neither successful nor failed execution. +- Truncation, refusal, missing final text, and tool-turn exhaustion cannot + produce successful Anthropic execution. Partial output and known token usage + survive incomplete responses. A batch API `succeeded` envelope is insufficient. +- Direct and message-batch adapters reject explicit shell commands; they do not + have a command-execution capability. Free-text capability inference is not solved. +- Persisted results and routing receipts say `verification_status=unverified` + and `learning_eligible=false`. A successful response without independent + verification is counted as unverified and flagged for review. +- `VerificationEvidence` is a legacy heuristic assessment. Even its maximum + score cannot verify a task or automatically requeue it. It is serialized under + `assessment`, not `verification`. +- Each dispatch has an attempt ID; batch IDs plus custom IDs provide stable + attempt identities on retrieval. Assessments attach to attempts. Ambiguous + work-item-only attachments fail rather than attach to the wrong retry. +- Repeated collection of the same receipt in one collector is idempotent; + conflicting receipts fail. Run directories cannot overwrite another run in + the same second. Cross-process journal deduplication is still pending. + +## Routing behavior change + +ModelRouter retains existing journals and appends execution observations, but +does not ingest them into adaptive outcome statistics or update those statistics +from new execution-only observations. Static task/complexity routing continues. +Batch completion no longer fabricates quality 1.0. Existing quality heuristics +remain advisory telemetry under `heuristic_quality_score`; measured `quality_score` +remains null, not a learning signal. + +There is deliberately no switch to re-enable unverified history. Adaptive +outcome routing needs a separate ingestion contract that binds task, attempt, +artifact revision, check version, coverage and result, and supports corrections. +This pause concerns this supervisor router; it does not claim to disable every +other learning or retrieval heuristic in Cortex. + +## Compatibility and limits + +Consumers reading `verification` must migrate to the explicitly advisory +`assessment` field. Summary success counts are retained for compatibility but +must be labeled execution counts in downstream interfaces. Evidence attachment +before collection, or ambiguous attachment after retry, now raises ValueError. + +Local providers currently return only text/token tuples, so their truncation +status cannot be established here. Nonempty local output remains unverified. +Timeout usage may be unknown, and cancelling an asyncio thread wrapper does not +prove an underlying network request stopped. No exactly-once external execution +guarantee is made. + +Path validation resolves components instead of comparing string prefixes. It +rejects sibling-prefix and symlink escapes under a stable filesystem. This is +not an OS sandbox against a hostile concurrent filesystem mutation. + +## Release and next gates + +The focused regression suite is `tests/test_dispatch_integrity.py`. It exercises +real dispatcher/collector/router code with a synthetic provider and actual +temporary filesystem persistence. No live-model or installed-machine claim is +made by that suite. CI also gates touched files and installed-wheel imports. + +Next: transactional event identity and correction, unified memory reads/writes, +atomic supersession, capture privacy, dependency/lease/cancellation enforcement, +and an independent exact-artifact verifier. Then prove a real interrupted work +loop and useful lesson reuse. Machine cutover requires a verified source/state +inventory, a consistent backup, and an explicit rollback path. + +The offline pytest plugin also checks intended HTTP destinations before proxy +transport, permits in-memory test transports, and blocks external Python sockets. +Use a credential-free environment for broad test runs. This is not an OS-level +sandbox and does not constrain arbitrary subprocess network access. diff --git a/pyproject.toml b/pyproject.toml index 53cf888..a43df04 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -92,11 +92,12 @@ Repository = "https://github.com/jessekemp1/cortex" # wheel (mcp_handlers, then state_paths, then config… ModuleNotFoundError one # import deeper each time), so we package every root module. All 42 are # import-safe (no top-level side effects), so unused ones are inert. -py-modules = ["agent_factory", "ai_intelligence", "alert_monitor", "bridge", "bridge_intelligence", "bridge_system", "briefing_resilient", "config", "context_intelligence", "data_migration", "deep_assessment", "env_loader", "feedback", "formatter", "goal_commands", "goal_parser", "goal_velocity", "heartbeat", "learning", "learning_config", "mcp_handlers", "mcp_server", "metrics_tracker", "orchestrator", "portfolio_analyzer", "portfolio_memory", "recommendation_engine", "recommendations", "reflection", "scheduler", "security", "self_audit", "session_cache", "session_delta", "session_manager", "session_snapshot", "session_watcher", "setup_ops", "spec_knowledge_base", "state_paths", "task_discovery", "validation", "watches", "weekly_planner"] +py-modules = ["agent_factory", "ai_intelligence", "alert_monitor", "bridge", "bridge_intelligence", "bridge_system", "briefing_resilient", "config", "context_intelligence", "data_migration", "deep_assessment", "env_loader", "feedback", "formatter", "goal_commands", "goal_parser", "goal_velocity", "heartbeat", "learning", "learning_config", "mcp_handlers", "mcp_server", "metrics_tracker", "orchestrator", "portfolio_analyzer", "portfolio_memory", "recommendation_engine", "recommendations", "reflection", "scheduler", "security", "self_audit", "session_cache", "session_delta", "session_manager", "session_snapshot", "session_watcher", "setup_ops", "spec_knowledge_base", "state_paths", "store_registry", "recall_confidence", "task_discovery", "validation", "watches", "weekly_planner"] [tool.setuptools.packages.find] exclude = [ "tests*", + "build*", "docs*", "examples*", "reports*", diff --git a/store_registry.py b/store_registry.py index 05be9f7..533891b 100644 --- a/store_registry.py +++ b/store_registry.py @@ -28,11 +28,13 @@ from pathlib import Path from typing import List, Optional +from state_paths import get_cortex_dir + WRITER_NONE = "NONE" def _cortex_dir() -> Path: - return Path.home() / ".cortex" + return get_cortex_dir() def _portfolio_dir() -> Path: @@ -134,8 +136,8 @@ class StoreStatus: store: Store exists: bool age_days: Optional[float] - problems: List[str] = field(default_factory=list) # hard failures - warnings: List[str] = field(default_factory=list) # advisory, not a failure + problems: List[str] = field(default_factory=list) # hard failures + warnings: List[str] = field(default_factory=list) # advisory, not a failure @property def ok(self) -> bool: @@ -166,9 +168,7 @@ def evaluate(stores: Optional[List[Store]] = None) -> List[StoreStatus]: if not exists: problems.append(f"missing at {s.path}") elif not s.event_driven: - past_sla = ( - s.max_age_days is not None and age is not None and age > s.max_age_days - ) + past_sla = s.max_age_days is not None and age is not None and age > s.max_age_days if past_sla and s.scheduled: # A store its own scheduler is supposed to keep fresh is stale: # the scheduler is broken. Hard fail — someone must fix it. @@ -187,7 +187,11 @@ def evaluate(stores: Optional[List[Store]] = None) -> List[StoreStatus]: out.append( StoreStatus( - store=s, exists=exists, age_days=age, problems=problems, warnings=warnings + store=s, + exists=exists, + age_days=age, + problems=problems, + warnings=warnings, ) ) return out diff --git a/supervisor/collector.py b/supervisor/collector.py index 4a960d3..00b4411 100644 --- a/supervisor/collector.py +++ b/supervisor/collector.py @@ -26,7 +26,7 @@ @dataclass class VerificationEvidence: - """Structured evidence of verification quality for a dispatched task. + """Legacy heuristic assessment, not an independent verification receipt. Score breakdown (0-100): - tests_ran: up to 20 pts (10+ tests = max) @@ -35,10 +35,8 @@ class VerificationEvidence: - regression_free: 20 pts - golden_match: 8 pts - Interpretation: - score >= 60 → verified, proceed - 40 <= score < 60 → flag for human review - score < 40 → auto-reject, re-queue with different approach + A score cannot prove which artifact was checked or what passed. These + fields remain useful review hints; they never verify or auto-requeue work. """ tests_ran: int = 0 # how many tests actually executed @@ -61,13 +59,13 @@ def score(self) -> float: @property def needs_review(self) -> bool: - """Score below 60 or human explicitly required.""" - return self.score < 60.0 or self.human_required + """Independent verification is required regardless of heuristic score.""" + return True @property def auto_reject(self) -> bool: - """Score below 40 — re-queue with different approach.""" - return self.score < 40.0 + """A heuristic alone must not trigger automatic repeated execution.""" + return False def to_dict(self) -> dict: return { @@ -85,7 +83,7 @@ def to_dict(self) -> dict: @dataclass class BatchSummary: - """Aggregated summary for a batch of dispatches.""" + """Execution counts only; succeeded does not mean verified task success.""" total: int succeeded: int @@ -96,6 +94,8 @@ class BatchSummary: errors: list[str] flagged_for_review: int = 0 # results where verification.needs_review auto_rejected: int = 0 # results where verification.auto_reject + unverified: int = 0 + dry_runs: int = 0 class ResultCollector: @@ -118,7 +118,7 @@ def __init__( self._results_dir = results_dir or _DEFAULT_RESULTS_DIR self._outcomes_path = outcomes_path or _DEFAULT_OUTCOMES_PATH self._results: list[DispatchResult] = [] - self._evidence: dict[str, VerificationEvidence] = {} # work_item_id -> evidence + self._evidence: dict[str, VerificationEvidence] = {} # attempt_id -> assessment # ------------------------------------------------------------------ # Collection @@ -126,6 +126,11 @@ def __init__( def collect(self, result: DispatchResult) -> None: """Add a single :class:`DispatchResult` to the current batch.""" + for existing in self._results: + if existing.attempt_id == result.attempt_id: + if existing != result: + raise ValueError("Conflicting execution receipts for the same attempt") + return self._results.append(result) def collect_batch( @@ -140,10 +145,12 @@ def collect_batch( evidence: Optional per-result verification evidence (parallel list). If provided, flagged/rejected counts are updated in the summary. """ + if evidence is not None and len(evidence) != len(results): + raise ValueError("Evidence must have exactly one assessment per result") for i, r in enumerate(results): - self._results.append(r) + self.collect(r) if evidence and i < len(evidence): - self._evidence[r.work_item_id] = evidence[i] + self._evidence[r.attempt_id] = evidence[i] return self.get_summary() # ------------------------------------------------------------------ @@ -169,16 +176,21 @@ def record_outcome( quality is persisted alongside the outcome so routing can learn which task types produce trustworthy outputs. """ - ev = evidence or self._evidence.get(result.work_item_id) + ev = evidence or self._evidence.get(result.attempt_id) model_tier = _tier_from_model_id(result.model_used) entry: dict = { "timestamp": datetime.now(tz=timezone.utc).isoformat(), "work_item_id": result.work_item_id, + "attempt_id": result.attempt_id, + "execution_status": result.execution_status, + "verification_status": "unverified", + "learning_eligible": False, "model_tier": model_tier, "model_id": result.model_used, "task_type": result.task_type, "success": result.success, - "quality_score": quality_score, + "quality_score": None, + "heuristic_quality_score": quality_score, "tokens_used": result.tokens_used, "duration_seconds": result.duration_seconds, "provider": getattr(result, "provider", "anthropic"), @@ -186,7 +198,7 @@ def record_outcome( "is_baseline": is_baseline, } if ev is not None: - entry["verification"] = ev.to_dict() + entry["assessment"] = ev.to_dict() self._outcomes_path.parent.mkdir(parents=True, exist_ok=True) with self._outcomes_path.open("a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n") @@ -199,7 +211,7 @@ def record_outcome( def get_summary(self) -> BatchSummary: """Build and return a :class:`BatchSummary` for collected results.""" succeeded = sum(1 for r in self._results if r.success) - failed = sum(1 for r in self._results if not r.success) + failed = sum(1 for r in self._results if not r.success and r.execution_status != "dry_run") total_tokens = sum(r.tokens_used for r in self._results) total_duration = sum(r.duration_seconds for r in self._results) @@ -210,16 +222,9 @@ def get_summary(self) -> BatchSummary: errors = [r.error for r in self._results if r.error] - flagged = sum( - 1 - for r in self._results - if (ev := self._evidence.get(r.work_item_id)) and ev.needs_review - ) - rejected = sum( - 1 - for r in self._results - if (ev := self._evidence.get(r.work_item_id)) and ev.auto_reject - ) + # No independent verifier is wired here. Missing evidence and even a + # perfect heuristic assessment leave successful executions unverified. + flagged = succeeded return BatchSummary( total=len(self._results), @@ -230,7 +235,9 @@ def get_summary(self) -> BatchSummary: model_breakdown=breakdown, errors=errors, flagged_for_review=flagged, - auto_rejected=rejected, + auto_rejected=0, + unverified=succeeded, + dry_runs=sum(r.execution_status == "dry_run" for r in self._results), ) # ------------------------------------------------------------------ @@ -240,12 +247,17 @@ def get_summary(self) -> BatchSummary: def persist(self) -> Path: """Write results and summary to disk. Returns the run directory path.""" timestamp = datetime.now(tz=timezone.utc).strftime("%Y%m%dT%H%M%SZ") - run_dir = self._results_dir / timestamp - run_dir.mkdir(parents=True, exist_ok=True) + from uuid import uuid4 + + run_dir = self._results_dir / f"{timestamp}-{uuid4().hex}" + run_dir.mkdir(parents=True, exist_ok=False) # results.json — full result list results_path = run_dir / "results.json" results_data = [_dispatch_result_to_dict(r) for r in self._results] + for record, result in zip(results_data, self._results): + if result.attempt_id in self._evidence: + record["assessment"] = self._evidence[result.attempt_id].to_dict() results_path.write_text(json.dumps(results_data, indent=2) + "\n") # summary.json @@ -265,12 +277,25 @@ def clear(self) -> None: self._results.clear() self._evidence.clear() - def attach_evidence(self, work_item_id: str, evidence: VerificationEvidence) -> None: + def attach_evidence( + self, + work_item_id: str, + evidence: VerificationEvidence, + *, + attempt_id: str | None = None, + ) -> None: """Attach verification evidence to a previously collected result. Useful when verification runs asynchronously after dispatch. """ - self._evidence[work_item_id] = evidence + matches = [ + r + for r in self._results + if r.work_item_id == work_item_id and (attempt_id is None or r.attempt_id == attempt_id) + ] + if len(matches) != 1: + raise ValueError("Evidence requires exactly one collected attempt; supply attempt_id") + self._evidence[matches[0].attempt_id] = evidence log.debug("evidence attached: work_item=%s score=%.1f", work_item_id, evidence.score) @@ -292,6 +317,10 @@ def _dispatch_result_to_dict(result: DispatchResult) -> dict: """Serialise a :class:`DispatchResult` to a plain dict.""" return { "work_item_id": result.work_item_id, + "attempt_id": result.attempt_id, + "execution_status": result.execution_status, + "verification_status": "unverified", + "learning_eligible": False, "success": result.success, "output": result.output, "model_used": result.model_used, diff --git a/supervisor/dispatch.py b/supervisor/dispatch.py index 2e333c4..3c9b82b 100644 --- a/supervisor/dispatch.py +++ b/supervisor/dispatch.py @@ -16,9 +16,10 @@ import os import subprocess import time -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any, Optional +from uuid import uuid4 from cortex.supervisor.models import WorkItem @@ -110,7 +111,7 @@ @dataclass class DispatchResult: - """Outcome of a single agent dispatch.""" + """Execution receipt, not proof that a task's acceptance criteria passed.""" work_item_id: str success: bool @@ -123,6 +124,23 @@ class DispatchResult: checkpoint_id: Optional[str] = None # For resume provider: str = "anthropic" # Which provider handled this dispatch cost_usd: float = 0.0 # Estimated cost in USD + attempt_id: str = field(default_factory=lambda: uuid4().hex) + execution_status: str = "" + + def __post_init__(self) -> None: + if not self.execution_status: + self.execution_status = "completed" if self.success else "failed" + if self.execution_status != "completed": + self.success = False + + +class IncompleteDispatch(RuntimeError): + """Provider stopped without completing, preserving partial output and usage.""" + + def __init__(self, reason: str, output: str, tokens: int) -> None: + super().__init__(reason) + self.output = output + self.tokens = tokens class AgentDispatcher: @@ -171,15 +189,29 @@ async def dispatch_async( ) return DispatchResult( work_item_id=work_item.id, - success=True, + success=False, output=f"[DRY RUN] Would dispatch to {model_selection.model_id}", model_used=model_selection.model_id, tokens_used=0, duration_seconds=0.0, task_type=work_item.task_type, + execution_status="dry_run", ) timeout = MODEL_TIMEOUTS.get(model_selection.model_tier, 120.0) + if work_item.command: + return DispatchResult( + work_item_id=work_item.id, + success=False, + output="", + model_used=model_selection.model_id, + tokens_used=0, + duration_seconds=0, + task_type=work_item.task_type, + provider=model_selection.provider, + execution_status="unsupported", + error="This adapter cannot execute commands; route to a command-capable harness", + ) prompt = self._build_prompt(work_item) system_prompt = self._build_system_prompt(work_item) project_root = self._resolve_project_root(work_item) @@ -206,6 +238,20 @@ async def dispatch_async( tokens_used=tokens, duration_seconds=round(elapsed, 3), task_type=work_item.task_type, + provider=model_selection.provider, + ) + except IncompleteDispatch as exc: + return DispatchResult( + work_item_id=work_item.id, + success=False, + output=exc.output, + model_used=model_selection.model_id, + tokens_used=exc.tokens, + duration_seconds=round(time.monotonic() - start, 3), + task_type=work_item.task_type, + provider=model_selection.provider, + error=str(exc), + execution_status="incomplete", ) except asyncio.TimeoutError: elapsed = time.monotonic() - start @@ -224,6 +270,8 @@ async def dispatch_async( duration_seconds=round(elapsed, 3), task_type=work_item.task_type, error=f"Timed out after {timeout}s", + execution_status="timed_out", + provider=model_selection.provider, ) except Exception as exc: elapsed = time.monotonic() - start @@ -241,6 +289,7 @@ async def dispatch_async( duration_seconds=round(elapsed, 3), task_type=work_item.task_type, error=str(exc), + provider=model_selection.provider, ) def dispatch_and_record( @@ -296,6 +345,12 @@ def submit_batch( """ if not items: raise ValueError("Batch must contain at least one item") + if self._dry_run: + raise ValueError("Dry-run dispatchers cannot submit remote batches") + if len({work_item.id for work_item, _ in items}) != len(items): + raise ValueError("Batch work item IDs must be unique") + if any(work_item.command for work_item, _ in items): + raise ValueError("Message batches cannot execute commands") client = self._get_client() requests = [] @@ -384,12 +439,23 @@ def retrieve_batch( tokens = msg.usage.input_tokens + msg.usage.output_tokens dr = DispatchResult( work_item_id=cid, - success=True, + success=(msg.stop_reason == "end_turn" and bool(text.strip())), output=text, model_used=meta_entry.get("model_id", ""), tokens_used=tokens, duration_seconds=0.0, task_type=meta_entry.get("task_type", ""), + attempt_id=f"{batch_id}:{cid}", + execution_status=( + "completed" + if msg.stop_reason == "end_turn" and text.strip() + else "incomplete" + ), + error=( + None + if msg.stop_reason == "end_turn" and text.strip() + else f"Incomplete batch response: {msg.stop_reason}" + ), ) else: dr = DispatchResult( @@ -401,6 +467,7 @@ def retrieve_batch( duration_seconds=0.0, task_type=meta_entry.get("task_type", ""), error=f"Batch result: {entry.result.type}", + attempt_id=f"{batch_id}:{cid}", ) results.append(dr) @@ -411,7 +478,7 @@ def retrieve_batch( work_item_id=cid, model_tier=meta_entry.get("model_tier", "sonnet"), success=dr.success, - quality_score=1.0 if dr.success else 0.0, + quality_score=None, task_type=dr.task_type, ) @@ -529,6 +596,8 @@ async def _run_local_dispatch( "Is Ollama running? `ollama serve`" ) text, tokens = await asyncio.to_thread(provider.complete, prompt, system_prompt, max_tokens) + if not text.strip(): + raise IncompleteDispatch("Empty local response", text, tokens) return text, tokens async def _run_dispatch( @@ -544,7 +613,7 @@ async def _run_dispatch( Sends initial prompt, then loops: 1. If response has tool_use blocks → execute tools, send results back 2. If response has only text blocks → done, return text - 3. If max turns exceeded → return accumulated text + 3. If max turns exceeded → report incomplete execution """ client = self._get_client() messages: list[dict] = [{"role": "user", "content": prompt}] @@ -563,9 +632,14 @@ async def _run_dispatch( # Check for tool_use blocks tool_uses = [b for b in response.content if b.type == "tool_use"] + text = "\n".join(b.text for b in response.content if b.type == "text") + if response.stop_reason not in ("end_turn", "tool_use"): + raise IncompleteDispatch( + f"Incomplete response: {response.stop_reason}", text, total_tokens + ) if not tool_uses: - # Final text response — no more tools needed - text = "\n".join(b.text for b in response.content if b.type == "text") + if response.stop_reason != "end_turn" or not text.strip(): + raise IncompleteDispatch("Missing final response", text, total_tokens) return text, total_tokens # Append assistant response, then execute tools and send results back @@ -585,7 +659,7 @@ async def _run_dispatch( # Max turns exceeded — extract whatever text is available text_parts = [b.text for b in response.content if b.type == "text"] text = "\n".join(text_parts) if text_parts else "(max turns exceeded)" - return text, total_tokens + raise IncompleteDispatch("Maximum tool turns exceeded", text, total_tokens) # ------------------------------------------------------------------ # Tool execution (read-only) @@ -613,7 +687,7 @@ def _validate_path(self, raw_path: str, root: Path) -> tuple[Path | None, str]: target = (root / raw_path).resolve() except (ValueError, OSError) as exc: return None, f"Error: invalid path: {exc}" - if not str(target).startswith(str(root)): + if not target.is_relative_to(root.resolve()): return None, "Error: access denied — path outside project root" return target, "" @@ -644,10 +718,17 @@ def _tool_search_code(self, tool_input: dict, root: Path) -> str: if err: return err - cmd = ["grep", "-rn", "--include=*.py", pattern, str(target)] + cmd = ["grep", "-rn", "--include=*.py", "--", pattern, str(target)] glob_filter = tool_input.get("glob") if glob_filter: - cmd = ["grep", "-rn", f"--include={glob_filter}", pattern, str(target)] + cmd = [ + "grep", + "-rn", + f"--include={glob_filter}", + "--", + pattern, + str(target), + ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=_SEARCH_TIMEOUT) @@ -673,6 +754,10 @@ def _tool_list_files(self, tool_input: dict, root: Path) -> str: matches = sorted(target.glob(pattern))[:_MAX_LIST_FILES] if not matches: return "(no files match)" - return "\n".join(str(m.relative_to(root)) for m in matches if m.is_file()) + return "\n".join( + str(m.relative_to(root)) + for m in matches + if m.resolve().is_relative_to(root) and m.is_file() + ) except (OSError, ValueError) as exc: return f"Error listing files: {exc}" diff --git a/supervisor/router.py b/supervisor/router.py index 89238ee..d035e72 100644 --- a/supervisor/router.py +++ b/supervisor/router.py @@ -1,8 +1,8 @@ """ Model complexity router -- selects opus/sonnet/haiku by task complexity. -Uses outcome data from ~/.cortex/metrics/model_outcomes.jsonl to learn -which model tier handles which task types most effectively. +Execution telemetry is preserved, but adaptive outcome routing is paused until +independent, version-bound verification is available. Complexity scoring: - Token estimate (high tokens -> opus) @@ -17,7 +17,10 @@ from dataclasses import dataclass from datetime import datetime from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +if TYPE_CHECKING: + from .providers import ProviderRegistry from .models import WorkItem, WorkItemPriority @@ -209,7 +212,13 @@ class ModelRouter: } # Keywords for complexity classification - _COMPLEX_KEYWORDS = {"refactor entire", "redesign", "migrate", "rewrite", "overhaul"} + _COMPLEX_KEYWORDS = { + "refactor entire", + "redesign", + "migrate", + "rewrite", + "overhaul", + } _SIMPLE_KEYWORDS = {"fix typo", "rename", "update version", "bump", "formatting"} def __init__(self, outcomes_path: Optional[Path] = None) -> None: @@ -219,35 +228,16 @@ def __init__(self, outcomes_path: Optional[Path] = None) -> None: self._build_outcome_stats() def _load_outcomes(self) -> List[_OutcomeRecord]: - """Load historical outcome records from JSONL.""" - if not self._outcomes_path.exists(): - return [] - - records: List[_OutcomeRecord] = [] - try: - for line in self._outcomes_path.read_text(encoding="utf-8").splitlines(): - line = line.strip() - if not line: - continue - try: - data = json.loads(line) - records.append( - _OutcomeRecord( - work_item_id=data.get("work_item_id", ""), - model_tier=data.get("model_tier", ""), - task_type=data.get("task_type", ""), - success=data.get("success", False), - quality_score=data.get("quality_score") or 0.0, - timestamp=data.get("timestamp", ""), - ) - ) - except json.JSONDecodeError: - continue - except OSError as exc: - log.error("Failed to read outcomes file: %s", exc) + """Keep legacy execution logs intact, but do not learn from them. - log.info("Loaded %d historical outcomes", len(records)) - return records + No independent task verifier currently writes this store. Neither a + successful API response nor a heuristic quality score is task evidence. + Verified-outcome ingestion needs an explicit version/attempt-bound + contract before adaptive routing can safely resume. + """ + if self._outcomes_path.exists(): + log.info("Adaptive outcome routing paused: execution history is unverified") + return [] def select_model(self, work_item: WorkItem) -> ModelSelection: """Select the best model tier for the given work item. @@ -379,53 +369,32 @@ def record_outcome( work_item_id: str, model_tier: str, success: bool, - quality_score: float, + quality_score: float | None, task_type: str = "", ) -> None: """Append an outcome record to the JSONL file. - This data feeds back into future routing decisions via - ``_get_historical_performance``. + This preserves execution telemetry only. It is deliberately excluded + from adaptive routing until independent verification is available. """ record = { "work_item_id": work_item_id, "model_tier": model_tier, "task_type": task_type, "success": success, - "quality_score": quality_score, + "quality_score": None, + "heuristic_quality_score": quality_score, "timestamp": datetime.now().isoformat(), + "verification_status": "unverified", + "learning_eligible": False, + "outcome_kind": "execution", } self._outcomes_path.parent.mkdir(parents=True, exist_ok=True) with self._outcomes_path.open("a", encoding="utf-8") as f: f.write(json.dumps(record) + "\n") - # Update in-memory cache - self._outcomes.append( - _OutcomeRecord( - work_item_id=work_item_id, - model_tier=model_tier, - task_type=task_type, - success=success, - quality_score=quality_score, - timestamp=record["timestamp"], - ) - ) - log.info( - "Recorded outcome for %s: %s %s (quality=%.2f)", - work_item_id, - model_tier, - "SUCCESS" if success else "FAILURE", - quality_score, - ) - - # Update outcome stats cache - stats = self._outcome_stats.setdefault(task_type, {}).setdefault( - model_tier, {"success": 0, "total": 0} - ) - stats["total"] += 1 - if success: - stats["success"] += 1 + log.info("Recorded unverified execution for %s (%s)", work_item_id, model_tier) # ------------------------------------------------------------------ # Phase 2: route() API with task-type mapping + complexity overrides @@ -582,7 +551,7 @@ def update_from_outcome(self, task_type: str, model: str, success: bool) -> None work_item_id="", model_tier=model, success=success, - quality_score=1.0 if success else 0.0, + quality_score=None, task_type=task_type, ) diff --git a/tests/network_guard.py b/tests/network_guard.py new file mode 100644 index 0000000..2f068bb --- /dev/null +++ b/tests/network_guard.py @@ -0,0 +1,113 @@ +"""Opt-in pytest plugin: refuse external Python socket connections in CI. + +Load with ``python -m pytest -p tests.network_guard``. Loopback test servers +and Unix sockets remain available. This is not a subprocess/OS network sandbox. +""" + +import ipaddress +import importlib +import socket +from urllib.parse import urlsplit + +import pytest + + +def _check_host(host): + if host in (None, "", "localhost", b"localhost"): + return + if isinstance(host, bytes): + host = host.decode("ascii") + try: + if ipaddress.ip_address(host).is_loopback: + return + except ValueError: + pass + raise RuntimeError("External network disabled by tests.network_guard") + + +def pytest_configure(config): + patcher = pytest.MonkeyPatch() + original_connect = socket.socket.connect + original_connect_ex = socket.socket.connect_ex + original_getaddrinfo = socket.getaddrinfo + original_sendto = socket.socket.sendto + + def connect(sock, address): + if sock.family != socket.AF_UNIX: + _check_host(address[0]) + return original_connect(sock, address) + + def connect_ex(sock, address): + if sock.family != socket.AF_UNIX: + _check_host(address[0]) + return original_connect_ex(sock, address) + + def getaddrinfo(host, *args, **kwargs): + _check_host(host) + return original_getaddrinfo(host, *args, **kwargs) + + def sendto(sock, data, *args): + if sock.family != socket.AF_UNIX: + _check_host(args[-1][0]) + return original_sendto(sock, data, *args) + + patcher.setattr(socket.socket, "connect", connect) + patcher.setattr(socket.socket, "connect_ex", connect_ex) + patcher.setattr(socket, "getaddrinfo", getaddrinfo) + patcher.setattr(socket.socket, "sendto", sendto) + # Validate the intended HTTP destination too: a loopback proxy must not + # turn the loopback exception into access to a live external provider. + for module_name in ("httpx", "httpx2"): + try: + module = importlib.import_module(module_name) + except ImportError: + continue + + def in_memory(client): + transport = getattr(client, "_transport", None) + return type(transport).__name__ in ( + "MockTransport", + "ASGITransport", + "WSGITransport", + "_TestClientTransport", + ) and type(transport).__module__ in ( + "httpx", + "httpx2", + "starlette.testclient", + "httpx._transports.mock", + "httpx._transports.asgi", + "httpx._transports.wsgi", + "httpx2._transports.mock", + "httpx2._transports.asgi", + "httpx2._transports.wsgi", + ) + + def wrap_send(original): + def send(client, request, *args, **kwargs): + if not in_memory(client): + _check_host(request.url.host) + return original(client, request, *args, **kwargs) + + return send + + def wrap_async_send(original): + async def send(client, request, *args, **kwargs): + if not in_memory(client): + _check_host(request.url.host) + return await original(client, request, *args, **kwargs) + + return send + + patcher.setattr(module.Client, "send", wrap_send(module.Client.send)) + patcher.setattr(module.AsyncClient, "send", wrap_async_send(module.AsyncClient.send)) + + import requests + + original_requests_send = requests.Session.send + + def requests_send(session, request, *args, **kwargs): + _check_host(urlsplit(request.url).hostname) + return original_requests_send(session, request, *args, **kwargs) + + patcher.setattr(requests.Session, "send", requests_send) + config.add_cleanup(patcher.undo) diff --git a/tests/test_dispatch_integrity.py b/tests/test_dispatch_integrity.py new file mode 100644 index 0000000..e3e1f44 --- /dev/null +++ b/tests/test_dispatch_integrity.py @@ -0,0 +1,195 @@ +"""Exercise the real dispatcher/collector/router boundary with synthetic providers.""" + +import json +from types import SimpleNamespace as NS +from unittest.mock import Mock + +import pytest + +from cortex.supervisor.collector import ResultCollector, VerificationEvidence +from cortex.supervisor.dispatch import AgentDispatcher, DispatchResult, ModelSelection +from cortex.supervisor.models import WorkItem +from cortex.supervisor.router import ModelRouter + + +def response(reason="end_turn", text="analysis"): + return NS( + stop_reason=reason, + content=[NS(type="text", text=text)], + usage=NS(input_tokens=7, output_tokens=3), + ) + + +def item(): + return WorkItem(id="same-task", source="test", task_type="review", description="Review code") + + +def selection(): + return ModelSelection("sonnet", "claude-sonnet-4-6", "test", 0.5, 0.5) + + +@pytest.mark.parametrize( + "reason,text", + [ + ("max_tokens", "partial"), + ("refusal", "cannot"), + (None, "unknown"), + ("end_turn", " "), + ("tool_use", "no tools provided"), + ], +) +def test_incomplete_provider_response_is_not_success(monkeypatch, reason, text): + dispatcher = AgentDispatcher() + client = NS(messages=NS(create=Mock(return_value=response(reason, text)))) + monkeypatch.setattr(dispatcher, "_get_client", lambda: client) + result = dispatcher.dispatch(item(), selection()) + assert result.success is False + assert result.execution_status == "incomplete" + assert result.output == text + assert result.tokens_used == 10 + + +def test_execution_stays_unverified_across_persistence_and_router_restart(tmp_path, monkeypatch): + dispatcher = AgentDispatcher() + client = NS(messages=NS(create=Mock(return_value=response()))) + monkeypatch.setattr(dispatcher, "_get_client", lambda: client) + result = dispatcher.dispatch(item(), selection()) + outcomes = tmp_path / "outcomes.jsonl" + collector = ResultCollector(tmp_path / "runs", outcomes) + collector.collect(result) + # Even a perfect heuristic is not an independent check of an artifact. + assessment = VerificationEvidence(10, True, 4, True, True) + collector.attach_evidence(result.work_item_id, assessment, attempt_id=result.attempt_id) + collector.record_outcome(result, quality_score=1.0) + run = collector.persist() + summary = json.loads((run / "summary.json").read_text()) + receipt = json.loads(outcomes.read_text()) + assert summary["succeeded"] == 1 # execution only + assert summary["unverified"] == summary["flagged_for_review"] == 1 + assert summary["auto_rejected"] == 0 + assert receipt["attempt_id"] == result.attempt_id + assert receipt["verification_status"] == "unverified" + assert receipt["learning_eligible"] is False + assert receipt["quality_score"] is None + assert receipt["heuristic_quality_score"] == 1.0 + assert receipt["assessment"]["score"] == 100 + assert ModelRouter(outcomes)._outcomes == [] + + +def test_missing_evidence_is_flagged(tmp_path): + result = DispatchResult("task", True, "done", "sonnet", 0, 0) + collector = ResultCollector(tmp_path / "runs", tmp_path / "outcomes") + collector.collect(result) + assert collector.get_summary().flagged_for_review == 1 + + +def test_attempts_do_not_share_assessments_and_runs_do_not_overwrite(tmp_path): + collector = ResultCollector(tmp_path / "runs", tmp_path / "outcomes") + first = DispatchResult("task", True, "first", "sonnet", 0, 0) + second = DispatchResult("task", True, "second", "sonnet", 0, 0) + collector.collect_batch([first, second]) + assert first.attempt_id != second.attempt_id + with pytest.raises(ValueError, match="attempt_id"): + collector.attach_evidence("task", VerificationEvidence()) + collector.attach_evidence("task", VerificationEvidence(), attempt_id=first.attempt_id) + a, b = collector.persist(), collector.persist() + assert a != b and a.exists() and b.exists() + rows = json.loads((a / "results.json").read_text()) + assert "assessment" in rows[0] and "assessment" not in rows[1] + + +@pytest.mark.parametrize( + "tool,extra", + [ + ("read_file", {}), + ("search_code", {"pattern": "secret"}), + ("list_files", {"pattern": "*"}), + ], +) +def test_sibling_prefix_and_symlink_escape_are_rejected(tmp_path, tool, extra): + root = tmp_path / "project" + root.mkdir() + outside = tmp_path / "project-private" + outside.mkdir() + (outside / "secret.py").write_text("secret synthetic fixture") + (root / "escape").symlink_to(outside, target_is_directory=True) + for path in ("../project-private", "escape"): + if tool == "read_file": + path += "/secret.py" + output = AgentDispatcher()._execute_tool(tool, {"path": path, **extra}, str(root)) + assert output.startswith("Error: access denied") + + +def test_dry_run_does_not_count_as_execution_success_or_failure(tmp_path): + result = AgentDispatcher(dry_run=True).dispatch(item(), selection()) + collector = ResultCollector(tmp_path / "runs", tmp_path / "outcomes") + collector.collect(result) + summary = collector.get_summary() + assert (summary.total, summary.succeeded, summary.failed, summary.dry_runs) == ( + 1, + 0, + 0, + 1, + ) + + +def test_batch_transport_success_is_not_complete_or_verified(tmp_path, monkeypatch): + dispatcher = AgentDispatcher() + entry = NS(custom_id="task", result=NS(type="succeeded", message=response("max_tokens"))) + batches = NS(retrieve=lambda _: NS(processing_status="ended"), results=lambda _: [entry]) + monkeypatch.setattr(dispatcher, "_get_client", lambda: NS(messages=NS(batches=batches))) + monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path) + router = ModelRouter(tmp_path / "outcomes.jsonl") + first = dispatcher.retrieve_batch("batch", router=router)[0] + second = dispatcher.retrieve_batch("batch")[0] + assert first.success is False + assert first.attempt_id == second.attempt_id == "batch:task" + receipt = json.loads((tmp_path / "outcomes.jsonl").read_text()) + assert receipt["quality_score"] is None + assert receipt["learning_eligible"] is False + assert router._outcomes == [] + + +def test_command_rejected_before_provider_call(monkeypatch): + dispatcher = AgentDispatcher() + client = Mock() + monkeypatch.setattr(dispatcher, "_get_client", client) + work = item() + work.command = "touch should-not-exist" + result = dispatcher.dispatch(work, selection()) + assert result.success is False + assert result.execution_status == "unsupported" + client.assert_not_called() + + +def test_repeated_collection_is_idempotent(tmp_path): + result = DispatchResult("task", True, "done", "sonnet", 0, 0) + collector = ResultCollector(tmp_path / "runs", tmp_path / "outcomes") + collector.collect(result) + collector.collect_batch([result]) + assert collector.get_summary().total == 1 + + +def test_registry_uses_explicit_state_directory(tmp_path, monkeypatch): + from store_registry import registry + + monkeypatch.setenv("CORTEX_HOME", str(tmp_path / "wrong")) + monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path / "right")) + paths = {s.name: s.path for s in registry()} + assert paths["decisions"] == tmp_path / "right" / "decisions.jsonl" + assert paths["outcomes"] == tmp_path / "right" / "outcomes.jsonl" + + +def test_batch_rejects_dry_run_and_duplicate_ids_before_provider_call(monkeypatch): + from cortex.supervisor.router import ModelAssignment + + assignment = ModelAssignment(model="sonnet", confidence=0.5, rationale="test") + for dispatcher, work_items, message in ( + (AgentDispatcher(dry_run=True), [(item(), assignment)], "Dry-run"), + (AgentDispatcher(), [(item(), assignment), (item(), assignment)], "unique"), + ): + client = Mock() + monkeypatch.setattr(dispatcher, "_get_client", client) + with pytest.raises(ValueError, match=message): + dispatcher.submit_batch(work_items) + client.assert_not_called() diff --git a/tests/test_doctor_json_reset.py b/tests/test_doctor_json_reset.py index 9b93b9a..89c5d8e 100644 --- a/tests/test_doctor_json_reset.py +++ b/tests/test_doctor_json_reset.py @@ -22,6 +22,19 @@ def state_dir(tmp_path, monkeypatch): monkeypatch.setenv("CORTEX_STATE_DIR", str(tmp_path)) monkeypatch.setenv("CORTEX_CLAUDE_BIN", "") # force claude-not-installed path + # Healthy means the declared stores exist, not merely an empty directory. + # Isolate the legacy portfolio store too; do not inspect the operator's home. + from pathlib import Path + import pickle + from store_registry import registry + + monkeypatch.setattr(Path, "home", lambda: tmp_path / "home") + for store in registry(): + store.path.parent.mkdir(parents=True, exist_ok=True) + if store.path.suffix == ".pkl": + store.path.write_bytes(pickle.dumps({})) + else: + store.path.write_text("{}" if store.path.suffix == ".json" else "") return tmp_path @@ -53,12 +66,23 @@ def test_doctor_exit_nonzero_on_failure(state_dir): def test_doctor_exit_zero_when_healthy(state_dir): - # No seeded failure; golden path (no key) must NOT fail doctor. + # Existing fresh stores, no seeded failure; no key remains advisory. with pytest.raises(SystemExit) as exc: cmd_doctor(SimpleNamespace(fix=False, json=True)) assert exc.value.code == 0 +def test_doctor_missing_store_still_fails(state_dir, capsys): + (state_dir / "decisions.jsonl").unlink() + with pytest.raises(SystemExit) as exc: + cmd_doctor(SimpleNamespace(fix=False, json=True)) + payload = json.loads(capsys.readouterr().out) + check = next(c for c in payload["checks"] if c["check"] == "store fresh: decisions") + assert exc.value.code == 1 + assert check["status"] == "fail" + assert str(state_dir / "decisions.jsonl") in check["detail"] + + def test_doctor_fix_repairs_broken_spool_and_exits_zero(state_dir, capsys): _spool(state_dir) with pytest.raises(SystemExit) as exc: diff --git a/tests/test_mcp_research_tools.py b/tests/test_mcp_research_tools.py index 29120dc..278eaf7 100644 --- a/tests/test_mcp_research_tools.py +++ b/tests/test_mcp_research_tools.py @@ -222,6 +222,17 @@ def _skip_if_no_mcp(self): """Skip if MCP SDK is not installed.""" pytest.importorskip("mcp") + @staticmethod + def _registered_names(server): + import asyncio + from fastmcp import Client + + async def discover(): + async with Client(server) as client: + return {tool.name for tool in await client.list_tools()} + + return asyncio.run(discover()) + _GOLDEN_TOOLS = { "cortex_intelligence", "cortex_record_decision", @@ -252,7 +263,7 @@ def test_golden_tools_always_loaded(self): """The golden five are registered at import — no enable step.""" from cortex.mcp_server import mcp as mcp_instance - registered = set(mcp_instance._tool_manager._tools.keys()) + registered = self._registered_names(mcp_instance) missing = self._GOLDEN_TOOLS - registered assert missing == set(), f"Golden-five tools not registered: {missing}" @@ -264,7 +275,7 @@ def test_only_golden_five_visible_by_default(self): pytest.skip("suite running with CORTEX_EXPERIMENTAL set") from cortex.mcp_server import mcp as mcp_instance - registered = set(mcp_instance._tool_manager._tools.keys()) + registered = self._registered_names(mcp_instance) assert registered == self._GOLDEN_TOOLS, ( f"default MCP surface must be exactly the golden five. " f"unexpected: {sorted(registered - self._GOLDEN_TOOLS)}; " @@ -279,7 +290,7 @@ def test_experimental_tools_gated_by_default(self): pytest.skip("suite running with CORTEX_EXPERIMENTAL set") from cortex.mcp_server import mcp as mcp_instance - registered = set(mcp_instance._tool_manager._tools.keys()) + registered = self._registered_names(mcp_instance) leaked = self._EXPERIMENTAL_TOOLS & registered assert leaked == set(), f"Experimental tools registered by default: {leaked}" @@ -333,7 +344,7 @@ def test_no_enable_tools_tool(self): """cortex_enable_tools no longer exists (deferred loading removed).""" from cortex.mcp_server import mcp as mcp_instance - registered = set(mcp_instance._tool_manager._tools.keys()) + registered = self._registered_names(mcp_instance) assert "cortex_enable_tools" not in registered, "cortex_enable_tools should be removed" diff --git a/tests/test_model_router.py b/tests/test_model_router.py index 2f63fcf..8263f27 100644 --- a/tests/test_model_router.py +++ b/tests/test_model_router.py @@ -10,6 +10,12 @@ import pytest +from cortex.supervisor.router import ( + ModelAssignment, + ModelRouter as ActualModelRouter, +) + + from cortex.supervisor.models import WorkItem, WorkItemPriority @@ -154,10 +160,30 @@ def router() -> ModelRouter: def router_with_outcomes(tmp_path: Path) -> ModelRouter: outcomes_file = tmp_path / "outcomes.jsonl" records = [ - {"work_item_id": "w1", "model_tier": "opus", "success": True, "tokens_used": 8000}, - {"work_item_id": "w2", "model_tier": "opus", "success": True, "tokens_used": 6000}, - {"work_item_id": "w3", "model_tier": "sonnet", "success": False, "tokens_used": 3000}, - {"work_item_id": "w4", "model_tier": "haiku", "success": True, "tokens_used": 500}, + { + "work_item_id": "w1", + "model_tier": "opus", + "success": True, + "tokens_used": 8000, + }, + { + "work_item_id": "w2", + "model_tier": "opus", + "success": True, + "tokens_used": 6000, + }, + { + "work_item_id": "w3", + "model_tier": "sonnet", + "success": False, + "tokens_used": 3000, + }, + { + "work_item_id": "w4", + "model_tier": "haiku", + "success": True, + "tokens_used": 500, + }, ] outcomes_file.write_text("\n".join(json.dumps(r) for r in records) + "\n") return ModelRouter(outcomes_path=outcomes_file) @@ -197,7 +223,10 @@ def test_complexity_score_ranges(self, router: ModelRouter) -> None: files=30, ), _make_item( - task_type="review", priority=WorkItemPriority.MEDIUM, tokens=10_000, files=5 + task_type="review", + priority=WorkItemPriority.MEDIUM, + tokens=10_000, + files=5, ), ] for item in items: @@ -341,11 +370,6 @@ def test_historical_performance_with_data( # Phase 2: Tests for route() API with TASK_MODEL_MAP + complexity overrides # --------------------------------------------------------------------------- -from cortex.supervisor.router import ( - ModelAssignment, - ModelRouter as ActualModelRouter, -) - def _make_phase2_item( task_type: str = "research", @@ -472,23 +496,23 @@ def test_moderate_complexity_uses_task_map(self, phase2_router: ActualModelRoute assert result.model == "sonnet" assert "Complexity override" not in result.rationale - def test_outcome_adjustment_escalates( + def test_legacy_failures_do_not_escalate( self, phase2_router_with_outcomes: ActualModelRouter ) -> None: - """When sonnet has <40% success on 'implement', escalate to opus.""" + """Legacy execution failures cannot calibrate task success.""" item = _make_phase2_item(task_type="implement") result = phase2_router_with_outcomes.route(item) - assert result.model == "opus" - assert "Escalated" in result.rationale + assert result.model == "sonnet" + assert "Escalated" not in result.rationale - def test_outcome_adjustment_downgrades( + def test_legacy_successes_do_not_downgrade( self, phase2_router_with_outcomes: ActualModelRouter ) -> None: - """When opus has >90% success on 'planning', downgrade to sonnet.""" + """Legacy execution successes cannot certify cheaper routing.""" item = _make_phase2_item(task_type="planning") result = phase2_router_with_outcomes.route(item) - assert result.model == "sonnet" - assert "Downgraded" in result.rationale + assert result.model == "opus" + assert "Downgraded" not in result.rationale def test_missing_outcomes_file_graceful(self, tmp_path: Path) -> None: """Router should not crash when outcomes file doesn't exist.""" @@ -512,10 +536,10 @@ def test_update_from_outcome_records(self, tmp_path: Path) -> None: lines = outcomes_file.read_text().strip().splitlines() assert len(lines) == 5 - # Verify in-memory stats updated - stats = router._outcome_stats.get("test", {}).get("sonnet", {}) - assert stats["total"] == 5 - assert stats["success"] == 5 + # Execution receipts persist without changing routing calibration. + assert router._outcome_stats == {} + assert all(json.loads(line)["learning_eligible"] is False for line in lines) + assert ActualModelRouter(outcomes_path=outcomes_file)._outcomes == [] def test_route_returns_model_assignment(self, phase2_router: ActualModelRouter) -> None: """route() must return a ModelAssignment with correct fields.""" diff --git a/tests/test_orchestration_pipeline.py b/tests/test_orchestration_pipeline.py index 9c6761e..ec0f22c 100644 --- a/tests/test_orchestration_pipeline.py +++ b/tests/test_orchestration_pipeline.py @@ -744,13 +744,19 @@ def _make_text_response(self, text, input_tokens=100, output_tokens=50): text_block.type = "text" text_block.text = text mock_resp.content = [text_block] + mock_resp.stop_reason = "end_turn" mock_resp.usage = MagicMock() mock_resp.usage.input_tokens = input_tokens mock_resp.usage.output_tokens = output_tokens return mock_resp def _make_tool_use_response( - self, tool_name, tool_input, tool_id="tool_1", input_tokens=100, output_tokens=50 + self, + tool_name, + tool_input, + tool_id="tool_1", + input_tokens=100, + output_tokens=50, ): """Create a mock response with a tool_use block.""" mock_resp = MagicMock() @@ -760,6 +766,7 @@ def _make_tool_use_response( tool_block.input = tool_input tool_block.id = tool_id mock_resp.content = [tool_block] + mock_resp.stop_reason = "tool_use" mock_resp.usage = MagicMock() mock_resp.usage.input_tokens = input_tokens mock_resp.usage.output_tokens = output_tokens @@ -824,7 +831,9 @@ def test_dispatch_max_turns_exceeded(self, dispatcher, work_item, model_selectio result = dispatcher.dispatch(work_item, model_selection) - assert result.success is True + assert result.success is False + assert result.execution_status == "incomplete" + assert result.tokens_used == 1500 # Should have stopped at max_turns (10) assert mock_client.messages.create.call_count == 10 assert "max turns" in result.output.lower() or result.output != "" @@ -1218,7 +1227,8 @@ def test_dry_run_does_not_call_api(self): # Client should never have been created mock_client_fn.assert_not_called() - assert result.success is True + assert result.success is False + assert result.execution_status == "dry_run" assert "[DRY RUN]" in result.output assert result.tokens_used == 0 @@ -1480,6 +1490,7 @@ def _make_result(cid, succeeded=True): text_block.type = "text" text_block.text = f"Result for {cid}" r.result.message.content = [text_block] + r.result.message.stop_reason = "end_turn" r.result.message.usage.input_tokens = 100 r.result.message.usage.output_tokens = 200 else: @@ -1555,6 +1566,7 @@ def test_retrieve_batch_records_outcomes(self, tmp_path): tb.type = "text" tb.text = "done" r.result.message.content = [tb] + r.result.message.stop_reason = "end_turn" r.result.message.usage.input_tokens = 50 r.result.message.usage.output_tokens = 100 mock_client.messages.batches.results.return_value = [r]