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.

Author: etr

.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

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])

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