Fail closed when host deployment evidence is missing

platform/inneros_core_runtime/temporal_activities.py

@@ -170,6 +170,38 @@ def _hydrate_gate_envelope(envelope_dict: Dict[str, Any]) -> Dict[str, Any]:
     return out
 
 
+def _requires_observed_host_mutation(envelope_dict: Dict[str, Any]) -> bool:
+    """Host/deploy operations must prove real execution, not just generated code/tests."""
+    task_class = str(envelope_dict.get("task_class") or "").strip().lower()
+    if task_class == "deployment":
+        return True
+    policy = str(envelope_dict.get("execution_policy") or "").strip().lower()
+    payload = envelope_dict.get("payload") if isinstance(envelope_dict.get("payload"), dict) else {}
+    objective = " ".join(
+        str(envelope_dict.get(key) or "")
+        for key in ("title", "objective")
+    ).lower()
+    objective += " " + " ".join(str(item or "") for item in envelope_dict.get("checklist") or []).lower()
+    host_markers = ("deploy", "deployment", "restart", "runtime", "platform sync", "platform-sync", "sync intel", "sync amd")
+    policy_markers = ("deploy", "host", "platform-sync", "platform_sync")
+    return (
+        any(marker in policy for marker in policy_markers)
+        or (
+            any(marker in objective for marker in host_markers)
+            and bool(payload.get("rollback_required") or payload.get("verify_dual_node"))
+        )
+    )
+
+
+def _observed_host_mutation_evidence(agent_result: Dict[str, Any]) -> tuple[bool, dict[str, Any], dict[str, Any]]:
+    nested = agent_result.get("evidence") if isinstance(agent_result.get("evidence"), dict) else {}
+    host = agent_result.get("host_execution") or agent_result.get("deployment_evidence") or nested.get("host_execution") or nested.get("deployment_evidence") or {}
+    post = agent_result.get("postconditions") or agent_result.get("observed_postconditions") or nested.get("postconditions") or nested.get("observed_postconditions") or {}
+    host_ok = isinstance(host, dict) and host.get("executed") is True
+    post_ok = isinstance(post, dict) and post.get("verified") is True
+    return bool(host_ok and post_ok), host if isinstance(host, dict) else {}, post if isinstance(post, dict) else {}
+
+
 @activity.defn
 async def activity_validate_completion_gate(envelope_dict: Dict[str, Any], agent_result: Dict[str, Any]) -> Dict[str, Any]:
     _safe_heartbeat("validating_completion_gate")
@@ -183,6 +215,19 @@ async def activity_validate_completion_gate(envelope_dict: Dict[str, Any], agent
     evidence_required = list(envelope_dict.get("evidence_required") or [])
     commit_sha = str(agent_result.get("commit_sha") or "").strip()
 
+    if _requires_observed_host_mutation(envelope_dict):
+        host_verified, host_evidence, postconditions = _observed_host_mutation_evidence(agent_result)
+        if not host_verified:
+            return {
+                "passed": False,
+                "error": (
+                    "Completion prohibited: host/deployment task requires observed host mutation "
+                    "and verified postconditions; generated files or passing unit tests are not deployment evidence"
+                ),
+                "host_execution": host_evidence,
+                "postconditions": postconditions,
+            }
+
     if commit_sha and task_class == "verification" and (
         not evidence_required or set(evidence_required) <= {"commit_sha"}
     ):

platform/tests/test_completion_gate_evidence_keys.py

@@ -24,5 +24,70 @@ def test_dotted_test_results_exit_code_satisfies_requirement(self) -> None:
         self.assertTrue(result.get("passed"), result)
 
 
+    def test_host_deploy_cannot_pass_with_generated_files_and_pytest_only(self) -> None:
+        envelope = {
+            "task_class": "coding",
+            "execution_lane": "internal",
+            "execution_policy": "owner-approved-bounded-platform-sync",
+            "payload": {"rollback_required": True, "verify_dual_node": True},
+            "title": "Deploy platform and restart runtime",
+        }
+        agent_result = {
+            "ok": True,
+            "files_count": 3,
+            "objective_files_count": 3,
+            "objective_paths": ["src/fake.py", "tests/fake_test.py", "package.json"],
+            "code_diff": "+ fake",
+            "test_results": {"exit_code": 0, "ok": True},
+        }
+        result = asyncio.run(ta.activity_validate_completion_gate(envelope, agent_result))
+        self.assertFalse(result.get("passed"), result)
+        self.assertIn("observed host mutation", result.get("error", ""))
+
+    def test_host_deploy_requires_real_execution_and_postcondition_evidence(self) -> None:
+        envelope = {
+            "task_class": "deployment",
+            "execution_lane": "internal",
+            "execution_policy": "owner-approved-bounded-platform-sync",
+            "payload": {"rollback_required": True, "verify_dual_node": True},
+            "title": "Deploy platform and restart runtime",
+        }
+        agent_result = {
+            "ok": True,
+            "test_results": {"exit_code": 0, "ok": True},
+            "host_execution": {
+                "executed": True,
+                "commands": ["git pull --ff-only", "systemctl --user restart inneros-mcp"],
+                "before_sha": "a" * 40,
+                "after_sha": "b" * 40,
+            },
+            "postconditions": {
+                "verified": True,
+                "service_health": {"amd": "healthy", "intel": "healthy"},
+                "registry_resolution": {"ok": True},
+            },
+        }
+        result = asyncio.run(ta.activity_validate_completion_gate(envelope, agent_result))
+        self.assertTrue(result.get("passed"), result)
+
+
+    def test_host_deploy_verification_sha_cannot_bypass_host_evidence(self) -> None:
+        envelope = {
+            "task_class": "verification",
+            "execution_lane": "internal",
+            "execution_policy": "owner-approved-bounded-platform-sync",
+            "payload": {"rollback_required": True},
+            "title": "Verify deployed runtime",
+        }
+        agent_result = {
+            "ok": True,
+            "commit_sha": "c" * 40,
+            "test_results": {"exit_code": 0, "ok": True},
+        }
+        result = asyncio.run(ta.activity_validate_completion_gate(envelope, agent_result))
+        self.assertFalse(result.get("passed"), result)
+        self.assertIn("observed host mutation", result.get("error", ""))
+
+
 if __name__ == "__main__":
     unittest.main()