fix(ci): bound diagnostic process inspection within phase deadlines

Reduce diagnostic polling pressure while retaining fatal inspection failures and exact process cleanup. Keep shared inspection defaults and original startup assertions unchanged.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

scripts/ci/pi-intel/diagnostic_lifecycle.py

@@ -2,8 +2,28 @@
 import time,subprocess
 from launch_gate import LaunchGate
 from source_guard import require
+from owned_processes import process_table,command_tokens,ProcessInspectionUnavailable
+
+class DiagnosticInspection:
+ """Same identity inspection, bounded by the current diagnostic lifecycle phase."""
+ def __init__(self,owner,child):self.owner=owner;self.child=child
+ def budget(self):
+  deadline=self.owner.stop_deadline if self.owner.stop_deadline is not None else self.child.deadline
+  require(deadline is not None,'Diagnostic inspection before deadline publication')
+  remaining=deadline-self.child.clock()
+  require(remaining>0,'Diagnostic inspection deadline')
+  return min(5,remaining)
+ def table(self):return process_table(timeout=self.budget())
+ def argv(self,pid):
+  try:return command_tokens(pid,timeout=self.budget())
+  except ProcessInspectionUnavailable as error:
+   # Active diagnosis must not continue with an unresolved identity. Cleanup
+   # retains its existing bounded fresh-inspection/revalidation policy.
+   if self.owner.stop_deadline is None:raise RuntimeError('Diagnostic active argv inspection failed') from error
+   raise
+
 class DiagnosticChild:
- def __init__(self,owner):self.owner=owner;self.child=None;self.attempted=False;self.uncertain=False;self.retired=False;self.cancelled=None;self.gate=LaunchGate(self.retire)
+ def __init__(self,owner):self.owner=owner;self.child=None;self.attempted=False;self.uncertain=False;self.retired=False;self.cancelled=None;self.gate=LaunchGate(self.retire);self.deadline=None;self.clock=time.monotonic
  def cancel(self,reason):self.cancelled=reason;self.gate.cancel(reason)
  def retire(self,child):
   if self.retired:return
@@ -14,7 +34,7 @@ class DiagnosticChild:
   except BaseException:self.uncertain=True;raise
  def execute(self,factory,timeout,clock=time.monotonic,sleep=time.sleep):
   require(not self.attempted and not self.uncertain,'Diagnostic may execute once only')
-  start=clock()
+  self.clock=clock;self.deadline=clock()+timeout
   def before():
    if self.cancelled:self.gate.cancel(self.cancelled)
   def launch():self.attempted=True;return factory()
@@ -22,7 +42,8 @@ class DiagnosticChild:
   try:
    self.gate.spawn(launch,publish,before)
    while self.child.poll() is None:
-    self.owner.observe(self.child.pid);require(clock()-start<timeout,'Diagnostic outer deadline');sleep(.1)
+    require(clock()<self.deadline,'Diagnostic outer deadline');self.owner.observe(self.child.pid)
+    require(clock()<self.deadline,'Diagnostic outer deadline');sleep(min(.5,self.deadline-clock()))
    self.owner.observe(self.child.pid)
   finally:
    if self.child is not None:self.retire(self.child)

scripts/ci/pi-intel/owned_processes.py

@@ -11,9 +11,9 @@ def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat()
 def atomic_json(path, value):
     path=pathlib.Path(path); tmp=path.with_suffix('.tmp')
     tmp.write_text(json.dumps(value,indent=2)+'\n'); tmp.chmod(0o600); tmp.replace(path)
-def process_table():
+def process_table(*, timeout=2):
     result={}
-    for line in subprocess.check_output(['/bin/ps','-axo','pid=,ppid=,lstart=,stat=,comm='],text=True,timeout=2).splitlines():
+    for line in subprocess.check_output(['/bin/ps','-axo','pid=,ppid=,lstart=,stat=,comm='],text=True,timeout=timeout).splitlines():
         row=line.split(maxsplit=8)
         if len(row)!=9: continue
         status={'R':'running','S':'sleeping','I':'idle','T':'stopped','Z':'zombie','U':'uninterruptible'}.get(row[7][:1],'unknown')
@@ -23,9 +23,9 @@ def process_table():
 class ProcessInspectionUnavailable(Exception):
     """No trustworthy argv snapshot; caller must revalidate identity."""
 
-def command_tokens(pid):
+def command_tokens(pid, *, timeout=2):
     # Inspect only an already owned PID; never persist or print this output.
-    try:value=subprocess.check_output(['/bin/ps','-p',str(pid),'-o','command='],text=True,stderr=subprocess.DEVNULL,timeout=2)
+    try:value=subprocess.check_output(['/bin/ps','-p',str(pid),'-o','command='],text=True,stderr=subprocess.DEVNULL,timeout=timeout)
     except (subprocess.SubprocessError,OSError) as error:raise ProcessInspectionUnavailable() from error
     if not value.strip() or len(value)>16384:raise ProcessInspectionUnavailable()
     try:return shlex.split(value)

scripts/ci/pi-intel/startup_diagnostic.py

@@ -8,7 +8,7 @@ from retain_pack import verify_archive
 from closed_inventory import closed_tree
 from startup_timing_patch import patch_sidecar
 from owned_processes import OwnedProcesses,atomic_json
-from diagnostic_lifecycle import DiagnosticChild
+from diagnostic_lifecycle import DiagnosticChild,DiagnosticInspection
 HERE=Path(__file__).resolve().parent
 PIN=json.loads((HERE/'startup-diagnostic-inputs.json').read_text())
 
@@ -122,6 +122,8 @@ def execute(args):
   retained=out/'retained-test-state';retained.mkdir(mode=0o700)
   tenv={**env,'PAPERCLIP_TEST_PI_STARTUP_PACKAGE_ROOT':str(pack),'PAPERCLIP_TEST_PI_STARTUP_RUNNER_BINARY':str(daemon),'PI_INTEL_OWNED_TMP':str(scratch),'PI_INTEL_RETAINED_STATE':str(retained)}
   owner=OwnedProcesses(out/'closed-startup-processes.json',scratch,pack,sidecar);active=DiagnosticChild(owner)
+  inspection=DiagnosticInspection(owner,active);owner.table=inspection.table;owner.argv=inspection.argv
+  proof['processInspection']={'activeCadenceSeconds':0.5,'maximumCallSeconds':5,'defaultSharedCallSeconds':2,'deadlineCap':'remaining active70s or existing cleanup100s','activeFailuresFatal':True,'activeInspectionRetries':0}
   row={'label':'original-closed-startup-test-with-diagnostic-sidecar','argv':[str(node),'--import',str(HERE/'retain-test-state.mjs'),'--test',str(test)],'outerDeadlineSeconds':70,'originalTestTimeoutMs':60000,'originalAdmissionLimitMs':30000,'startedAt':datetime.datetime.now(datetime.timezone.utc).isoformat(),'startedMonotonicNs':str(time.monotonic_ns())};proof['commands'].append(row);proof['status']='diagnostic_running';save()
   try:
    with (out/'closed-startup.log').open('xb') as log:

scripts/ci/pi-intel/test_startup_diagnostic.py

@@ -2,7 +2,10 @@ import copy,hashlib,json,os,subprocess,tempfile,unittest
 from pathlib import Path
 from startup_timing_patch import patch_sidecar,ORIGINAL_SHA
 from startup_diagnostic import validate_pack_delta,validate_timing_sink,reject_sink_failure
-from diagnostic_lifecycle import DiagnosticChild
+from diagnostic_lifecycle import DiagnosticChild,DiagnosticInspection
+from owned_processes import process_table,command_tokens,ProcessInspectionUnavailable
+from unittest.mock import patch
+from types import SimpleNamespace
 
 class FakeProcess:
  pid=111;returncode=0
@@ -124,4 +127,69 @@ class DiagnosticTests(unittest.TestCase):
   with self.assertRaises(OSError):c.execute(factory,1)
   self.assertTrue(c.uncertain)
 
+class DiagnosticInspectionTests(unittest.TestCase):
+ def fixture(self,now=10,active_end=80,cleanup_end=None):
+  self.now=now
+  owner=SimpleNamespace(stop_deadline=cleanup_end)
+  child=SimpleNamespace(deadline=active_end,clock=lambda:self.now)
+  return owner,child,DiagnosticInspection(owner,child)
+ def test_shared_inspection_defaults_and_exact_commands(self):
+  with patch('owned_processes.subprocess.check_output',return_value='') as run:
+   self.assertEqual(process_table(),{})
+   run.assert_called_once_with(['/bin/ps','-axo','pid=,ppid=,lstart=,stat=,comm='],text=True,timeout=2)
+  with patch('owned_processes.subprocess.check_output',return_value='node /owned/sidecar') as run:
+   self.assertEqual(command_tokens(123),['node','/owned/sidecar'])
+   run.assert_called_once_with(['/bin/ps','-p','123','-o','command='],text=True,stderr=subprocess.DEVNULL,timeout=2)
+ def test_diagnostic_maximum_and_remaining_active_budget(self):
+  _,_,inspection=self.fixture()
+  with patch('owned_processes.subprocess.check_output',return_value='node') as run:
+   inspection.table();self.assertEqual(run.call_args.kwargs['timeout'],5)
+   inspection.argv(123);self.assertEqual(run.call_args.kwargs['timeout'],5)
+   self.now=79.75;inspection.table();self.assertEqual(run.call_args.kwargs['timeout'],.25)
+   self.now=79.9;inspection.argv(123);self.assertAlmostEqual(run.call_args.kwargs['timeout'],.1)
+ def test_cleanup_budget_replaces_expired_active_deadline(self):
+  owner,_,inspection=self.fixture(now=90,active_end=80,cleanup_end=190)
+  with patch('owned_processes.subprocess.check_output',return_value='node') as run:
+   inspection.table();self.assertEqual(run.call_args.kwargs['timeout'],5)
+   self.now=189.75;inspection.argv(123);self.assertEqual(run.call_args.kwargs['timeout'],.25)
+ def test_expired_or_unpublished_deadlines_never_launch_inspection(self):
+  for active_end,cleanup_end in [(10,None),(None,None),(80,10)]:
+   _,_,inspection=self.fixture(active_end=active_end,cleanup_end=cleanup_end)
+   with patch('owned_processes.subprocess.check_output') as run:
+    for invoke in [inspection.table,lambda:inspection.argv(123)]:
+     with self.assertRaises(RuntimeError):invoke()
+    run.assert_not_called()
+ def test_active_argv_failure_fatal_cleanup_policy_preserved(self):
+  owner,_,inspection=self.fixture()
+  with patch('owned_processes.subprocess.check_output',side_effect=subprocess.TimeoutExpired('ps',5)) as run:
+   with self.assertRaisesRegex(RuntimeError,'active argv inspection'):inspection.argv(123)
+   self.assertEqual(run.call_count,1)
+   owner.stop_deadline=110
+   with self.assertRaises(ProcessInspectionUnavailable):inspection.argv(123)
+   self.assertEqual(run.call_count,2)
+ def test_active_cadence_half_second_and_deadline_capped(self):
+  owner=FakeOwner();child=DiagnosticChild(owner);process=FakeProcess();process.poll=lambda:None
+  self.now=0;sleeps=[]
+  def sleep(duration):sleeps.append(duration);self.now+=duration
+  with self.assertRaisesRegex(RuntimeError,'outer deadline'):
+   child.execute(lambda:process,1.2,clock=lambda:self.now,sleep=sleep)
+  self.assertEqual(len(sleeps),3);self.assertEqual(sleeps[:2],[.5,.5]);self.assertAlmostEqual(sleeps[2],.2)
+  self.assertTrue(child.retired);self.assertEqual(owner.stops,1)
+ def test_active_table_timeout_fatal_retired_once_without_retry(self):
+  owner=FakeOwner();child=DiagnosticChild(owner);process=FakeProcess();process.poll=lambda:None
+  calls=[]
+  def observe(pid):
+   calls.append(pid)
+   if len(calls)==2:raise subprocess.TimeoutExpired('ps',5)
+  owner.observe=observe
+  with self.assertRaises(subprocess.TimeoutExpired):child.execute(lambda:process,70)
+  self.assertEqual(calls,[111,111]);self.assertEqual(owner.stops,1);self.assertTrue(child.retired)
+  with self.assertRaises(RuntimeError):child.execute(lambda:self.fail('second launch'),70)
+ def test_inspection_failure_with_cleanup_error_stays_uncertain(self):
+  owner=FakeOwner(fail_stop=True);child=DiagnosticChild(owner)
+  owner.observe=lambda pid:(_ for _ in ()).throw(subprocess.TimeoutExpired('ps',5))
+  with self.assertRaisesRegex(RuntimeError,'stop failed'):child.execute(lambda:FakeProcess(),70)
+  self.assertTrue(child.uncertain);self.assertEqual(owner.stops,1)
+  with self.assertRaises(RuntimeError):child.execute(lambda:self.fail('second launch'),70)
+
 if __name__=='__main__':unittest.main()