Proof: вынести общую execution-границу

proof/region/v1/PROTOCOL.md

@@ -187,13 +187,28 @@ Git executable/version, repository URL и tag являются диагност
 
 ## Diagnostic execution boundary
 
+`proof/region/v1/executor.py` — общий для enclosure engines leaf без импорта
+Arb/MPFI, formula или comparator semantics. Он же единолично кодирует
+`execution-invocation.v1` и `execution-platform.v1`; engine pipeline не может
+вводить параллельную identity того же процесса. Sandbox release
+`labcolors.proof-region.executor.linux-x86_64.v1` намеренно не сохраняет
+старую Arb-domain identity: это hard cut, а не compatibility alias.
+
 `ControlledExecutorV1` — единственный владелец one-shot capability: новый,
 неуспешный, перекрывающийся probe или замена backend отзывают ранее выданный
 объект до RUN. Capability выпускается контроллером для одного probe-поколения
 и одного process id; fork не дублирует право запуска. Backend сообщает только
 наблюдённые свойства хоста, получает guard текущего probe и не может продлить
 жизнь capability повторно используемым report-объектом.
 
+`ExecutionRequestV1`, его limits и `SupportedV1` являются структурно
+неизменяемыми значениями. Публичные execution identity functions воспроизводят
+admission из точных координат и возвращают
+`bytes | ExecutionIdentityRejectedV1`: foreign или даже намеренно forged
+malformed value становится versioned typed rejection, а не новой identity и не
+exception-channel. `ControlledExecutorV1` отвергает такой request до probe и
+backend run как `ObserverFailureV1(REQUEST_NOT_ADMITTED)`.
+
 Linux backend допускается лишь в отдельном helper process. Helper находится в
 прямом дочернем cgroup объявленного parent, а весь parent subtree имеет
 `pids.max = 2` и перед probe содержит ровно observer. Эти два task slots имеют

proof/region/v1/arb/pipeline.py

@@ -132,8 +132,6 @@
 _BUILD_INPUT_BUNDLE_ID_LABEL_V1 = (
     b"labcolors.proof-region.arb-build-input-bundle.v1\0"
 )
-_INVOCATION_ID_LABEL_V1 = b"labcolors.proof-region.arb-invocation.v1\0"
-_PLATFORM_ID_LABEL_V1 = b"labcolors.proof-region.arb-run-platform.v1\0"
 _BUILD_SOURCES_TOKEN = object()
 _COMPARATOR_TOKEN = object()
 _BUILD_OBSERVATION_TOKEN = object()
@@ -2515,44 +2513,6 @@ def input_bundle(self) -> SealedBuildInputBundleV1:
 )
 
 
-def invocation_identity_v1(request: executor.ExecutionRequestV1) -> bytes:
-    if type(request) is not executor.ExecutionRequestV1:
-        raise TypeError("request must be ExecutionRequestV1")
-    chunks: list[bytes] = [hashlib.sha256(request.executable).digest()]
-    chunks.append(len(request.argv).to_bytes(4, "big"))
-    chunks.extend(request.argv)
-    chunks.append(len(request.environment).to_bytes(4, "big"))
-    for key, value in request.environment:
-        chunks.extend((key, value))
-    chunks.extend(
-        (
-            request.cwd,
-            hashlib.sha256(request.stdin).digest(),
-            len(request.stdin).to_bytes(8, "big"),
-            request.umask.to_bytes(4, "big"),
-        )
-    )
-    for item in fields(request.limits):
-        chunks.append(getattr(request.limits, item.name).to_bytes(8, "big"))
-    return _identity(_INVOCATION_ID_LABEL_V1, tuple(chunks))
-
-
-def platform_identity_v1(report: executor.SupportedV1) -> bytes:
-    if (
-        type(report) is not executor.SupportedV1
-        or report.platform != executor.EXECUTION_PLATFORM_V1
-        or report.sandbox_policy_release != executor.SANDBOX_POLICY_RELEASE_V1
-    ):
-        raise TypeError("report must be the exact V1 supported platform")
-    return _identity(
-        _PLATFORM_ID_LABEL_V1,
-        (
-            report.platform.encode("ascii"),
-            report.sandbox_policy_release.encode("ascii"),
-        ),
-    )
-
-
 class ControlledPipelineV1:
     def __init__(
         self,

proof/region/v1/arb/receipt.py

@@ -60,6 +60,7 @@ def source_bound_policy_identity_v1() -> bytes:
             b"source=lock-plus-owned-archive-and-build-input-replay",
             b"build=one-sealed-bundle-two-fresh-byte-equal-attempts",
             b"run=retained-executable-object-one-contained-process",
+            b"identity=immutable-coordinates-total-rejection-v1",
             b"claim=provenance-only-no-numerical-semantics",
             b"trust=unsealed-linux-x64-host-and-docker-daemon",
         ),
@@ -288,8 +289,13 @@ def _run_identity_v1(
         transcript.counters,
         transcript.witness_store,
     )
-    invocation_identity = pipeline.invocation_identity_v1(invocation)
-    platform_identity = pipeline.platform_identity_v1(platform_value)
+    invocation_identity = executor.invocation_identity_v1(invocation)
+    platform_identity = executor.platform_identity_v1(platform_value)
+    if (
+        type(invocation_identity) is not bytes
+        or type(platform_identity) is not bytes
+    ):
+        raise TypeError("execution identity replay was rejected")
     expected_claim = protocol.RunClaimV1.for_transcript(
         request.job,
         build.comparator.manifest,
@@ -526,7 +532,7 @@ def __post_init__(self) -> None:
 
 
 def _limits_copy_v1(value: executor.ExecutionLimitsV1) -> executor.ExecutionLimitsV1:
-    return executor.ExecutionLimitsV1(*(getattr(value, item.name) for item in fields(value)))
+    return executor.ExecutionLimitsV1(*value)
 
 
 def _resolve_request_v1(
@@ -716,8 +722,26 @@ def execute(self, request: pipeline.PipelineRequestV1) -> SourceBoundResultV1:
                 str(error),
             )
         try:
-            invocation_identity = pipeline.invocation_identity_v1(invocation)
-            platform_identity = pipeline.platform_identity_v1(capability)
+            invocation_identity = executor.invocation_identity_v1(invocation)
+            if type(invocation_identity) is executor.ExecutionIdentityRejectedV1:
+                return pipeline.ExecutionRejectedV1(
+                    pipeline.ExecutionFailureReasonV1.BACKEND_CONTRACT,
+                    invocation_identity,
+                )
+            platform_identity = executor.platform_identity_v1(capability)
+            if type(platform_identity) is executor.ExecutionIdentityRejectedV1:
+                return pipeline.ExecutionRejectedV1(
+                    pipeline.ExecutionFailureReasonV1.BACKEND_CONTRACT,
+                    platform_identity,
+                )
+            if (
+                type(invocation_identity) is not bytes
+                or type(platform_identity) is not bytes
+            ):
+                return pipeline.ExecutionRejectedV1(
+                    pipeline.ExecutionFailureReasonV1.BACKEND_CONTRACT,
+                    (invocation_identity, platform_identity),
+                )
             run_claim = protocol.RunClaimV1.for_transcript(
                 replay_request.job,
                 built.comparator.manifest,

proof/region/v1/arb/tests/gate.py

@@ -11,10 +11,11 @@
 
 
 TEST_DIRECTORY = Path(__file__).resolve().parent
+SHARED_TEST_DIRECTORY = TEST_DIRECTORY.parents[1] / "tests"
 REPO = Path(__file__).resolve().parents[5]
 sys.path.insert(0, str(REPO))
 EXPECTED_TEST_INVENTORY_SHA256 = (
-    "6e73ade1e7d5b21d50fe9826a1b39e4043e63bcd090b504dbee5e1c38515e373"
+    "9df49e5bc78ab7cf2386570f500c80f3759f21385f597f2720aa27b1e9700a76"
 )
 _EVALUATOR_REASON = "set LABCOLORS_ARB_EVALUATOR to the controlled C17 binary"
 EXPECTED_SKIPS = frozenset(
@@ -69,6 +70,23 @@ def test_inventory_sha256_v1(suite: unittest.TestSuite) -> str:
     return hashlib.sha256(_inventory_preimage_v1(test_ids)).hexdigest()
 
 
+def full_suite_v1() -> unittest.TestSuite:
+    """Compose the shared execution contract and Arb-only contract once each."""
+
+    return unittest.TestSuite(
+        (
+            unittest.defaultTestLoader.discover(
+                str(SHARED_TEST_DIRECTORY),
+                pattern="test_executor.py",
+            ),
+            unittest.defaultTestLoader.discover(
+                str(TEST_DIRECTORY),
+                pattern="test_*.py",
+            ),
+        )
+    )
+
+
 def run_exact_suite_v1(
     suite: unittest.TestSuite,
     *,
@@ -121,10 +139,7 @@ def run_exact_suite_v1(
 
 
 def main() -> int:
-    suite = unittest.defaultTestLoader.discover(
-        str(TEST_DIRECTORY),
-        pattern="test_*.py",
-    )
+    suite = full_suite_v1()
     return run_exact_suite_v1(
         suite,
         expected_inventory_sha256=EXPECTED_TEST_INVENTORY_SHA256,

proof/region/v1/arb/tests/native_gate.py

@@ -12,7 +12,7 @@
 sys.path.insert(0, str(REPO))
 
 from proof.region.v1.arb.tests import gate  # noqa: E402
-from proof.region.v1.arb.tests.test_executor import (  # noqa: E402
+from proof.region.v1.tests.test_executor import (  # noqa: E402
     NativeLinuxIntegrationTests,
 )
 from proof.region.v1.arb.tests.test_receipt import (  # noqa: E402

proof/region/v1/arb/tests/test_build_recipe.py

@@ -19,6 +19,26 @@
 
 
 class ArbBuildRecipeTests(unittest.TestCase):
+    def test_fast_gate_includes_the_shared_executor_suite_exactly_once(self) -> None:
+        tests = tuple(arb_gate._iter_tests_v1(arb_gate.full_suite_v1()))
+        identifiers = tuple(test.id() for test in tests)
+        executor_identifiers = tuple(
+            identifier for identifier in identifiers if identifier.startswith("test_executor.")
+        )
+        expected = tuple(
+            test.id()
+            for test in arb_gate._iter_tests_v1(
+                unittest.defaultTestLoader.discover(
+                    str(arb_gate.SHARED_TEST_DIRECTORY),
+                    pattern="test_executor.py",
+                )
+            )
+        )
+
+        self.assertTrue(executor_identifiers)
+        self.assertEqual(executor_identifiers, expected)
+        self.assertEqual(len(identifiers), len(set(identifiers)))
+
     def test_pr_gate_requires_a_disposable_exact_workflow_runner(self) -> None:
         source = WORKFLOW.read_text(encoding="utf-8")
         runner_contracts = [

proof/region/v1/arb/tests/test_pipeline.py

@@ -221,17 +221,19 @@ def _foreign_comparator() -> ContentResolvedComparatorManifestV2:
     return ContentResolvedComparatorManifestV2.admit(manifest, by_digest.get)
 
 
-def _limits() -> executor.ExecutionLimitsV1:
-    return executor.ExecutionLimitsV1(
-        max_executable_bytes=16 * 1024 * 1024,
-        max_stdin_bytes=16 * 1024 * 1024,
-        max_argument_bytes=4096,
-        max_stdout_bytes=16 * 1024 * 1024,
-        max_stderr_bytes=64 * 1024,
-        wall_timeout_ns=60_000_000_000,
-        memory_max_bytes=1024 * 1024 * 1024,
-        pids_max=1,
-    )
+def _limits(**changes: int) -> executor.ExecutionLimitsV1:
+    values = {
+        "max_executable_bytes": 16 * 1024 * 1024,
+        "max_stdin_bytes": 16 * 1024 * 1024,
+        "max_argument_bytes": 4096,
+        "max_stdout_bytes": 16 * 1024 * 1024,
+        "max_stderr_bytes": 64 * 1024,
+        "wall_timeout_ns": 60_000_000_000,
+        "memory_max_bytes": 1024 * 1024 * 1024,
+        "pids_max": 1,
+    }
+    values.update(changes)
+    return executor.ExecutionLimitsV1(**values)
 
 
 def _request(**changes: object) -> pipeline.PipelineRequestV1:
@@ -581,6 +583,10 @@ def test_admission_uses_only_explicit_cross_module_verification_api(self) -> Non
         ):
             with self.subTest(forbidden=forbidden):
                 self.assertNotIn(forbidden, source)
+        self.assertTrue(callable(executor.invocation_identity_v1))
+        self.assertTrue(callable(executor.platform_identity_v1))
+        self.assertFalse(hasattr(pipeline, "invocation_identity_v1"))
+        self.assertFalse(hasattr(pipeline, "platform_identity_v1"))
 
     def test_host_trust_claims_only_backend_observable_facts(self) -> None:
         trust = pipeline.HostTrustBoundaryV1.UNSEALED_LINUX_X64_DOCKER_HOST
@@ -754,10 +760,7 @@ def test_input_transport_or_invalid_binary_is_typed_failure(self) -> None:
     def test_job_that_exceeds_exact_run_limits_is_rejected_before_build(self) -> None:
         with self.assertRaises(pipeline.PipelineInputErrorV1) as caught:
             _request(
-                execution_limits=replace(
-                    _limits(),
-                    max_stdin_bytes=1,
-                )
+                execution_limits=_limits(max_stdin_bytes=1)
             )
 
         self.assertEqual(
@@ -768,8 +771,7 @@ def test_job_that_exceeds_exact_run_limits_is_rejected_before_build(self) -> Non
     def test_build_output_limit_is_rejected_at_pipeline_admission(self) -> None:
         with self.assertRaises(pipeline.PipelineInputErrorV1) as caught:
             _request(
-                execution_limits=replace(
-                    _limits(),
+                execution_limits=_limits(
                     max_executable_bytes=pipeline.BUILD_STDOUT_LIMIT_V1 + 1,
                 )
             )