test: harden processing failure diagnostics and cleanup

agents/rocko/work/vikunja-502/run-mutants.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Run only with Sage's explicit slot grant. Mutations are exact-once and restored."""
+from pathlib import Path
+import argparse, hashlib, json, os, re, signal, subprocess, sys, time
+packet = Path(__file__).resolve().parent
+root = Path('/mnt/storage/scratch/rocko-v502')
+all_sites = json.loads((packet/'all-mutants.json').read_text())
+parser = argparse.ArgumentParser()
+parser.add_argument('ids', nargs='*')
+parser.add_argument('--resume-from')
+parser.add_argument('--control-file')
+parser.add_argument('--adjudicate-timeout')
+args = parser.parse_args()
+sites = all_sites
+if args.resume_from:
+    ids = [s['id'] for s in all_sites]
+    if args.ids or args.resume_from not in ids: raise SystemExit('invalid resume selection')
+    sites = all_sites[ids.index(args.resume_from):]
+elif args.ids:
+    wanted = set(args.ids); sites = [s for s in all_sites if s['id'] in wanted]
+    if {s['id'] for s in sites} != wanted: raise SystemExit('unknown mutant id')
+if args.adjudicate_timeout and not args.control_file:
+    raise SystemExit('timeout adjudication requires an unmutated whole-file control')
+originals = {e['file']: (root/e['file']).read_bytes() for s in sites for e in [s, *s.get('extra_edits', [])]}
+for s in sites:
+    for e in [s, *s.get('extra_edits', [])]:
+        if originals[e['file']].decode().count(e['old']) != 1: raise SystemExit('preflight '+s['id'])
+logs = packet/('resume-mutant-logs' if args.resume_from else 'mutant-logs'); logs.mkdir(exist_ok=True)
+result_file = packet/('resume-mutant-results.json' if args.resume_from else 'mutant-results.json')
+results = []
+controls = {}
+run_id = str(time.time_ns())
+
+def verify_candidate():
+    expected = (packet/'commit.txt').read_text().strip()
+    if subprocess.check_output(['git','rev-parse','HEAD'],cwd=root,text=True).strip()!=expected:
+        raise RuntimeError('candidate HEAD changed')
+    manifest = packet/'candidate-manifest.sha256'
+    digest = hashlib.sha256(manifest.read_bytes()).hexdigest()
+    if digest!=(packet/'candidate.txt').read_text().strip(): raise RuntimeError('manifest digest changed')
+    for line in manifest.read_text().splitlines():
+        wanted,name = line.split('  ',1)
+        if hashlib.sha256((root/name).read_bytes()).hexdigest()!=wanted: raise RuntimeError('candidate drift '+name)
+    return {'head':expected,'manifest':digest}
+
+def stat(pid):
+    try:
+        raw = Path('/proc',str(pid),'stat').read_text(); fields = raw[raw.rfind(')')+2:].split()
+        return dict(ppid=int(fields[1]), group=int(fields[2]), start=fields[19], state=fields[0])
+    except (FileNotFoundError, PermissionError, ProcessLookupError): return None
+def capture(child, known):
+    current = {}
+    for p in Path('/proc').iterdir():
+        if p.name.isdecimal():
+            value=stat(int(p.name))
+            if value: current[int(p.name)]=value
+    changed=True
+    while changed:
+        changed=False
+        for pid,value in current.items():
+            if pid not in known and (pid==child or (value['ppid'] in known and current.get(value['ppid'],{}).get('start')==known[value['ppid']]['start'])): known[pid]=value;changed=True
+
+def cleanup(known):
+    groups={}
+    for pid,old in known.items():
+        cur=stat(pid)
+        if cur and cur['start']==old['start'] and cur['state']!='Z': groups.setdefault(cur['group'],[]).append(pid)
+    for group,anchors in groups.items():
+        # Every member here descends from this test runner, identified by start ticks.
+        if not any((cur:=stat(pid)) and cur['start']==known[pid]['start'] and cur['group']==group for pid in anchors):continue
+        try: os.killpg(group, signal.SIGKILL)
+        except ProcessLookupError: pass
+def execute(test, pattern, log):
+    child = None; known = {}; timedout = False; started = time.monotonic()
+    command = ['node','--test','--test-reporter=tap']
+    if pattern: command += ['--test-name-pattern='+re.escape(pattern)]
+    command += [test]
+    try:
+        with log.open('w') as out:
+            child = subprocess.Popen(command,cwd=root,stdout=out,stderr=subprocess.STDOUT,start_new_session=True,
+                env={**os.environ,'TMPDIR':os.environ['TMPDIR'],'DOCKER_HOST':'unix:///nonexistent.sock',
+                     'MOSAIC_CONFIG':str(Path(os.environ['TMPDIR'])/'no-live-config.json')})
+            first = stat(child.pid)
+            if first: known[child.pid] = first
+            deadline = started+90
+            while child.poll() is None:
+                capture(child.pid,known)
+                if time.monotonic()>=deadline:
+                    timedout = True; cleanup(known)
+                    try: os.killpg(child.pid,signal.SIGKILL)
+                    except ProcessLookupError: pass
+                    break
+                time.sleep(.05)
+            rc = child.wait()
+    finally:
+        if child and child.poll() is None:
+            cleanup(known)
+            try: os.killpg(child.pid,signal.SIGKILL)
+            except ProcessLookupError: pass
+            child.wait()
+        cleanup(known)
+    text = log.read_text()
+    return {'exit':rc,'timeout':timedout,'seconds':time.monotonic()-started,'log':str(log),
+            'failing':re.findall(r'^\s*not ok \d+ - (.*)$',text,re.M),
+            'counts':{k:int(v) for k,v in re.findall(r'^# (tests|pass|fail|skipped|cancelled) (\d+)\s*$',text,re.M)}}
+
+def timeout_cause(site, control=None):
+    if site['id']=='SR08':
+        return ('runtimes.mjs:333 loops on now() <= deadline; stop-run.test.mjs:106-108 '
+                'gives the second host a fixed fake clock with a real sleep. Removing the join '
+                'branch enters the delivery loop without advancing that clock. Original log: TAP header only.')
+    if not control or control.get('passed') is not True:
+        return ('Mutated '+site['file']+' timed out after 90s; no passing same-file control '
+                'exists in this invocation. Detection is not established. '
+                'Detailed cause must be recorded before review.')
+    return ('Mutated '+site['file']+' timed out after 90s while the same whole test file '
+            'passed unmutated in this invocation. Detailed cause must be recorded before review.')
+
+try:
+    binding = verify_candidate()
+    if args.control_file:
+        if args.control_file not in {s['test'] for s in all_sites}: raise SystemExit('unknown control file')
+        before = hashlib.sha256((root/args.control_file).read_bytes()).hexdigest()
+        control = execute(args.control_file,'',logs/'unmutated-control.tap')
+        after = hashlib.sha256((root/args.control_file).read_bytes()).hexdigest()
+        counts = control['counts']
+        passed = (control['exit']==0 and not control['timeout'] and not control['failing'] and before==after
+                  and counts.get('tests',0)>0 and counts.get('pass')==counts.get('tests')
+                  and counts.get('fail')==0 and counts.get('cancelled')==0 and counts.get('skipped')==0)
+        control.update(run_id=run_id,test=args.control_file,test_sha256=before,passed=passed,**binding)
+        (packet/'unmutated-control.json').write_text(json.dumps(control,indent=2)+'\n')
+        print('CONTROL',json.dumps(control),flush=True)
+        if not passed: raise SystemExit('control failed; stop for Mos, no mutation applied')
+        verify_candidate()
+        controls[args.control_file] = control
+    if args.adjudicate_timeout:
+        # This is the already-observed SR08 timeout, not a rerun or an assertion kill.
+        if args.adjudicate_timeout!='SR08': raise SystemExit('only SR08 historical adjudication approved')
+        prior = json.loads((packet/'mutant-results.json').read_text())
+        old = next((s for s in prior if s['id']=='SR08'),None)
+        site = next(s for s in all_sites if s['id']=='SR08')
+        if not old or old['status']!='timeout' or old['test']!=args.control_file or old['failing']:
+            raise SystemExit('historical timeout does not match approved control')
+        if Path(old['log']).read_text().strip()!='TAP version 13': raise SystemExit('historical timeout log changed')
+        adjudication = {**old,'status':'killed-by-timeout','cause':timeout_cause(site, controls[args.control_file]),'control':controls[args.control_file],
+                        'original_log_sha256':hashlib.sha256(Path(old['log']).read_bytes()).hexdigest()}
+        (packet/'SR08-timeout-adjudication.json').write_text(json.dumps(adjudication,indent=2)+'\n')
+        print('TIMEOUT_KILL SR08',adjudication['cause'],flush=True)
+    for i,s in enumerate(sites,1):
+        verify_candidate()
+        edits = [s, *s.get('extra_edits', [])]
+        for e in edits:
+            if (root/e['file']).read_bytes()!=originals[e['file']]: raise RuntimeError('source drift '+s['id'])
+        print(f'{i}/{len(sites)} {s["id"]}: {s["check"]}',flush=True)
+        try:
+            for e in edits:
+                (root/e['file']).write_bytes(originals[e['file']].decode().replace(e['old'],e['new'],1).encode())
+            observed = execute(s['test'],s['pattern'],logs/(s['id']+'.tap'))
+        finally:
+            for e in edits: (root/e['file']).write_bytes(originals[e['file']])
+        expected = not s['pattern'] or any(s['pattern'] in name for name in observed['failing'])
+        control = controls.get(s['test'])
+        if observed['timeout']:
+            status = 'killed-by-timeout' if control and control['passed'] and control['run_id']==run_id else 'timeout-without-control'
+        elif observed['exit']==0: status = 'survived'
+        elif observed['failing'] and expected: status = 'assertion-killed'
+        else: status = 'invalid-failure'
+        row = {**{k:s[k] for k in ['id','file','test','pattern','check']},**observed,'status':status}
+        if observed['timeout']: row.update(cause=timeout_cause(s, control),control=control)
+        results.append(row)
+        result_file.write_text(json.dumps(results,indent=2)+'\n')
+        print(status,flush=True)
+        if status not in ('assertion-killed','killed-by-timeout'): raise SystemExit(1)
+finally:
+    for name,data in originals.items(): (root/name).write_bytes(data)
+    if any((root/name).read_bytes()!=data for name,data in originals.items()): raise RuntimeError('restore failed')
+summary = {'assertion_kills':sum(s['status']=='assertion-killed' for s in results),
+           'timeout_kills':sum(s['status']=='killed-by-timeout' for s in results),
+           'historical_timeout_kills':int(bool(args.adjudicate_timeout)),'selected':len(sites),
+           'run_id':run_id,'candidate':binding}
+(packet/'mutant-resume-summary.json').write_text(json.dumps(summary,indent=2)+'\n')
+print('MUTANT_SUMMARY',json.dumps(summary),flush=True)

agents/rocko/work/vikunja-502/tests/test_timeout_cause.py

@@ -0,0 +1,44 @@
+"""Exercise the actual pure cause renderer without executing the mutant driver."""
+import ast
+from pathlib import Path
+import unittest
+
+RUNNER = Path(__file__).resolve().parents[1] / 'run-mutants.py'
+
+
+def cause_renderer():
+    tree = ast.parse(RUNNER.read_text())
+    function = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'timeout_cause')
+    namespace = {}
+    exec(compile(ast.Module(body=[function], type_ignores=[]), str(RUNNER), 'exec'), namespace)
+    return namespace['timeout_cause']
+
+
+class TimeoutCause(unittest.TestCase):
+    def setUp(self):
+        self.cause = cause_renderer()
+        self.site = {'id': 'KS54', 'file': 'packages/bus/src/runtime.mjs'}
+
+    def test_missing_control_never_claims_an_unmutated_pass(self):
+        text = self.cause(self.site)
+        self.assertIn('no passing same-file control', text)
+        self.assertNotIn('passed unmutated', text)
+
+    def test_failed_control_never_claims_an_unmutated_pass(self):
+        text = self.cause(self.site, {'passed': False})
+        self.assertIn('no passing same-file control', text)
+        self.assertNotIn('passed unmutated', text)
+
+    def test_passing_control_can_state_its_observed_pass(self):
+        text = self.cause(self.site, {'passed': True})
+        self.assertIn('passed unmutated', text)
+        self.assertNotIn('no passing same-file control', text)
+
+    def test_known_sr08_cause_remains_specific_without_claiming_a_control(self):
+        text = self.cause({'id': 'SR08', 'file': 'packages/processing-control/src/runtimes.mjs'})
+        self.assertIn('fixed fake clock', text)
+        self.assertNotIn('passed unmutated', text)
+
+
+if __name__ == '__main__':
+    unittest.main()

packages/bus/tests/tasks.test.mjs

@@ -259,10 +259,13 @@ test('the runtime refuses an invalid adapter and closes a valid one', async (t)
       return { handle: async () => ({ ok: 1 }), timeout: 1000, close: async () => closed++ };
     },
   });
-  // Row 191: runtime injects cancellation alongside the existing factory dependencies.
-  assert.deepEqual(given, ['broker', 'businesses', 'credentials', 'processing']);
-  assert.equal(typeof processing.work, 'function');
-  await runtime.close();
+  try {
+    // Row 191: runtime injects cancellation alongside the existing factory dependencies.
+    assert.deepEqual(given, ['broker', 'businesses', 'credentials', 'processing']);
+    assert.equal(typeof processing.work, 'function');
+  } finally {
+    await runtime.close();
+  }
   assert.equal(closed, 5);
   await assert.rejects(
     startBroker({

packages/processing-control/tests/stop-run.test.mjs

@@ -105,7 +105,7 @@ test('a separate host joins durable first ownership without duplicate signals',
   const messages = []; child.on('message', value => messages.push(value));
   await once(child, 'message');
   const joined = stopRun({ ...f.args, owner: { decision: 'second-host' } },
-    { ...f.options, sleep: ms => new Promise(resolve => setTimeout(resolve, ms)) });
+    { ...f.options, now: Date.now, sleep: ms => new Promise(resolve => setTimeout(resolve, ms)) });
   setTimeout(() => child.send('continue'), 30);
   const result = await joined;
   const exit = child.exitCode === null ? await once(child, 'exit') : [child.exitCode];

packages/webui/tests/processing.test.mjs

@@ -52,15 +52,26 @@ test('Console halt and generation-bound resume use only the management writer an
 test('processing write fences refuse cross-site, wrong origin, unnamed host, duplicate identity and bad payload before writer', async t => {
   const f = await fixture(t);
   const cases = [
-    { headers: { ...POST.headers, origin: 'https://elsewhere' } },
-    { headers: { ...POST.headers, 'sec-fetch-site': 'cross-site' } },
-    { headers: { ...POST.headers, host: `127.0.0.1:${f.server.address().port}`, 'x-user': undefined } },
-    { headers: { ...POST.headers, 'x-user': ['jason', 'jason'] } },
-    { headers: { ...POST.headers, 'content-type': 'text/plain' } },
-    { body: '{"human":"jason"}' }, { body: '[]' }, { body: 'null' }, { body: ' '.repeat(1025) + '{}' },
+    ['wrong origin', { headers: { ...POST.headers, origin: 'https://elsewhere' } }, 403, 'cross-origin request refused'],
+    ['cross-site', { headers: { ...POST.headers, 'sec-fetch-site': 'cross-site' } }, 403, 'cross-site'],
+    ['unnamed host', { headers: { ...POST.headers, host: `127.0.0.1:${f.server.address().port}`,
+      origin: `http://127.0.0.1:${f.server.address().port}`, 'x-user': undefined } }, 403, 'read-only'],
+    ['duplicate identity', { headers: { ...POST.headers, 'x-user': ['jason', 'jason'] } }, 401, 'sign-in required'],
+    ['content type', { headers: { ...POST.headers, 'content-type': 'text/plain' } }, 415, 'invalid-request'],
+    ['extra key', { body: '{"human":"jason"}' }, 400, 'invalid-request'],
+    ['array', { body: '[]' }, 400, 'invalid-request'],
+    ['null', { body: 'null' }, 400, 'invalid-request'],
+    ['body bound', { body: ' '.repeat(1025) + '{}' }, 400, 'invalid-request'],
   ];
-  for (const c of cases) assert.ok((await http(f.server, '/api/processing/halt', { ...POST, ...c })).status >= 400);
-  assert.equal((await http(f.server, '/api/processing/resume', POST)).status, 400);
+  for (const [name, changed, status, error] of cases) {
+    const response = await http(f.server, '/api/processing/halt', { ...POST, ...changed });
+    assert.equal(response.status, status, name);
+    assert.equal(response.value.error, error, name);
+    assert.deepEqual(f.calls, [], name + ' must refuse before the writer');
+  }
+  const missingGeneration = await http(f.server, '/api/processing/resume', POST);
+  assert.equal(missingGeneration.status, 400);
+  assert.equal(missingGeneration.value.error, 'invalid-request');
   assert.deepEqual(f.calls, []);
 });
 test('tester cannot invoke either control writer and failures never echo backend text', async t => {