feat(phase3): add sanctum execution foundation

docs/en/README.md

@@ -33,6 +33,11 @@ English is the canonical source for all Benchwork documentation.
 - [RFC-0008: Codex Plugin-first Host Architecture](rfcs/RFC-0008-codex-plugin-first-host-architecture.md)
 - [RFC-0009: MCP as the Scientific Control Plane](rfcs/RFC-0009-mcp-scientific-control-plane.md)
 - [RFC-0010: The Invitation Installer Contract](rfcs/RFC-0010-invitation-installer-contract.md)
+- [RFC-0011: Sanctum Execution Model](rfcs/RFC-0011-sanctum-execution-model.md)
+- [RFC-0012: Job, Lease, and Worker Protocol](rfcs/RFC-0012-job-lease-worker-protocol.md)
+- [RFC-0013: Artifact Storage Model](rfcs/RFC-0013-artifact-storage-model.md)
+- [RFC-0014: Patch Promotion Protocol](rfcs/RFC-0014-patch-promotion-protocol.md)
+- [RFC-0015: Experiment Executor API](rfcs/RFC-0015-experiment-executor-api.md)
 
 ## Architecture
 

src/benchwork/athanor.py

@@ -3729,6 +3729,13 @@ def build(events: list[dict[str, Any]]) -> tuple[str, str, dict[str, Any]]:
                 artifact_path.relative_to(self.root.resolve())
             except ValueError as error:
                 raise AthanorError("Artifact location URI escapes the project") from error
+            managed_storage = (self.root / ".benchwork" / "storage").resolve()
+            try:
+                artifact_path.relative_to(managed_storage)
+            except ValueError:
+                pass
+            else:
+                raise AthanorError("Artifact location cannot use managed storage namespace")
             try:
                 artifact_blob = artifact_path.read_bytes()
             except OSError as error:

src/benchwork/mcp/runtime.py

@@ -15,6 +15,7 @@
 from ..circle import CapsuleStore, CapabilityRegistry, Ward
 from ..doctor import deep_doctor
 from ..hosts import HOSTS
+from ..execution import ExecutionService
 from ..project import ProjectContext, discover_project_root
 from ..schema_validation import _schema_directory, validate_instance
 from ..tasks import TaskService
@@ -85,6 +86,36 @@ def _root(self) -> Path:
     def _athanor(self) -> Athanor:
         return Athanor(self._root())
 
+    def _execution(self) -> ExecutionService:
+        return ExecutionService(self._root())
+
+    def _execution_run(
+        self,
+        tool: str,
+        operation: Callable[[], dict[str, Any]],
+    ) -> dict[str, Any]:
+        """Render closed executor failures without leaking local runtime detail."""
+        try:
+            return success(tool, operation())
+        except AthanorError as error:
+            message = str(error)
+            lowered = message.lower()
+            if "unknown execution job" in lowered:
+                code = "EXECUTION_NOT_FOUND"
+            elif "no terminal outcome" in lowered:
+                code = "EXECUTION_NOT_READY"
+            elif "idempotency conflict" in lowered:
+                code = "IDEMPOTENCY_CONFLICT"
+            elif "cursor" in lowered:
+                code = "STALE_CURSOR"
+            elif "outcome is ineligible" in lowered:
+                code = "RESULT_INELIGIBLE"
+            elif "revision conflict" in lowered or "not terminalizable" in lowered:
+                code = "EXECUTION_CONFLICT"
+            else:
+                code = "EXECUTION_POLICY_REJECTED"
+            return failure(tool, error, code=code, project_root=self._configured_root or Path.cwd())
+
     def _run(
         self,
         tool: str,
@@ -385,6 +416,80 @@ def operation() -> dict[str, Any]:
 
         return self._run(tool, operation)
 
+    def benchwork_start_job(
+        self,
+        execution_specification: dict[str, Any],
+        idempotency_key: str,
+    ) -> dict[str, Any]:
+        """Submit a typed pre-conformance Job without accepting a command or shell.
+
+        This Host-neutral adapter is intentionally absent from the published
+        MCP registry until RFC-0012 through RFC-0015 schemas and conformance
+        requirements are implemented.
+        """
+        return self._execution_run(
+            "benchwork_start_job",
+            lambda: self._execution().start(execution_specification, idempotency_key),
+        )
+
+    def benchwork_observe_job(
+        self,
+        job_id: str,
+        limit: int = 50,
+        cursor: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        """Read a bounded, fixed-prefix operational Job observation."""
+        return self._execution_run(
+            "benchwork_observe_job",
+            lambda: self._execution().observe(job_id, limit=limit, cursor=cursor),
+        )
+
+    def benchwork_cancel_job(
+        self,
+        job_id: str,
+        job_binding_sigil: str,
+        expected_job_revision: int,
+        idempotency_key: str,
+        reason: str,
+    ) -> dict[str, Any]:
+        """Record an idempotent stop request; no request is a process signal."""
+        return self._execution_run(
+            "benchwork_cancel_job",
+            lambda: self._execution().cancel(
+                job_id,
+                job_binding_sigil,
+                expected_job_revision,
+                idempotency_key,
+                reason,
+            ),
+        )
+
+    def benchwork_get_job_result(self, job_id: str) -> dict[str, Any]:
+        """Derive a terminal operational Outcome without canonical acceptance."""
+        return self._execution_run(
+            "benchwork_get_job_result",
+            lambda: self._execution().get_outcome(job_id),
+        )
+
+    def benchwork_accept_job_result(
+        self,
+        job_id: str,
+        execution_job_outcome_sigil: str,
+        idempotency_key: str,
+    ) -> dict[str, Any]:
+        """Fail closed until an Outcome has complete Agent Result v2 evidence."""
+        def operation() -> dict[str, Any]:
+            if not isinstance(idempotency_key, str) or not idempotency_key:
+                raise AthanorError("execution acceptance idempotency key is invalid")
+            outcome = self._execution().get_outcome(job_id)
+            if execution_job_outcome_sigil != outcome["outcome_sigil"]:
+                raise AthanorError("execution Outcome is ineligible")
+            if not outcome["eligible_for_acceptance"]:
+                raise AthanorError("execution Outcome is ineligible")
+            raise AthanorError("execution Outcome is ineligible")
+
+        return self._execution_run("benchwork_accept_job_result", operation)
+
     def benchwork_open_task(
         self,
         capability: str,

tests/test_athanor.py

@@ -90,6 +90,23 @@ def test_working_requires_frozen_protocol_and_replays_lifecycle(self) -> None:
         self.assertEqual(self.athanor.workings()[working_id]["stage"], "PILOT")
         self.assertEqual(len(self.athanor.workings()[working_id]["history"]), 2)
 
+    def test_artifact_registration_rejects_phase_three_managed_storage(self) -> None:
+        program_id, _ = self.athanor.create_program("storage-boundary", "Storage boundary")
+        managed = self.root / ".benchwork" / "storage" / "blobs" / "candidate"
+        managed.parent.mkdir(parents=True)
+        managed.write_bytes(b"operational-only")
+        with self.assertRaisesRegex(AthanorError, "managed storage namespace"):
+            self.athanor.register_artifact(
+                "AR-001",
+                program_id,
+                "implementation",
+                {
+                    "uri": ".benchwork/storage/blobs/candidate",
+                    "sigil": "sha256:" + hashlib.sha256(b"operational-only").hexdigest(),
+                },
+                program_id,
+            )
+
     def test_schema_files_are_versioned_json_contracts(self) -> None:
         schemas = Path(__file__).parents[1] / "schemas"
         for schema_path in schemas.glob("*.json"):