fix: reduce receipt startup and repair APT cache fixtures

docs/APT-CACHE.md

@@ -166,6 +166,17 @@ also checks project images for endpoint leakage and tests outage failure and
 direct HTTPS. It removes only its own containers, images, and
 builders. It never clears or changes a production cache.
 
+The disposable cache uses the pinned Go build image's APT 2.6.1 client identifier
+for outbound requests. The disposable cache returned 404 for a signed Bookworm
+index with its default identifier, while the APT-compatible identifier
+served the exact signed bytes. This fixture setting changes no package pins,
+client timeouts, cache-outage checks or operator-managed cache configuration.
+
+The tooling image installs its dependencies over HTTPS from the same reviewed
+snapshot. A fresh HTTP install returned 503 while the HTTPS install completed
+with the same package selection. The build restores its original HTTP source
+afterward so the cache and outage checks still exercise HTTP proxy behavior.
+
 Before live acceptance, repeat representative shards on the approved host.
 Record elapsed time, upstream package bytes, cache-served package bytes, and
 repeated package hits. Report index revalidation separately. `apt-get update`

docs/LOCAL-VALIDATION.md

@@ -201,6 +201,9 @@ records that executor revision. An unrelated sibling helper cannot execute
 stages, and a mismatched installation fails before launching any service.
 The executor reads a root-protected `ExecStopPost` receipt and verifies the
 service cgroup is empty before recording success or starting another stage.
+The completion hook uses isolated Python without site initialization or optional
+module imports to reduce startup work within the same cleanup allowance. It writes the
+same three manager fields as JSON, rejecting unexpected field bytes or lengths.
 Missing receipts, unsuccessful service results, forced termination and incomplete
 cleanup fail validation. There is no `/proc` walk, pidfd table or child-subreaper
 in the executor. Ordinary configuration stages cannot write the trusted policy,

scripts/fixtures/apt-cache/Dockerfile

@@ -3,8 +3,13 @@ ARG RUNNER_IMAGE=ci-fleet-runner:dev
 FROM ${RUNNER_IMAGE}
 USER root
 # Test tooling only. Package inputs use the runner's reviewed Debian snapshot.
-RUN apt-get update && apt-get install -y --no-install-recommends apt-cacher-ng python3 \
+# Restore the HTTP source afterward for the cache and outage checks.
+RUN cp /etc/apt/sources.list /tmp/ci-fleet-fixture-sources.list \
+    && sed -i 's#http://snapshot.debian.org/#https://snapshot.debian.org/#' /etc/apt/sources.list \
+    && apt-get update && apt-get install -y --no-install-recommends apt-cacher-ng python3 \
+    && mv /tmp/ci-fleet-fixture-sources.list /etc/apt/sources.list \
     && install -d -o apt-cacher-ng -g apt-cacher-ng /var/cache/apt-cacher-ng /var/log/apt-cacher-ng /run/apt-cacher-ng \
     && rm -rf /var/lib/apt/lists/*
 ENTRYPOINT ["/usr/sbin/apt-cacher-ng"]
-CMD ["-c", "/etc/apt-cacher-ng", "ForeGround=1", "BindAddress=0.0.0.0"]
+# Use the pinned APT client's identifier for snapshot index requests.
+CMD ["-c", "/etc/apt-cacher-ng", "ForeGround=1", "BindAddress=0.0.0.0", "UserAgent=Debian APT-HTTP/1.3 (2.6.1)"]

scripts/local_validation_executor.py

@@ -24,11 +24,15 @@
 HOST_ENV = {"PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C.UTF-8"}
 # ExecStopPost runs after systemd has terminated the stage's remaining processes.
 # Its receipt lives in a root-only directory, outside candidate-writable paths.
-FINISH = '''import json, os
-from pathlib import Path
-result = {key: os.environ.get(key) for key in ("SERVICE_RESULT", "EXIT_CODE", "EXIT_STATUS")}
-with Path(__file__).with_name("result.json").open("x") as output:
-    json.dump(result, output)
+FINISH = '''import os
+names = ("SERVICE_RESULT", "EXIT_CODE", "EXIT_STATUS")
+values = [os.environ.get(name) for name in names]
+allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-"
+if any(value is not None and (len(value) > 64 or any(char not in allowed for char in value)) for value in values):
+    raise ValueError("invalid system manager receipt field")
+fields = ['"' + name + '":' + ('null' if value is None else '"' + value + '"') for name, value in zip(names, values)]
+with open(__file__.rsplit("/", 1)[0] + "/result.json", "x", encoding="utf-8") as output:
+    output.write("{" + ",".join(fields) + "}")
 '''
 
 
@@ -112,7 +116,7 @@ def stop():
         properties = ["Type=exec", "ExitType=cgroup", "Slice=system.slice", "Restart=no", "Delegate=no",
                       "KillMode=control-group", "SendSIGKILL=yes", "RuntimeMaxSec=" + str(timeout) + "s",
                       "TimeoutStopSec=" + str(max(grace, .001)) + "s", "TasksMax=512",
-                      "User=" + user, "ExecStopPost=+/usr/bin/python3 -I " + str(finish)]
+                      "User=" + user, "ExecStopPost=+/usr/bin/python3 -I -S " + str(finish)]
         if user != "root":
             writable = list(dict.fromkeys([str(cwd), *(env[key] for key in ("HOME", "TMPDIR") if key in env)]))
             if any(not Path(path).is_absolute() for path in writable):

scripts/test_local_validation_executor.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import importlib.util
+import builtins
 import hashlib
 import json
 import os
@@ -39,6 +40,37 @@ def wait_ready():
 '''
 
 
+class ReceiptTests(unittest.TestCase):
+    def test_receipt_writer_uses_only_os_and_preserves_manager_fields(self):
+        original_import = builtins.__import__
+
+        def frozen_import(name, *args, **kwargs):
+            if name != "os":
+                raise ImportError("completion hook must not load " + name)
+            return original_import(name, *args, **kwargs)
+
+        for values in ({"SERVICE_RESULT": "success", "EXIT_CODE": "exited", "EXIT_STATUS": "0"},
+                       {"SERVICE_RESULT": "timeout", "EXIT_CODE": "killed", "EXIT_STATUS": "TERM"},
+                       {}):
+            with self.subTest(values=values), tempfile.TemporaryDirectory() as temporary:
+                finish = Path(temporary) / "finish.py"
+                with patch.dict(os.environ, values, clear=True), \
+                        patch.object(builtins, "__import__", side_effect=frozen_import):
+                    exec(compile(executor.FINISH, str(finish), "exec"), {"__file__": str(finish)})
+                expected = {name: values.get(name) for name in ("SERVICE_RESULT", "EXIT_CODE", "EXIT_STATUS")}
+                self.assertEqual(json.loads(finish.with_name("result.json").read_text()), expected)
+                with patch.dict(os.environ, values, clear=True), self.assertRaises(FileExistsError):
+                    exec(compile(executor.FINISH, str(finish), "exec"), {"__file__": str(finish)})
+
+    def test_receipt_writer_refuses_unexpected_bytes_without_a_receipt(self):
+        for value in ('success"', "success\n", "x" * 65):
+            with self.subTest(value=value), tempfile.TemporaryDirectory() as temporary:
+                finish = Path(temporary) / "finish.py"
+                with patch.dict(os.environ, {"SERVICE_RESULT": value}, clear=True), self.assertRaises(ValueError):
+                    exec(compile(executor.FINISH, str(finish), "exec"), {"__file__": str(finish)})
+                self.assertFalse(finish.with_name("result.json").exists())
+
+
 class AdmissionTests(unittest.TestCase):
     def test_job_account_rejects_privileged_primary_and_supplementary_groups_before_launch(self):
         account = SimpleNamespace(pw_uid=1001, pw_gid=1001, pw_shell="/usr/sbin/nologin")
@@ -125,7 +157,9 @@ def test_missing_failed_or_malformed_manager_receipts_cannot_pass(self):
                         {"SERVICE_RESULT": "success", "EXIT_CODE": "exited", "EXIT_STATUS": "7"}):
             with self.subTest(outcome=outcome), tempfile.TemporaryDirectory() as temporary:
                 def launch(command, **_):
-                    finish = Path(next(value for value in command if value.startswith("--property=ExecStopPost=")).split()[-1])
+                    hook = next(value for value in command if value.startswith("--property=ExecStopPost="))
+                    self.assertIn(" -I -S ", hook)
+                    finish = Path(hook.split()[-1])
                     if outcome is not None:
                         finish.with_name("result.json").write_text(json.dumps(outcome))
                     return type("Process", (), {"returncode": 0, "poll": lambda _: 0})()
@@ -197,6 +231,18 @@ def test_success_has_exact_output_and_no_inherited_environment(self):
         self.assertEqual(self.log.read_text(), "$HOME %n literal\n")
         self.assert_empty()
 
+    def test_timeout_with_short_cleanup_produces_a_receipt(self):
+        with self.assertRaises(subprocess.TimeoutExpired):
+            self.execute('''
+if os.fork() == 0:
+    os.setsid(); record()
+    signal.signal(signal.SIGTERM, signal.SIG_IGN)
+    ready(); time.sleep(60)
+else:
+    wait_ready(); time.sleep(60)
+''', timeout=1, grace=.2)
+        self.assert_empty()
+
     def test_detached_child_is_cleaned_before_next_stage(self):
         self.execute('''
 if os.fork() == 0: