feat(sandbox): add bounded memory profiles

CHANGELOG.md

@@ -3,6 +3,14 @@
 All notable changes to Echo Certification Forge are documented here. Versions follow Semantic
 Versioning; dates use ISO 8601.
 
+## [Unreleased]
+
+### Changed
+
+- Allow operators to select a validated 128 MiB to 4 GiB journey cgroup while preserving no-swap
+  containment, and record the effective memory, CPU, PID, and scratch limits in journey evidence.
+- Bind the exact sandbox image digest and resource profile into the certification environment identity.
+
 ## [1.1.0] - 2026-08-09
 
 ### Added

docs/OPERATIONS.md

@@ -18,6 +18,13 @@
 6. Promote atomically, health-check production, and roll back on any mismatch.
 7. Publish immutable machine certificates and the repository certificate graphic.
 
+The execution cgroup remains mandatory. `ECHO_CERTFORGE_SANDBOX_MEMORY` (or
+`--sandbox-memory`) may select a whole-MiB/GiB limit from `128m` through `4g`; the default is
+`512m`. CertForge rejects malformed, lower, higher, or unbounded values, applies the same value to
+memory and memory-swap, and records the effective resource profile in critical-journey evidence.
+The pinned sandbox image digest and resource profile are also incorporated into the authoritative
+certification environment identity; changing either invalidates reuse of the prior environment digest.
+
 ## Incident triage
 
 Capture the run ID, target SHA, environment digest, policy version, service health, dispatcher state,

src/echo_certification_forge/app.py

@@ -10,6 +10,7 @@
 from .runner import TrustedTransportRegistry
 from .adapters import adapter_set_digest
 from .run_worker import _worker_environment, load_adapter_execution_profile
+from .sandbox import DEFAULT_IMAGE, DEFAULT_MEMORY, DockerSandbox
 from .service import ServiceContext, create_app
 from .signing import TrustedPublicKeyRegistry
 from .subscriber import SubscriberGovernance, SubscriberPolicy
@@ -82,7 +83,11 @@ def _load_certification_environment() -> dict[str, str] | None:
         **{name: Path(value) for name, value in required.items() if value is not None}
     )
     adapter_sha256 = adapter_set_digest(records)
-    environment = _worker_environment(adapter_sha256, profile_sha256)
+    sandbox = DockerSandbox(
+        image=os.environ.get("ECHO_CERTFORGE_SANDBOX_IMAGE", DEFAULT_IMAGE),
+        memory=os.environ.get("ECHO_CERTFORGE_SANDBOX_MEMORY", DEFAULT_MEMORY),
+    )
+    environment = _worker_environment(adapter_sha256, profile_sha256, sandbox)
     return {
         "certification_environment_identity_digest": environment.identity_digest,
         "runner_image_digest": "sha256:" + environment.runner_image_sha256,

src/echo_certification_forge/dispatch_worker.py

@@ -12,7 +12,7 @@
 from .intake import SubmitRequest
 from .policy import RuleManifest
 from .run_worker import _load_adapter_inputs, _load_signer, run
-from .sandbox import DEFAULT_IMAGE, DockerSandbox
+from .sandbox import DEFAULT_IMAGE, DEFAULT_MEMORY, DockerSandbox, normalize_memory_limit
 from .subscriber import SubscriberDispatch, SubscriberError, SubscriberGovernance, SubscriberPolicy
 
 _REPO = Path(__file__).resolve().parents[2]
@@ -187,6 +187,12 @@ def main(argv: list[str] | None = None) -> int:
         "--sandbox-image",
         default=os.environ.get("ECHO_CERTFORGE_SANDBOX_IMAGE", DEFAULT_IMAGE),
     )
+    parser.add_argument(
+        "--sandbox-memory",
+        type=normalize_memory_limit,
+        default=os.environ.get("ECHO_CERTFORGE_SANDBOX_MEMORY", DEFAULT_MEMORY),
+        help="bounded container memory limit (128m through 4g)",
+    )
     parser.add_argument(
         "--sandbox-docker",
         default=os.environ.get("ECHO_CERTFORGE_SANDBOX_DOCKER", "docker"),
@@ -273,6 +279,7 @@ def main(argv: list[str] | None = None) -> int:
     sandbox = (
         DockerSandbox(
             image=args.sandbox_image,
+            memory=args.sandbox_memory,
             docker=tuple(args.sandbox_docker.split()),
         )
         if args.sandbox

src/echo_certification_forge/run_worker.py

@@ -39,7 +39,13 @@
 from .models import EnvironmentIdentity, RunOutcome, RunState, TargetIdentity
 from .policy import RuleManifest
 from .runner import RunnerResponse
-from .sandbox import DEFAULT_IMAGE, DockerSandbox, sandboxed_journey_runner
+from .sandbox import (
+    DEFAULT_IMAGE,
+    DEFAULT_MEMORY,
+    DockerSandbox,
+    normalize_memory_limit,
+    sandboxed_journey_runner,
+)
 from .signing import Ed25519VerdictSigner
 from .subscriber import SubscriberError, SubscriberGovernance, SubscriberPolicy
 
@@ -99,18 +105,28 @@ def _env_digest(component: str) -> str:
 def _worker_environment(
     adapter_set_sha256: str | None = None,
     adapter_execution_profile_sha256: str | None = None,
+    sandbox: DockerSandbox | None = None,
 ) -> EnvironmentIdentity:
     """Declared certification environment.
 
     The legacy v1 path retains its historical environment commitment. P5/v2 callers pass the exact
     digest derived from the verified signed adapter execution records.
     """
+    runner_image_sha256 = sandbox.image_sha256() if sandbox is not None else _env_digest("runner-image")
+    harness_sha256 = _env_digest("harness")
+    if sandbox is not None:
+        harness_sha256 = sha256_json(
+            {
+                "base_harness_sha256": harness_sha256,
+                "sandbox_resource_limits": sandbox.resource_limits(),
+            }
+        )
     return EnvironmentIdentity(
-        runner_image_sha256=_env_digest("runner-image"),
+        runner_image_sha256=runner_image_sha256,
         adapter_set_sha256=adapter_set_sha256 or _env_digest("adapter-set"),
         test_plan_sha256=_env_digest("test-plan"),
         policy_sha256=_env_digest("policy"),
-        harness_sha256=_env_digest("harness"),
+        harness_sha256=harness_sha256,
         prompt_set_sha256=_env_digest("prompt-set"),
         model_route_sha256=(
             sha256_json(
@@ -347,7 +363,11 @@ def run(
         if adapter_bundle_response is not None
         else None
     )
-    environment = _worker_environment(adapter_digest, adapter_execution_profile_sha256)
+    environment = _worker_environment(
+        adapter_digest,
+        adapter_execution_profile_sha256,
+        sandbox_effective,
+    )
     if subscribers is not None:
         if existing is None or existing["state"] != RunState.QUEUED.value:
             if claim is not None:
@@ -720,6 +740,12 @@ def main(argv: list[str] | None = None) -> int:
         "--sandbox-image",
         default=os.environ.get("ECHO_CERTFORGE_SANDBOX_IMAGE", DEFAULT_IMAGE),
     )
+    parser.add_argument(
+        "--sandbox-memory",
+        type=normalize_memory_limit,
+        default=os.environ.get("ECHO_CERTFORGE_SANDBOX_MEMORY", DEFAULT_MEMORY),
+        help="bounded container memory limit (128m through 4g)",
+    )
     parser.add_argument(
         "--sandbox-docker",
         default=os.environ.get("ECHO_CERTFORGE_SANDBOX_DOCKER", "docker"),
@@ -879,6 +905,7 @@ def main(argv: list[str] | None = None) -> int:
     if args.sandbox:
         sandbox = DockerSandbox(
             image=args.sandbox_image,
+            memory=args.sandbox_memory,
             docker=tuple(args.sandbox_docker.split()),
         )
 

src/echo_certification_forge/sandbox.py

@@ -18,6 +18,7 @@
 """
 from __future__ import annotations
 
+import re
 import subprocess
 import time
 from dataclasses import dataclass, field
@@ -26,12 +27,32 @@
 
 # Pinned minimal Python base (same digest the P4 supply-chain pipeline pins). Override per policy.
 DEFAULT_IMAGE = "python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df"
+DEFAULT_MEMORY = "512m"
+MIN_MEMORY_MIB = 128
+MAX_MEMORY_MIB = 4096
+_MEMORY_LIMIT = re.compile(r"^([1-9][0-9]*)([mMgG])$")
+_PINNED_IMAGE = re.compile(r"(?:^|@)sha256:([0-9a-f]{64})$")
 
 
 class SandboxError(RuntimeError):
     """The sandbox runtime is unavailable or failed to launch (distinct from a target-code failure)."""
 
 
+def normalize_memory_limit(value: str) -> str:
+    """Validate and normalize a bounded Docker memory limit."""
+    match = _MEMORY_LIMIT.fullmatch(value.strip())
+    if match is None:
+        raise ValueError("sandbox memory must be a whole number followed by m or g")
+    amount = int(match.group(1))
+    unit = match.group(2).lower()
+    memory_mib = amount if unit == "m" else amount * 1024
+    if not MIN_MEMORY_MIB <= memory_mib <= MAX_MEMORY_MIB:
+        raise ValueError(
+            f"sandbox memory must be between {MIN_MEMORY_MIB}m and {MAX_MEMORY_MIB // 1024}g"
+        )
+    return f"{amount}{unit}"
+
+
 @dataclass(frozen=True, slots=True)
 class SandboxResult:
     returncode: int
@@ -43,7 +64,7 @@ class SandboxResult:
 @dataclass(frozen=True, slots=True)
 class DockerSandbox:
     image: str = DEFAULT_IMAGE
-    memory: str = "512m"
+    memory: str = DEFAULT_MEMORY
     cpus: str = "1.0"
     pids_limit: int = 128
     tmpfs_size: str = "64m"
@@ -52,15 +73,35 @@ class DockerSandbox:
     docker: tuple[str, ...] = ("docker",)
     extra_env: dict[str, str] = field(default_factory=dict)
 
+    def image_sha256(self) -> str:
+        match = _PINNED_IMAGE.search(self.image)
+        if match is None:
+            raise SandboxError("sandbox image must be pinned by sha256 digest")
+        return match.group(1)
+
+    def resource_limits(self) -> dict[str, str | int]:
+        try:
+            memory = normalize_memory_limit(self.memory)
+        except ValueError as exc:
+            raise SandboxError(str(exc)) from exc
+        return {
+            "memory": memory,
+            "cpus": self.cpus,
+            "pids": self.pids_limit,
+            "tmpfs": self.tmpfs_size,
+        }
+
     def build_command(self, argv: list[str], workdir: Path) -> list[str]:
         """Construct the fully-hardened `docker run` argv. Pure — no side effects, unit-testable."""
         if not argv:
             raise SandboxError("empty journey argv")
+        self.image_sha256()
+        memory = str(self.resource_limits()["memory"])
         cmd: list[str] = [
             *self.docker, "run", "--rm",
             "--network", "none",
-            "--memory", self.memory,
-            "--memory-swap", self.memory,          # no swap escape past the memory cap
+            "--memory", memory,
+            "--memory-swap", memory,               # no swap escape past the memory cap
             "--cpus", self.cpus,
             "--pids-limit", str(self.pids_limit),
             "--read-only",
@@ -139,6 +180,7 @@ def _run(argv: list[str], workdir: Path) -> tuple[bool, dict]:
             return False, {"executed": True, "isolation": "docker", "error": f"sandbox_unavailable:{exc}"}
         return result.returncode == 0, {
             "executed": True, "isolation": "docker", "image": sandbox.image,
+            "resource_limits": sandbox.resource_limits(),
             "argv": argv, "returncode": result.returncode, "timed_out": result.timed_out,
             "stdout_tail": result.stdout[-2000:], "stderr_tail": result.stderr[-2000:],
         }

tests/test_p5_adapter_execution.py

@@ -39,6 +39,7 @@
 from echo_certification_forge.family_r5 import execute as execute_r5
 from echo_certification_forge.evidence import merkle_root
 from echo_certification_forge.runner import RunnerEphemeralIdentity
+from echo_certification_forge.sandbox import DockerSandbox
 from echo_certification_forge.run_worker import (
     _load_adapter_inputs,
     load_adapter_execution_profile,
@@ -720,6 +721,7 @@ def test_production_router_arguments_rebind_bundle_and_reach_worker_execution(
     expected_environment = _worker_environment(
         adapter_set_digest(records),
         result["adapter_execution_profile_sha256"],
+        DockerSandbox(),
     )
     assert result["environment_identity_digest"] == expected_environment.identity_digest
     store = EvidenceStore(db_path, evidence_root)

tests/test_p6_platform_integration.py

@@ -1831,6 +1831,7 @@ def test_registry_webhook_oci_run_certifies_and_deploys_end_to_end(
     try:
         manifest_digest = _push_test_image(registry)
         repo = f"http://127.0.0.1:{registry.port}/testapp"
+        exact_image = f"127.0.0.1:{registry.port}/testapp@{manifest_digest}"
         event = {
             "event_id": "evt-oci-0001",
             "event_type": "registry.image.pushed",
@@ -1839,7 +1840,9 @@ def test_registry_webhook_oci_run_certifies_and_deploys_end_to_end(
             "image_repository": repo,
             "source_commit": "abc123def456",
             # the platform declares the WORKER's environment commitment for the run
-            "environment_identity_digest": run_worker._worker_environment().identity_digest,
+            "environment_identity_digest": run_worker._worker_environment(
+                sandbox=DockerSandbox(image=exact_image)
+            ).identity_digest,
             "policy_version": manifest.manifest_id,
         }
         body = json.dumps(event).encode("utf-8")

tests/test_t4p8_sandbox.py

@@ -11,8 +11,14 @@
 import pytest
 
 from echo_certification_forge.executor import RunExecutor, StaticEntitlement
+from echo_certification_forge.run_worker import _worker_environment
 from echo_certification_forge.sandbox import (
-    DEFAULT_IMAGE, DockerSandbox, SandboxError, sandboxed_journey_runner,
+    DEFAULT_IMAGE,
+    DockerSandbox,
+    SandboxError,
+    SandboxResult,
+    normalize_memory_limit,
+    sandboxed_journey_runner,
 )
 from echo_certification_forge.signing import Ed25519VerdictSigner
 
@@ -41,6 +47,76 @@ def test_empty_argv_rejected(tmp_path):
         DockerSandbox().build_command([], tmp_path)
 
 
+@pytest.mark.parametrize(
+    ("raw", "normalized"),
+    (("128m", "128m"), ("1024M", "1024m"), ("1G", "1g"), ("4g", "4g")),
+)
+def test_memory_limit_is_bounded_and_normalized(raw, normalized):
+    assert normalize_memory_limit(raw) == normalized
+
+
+@pytest.mark.parametrize("raw", ("", "0m", "127m", "5g", "512", "1.5g", "unlimited"))
+def test_memory_limit_rejects_unbounded_or_malformed_values(raw):
+    with pytest.raises(ValueError, match="sandbox memory"):
+        normalize_memory_limit(raw)
+
+
+def test_custom_memory_limit_stays_cgroup_bounded(tmp_path):
+    cmd = DockerSandbox(memory="1G").build_command(["python3", "hello.py"], tmp_path)
+    joined = " ".join(cmd)
+    assert "--memory 1g" in joined
+    assert "--memory-swap 1g" in joined
+
+
+def test_environment_identity_binds_pinned_image_and_resource_profile():
+    image_a = "example.invalid/runtime@sha256:" + ("a" * 64)
+    image_b = "example.invalid/runtime@sha256:" + ("b" * 64)
+    baseline = _worker_environment(sandbox=DockerSandbox(image=image_a, memory="1g"))
+    different_image = _worker_environment(
+        sandbox=DockerSandbox(image=image_b, memory="1g")
+    )
+    different_memory = _worker_environment(
+        sandbox=DockerSandbox(image=image_a, memory="2g")
+    )
+    assert baseline.runner_image_sha256 == "a" * 64
+    assert baseline.identity_digest != different_image.identity_digest
+    assert baseline.identity_digest != different_memory.identity_digest
+
+
+def test_unpinned_image_is_rejected_before_execution(tmp_path):
+    with pytest.raises(SandboxError, match="pinned by sha256"):
+        DockerSandbox(image="python:latest").build_command(["python3", "hello.py"], tmp_path)
+
+
+def test_effective_resource_limits_are_recorded_in_journey_evidence(tmp_path):
+    class StubSandbox:
+        image = "example.invalid/runtime@sha256:" + ("a" * 64)
+        memory = "1G"
+        cpus = "1.5"
+        pids_limit = 96
+        tmpfs_size = "80m"
+
+        @staticmethod
+        def run(argv, workdir, execution_guard=None):
+            return SandboxResult(0, "ok", "", False)
+
+        @staticmethod
+        def resource_limits():
+            return {"memory": "1g", "cpus": "1.5", "pids": 96, "tmpfs": "80m"}
+
+    passed, detail = sandboxed_journey_runner(StubSandbox())(
+        ["python3", "hello.py"],
+        tmp_path,
+    )
+    assert passed is True
+    assert detail["resource_limits"] == {
+        "memory": "1g",
+        "cpus": "1.5",
+        "pids": 96,
+        "tmpfs": "80m",
+    }
+
+
 def test_unavailable_runtime_is_a_harness_failure_not_a_pass(tmp_path):
     # a bogus docker binary -> SandboxError -> runner reports passed=False (never a silent pass)
     runner = sandboxed_journey_runner(DockerSandbox(docker=("definitely-not-docker-xyz",)))