fix(import): archive probe → subprocess (was multiprocessing.Process)

Every archive import was failing immediately with "AssertionError:
daemonic processes are not allowed to have children" (operator-flagged
2026-05-30 — import_archive_file crashes in 49ms with the assertion).
Celery's prefork pool runs tasks in daemon processes; Python's
multiprocessing module refuses to let daemons spawn children, which is
exactly what probe_archive was doing via mp.get_context("spawn")
.Process. The Layer-3 crash-isolation feature added 2026-05-28 was
effectively a hard-blocker on the very import path it was meant to
protect.

Switched probe_archive to subprocess.run — no daemon restriction, still
isolates the probe (a probe segfault/OOM exits non-zero, doesn't kill
the worker). The probe body lifted to a tiny runner module
(_archive_probe_runner) that imports the unchanged _run_probe helper
and prints a single JSON line; parent parses stdout, returns the
ProbeResult exactly as before (timeout, signal, OOM, clean-rejection,
ok — all preserved).

cwd for the subprocess is the repo root derived from __file__ parents
so `python -m backend.app.utils._archive_probe_runner` resolves both in
the Celery container and pytest, regardless of where the worker was
launched.

Test refactor: test_archive_probe_target_bomb_guard →
test_run_probe_bomb_guard. Same in-process call to the (renamed) probe
body so the monkeypatched cap still takes effect; the real subprocess
path is exercised by the existing test_probe_archive_valid_zip /
test_probe_archive_corrupt_zip_clean_rejection tests.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/utils/_archive_probe_runner.py

@@ -0,0 +1,33 @@
+"""Subprocess entrypoint for safe_probe.probe_archive — see safe_probe.py.
+
+probe_archive spawns this via subprocess (not multiprocessing.Process)
+because Celery's prefork worker pool runs tasks in DAEMON processes and
+Python's multiprocessing forbids daemon processes from spawning children
+("AssertionError: daemonic processes are not allowed to have children",
+operator-flagged 2026-05-30 — every archive import failed at task
+startup). subprocess has no such restriction; we still get crash-
+isolation because a probe segfault/OOM exits non-zero rather than
+killing the worker.
+
+Prints a single JSON line on stdout: {"status": "ok"|"error",
+"detail": "..."?}. Exit code 0 for clean outcomes; non-zero exit
+(signal / OOM-kill / unhandled exception) is the poison-pill signature
+the parent maps to ProbeResult(crashed=True).
+"""
+import json
+import sys
+
+from .safe_probe import _run_probe
+
+
+def main() -> int:
+    if len(sys.argv) != 2:
+        print(json.dumps({"status": "error", "detail": "usage: <path>"}))
+        return 2
+    status, detail = _run_probe(sys.argv[1])
+    print(json.dumps({"status": status, "detail": detail}))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backend/app/utils/safe_probe.py

@@ -28,8 +28,8 @@ Operator-requested 2026-05-28 (Layer 3).
 """
 
 import json
-import multiprocessing as mp
 import subprocess
+import sys
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -40,6 +40,13 @@ ARCHIVE_PROBE_TIMEOUT_SECONDS = 120
 # art-pack archives while stopping a few-KB zip that expands to TB).
 MAX_ARCHIVE_UNCOMPRESSED_BYTES = 4 * 1024 * 1024 * 1024
 
+# Repo root for the subprocess cwd so `python -m backend.app.utils.*`
+# resolves regardless of where Celery / pytest started. backend/app/utils
+# = parents[0]; backend/app = parents[1]; backend = parents[2]; repo root
+# = parents[3].
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+_PROBE_RUNNER_MODULE = "backend.app.utils._archive_probe_runner"
+
 
 @dataclass(frozen=True)
 class ProbeResult:
@@ -91,54 +98,68 @@ def probe_video(path: Path, *, timeout: float = VIDEO_PROBE_TIMEOUT_SECONDS) ->
 
 
 def probe_archive(path: Path, *, timeout: float = ARCHIVE_PROBE_TIMEOUT_SECONDS) -> ProbeResult:
-    """Bomb-size guard + isolated integrity test for an archive."""
-    ctx = mp.get_context("spawn")
-    q = ctx.Queue()
-    proc = ctx.Process(target=_archive_probe_target, args=(str(path), q))
-    proc.start()
-    proc.join(timeout)
-    if proc.is_alive():
-        proc.terminate()
-        proc.join(5)
+    """Bomb-size guard + isolated integrity test for an archive.
+
+    Runs via subprocess (not multiprocessing.Process) because Celery's
+    prefork worker pool is daemon-mode and Python's multiprocessing
+    forbids daemon processes from spawning children ("AssertionError:
+    daemonic processes are not allowed to have children"). subprocess
+    has no such restriction and still gives the crash isolation: a probe
+    segfault/OOM exits non-zero rather than killing the worker.
+    """
+    try:
+        result = subprocess.run(
+            [sys.executable, "-m", _PROBE_RUNNER_MODULE, str(path)],
+            capture_output=True, text=True, timeout=timeout,
+            cwd=str(_REPO_ROOT),
+        )
+    except subprocess.TimeoutExpired:
         return ProbeResult(ok=False, crashed=True, reason="archive probe timed out")
-    if proc.exitcode != 0:
-        # Negative exitcode = killed by signal (segfault); positive =
-        # the child os._exit'd or was OOM-killed. Either way the file
-        # hard-crashed the probe — the poison-pill signature.
+    if result.returncode != 0:
+        # Negative = killed by signal (segfault); positive = unhandled
+        # exception or OOM-kill. Either way: poison-pill signature.
         return ProbeResult(
             ok=False, crashed=True,
-            reason=f"archive probe crashed (exit {proc.exitcode})",
+            reason=f"archive probe crashed (exit {result.returncode})",
         )
+    last_line = result.stdout.strip().splitlines()[-1:] or [""]
     try:
-        outcome = q.get(timeout=5)
-    except Exception:  # noqa: BLE001 — empty queue / broken pipe
-        return ProbeResult(ok=False, crashed=True, reason="archive probe produced no result")
-    status, detail = outcome
-    if status == "ok":
+        outcome = json.loads(last_line[0])
+    except json.JSONDecodeError as exc:
+        return ProbeResult(
+            ok=False, crashed=True,
+            reason=f"archive probe produced no parseable result: {exc}",
+        )
+    if outcome.get("status") == "ok":
         return ProbeResult(ok=True)
-    return ProbeResult(ok=False, crashed=False, reason=detail)
+    return ProbeResult(
+        ok=False, crashed=False,
+        reason=outcome.get("detail") or "archive probe rejected",
+    )
 
 
-def _archive_probe_target(path_str: str, q) -> None:
-    """Runs in the spawned child. Reads member sizes (bomb guard) then
-    runs the format's integrity test. Puts ('ok', None) or
-    ('error', reason). A crash/OOM here never reaches the queue — the
-    parent reads the non-zero exit code instead."""
+def _run_probe(path_str: str) -> tuple[str, str | None]:
+    """Pure-Python body of the archive probe — bomb-guard + integrity test.
+
+    Returns ('ok', None) or ('error', reason). Caught exceptions become
+    clean 'error' rejections; uncaught crashes in the subprocess become
+    non-zero exit codes (poison-pill signature) handled by probe_archive.
+
+    Exposed at the module level so the subprocess runner and tests both
+    call the same code path.
+    """
     path = Path(path_str)
     ext = path.suffix.lower()
     try:
         total, test_bad = _inspect_archive(path, ext)
     except Exception as exc:  # noqa: BLE001 — clean rejection
-        q.put(("error", f"{type(exc).__name__}: {exc}"))
-        return
+        return ("error", f"{type(exc).__name__}: {exc}")
     if total is not None and total > MAX_ARCHIVE_UNCOMPRESSED_BYTES:
         gib = total / (1024 ** 3)
-        q.put(("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap"))
-        return
+        return ("error", f"uncompressed size {gib:.1f} GiB exceeds the bomb-guard cap")
     if test_bad is not None:
-        q.put(("error", f"integrity test failed at member {test_bad!r}"))
-        return
-    q.put(("ok", None))
+        return ("error", f"integrity test failed at member {test_bad!r}")
+    return ("ok", None)
 
 
 def _inspect_archive(path: Path, ext: str):

tests/test_safe_probe.py

@@ -1,12 +1,11 @@
 """Layer-3 subprocess-isolated probe tests.
 
-The bomb-guard cap is exercised against `_archive_probe_target` directly
-(in-process, where a monkeypatch on the module constant takes effect) —
-spawn re-imports the module in the child, so a parent-process
-monkeypatch wouldn't reach the spawned worker.
+The bomb-guard cap is exercised against `_run_probe` directly (in-process,
+where a monkeypatch on the module constant takes effect) — the real probe
+runs in a subprocess that re-imports the module, so a parent-process
+monkeypatch wouldn't reach the spawned interpreter.
 """
 
-import multiprocessing as mp
 import zipfile
 from pathlib import Path
 
@@ -46,16 +45,14 @@ def test_inspect_archive_reports_size_and_clean_integrity(tmp_path):
     assert bad is None
 
 
-def test_archive_probe_target_bomb_guard(tmp_path, monkeypatch):
-    """In-process call to the child target so the monkeypatched cap
-    takes effect. A normal zip whose uncompressed size exceeds the
-    (lowered) cap is rejected with the bomb-guard reason."""
+def test_run_probe_bomb_guard(tmp_path, monkeypatch):
+    """In-process call to _run_probe so the monkeypatched cap takes effect.
+    The public probe_archive runs in a subprocess which re-imports the
+    module and wouldn't see the patched constant."""
     monkeypatch.setattr(safe_probe, "MAX_ARCHIVE_UNCOMPRESSED_BYTES", 10)
     z = tmp_path / "bomb.zip"
     _zip(z, {"big.txt": b"x" * 5000})  # 5000 uncompressed > 10-byte cap
-    q = mp.get_context("spawn").Queue()
-    safe_probe._archive_probe_target(str(z), q)
-    status, detail = q.get(timeout=5)
+    status, detail = safe_probe._run_probe(str(z))
     assert status == "error"
     assert "bomb-guard cap" in detail