diff --git a/apps/web/vercel.json b/apps/web/vercel.json new file mode 100644 index 000000000..d277b8ca5 --- /dev/null +++ b/apps/web/vercel.json @@ -0,0 +1,7 @@ +{ + "git": { + "deploymentEnabled": { + "fix/2168-mcp-routed-false-success": false + } + } +} diff --git a/src/agents/mcp_agent_network.py b/src/agents/mcp_agent_network.py index c29f074d6..147cc6b94 100644 --- a/src/agents/mcp_agent_network.py +++ b/src/agents/mcp_agent_network.py @@ -14,6 +14,22 @@ logger = logging.getLogger(__name__) +# Status for a route the network knows about but cannot execute. It always +# carries an "error" key so every caller that checks `"error" in result` +# (e.g. VideoPipelineOrchestrator._execute_agent_stage) treats it as failure. +NOT_IMPLEMENTED = "not_implemented" + + +def not_implemented_result(route: str, action: str, reason: str) -> dict[str, Any]: + """Explicit failure for a tool route that was not executed.""" + return { + "status": NOT_IMPLEMENTED, + "executed": False, + "route": route, + "action": action, + "error": f"{route} {action}: not implemented ({reason}); nothing was executed", + } + # Agent Network Configuration AGENT_NETWORK_CONFIG = Path(__file__).parent.parent.parent / "config" / "agent_network.json" @@ -204,12 +220,22 @@ async def _call_internal(self, endpoint: str, action: str, payload: dict) -> dic return await self._call_knowledge_base(action, payload) except Exception as e: return {"error": f"Internal tool error: {str(e)}"} + return not_implemented_result(endpoint, action, f"no internal module handler for '{module_name}'") async def _call_mcp(self, endpoint: str, action: str, payload: dict) -> dict: """Call MCP server via stdio/protocol""" server_name = endpoint.replace("mcp://", "") - logger.info(f"MCP call: {server_name}.{action}") - return {"status": "routed", "server": server_name, "action": action} + # No MCP client/transport is wired into this network. Previously this + # returned {"status": "routed"} without doing anything, so the + # pipeline's deploy step (create_repo via mcp://github) reported + # success without deploying. Report the truth instead. + logger.error( + "MCP route not implemented: %s.%s was not executed", + server_name, + action, + extra={"route": endpoint, "action": action, "status": NOT_IMPLEMENTED}, + ) + return not_implemented_result(endpoint, action, "no MCP client transport is configured") async def _call_http(self, endpoint: str, action: str, payload: dict) -> dict: """Call HTTP endpoint""" @@ -217,6 +243,7 @@ async def _call_http(self, endpoint: str, action: str, payload: dict) -> dict: if endpoint.startswith("https://api.vercel.com"): return await self._call_vercel_api(action, payload) + return not_implemented_result(endpoint, action, "no HTTP handler for this endpoint") def get_pipeline_agents(self) -> list[str]: """Get ordered list of agents for video-to-software pipeline""" @@ -242,7 +269,7 @@ async def _call_skill_builder(self, action: str, payload: dict) -> dict: elif action == "suggest_fix": return await build_validator.suggest_fix(**payload) else: - return {"status": "pending_implementation", "action": action} + return not_implemented_result("internal://skill_builder", action, "no skill builder handler") except Exception as e: logger.error(f"Build validator error: {e}") @@ -302,7 +329,7 @@ async def _call_knowledge_base(self, action: str, payload: dict) -> dict: return {"status": "success", "stats": stats} else: - return {"status": "unknown_action", "action": action} + return not_implemented_result("internal://knowledge_base", action, "no knowledge base handler") except Exception as e: logger.error(f"Knowledge base error: {e}") @@ -326,8 +353,7 @@ async def _call_vercel_api(self, action: str, payload: dict) -> dict: return await deployment_tool.deploy_to_github_and_vercel(**payload) else: - # TODO: Implement other deployment features - return {"status": "pending_implementation", "action": action} + return not_implemented_result("https://api.vercel.com", action, "no deployment handler") except Exception as e: logger.error(f"Deployment error: {e}") diff --git a/src/agents/pipeline_orchestrator.py b/src/agents/pipeline_orchestrator.py index a40a07be4..79272f411 100644 --- a/src/agents/pipeline_orchestrator.py +++ b/src/agents/pipeline_orchestrator.py @@ -59,8 +59,34 @@ def filter(self, record): # --- VERA imports (lazy-loaded to avoid import-time side effects) --- +class VeraInitError(RuntimeError): + """VERA modules are installed but failed to initialize.""" + + +# Stage results with one of these statuses are failures even without an +# "error" key (e.g. DeploymentMCPTool returns status "failed"). +FAILED_STATUSES = frozenset({"failed", "error", "not_implemented"}) + + +def _stage_failure_reason(result: Any) -> str | None: + """Return why a routed stage result is a failure, or None if it succeeded.""" + if not isinstance(result, dict): + return f"stage returned {type(result).__name__}, not a result dict" + if "error" in result: + return str(result["error"]) + status = result.get("status") + if status in FAILED_STATUSES: + return f"stage reported status '{status}'" + return None + + def _get_vera(): - """Lazy-load VERA components. Returns None for each component if VERA is unavailable.""" + """Lazy-load VERA components. + + Returns None when the VERA modules are not installed. Raises VeraInitError + when they are installed but fail to initialize, so the orchestrator can + fail closed instead of silently running without its security layer. + """ try: from vera.identity import get_identity_service from vera.firewall import get_firewall @@ -81,12 +107,17 @@ def _get_vera(): "maturity": get_maturity_runtime(), "enforcer": get_enforcer(), } - except ImportError: - logger.info("VERA modules not available in this env (hardened local maturity still applied for pipeline agents)") - return None except Exception as e: - logger.error(f"VERA initialization failed: {e} — running without security layer") - return None + if isinstance(e, ModuleNotFoundError) and e.name == "vera": + logger.info("VERA modules not available in this env (hardened local maturity still applied for pipeline agents)") + return None + logger.error( + "VERA initialization failed (%s): %s — all stages will be blocked", + type(e).__name__, + e, + exc_info=True, + ) + raise VeraInitError(f"{type(e).__name__}: {e}") from e # Local hardened maturity for pipeline agents (always-on fallback / hardening when full VERA unavailable) # This ensures code-gen, deployer etc. are treated as level 1+ for security intent. @@ -176,7 +207,13 @@ def __init__(self): self.emitter = get_emitter() # VERA security layer (None if VERA modules unavailable) - self._vera = _get_vera() + self._vera_init_error: str | None = None + try: + self._vera = _get_vera() + except VeraInitError as exc: + # Fail closed: VERA is present but broken, so no stage may run. + self._vera = None + self._vera_init_error = str(exc) self._vera_credentials: dict[str, Any] = {} # agent_id → (token, credential) if self._vera: logger.info("VERA security layer active — all agent stages will be verified") @@ -419,6 +456,15 @@ async def _execute_agent_stage(self, agent_id: str) -> PipelineResult: action, payload = self._prepare_agent_action(agent_id) # --- VERA pre-execution checks --- + vera_init_error = getattr(self, "_vera_init_error", None) + if vera_init_error: + blocked = PipelineResult( + agent_id, False, {}, (time.monotonic() - start_time) * 1000, + f"VERA: security layer failed to initialize ({vera_init_error}) — stage blocked (fail-closed)", + ) + self._record_audit_stage(agent_id, action, blocked) + return blocked + if self._vera: vera_check = self._vera_pre_check(agent_id, agent.name, action, payload) if vera_check is not None: @@ -442,20 +488,25 @@ async def _execute_agent_stage(self, agent_id: str) -> PipelineResult: duration = (time.monotonic() - start_time) * 1000 - if "error" in result: + failure_reason = _stage_failure_reason(result) + if failure_reason is not None: # --- VERA: record failure proof + notify enforcer --- if self._vera: + proof_result = dict(result) if isinstance(result, dict) else {} + proof_result.setdefault("error", failure_reason) self._vera_record_proof( - agent_id, action, payload, result, duration, + agent_id, action, payload, proof_result, duration, authorized=True, success=False, ) self._vera["enforcer"].on_enforcement_event( agent_id, "stage_error", - details=f"Stage {agent_id} returned error: {result.get('error', '')}", + details=f"Stage {agent_id} returned error: {failure_reason}", ) self._vera["breakers"].get_breaker(agent_id).record_failure("error") - failed = PipelineResult(agent_id, False, result, duration, result["error"]) + failed = PipelineResult( + agent_id, False, result if isinstance(result, dict) else {}, duration, failure_reason + ) self._record_audit_stage(agent_id, action, failed) return failed @@ -534,8 +585,12 @@ def _vera_pre_check( maturity_level=initial_level, ) self._vera_credentials[agent_id] = (token, credential) - except Exception: - self._vera_credentials[agent_id] = (None, None) + except Exception as exc: # exception-boundary: identity check failed; fail closed (block the stage) + logger.error( + "VERA credential issuance failed for %s (%s) — stage blocked", + agent_id, type(exc).__name__, exc_info=True, + ) + return f"VERA: credential issuance failed ({type(exc).__name__}) — stage blocked (fail-closed)" else: self._vera_credentials[agent_id] = (None, None) @@ -548,8 +603,12 @@ def _vera_pre_check( verified = v["identity"].verify_credential(token) if token else None if verified is None and not (v["identity"].is_revoked(agent_id) if hasattr(v["identity"], "is_revoked") else False): maturity_level = _get_effective_maturity(agent_id) - except Exception: - pass + except Exception as exc: # exception-boundary: identity check failed; fail closed (block the stage) + logger.error( + "VERA credential verification failed for %s (%s) — stage blocked", + agent_id, type(exc).__name__, exc_info=True, + ) + return f"VERA: credential verification failed ({type(exc).__name__}) — stage blocked (fail-closed)" # 3. Gateway permission check (hardened: use effective level even without full VERA) decision = None @@ -568,12 +627,15 @@ def _vera_pre_check( ) breaker.record_failure("auth_failure") return f"VERA: Permission denied — {decision.reason}" - except Exception as exc: - logger.warning( - "VERA gateway check failed (non-blocking) for %s: %s", + except Exception as exc: # exception-boundary: gateway check failed; fail closed (block the stage) + logger.error( + "VERA gateway check failed for %s (%s): %s — stage blocked", agent_id, + type(exc).__name__, exc, + exc_info=True, ) + return f"VERA: gateway check failed ({type(exc).__name__}) — stage blocked (fail-closed)" else: # No full VERA: log once per agent but allow (hardened local maturity already applied above) if not hasattr(self, "_vera_harden_logged"): @@ -605,8 +667,15 @@ def _vera_pre_check( details=f"Blocked input for {action}: {scan.details}", ) return f"VERA: Input blocked by firewall — {scan.threat_level.value} threat" - except Exception as e: - logger.warning(f"VERA firewall scan error (non-blocking): {e}") + except Exception as e: # exception-boundary: firewall scan failed; fail closed (block the stage) + logger.error( + "VERA firewall scan failed for %s (%s): %s — stage blocked", + agent_id, + type(e).__name__, + e, + exc_info=True, + ) + return f"VERA: firewall scan failed ({type(e).__name__}) — stage blocked (fail-closed)" return None # All checks passed diff --git a/tests/unit/test_mcp_routed_false_success_2168.py b/tests/unit/test_mcp_routed_false_success_2168.py new file mode 100644 index 000000000..4930c6def --- /dev/null +++ b/tests/unit/test_mcp_routed_false_success_2168.py @@ -0,0 +1,292 @@ +"""Issue #2168: the agent network and pipeline must not report success they did not achieve. + +- An MCP route with no transport used to return {"status": "routed"} without + executing anything, so the pipeline's deploy step (create_repo via + mcp://github) reported success without deploying. +- VERA gateway / firewall / identity exceptions used to be logged as + "non-blocking" and the stage ran anyway (fail-open). + +No subprocess, network or database is touched: VERA is faked and the network +routes either to the real (unimplemented) MCP path or to a stub. +""" + +from __future__ import annotations + +import enum +import sys +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from agents import pipeline_orchestrator as po +from agents.mcp_agent_network import MCPAgentNetwork + +NOT_IMPLEMENTED = "not_implemented" # contract: same status scripts/ci/autonomous_video_processing.py uses + + +class _Emitter: + def __init__(self): + self.events = [] + + async def emit(self, channel, payload): + self.events.append((channel, payload)) + + +def _orchestrator(network=None, vera=None, vera_init_error=None): + orch = object.__new__(po.VideoPipelineOrchestrator) + orch.network = network or MCPAgentNetwork() + orch.pipeline_state = {} + orch.results = {} + orch.emitter = _Emitter() + orch._vera = vera + orch._vera_init_error = vera_init_error + orch._vera_credentials = {} + return orch + + +# --------------------------------------------------------------------------- +# MCP route: not implemented is a failure, never "routed" +# --------------------------------------------------------------------------- + + +async def test_mcp_route_reports_not_implemented_with_error(): + result = await MCPAgentNetwork()._call_mcp("mcp://github", "create_repo", {}) + assert result["status"] == NOT_IMPLEMENTED + assert result["executed"] is False + assert "error" in result + assert result["status"] != "routed" + + +async def test_deployer_create_repo_route_is_not_executed(): + result = await MCPAgentNetwork().route_to_agent("deployer", "create_repo", {}) + assert result["status"] == NOT_IMPLEMENTED + assert "not implemented" in result["error"] + + +@pytest.mark.parametrize( + "call", + [ + lambda n: n._call_http("https://example.invalid", "anything", {}), + lambda n: n._call_internal("internal://nowhere", "anything", {}), + lambda n: n._call_vercel_api("get_deployment_status", {}), + ], + ids=["http-unknown-endpoint", "internal-unknown-module", "vercel-unhandled-action"], +) +async def test_other_unimplemented_routes_fail_explicitly(call): + result = await call(MCPAgentNetwork()) + assert result["status"] == NOT_IMPLEMENTED + assert result["executed"] is False + assert "error" in result + + +async def test_pipeline_deploy_stage_fails_when_route_not_executed(): + orch = _orchestrator() + stage = await orch._execute_agent_stage("deployer") + assert stage.success is False + assert "not implemented" in stage.error + assert stage.data["status"] == NOT_IMPLEMENTED + + +async def test_pipeline_report_is_not_success_when_deploy_not_executed(monkeypatch): + monkeypatch.setattr(po, "get_audit_store", None) # no audit file writes from a unit test + network = MCPAgentNetwork() + real_route = network.route_to_agent + + async def route(agent_id, action, payload): + if agent_id == "deployer": + return await real_route(agent_id, action, payload) # the real, unimplemented MCP path + return {"status": "success"} + + network.route_to_agent = route + orch = _orchestrator(network=network) + report = await orch._run_sequential_pipeline("https://youtu.be/auJzb1D-fag", {"continue_on_error": True}) + assert report["success"] is False + assert report["stage_results"]["deployer"]["success"] is False + assert "not implemented" in report["stage_results"]["deployer"]["error"] + others = {k: v for k, v in report["stage_results"].items() if k != "deployer"} + assert all(v["success"] for v in others.values()) + + +@pytest.mark.parametrize( + "result", + [ + {"status": "failed", "errors": [{"stage": "github", "message": "x"}]}, + {"status": "not_implemented"}, + {"status": "error"}, + None, + ], + ids=["deploy-tool-failed-shape", "not-implemented", "error-status", "non-dict"], +) +async def test_failed_status_without_error_key_is_stage_failure(result): + network = MagicMock(spec=MCPAgentNetwork) + network.get_agent.return_value = SimpleNamespace(name="Deployment Agent", role="deploy") + network.route_to_agent = AsyncMock(return_value=result) + stage = await _orchestrator(network=network)._execute_agent_stage("deployer") + assert stage.success is False + assert stage.error + + +# --------------------------------------------------------------------------- +# VERA: security-check exceptions fail closed +# --------------------------------------------------------------------------- + + +class _FirewallAction(enum.Enum): + ALLOW = "allow" + BLOCK = "block" + + +@pytest.fixture +def fake_vera_firewall_module(monkeypatch): + vera_pkg = types.ModuleType("vera") + firewall_mod = types.ModuleType("vera.firewall") + firewall_mod.FirewallAction = _FirewallAction + vera_pkg.firewall = firewall_mod + monkeypatch.setitem(sys.modules, "vera", vera_pkg) + monkeypatch.setitem(sys.modules, "vera.firewall", firewall_mod) + + +def _fake_vera(): + breaker = MagicMock() + breaker.allow_action.return_value = True + breakers = MagicMock() + breakers.get_breaker.return_value = breaker + identity = MagicMock() + identity.issue_credential.return_value = ("token", SimpleNamespace(token_id="jti", maturity_level=1)) + maturity = MagicMock() + maturity.get_level.return_value = 1 + gateway = MagicMock() + gateway.check_permission.return_value = SimpleNamespace(allowed=True, reason="") + firewall = MagicMock() + firewall.scan_input.return_value = SimpleNamespace(action=_FirewallAction.ALLOW) + return { + "identity": identity, + "firewall": firewall, + "gateway": gateway, + "proof_store": MagicMock(), + "ExecutionProof": MagicMock(), + "hash_data": lambda text: "h", + "breakers": breakers, + "maturity": maturity, + "enforcer": MagicMock(), + } + + +def _stub_network(): + network = MagicMock(spec=MCPAgentNetwork) + network.get_agent.return_value = SimpleNamespace(name="Deployment Agent", role="deploy") + network.route_to_agent = AsyncMock(return_value={"status": "success"}) + return network + + +async def test_vera_all_checks_pass_runs_stage(fake_vera_firewall_module): + """Positive control: with VERA healthy the stage executes.""" + network = _stub_network() + stage = await _orchestrator(network=network, vera=_fake_vera())._execute_agent_stage("deployer") + assert stage.success is True + network.route_to_agent.assert_awaited_once() + + +async def test_vera_gateway_exception_blocks_stage(fake_vera_firewall_module): + vera = _fake_vera() + vera["gateway"].check_permission.side_effect = RuntimeError("policy store unreachable") + network = _stub_network() + stage = await _orchestrator(network=network, vera=vera)._execute_agent_stage("deployer") + assert stage.success is False + assert "gateway check failed" in stage.error and "fail-closed" in stage.error + network.route_to_agent.assert_not_awaited() + + +async def test_vera_firewall_exception_blocks_stage(fake_vera_firewall_module): + vera = _fake_vera() + vera["firewall"].scan_input.side_effect = ValueError("scanner crashed") + network = _stub_network() + stage = await _orchestrator(network=network, vera=vera)._execute_agent_stage("deployer") + assert stage.success is False + assert "firewall scan failed" in stage.error and "fail-closed" in stage.error + network.route_to_agent.assert_not_awaited() + + +async def test_vera_credential_issuance_exception_blocks_stage(fake_vera_firewall_module): + vera = _fake_vera() + vera["identity"].issue_credential.side_effect = OSError("key file missing") + network = _stub_network() + stage = await _orchestrator(network=network, vera=vera)._execute_agent_stage("deployer") + assert stage.success is False + assert "credential issuance failed" in stage.error + network.route_to_agent.assert_not_awaited() + + +async def test_vera_cached_credential_verification_exception_blocks_stage(fake_vera_firewall_module): + vera = _fake_vera() + vera["identity"].verify_credential.side_effect = RuntimeError("verification unavailable") + network = _stub_network() + orch = _orchestrator(network=network, vera=vera) + orch._vera_credentials["deployer"] = ("cached-token", SimpleNamespace(token_id="jti", maturity_level=1)) + stage = await orch._execute_agent_stage("deployer") + assert stage.success is False + assert "credential verification failed" in stage.error and "fail-closed" in stage.error + vera["identity"].verify_credential.assert_called_once_with("cached-token") + network.route_to_agent.assert_not_awaited() + + +@pytest.mark.parametrize("result", [None, ["invalid"], {"status": "failed"}]) +async def test_vera_records_failure_proof_for_invalid_stage_results(fake_vera_firewall_module, result): + vera = _fake_vera() + network = _stub_network() + network.route_to_agent.return_value = result + stage = await _orchestrator(network=network, vera=vera)._execute_agent_stage("deployer") + assert stage.success is False + vera["proof_store"].append.assert_called_once_with(vera["ExecutionProof"].return_value) + assert vera["ExecutionProof"].call_args.kwargs["output_summary"] == f"error: {stage.error}" + + +async def test_vera_init_failure_blocks_every_stage(): + network = _stub_network() + with patch.object(po, "_get_vera", side_effect=po.VeraInitError("RuntimeError: bad key")), patch.object( + po, "get_agent_network", return_value=network + ), patch.object(po, "get_emitter", return_value=_Emitter()): + orch = po.VideoPipelineOrchestrator() + stage = await orch._execute_agent_stage("deployer") + assert stage.success is False + assert "failed to initialize" in stage.error + network.route_to_agent.assert_not_awaited() + + +@pytest.mark.parametrize( + "error", + [ + RuntimeError("identity key unreadable"), + ImportError("identity key unreadable"), + ModuleNotFoundError("identity key unreadable", name="identity_backend"), + ], +) +def test_get_vera_raises_when_installed_modules_fail_to_initialize(monkeypatch, error): + def boom(): + raise error + + modules = { + "vera": types.ModuleType("vera"), + "vera.identity": types.ModuleType("vera.identity"), + "vera.firewall": types.ModuleType("vera.firewall"), + "vera.gateway": types.ModuleType("vera.gateway"), + "vera.proof_chain": types.ModuleType("vera.proof_chain"), + "vera.enforcement": types.ModuleType("vera.enforcement"), + "vera.maturity": types.ModuleType("vera.maturity"), + "vera.enforcer": types.ModuleType("vera.enforcer"), + } + modules["vera.identity"].get_identity_service = boom + modules["vera.firewall"].get_firewall = MagicMock() + modules["vera.gateway"].get_gateway = MagicMock() + modules["vera.proof_chain"].ExecutionProof = MagicMock() + modules["vera.proof_chain"].get_proof_store = MagicMock() + modules["vera.proof_chain"].hash_data = MagicMock() + modules["vera.enforcement"].get_breaker_manager = MagicMock() + modules["vera.maturity"].get_maturity_runtime = MagicMock() + modules["vera.enforcer"].get_enforcer = MagicMock() + for name, module in modules.items(): + monkeypatch.setitem(sys.modules, name, module) + with pytest.raises(po.VeraInitError, match="identity key unreadable"): + po._get_vera() diff --git a/vercel.json b/vercel.json index d77369556..c95a57bff 100644 --- a/vercel.json +++ b/vercel.json @@ -7,7 +7,10 @@ { "source": "/(.*)", "has": [ - { "type": "host", "value": "event-relay-web.vercel.app" } + { + "type": "host", + "value": "event-relay-web.vercel.app" + } ], "destination": "https://uvai.io/$1", "permanent": true @@ -15,10 +18,18 @@ { "source": "/(.*)", "has": [ - { "type": "host", "value": "v0-uvai.vercel.app" } + { + "type": "host", + "value": "v0-uvai.vercel.app" + } ], "destination": "https://uvai.io/$1", "permanent": true } - ] + ], + "git": { + "deploymentEnabled": { + "fix/2168-mcp-routed-false-success": false + } + } }