Fix v3 CI process cleanup on macOS

.github/workflows/v3-cleanup-diagnostic.yml

@@ -0,0 +1,271 @@
+name: v3 cleanup diagnostic
+on:
+  pull_request:
+    branches: [v3]
+    paths: ['.github/workflows/v3-cleanup-diagnostic.yml', 'scripts/lib/package_commands.py', 'scripts/test-v3-installed-consumer.py']
+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: 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, 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})
+          (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 "$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"
+      - name: Observe actual helper and native zombie signal semantics
+        if: always()
+        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 hashlib
+          import json
+          import os
+          from pathlib import Path
+          import platform
+          import select
+          import signal
+          import subprocess
+          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(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=[])
+
+          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"
+      - 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'
+          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']) == 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'])
+          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
+            ${{ runner.temp }}/v3-cleanup-contracts
+          if-no-files-found: error

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:
@@ -251,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])