Contain ordinary POSIX descendants of shared Git blob readers

docs/research/git-blob-descendant-ownership.md

@@ -0,0 +1,104 @@
+# POSIX ownership of shared Git blob readers
+
+A response deadline that kills only `git cat-file` does not bound a pipe read
+when an ordinary descendant retains stdout after Git exits. The pre-change
+reader reproduced this with actual processes: a controlled Git launcher exited,
+its ready child retained the response pipe, and the read remained blocked after
+a 0.7-second deadline until the outer 2-second test guard intervened. A launcher
+without that child rejected promptly. This is a process-boundary reproduction,
+not evidence that an installed Git version spontaneously exhibits this behavior.
+
+The shared `git_blobs` reader now starts each POSIX batch process in its own
+session. Its response watchdog and generator finalizer share an idempotent stop
+operation that sends SIGKILL to that process group, including after leader exit.
+An exited leader is reaped before signaling; this avoids macOS EPERM for a group
+containing only its unreaped zombie. The watchdog is cancelled and joined before
+control returns to the caller, so the finalizer and watchdog do not race their
+stop operations. Finalization stops the group before closing buffered streams.
+Normal completion, malformed responses and early generator close use the same
+ownership path. Non-POSIX retains its existing direct-child behavior.
+
+The stop action does not alter authenticated bytes, source roles or Git
+repository selection. Streaming still retains one blob, and the deadline ends
+before yielding it: a slow or suspended consumer is not charged as a slow Git
+response. The unsupported-NUL fallback still requires the first request, no
+response and exit 129; it retains the original request and exact legacy framing.
+Live-input drift continues to reject, including old research packages whose
+historical helper pins differ from today's shared code. This repair neither
+repins those artifacts nor makes their historical replay green.
+
+## Actual consumers
+
+All production calls share this one reader:
+
+| Consumer | Blob use | Compatibility checks |
+| --- | --- | --- |
+| `research_package.validate` | Source-pin objects | Git compatibility, replacements, root-binding suites |
+| `research_package.planning_manifests` | Sparse HEAD manifest objects | Sparse-planning suite |
+| `new_research_package.create` | New package source pins | Scaffold-batch suite |
+| `check_research_commit.check` | Committed package/input bytes | Commit-bytes suite |
+| `export_research_package.inventory/export` | Committed manifest and archive bodies | Export suite |
+
+The older `git_source_rows` helper, individual `git()` calls, HEAD inventory
+commands and history-helper subprocesses are separate boundaries. This change
+does not silently extend their deadlines. No frozen historical runner is edited.
+
+## Validation
+
+The dedicated test module launches the actual shared reader in an independently
+bounded worker. A real child writes its PID/readiness file before the simulated
+Git leader exits or responds. Tests inspect that readiness and stopped process
+state before fallback fixture cleanup. An adopted zombie counts as stopped,
+not as a running descendant. The direct-child-only semantic mutant still blocks
+past the external guard with the descendant alive, so teardown cannot manufacture
+a passing cleanup result.
+
+Six test methods cover header, partial-payload, terminator and blocked-write
+stalls; successful and malformed responses; early generator close with a live
+leader; consumer suspension longer than the response deadline; exact binary
+legacy fallback; and the direct-child-only mutant. The existing closed-stdin
+fixture models the new POSIX operation while retaining native-Windows-compatible
+simulation; it does not pretend to test Windows process groups.
+
+Local validation uses CPython 3.12.7 on macOS arm64, with warnings as errors:
+
+```sh
+python3.12 -W error -B -m unittest discover -s tests -p 'test_research_git_*.py' -v
+python3.12 -W error -B -O -m unittest discover -s tests -p 'test_research_git_*.py' -v
+```
+
+Also run each of `test_research_scaffold_batch.py`,
+`test_research_commit_bytes.py`, `test_research_sparse_planning.py` and
+`test_research_export.py` with the same normal/optimized discovery commands.
+All 57 Git-reader tests passed normally (19.963s) and optimized (21.260s),
+and all 35 caller tests passed in each mode. A fresh committed three-file source
+snapshot, without `.git` and with only the `ps` observer on PATH, passed all six
+descendant tests normally (9.589s) and optimized (10.767s); every source hash
+remained unchanged. This is a focused tooling snapshot, not a registered research
+package export. An earlier sandboxed optimized attempt denied the `ps` observer;
+the final optimized run used process-observation permission. That denied attempt
+is not counted as passing evidence.
+
+These real-Git checks cover exact binary transport, drift rejection, linked
+worktrees, retained source authority, sparse planning, immutable publication
+bytes and Git-free export integrity. They are tooling checks, not mathematical
+replays. No full selected, historical all-suite or heavy scientific run is used.
+
+## Boundaries and open next questions
+
+The contract covers ordinary POSIX descendants that remain in the owned process
+group and retain the caller's signal permissions. It does not contain escaped
+sessions/groups, privileged children, uninterruptible kernel operations, or an
+abruptly killed outer Python process. Windows descendant ownership and native
+Linux execution are unverified here. Individual Git commands and the legacy
+reader still need separate bounded ownership work if their concrete callers
+require it. A platform-specific ownership implementation must preserve the
+streamed protocol and consumer-suspension boundary before extending this claim.
+
+This takes the explicitly deferred descendant boundary in
+[the shared blob-response deadline handoff](https://github.com/CrocSwap/integer-mult-bounds/discussions/4681#discussioncomment-19130292).
+The source reader, immutable-object policy, fallback and deadline work retain
+their original attribution, including Douglas Colkitt, Nick Petalas, Arham Tahir,
+Woklerx and Mihir Srivastava. The process ownership repair and regression evidence
+were prepared by JoyboyBrian with substantial OpenAI Codex assistance;
+Apache-2.0. No independent human or formal proof-kernel review is claimed.

scripts/research_package.py

@@ -168,21 +168,34 @@ def _closed_stdin_error(error):
             or (os.name == "nt" and error.errno == errno.EINVAL))
 
 
+def _stop_git_blob_process(process):
+    """Stop this reader's owned POSIX group even after its leader exits."""
+    try:
+        if os.name == "posix":
+            # Reap an exited leader first: signaling a group containing only
+            # its unreaped zombie can report EPERM on macOS. A surviving
+            # descendant keeps the group alive after the leader is reaped.
+            process.poll()
+            os.killpg(process.pid, signal.SIGKILL)
+        else:
+            process.kill()
+    except ProcessLookupError:
+        pass
+
+
 @contextmanager
-def _git_blob_response_deadline(process):
-    """Bound one direct-child response, excluding time spent by its consumer.
+def _git_blob_response_deadline(stop):
+    """Bound one response, excluding time spent by its consumer.
 
-    Descendants retaining a pipe and separate git() calls need their own
-    containment/deadlines; killing the direct child cannot close their handles.
+    On POSIX the reader owns a session: ordinary descendants retaining a pipe
+    are stopped with the leader. Escaped sessions and separate git() calls are
+    outside this contract; non-POSIX retains the direct-child deadline.
     """
     expired = threading.Event()
 
     def terminate():
         expired.set()
-        try:
-            process.kill()
-        except ProcessLookupError:
-            pass
+        stop()
 
     timer = threading.Timer(GIT_BLOB_RESPONSE_TIMEOUT, terminate)
     timer.start()
@@ -200,7 +213,16 @@ def _git_blob_batch(root, specs, nul):
     # Usage diagnostics from the unsupported-option probe are expected. Avoid
     # a second pipe that could block while we consume the framed stdout reply.
     with subprocess.Popen(argv, cwd=root, env=git_source_env(), stdin=subprocess.PIPE,
-                          stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) as process:
+                          stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+                          start_new_session=(os.name == "posix")) as process:
+        stopped = False
+
+        def stop():
+            nonlocal stopped
+            if not stopped:
+                _stop_git_blob_process(process)
+                stopped = True
+
         try:
             for index, spec in enumerate(specs):
                 require("\0" not in spec, "invalid Git object request")
@@ -209,7 +231,7 @@ def _git_blob_batch(root, specs, nul):
                     # Resolve one literal argument, never split a path into requests.
                     request = git(root, "rev-parse", "--verify", "--end-of-options", spec).decode("ascii").strip()
                     require(re.fullmatch(r"[0-9a-f]{40,64}", request), "invalid Git object ID")
-                with _git_blob_response_deadline(process):
+                with _git_blob_response_deadline(stop):
                     try:
                         process.stdin.write(request.encode("utf-8") + (b"\0" if nul else b"\n"))
                         process.stdin.flush()
@@ -233,6 +255,10 @@ def _git_blob_batch(root, specs, nul):
             process.stdin.close()
             require(process.wait(timeout=90) == 0, "Git source reader failed")
         finally:
+            # Terminate ordinary descendants before closing buffered pipes. This
+            # also covers successful leader exit and an early generator close.
+            if os.name == "posix":
+                stop()
             # Option rejection can close stdin before our first write/flush.
             try:
                 process.stdin.close()

tests/test_research_git_descendants.py

@@ -0,0 +1,167 @@
+"""Real POSIX descendants must not outlive a shared Git blob read.
+
+Builds on the retained source-reader deadline and compatibility contracts.
+JoyboyBrian, with substantial OpenAI Codex assistance. Apache-2.0.
+"""
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+# Run the reader in a separate process so the unfixed regression is bounded by
+# the test, without releasing its blocked read before we record the failure.
+WORKER = r'''
+import importlib.util,json,os,subprocess,sys,time
+from pathlib import Path
+from unittest.mock import patch
+root, directory, mode, mutant = sys.argv[1:]
+directory=Path(directory)
+spec=importlib.util.spec_from_file_location('blob_reader',Path(root)/'scripts/research_package.py')
+p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p)
+p.GIT_BLOB_RESPONSE_TIMEOUT=.8
+real_popen=subprocess.Popen
+calls=[]
+leader = r"""
+import os,subprocess,sys,time
+from pathlib import Path
+directory=Path(sys.argv[1]);mode=sys.argv[2]
+child=subprocess.Popen([sys.executable,'-c',
+ 'import os,time;from pathlib import Path;Path('+repr(str(directory/'child-ready'))+').write_text(str(os.getpid()));time.sleep(60)'])
+while not (directory/'child-ready').exists(): time.sleep(.005)
+if mode=='write': sys.exit(0)
+while sys.stdin.buffer.read(1) not in (b'\0',b'\n',b''): pass
+reply=b'a'*40+b' blob 4\n'+b'\0\xff\r\n'+b'\n'
+if mode=='header': sys.exit(0)
+if mode=='payload': sys.stdout.buffer.write(reply[:50]);sys.stdout.buffer.flush();sys.exit(0)
+if mode=='terminator': sys.stdout.buffer.write(reply[:-1]);sys.stdout.buffer.flush();sys.exit(0)
+if mode=='malformed': sys.stdout.buffer.write(b'bad\n');sys.stdout.buffer.flush();sys.exit(0)
+sys.stdout.buffer.write(reply);sys.stdout.buffer.flush()
+if mode=='suspended':
+ while sys.stdin.buffer.read(1) not in (b'\0',b'\n',b''): pass
+ sys.stdout.buffer.write(reply);sys.stdout.buffer.flush()
+if mode=='early': time.sleep(60)
+"""
+def start(argv,**kwargs):
+ calls.append(argv)
+ if mode=='fallback' and len(calls)==1:
+  child=real_popen([sys.executable,'-I','-B','-c','import sys;sys.exit(129)'],**kwargs)
+ else:
+  child=real_popen([sys.executable,'-I','-B','-c',leader,str(directory),mode],**kwargs)
+ (directory/('leader-'+str(len(calls)))).write_text(str(child.pid))
+ return child
+if mutant=='direct':
+ def direct(process):
+  try: process.kill()
+  except ProcessLookupError: pass
+ p._stop_git_blob_process=direct
+result={}
+try:
+ with patch.object(p.subprocess,'Popen',side_effect=start):
+  rows=p.git_blobs(directory,['x'*2_000_000] if mode=='write' else ['HEAD:first','HEAD:second'] if mode=='suspended' else ['HEAD:first'])
+  if mode in ('early','suspended'):
+   result['first']=next(rows)==('a'*40,b'\0\xff\r\n')
+   if mode=='suspended':
+    time.sleep(1.1)
+    os.kill(int((directory/'child-ready').read_text()),0)
+    result['second']=next(rows)==('a'*40,b'\0\xff\r\n')
+    result['exhausted']=list(rows)==[]
+   rows.close()
+  else: result['values']=list(rows)==[('a'*40,b'\0\xff\r\n')]
+except Exception as error:
+ import traceback;traceback.print_exc()
+ result['error']=type(error).__name__
+result['calls']=len(calls)
+print(json.dumps(result),flush=True)
+'''
+
+
+def running(pid):
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    # An adopted zombie is stopped, even before the system reaps its PID.
+    state = subprocess.run(['ps', '-o', 'stat=', '-p', str(pid)],
+                           capture_output=True, text=True, timeout=2).stdout.strip()
+    return bool(state) and not state.startswith('Z')
+
+
+@unittest.skipUnless(os.name == 'posix', 'POSIX owned process-group contract')
+class GitBlobDescendantTests(unittest.TestCase):
+    def exercise(self, mode, mutant=False):
+        with tempfile.TemporaryDirectory() as temporary:
+            directory = Path(temporary)
+            argv = [sys.executable, '-I', '-B']
+            if sys.flags.optimize:
+                argv.append('-O')
+            argv += ['-c', WORKER, str(ROOT), temporary, mode, 'direct' if mutant else 'none']
+            process = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                                       text=True, start_new_session=True)
+            try:
+                try:
+                    stdout, stderr = process.communicate(timeout=4)
+                    expired = False
+                except subprocess.TimeoutExpired:
+                    expired = True
+                    process.kill()
+                    stdout, stderr = process.communicate(timeout=2)
+                # Readiness and process state are inspected BEFORE fallback cleanup.
+                self.assertTrue((directory / 'child-ready').exists(), stderr)
+                child = int((directory / 'child-ready').read_text())
+                if mutant:
+                    self.assertTrue(expired, stdout + stderr)
+                    self.assertTrue(running(child))
+                    return
+                self.assertFalse(expired, 'reader exceeded its deadline: ' + stderr)
+                self.assertEqual(process.returncode, 0, stderr)
+                if 'PermissionError' in stderr: self.fail(stderr)
+                end = time.monotonic() + 1
+                while running(child) and time.monotonic() < end:
+                    time.sleep(.01)
+                self.assertFalse(running(child), 'descendant survived reader completion')
+                return json.loads(stdout)
+            finally:
+                # This is failure containment, never evidence of successful cleanup.
+                for path in directory.glob('leader-*'):
+                    try:
+                        os.killpg(int(path.read_text()), signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+                if process.poll() is None:
+                    process.kill()
+                process.wait(timeout=2)
+                for stream in (process.stdout, process.stderr):
+                    stream.close()
+
+    def test_exited_leader_header_payload_terminator_and_write_stalls(self):
+        for mode in ('header', 'payload', 'terminator', 'write'):
+            with self.subTest(mode=mode):
+                result = self.exercise(mode)
+                self.assertEqual(result, {'error': 'TimeoutExpired', 'calls': 1})
+
+    def test_success_and_malformed_reply_stop_ready_descendant(self):
+        self.assertEqual(self.exercise('success'), {'values': True, 'calls': 1})
+        self.assertEqual(self.exercise('malformed'), {'error': 'ValueError', 'calls': 1})
+
+    def test_early_close_stops_live_leader_and_descendant(self):
+        self.assertEqual(self.exercise('early'), {'first': True, 'calls': 1})
+
+    def test_consumer_suspension_does_not_expire_the_next_response(self):
+        self.assertEqual(self.exercise('suspended'),
+                         {'first': True, 'second': True, 'exhausted': True, 'calls': 1})
+
+    def test_legacy_fallback_retains_exact_reply_and_owns_descendant(self):
+        self.assertEqual(self.exercise('fallback'), {'values': True, 'calls': 2})
+
+    def test_direct_child_only_mutant_still_hangs_with_ready_descendant(self):
+        self.exercise('header', mutant=True)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_research_git_pipe_errors.py

@@ -90,8 +90,19 @@ def start(argv, **kwargs):
             children.append(child)
             return child
 
+        def kill_group(pid, signum):
+            # The simulated POSIX branch still runs on native Windows. Group
+            # ownership itself is tested separately with actual POSIX processes.
+            if hasattr(os, "killpg"):
+                os.killpg(pid, signum)
+            else:
+                next(child for child in children if child.pid == pid).kill()
+
         try:
-            with (patch.object(PACKAGE, "os", SimpleNamespace(name=platform, environ=os.environ)),
+            with (patch.object(PACKAGE, "os", SimpleNamespace(
+                    name=platform, environ=os.environ, killpg=kill_group)),
+                  patch.object(PACKAGE.signal, "SIGKILL",
+                               getattr(PACKAGE.signal, "SIGKILL", 9), create=True),
                   patch.object(PACKAGE.subprocess, "Popen", side_effect=start)):
                 yield calls, pipes
         finally: