Repair Eggcracker fail-closed lifecycle for 0.1.1

.github/pull_request_template.md

@@ -1,6 +1,6 @@
 ## Summary
 
-Describe the user-visible change and why it belongs in Lumi Nutcracker's narrow kill-switch scope.
+Describe the user-visible change and why it belongs in Lumi Eggcracker's narrow kill-switch scope.
 
 ## Verification
 

.github/workflows/ci.yml

@@ -13,11 +13,11 @@ jobs:
   test-and-package:
     runs-on: ubuntu-24.04
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@v6
         with:
           fetch-depth: 0
 
-      - uses: actions/setup-python@v5
+      - uses: actions/setup-python@v6
         with:
           python-version: "3.11"
 
@@ -38,19 +38,19 @@ jobs:
       - name: Verify public release
         run: >-
           python scripts/verify_release.py
-          --artifact dist/github/lumi-nutcracker-0.1.0.pyz
-          --source-archive dist/github/lumi-nutcracker-0.1.0-source.zip
-          --release-bundle dist/github/lumi-nutcracker-0.1.0-linux.zip
+          --artifact dist/github/lumi-eggcracker-0.1.1.pyz
+          --source-archive dist/github/lumi-eggcracker-0.1.1-source.zip
+          --release-bundle dist/github/lumi-eggcracker-0.1.1-linux.zip
 
       - name: Upload tag artifacts
         if: startsWith(github.ref, 'refs/tags/v')
-        uses: actions/upload-artifact@v4
+        uses: actions/upload-artifact@v6
         with:
-          name: lumi-nutcracker-${{ github.ref_name }}
+          name: lumi-eggcracker-${{ github.ref_name }}
           path: |
-            dist/github/lumi-nutcracker-0.1.0.pyz
-            dist/github/lumi-nutcracker-0.1.0-source.zip
-            dist/github/lumi-nutcracker-0.1.0-linux.zip
+            dist/github/lumi-eggcracker-0.1.1.pyz
+            dist/github/lumi-eggcracker-0.1.1-source.zip
+            dist/github/lumi-eggcracker-0.1.1-linux.zip
             dist/github/release-manifest.json
             dist/github/SHA256SUMS
           if-no-files-found: error

AGENTS.md

@@ -1,10 +0,0 @@
-# Lumi Nutcracker repository rules
-
-- This product supports only explicitly launched workloads on Linux with systemd and unified cgroup v2.
-- Tests may affect only fresh Nutcracker-owned units and fixture processes they create. Never signal PID 1, the test runner, its ancestors, a pre-existing process or an unrelated cgroup.
-- The root supervisor is the sole enforcement component. The workload identity must never gain control-socket access or root authority.
-- Direct `cgroup.kill` is authoritative. Do not replace it with process ancestry, compatibility backends or service-stop-only proof.
-- Keep a trigger-side path free of durable state, evidence, hash, model and networking operations until direct containment has been attempted.
-- Do not add network isolation, behavioural models, AI identification, BPF, cloud services or unsupported-platform claims without a separately approved evidence boundary.
-- Preserve unrelated user changes. Keep generated artifacts, models, VM images, keys and local evidence out of source control.
-- A partial or ambiguous result must never be reported as a successful termination.

CHANGELOG.md

@@ -1,13 +1,10 @@
 # Changelog
 
-All notable public changes are recorded here.
+## 0.1.1
+
+- Renamed the public product to Lumi Eggcracker.
+- Repaired launch, watcher, lifecycle, receipt, identity, and release hygiene paths.
 
 ## 0.1.0
 
-- Initial Lumi Nutcracker engineering preview.
-- Root-supervised launch of explicitly selected workloads in owned cgroup-v2 units.
-- Operator-triggered complete workload-tree termination using `cgroup.kill`.
-- Automatic PID-limit tripwire containment.
-- Dedicated workload/operator identity separation and control-socket authentication.
-- Post-containment receipts, status, list, doctor and version commands.
-- Clean installer, uninstaller, native qualification matrix and local AI smoke demonstration.
+- Retained public engineering-preview predecessor; not rewritten by this release.

CONTRIBUTING.md

@@ -1,19 +1,5 @@
 # Contributing
 
-Lumi Nutcracker is intentionally narrow. Changes should preserve the supported release claim and keep direct cgroup containment as the first trigger-side effect.
+Keep the product claim narrow and mechanically testable. Direct `cgroup.kill` plus exact empty proof remains authoritative; do not add unsupported backends, automatic AI recognition, behavioural models, networking, or host-wide claims without a separately qualified release boundary.
 
-## Local checks
-
-```bash
-PYTHONPATH=src python3 -m unittest discover -s tests -v
-ruff check src tests scripts
-python3 scripts/build_release.py --output dist
-python3 scripts/verify_release.py \
-  --artifact dist/lumi-nutcracker-0.1.0.pyz \
-  --source-archive dist/lumi-nutcracker-0.1.0-source.zip \
-  --release-bundle dist/lumi-nutcracker-0.1.0-linux.zip
-```
-
-Native tests create and terminate root-owned systemd cgroups. Run them only inside a disposable Linux VM you own, and never modify them to target pre-existing processes or cgroups.
-
-Open an issue before adding a new containment backend, network isolation, AI identification, behavioural models or a broader security claim.
+Run unit tests, lint, release construction, and public-artifact verification before proposing a change.

LIMITATIONS.md

@@ -1,9 +1,7 @@
 # Limitations
 
-Lumi Nutcracker protects only commands explicitly launched through its local root supervisor. It does not inspect the host for AI, agent, malware or intrusion activity, and it cannot attach an existing process to its protected boundary.
+Lumi Eggcracker protects only commands explicitly launched through its local root supervisor on the qualified Linux/systemd/cgroup-v2 environment.
 
-The product terminates the exact cgroup it created. It does not provide network isolation, credential isolation, filesystem isolation, container isolation, virtual-machine isolation or host isolation. It is not an EDR, antivirus or general malware-prevention product.
+It does not discover AI processes, attach existing processes, identify malware, detect unknown intrusions, isolate networks, credentials or filesystems, provide general malware prevention, or replace EDR.
 
-The PID tripwire is a cgroup process-count ceiling. It is not behavioural detection, a resource-governance system or AI attribution. A supervisor restart intentionally fail-closed terminates active owned workloads rather than continuing them without a live watcher.
-
-Qualification is limited to the published tested native Linux environment. Passing local tests does not establish universal Linux compatibility or production readiness for workloads outside that environment.
+The `start` command is non-interactive in 0.1.1: there is no terminal attachment, current-working-directory forwarding, timeout, memory/CPU limit UX, or output streaming. The supervisor supports one active protected workload globally. Terminal records are retained locally up to 128 entries.

QUALIFICATION.md

@@ -0,0 +1,11 @@
+# Qualification
+
+The 0.1.1 candidate is releaseable only when its exact commit passes the native Ubuntu matrix and produces the minimal evidence pack. Required gates include 100/100 fork-race workload-tree kills, 100/100 unrelated-canary survivals, zero workload control-socket accesses or replacement launches, PID-tripwire containment, benign near-limit completion, fail-closed supervisor restart recovery, and p95 trigger-to-empty latency below 500 ms.
+
+Run the matrix only on a disposable or dedicated Linux test host:
+
+```sh
+sudo python3 scripts/run_native_matrix.py --fork-race-repetitions 100 --benign-repetitions 50 --restart-repetitions 20 --socket-attempts 100 --output ./native-matrix.json
+```
+
+The evidence establishes behavior only for the tested environment and commit. It does not establish universal Linux compatibility or any unsupported security claim.

README.md

@@ -1,125 +1,51 @@
-# Lumi Nutcracker
+# Lumi Eggcracker
 
-**The local Linux kill switch for AI and agent workloads.**
+Lumi Eggcracker is a small, local Linux kill switch for an AI or agent workload you explicitly select.
 
-Lumi Nutcracker launches a command you explicitly select under a dedicated no-login identity and inside a root-controlled cgroup-v2 boundary. The operator can terminate the complete workload tree, or configure a PID ceiling that triggers automatic containment when the workload starts creating too many processes.
+> Lumi Eggcracker launches one explicitly selected AI or agent workload inside a root-controlled Linux cgroup and terminates the complete workload tree on operator command or when its configured PID-runaway tripwire fires.
 
-> Lumi Nutcracker launches an explicitly selected AI or agent workload inside a root-controlled Linux cgroup and terminates the complete workload tree on operator command or when its configured PID-runaway tripwire fires.
+It uses one root-owned supervisor, direct cgroup-v2 `cgroup.kill`, and an exact empty-cgroup proof. The workload runs under a dedicated no-login account that cannot access the control socket.
 
-This is a Linux engineering preview. Read [Limitations](LIMITATIONS.md) before relying on it.
+## It does
 
-## What it does
+- launch one selected command in an exact owned cgroup;
+- kill its complete cgroup tree on `eggcracker kill`;
+- kill it automatically when its PID limit trips;
+- write a post-containment receipt only after an exact empty proof;
+- provide `start`, `kill`, `status`, `list`, `doctor`, and `version`.
 
-- launches one selected command in a Nutcracker-owned cgroup;
-- runs that command as a dedicated unprivileged, no-login user;
-- kills the complete owned cgroup with the kernel's `cgroup.kill` primitive;
-- verifies that the cgroup and all descendants are empty before issuing a success receipt;
-- automatically contains workloads that hit their configured PID limit;
-- fails closed for active workloads when the supervisor restarts.
+## It does not do
 
-## What it does not do
-
-Lumi Nutcracker does not discover AI processes, attach to existing processes, identify malware, detect unknown intrusions, isolate networks or credentials, prevent general malware, or replace endpoint security.
+It does not identify AI automatically, find unknown processes, detect intrusions or malware, isolate networks or credentials, prevent general malware, replace EDR, or attach an existing process to its boundary.
 
 ## Requirements
 
-- Ubuntu 24.04 or a comparable Linux distribution using systemd;
-- unified cgroup v2 with `cgroup.kill` and the PID controller;
-- Python 3.11 or later;
-- root access for installation;
-- a non-root local operator account.
-
-## Install a GitHub release
-
-Download and extract `lumi-nutcracker-0.1.0-linux.zip`, then run:
-
-```bash
-cd lumi-nutcracker-0.1.0
-sha256sum -c SHA256SUMS
-sudo python3 scripts/install.py \
-  --operator "$(id -un)" \
-  --artifact ./lumi-nutcracker-0.1.0.pyz
-nutcracker doctor
-```
-
-The installer creates the dedicated `lumi-nutcracker-workload` identity dynamically. It does not assume a particular operator UID.
-
-## Build from source
-
-```bash
-python3 scripts/build_release.py --output dist
-python3 scripts/verify_release.py \
-  --artifact dist/lumi-nutcracker-0.1.0.pyz \
-  --source-archive dist/lumi-nutcracker-0.1.0-source.zip \
-  --release-bundle dist/lumi-nutcracker-0.1.0-linux.zip
-sudo python3 scripts/install.py \
-  --operator "$(id -un)" \
-  --artifact dist/lumi-nutcracker-0.1.0.pyz
-```
-
-Release builds refuse a dirty Git tree so the embedded source commit remains meaningful.
+- Linux with systemd and unified cgroup v2;
+- `pids` controller and `cgroup.kill` on transient service cgroups;
+- root for install/uninstall; a non-root local operator account;
+- Python 3.11 or newer.
 
 ## Quick start
 
-```bash
-nutcracker start --name local-agent --max-pids 64 -- /usr/bin/python3 agent.py
-nutcracker status --name local-agent
-nutcracker kill --name local-agent --receipt ./kill-receipt.json
-```
-
-The receipt is written only after direct cgroup kill and empty-hierarchy verification succeed.
+Download and extract `lumi-eggcracker-0.1.1-linux.zip`, then run as root:
 
-## Commands
-
-```text
-nutcracker start --name NAME --max-pids LIMIT -- COMMAND [ARGS...]
-nutcracker kill --name NAME --receipt FILE
-nutcracker status --name NAME
-nutcracker list
-nutcracker doctor
-nutcracker version
-```
-
-## Local AI smoke demonstration
-
-With a locally installed `llama-cli` and a GGUF model readable by the workload identity:
-
-```bash
-python3 scripts/smoke_local_ai.py \
-  --llama-cli /opt/llama/llama-cli \
-  --model /opt/models/local.gguf \
-  --output ai-smoke.json
-```
-
-The script waits for visible model output, kills that selected inference workload through Nutcracker, and verifies that an unrelated canary survives.
-
-## Uninstall
-
-Terminate active protected workloads first, then run from the extracted release or source tree:
-
-```bash
+```sh
+cd lumi-eggcracker-0.1.1
+sudo python3 scripts/install.py --operator "$USER" --artifact ./lumi-eggcracker-0.1.1.pyz
+eggcracker doctor
+eggcracker start --name demo --max-pids 8 -- /bin/sleep 60
+eggcracker kill --name demo --receipt ./demo-receipt.json
 sudo python3 scripts/uninstall.py
-sudo python3 scripts/verify_uninstalled.py
 ```
 
-## Qualification
-
-The 0.1.0 native Ubuntu qualification completed:
+The selected command is launched through a systemd transient unit. It does not attach an interactive terminal, preserve your current working directory, or stream output to the CLI in this preview. Use absolute paths and redirect output in a wrapper script when needed.
 
-- 100/100 complete fork-race workload-tree kills;
-- 100/100 unrelated-canary survivals;
-- zero successful control-socket accesses from the workload identity;
-- zero successful replacement launches;
-- automatic PID-tripwire containment;
-- 50/50 bounded benign completions without a false kill;
-- 20/20 fail-closed supervisor restart recoveries;
-- measured p95 trigger-to-empty latency below 500 ms;
-- a real local llama.cpp workload smoke test;
-- clean installation and complete uninstallation;
-- no firewall-rule objects or dependencies.
+## Real local AI smoke
 
-See [Security](SECURITY.md), [Limitations](LIMITATIONS.md), and [Release notes](RELEASE_NOTES.md).
+With a local `llama-cli` and GGUF model available, the release bundle includes a smoke demonstration that waits for visible model output, kills the explicitly selected inference workload, and verifies an unrelated canary survives:
 
-## Licence
+```sh
+sudo python3 scripts/smoke_local_ai.py --llama-cli /path/to/llama-cli --model /path/to/model.gguf --output ./ai-smoke.json
+```
 
-Apache-2.0. See [LICENSE](LICENSE).
+See [SECURITY_MODEL.md](SECURITY_MODEL.md), [LIMITATIONS.md](LIMITATIONS.md), and [QUALIFICATION.md](QUALIFICATION.md) before using the preview.

RELEASE_NOTES.md

@@ -1,23 +1,5 @@
-# Lumi Nutcracker 0.1.0
+# Lumi Eggcracker 0.1.1
 
-Lumi Nutcracker 0.1.0 is a Linux engineering preview of a deliberately small AI and agent workload kill switch.
+0.1.1 is a corrective engineering-preview candidate. It introduces a fail-closed launch gate, cgroup-event lifecycle authority, direct cgroup-kill-first enforcement, receipt-before-cleanup ordering, bounded run records, stricter workload-account isolation, and a consistent Lumi Eggcracker public surface.
 
-## Supported claim
-
-> Lumi Nutcracker launches an explicitly selected AI or agent workload inside a root-controlled Linux cgroup and terminates the complete workload tree on operator command or when its configured PID-runaway tripwire fires.
-
-## Highlights
-
-- One supported root-supervisor backend on systemd and unified cgroup v2.
-- Dedicated no-login workload identity separated from the human operator.
-- Direct `cgroup.kill` enforcement followed by exact empty-hierarchy verification.
-- Operator kill, automatic PID tripwire, post-containment receipts and restart fail-closed recovery.
-- Public commands: `start`, `kill`, `status`, `list`, `doctor` and `version`.
-
-## Qualification
-
-The release passed 100/100 fork-race kills, 100/100 unrelated-canary survivals, 100 denied workload socket attempts, zero replacement launches, five automatic PID-tripwire trials, 50 benign completions, 20 supervisor restart recoveries, clean install/uninstall and a real local AI smoke demonstration. Native p95 trigger-to-empty latency was below the precommitted 500 ms ceiling.
-
-## Important limits
-
-This release does not identify AI, discover unknown processes, attach to existing workloads, isolate network or credentials, detect malware, or replace EDR. See [LIMITATIONS.md](LIMITATIONS.md).
+It is not published automatically by this repository. Native Ubuntu qualification and explicit release approval are required before tag or promotion.

SECURITY.md

@@ -1,5 +1,7 @@
 # Security policy
 
-Report security issues privately through GitHub's **Security → Report a vulnerability** flow when it is enabled for the repository. Otherwise, contact the project maintainer privately. Do not open a public issue containing exploit details, live credentials, private model files or VM keys.
+Lumi Eggcracker is a privileged local process-containment preview. Test only workloads and machines you own or administer.
 
-Lumi Nutcracker is a local privileged process-containment tool. Test only workloads and machines you own or administer. Do not use it against third-party processes or systems.
+Report vulnerabilities privately through the repository security-advisory channel. Do not include exploit code, machine-specific paths, credentials, or live workload details in public issues.
+
+The security boundary and non-claims are documented in [SECURITY_MODEL.md](SECURITY_MODEL.md).

SECURITY_MODEL.md

@@ -0,0 +1,7 @@
+# Security model
+
+The root-owned `lumi-eggcracker.service` is the only enforcement component. It accepts requests only from root or the configured operator UID over a root-owned Unix socket. A selected workload runs as a dedicated no-login unprivileged user with no supplementary groups; that identity cannot read, connect to, or invoke the socket.
+
+Each workload is placed in an exactly identified systemd transient cgroup. On an operator request, PID-limit tripwire, watcher failure, or supervisor restart recovery, the supervisor validates that exact cgroup identity and writes `1` to its `cgroup.kill` control file. It then proves that the complete cgroup hierarchy is empty. Systemd stop is cleanup only.
+
+This is not a general sandbox, host intrusion detector, malware detector, network isolator, credential isolator, or EDR replacement. The selected workload may still access everything normally available to its unprivileged account.

pyproject.toml

@@ -3,8 +3,8 @@ requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "lumi-nutcracker"
-version = "0.1.0"
+name = "lumi-eggcracker"
+version = "0.1.1"
 description = "A local Linux kill switch for explicitly selected AI and agent workloads"
 readme = "README.md"
 requires-python = ">=3.11"
@@ -22,7 +22,7 @@ classifiers = [
 ]
 
 [project.scripts]
-nutcracker = "lumi_nutcracker.cli:main"
+eggcracker = "lumi_eggcracker.cli:main"
 
 [tool.setuptools]
 package-dir = {"" = "src"}

scripts/build_release.py

@@ -1,4 +1,4 @@
-"""Build a clean Lumi Nutcracker zipapp, source archive, and checksums."""
+"""Build a clean Lumi Eggcracker zipapp, source archive, and checksums."""
 
 from __future__ import annotations
 
@@ -24,7 +24,7 @@ def digest(path: Path) -> str:
 
 
 def version() -> str:
-    value = (ROOT / "src" / "lumi_nutcracker" / "__init__.py").read_text(encoding="utf-8")
+    value = (ROOT / "src" / "lumi_eggcracker" / "__init__.py").read_text(encoding="utf-8")
     match = re.search(r'__version__ = "([0-9]+\.[0-9]+\.[0-9]+)"', value)
     if not match:
         raise RuntimeError("cannot determine public version")
@@ -48,20 +48,20 @@ def main() -> int:
     release_version = version()
     commit = command(["git", "-C", str(ROOT), "rev-parse", "HEAD"])
     args.output.mkdir(parents=True, exist_ok=True)
-    artifact = args.output / f"lumi-nutcracker-{release_version}.pyz"
-    source = args.output / f"lumi-nutcracker-{release_version}-source.zip"
-    bundle = args.output / f"lumi-nutcracker-{release_version}-linux.zip"
-    with tempfile.TemporaryDirectory(prefix="lumi-nutcracker-build-") as raw:
+    artifact = args.output / f"lumi-eggcracker-{release_version}.pyz"
+    source = args.output / f"lumi-eggcracker-{release_version}-source.zip"
+    bundle = args.output / f"lumi-eggcracker-{release_version}-linux.zip"
+    with tempfile.TemporaryDirectory(prefix="lumi-eggcracker-build-") as raw:
         stage = Path(raw) / "src"
-        shutil.copytree(ROOT / "src", stage, ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
-        (stage / "__main__.py").write_text("from lumi_nutcracker.cli import main\nraise SystemExit(main())\n", encoding="utf-8")
-        (stage / "lumi_nutcracker" / "build_info.py").write_text(f'SOURCE_COMMIT = "{commit}"\n', encoding="utf-8")
+        shutil.copytree(ROOT / "src", stage, ignore=shutil.ignore_patterns("__pycache__", "*.pyc", "*.egg-info"))
+        (stage / "__main__.py").write_text("from lumi_eggcracker.cli import main\nraise SystemExit(main())\n", encoding="utf-8")
+        (stage / "lumi_eggcracker" / "build_info.py").write_text(f'SOURCE_COMMIT = "{commit}"\n', encoding="utf-8")
         zipapp.create_archive(stage, target=artifact, interpreter="/usr/bin/env python3")
-    command(["git", "-C", str(ROOT), "archive", "--format=zip", f"--prefix=lumi-nutcracker-{release_version}/", "-o", str(source), "HEAD"])
+    command(["git", "-C", str(ROOT), "archive", "--format=zip", f"--prefix=lumi-eggcracker-{release_version}/", "-o", str(source), "HEAD"])
     manifest = {"artifact": artifact.name, "sha256": digest(artifact), "source_archive": source.name, "source_archive_sha256": digest(source), "source_commit": commit, "version": release_version}
     (args.output / "release-manifest.json").write_text(__import__("json").dumps(manifest, sort_keys=True) + "\n", encoding="utf-8")
     (args.output / "SHA256SUMS").write_text(f"{manifest['sha256']}  {artifact.name}\n{manifest['source_archive_sha256']}  {source.name}\n", encoding="utf-8")
-    prefix = f"lumi-nutcracker-{release_version}"
+    prefix = f"lumi-eggcracker-{release_version}"
     release_files = {
         artifact: artifact.name,
         source: source.name,
@@ -71,6 +71,8 @@ def main() -> int:
         ROOT / "LICENSE": "LICENSE",
         ROOT / "LIMITATIONS.md": "LIMITATIONS.md",
         ROOT / "SECURITY.md": "SECURITY.md",
+        ROOT / "SECURITY_MODEL.md": "SECURITY_MODEL.md",
+        ROOT / "QUALIFICATION.md": "QUALIFICATION.md",
         ROOT / "RELEASE_NOTES.md": "RELEASE_NOTES.md",
         ROOT / "scripts" / "install.py": "scripts/install.py",
         ROOT / "scripts" / "uninstall.py": "scripts/uninstall.py",
@@ -80,6 +82,12 @@ def main() -> int:
     with zipfile.ZipFile(bundle, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive:
         for path, name in release_files.items():
             archive.write(path, f"{prefix}/{name}")
+    # The bundle cannot contain a checksum of itself without recursion.  The
+    # detached checksum file intentionally covers all published binary assets.
+    (args.output / "SHA256SUMS").write_text(
+        f"{manifest['sha256']}  {artifact.name}\n{manifest['source_archive_sha256']}  {source.name}\n{digest(bundle)}  {bundle.name}\n",
+        encoding="utf-8",
+    )
     print(bundle)
     return 0
 

scripts/install.py

@@ -1,4 +1,4 @@
-"""Manifest-bound installer for one local root supervisor and workload account."""
+"""Manifest-bound installer for Lumi Eggcracker's root supervisor."""
 
 from __future__ import annotations
 
@@ -16,14 +16,15 @@ import time
 from pathlib import Path
 from typing import Any
 
-LIB = Path("/usr/local/lib/lumi-nutcracker")
-BIN = Path("/usr/local/bin/nutcracker")
-ETC = Path("/etc/lumi-nutcracker")
-UNIT = Path("/etc/systemd/system/lumi-nutcracker.service")
-STATE = Path("/var/lib/lumi-nutcracker")
-RUNTIME = Path("/run/lumi-nutcracker")
+VERSION = "0.1.1"
+LIB = Path("/usr/local/lib/lumi-eggcracker")
+BIN = Path("/usr/local/bin/eggcracker")
+ETC = Path("/etc/lumi-eggcracker")
+UNIT = Path("/etc/systemd/system/lumi-eggcracker.service")
+STATE = Path("/var/lib/lumi-eggcracker")
+RUNTIME = Path("/run/lumi-eggcracker")
 SOCKET = RUNTIME / "control.sock"
-WORKLOAD_NAME = "lumi-nutcracker-workload"
+WORKLOAD_NAME = "lumi-eggcracker-workload"
 TARGETS = (LIB, BIN, ETC, UNIT, STATE, RUNTIME)
 
 
@@ -54,7 +55,7 @@ def manifest_for(artifact: Path) -> dict[str, Any]:
     expected = {"artifact", "sha256", "source_archive", "source_archive_sha256", "source_commit", "version"}
     if set(value) != expected or value["artifact"] != artifact.name or value["sha256"] != digest(artifact):
         raise RuntimeError("release artifact identity does not match manifest")
-    if value["version"] != "0.1.0" or len(value["source_commit"]) != 40:
+    if value["version"] != VERSION or len(value["source_commit"]) != 40:
         raise RuntimeError("release manifest version or source identity is invalid")
     return value
 
@@ -63,7 +64,7 @@ def workload_account() -> tuple[pwd.struct_passwd, bool]:
     try:
         account = pwd.getpwnam(WORKLOAD_NAME)
         if account.pw_uid == 0 or account.pw_shell not in {"/usr/sbin/nologin", "/sbin/nologin"} or account.pw_dir != "/nonexistent":
-            raise RuntimeError("existing workload account does not meet Nutcracker contract")
+            raise RuntimeError("existing workload account does not meet Eggcracker contract")
         return account, False
     except KeyError:
         nologin = "/usr/sbin/nologin" if Path("/usr/sbin/nologin").is_file() else "/sbin/nologin"
@@ -74,12 +75,11 @@ def workload_account() -> tuple[pwd.struct_passwd, bool]:
 
 
 def service() -> bytes:
-    return b"""[Unit]\nDescription=Lumi Nutcracker protected workload supervisor\nAfter=multi-user.target\nStartLimitIntervalSec=60\nStartLimitBurst=30\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/lib/lumi-nutcracker/lumi-nutcracker.pyz _supervisor --policy /etc/lumi-nutcracker/policy.json\nRestart=always\nRestartSec=0.1\nRuntimeDirectory=lumi-nutcracker\nRuntimeDirectoryMode=0710\nUMask=0077\nNoNewPrivileges=yes\n\n[Install]\nWantedBy=multi-user.target\n"""
+    return b"""[Unit]\nDescription=Lumi Eggcracker protected workload supervisor\nAfter=multi-user.target\nStartLimitIntervalSec=60\nStartLimitBurst=60\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/lib/lumi-eggcracker/lumi-eggcracker.pyz _supervisor --policy /etc/lumi-eggcracker/policy.json\nRestart=always\nRestartSec=0.1\nRuntimeDirectory=lumi-eggcracker\nRuntimeDirectoryMode=0710\nUMask=0077\nNoNewPrivileges=yes\n\n[Install]\nWantedBy=multi-user.target\n"""
 
 
 def cgroup_kill_available() -> bool:
-    """Probe a disposable child cgroup; cgroup.kill is intentionally absent at root."""
-    unit = f"lumi-nutcracker-preflight-{secrets.token_hex(8)}.service"
+    unit = f"lumi-eggcracker-preflight-{secrets.token_hex(8)}.service"
     started = run(["/usr/bin/systemd-run", f"--unit={unit}", "--collect", "--property=Type=exec", "--", "/bin/sleep", "5"])
     if started.returncode:
         return False
@@ -87,14 +87,13 @@ def cgroup_kill_available() -> bool:
         shown = run(["/usr/bin/systemctl", "show", unit, "--property=ControlGroup"])
         values = dict(line.split("=", 1) for line in shown.stdout.splitlines() if "=" in line)
         cgroup = values.get("ControlGroup", "")
-        parts = cgroup.lstrip("/").split("/")
-        return bool(cgroup.startswith("/system.slice/")) and (Path("/sys/fs/cgroup").joinpath(*parts) / "cgroup.kill").is_file()
+        return bool(cgroup.startswith("/system.slice/")) and (Path("/sys/fs/cgroup").joinpath(*cgroup.lstrip("/").split("/")) / "cgroup.kill").is_file()
     finally:
         run(["/usr/bin/systemctl", "stop", unit])
 
 
-def cleanup(created: list[Path], created_user: bool, created_group: bool = False) -> None:
-    run(["/usr/bin/systemctl", "disable", "--now", "lumi-nutcracker.service"])
+def cleanup(created: list[Path], created_user: bool, created_group: bool) -> None:
+    run(["/usr/bin/systemctl", "disable", "--now", "lumi-eggcracker.service"])
     for path in reversed(created):
         if path.is_dir() and not path.is_symlink():
             shutil.rmtree(path)
@@ -121,39 +120,32 @@ def main() -> int:
     if operator.pw_uid == 0:
         raise SystemExit("operator must be non-root")
     if any(path.exists() or path.is_symlink() for path in TARGETS):
-        raise SystemExit("refusing pre-existing Nutcracker installation target")
-    if not Path("/sys/fs/cgroup/cgroup.controllers").is_file():
-        raise SystemExit("unified cgroup v2 is required")
-    if "pids" not in Path("/sys/fs/cgroup/cgroup.controllers").read_text(encoding="ascii").split():
-        raise SystemExit("cgroup PID controller is required")
-    if not cgroup_kill_available():
-        raise SystemExit("a transient system workload cgroup with cgroup.kill is required")
+        raise SystemExit("refusing pre-existing Eggcracker installation target")
+    controllers = Path("/sys/fs/cgroup/cgroup.controllers")
+    if not controllers.is_file() or "pids" not in controllers.read_text(encoding="ascii").split() or not cgroup_kill_available():
+        raise SystemExit("unified cgroup v2 with cgroup.kill and PID controller is required")
     account, created_user = workload_account()
     group = grp.getgrgid(account.pw_gid)
     created_group = created_user and group.gr_name == WORKLOAD_NAME
-    if account.pw_uid in {0, operator.pw_uid} or account.pw_gid == operator.pw_gid:
+    supplementary = set(os.getgrouplist(account.pw_name, account.pw_gid))
+    if account.pw_uid in {0, operator.pw_uid} or account.pw_gid == operator.pw_gid or supplementary != {account.pw_gid}:
         cleanup([], created_user, created_group)
-        raise SystemExit("workload identity must be separate from operator")
+        raise SystemExit("workload identity must be isolated from the operator and have no supplementary groups")
     created: list[Path] = []
     try:
-        LIB.mkdir(mode=0o755)
-        created.append(LIB)
-        ETC.mkdir(mode=0o700)
-        created.append(ETC)
-        STATE.mkdir(mode=0o700)
-        created.append(STATE)
-        shutil.copyfile(args.artifact, LIB / "lumi-nutcracker.pyz")
-        os.chmod(LIB / "lumi-nutcracker.pyz", 0o755)
-        write_new(BIN, b"#!/bin/sh\nexec /usr/bin/python3 /usr/local/lib/lumi-nutcracker/lumi-nutcracker.pyz \"$@\"\n", 0o755)
-        created.append(BIN)
-        policy = {"operator_gid": operator.pw_gid, "operator_uid": operator.pw_uid, "schema_version": "lumi-nutcracker.policy.v1", "socket_path": str(SOCKET), "source_commit": release["source_commit"], "state_dir": str(STATE), "unit_prefix": "lumi-nutcracker-workload-", "version": release["version"], "workload_gid": account.pw_gid, "workload_uid": account.pw_uid}
+        LIB.mkdir(mode=0o755); created.append(LIB)
+        ETC.mkdir(mode=0o700); created.append(ETC)
+        STATE.mkdir(mode=0o700); created.append(STATE)
+        shutil.copyfile(args.artifact, LIB / "lumi-eggcracker.pyz")
+        os.chmod(LIB / "lumi-eggcracker.pyz", 0o755)
+        write_new(BIN, b"#!/bin/sh\nexec /usr/bin/python3 /usr/local/lib/lumi-eggcracker/lumi-eggcracker.pyz \"$@\"\n", 0o755); created.append(BIN)
+        policy = {"operator_gid": operator.pw_gid, "operator_uid": operator.pw_uid, "schema_version": "lumi-eggcracker.policy.v2", "socket_path": str(SOCKET), "source_commit": release["source_commit"], "state_dir": str(STATE), "unit_prefix": "lumi-eggcracker-workload-", "version": release["version"], "workload_gid": account.pw_gid, "workload_uid": account.pw_uid}
         write_new(ETC / "policy.json", (json.dumps(policy, sort_keys=True) + "\n").encode(), 0o600)
-        write_new(UNIT, service(), 0o644)
-        created.append(UNIT)
-        manifest = {"created_workload_group": created_group, "created_workload_user": created_user, "files": {str(BIN): digest(BIN), str(ETC / "policy.json"): digest(ETC / "policy.json"), str(LIB / "lumi-nutcracker.pyz"): digest(LIB / "lumi-nutcracker.pyz"), str(UNIT): digest(UNIT)}, "operator": operator.pw_name, "operator_uid": operator.pw_uid, "schema_version": "lumi-nutcracker.install.v1", "targets": [str(path) for path in TARGETS], "workload_group": group.gr_name, "workload_uid": account.pw_uid, "workload_user": account.pw_name}
+        write_new(UNIT, service(), 0o644); created.append(UNIT)
+        manifest = {"created_workload_group": created_group, "created_workload_user": created_user, "files": {str(BIN): digest(BIN), str(ETC / "policy.json"): digest(ETC / "policy.json"), str(LIB / "lumi-eggcracker.pyz"): digest(LIB / "lumi-eggcracker.pyz"), str(UNIT): digest(UNIT)}, "operator": operator.pw_name, "operator_uid": operator.pw_uid, "schema_version": "lumi-eggcracker.install.v2", "targets": [str(path) for path in TARGETS], "workload_group": group.gr_name, "workload_uid": account.pw_uid, "workload_user": account.pw_name}
         write_new(STATE / "install-manifest.json", (json.dumps(manifest, sort_keys=True) + "\n").encode(), 0o600)
         checked = run(["/usr/bin/systemctl", "daemon-reload"])
-        started = run(["/usr/bin/systemctl", "enable", "--now", "lumi-nutcracker.service"])
+        started = run(["/usr/bin/systemctl", "enable", "--now", "lumi-eggcracker.service"])
         if checked.returncode or started.returncode:
             raise RuntimeError((checked.stderr + started.stderr).strip() or "cannot start supervisor")
         deadline = time.monotonic() + 15
@@ -162,7 +154,7 @@ def main() -> int:
         metadata = SOCKET.stat()
         if not SOCKET.exists() or stat.S_IMODE(metadata.st_mode) != 0o660 or metadata.st_uid != 0 or metadata.st_gid != operator.pw_gid:
             raise RuntimeError("control socket ownership contract failed")
-        print(json.dumps({"result": "INSTALLED", "service": "lumi-nutcracker.service", "workload_uid": account.pw_uid}, sort_keys=True))
+        print(json.dumps({"result": "INSTALLED", "service": "lumi-eggcracker.service", "workload_uid": account.pw_uid}, sort_keys=True))
         return 0
     except Exception:
         cleanup(created, created_user, created_group)

scripts/run_native_matrix.py

@@ -1,4 +1,4 @@
-"""Root-only native qualification for fresh Nutcracker-owned fixture workloads."""
+"""Root-only native qualification for fresh Eggcracker-owned fixture workloads."""
 
 from __future__ import annotations
 
@@ -14,7 +14,7 @@ from pathlib import Path
 from typing import Any
 
 ROOT = Path(__file__).resolve().parents[1]
-CLI = "/usr/local/bin/nutcracker"
+CLI = "/usr/local/bin/eggcracker"
 
 
 def run(argv: list[str], *, check: bool = True, timeout: int = 30) -> subprocess.CompletedProcess[str]:
@@ -77,7 +77,7 @@ def main() -> int:
         raise SystemExit("output must be a new file under an existing directory")
     if args.fork_race_repetitions != 100 or args.benign_repetitions < 50 or args.restart_repetitions < 20 or args.socket_attempts < 100:
         raise SystemExit("qualification counts are below the precommitted gates")
-    install = json.loads(Path("/var/lib/lumi-nutcracker/install-manifest.json").read_text(encoding="utf-8"))
+    install = json.loads(Path("/var/lib/lumi-eggcracker/install-manifest.json").read_text(encoding="utf-8"))
     operator = str(install["operator"])
     workload = str(install["workload_user"])
     workload_uid = pwd.getpwnam(workload).pw_uid
@@ -97,7 +97,7 @@ def main() -> int:
                 name = f"race-{token}-{index}"
                 call(operator, ["start", "--name", name, "--max-pids", "4096", "--", "/usr/bin/python3", str(ROOT / "tests/fixtures/fork_race.py"), modes[index % len(modes)]])
                 time.sleep(0.12)
-                receipt_path = Path("/tmp") / f"lumi-nutcracker-receipt-{token}-{index}.json"
+                receipt_path = Path("/tmp") / f"lumi-eggcracker-receipt-{token}-{index}.json"
                 receipt = call(operator, ["kill", "--name", name, "--receipt", str(receipt_path)])
                 if receipt.get("result") != "TERMINATED" or receipt["trigger"]["kind"] != "OPERATOR" or receipt["containment"]["surviving_pids"]:
                     raise RuntimeError("fork race did not produce exact termination receipt")
@@ -113,7 +113,7 @@ def main() -> int:
             name = f"pressure-{token}-{index}"
             call(operator, ["start", "--name", name, "--max-pids", "4", "--", "/usr/bin/python3", str(ROOT / "tests/fixtures/pid_pressure.py")])
             state = wait_state(operator, name, "TERMINATED")
-            receipt_files = sorted(Path("/var/lib/lumi-nutcracker/receipts").glob("*.json"), key=lambda path: path.stat().st_mtime_ns)
+            receipt_files = sorted(Path("/var/lib/lumi-eggcracker/receipts").glob("*.json"), key=lambda path: path.stat().st_mtime_ns)
             receipt = json.loads(receipt_files[-1].read_text(encoding="utf-8"))
             if receipt["trigger"]["kind"] != "PID_LIMIT":
                 raise RuntimeError(f"PID tripwire did not create PID receipt: {state}")
@@ -126,7 +126,7 @@ def main() -> int:
             if state.get("state") != "COMPLETED_ALLOWED":
                 raise RuntimeError("benign near-limit workload was not allowed")
             results["benign"].append(state["state"])
-        hostile_path = Path("/tmp") / f"lumi-nutcracker-hostile-{token}.json"
+        hostile_path = Path("/tmp") / f"lumi-eggcracker-hostile-{token}.json"
         hostile = f"hostile-{token}"
         call(operator, ["start", "--name", hostile, "--max-pids", "8", "--", "/usr/bin/python3", str(ROOT / "tests/fixtures/hostile_client.py"), str(hostile_path), str(args.socket_attempts)])
         wait_state(operator, hostile, "COMPLETED_ALLOWED", timeout=30)
@@ -141,9 +141,9 @@ def main() -> int:
         for index in range(args.restart_repetitions):
             name = f"restart-{token}-{index}"
             call(operator, ["start", "--name", name, "--max-pids", "8", "--", "/bin/sleep", "30"])
-            run(["/usr/bin/systemctl", "kill", "--kill-who=main", "-s", "SIGKILL", "lumi-nutcracker.service"])
+            run(["/usr/bin/systemctl", "kill", "--kill-who=main", "-s", "SIGKILL", "lumi-eggcracker.service"])
             state = wait_state(operator, name, "TERMINATED", timeout=12)
-            receipt_files = sorted(Path("/var/lib/lumi-nutcracker/receipts").glob("*.json"), key=lambda path: path.stat().st_mtime_ns)
+            receipt_files = sorted(Path("/var/lib/lumi-eggcracker/receipts").glob("*.json"), key=lambda path: path.stat().st_mtime_ns)
             receipt = json.loads(receipt_files[-1].read_text(encoding="utf-8"))
             if receipt["trigger"]["kind"] != "SUPERVISOR_RESTART_FAIL_CLOSED":
                 raise RuntimeError(f"restart did not fail closed: {state}")
@@ -162,9 +162,9 @@ def main() -> int:
             args.output.write_text(json.dumps(results, sort_keys=True) + "\n", encoding="utf-8")
         raise
     finally:
-        active = run(["/usr/bin/systemctl", "list-units", "lumi-nutcracker-workload-*", "--type=service", "--state=active", "--plain", "--no-legend"], check=False)
+        active = run(["/usr/bin/systemctl", "list-units", "lumi-eggcracker-workload-*", "--type=service", "--state=active", "--plain", "--no-legend"], check=False)
         for line in active.stdout.splitlines():
-            if line.split() and line.split()[0].startswith("lumi-nutcracker-workload-"):
+            if line.split() and line.split()[0].startswith("lumi-eggcracker-workload-"):
                 run(["/usr/bin/systemctl", "stop", line.split()[0]], check=False)
 
 

scripts/smoke_local_ai.py

@@ -23,9 +23,9 @@ def digest(path: Path) -> str:
 
 
 def call(args: list[str]) -> dict[str, object]:
-    result = subprocess.run(["/usr/local/bin/nutcracker", *args], capture_output=True, text=True, check=False, timeout=30)
+    result = subprocess.run(["/usr/local/bin/eggcracker", *args], capture_output=True, text=True, check=False, timeout=30)
     if result.returncode:
-        raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "Nutcracker command failed")
+        raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "Eggcracker command failed")
     return json.loads(result.stdout)
 
 
@@ -39,7 +39,7 @@ def main() -> int:
         raise SystemExit("output must be a new file under an existing directory")
     if not args.llama_cli.is_file() or not os.access(args.llama_cli, os.X_OK) or not args.model.is_file():
         raise SystemExit("llama-cli executable and GGUF model are required")
-    with tempfile.TemporaryDirectory(prefix="lumi-nutcracker-ai-smoke-", dir="/tmp") as raw:
+    with tempfile.TemporaryDirectory(prefix="lumi-eggcracker-ai-smoke-", dir="/tmp") as raw:
         root = Path(raw)
         # The operator owns this short-lived directory; the dedicated workload
         # identity needs a writable output location for the visible demo.

scripts/uninstall.py

@@ -1,4 +1,4 @@
-"""Exact manifest-bound uninstaller for Lumi Nutcracker."""
+"""Exact manifest-bound uninstaller for Lumi Eggcracker."""
 
 from __future__ import annotations
 
@@ -11,12 +11,13 @@ import shutil
 import subprocess
 from pathlib import Path
 
-LIB = Path("/usr/local/lib/lumi-nutcracker")
-BIN = Path("/usr/local/bin/nutcracker")
-ETC = Path("/etc/lumi-nutcracker")
-UNIT = Path("/etc/systemd/system/lumi-nutcracker.service")
-STATE = Path("/var/lib/lumi-nutcracker")
-RUNTIME = Path("/run/lumi-nutcracker")
+LIB = Path("/usr/local/lib/lumi-eggcracker")
+BIN = Path("/usr/local/bin/eggcracker")
+ETC = Path("/etc/lumi-eggcracker")
+UNIT = Path("/etc/systemd/system/lumi-eggcracker.service")
+STATE = Path("/var/lib/lumi-eggcracker")
+RUNTIME = Path("/run/lumi-eggcracker")
+PREFIX = "lumi-eggcracker-workload-"
 
 
 def digest(path: Path) -> str:
@@ -31,6 +32,23 @@ def run(argv: list[str]) -> subprocess.CompletedProcess[str]:
     return subprocess.run(argv, capture_output=True, text=True, check=False, timeout=60)
 
 
+def owned_cgroups_empty() -> bool:
+    root = Path("/sys/fs/cgroup/system.slice")
+    if not root.is_dir():
+        return False
+    for path in root.glob(f"{PREFIX}*.service"):
+        if not path.is_dir() or path.is_symlink():
+            return False
+        events = path / "cgroup.events"
+        try:
+            populated = dict(line.split(" ", 1) for line in events.read_text(encoding="ascii").splitlines()).get("populated")
+        except OSError:
+            return False
+        if populated != "0":
+            return False
+    return True
+
+
 def main() -> int:
     if os.geteuid() != 0:
         raise SystemExit("uninstaller must run as root")
@@ -38,23 +56,19 @@ def main() -> int:
     if manifest_path.is_symlink() or not manifest_path.is_file():
         raise SystemExit("installed manifest is missing")
     manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-    if manifest.get("schema_version") != "lumi-nutcracker.install.v1":
+    if manifest.get("schema_version") != "lumi-eggcracker.install.v2":
         raise SystemExit("installed manifest schema is invalid")
     for name, expected in manifest.get("files", {}).items():
         path = Path(name)
         if path.is_symlink() or not path.is_file() or digest(path) != expected:
             raise SystemExit(f"refusing uninstall because installed file drifted: {path}")
-    active = run(["/usr/bin/systemctl", "list-units", "lumi-nutcracker-workload-*", "--type=service", "--state=active", "--no-legend", "--no-pager"])
-    if active.stdout.strip():
-        raise SystemExit("refusing uninstall with active Nutcracker workloads")
-    run(["/usr/bin/systemctl", "disable", "--now", "lumi-nutcracker.service"])
-    run(["/usr/bin/systemctl", "reset-failed", "lumi-nutcracker.service"])
+    if not owned_cgroups_empty():
+        raise SystemExit("refusing uninstall with populated or uncertain Eggcracker cgroups")
+    run(["/usr/bin/systemctl", "disable", "--now", "lumi-eggcracker.service"])
+    run(["/usr/bin/systemctl", "reset-failed", "lumi-eggcracker.service"])
     for path in (UNIT, BIN, LIB, ETC, STATE, RUNTIME):
         if path.exists() and not path.is_symlink():
-            if path.is_dir():
-                shutil.rmtree(path)
-            else:
-                path.unlink()
+            shutil.rmtree(path) if path.is_dir() else path.unlink()
     if manifest.get("created_workload_user"):
         try:
             account = pwd.getpwnam(str(manifest["workload_user"]))
@@ -65,9 +79,8 @@ def main() -> int:
             if result.returncode:
                 raise SystemExit(result.stderr.strip() or "cannot remove created workload account")
     if manifest.get("created_workload_group"):
-        group_name = str(manifest.get("workload_group", ""))
         try:
-            group = grp.getgrnam(group_name)
+            group = grp.getgrnam(str(manifest.get("workload_group", "")))
         except KeyError:
             group = None
         if group:

scripts/verify_release.py

@@ -10,18 +10,19 @@ import sys
 import zipfile
 from pathlib import Path
 
-FORBIDDEN = ("/mnt/f/", "f:\\", "network-deny", "network_rule", "nftables", "/usr/sbin/nft", "b20", "brief 1", "brief-", "rootless", "skylark sentinel", "skylark-sentinel", "/usr/local/bin/lumi-nutcracker")
+VERSION = "0.1.1"
+PREFIX = f"lumi-eggcracker-{VERSION}/"
+FORBIDDEN = (
+    "/mnt/" + "f/", "f" + ":\\", "network" + "-deny", "network" + "_rule",
+    "nft" + "ables", "/usr/sbin/" + "nft", "b" + "20", "brief" + " 1",
+    "brief" + "-", "root" + "less", "skylark" + " sentinel",
+    "skylark" + "-sentinel", "nut" + "cracker",
+)
 
 
 def text_from_zip(path: Path) -> str:
     with zipfile.ZipFile(path) as archive:
-        values: list[str] = []
-        for name in archive.namelist():
-            if name.endswith("scripts/verify_release.py"):
-                continue
-            if name.endswith((".py", ".md", ".toml", ".txt")):
-                values.append(archive.read(name).decode("utf-8", errors="ignore").lower())
-        return "\n".join(values)
+        return "\n".join(archive.read(name).decode("utf-8", errors="ignore").lower() for name in archive.namelist() if name.endswith((".py", ".md", ".toml", ".txt")))
 
 
 def digest_bytes(value: bytes) -> str:
@@ -34,43 +35,28 @@ def main() -> int:
     parser.add_argument("--source-archive", required=True, type=Path)
     parser.add_argument("--release-bundle", required=True, type=Path)
     args = parser.parse_args()
-    if not args.artifact.is_file() or not args.source_archive.is_file() or not args.release_bundle.is_file():
+    if not all(path.is_file() for path in (args.artifact, args.source_archive, args.release_bundle)):
         raise SystemExit("release artifacts are missing")
     for path in (args.artifact, args.source_archive, args.release_bundle):
-        value = text_from_zip(path)
-        leaks = [item for item in FORBIDDEN if item in value]
+        leaks = [item for item in FORBIDDEN if item in text_from_zip(path)]
         if leaks:
             raise SystemExit(f"forbidden public artifact content in {path.name}: {leaks}")
     result = subprocess.run([sys.executable, str(args.artifact), "version"], capture_output=True, text=True, check=False)
-    if result.returncode or result.stdout.strip() != "0.1.0":
+    if result.returncode or result.stdout.strip() != VERSION:
         raise SystemExit("artifact version is inconsistent")
     manifest = json.loads((args.artifact.parent / "release-manifest.json").read_text(encoding="utf-8"))
-    if manifest.get("version") != "0.1.0" or manifest.get("artifact") != args.artifact.name:
+    if manifest.get("version") != VERSION or manifest.get("artifact") != args.artifact.name:
         raise SystemExit("release manifest is inconsistent")
-    prefix = "lumi-nutcracker-0.1.0/"
-    expected = {
-        prefix + args.artifact.name,
-        prefix + args.source_archive.name,
-        prefix + "release-manifest.json",
-        prefix + "SHA256SUMS",
-        prefix + "README.md",
-        prefix + "LICENSE",
-        prefix + "LIMITATIONS.md",
-        prefix + "SECURITY.md",
-        prefix + "RELEASE_NOTES.md",
-        prefix + "scripts/install.py",
-        prefix + "scripts/uninstall.py",
-        prefix + "scripts/verify_uninstalled.py",
-        prefix + "scripts/smoke_local_ai.py",
-    }
+    expected = {PREFIX + name for name in (args.artifact.name, args.source_archive.name, "release-manifest.json", "SHA256SUMS", "README.md", "LICENSE", "LIMITATIONS.md", "SECURITY.md", "SECURITY_MODEL.md", "QUALIFICATION.md", "RELEASE_NOTES.md", "scripts/install.py", "scripts/uninstall.py", "scripts/verify_uninstalled.py", "scripts/smoke_local_ai.py")}
     with zipfile.ZipFile(args.release_bundle) as archive:
         if set(archive.namelist()) != expected:
             raise SystemExit("release bundle contents are inconsistent")
-        if digest_bytes(archive.read(prefix + args.artifact.name)) != manifest["sha256"]:
+        if digest_bytes(archive.read(PREFIX + args.artifact.name)) != manifest["sha256"] or digest_bytes(archive.read(PREFIX + args.source_archive.name)) != manifest["source_archive_sha256"]:
             raise SystemExit("bundled artifact digest is inconsistent")
-        if digest_bytes(archive.read(prefix + args.source_archive.name)) != manifest["source_archive_sha256"]:
-            raise SystemExit("bundled source digest is inconsistent")
-    print(json.dumps({"result": "PASS", "version": "0.1.0"}, sort_keys=True))
+    sums = (args.artifact.parent / "SHA256SUMS").read_text(encoding="utf-8")
+    if args.artifact.name not in sums or args.source_archive.name not in sums or args.release_bundle.name not in sums:
+        raise SystemExit("checksums do not cover every release asset")
+    print(json.dumps({"result": "PASS", "version": VERSION}, sort_keys=True))
     return 0
 
 

scripts/verify_uninstalled.py

@@ -1,4 +1,4 @@
-"""Verify that a clean-install test left no Nutcracker service or paths behind."""
+"""Verify that a clean-install test left no Eggcracker paths behind."""
 
 from __future__ import annotations
 
@@ -8,25 +8,25 @@ import pwd
 import subprocess
 from pathlib import Path
 
-PATHS = (Path("/usr/local/lib/lumi-nutcracker"), Path("/usr/local/bin/nutcracker"), Path("/etc/lumi-nutcracker"), Path("/etc/systemd/system/lumi-nutcracker.service"), Path("/var/lib/lumi-nutcracker"), Path("/run/lumi-nutcracker"))
+PATHS = (Path("/usr/local/lib/lumi-eggcracker"), Path("/usr/local/bin/eggcracker"), Path("/etc/lumi-eggcracker"), Path("/etc/systemd/system/lumi-eggcracker.service"), Path("/var/lib/lumi-eggcracker"), Path("/run/lumi-eggcracker"))
 
 
 def main() -> int:
     if os.geteuid() != 0:
         raise SystemExit("uninstall verifier must run as root")
     if any(path.exists() or path.is_symlink() for path in PATHS):
-        raise SystemExit("Nutcracker installation path remains")
-    units = subprocess.run(["/usr/bin/systemctl", "list-units", "lumi-nutcracker*", "--all", "--plain", "--no-legend"], capture_output=True, text=True, check=False)
+        raise SystemExit("Eggcracker installation path remains")
+    units = subprocess.run(["/usr/bin/systemctl", "list-units", "lumi-eggcracker*", "--all", "--plain", "--no-legend"], capture_output=True, text=True, check=False)
     if units.stdout.strip():
-        raise SystemExit("Nutcracker unit remains")
+        raise SystemExit("Eggcracker unit remains")
     try:
-        pwd.getpwnam("lumi-nutcracker-workload")
+        pwd.getpwnam("lumi-eggcracker-workload")
     except KeyError:
         pass
     else:
         raise SystemExit("created workload account remains")
     try:
-        grp.getgrnam("lumi-nutcracker-workload")
+        grp.getgrnam("lumi-eggcracker-workload")
     except KeyError:
         print('{"result":"PASS"}')
         return 0

src/lumi_eggcracker/__init__.py

@@ -0,0 +1,3 @@
+"""Lumi Eggcracker: explicit protected Linux workload termination."""
+
+__version__ = "0.1.1"

src/{lumi_nutcracker → lumi_eggcracker}/cli.py

@@ -9,12 +9,13 @@ from pathlib import Path
 
 from . import __version__
 from .client import request
+from .gate import main as gate_main
 from .jsonio import JsonInputError, write_new_json
 from .supervisor import main as supervisor_main
 
 
 def _parser() -> argparse.ArgumentParser:
-    parser = argparse.ArgumentParser(prog="nutcracker")
+    parser = argparse.ArgumentParser(prog="eggcracker")
     commands = parser.add_subparsers(dest="command", required=True)
     start = commands.add_parser("start", help="launch an explicitly selected protected workload")
     start.add_argument("--name", required=True)
@@ -39,6 +40,8 @@ def main(argv: list[str] | None = None) -> int:
     values = sys.argv[1:] if argv is None else argv
     if values[:1] == ["_supervisor"]:
         return supervisor_main(values[1:])
+    if values[:1] == ["_gate"]:
+        return gate_main(values[1:])
     args = _parser().parse_args(values)
     if args.command == "version":
         print(__version__)
@@ -63,5 +66,5 @@ def main(argv: list[str] | None = None) -> int:
         print(json.dumps(value, sort_keys=True))
         return 0
     except (JsonInputError, OSError) as error:
-        print(f"nutcracker: {error}", file=sys.stderr)
+        print(f"eggcracker: {error}", file=sys.stderr)
         return 4

src/{lumi_nutcracker → lumi_eggcracker}/client.py

@@ -10,7 +10,7 @@ from typing import Any
 from .jsonio import JsonInputError
 
 MAX_FRAME = 32 * 1024
-SOCKET_PATH = "/run/lumi-nutcracker/control.sock"
+SOCKET_PATH = "/run/lumi-eggcracker/control.sock"
 
 
 def _receive(connection: socket.socket) -> dict[str, Any]:

src/{lumi_nutcracker → lumi_eggcracker}/containment.py

@@ -12,7 +12,7 @@ from pathlib import Path
 from .jsonio import JsonInputError
 
 CGROUP_ROOT = Path("/sys/fs/cgroup")
-UNIT_RE = re.compile(r"^/system\.slice/lumi-nutcracker-workload-([0-9a-f]{24})\.service$")
+UNIT_RE = re.compile(r"^/system\.slice/lumi-eggcracker-workload-([0-9a-f]{24})\.service$")
 
 
 @dataclass(frozen=True)
@@ -24,16 +24,6 @@ class CgroupIdentity:
     run_id: str
     unit: str
 
-    def as_dict(self) -> dict[str, object]:
-        return {
-            "boot_id": self.boot_id,
-            "cgroup": self.cgroup,
-            "cgroup_device": self.device,
-            "cgroup_inode": self.inode,
-            "run_id": self.run_id,
-            "unit": self.unit,
-        }
-
 
 @dataclass(frozen=True)
 class EmptyProof:
@@ -52,35 +42,40 @@ def boot_id() -> str:
 
 def _path(cgroup: str, *, root: Path = CGROUP_ROOT) -> Path:
     if not UNIT_RE.fullmatch(cgroup):
-        raise JsonInputError("cgroup is outside the Nutcracker unit namespace")
+        raise JsonInputError("cgroup is outside the Eggcracker unit namespace")
     candidate = root.joinpath(*cgroup.lstrip("/").split("/"))
     if candidate.is_symlink() or not candidate.is_dir():
         raise JsonInputError("owned cgroup is unavailable")
     return candidate
 
 
+def events_from_fd(descriptor: int) -> dict[str, int]:
+    os.lseek(descriptor, 0, os.SEEK_SET)
+    raw = os.read(descriptor, 4096).decode("ascii")
+    pairs = [line.split(" ", 1) for line in raw.splitlines()]
+    value = {key: item for key, item in pairs if key and item}
+    if not value or any(not item.isdigit() for item in value.values()):
+        raise JsonInputError("cgroup event file has invalid values")
+    return {key: int(item) for key, item in value.items()}
+
+
 def _events(path: Path) -> dict[str, int]:
     try:
-        pairs = [line.split(" ", 1) for line in path.read_text(encoding="ascii").splitlines()]
+        with path.open("rb", buffering=0) as handle:
+            return events_from_fd(handle.fileno())
     except OSError as error:
         raise JsonInputError(f"cannot read {path.name}: {error}") from error
-    value = {key: raw for key, raw in pairs if key and raw}
-    if any(not raw.isdigit() for raw in value.values()):
-        raise JsonInputError(f"{path.name} has invalid values")
-    return {key: int(raw) for key, raw in value.items()}
 
 
 def capture_identity(cgroup: str, run_id: str, unit: str, *, root: Path = CGROUP_ROOT) -> CgroupIdentity:
-    if unit != f"lumi-nutcracker-workload-{run_id}.service":
+    if unit != f"lumi-eggcracker-workload-{run_id}.service":
         raise JsonInputError("unit and run identity disagree")
     path = _path(cgroup, root=root)
     required = ("cgroup.kill", "cgroup.events", "cgroup.procs", "pids.events", "pids.max")
     if not all((path / name).is_file() for name in required):
         raise JsonInputError("owned cgroup lacks required v2 control files")
-    if "populated" not in _events(path / "cgroup.events"):
-        raise JsonInputError("owned cgroup does not expose populated state")
-    if "max" not in _events(path / "pids.events"):
-        raise JsonInputError("owned cgroup does not expose PID events")
+    if "populated" not in _events(path / "cgroup.events") or "max" not in _events(path / "pids.events"):
+        raise JsonInputError("owned cgroup does not expose required events")
     metadata = path.stat(follow_symlinks=False)
     return CgroupIdentity(boot_id(), cgroup, metadata.st_dev, metadata.st_ino, run_id, unit)
 
@@ -89,21 +84,13 @@ def validate_identity(identity: CgroupIdentity, *, root: Path = CGROUP_ROOT) ->
     if identity.boot_id != boot_id():
         raise JsonInputError("workload boot identity changed")
     path = _path(identity.cgroup, root=root)
-    try:
-        metadata = path.stat(follow_symlinks=False)
-    except OSError as error:
-        raise JsonInputError("owned cgroup is unavailable") from error
+    metadata = path.stat(follow_symlinks=False)
     if metadata.st_dev != identity.device or metadata.st_ino != identity.inode:
         raise JsonInputError("owned cgroup identity drifted")
     return path
 
 
-def pids_max_event(identity: CgroupIdentity, *, root: Path = CGROUP_ROOT) -> int:
-    return _events(validate_identity(identity, root=root) / "pids.events")["max"]
-
-
 def kill_path(path: Path) -> tuple[int, int]:
-    """Write a previously validated exact cgroup.kill file directly."""
     control = path / "cgroup.kill"
     started = time.monotonic_ns()
     descriptor = os.open(control, os.O_WRONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0))
@@ -116,55 +103,41 @@ def kill_path(path: Path) -> tuple[int, int]:
 
 
 def kill(identity: CgroupIdentity, *, root: Path = CGROUP_ROOT) -> tuple[int, int]:
-    """Validate then write cgroup.kill; use kill_path when trigger order matters."""
+    """Validate the exact identity, then apply the authoritative primitive."""
     return kill_path(validate_identity(identity, root=root))
 
 
 def _survivors(path: Path) -> tuple[list[int], int]:
-    directories = [path]
-    directories.extend(item for item in path.rglob("*") if item.is_dir() and not item.is_symlink())
+    directories = [path, *(item for item in path.rglob("*") if item.is_dir() and not item.is_symlink())]
     values: list[int] = []
     for directory in directories:
-        control = directory / "cgroup.procs"
-        if control.is_file():
-            for line in control.read_text(encoding="ascii").splitlines():
-                if not line.isdigit():
-                    raise JsonInputError("cgroup.procs contains invalid PID")
-                values.append(int(line))
+        for line in (directory / "cgroup.procs").read_text(encoding="ascii").splitlines():
+            if not line.isdigit():
+                raise JsonInputError("cgroup.procs contains invalid PID")
+            values.append(int(line))
     return sorted(set(values)), len(directories)
 
 
-def _collected(error: Exception) -> bool:
+def _collected(error: BaseException) -> bool:
     return isinstance(error, OSError) and error.errno in {errno.ENOENT, errno.ENODEV}
 
 
 def verify_empty(identity: CgroupIdentity, *, deadline_seconds: float = 0.9, root: Path = CGROUP_ROOT) -> tuple[int, EmptyProof]:
-    """Prove an empty hierarchy after direct kill, re-applying it across fork races."""
     deadline = time.monotonic() + deadline_seconds
     last = EmptyProof(False, 0, -1, [])
     while time.monotonic() <= deadline:
         try:
             path = validate_identity(identity, root=root)
-        except JsonInputError as error:
-            if "unavailable" in str(error):
-                return time.monotonic_ns(), EmptyProof(True, 0, 0, [])
-            raise
-        try:
             events = _events(path / "cgroup.events")
             survivors, descendants = _survivors(path)
         except (JsonInputError, OSError) as error:
-            # A transient service may be collected immediately after cgroup.kill.
-            # ENODEV/ENOENT from cgroupfs is proof that no live hierarchy remains.
-            if _collected(error) or "No such device" in str(error) or "No such file" in str(error):
+            if _collected(error) or "unavailable" in str(error) or "No such file" in str(error) or "No such device" in str(error):
                 return time.monotonic_ns(), EmptyProof(True, 0, 0, [])
             raise
         populated = events.get("populated", -1)
         last = EmptyProof(populated == 0 and not survivors, descendants, populated, survivors)
         if last.complete:
             return time.monotonic_ns(), last
-        # A process can fork while the kernel is walking a busy hierarchy for a
-        # prior cgroup.kill.  Re-applying the same direct primitive closes that
-        # narrow race; nothing is persisted or explained before enforcement.
         try:
             kill_path(path)
         except OSError as error:

src/lumi_eggcracker/gate.py

@@ -0,0 +1,22 @@
+"""Tiny workload-side pre-exec gate; it has no supervisor privileges."""
+
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="eggcracker internal-gate")
+    parser.add_argument("--fifo", required=True, type=Path)
+    parser.add_argument("command", nargs=argparse.REMAINDER)
+    args = parser.parse_args(argv)
+    command = args.command[1:] if args.command[:1] == ["--"] else args.command
+    if not command:
+        raise SystemExit("missing gated command")
+    with args.fifo.open("rb", buffering=0) as gate:
+        if gate.read(3) != b"GO\n":
+            raise SystemExit("invalid launch-gate release")
+    os.execvp(command[0], command)
+    return 127

src/{lumi_nutcracker → lumi_eggcracker}/jsonio.py

@@ -54,14 +54,17 @@ def write_new_json(destination: Path, value: dict[str, Any], *, mode: int = 0o60
     descriptor, raw = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent)
     temporary = Path(raw)
     try:
-        os.fchmod(descriptor, mode)
+        if hasattr(os, "fchmod"):
+            os.fchmod(descriptor, mode)
         os.write(descriptor, canonical_bytes(value))
         os.fsync(descriptor)
     finally:
         os.close(descriptor)
+    if not hasattr(os, "fchmod"):
+        os.chmod(temporary, mode)
     try:
         os.link(temporary, destination)
-        directory = os.open(destination.parent, os.O_RDONLY)
+        directory = os.open(destination.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
         try:
             os.fsync(directory)
         finally:

src/{lumi_nutcracker → lumi_eggcracker}/records.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import os
 import re
 import tempfile
@@ -11,54 +12,80 @@ from typing import Any
 from .containment import CgroupIdentity, EmptyProof
 from .jsonio import JsonInputError, canonical_bytes, load_regular_json
 
-RUN_SCHEMA = "lumi-nutcracker.run.v1"
-RECEIPT_SCHEMA = "lumi-nutcracker.receipt.v1"
+RUN_SCHEMA = "lumi-eggcracker.run.v2"
+RECEIPT_SCHEMA = "lumi-eggcracker.receipt.v2"
 NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\Z")
+RUN_ID = re.compile(r"[0-9a-f]{24}\Z")
+ACTIVE_STATES = {"STARTING", "RUNNING"}
+TERMINAL_STATES = {"COMPLETED_ALLOWED", "TERMINATED", "CONTAINMENT_FAILED", "CONTAINED_RECEIPT_FAILED"}
 
 
 def write_atomic(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
     descriptor, raw = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
     temporary = Path(raw)
     try:
-        os.fchmod(descriptor, 0o600)
+        if hasattr(os, "fchmod"):
+            os.fchmod(descriptor, 0o600)
         os.write(descriptor, canonical_bytes(value))
         os.fsync(descriptor)
     finally:
         os.close(descriptor)
+    if not hasattr(os, "fchmod"):
+        os.chmod(temporary, 0o600)
     os.replace(temporary, path)
-    directory = os.open(path.parent, os.O_RDONLY)
+    try:
+        directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    except OSError:
+        return  # Directory fsync is unavailable on Windows test hosts.
     try:
         os.fsync(directory)
+    except OSError:
+        return
     finally:
         os.close(directory)
 
 
-def record_path(runs: Path, name: str) -> Path:
-    if not NAME.fullmatch(name):
+def run_path(runs: Path, run_id: str) -> Path:
+    if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
+        raise JsonInputError("workload run identity is invalid")
+    return runs / f"{run_id}.json"
+
+
+def name_path(names: Path, name: str) -> Path:
+    if not isinstance(name, str) or not NAME.fullmatch(name):
         raise JsonInputError("workload name is invalid")
-    return runs / f"{name}.json"
+    return names / f"{name}.json"
+
+
+def command_summary(argv: list[str]) -> dict[str, Any]:
+    if not isinstance(argv, list) or not argv or not all(isinstance(item, str) and item for item in argv):
+        raise JsonInputError("workload argv is invalid")
+    return {"argv_count": len(argv), "argv_sha256": hashlib.sha256("\0".join(argv).encode("utf-8")).hexdigest(), "executable": argv[0]}
 
 
 def validate_run(value: dict[str, Any]) -> dict[str, Any]:
     expected = {
-        "argv", "boot_id", "cgroup", "cgroup_device", "cgroup_inode", "created_monotonic_ns",
-        "max_pids", "name", "operator_uid", "run_id", "schema_version", "state", "unit",
-        "workload_gid", "workload_uid",
+        "argv_count", "argv_sha256", "boot_id", "cgroup", "cgroup_device", "cgroup_inode",
+        "created_monotonic_ns", "executable", "max_pids", "name", "operator_uid", "run_id",
+        "schema_version", "state", "unit", "workload_gid", "workload_uid",
     }
     if set(value) != expected or value.get("schema_version") != RUN_SCHEMA:
         raise JsonInputError("run record schema is invalid")
     if not isinstance(value["name"], str) or not NAME.fullmatch(value["name"]):
         raise JsonInputError("run record name is invalid")
-    if not isinstance(value["run_id"], str) or not re.fullmatch(r"[0-9a-f]{24}", value["run_id"]):
+    if not isinstance(value["run_id"], str) or not RUN_ID.fullmatch(value["run_id"]):
         raise JsonInputError("run record identity is invalid")
-    if value["unit"] != f"lumi-nutcracker-workload-{value['run_id']}.service":
+    if value["unit"] != f"lumi-eggcracker-workload-{value['run_id']}.service":
         raise JsonInputError("run record unit is invalid")
-    if not isinstance(value["argv"], list) or not value["argv"] or not all(isinstance(item, str) and item for item in value["argv"]):
-        raise JsonInputError("run record argv is invalid")
-    keys = ("cgroup_device", "cgroup_inode", "created_monotonic_ns", "operator_uid", "workload_gid", "workload_uid", "max_pids")
+    if not isinstance(value["executable"], str) or not value["executable"]:
+        raise JsonInputError("run record executable is invalid")
+    if not isinstance(value["argv_sha256"], str) or not re.fullmatch(r"[0-9a-f]{64}", value["argv_sha256"]):
+        raise JsonInputError("run record command hash is invalid")
+    keys = ("cgroup_device", "cgroup_inode", "created_monotonic_ns", "operator_uid", "workload_gid", "workload_uid", "max_pids", "argv_count")
     if any(isinstance(value[key], bool) or not isinstance(value[key], int) or value[key] < 0 for key in keys):
         raise JsonInputError("run record integer field is invalid")
-    if value["state"] not in {"RUNNING", "COMPLETED_ALLOWED", "TERMINATED", "CONTAINMENT_FAILED", "CONTAINED_RECEIPT_FAILED"}:
+    if value["state"] not in ACTIVE_STATES | TERMINAL_STATES:
         raise JsonInputError("run record state is invalid")
     return value
 
@@ -68,20 +95,20 @@ def identity_from_run(value: dict[str, Any]) -> CgroupIdentity:
     return CgroupIdentity(record["boot_id"], record["cgroup"], record["cgroup_device"], record["cgroup_inode"], record["run_id"], record["unit"])
 
 
-def load_run(runs: Path, name: str) -> dict[str, Any]:
-    return validate_run(load_regular_json(record_path(runs, name)))
+def load_run(runs: Path, run_id: str) -> dict[str, Any]:
+    return validate_run(load_regular_json(run_path(runs, run_id)))
 
 
 def make_receipt(
     *, record: dict[str, Any], trigger: str, trigger_ns: int, kill_started_ns: int, kill_complete_ns: int,
-    empty_ns: int, proof: EmptyProof, version: str, source_commit: str, cleanup: dict[str, Any], event_id: str,
+    empty_ns: int, proof: EmptyProof, version: str, source_commit: str, event_id: str,
 ) -> dict[str, Any]:
-    if trigger not in {"OPERATOR", "PID_LIMIT", "SUPERVISOR_RESTART_FAIL_CLOSED"} or not re.fullmatch(r"[0-9a-f]{24}", event_id):
+    if trigger not in {"OPERATOR", "PID_LIMIT", "SUPERVISOR_FAILURE", "SUPERVISOR_RESTART_FAIL_CLOSED"} or not RUN_ID.fullmatch(event_id):
         raise JsonInputError("receipt trigger or event identity is invalid")
     if not proof.complete or proof.root_populated != 0 or proof.surviving_pids:
         raise JsonInputError("cannot issue a success receipt before exact emptiness proof")
     return {
-        "cleanup": cleanup,
+        "cleanup": {"attempted": False},
         "containment": {
             "cgroup_kill_written": True,
             "descendant_cgroups_checked": proof.descendant_cgroups_checked,

src/{lumi_nutcracker → lumi_eggcracker}/supervisor.py

@@ -1,9 +1,10 @@
-"""Root-owned single-backend supervisor for explicit protected workloads."""
+"""Root-owned, fail-closed supervisor for one explicit protected workload."""
 
 from __future__ import annotations
 
 import argparse
 import datetime as dt
+import errno
 import json
 import os
 import re
@@ -12,6 +13,7 @@ import signal
 import socket
 import struct
 import subprocess
+import sys
 import threading
 import time
 from pathlib import Path
@@ -20,29 +22,34 @@ from typing import Any
 from . import __version__
 from .containment import (
     capture_identity,
+    events_from_fd,
     kill_path,
-    pids_max_event,
     validate_identity,
     verify_empty,
 )
 from .jsonio import JsonInputError, load_regular_json
 from .records import (
+    ACTIVE_STATES,
+    RUN_ID,
     RUN_SCHEMA,
+    command_summary,
     identity_from_run,
     load_run,
     make_receipt,
-    record_path,
+    name_path,
+    run_path,
     validate_run,
     write_atomic,
 )
 
 MAX_FRAME = 32 * 1024
 NAME = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}\Z")
-RUN_ID = re.compile(r"[0-9a-f]{24}\Z")
-POLICY_SCHEMA = "lumi-nutcracker.policy.v1"
-SOCKET_PATH = Path("/run/lumi-nutcracker/control.sock")
-STATE_DIR = Path("/var/lib/lumi-nutcracker")
-UNIT_PREFIX = "lumi-nutcracker-workload-"
+POLICY_SCHEMA = "lumi-eggcracker.policy.v2"
+SOCKET_PATH = Path("/run/lumi-eggcracker/control.sock")
+STATE_DIR = Path("/var/lib/lumi-eggcracker")
+UNIT_PREFIX = "lumi-eggcracker-workload-"
+GATES_DIR = Path("/run/lumi-eggcracker/gates")
+MAX_TERMINAL_RECORDS = 128
 
 
 def _pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
@@ -103,11 +110,14 @@ class Supervisor:
             raise JsonInputError("supervisor build identity is invalid")
         self.policy = policy
         self.runs = STATE_DIR / "runs"
+        self.names = STATE_DIR / "names"
         self.receipts = STATE_DIR / "receipts"
         self.stop_event = threading.Event()
         self.locks: dict[str, threading.Lock] = {}
+        self.lock_guard = threading.Lock()
+        self.start_lock = threading.Lock()
         self.completed: dict[str, dict[str, Any]] = {}
-        self.operations: list[str] = []  # Test-visible in-memory ordering only.
+        self.operations: list[str] = []  # Test-visible ordering only.
 
     @property
     def operator_uid(self) -> int:
@@ -118,24 +128,75 @@ class Supervisor:
 
     def _show(self, unit: str) -> dict[str, str]:
         if not unit.startswith(UNIT_PREFIX) or not unit.endswith(".service"):
-            raise JsonInputError("unit is outside Nutcracker namespace")
+            raise JsonInputError("unit is outside Eggcracker namespace")
         result = self._run(["/usr/bin/systemctl", "show", unit, "--property=ActiveState", "--property=ControlGroup", "--property=TasksMax"])
         if result.returncode:
-            return {"ActiveState": "inactive", "ControlGroup": "", "TasksMax": ""}
+            return {"ActiveState": "unknown", "ControlGroup": "", "TasksMax": ""}
         return {key: value for key, value in (line.split("=", 1) for line in result.stdout.splitlines() if "=" in line)}
 
+    def _new_lock(self, name: str) -> threading.Lock:
+        with self.lock_guard:
+            return self.locks.setdefault(name, threading.Lock())
+
     def _store(self, record: dict[str, Any]) -> None:
+        record = validate_run(record)
         self.operations.append("durable-state")
-        write_atomic(record_path(self.runs, record["name"]), validate_run(record))
+        write_atomic(run_path(self.runs, record["run_id"]), record)
+        pointer = name_path(self.names, record["name"])
+        if record["state"] in ACTIVE_STATES:
+            write_atomic(pointer, {"run_id": record["run_id"]})
+        elif pointer.exists():
+            try:
+                current = load_regular_json(pointer)
+                if current == {"run_id": record["run_id"]}:
+                    pointer.unlink()
+            except JsonInputError:
+                raise JsonInputError("workload name index is invalid")
+        self._prune_terminal_records()
+
+    def _prune_terminal_records(self) -> None:
+        records: list[tuple[int, Path]] = []
+        for path in self.runs.glob("*.json"):
+            try:
+                value = load_run(self.runs, path.stem)
+            except JsonInputError:
+                continue
+            if value["state"] not in ACTIVE_STATES:
+                records.append((int(value["created_monotonic_ns"]), path))
+        for _, path in sorted(records, reverse=True)[MAX_TERMINAL_RECORDS:]:
+            path.unlink(missing_ok=True)
 
     def _load(self, name: str) -> dict[str, Any]:
-        return load_run(self.runs, name)
+        pointer = load_regular_json(name_path(self.names, name))
+        if set(pointer) != {"run_id"}:
+            raise JsonInputError("workload name index is invalid")
+        record = load_run(self.runs, pointer["run_id"])
+        if record["name"] != name or record["state"] not in ACTIVE_STATES:
+            raise JsonInputError("workload name is unavailable")
+        return record
+
+    def _latest_by_name(self, name: str) -> dict[str, Any]:
+        """Resolve a completed run only when its name is no longer active."""
+        candidates: list[dict[str, Any]] = []
+        for path in self.runs.glob("*.json"):
+            try:
+                record = load_run(self.runs, path.stem)
+            except JsonInputError:
+                continue
+            if record["name"] == name:
+                candidates.append(record)
+        if not candidates:
+            raise JsonInputError("workload name is unavailable")
+        return max(candidates, key=lambda item: int(item["created_monotonic_ns"]))
 
     def _receipt_path(self, event_id: str) -> Path:
         return self.receipts / f"{event_id}.json"
 
     def _prepare(self) -> None:
-        for path, mode, gid in ((SOCKET_PATH.parent, 0o710, self.policy["operator_gid"]), (STATE_DIR, 0o700, 0), (self.runs, 0o700, 0), (self.receipts, 0o700, 0)):
+        # Workloads need traversal only to their group-readable gate.  The
+        # control socket remains a root/operator 0660 inode, so traversal does
+        # not grant control-socket access or directory listing.
+        for path, mode, gid in ((SOCKET_PATH.parent, 0o711, self.policy["operator_gid"]), (GATES_DIR, 0o710, self.policy["workload_gid"]), (STATE_DIR, 0o700, 0), (self.runs, 0o700, 0), (self.names, 0o700, 0), (self.receipts, 0o700, 0)):
             path.mkdir(mode=mode, parents=True, exist_ok=True)
             os.chown(path, 0, gid)
             os.chmod(path, mode)
@@ -143,38 +204,25 @@ class Supervisor:
             raise JsonInputError("control socket already exists")
         self._recover()
 
-    def _new_lock(self, name: str) -> threading.Lock:
-        if name not in self.locks:
-            self.locks[name] = threading.Lock()
-        return self.locks[name]
-
     def _cleanup(self, unit: str) -> dict[str, Any]:
         result = self._run(["/usr/bin/systemctl", "stop", unit])
-        return {"systemctl_stop_attempted": True, "systemctl_stop_returncode": result.returncode, "systemctl_stop_stderr": result.stderr.strip()}
+        return {"attempted": True, "systemctl_stop_returncode": result.returncode, "systemctl_stop_stderr": result.stderr.strip()}
 
-    def _complete_allowed(self, record: dict[str, Any]) -> bool:
-        """Persist normal completion only after systemd reports the unit inactive."""
-        lock = self._new_lock(record["name"])
-        with lock:
-            if record["run_id"] in self.completed or record["state"] != "RUNNING":
-                return False
-            if self._show(record["unit"])["ActiveState"] == "active":
-                return False
-            record["state"] = "COMPLETED_ALLOWED"
-            self._store(record)
-            return True
+    def _write_receipt(self, receipt: dict[str, Any], event_id: str) -> None:
+        receipt["receipt_written_utc"] = dt.datetime.now(dt.UTC).isoformat().replace("+00:00", "Z")
+        self.operations.append("durable-receipt")
+        write_atomic(self._receipt_path(event_id), receipt)
 
     def _contain(self, record: dict[str, Any], trigger: str, trigger_ns: int | None = None) -> dict[str, Any]:
-        """Contain exactly one verified cgroup. First trigger-side effect is cgroup.kill."""
-        lock = self._new_lock(record["name"])
+        """Direct cgroup.kill is the first trigger-side effect; cleanup is post-proof only."""
+        lock = self._new_lock(record["run_id"])
         with lock:
             if record["run_id"] in self.completed:
                 return self.completed[record["run_id"]]
             identity = identity_from_run(record)
-            # Identity validation completes before the trigger is declared.
-            path = validate_identity(identity)
             observed = trigger_ns if trigger_ns is not None else time.monotonic_ns()
             try:
+                path = validate_identity(identity)
                 self.operations.append("cgroup.kill")
                 kill_started, kill_completed = kill_path(path)
                 empty_ns, proof = verify_empty(identity)
@@ -184,112 +232,205 @@ class Supervisor:
                 record["state"] = "CONTAINMENT_FAILED"
                 self._store(record)
                 raise JsonInputError(f"containment failed: {error}") from error
-            cleanup = self._cleanup(record["unit"])
             event_id = os.urandom(12).hex()
-            receipt = make_receipt(record=record, trigger=trigger, trigger_ns=observed, kill_started_ns=kill_started, kill_complete_ns=kill_completed, empty_ns=empty_ns, proof=proof, version=__version__, source_commit=self.policy["source_commit"], cleanup=cleanup, event_id=event_id)
-            receipt["receipt_written_utc"] = dt.datetime.now(dt.UTC).isoformat().replace("+00:00", "Z")
+            receipt = make_receipt(record=record, trigger=trigger, trigger_ns=observed, kill_started_ns=kill_started, kill_complete_ns=kill_completed, empty_ns=empty_ns, proof=proof, version=__version__, source_commit=self.policy["source_commit"], event_id=event_id)
             try:
-                self.operations.append("durable-receipt")
-                write_atomic(self._receipt_path(event_id), receipt)
+                self._write_receipt(receipt, event_id)
                 record["state"] = "TERMINATED"
                 self._store(record)
             except Exception as error:
                 record["state"] = "CONTAINED_RECEIPT_FAILED"
                 self._store(record)
                 raise JsonInputError(f"contained but receipt persistence failed: {error}") from error
+            # Cleanup is deliberately not in the enforcement or proof path.
+            cleanup = self._cleanup(record["unit"])
+            receipt["cleanup"] = cleanup
+            try:
+                write_atomic(self._receipt_path(event_id), receipt)
+            except (OSError, JsonInputError):
+                receipt["cleanup_update_error"] = True
             receipt["receipt_path"] = str(self._receipt_path(event_id))
             self.completed[record["run_id"]] = receipt
             return receipt
 
-    def _watch(self, record: dict[str, Any]) -> None:
+    def _complete_allowed(self, record: dict[str, Any], cgroup_events_fd: int) -> bool:
+        """Only exact cgroup population, not systemd state, permits completion."""
+        events = events_from_fd(cgroup_events_fd)
+        if events.get("populated") != 0:
+            return False
+        lock = self._new_lock(record["run_id"])
+        with lock:
+            if record["state"] not in ACTIVE_STATES:
+                return False
+            record["state"] = "COMPLETED_ALLOWED"
+            self._store(record)
+            return True
+
+    def _watch_once(self, record: dict[str, Any], ready: threading.Event | None = None) -> None:
         identity = identity_from_run(record)
+        path = validate_identity(identity)
+        pids_fd = os.open(path / "pids.events", os.O_RDONLY | os.O_CLOEXEC)
+        cgroup_fd = os.open(path / "cgroup.events", os.O_RDONLY | os.O_CLOEXEC)
         try:
-            baseline = pids_max_event(identity)
-            descriptor = os.open(validate_identity(identity) / "pids.events", os.O_RDONLY | os.O_CLOEXEC)
+            baseline = events_from_fd(pids_fd).get("max")
+            if baseline is None:
+                raise JsonInputError("pids.events lacks max counter")
             poller = select.poll()
-            poller.register(descriptor, select.POLLPRI | select.POLLERR | select.POLLIN)
+            poller.register(pids_fd, select.POLLPRI)
+            poller.register(cgroup_fd, select.POLLPRI)
+            if ready is not None:
+                ready.set()
             while not self.stop_event.is_set():
-                if not poller.poll(250):
-                    if self._complete_allowed(record):
-                        return
-                    continue
-                # --collect may remove the cgroup between poll wake-up and the
-                # pids.events read.  An already inactive unit completed normally;
-                # do not turn that lifecycle race into a false containment failure.
-                if self._complete_allowed(record):
-                    return
-                os.lseek(descriptor, 0, os.SEEK_SET)
-                if pids_max_event(identity) > baseline:
+                poller.poll(250)
+                current = events_from_fd(pids_fd).get("max")
+                if current is None:
+                    raise JsonInputError("pids.events lacks max counter")
+                if current > baseline:
                     self._contain(record, "PID_LIMIT", time.monotonic_ns())
                     return
-        except Exception:  # noqa: BLE001 - any watcher failure must fail closed
-            # A lost watcher must not leave an active owned workload running.
-            try:
-                self._contain(record, "SUPERVISOR_RESTART_FAIL_CLOSED", time.monotonic_ns())
-            except JsonInputError:
-                pass
+                if self._complete_allowed(record, cgroup_fd):
+                    return
         finally:
-            if "descriptor" in locals():
+            os.close(pids_fd)
+            os.close(cgroup_fd)
+
+    def _watch(self, record: dict[str, Any], ready: threading.Event | None = None) -> None:
+        failure: BaseException | None = None
+        for attempt in range(2):
+            try:
+                self._watch_once(record, ready)
+                return
+            except (JsonInputError, OSError, RuntimeError) as error:
+                failure = error
+                if attempt == 0:
+                    continue
+        try:
+            self._contain(record, "SUPERVISOR_FAILURE", time.monotonic_ns())
+        except JsonInputError:
+            # A failed containment is intentionally left as a durable failure state.
+            pass
+        if failure is not None:
+            print(f"eggcracker watcher failed closed: {failure}", file=sys.stderr, flush=True)
+
+    def _make_gate(self, run_id: str) -> Path:
+        gate = GATES_DIR / run_id
+        if gate.exists() or gate.is_symlink():
+            raise JsonInputError("launch gate already exists")
+        os.mkfifo(gate, 0o640)
+        os.chown(gate, 0, self.policy["workload_gid"])
+        os.chmod(gate, 0o640)
+        return gate
+
+    def _release_gate(self, gate: Path) -> None:
+        deadline = time.monotonic() + 1.5
+        try:
+            while True:
+                try:
+                    descriptor = os.open(gate, os.O_WRONLY | os.O_NONBLOCK | os.O_CLOEXEC)
+                    break
+                except OSError as error:
+                    if error.errno != errno.ENXIO or time.monotonic() >= deadline:
+                        raise JsonInputError("workload gate did not attach") from error
+                    time.sleep(0.01)
+            try:
+                if os.write(descriptor, b"GO\n") != 3:
+                    raise JsonInputError("short launch-gate release")
+            finally:
                 os.close(descriptor)
+        finally:
+            gate.unlink(missing_ok=True)
+
+    def _active_exists(self) -> bool:
+        for path in self.runs.glob("*.json"):
+            try:
+                if load_run(self.runs, path.stem)["state"] in ACTIVE_STATES:
+                    return True
+            except JsonInputError:
+                raise JsonInputError("run record is invalid")
+        return False
 
     def _start(self, args: dict[str, Any]) -> dict[str, Any]:
         if set(args) != {"argv", "max_pids", "name"}:
             raise JsonInputError("start arguments are invalid")
         name, argv, maximum = args["name"], args["argv"], args["max_pids"]
-        if not isinstance(name, str) or not NAME.fullmatch(name) or record_path(self.runs, name).exists():
-            raise JsonInputError("workload name is unavailable")
-        if not isinstance(argv, list) or not argv or not all(isinstance(item, str) and item for item in argv):
-            raise JsonInputError("workload argv is invalid")
+        if not isinstance(name, str) or not NAME.fullmatch(name):
+            raise JsonInputError("workload name is invalid")
+        summary = command_summary(argv)
         if isinstance(maximum, bool) or not isinstance(maximum, int) or not 4 <= maximum <= 4096:
             raise JsonInputError("max_pids must be from 4 to 4096")
-        run_id = os.urandom(12).hex()
-        unit = f"{UNIT_PREFIX}{run_id}.service"
-        result = self._run(["/usr/bin/systemd-run", f"--unit={unit}", "--collect", f"--uid={self.policy['workload_uid']}", f"--gid={self.policy['workload_gid']}", "--property=Type=exec", "--property=ExitType=cgroup", "--property=KillMode=control-group", "--property=NoNewPrivileges=yes", "--property=UMask=0077", f"--property=TasksMax={maximum}", "--", *argv])
-        if result.returncode:
-            raise JsonInputError(result.stderr.strip() or "system workload launch failed")
-        deadline = time.monotonic() + 2.0
-        props: dict[str, str] = {}
-        while time.monotonic() < deadline:
-            props = self._show(unit)
-            if props["ActiveState"] == "active" and props["ControlGroup"]:
-                break
-            time.sleep(0.01)
-        if props.get("ActiveState") != "active" or not props.get("ControlGroup"):
-            self._run(["/usr/bin/systemctl", "stop", unit])
-            raise JsonInputError("workload did not become active")
-        identity = capture_identity(props["ControlGroup"], run_id, unit)
-        record = {"argv": argv, "boot_id": identity.boot_id, "cgroup": identity.cgroup, "cgroup_device": identity.device, "cgroup_inode": identity.inode, "created_monotonic_ns": time.monotonic_ns(), "max_pids": maximum, "name": name, "operator_uid": self.operator_uid, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": unit, "workload_gid": self.policy["workload_gid"], "workload_uid": self.policy["workload_uid"]}
-        self._store(record)
-        threading.Thread(target=self._watch, args=(record,), daemon=True).start()
-        return {"name": name, "state": "RUNNING", "unit": unit, "workload_uid": record["workload_uid"]}
-
-    def _orphan_record(self, unit: str, cgroup: str) -> dict[str, Any]:
+        with self.start_lock:
+            if name_path(self.names, name).exists() or self._active_exists():
+                raise JsonInputError("one protected workload is already active or name is unavailable")
+            run_id = os.urandom(12).hex()
+            unit = f"{UNIT_PREFIX}{run_id}.service"
+            gate = self._make_gate(run_id)
+            result = self._run(["/usr/bin/systemd-run", "--no-block", f"--unit={unit}", "--collect", f"--uid={self.policy['workload_uid']}", f"--gid={self.policy['workload_gid']}", "--property=Type=exec", "--property=ExitType=cgroup", "--property=KillMode=control-group", "--property=NoNewPrivileges=yes", "--property=UMask=0077", f"--property=TasksMax={maximum}", "--", "/usr/bin/python3", "/usr/local/lib/lumi-eggcracker/lumi-eggcracker.pyz", "_gate", "--fifo", str(gate), "--", *argv])
+            if result.returncode:
+                gate.unlink(missing_ok=True)
+                raise JsonInputError(result.stderr.strip() or "system workload launch failed")
+            try:
+                deadline = time.monotonic() + 2.0
+                props: dict[str, str] = {}
+                while time.monotonic() < deadline:
+                    props = self._show(unit)
+                    if props["ActiveState"] == "active" and props["ControlGroup"]:
+                        break
+                    time.sleep(0.01)
+                if props.get("ActiveState") != "active" or not props.get("ControlGroup"):
+                    raise JsonInputError("gated workload did not become active")
+                identity = capture_identity(props["ControlGroup"], run_id, unit)
+                record = {**summary, "boot_id": identity.boot_id, "cgroup": identity.cgroup, "cgroup_device": identity.device, "cgroup_inode": identity.inode, "created_monotonic_ns": time.monotonic_ns(), "max_pids": maximum, "name": name, "operator_uid": self.operator_uid, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "STARTING", "unit": unit, "workload_gid": self.policy["workload_gid"], "workload_uid": self.policy["workload_uid"]}
+                self._store(record)
+                ready = threading.Event()
+                watcher = threading.Thread(target=self._watch, args=(record, ready), daemon=True)
+                watcher.start()
+                if not ready.wait(2.0):
+                    raise JsonInputError("watcher did not become ready before target release")
+                # The gated target has not executed yet.  The watcher has opened
+                # both event descriptors and captured the PID baseline.
+                self._release_gate(gate)
+                if record["state"] in ACTIVE_STATES:
+                    record["state"] = "RUNNING"
+                    self._store(record)
+                return {"name": name, "run_id": run_id, "state": record["state"], "unit": unit, "workload_uid": record["workload_uid"]}
+            except Exception:
+                gate.unlink(missing_ok=True)
+                try:
+                    # Best effort rollback after a launch failure; direct kill is still authoritative.
+                    identity = capture_identity(props["ControlGroup"], run_id, unit)
+                    kill_path(validate_identity(identity))
+                    verify_empty(identity)
+                except (JsonInputError, OSError, RuntimeError):
+                    self._run(["/usr/bin/systemctl", "stop", unit])
+                raise
+
+    def _orphan_record(self, cgroup: str) -> dict[str, Any]:
+        unit = Path(cgroup).name
         run_id = unit.removeprefix(UNIT_PREFIX).removesuffix(".service")
         if not RUN_ID.fullmatch(run_id):
-            raise JsonInputError("orphan unit identity is invalid")
+            raise JsonInputError("orphan cgroup identity is invalid")
         identity = capture_identity(cgroup, run_id, unit)
-        return {"argv": ["<orphaned-owned-unit>"], "boot_id": identity.boot_id, "cgroup": identity.cgroup, "cgroup_device": identity.device, "cgroup_inode": identity.inode, "created_monotonic_ns": time.monotonic_ns(), "max_pids": 0, "name": f"orphan-{run_id}", "operator_uid": self.operator_uid, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": unit, "workload_gid": self.policy["workload_gid"], "workload_uid": self.policy["workload_uid"]}
-
-    def _active_units(self) -> list[str]:
-        result = self._run(["/usr/bin/systemctl", "list-units", f"{UNIT_PREFIX}*", "--type=service", "--all", "--plain", "--no-legend"])
-        if result.returncode:
-            raise JsonInputError("cannot enumerate owned units")
-        return [line.split()[0] for line in result.stdout.splitlines() if line.split() and line.split()[0].startswith(UNIT_PREFIX)]
+        return {"argv_count": 0, "argv_sha256": "0" * 64, "boot_id": identity.boot_id, "cgroup": identity.cgroup, "cgroup_device": identity.device, "cgroup_inode": identity.inode, "created_monotonic_ns": time.monotonic_ns(), "executable": "<orphaned-owned-cgroup>", "max_pids": 0, "name": f"orphan-{run_id}", "operator_uid": self.operator_uid, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": unit, "workload_gid": self.policy["workload_gid"], "workload_uid": self.policy["workload_uid"]}
 
     def _recover(self) -> None:
-        records: dict[str, dict[str, Any]] = {}
+        recorded_ids: set[str] = set()
         for path in self.runs.glob("*.json"):
             try:
                 record = load_run(self.runs, path.stem)
-                records[record["unit"]] = record
             except JsonInputError:
                 continue
-        for unit in self._active_units():
-            props = self._show(unit)
-            if props["ActiveState"] != "active" or not props["ControlGroup"]:
+            recorded_ids.add(record["run_id"])
+            if record["state"] in ACTIVE_STATES:
+                self._contain(record, "SUPERVISOR_RESTART_FAIL_CLOSED")
+        root = Path("/sys/fs/cgroup/system.slice")
+        if not root.is_dir():
+            return
+        for path in root.glob(f"{UNIT_PREFIX}*.service"):
+            if not path.is_dir() or path.is_symlink():
                 continue
-            record = records.get(unit) or self._orphan_record(unit, props["ControlGroup"])
-            self._contain(record, "SUPERVISOR_RESTART_FAIL_CLOSED")
+            run_id = path.name.removeprefix(UNIT_PREFIX).removesuffix(".service")
+            if RUN_ID.fullmatch(run_id) and run_id not in recorded_ids:
+                self._contain(self._orphan_record("/system.slice/" + path.name), "SUPERVISOR_RESTART_FAIL_CLOSED")
 
     def handle(self, value: dict[str, Any]) -> dict[str, Any]:
         if set(value) != {"action", "args"} or not isinstance(value["action"], str) or not isinstance(value["args"], dict):
@@ -301,13 +442,14 @@ class Supervisor:
         if action == "start":
             return self._start(args)
         if action == "status" and set(args) == {"name"}:
-            record = self._load(args["name"])
-            if record["state"] == "RUNNING":
-                self._complete_allowed(record)
-            props = self._show(record["unit"])
-            return {"active_state": props["ActiveState"], "name": record["name"], "state": record["state"], "unit": record["unit"], "workload_uid": record["workload_uid"]}
+            try:
+                record = self._load(args["name"])
+            except JsonInputError:
+                record = self._latest_by_name(args["name"])
+            return {"name": record["name"], "run_id": record["run_id"], "state": record["state"], "unit": record["unit"], "workload_uid": record["workload_uid"]}
         if action == "list" and not args:
-            return {"runs": [self.handle({"action": "status", "args": {"name": path.stem}}) for path in sorted(self.runs.glob("*.json"))]}
+            runs = [self.handle({"action": "status", "args": {"name": path.stem}}) for path in sorted(self.names.glob("*.json"))]
+            return {"runs": runs}
         if action == "kill" and set(args) == {"name"}:
             return self._contain(self._load(args["name"]), "OPERATOR")
         raise JsonInputError("unsupported supervisor action")
@@ -341,11 +483,15 @@ class Supervisor:
 
 
 def main(argv: list[str] | None = None) -> int:
-    parser = argparse.ArgumentParser(prog="nutcracker internal-supervisor")
+    parser = argparse.ArgumentParser(prog="eggcracker internal-supervisor")
     parser.add_argument("--policy", required=True, type=Path)
     args = parser.parse_args(argv)
     supervisor = Supervisor(args.policy)
-    signal.signal(signal.SIGTERM, lambda *_: supervisor.stop_event.set())
+    def stop(*_: object) -> None:
+        supervisor.stop_event.set()
+        raise SystemExit(0)
+
+    signal.signal(signal.SIGTERM, stop)
     return supervisor.serve()
 
 

src/lumi_nutcracker/__init__.py

@@ -1,3 +0,0 @@
-"""Lumi Nutcracker: explicit protected Linux workload termination."""
-
-__version__ = "0.1.0"

tests/fixtures/hostile_client.py

@@ -11,7 +11,7 @@ from pathlib import Path
 
 output = Path(sys.argv[1])
 attempts = int(sys.argv[2])
-socket_path = "/run/lumi-nutcracker/control.sock"
+socket_path = "/run/lumi-eggcracker/control.sock"
 successes = 0
 replacement_successes = 0
 for index in range(attempts):
@@ -22,7 +22,7 @@ for index in range(attempts):
         successes += 1
     except OSError:
         pass
-result = subprocess.run(["/usr/local/bin/nutcracker", "start", "--name", "replacement-hostile", "--max-pids", "4", "--", "/bin/sleep", "2"], capture_output=True, text=True, check=False)
+result = subprocess.run(["/usr/local/bin/eggcracker", "start", "--name", "replacement-hostile", "--max-pids", "4", "--", "/bin/sleep", "2"], capture_output=True, text=True, check=False)
 if result.returncode == 0:
     replacement_successes += 1
 output.write_text(json.dumps({"connection_successes": successes, "replacement_successes": replacement_successes, "uid": os.getuid()}) + "\n", encoding="utf-8")

tests/test_cli.py

@@ -5,7 +5,7 @@ import unittest
 from contextlib import redirect_stdout
 from unittest.mock import patch
 
-from lumi_nutcracker.cli import main
+from lumi_eggcracker.cli import main
 
 
 class CliTests(unittest.TestCase):
@@ -13,19 +13,17 @@ class CliTests(unittest.TestCase):
         output = io.StringIO()
         with redirect_stdout(output):
             self.assertEqual(0, main(["version"]))
-        self.assertEqual("0.1.0", output.getvalue().strip())
+        self.assertEqual("0.1.1", output.getvalue().strip())
 
     def test_public_help_has_only_supported_commands(self) -> None:
-        from lumi_nutcracker.cli import _parser
-
+        from lumi_eggcracker.cli import _parser
         help_text = _parser().format_help()
         for command in ("start", "kill", "status", "list", "doctor", "version"):
             self.assertIn(command, help_text)
         self.assertNotIn("_supervisor", help_text)
         self.assertNotIn("network" + "-deny", help_text)
-        self.assertNotIn("root" + "less", help_text)
 
     def test_internal_supervisor_dispatch_remains_available(self) -> None:
-        with patch("lumi_nutcracker.cli.supervisor_main", return_value=7) as supervisor:
+        with patch("lumi_eggcracker.cli.supervisor_main", return_value=7) as supervisor:
             self.assertEqual(main(["_supervisor", "--policy", "/tmp/policy.json"]), 7)
         supervisor.assert_called_once_with(["--policy", "/tmp/policy.json"])

tests/test_containment.py

@@ -5,14 +5,14 @@ import unittest
 from pathlib import Path
 from unittest.mock import patch
 
-from lumi_nutcracker import containment
-from lumi_nutcracker.jsonio import JsonInputError
+from lumi_eggcracker import containment
+from lumi_eggcracker.jsonio import JsonInputError
 
 
 class ContainmentTests(unittest.TestCase):
     def _root(self, temporary: Path) -> tuple[Path, str, str, str]:
         run_id = "a" * 24
-        unit = f"lumi-nutcracker-workload-{run_id}.service"
+        unit = f"lumi-eggcracker-workload-{run_id}.service"
         cgroup = f"/system.slice/{unit}"
         path = temporary / "system.slice" / unit
         path.mkdir(parents=True)
@@ -51,11 +51,10 @@ class ContainmentTests(unittest.TestCase):
             self.assertTrue(direct.called)
             self.assertTrue(proof.complete)
 
-    def test_collected_cgroup_after_direct_kill_is_empty(self) -> None:
+    def test_collected_cgroup_wrapped_as_json_error_is_still_an_empty_proof(self) -> None:
         with tempfile.TemporaryDirectory() as raw, patch.object(containment, "boot_id", return_value="a" * 36):
-            path, cgroup, run_id, unit = self._root(Path(raw))
+            _, cgroup, run_id, unit = self._root(Path(raw))
             identity = containment.capture_identity(cgroup, run_id, unit, root=Path(raw))
-            (path / "cgroup.events").write_text("populated 1\n", encoding="ascii")
-            with patch.object(containment, "kill_path", side_effect=FileNotFoundError(2, "No such file")):
+            with patch.object(containment, "_events", side_effect=JsonInputError("cannot read cgroup.events: No such file")):
                 _, proof = containment.verify_empty(identity, root=Path(raw))
             self.assertTrue(proof.complete)

tests/test_records.py

@@ -2,25 +2,29 @@ from __future__ import annotations
 
 import unittest
 
-from lumi_nutcracker.containment import EmptyProof
-from lumi_nutcracker.jsonio import JsonInputError
-from lumi_nutcracker.records import RUN_SCHEMA, make_receipt, validate_run
+from lumi_eggcracker.containment import EmptyProof
+from lumi_eggcracker.jsonio import JsonInputError
+from lumi_eggcracker.records import RUN_SCHEMA, command_summary, make_receipt, validate_run
 
 
 def record() -> dict[str, object]:
     run_id = "a" * 24
-    return {"argv": ["/bin/true"], "boot_id": "b" * 36, "cgroup": f"/system.slice/lumi-nutcracker-workload-{run_id}.service", "cgroup_device": 1, "cgroup_inode": 2, "created_monotonic_ns": 3, "max_pids": 8, "name": "demo", "operator_uid": 1001, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": f"lumi-nutcracker-workload-{run_id}.service", "workload_gid": 2001, "workload_uid": 2001}
+    return {**command_summary(["/bin/true", "--safe"]), "boot_id": "b" * 36, "cgroup": f"/system.slice/lumi-eggcracker-workload-{run_id}.service", "cgroup_device": 1, "cgroup_inode": 2, "created_monotonic_ns": 3, "max_pids": 8, "name": "demo", "operator_uid": 1001, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": f"lumi-eggcracker-workload-{run_id}.service", "workload_gid": 2001, "workload_uid": 2001}
 
 
 class RecordTests(unittest.TestCase):
     def test_receipt_requires_exact_empty_proof(self) -> None:
         value = record()
-        receipt = make_receipt(record=value, trigger="OPERATOR", trigger_ns=10, kill_started_ns=11, kill_complete_ns=12, empty_ns=13, proof=EmptyProof(True, 1, 0, []), version="0.1.0", source_commit="c" * 40, cleanup={}, event_id="d" * 24)
+        receipt = make_receipt(record=value, trigger="OPERATOR", trigger_ns=10, kill_started_ns=11, kill_complete_ns=12, empty_ns=13, proof=EmptyProof(True, 1, 0, []), version="0.1.1", source_commit="c" * 40, event_id="d" * 24)
         self.assertEqual("TERMINATED", receipt["result"])
         self.assertEqual("cgroup.kill", receipt["containment"]["primitive"])
 
-    def test_schema_rejects_unknown_field(self) -> None:
+    def test_durable_record_redacts_command_arguments(self) -> None:
         value = record()
-        value["extra"] = True
+        self.assertNotIn("argv", value)
+        self.assertNotIn("--safe", str(value))
+
+    def test_schema_rejects_unknown_field(self) -> None:
+        value = record(); value["extra"] = True
         with self.assertRaises(JsonInputError):
             validate_run(value)

tests/test_supervisor.py

@@ -1,45 +1,78 @@
 from __future__ import annotations
 
+import tempfile
+import threading
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 
-from lumi_nutcracker.containment import EmptyProof
-from lumi_nutcracker.records import RUN_SCHEMA
-from lumi_nutcracker.supervisor import Supervisor
+from lumi_eggcracker.containment import EmptyProof
+from lumi_eggcracker.jsonio import JsonInputError
+from lumi_eggcracker.records import RUN_SCHEMA, command_summary
+from lumi_eggcracker.supervisor import Supervisor
 
 
 def record() -> dict[str, object]:
     run_id = "a" * 24
-    return {"argv": ["/bin/true"], "boot_id": "b" * 36, "cgroup": f"/system.slice/lumi-nutcracker-workload-{run_id}.service", "cgroup_device": 1, "cgroup_inode": 2, "created_monotonic_ns": 3, "max_pids": 8, "name": "demo", "operator_uid": 1001, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": f"lumi-nutcracker-workload-{run_id}.service", "workload_gid": 2001, "workload_uid": 2001}
+    return {**command_summary(["/bin/true"]), "boot_id": "b" * 36, "cgroup": f"/system.slice/lumi-eggcracker-workload-{run_id}.service", "cgroup_device": 1, "cgroup_inode": 2, "created_monotonic_ns": 3, "max_pids": 8, "name": "demo", "operator_uid": 1001, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": f"lumi-eggcracker-workload-{run_id}.service", "workload_gid": 2001, "workload_uid": 2001}
 
 
 class SupervisorTests(unittest.TestCase):
     def _instance(self) -> Supervisor:
         value = object.__new__(Supervisor)
-        value.policy = {"source_commit": "c" * 40}
+        value.policy = {"source_commit": "c" * 40, "workload_uid": 2001}
         value.runs = Path(".")
+        value.names = Path(".")
         value.receipts = Path(".")
         value.locks = {}
+        value.lock_guard = threading.Lock()
+        value.start_lock = threading.Lock()
         value.completed = {}
         value.operations = []
         return value
 
-    def test_containment_orders_direct_kill_before_durable_writes(self) -> None:
+    def test_containment_orders_direct_kill_before_receipt_state_and_cleanup(self) -> None:
         supervisor = self._instance()
-        saved: list[str] = []
-        response = {"result": "TERMINATED"}
-        with patch("lumi_nutcracker.supervisor.validate_identity", return_value=Path("/owned")), patch("lumi_nutcracker.supervisor.kill_path", return_value=(11, 12)), patch("lumi_nutcracker.supervisor.verify_empty", return_value=(13, EmptyProof(True, 1, 0, []))), patch.object(supervisor, "_cleanup", return_value={}), patch("lumi_nutcracker.supervisor.make_receipt", return_value=response), patch("lumi_nutcracker.supervisor.write_atomic", side_effect=lambda *_: saved.append("write")), patch.object(supervisor, "_store", side_effect=lambda *_: supervisor.operations.append("durable-state")):
+        response = {"result": "TERMINATED", "cleanup": {"attempted": False}}
+        with patch("lumi_eggcracker.supervisor.validate_identity", return_value=Path("/owned")), patch("lumi_eggcracker.supervisor.kill_path", return_value=(11, 12)), patch("lumi_eggcracker.supervisor.verify_empty", return_value=(13, EmptyProof(True, 1, 0, []))), patch.object(supervisor, "_cleanup", side_effect=lambda _: supervisor.operations.append("cleanup") or {}), patch("lumi_eggcracker.supervisor.make_receipt", return_value=response), patch("lumi_eggcracker.supervisor.write_atomic"), patch.object(supervisor, "_store", side_effect=lambda *_: supervisor.operations.append("durable-state")):
             result = supervisor._contain(record(), "OPERATOR", 10)
         self.assertEqual("TERMINATED", result["result"])
         self.assertEqual("cgroup.kill", supervisor.operations[0])
         self.assertLess(supervisor.operations.index("cgroup.kill"), supervisor.operations.index("durable-receipt"))
-        self.assertLess(supervisor.operations.index("cgroup.kill"), supervisor.operations.index("durable-state"))
-        self.assertEqual(["write"], saved)
+        self.assertLess(supervisor.operations.index("durable-receipt"), supervisor.operations.index("cleanup"))
 
-    def test_normal_completion_is_reconciled_when_cgroup_is_collected(self) -> None:
+    def test_exact_empty_cgroup_allows_normal_completion_without_systemctl(self) -> None:
         supervisor = self._instance()
         saved: list[dict[str, object]] = []
-        with patch.object(supervisor, "_show", return_value={"ActiveState": "inactive"}), patch.object(supervisor, "_store", side_effect=lambda value: saved.append(value.copy())):
-            self.assertTrue(supervisor._complete_allowed(record()))
+        with patch("lumi_eggcracker.supervisor.events_from_fd", return_value={"populated": 0}), patch.object(supervisor, "_store", side_effect=lambda value: saved.append(value.copy())):
+            self.assertTrue(supervisor._complete_allowed(record(), 42))
         self.assertEqual("COMPLETED_ALLOWED", saved[0]["state"])
+
+    def test_status_is_read_only(self) -> None:
+        supervisor = self._instance()
+        item = record()
+        with patch.object(supervisor, "_load", return_value=item), patch.object(supervisor, "_store") as stored:
+            value = supervisor.handle({"action": "status", "args": {"name": "demo"}})
+        self.assertEqual("RUNNING", value["state"])
+        stored.assert_not_called()
+
+    def test_status_resolves_the_latest_terminal_run_after_name_reuse_is_enabled(self) -> None:
+        supervisor = self._instance()
+        item = record(); item["state"] = "COMPLETED_ALLOWED"
+        with patch.object(supervisor, "_load", side_effect=JsonInputError("gone")), patch.object(supervisor, "_latest_by_name", return_value=item), patch.object(supervisor, "_store") as stored:
+            value = supervisor.handle({"action": "status", "args": {"name": "demo"}})
+        self.assertEqual("COMPLETED_ALLOWED", value["state"])
+        stored.assert_not_called()
+
+    def test_run_records_are_keyed_by_run_id_and_name_pointer_is_removed_when_terminal(self) -> None:
+        supervisor = self._instance()
+        with tempfile.TemporaryDirectory() as raw:
+            root = Path(raw)
+            supervisor.runs = root / "runs"; supervisor.names = root / "names"
+            item = record()
+            supervisor._store(item)
+            self.assertTrue((supervisor.runs / ("a" * 24 + ".json")).is_file())
+            self.assertTrue((supervisor.names / "demo.json").is_file())
+            item["state"] = "TERMINATED"
+            supervisor._store(item)
+            self.assertFalse((supervisor.names / "demo.json").exists())