From 175bf00802b387f011d80988823b428ff133db4c Mon Sep 17 00:00:00 2001
From: Sebastiano Merlino <sebastiano@hey.com>
Date: Thu, 15 Oct 2026 18:35:01 -0700
Subject: [PATCH 1/5] FEATURE-ci-process-cleanup: observe native macOS cleanup
 failure

---
 .github/workflows/v3-cleanup-diagnostic.yml | 130 ++++++++++++++++++++
 1 file changed, 130 insertions(+)
 create mode 100644 .github/workflows/v3-cleanup-diagnostic.yml

diff --git a/.github/workflows/v3-cleanup-diagnostic.yml b/.github/workflows/v3-cleanup-diagnostic.yml
new file mode 100644
index 000000000..da0742881
--- /dev/null
+++ b/.github/workflows/v3-cleanup-diagnostic.yml
@@ -0,0 +1,130 @@
+name: v3 cleanup diagnostic
+on:
+  pull_request:
+    branches: [v3]
+    paths: ['.github/workflows/v3-cleanup-diagnostic.yml']
+permissions:
+  contents: read
+jobs:
+  macos-cleanup:
+    runs-on: macos-14
+    timeout-minutes: 5
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
+      - name: Match Python prerequisite
+        run: brew install python
+      - name: Observe actual helper and native zombie signal semantics
+        timeout-minutes: 2
+        shell: bash
+        run: |
+          cat > "$RUNNER_TEMP/v3-cleanup-native-probe.py" <<'PY'
+          #!/usr/bin/env python3
+          """Bounded observational probe: execute the checked-out helper unchanged."""
+          import argparse
+          import importlib.util
+          import json
+          import os
+          from pathlib import Path
+          import platform
+          import select
+          import signal
+          import sys
+          import time
+          import traceback
+          
+          parser = argparse.ArgumentParser()
+          parser.add_argument('--source', type=Path, default=Path.cwd())
+          parser.add_argument('--output', type=Path, required=True)
+          args = parser.parse_args()
+          source = args.source.resolve()
+          args.output.mkdir(parents=True, exist_ok=False)
+          sys.path.insert(0, str(source / 'scripts/lib'))
+          import package_commands as commands
+          spec = importlib.util.spec_from_file_location('audit', source / 'scripts/audit-v3-installed-package.py')
+          audit = importlib.util.module_from_spec(spec)
+          spec.loader.exec_module(audit)
+          
+          receipt = dict(platform=platform.platform(), uname=list(platform.uname()), python=sys.version,
+                         waitid=callable(getattr(os, 'waitid', None)), wnowait=hasattr(os, 'WNOWAIT'),
+                         source=str(source), uid=os.getuid(), observations=[], commands=[], controls=[])
+          
+          def state(tree, deadline):
+              members = commands._process_snapshot(deadline)
+              tree._check_leader(members.get(tree.child.pid))
+              return dict(pid=tree.child.pid, parent=os.getpid(), returncode=tree.child.returncode,
+                          anchor_exited=tree._anchor(deadline),
+                          group={str(pid): value for pid, value in members.items() if value[1] == tree.child.pid})
+          
+          original_kill_group = commands._CommandTree._kill_group
+          
+          def observed_kill_group(tree, deadline):
+              item = state(tree, deadline)
+              receipt['observations'].append(item)
+              try:
+                  result = original_kill_group(tree, deadline)
+                  item['signal'] = 'success'
+                  return result
+              except BaseException as error:
+                  item.update(signal=type(error).__name__, errno=getattr(error, 'errno', None), error=str(error))
+                  item['after'] = state(tree, deadline)
+                  raise
+          
+          # Observation only; the original helper performs every signal and cleanup.
+          commands._CommandTree._kill_group = observed_kill_group
+          runner = commands.Commands(args.output)
+          for argv in (['git', 'branch', '--show-current'], [sys.executable, '-c', 'pass'],
+                       [sys.executable, '-c', 'raise SystemExit(7)']):
+              item = dict(argv=argv)
+              try:
+                  item['output'] = runner(argv, source, audit.clean_environment(), timeout=3, allow_failure=True)
+                  item['result'] = 'success'
+              except Exception as error:
+                  item.update(result=type(error).__name__, error=str(error), traceback=traceback.format_exc())
+              receipt['commands'].append(item)
+          
+          # A fresh owned live leader is the contrasting permission control.
+          tree = commands._CommandTree([sys.executable, '-u', '-c', "import time; print('ready', flush=True); time.sleep(30)"], source, os.environ)
+          deadline = time.monotonic() + 5
+          control = dict(kind='live-to-zombie')
+          try:
+              if not select.select([tree.child.stdout], [], [], 3)[0]:
+                  raise RuntimeError('live control readiness deadline exceeded')
+              if tree.child.stdout.readline() != b'ready\n':
+                  raise RuntimeError('live control did not become ready')
+              control['live'] = state(tree, deadline)
+              original_kill_group(tree, deadline)
+              control['live_sigkill'] = 'success'
+              while not tree.exited(deadline):
+                  commands._remaining(deadline)
+                  time.sleep(.001)
+              control['zombie'] = state(tree, deadline)
+              for label, operation in (
+                      ('zombie_pid_signal_zero', lambda: os.kill(tree.child.pid, 0)),
+                      ('zombie_group_signal_zero', lambda: os.killpg(tree.child.pid, 0)),
+                      ('zombie_helper_sigkill', lambda: original_kill_group(tree, deadline))):
+                  tree._anchor(deadline)
+                  try:
+                      operation()
+                      control[label] = dict(result='success')
+                  except Exception as error:
+                      control[label] = dict(result=type(error).__name__, errno=getattr(error, 'errno', None), error=str(error))
+              control['after'] = state(tree, deadline)
+          except Exception as error:
+              control.update(error=str(error), traceback=traceback.format_exc())
+          finally:
+              # Only our unreaped direct child; never signal after wait/reap.
+              if tree.child.returncode is None:
+                  tree.child.kill()
+                  tree.child.wait(timeout=2)
+              tree.child.stdout.close()
+          receipt['controls'].append(control)
+          (args.output / 'probe.json').write_text(json.dumps(receipt, indent=2) + '\n')
+          print(json.dumps(receipt, indent=2), flush=True)
+          PY
+          python3 "$RUNNER_TEMP/v3-cleanup-native-probe.py" --source "$GITHUB_WORKSPACE" --output "$RUNNER_TEMP/v3-cleanup-diagnostic"
+      - if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
+        with:
+          name: v3-cleanup-diagnostic-${{ github.sha }}
+          path: ${{ runner.temp }}/v3-cleanup-diagnostic
+          if-no-files-found: error

From 2848ebfe84ccbd63fe6cf77157cc6cd31ee99b8c Mon Sep 17 00:00:00 2001
From: Sebastiano Merlino <sebastiano@hey.com>
Date: Thu, 15 Oct 2026 18:50:07 -0700
Subject: [PATCH 2/5] FEATURE-ci-process-cleanup: Verify zombie retirement
 after denied POSIX group signals

Accept Darwin zombie-only EPERM only after bounded process discovery and ownership checks prove the group has no live members. Retain stable retirement, the shared cleanup budget, final leader reap, and original command receipts and failures.

Add real-process permission regressions and prepare source-bound native macOS contract and semantic-probe verification. Local installed-consumer contracts pass (45 tests), as do 120 retained affinity-controlled Linux overflow/socket trials. Native proof awaits publication.
---
 .github/workflows/v3-cleanup-diagnostic.yml |  89 +++++++++++++++-
 scripts/lib/package_commands.py             |  10 ++
 scripts/test-v3-installed-consumer.py       | 107 ++++++++++++++++++++
 3 files changed, 203 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/v3-cleanup-diagnostic.yml b/.github/workflows/v3-cleanup-diagnostic.yml
index da0742881..abd565e8b 100644
--- a/.github/workflows/v3-cleanup-diagnostic.yml
+++ b/.github/workflows/v3-cleanup-diagnostic.yml
@@ -2,7 +2,7 @@ name: v3 cleanup diagnostic
 on:
   pull_request:
     branches: [v3]
-    paths: ['.github/workflows/v3-cleanup-diagnostic.yml']
+    paths: ['.github/workflows/v3-cleanup-diagnostic.yml', 'scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py']
 permissions:
   contents: read
 jobs:
@@ -13,7 +13,27 @@ jobs:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
       - name: Match Python prerequisite
         run: brew install python
+      - name: Run native command contracts
+        timeout-minutes: 2
+        shell: bash
+        run: |
+          mkdir -p "$RUNNER_TEMP/v3-cleanup-contracts"
+          python3 - <<'PYTHON'
+          import hashlib, json, os, subprocess, sys
+          from pathlib import Path
+          source = Path.cwd()
+          paths = ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py')
+          receipt = dict(source_sha=subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip(),
+                         python=sys.version, hashes={p: hashlib.sha256((source / p).read_bytes()).hexdigest() for p in paths})
+          Path(os.environ['RUNNER_TEMP'], 'v3-cleanup-contracts/source.json').write_text(json.dumps(receipt, indent=2) + '\n')
+          PYTHON
+          set +e
+          python3 scripts/test-v3-installed-consumer.py CommandsTest 2>&1 | tee "$RUNNER_TEMP/v3-cleanup-contracts/commands-test.log"
+          status=${PIPESTATUS[0]}
+          printf '%s\n' "$status" > "$RUNNER_TEMP/v3-cleanup-contracts/exit-status"
+          exit "$status"
       - name: Observe actual helper and native zombie signal semantics
+        if: always()
         timeout-minutes: 2
         shell: bash
         run: |
@@ -22,12 +42,14 @@ jobs:
           """Bounded observational probe: execute the checked-out helper unchanged."""
           import argparse
           import importlib.util
+          import hashlib
           import json
           import os
           from pathlib import Path
           import platform
           import select
           import signal
+          import subprocess
           import sys
           import time
           import traceback
@@ -44,7 +66,10 @@ jobs:
           audit = importlib.util.module_from_spec(spec)
           spec.loader.exec_module(audit)
           
-          receipt = dict(platform=platform.platform(), uname=list(platform.uname()), python=sys.version,
+          receipt = dict(source_sha=subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=source, text=True).strip(),
+                         hashes={p: hashlib.sha256((source / p).read_bytes()).hexdigest() for p in
+                                 ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py')},
+                         platform=platform.platform(), uname=list(platform.uname()), python=sys.version,
                          waitid=callable(getattr(os, 'waitid', None)), wnowait=hasattr(os, 'WNOWAIT'),
                          source=str(source), uid=os.getuid(), observations=[], commands=[], controls=[])
           
@@ -122,9 +147,67 @@ jobs:
           print(json.dumps(receipt, indent=2), flush=True)
           PY
           python3 "$RUNNER_TEMP/v3-cleanup-native-probe.py" --source "$GITHUB_WORKSPACE" --output "$RUNNER_TEMP/v3-cleanup-diagnostic"
+      - name: Require source-bound native cleanup proof
+        if: always()
+        shell: bash
+        run: |
+          python3 - <<'PY'
+          import hashlib, json, os, subprocess
+          from pathlib import Path
+          root = Path(os.environ['RUNNER_TEMP'])
+          output = root / 'v3-cleanup-diagnostic'
+          receipt = json.loads((output / 'probe.json').read_text())
+          contracts = root / 'v3-cleanup-contracts'
+          binding = json.loads((contracts / 'source.json').read_text())
+          source_sha = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip()
+          hashes = {p: hashlib.sha256(Path(p).read_bytes()).hexdigest() for p in
+                    ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py')}
+          assert receipt['source_sha'] == binding['source_sha'] == source_sha == os.environ['GITHUB_SHA']
+          assert receipt['hashes'] == binding['hashes'] == hashes
+          assert receipt['python'] == binding['python']
+          assert (contracts / 'exit-status').read_text().strip() == '0'
+          assert '\nOK\n' in (contracts / 'commands-test.log').read_text()
+          assert receipt['uname'][0] == 'Darwin'
+          assert len(receipt['commands']) == 3
+          assert all(item['result'] == 'success' for item in receipt['commands'])
+          records = json.loads((output / 'commands.json').read_text())
+          assert len(records) == 3 and [row['exit'] for row in records] == [0, 0, 7]
+          for item, record in zip(receipt['commands'], records):
+              assert 'error' not in record and item['argv'] == record['argv']
+              data = Path(record['log']).read_bytes()
+              assert len(data) == record['bytes']
+              assert hashlib.sha256(data).hexdigest() == record['sha256']
+              assert item['output'] == data.decode(errors='replace')
+          def owned_zombie(item):
+              assert item['returncode'] is None and item['anchor_exited'] is True
+              assert item['group'] == {str(item['pid']): [item['parent'], item['pid'], 'Z']}
+          assert len(receipt['observations']) == 3
+          for item in receipt['observations']:
+              owned_zombie(item)
+              assert item['signal'] == 'success'
+          assert len(receipt['controls']) == 1
+          control = receipt['controls'][0]
+          assert 'error' not in control and control['kind'] == 'live-to-zombie'
+          live = control['live']
+          assert live['returncode'] is None and live['anchor_exited'] is False
+          assert live['group'][str(live['pid'])][:2] == [live['parent'], live['pid']]
+          assert live['group'][str(live['pid'])][2] != 'Z'
+          assert control['live_sigkill'] == 'success'
+          owned_zombie(control['zombie'])
+          owned_zombie(control['after'])
+          assert live['pid'] == control['zombie']['pid'] == control['after']['pid']
+          assert control['zombie_pid_signal_zero']['result'] == 'success'
+          assert control['zombie_group_signal_zero']['result'] == 'PermissionError'
+          assert control['zombie_group_signal_zero']['errno'] == 1
+          assert control['zombie_helper_sigkill']['result'] == 'success'
+          (output / 'semantic-proof.json').write_text(json.dumps(dict(status='passed', source_sha=source_sha,
+              hashes=hashes, command_exits=[row['exit'] for row in records], native_contracts='passed'), indent=2) + '\n')
+          PY
       - if: always()
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
         with:
           name: v3-cleanup-diagnostic-${{ github.sha }}
-          path: ${{ runner.temp }}/v3-cleanup-diagnostic
+          path: |
+            ${{ runner.temp }}/v3-cleanup-diagnostic
+            ${{ runner.temp }}/v3-cleanup-contracts
           if-no-files-found: error
diff --git a/scripts/lib/package_commands.py b/scripts/lib/package_commands.py
index 552f4cea0..c8aed996d 100644
--- a/scripts/lib/package_commands.py
+++ b/scripts/lib/package_commands.py
@@ -158,6 +158,16 @@ def _kill_group(self, deadline):
             os.killpg(self.child.pid, signal.SIGKILL)
         except ProcessLookupError:
             pass  # The snapshot must still prove that no live member remains.
+        except PermissionError:
+            # Darwin excludes zombies from group signals and can return EPERM.
+            self._anchor(deadline)
+            members = _process_snapshot(deadline)
+            self._check_leader(members.get(self.child.pid))
+            self._anchor(deadline)
+            _remaining(deadline)
+            if any(group == self.child.pid and state != 'Z'
+                   for _, group, state in members.values()):
+                raise
 
     def terminate(self, deadline=None):
         if self.job:
diff --git a/scripts/test-v3-installed-consumer.py b/scripts/test-v3-installed-consumer.py
index f39c91c1e..befa6f07c 100644
--- a/scripts/test-v3-installed-consumer.py
+++ b/scripts/test-v3-installed-consumer.py
@@ -285,6 +285,113 @@ def test_success_and_allowed_failure_keep_output_and_hash_receipts(self):
                 self.assertEqual(len(data), record['bytes'])
                 self.assertEqual(hashlib.sha256(data).hexdigest(), record['sha256'])
 
+    @unittest.skipUnless(os.name == 'posix', 'POSIX zombie group permissions')
+    def test_owned_zombie_permission_denial_preserves_completed_receipts(self):
+        snapshot = commands._process_snapshot
+        scans = []
+        denied = []
+        def observe(deadline):
+            members = snapshot(deadline)
+            scans.append(members)
+            return members
+        def signal_group(pid, sig):
+            members = snapshot(time.monotonic() + 2)
+            leader = members.get(pid)
+            self.assertEqual((os.getpid(), pid, 'Z'), leader)
+            self.assertTrue(all(state == 'Z' for _, group, state in members.values() if group == pid))
+            denied.append(pid)
+            raise PermissionError(1, 'zombie group has no signal targets')
+        for fallback in ('native', 'no-waitid', 'ps'):
+            for status in (0, 7):
+                with self.subTest(fallback=fallback, status=status):
+                    scans.clear()
+                    waitid = getattr(os, 'waitid', None) if fallback == 'native' else None
+                    platform = 'darwin' if fallback == 'ps' else sys.platform
+                    with patch.object(commands, '_process_snapshot', observe), patch.object(os, 'killpg', signal_group):
+                        with patch.object(os, 'waitid', waitid, create=True), patch.object(sys, 'platform', platform):
+                            output = self.runner(self.command(f"print('completed'); raise SystemExit({status})"),
+                                                 self.root, os.environ, timeout=2, allow_failure=True)
+                    self.assertEqual('completed\n', output)
+                    record = json.loads((self.root / 'commands.json').read_text())[-1]
+                    data = Path(record['log']).read_bytes()
+                    self.assertEqual(status, record['exit'])
+                    self.assertEqual(len(data), record['bytes'])
+                    self.assertEqual(hashlib.sha256(data).hexdigest(), record['sha256'])
+                    self.assertNotIn('error', record)
+                    self.assertGreaterEqual(len(scans), 3, 'permission proof must retain stable retirement scans')
+        self.assertEqual(6, len(denied))
+
+    def zombie_tree(self):
+        tree = commands._CommandTree(self.command('pass'), self.root, os.environ)
+        self.addCleanup(tree.child.stdout.close)
+        self.addCleanup(lambda: tree.child.wait(timeout=2))
+        deadline = time.monotonic() + 2
+        while not tree.exited(deadline):
+            self.assertLess(time.monotonic(), deadline)
+            time.sleep(.01)
+        self.assertIsNone(tree.child.returncode)
+        return tree
+
+    @unittest.skipUnless(os.name == 'posix', 'POSIX live group permissions')
+    def test_live_group_permission_denial_fails_closed(self):
+        tree = commands._CommandTree(self.command('import time; time.sleep(30)'),
+                                     self.root, os.environ)
+        try:
+            self.assertFalse(tree.exited(time.monotonic() + 2))
+            with patch.object(os, 'killpg', side_effect=PermissionError(1, 'live group denied')):
+                with self.assertRaisesRegex(PermissionError, 'live group denied'):
+                    tree._kill_group(time.monotonic() + 2)
+            self.assertFalse(tree.exited(time.monotonic() + 2))
+        finally:
+            tree.child.kill(); tree.child.wait(timeout=2); tree.child.stdout.close()
+
+    @unittest.skipUnless(os.name == 'posix', 'POSIX permission evidence boundary')
+    def test_zombie_permission_denial_requires_complete_owned_retirement_evidence(self):
+        failures = dict(unavailable='unavailable', malformed='malformed', unknown='unverified group denied',
+                        live='unverified group denied', missing='ownership', changed='ownership',
+                        reaped='ownership', deadline='cleanup deadline exceeded')
+        for failure, message in failures.items():
+            with self.subTest(failure=failure):
+                tree = self.zombie_tree()
+                snapshot = commands._process_snapshot
+                pid = tree.child.pid
+                elapsed = 0
+                def evidence(deadline):
+                    nonlocal elapsed
+                    members = snapshot(deadline)
+                    if failure in ('unavailable', 'malformed'):
+                        raise RuntimeError(failure + ' command process snapshot')
+                    if failure in ('unknown', 'live'):
+                        members[pid + 1] = (pid, pid, '?' if failure == 'unknown' else 'S')
+                    if failure == 'missing': del members[pid]
+                    if failure == 'changed': members[pid] = (os.getpid() + 1, pid, 'Z')
+                    if failure == 'reaped': tree.child.wait(timeout=2)
+                    if failure == 'deadline': elapsed = 2
+                    return members
+                # The initial ownership check is real; corrupt only the post-denial proof.
+                def deny(pid, sig):
+                    commands._process_snapshot = evidence
+                    raise PermissionError(1, 'unverified group denied')
+                clock = (lambda: elapsed) if failure == 'deadline' else time.monotonic
+                with patch.object(commands, '_process_snapshot', snapshot), patch.object(os, 'killpg', side_effect=deny) as kill:
+                    with patch.object(commands.time, 'monotonic', clock):
+                        with self.assertRaisesRegex((PermissionError, RuntimeError), message):
+                            tree._kill_group(1 if failure == 'deadline' else time.monotonic() + 2)
+                    if failure == 'reaped':
+                        self.assertEqual(0, tree.child.returncode)
+                        with self.assertRaisesRegex(RuntimeError, 'ownership'):
+                            tree._kill_group(time.monotonic() + 2)
+                        self.assertEqual(1, kill.call_count, 'lost ownership must prevent another signal')
+
+    @unittest.skipUnless(os.name == 'posix', 'POSIX cleanup error context')
+    def test_live_permission_failure_retains_original_command_timeout(self):
+        with patch.object(os, 'killpg', side_effect=PermissionError(1, 'live group denied')):
+            with self.assertRaisesRegex(RuntimeError, 'cleanup.*live group denied.*original command failure:.*deadline/output'):
+                self.runner(self.command('import time; time.sleep(.5)'), self.root, os.environ, timeout=.01)
+        receipt = json.loads((self.root / 'commands.json').read_text())[-1]
+        self.assertIn('original command failure:', receipt['error'])
+        self.assertNotIn('sha256', receipt)
+
     def assert_success_retires_socket_owner(self):
         ready = self.root / 'descendant.json'
         worker = ("import json,os,socket,time; from pathlib import Path; "

From 944fa196b46684e5b9187510a0050b96f4f31823 Mon Sep 17 00:00:00 2001
From: Sebastiano Merlino <sebastiano@hey.com>
Date: Thu, 15 Oct 2026 19:00:39 -0700
Subject: [PATCH 3/5] FEATURE-ci-process-cleanup: Correct padded ps fixture and
 retain native parser evidence

Trim valid trailing state-column padding before appending fixture modifiers, and assert owned-zombie EPERM reach independently for each receipt case. The real padded fixture reproduced its malformed-snapshot failure twice before this fix and passes afterward.

Retain bounded native snapshot failures and structured results from the complete actual CommandsTest class. Require the three named Linux-only Darwin skips exactly. Local 22 command contracts and 45 installed-consumer contracts pass; helper and Linux affinity fixture bytes are unchanged. Attempt 1 native proof remains incomplete pending fresh execution and raw-row diagnosis.
---
 .github/workflows/v3-cleanup-diagnostic.yml | 62 +++++++++++++++++++--
 scripts/test-v3-installed-consumer.py       |  5 +-
 2 files changed, 60 insertions(+), 7 deletions(-)

diff --git a/.github/workflows/v3-cleanup-diagnostic.yml b/.github/workflows/v3-cleanup-diagnostic.yml
index abd565e8b..fdf997f40 100644
--- a/.github/workflows/v3-cleanup-diagnostic.yml
+++ b/.github/workflows/v3-cleanup-diagnostic.yml
@@ -18,17 +18,60 @@ jobs:
         shell: bash
         run: |
           mkdir -p "$RUNNER_TEMP/v3-cleanup-contracts"
-          python3 - <<'PYTHON'
-          import hashlib, json, os, subprocess, sys
+          cat > "$RUNNER_TEMP/v3-cleanup-command-contracts.py" <<'PYTHON'
+          import hashlib, importlib.util, json, os, subprocess, sys, unittest
           from pathlib import Path
           source = Path.cwd()
+          output = Path(os.environ['RUNNER_TEMP'], 'v3-cleanup-contracts')
           paths = ('scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py')
           receipt = dict(source_sha=subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True).strip(),
                          python=sys.version, hashes={p: hashlib.sha256((source / p).read_bytes()).hexdigest() for p in paths})
-          Path(os.environ['RUNNER_TEMP'], 'v3-cleanup-contracts/source.json').write_text(json.dumps(receipt, indent=2) + '\n')
+          (output / 'source.json').write_text(json.dumps(receipt, indent=2) + '\n')
+          spec = importlib.util.spec_from_file_location('native_contracts', source / paths[1])
+          contracts = importlib.util.module_from_spec(spec)
+          spec.loader.exec_module(contracts)
+          snapshot = contracts.commands._process_snapshot
+          failures = []
+          current_test = None
+          def observed_snapshot(deadline):
+              try:
+                  return snapshot(deadline)
+              except Exception as error:
+                  item = dict(test=current_test, error=str(error))
+                  frame = error.__traceback__
+                  while frame:
+                      if frame.tb_frame.f_code.co_name == '_process_snapshot':
+                          values = frame.tb_frame.f_locals
+                          item.update(line=values.get('line'), fields=values.get('fields'))
+                          buffer = values.get('output')
+                          if buffer is not None:
+                              data = buffer.getvalue()
+                              path = output / ('snapshot-%04d.log' % len(failures))
+                              path.write_bytes(data)
+                              item.update(log=path.name, bytes=len(data), sha256=hashlib.sha256(data).hexdigest())
+                      frame = frame.tb_next
+                  failures.append(item)
+                  (output / 'snapshot-failures.json').write_text(json.dumps(failures, indent=2) + '\n')
+                  raise
+          # Retain parser evidence without replacing discovery, signals, or errors.
+          contracts.commands._process_snapshot = observed_snapshot
+          class Result(unittest.TextTestResult):
+              def startTest(self, test):
+                  global current_test
+                  current_test = test.id()
+                  super().startTest(test)
+          suite = unittest.defaultTestLoader.loadTestsFromTestCase(contracts.CommandsTest)
+          expected = [test.id() for test in suite]
+          result = unittest.TextTestRunner(verbosity=2, resultclass=Result).run(suite)
+          report = dict(expected=expected, tests_run=result.testsRun, successful=result.wasSuccessful(),
+                        failures=[test.id() for test, _ in result.failures],
+                        errors=[test.id() for test, _ in result.errors],
+                        skipped=[dict(test=test.id(), reason=reason) for test, reason in result.skipped])
+          (output / 'result.json').write_text(json.dumps(report, indent=2) + '\n')
+          raise SystemExit(not result.wasSuccessful())
           PYTHON
           set +e
-          python3 scripts/test-v3-installed-consumer.py CommandsTest 2>&1 | tee "$RUNNER_TEMP/v3-cleanup-contracts/commands-test.log"
+          python3 "$RUNNER_TEMP/v3-cleanup-command-contracts.py" 2>&1 | tee "$RUNNER_TEMP/v3-cleanup-contracts/commands-test.log"
           status=${PIPESTATUS[0]}
           printf '%s\n' "$status" > "$RUNNER_TEMP/v3-cleanup-contracts/exit-status"
           exit "$status"
@@ -166,7 +209,16 @@ jobs:
           assert receipt['hashes'] == binding['hashes'] == hashes
           assert receipt['python'] == binding['python']
           assert (contracts / 'exit-status').read_text().strip() == '0'
-          assert '\nOK\n' in (contracts / 'commands-test.log').read_text()
+          result = json.loads((contracts / 'result.json').read_text())
+          assert result['successful'] and not result['failures'] and not result['errors']
+          assert result['tests_run'] == len(result['expected']) == 22
+          assert len(set(result['expected'])) == 22
+          expected_skips = {
+              'native_contracts.CommandsTest.test_cleanup_discovery_failure_retains_original_error',
+              'native_contracts.CommandsTest.test_new_live_member_delays_retirement_and_zombies_do_not',
+              'native_contracts.CommandsTest.test_discovery_consumes_cleanup_deadline_without_restarting_wait'}
+          assert len(result['skipped']) == 3
+          assert {item['test'] for item in result['skipped']} == expected_skips
           assert receipt['uname'][0] == 'Darwin'
           assert len(receipt['commands']) == 3
           assert all(item['result'] == 'success' for item in receipt['commands'])
diff --git a/scripts/test-v3-installed-consumer.py b/scripts/test-v3-installed-consumer.py
index befa6f07c..84c8e6ac4 100644
--- a/scripts/test-v3-installed-consumer.py
+++ b/scripts/test-v3-installed-consumer.py
@@ -305,6 +305,7 @@ def signal_group(pid, sig):
             for status in (0, 7):
                 with self.subTest(fallback=fallback, status=status):
                     scans.clear()
+                    previous_denials = len(denied)
                     waitid = getattr(os, 'waitid', None) if fallback == 'native' else None
                     platform = 'darwin' if fallback == 'ps' else sys.platform
                     with patch.object(commands, '_process_snapshot', observe), patch.object(os, 'killpg', signal_group):
@@ -318,8 +319,8 @@ def signal_group(pid, sig):
                     self.assertEqual(len(data), record['bytes'])
                     self.assertEqual(hashlib.sha256(data).hexdigest(), record['sha256'])
                     self.assertNotIn('error', record)
+                    self.assertGreater(len(denied), previous_denials, 'each case must reach owned-zombie EPERM')
                     self.assertGreaterEqual(len(scans), 3, 'permission proof must retain stable retirement scans')
-        self.assertEqual(6, len(denied))
 
     def zombie_tree(self):
         tree = commands._CommandTree(self.command('pass'), self.root, os.environ)
@@ -513,7 +514,7 @@ def launch(argv, *args, **kwargs):
             if argv[0] in ('/bin/ps', '/usr/bin/ps'):
                 code = ("import subprocess; "
                         f"data=subprocess.check_output({argv!r}).decode('ascii'); "
-                        "print('\\n'.join(line + '>ACJ' for line in data.splitlines()))")
+                        "print('\\n'.join(line.rstrip() + '>ACJ' for line in data.splitlines()))")
                 argv = self.command(code)
             return popen(argv, *args, **kwargs)
         with patch.object(sys, 'platform', 'darwin'), patch.object(os, 'waitid', None, create=True), patch.object(subprocess, 'Popen', launch):

From 38d6678966d07b0409d0dca10e41aab560482c79 Mon Sep 17 00:00:00 2001
From: Sebastiano Merlino <sebastiano@hey.com>
Date: Thu, 15 Oct 2026 19:10:04 -0700
Subject: [PATCH 4/5] FEATURE-ci-process-cleanup: Bound one native receipt
 probe for historical snapshot errors

Execute only the actual owned-zombie receipt contract for up to 30 iterations in a 60-second start window, with failfast and existing bounded command cleanup. Retain offending raw ps rows, source binding, and structured per-iteration results. Preserve the already verified full native contract proof by enforcing its exact helper/test hashes.

Local diagnostic-runner verification passed 30 iterations (180 command cases) in 15.022s. The two historical unmodified native snapshot errors remain unexplained; this probe provides bounded evidence for review rather than a speculative parser fix.
---
 .github/workflows/v3-cleanup-diagnostic.yml | 57 +++++++++++++--------
 1 file changed, 37 insertions(+), 20 deletions(-)

diff --git a/.github/workflows/v3-cleanup-diagnostic.yml b/.github/workflows/v3-cleanup-diagnostic.yml
index fdf997f40..8fbd9656b 100644
--- a/.github/workflows/v3-cleanup-diagnostic.yml
+++ b/.github/workflows/v3-cleanup-diagnostic.yml
@@ -13,13 +13,13 @@ jobs:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
       - name: Match Python prerequisite
         run: brew install python
-      - name: Run native command contracts
+      - name: Bound the unexplained native snapshot observation
         timeout-minutes: 2
         shell: bash
         run: |
           mkdir -p "$RUNNER_TEMP/v3-cleanup-contracts"
           cat > "$RUNNER_TEMP/v3-cleanup-command-contracts.py" <<'PYTHON'
-          import hashlib, importlib.util, json, os, subprocess, sys, unittest
+          import hashlib, importlib.util, json, os, subprocess, sys, time, unittest
           from pathlib import Path
           source = Path.cwd()
           output = Path(os.environ['RUNNER_TEMP'], 'v3-cleanup-contracts')
@@ -60,15 +60,30 @@ jobs:
                   global current_test
                   current_test = test.id()
                   super().startTest(test)
-          suite = unittest.defaultTestLoader.loadTestsFromTestCase(contracts.CommandsTest)
-          expected = [test.id() for test in suite]
-          result = unittest.TextTestRunner(verbosity=2, resultclass=Result).run(suite)
-          report = dict(expected=expected, tests_run=result.testsRun, successful=result.wasSuccessful(),
-                        failures=[test.id() for test, _ in result.failures],
-                        errors=[test.id() for test, _ in result.errors],
-                        skipped=[dict(test=test.id(), reason=reason) for test, reason in result.skipped])
+          started = time.monotonic()
+          trials = []
+          for iteration in range(1, 31):
+              if time.monotonic() - started >= 60:
+                  break
+              suite = unittest.TestSuite([contracts.CommandsTest(
+                  'test_owned_zombie_permission_denial_preserves_completed_receipts')])
+              result = unittest.TextTestRunner(verbosity=2, resultclass=Result, failfast=True).run(suite)
+              trial = dict(iteration=iteration, tests_run=result.testsRun, successful=result.wasSuccessful(),
+                           failures=[test.id() for test, _ in result.failures],
+                           errors=[test.id() for test, _ in result.errors],
+                           skipped=[dict(test=test.id(), reason=reason) for test, reason in result.skipped])
+              trials.append(trial)
+              report = dict(mode='bounded-owned-zombie-receipts', max_iterations=30, max_seconds=60,
+                            elapsed_seconds=time.monotonic() - started, trials=trials,
+                            successful=all(item['successful'] for item in trials),
+                            stop='first-failure' if not result.wasSuccessful() else 'running')
+              (output / 'result.json').write_text(json.dumps(report, indent=2) + '\n')
+              if not result.wasSuccessful():
+                  break
+          report['stop'] = ('first-failure' if not report['successful'] else
+                            'iteration-limit' if len(trials) == 30 else 'time-limit')
           (output / 'result.json').write_text(json.dumps(report, indent=2) + '\n')
-          raise SystemExit(not result.wasSuccessful())
+          raise SystemExit(not report['successful'])
           PYTHON
           set +e
           python3 "$RUNNER_TEMP/v3-cleanup-command-contracts.py" 2>&1 | tee "$RUNNER_TEMP/v3-cleanup-contracts/commands-test.log"
@@ -210,15 +225,17 @@ jobs:
           assert receipt['python'] == binding['python']
           assert (contracts / 'exit-status').read_text().strip() == '0'
           result = json.loads((contracts / 'result.json').read_text())
-          assert result['successful'] and not result['failures'] and not result['errors']
-          assert result['tests_run'] == len(result['expected']) == 22
-          assert len(set(result['expected'])) == 22
-          expected_skips = {
-              'native_contracts.CommandsTest.test_cleanup_discovery_failure_retains_original_error',
-              'native_contracts.CommandsTest.test_new_live_member_delays_retirement_and_zombies_do_not',
-              'native_contracts.CommandsTest.test_discovery_consumes_cleanup_deadline_without_restarting_wait'}
-          assert len(result['skipped']) == 3
-          assert {item['test'] for item in result['skipped']} == expected_skips
+          assert result['mode'] == 'bounded-owned-zombie-receipts' and result['successful']
+          assert result['max_iterations'] == 30 and result['max_seconds'] == 60
+          assert 1 <= len(result['trials']) <= 30
+          assert result['stop'] in ('iteration-limit', 'time-limit')
+          for trial in result['trials']:
+              assert trial['tests_run'] == 1 and trial['successful']
+              assert not trial['failures'] and not trial['errors'] and not trial['skipped']
+          # Full 22-contract native GREEN is retained from run 38838310419 for these helper/test hashes.
+          assert hashes == {
+              'scripts/lib/package_commands.py': 'e04b86a71ca0fc9085ea2bb527afaba0fc3bf030e0934ef02f78c44d46b344cb',
+              'scripts/test-v3-installed-consumer.py': 'fa2b3e67ae3f0e5921c98e2dc10fa3653e401f5e0a31b9e4ff219b1d130f3492'}
           assert receipt['uname'][0] == 'Darwin'
           assert len(receipt['commands']) == 3
           assert all(item['result'] == 'success' for item in receipt['commands'])
@@ -253,7 +270,7 @@ jobs:
           assert control['zombie_group_signal_zero']['errno'] == 1
           assert control['zombie_helper_sigkill']['result'] == 'success'
           (output / 'semantic-proof.json').write_text(json.dumps(dict(status='passed', source_sha=source_sha,
-              hashes=hashes, command_exits=[row['exit'] for row in records], native_contracts='passed'), indent=2) + '\n')
+              hashes=hashes, command_exits=[row['exit'] for row in records], native_contracts='retained-run-38838310419', bounded_receipt_probe='passed'), indent=2) + '\n')
           PY
       - if: always()
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02

From ab3a7fa2bae4dc12a96cb108f5381fdd4787e9de Mon Sep 17 00:00:00 2001
From: Sebastiano Merlino <sebastiano@hey.com>
Date: Thu, 15 Oct 2026 19:20:15 -0700
Subject: [PATCH 5/5] FEATURE-ci-process-cleanup: Preserve unknown Darwin
 process states without claiming retirement

Actual native capture showed a complete numeric ps row with state ?< in an unrelated group. Recognize Darwin's unknown Mach state while retaining its membership and conservative non-Z semantics. Strict malformed input and owned leader checks remain unchanged.

Full captured stdout failed twice through the actual parser before the minimal whitelist repair and now preserves all 412 rows. Add captured-row, unknown owned-group denial, malformed input, and ownership regressions. Local 26 command/49 installed-consumer contracts and 120 affinity trials pass. Restore temporary full native verification for the new source; prior uncaptured parser errors remain only corroborated same-cause inference.
---
 .github/workflows/v3-cleanup-diagnostic.yml | 63 +++++++++------------
 scripts/lib/package_commands.py             |  3 +-
 scripts/test-v3-installed-consumer.py       | 55 ++++++++++++++++++
 3 files changed, 83 insertions(+), 38 deletions(-)

diff --git a/.github/workflows/v3-cleanup-diagnostic.yml b/.github/workflows/v3-cleanup-diagnostic.yml
index 8fbd9656b..71c8d8007 100644
--- a/.github/workflows/v3-cleanup-diagnostic.yml
+++ b/.github/workflows/v3-cleanup-diagnostic.yml
@@ -13,13 +13,13 @@ jobs:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
       - name: Match Python prerequisite
         run: brew install python
-      - name: Bound the unexplained native snapshot observation
+      - name: Run native command contracts
         timeout-minutes: 2
         shell: bash
         run: |
           mkdir -p "$RUNNER_TEMP/v3-cleanup-contracts"
           cat > "$RUNNER_TEMP/v3-cleanup-command-contracts.py" <<'PYTHON'
-          import hashlib, importlib.util, json, os, subprocess, sys, time, unittest
+          import hashlib, importlib.util, json, os, subprocess, sys, unittest
           from pathlib import Path
           source = Path.cwd()
           output = Path(os.environ['RUNNER_TEMP'], 'v3-cleanup-contracts')
@@ -60,30 +60,15 @@ jobs:
                   global current_test
                   current_test = test.id()
                   super().startTest(test)
-          started = time.monotonic()
-          trials = []
-          for iteration in range(1, 31):
-              if time.monotonic() - started >= 60:
-                  break
-              suite = unittest.TestSuite([contracts.CommandsTest(
-                  'test_owned_zombie_permission_denial_preserves_completed_receipts')])
-              result = unittest.TextTestRunner(verbosity=2, resultclass=Result, failfast=True).run(suite)
-              trial = dict(iteration=iteration, tests_run=result.testsRun, successful=result.wasSuccessful(),
-                           failures=[test.id() for test, _ in result.failures],
-                           errors=[test.id() for test, _ in result.errors],
-                           skipped=[dict(test=test.id(), reason=reason) for test, reason in result.skipped])
-              trials.append(trial)
-              report = dict(mode='bounded-owned-zombie-receipts', max_iterations=30, max_seconds=60,
-                            elapsed_seconds=time.monotonic() - started, trials=trials,
-                            successful=all(item['successful'] for item in trials),
-                            stop='first-failure' if not result.wasSuccessful() else 'running')
-              (output / 'result.json').write_text(json.dumps(report, indent=2) + '\n')
-              if not result.wasSuccessful():
-                  break
-          report['stop'] = ('first-failure' if not report['successful'] else
-                            'iteration-limit' if len(trials) == 30 else 'time-limit')
+          suite = unittest.defaultTestLoader.loadTestsFromTestCase(contracts.CommandsTest)
+          expected = [test.id() for test in suite]
+          result = unittest.TextTestRunner(verbosity=2, resultclass=Result).run(suite)
+          report = dict(expected=expected, tests_run=result.testsRun, successful=result.wasSuccessful(),
+                        failures=[test.id() for test, _ in result.failures],
+                        errors=[test.id() for test, _ in result.errors],
+                        skipped=[dict(test=test.id(), reason=reason) for test, reason in result.skipped])
           (output / 'result.json').write_text(json.dumps(report, indent=2) + '\n')
-          raise SystemExit(not report['successful'])
+          raise SystemExit(not result.wasSuccessful())
           PYTHON
           set +e
           python3 "$RUNNER_TEMP/v3-cleanup-command-contracts.py" 2>&1 | tee "$RUNNER_TEMP/v3-cleanup-contracts/commands-test.log"
@@ -225,17 +210,21 @@ jobs:
           assert receipt['python'] == binding['python']
           assert (contracts / 'exit-status').read_text().strip() == '0'
           result = json.loads((contracts / 'result.json').read_text())
-          assert result['mode'] == 'bounded-owned-zombie-receipts' and result['successful']
-          assert result['max_iterations'] == 30 and result['max_seconds'] == 60
-          assert 1 <= len(result['trials']) <= 30
-          assert result['stop'] in ('iteration-limit', 'time-limit')
-          for trial in result['trials']:
-              assert trial['tests_run'] == 1 and trial['successful']
-              assert not trial['failures'] and not trial['errors'] and not trial['skipped']
-          # Full 22-contract native GREEN is retained from run 38838310419 for these helper/test hashes.
-          assert hashes == {
-              'scripts/lib/package_commands.py': 'e04b86a71ca0fc9085ea2bb527afaba0fc3bf030e0934ef02f78c44d46b344cb',
-              'scripts/test-v3-installed-consumer.py': 'fa2b3e67ae3f0e5921c98e2dc10fa3653e401f5e0a31b9e4ff219b1d130f3492'}
+          assert result['successful'] and not result['failures'] and not result['errors']
+          assert result['tests_run'] == len(result['expected']) == 26
+          assert len(set(result['expected'])) == 26
+          assert {
+              'native_contracts.CommandsTest.test_ps_unknown_state_preserves_complete_numeric_membership',
+              'native_contracts.CommandsTest.test_ps_unknown_owned_member_cannot_justify_permission_denial',
+              'native_contracts.CommandsTest.test_ps_unknown_state_does_not_relax_snapshot_validation',
+              'native_contracts.CommandsTest.test_ps_unknown_state_cannot_hide_missing_or_changed_leader'
+          } <= set(result['expected'])
+          expected_skips = {
+              'native_contracts.CommandsTest.test_cleanup_discovery_failure_retains_original_error',
+              'native_contracts.CommandsTest.test_new_live_member_delays_retirement_and_zombies_do_not',
+              'native_contracts.CommandsTest.test_discovery_consumes_cleanup_deadline_without_restarting_wait'}
+          assert len(result['skipped']) == 3
+          assert {item['test'] for item in result['skipped']} == expected_skips
           assert receipt['uname'][0] == 'Darwin'
           assert len(receipt['commands']) == 3
           assert all(item['result'] == 'success' for item in receipt['commands'])
@@ -270,7 +259,7 @@ jobs:
           assert control['zombie_group_signal_zero']['errno'] == 1
           assert control['zombie_helper_sigkill']['result'] == 'success'
           (output / 'semantic-proof.json').write_text(json.dumps(dict(status='passed', source_sha=source_sha,
-              hashes=hashes, command_exits=[row['exit'] for row in records], native_contracts='retained-run-38838310419', bounded_receipt_probe='passed'), indent=2) + '\n')
+              hashes=hashes, command_exits=[row['exit'] for row in records], native_contracts='passed'), indent=2) + '\n')
           PY
       - if: always()
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
diff --git a/scripts/lib/package_commands.py b/scripts/lib/package_commands.py
index c8aed996d..e75901ded 100644
--- a/scripts/lib/package_commands.py
+++ b/scripts/lib/package_commands.py
@@ -261,8 +261,9 @@ def _process_snapshot(deadline):
     for line in output.getvalue().decode('ascii').splitlines():
         _remaining(deadline)
         fields = line.split()
+        # Darwin can report unknown Mach state '?'; retain it as non-retired.
         if (len(fields) != 4 or any(not item.isdigit() for item in fields[:3])
-                or not fields[3] or fields[3][0] not in 'RSDZTWtXxKIPLU'
+                or not fields[3] or fields[3][0] not in 'RSDZTWtXxKIPLU?'
                 or any(flag not in '+<>ACEJLNSsVWX' for flag in fields[3][1:])):
             raise RuntimeError('malformed command process snapshot')
         pid, parent, group = map(int, fields[:3])
diff --git a/scripts/test-v3-installed-consumer.py b/scripts/test-v3-installed-consumer.py
index 84c8e6ac4..21601def9 100644
--- a/scripts/test-v3-installed-consumer.py
+++ b/scripts/test-v3-installed-consumer.py
@@ -507,6 +507,61 @@ def launch(argv, *args, **kwargs):
             with self.assertRaisesRegex(RuntimeError, 'cleanup.*snapshot'):
                 self.runner(self.command('pass'), self.root, os.environ, timeout=2)
 
+    def ps_output(self, data):
+        popen = subprocess.Popen
+        def launch(argv, *args, **kwargs):
+            if argv[0] in ('/bin/ps', '/usr/bin/ps'):
+                argv = self.command(f'import os; os.write(1, {data.encode("ascii")!r})')
+            return popen(argv, *args, **kwargs)
+        return patch.object(subprocess, 'Popen', launch)
+
+    @unittest.skipUnless(os.name == 'posix', 'Darwin unknown scheduling state')
+    def test_ps_unknown_state_preserves_complete_numeric_membership(self):
+        # Captured native rows: the unknown process belongs to a different group.
+        data = ' 2255  2178  2255 Z<  \n 2258   910   892 ?<  \n'
+        with patch.object(sys, 'platform', 'darwin'), self.ps_output(data):
+            members = commands._process_snapshot(time.monotonic() + 2)
+        self.assertEqual({2255: (2178, 2255, 'Z'), 2258: (910, 892, '?')}, members)
+
+    @unittest.skipUnless(os.name == 'posix', 'unknown owned process cannot prove retirement')
+    def test_ps_unknown_owned_member_cannot_justify_permission_denial(self):
+        for unknown in ('leader', 'descendant'):
+            with self.subTest(unknown=unknown):
+                tree = self.zombie_tree()
+                pid = tree.child.pid
+                data = f'{pid} {os.getpid()} {pid} ' + ('?<' if unknown == 'leader' else 'Z') + '\n'
+                if unknown == 'descendant':
+                    data += f'{pid + 1} {pid} {pid} ?<\n'
+                with patch.object(sys, 'platform', 'darwin'), patch.object(os, 'waitid', None, create=True), self.ps_output(data):
+                    with patch.object(os, 'killpg', side_effect=PermissionError(1, 'unknown group denied')):
+                        with self.assertRaisesRegex(PermissionError, 'unknown group denied'):
+                            tree._kill_group(time.monotonic() + 2)
+                self.assertIsNone(tree.child.returncode, 'unknown state cannot reap the ownership anchor')
+
+    @unittest.skipUnless(os.name == 'posix', 'strict ps snapshot structure')
+    def test_ps_unknown_state_does_not_relax_snapshot_validation(self):
+        for data in ('pid 910 892 ?<', '2258 parent 892 ?<', '2258 910 group ?<',
+                     '-1 910 892 ?<', '0 910 892 ?<', '2258 -1 892 ?<', '2258 910 -1 ?<',
+                     '2258 910 ?<', '2258 910 892 ?< command', '2258 910 892 ?Q',
+                     '2258 910 892 !<', '2258 910 892 ?<\n2258 910 892 Z'):
+            with self.subTest(data=data):
+                with patch.object(sys, 'platform', 'darwin'), self.ps_output(data + '\n'):
+                    with self.assertRaisesRegex(RuntimeError, 'snapshot'):
+                        commands._process_snapshot(time.monotonic() + 2)
+
+    @unittest.skipUnless(os.name == 'posix', 'strict ps ownership anchor')
+    def test_ps_unknown_state_cannot_hide_missing_or_changed_leader(self):
+        tree = self.zombie_tree()
+        pid = tree.child.pid
+        for data in (f'{pid + 1} {pid} {pid} ?<',
+                     f'{pid} {os.getpid() + 1} {pid} ?<', f'{pid} {os.getpid()} 0 ?<'):
+            with self.subTest(data=data):
+                with patch.object(sys, 'platform', 'darwin'), patch.object(os, 'waitid', None, create=True), self.ps_output(data + '\n'):
+                    with patch.object(os, 'killpg') as kill:
+                        with self.assertRaisesRegex(RuntimeError, 'ownership'):
+                            tree._kill_group(time.monotonic() + 2)
+                    kill.assert_not_called()
+
     @unittest.skipUnless(os.name == 'posix', 'documented BSD/macOS state modifiers')
     def test_ps_fallback_accepts_documented_state_modifiers(self):
         popen = subprocess.Popen