fix(P0): fail-closed evidence and honest Temporal-to-Mongo reconciliation

.github/workflows/inneros-runtime-gates.yml

@@ -11,7 +11,14 @@ on:
       - "platform/inneros_core_runtime/capability_gateway_peer.py"
       - "platform/tests/test_capability_gateway.py"
       - "platform/tests/test_capability_gateway_regressions_p0.py"
+      - "platform/inneros_core_runtime/governed_model_writes.py"
+      - "platform/inneros_core_runtime/temporal_bounded_executor.py"
+      - "platform/inneros_core_runtime/interactive_ops_runner.py"
+      - "platform/scripts/cursor_ha_e2e_server_probe.py"
+      - "platform/tests/test_governed_model_writes.py"
       - "platform/inneros_core_runtime/execution_receipts.py"
+      - "platform/scripts/reconcile_ops_mongo_temporal.py"
+      - "platform/tests/test_reconcile_ops_mongo_temporal.py"
       - "platform/scripts/cursor_ops_headless_execute_one.py"
       - "platform/tests/test_execution_receipts.py"
       - "platform/inneros_core_runtime/temporal_activities.py"
@@ -39,6 +46,15 @@ on:
       - "platform/inneros_core_runtime/capability_gateway_peer.py"
       - "platform/tests/test_capability_gateway.py"
       - "platform/tests/test_capability_gateway_regressions_p0.py"
+      - "platform/inneros_core_runtime/governed_model_writes.py"
+      - "platform/inneros_core_runtime/temporal_bounded_executor.py"
+      - "platform/inneros_core_runtime/interactive_ops_runner.py"
+      - "platform/scripts/cursor_ha_e2e_server_probe.py"
+      - "platform/tests/test_governed_model_writes.py"
+      - "platform/inneros_core_runtime/execution_receipts.py"
+      - "platform/tests/test_execution_receipts.py"
+      - "platform/scripts/reconcile_ops_mongo_temporal.py"
+      - "platform/tests/test_reconcile_ops_mongo_temporal.py"
       - "platform/inneros_core_runtime/temporal_activities.py"
       - "platform/inneros_core_runtime/temporal_workflows.py"
       - "platform/inneros_core_runtime/agents/pool_agent_runners.py"
@@ -84,9 +100,16 @@ jobs:
             platform/inneros_core_runtime/capability_gateway_peer.py \
             scripts/inneros_local_release_executor.py \
             platform/tests/test_capability_gateway_regressions_p0.py \
+            platform/inneros_core_runtime/governed_model_writes.py \
+            platform/inneros_core_runtime/temporal_bounded_executor.py \
+            platform/inneros_core_runtime/interactive_ops_runner.py \
+            platform/scripts/cursor_ha_e2e_server_probe.py \
+            platform/tests/test_governed_model_writes.py \
             platform/inneros_core_runtime/execution_receipts.py \
             platform/scripts/cursor_ops_headless_execute_one.py \
             platform/tests/test_execution_receipts.py \
+            platform/scripts/reconcile_ops_mongo_temporal.py \
+            platform/tests/test_reconcile_ops_mongo_temporal.py \
             platform/inneros_core_runtime/temporal_activities.py \
             platform/inneros_core_runtime/agents/pool_agent_runners.py \
             platform/tests/test_global_coordination_guards.py \
@@ -101,6 +124,8 @@ jobs:
             platform/tests/test_execution_binding.py \
             platform/tests/test_local_model_router_fallback.py \
             platform/tests/test_execution_receipts.py \
+            platform/tests/test_reconcile_ops_mongo_temporal.py \
+            platform/tests/test_governed_model_writes.py \
             platform/tests/test_capability_gateway.py \
             platform/tests/test_capability_gateway_regressions_p0.py \
             platform/tests/test_peer_ops_runner_contract.py \

platform/inneros_core_runtime/execution_receipts.py

@@ -87,16 +87,28 @@ def completion_evidence_from_worktree(
     worktree: Path,
     before_sha: str,
     *,
-    test_exit_code: int = 0,
+    test_exit_code: int | None = None,
     **extra: Any,
 ) -> dict[str, Any]:
     """Build a Temporal/Mongo completion payload shared by Cursor, Codex, and Dev Swarm."""
     proof = committed_objective_receipt(worktree, before_sha)
     if not proof.get("ok"):
         return proof
-    evidence = {
-        **proof,
-        "test_results": {"exit_code": int(test_exit_code), "ok": int(test_exit_code) == 0},
-        **extra,
-    }
+    # Never invent a passing test result: it must be supplied by the runner
+    # after executing actual tests. Missing test evidence stays missing.
+    evidence = {**proof, **extra}
+    if test_exit_code is not None:
+        try:
+            verified_exit_code = int(test_exit_code)
+        except (TypeError, ValueError):
+            return {**proof, "ok": False, "error": "invalid_test_exit_code"}
+        evidence["test_results"] = {
+            "exit_code": verified_exit_code,
+            "ok": verified_exit_code == 0,
+        }
+        if verified_exit_code != 0:
+            return {
+                "ok": False, "error": "tests_failed",
+                "evidence": evidence, "commit_sha": proof["commit_sha"],
+            }
     return {"ok": True, "evidence": evidence, "commit_sha": proof["commit_sha"]}

platform/scripts/reconcile_ops_mongo_temporal.py

@@ -39,7 +39,22 @@ async def _temporal_describe(workflow_id: str) -> dict[str, Any]:
         desc = await handle.describe()
         status_name = str(desc.status)
         terminal = any(x in status_name for x in ("COMPLETED", "FAILED", "TERMINATED", "CANCELED"))
-        return {"ok": True, "workflow_id": workflow_id, "status": status_name, "run_id": desc.run_id, "terminal": terminal}
+        result_status = None
+        result_gate_passed = None
+        if "COMPLETED" in status_name:
+            # The RPC workflow may COMPLETE while returning {"status": "failed"}.
+            result = await handle.result()
+            if isinstance(result, dict):
+                result_status = str(result.get("status") or "").strip().lower() or None
+                evidence = result.get("evidence") or {}
+                gate = evidence.get("gate_verification") if isinstance(evidence, dict) else None
+                if isinstance(gate, dict):
+                    result_gate_passed = gate.get("passed") is True
+        return {
+            "ok": True, "workflow_id": workflow_id, "status": status_name,
+            "run_id": desc.run_id, "terminal": terminal,
+            "result_status": result_status, "result_gate_passed": result_gate_passed,
+        }
     except Exception as exc:
         return {"ok": False, "workflow_id": workflow_id, "error": type(exc).__name__, "detail": str(exc)[:240]}
 
@@ -55,29 +70,37 @@ def _proposed_action(task: dict[str, Any], temporal: dict[str, Any]) -> dict[str
         return None
     tid = str(task.get("task_id") or "")
     wf_id = str(task.get("workflow_id") or f"ops_task:{tid}")
-    headless_done = bool(task.get("headless_execution_finished_at"))
-    stale_canary = _is_stale_canary(task)
-    if not temporal.get("ok"):
-        if stale_canary or headless_done or _lease_expired(task):
-            return {
-                "task_id": tid,
-                "action": "set_status",
-                "new_status": "superseded" if stale_canary else "failed",
-                "reason": "temporal_workflow_missing",
-                "temporal": temporal,
-            }
+    # Never mark tasks failed merely because Temporal is temporarily unreachable.
+    # A lease timeout is not proof that an external process stopped.
+    if not temporal.get("ok") or not temporal.get("terminal"):
         return None
-    if temporal.get("terminal"):
-        st = str(temporal.get("status") or "")
-        new_status = "failed" if any(x in st for x in ("FAILED", "TERMINATED", "CANCELED")) else "completed"
-        return {
-            "task_id": tid,
-            "action": "set_status",
-            "new_status": new_status,
-            "reason": "temporal_terminal_mongo_stale",
-            "temporal": temporal,
-        }
-    return None
+    st = str(temporal.get("status") or "").upper()
+    if any(x in st for x in ("FAILED", "TERMINATED")):
+        new_status = "failed"
+    elif "CANCELED" in st:
+        new_status = "cancelled"
+    elif "COMPLETED" in st:
+        outcome = str(temporal.get("result_status") or "").lower()
+        if outcome in {"failed", "cancelled"}:
+            new_status = outcome
+        elif outcome == "completed":
+            is_coding = str(task.get("task_class") or "coding").lower() == "coding"
+            if is_coding and temporal.get("result_gate_passed") is not True:
+                return None
+            new_status = "completed"
+        else:
+            # A successful Temporal RPC can carry a blocked or failed task.
+            # Unknown legacy result shape must be reviewed, not completed.
+            return None
+    else:
+        return None
+    return {
+        "task_id": tid,
+        "action": "set_status",
+        "new_status": new_status,
+        "reason": "temporal_result_verified_mongo_stale",
+        "temporal": temporal,
+    }
 
 
 def main() -> int:
@@ -107,11 +130,19 @@ def main() -> int:
             actions.append({**prop, "mongo_status": doc.get("status"), "correlation_id": doc.get("correlation_id")})
 
     applied = 0
+    skipped_concurrent_updates = 0
     if args.apply:
         now = _utc()
         for act in actions:
-            col.update_one(
-                {"task_id": act["task_id"], "status": {"$nin": sorted(TERMINAL_MONGO)}},
+            # CAS: a claim/lease/terminal state that moved since the scan wins.
+            prior = next((row for row in rows if row.get("task_id") == act["task_id"]), {})
+            filt = {"task_id": act["task_id"], "status": act["mongo_status"]}
+            if "revision" in prior:
+                filt["revision"] = prior["revision"]
+            if "lease_until" in prior:
+                filt["lease_until"] = prior["lease_until"]
+            change = col.update_one(
+                filt,
                 {
                     "$set": {
                         "status": act["new_status"],
@@ -123,14 +154,18 @@ def main() -> int:
                     "$inc": {"revision": 1},
                 },
             )
-            applied += 1
+            if getattr(change, "modified_count", 0) == 1:
+                applied += 1
+            else:
+                skipped_concurrent_updates += 1
 
     report = {
         "ok": True,
         "generated_at": _utc(),
         "scanned": len(rows),
         "proposed": len(actions),
         "applied": applied,
+        "skipped_concurrent_updates": skipped_concurrent_updates,
         "dry_run": not args.apply,
         "actions": actions,
     }

platform/tests/test_execution_receipts.py

@@ -6,7 +6,9 @@
 import unittest
 from pathlib import Path
 
-from inneros_core_runtime.execution_receipts import committed_objective_receipt, git_head
+from inneros_core_runtime.execution_receipts import (
+    committed_objective_receipt, completion_evidence_from_worktree, git_head,
+)
 
 
 class ExecutionReceiptsTests(unittest.TestCase):
@@ -63,6 +65,30 @@ def test_missing_paths_after_commit_fail_closed(self) -> None:
         self.assertFalse(proof["ok"])
         self.assertEqual(proof["error"], "no_committed_objective_files")
 
+    def test_completed_receipt_never_fabricates_test_success(self) -> None:
+        (self.repo / "README.md").write_text("changed\n", encoding="utf-8")
+        self.git("add", "README.md")
+        self.git("commit", "-qm", "real task write")
+        proof = completion_evidence_from_worktree(self.repo, self.baseline)
+        self.assertTrue(proof["ok"], proof)
+        self.assertNotIn("test_results", proof["evidence"])
+        self.assertEqual(proof["evidence"]["objective_paths"], ["README.md"])
+
+    def test_completed_receipt_accepts_only_explicit_exit_code(self) -> None:
+        (self.repo / "README.md").write_text("changed\n", encoding="utf-8")
+        self.git("add", "README.md")
+        self.git("commit", "-qm", "real task write")
+        good = completion_evidence_from_worktree(self.repo, self.baseline, test_exit_code=0)
+        self.assertTrue(good["ok"], good)
+        self.assertEqual(good["evidence"]["test_results"]["exit_code"], 0)
+        failed = completion_evidence_from_worktree(self.repo, self.baseline, test_exit_code=1)
+        self.assertFalse(failed["ok"])
+        self.assertEqual(failed["error"], "tests_failed")
+        self.assertFalse(failed["evidence"]["test_results"]["ok"])
+        invalid = completion_evidence_from_worktree(self.repo, self.baseline, test_exit_code="invalid")
+        self.assertFalse(invalid["ok"])
+        self.assertEqual(invalid["error"], "invalid_test_exit_code")
+
 
 if __name__ == "__main__":
     unittest.main()

platform/tests/test_governed_model_writes.py

@@ -0,0 +1,106 @@
+"""Ensure Temporal materializes only governed local-model file payloads."""
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from inneros_core_runtime import governed_model_writes as gmw
+
+
+class GovernedMaterializationTests(unittest.TestCase):
+    def test_platform_prefix_rewritten_for_flat_runtime_worktree(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            wt = Path(tmp)
+            (wt / "tests").mkdir()
+            self.assertEqual(
+                gmw._rewrite_path_for_worktree(wt, "platform/tests/example.py"),
+                "tests/example.py",
+            )
+            self.assertEqual(
+                gmw._rewrite_path_for_worktree(wt, "tests/example.py"),
+                "tests/example.py",
+            )
+
+    def test_structured_candidate_invokes_allowlisted_writer(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            wt = Path(tmp)
+            (wt / "tests").mkdir()
+            envelope = {
+                "task_id": "ops_governed",
+                "correlation_id": "corr-governed",
+                "repo": "Rafa-Innerchispa/innerops-agentic-platform",
+                "work_branch": "chatgpt/ops-governed",
+                "objective": "Implement a tiny test marker",
+            }
+            payload = {"summary": "done", "files": [
+                {"path": "platform/tests/test_marker.py", "content": "assert True\n"},
+            ]}
+            def validate(parsed, *_args, **_kwargs):
+                return list(parsed["files"]), []
+            with (
+                patch("inneros_core_runtime.dev_swarm_scheduler._fanout_parse_model_json",
+                      return_value=payload),
+                patch("inneros_core_runtime.dev_swarm_scheduler._safe_generated_files",
+                      side_effect=validate) as validator,
+                patch("inneros_core_runtime.dev_swarm_scheduler._merge_node_scaffold",
+                      side_effect=lambda **kwargs: kwargs["files"]),
+                patch("inneros_core_runtime.local_execution_plane.write_file",
+                      return_value={"ok": True}) as writer,
+            ):
+                result = gmw.apply_candidate_response(
+                    envelope, wt, model_text='{"summary":"done"}',
+                )
+            self.assertTrue(result["ok"], result)
+            self.assertEqual(result["files_touched"], ["tests/test_marker.py"])
+            self.assertEqual(validator.call_args.args[0]["files"][0]["path"], "tests/test_marker.py")
+            self.assertEqual(writer.call_args.kwargs["path"], "tests/test_marker.py")
+            self.assertEqual(writer.call_args.kwargs["task_id"], "ops_governed")
+
+    def test_narrative_without_file_payload_does_not_write(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            with (
+                patch("inneros_core_runtime.dev_swarm_scheduler._fanout_parse_model_json",
+                      return_value=None),
+                patch("inneros_core_runtime.dev_swarm_scheduler._safe_generated_files",
+                      return_value=([], [])),
+                patch("inneros_core_runtime.dev_swarm_scheduler._merge_node_scaffold",
+                      return_value=[]),
+                patch("inneros_core_runtime.local_execution_plane.write_file") as writer,
+            ):
+                result = gmw.apply_candidate_response(
+                    {"task_id": "ops_narrative", "repo": "Rafa-Innerchispa/innerops-agentic-platform",
+                     "work_branch": "chatgpt/narrative"},
+                    Path(tmp), model_text="I've implemented all files",
+                )
+            self.assertFalse(result["ok"])
+            self.assertEqual(result["error"], "model_output_non_actionable")
+            writer.assert_not_called()
+
+    def test_partial_write_is_not_success(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            files = [{"path": "tests/a.py", "content": "a = 1\n"},
+                     {"path": "tests/b.py", "content": "b = 2\n"}]
+            with (
+                patch("inneros_core_runtime.dev_swarm_scheduler._fanout_parse_model_json",
+                      return_value={"files": files}),
+                patch("inneros_core_runtime.dev_swarm_scheduler._safe_generated_files",
+                      return_value=(files, [])),
+                patch("inneros_core_runtime.dev_swarm_scheduler._merge_node_scaffold",
+                      return_value=files),
+                patch("inneros_core_runtime.local_execution_plane.write_file",
+                      side_effect=[{"ok": True}, {"ok": False, "error": "denied"}]),
+            ):
+                result = gmw.apply_candidate_response(
+                    {"task_id": "ops_partial", "repo": "Rafa-Innerchispa/innerops-agentic-platform",
+                     "work_branch": "chatgpt/partial"}, Path(tmp),
+                    model_text='{"files":[]}',
+                )
+            self.assertFalse(result["ok"])
+            self.assertEqual(result["files_touched"], ["tests/a.py"])
+            self.assertEqual(result["write_count"], 2)
+
+
+if __name__ == "__main__":
+    unittest.main()