Merge candidate 20260908T194055Z-delivery-trust

Change-source: pulse-maintainer

docs/RELEASE_RESOURCE_EVIDENCE.md

@@ -4,8 +4,7 @@
 before and after the backend phase, plus `RELEASE_BACKEND_TOOLCHAIN` and
 `RELEASE_BACKEND_EXIT`. The exact-preflight launcher's retained log preserves
 these lines even when it removes its disposable worker directory. No success
-receipt is produced by this evidence collector. Backend commands, exit status,
-parallelism and assertion thresholds are unchanged.
+receipt is produced by this evidence collector. Backend exit status, parallelism and assertion thresholds are unchanged.
 
 The collector reads only allowlisted kernel resource counters and numeric Go
 runtime settings. It does not dump the environment, process command lines,
@@ -42,3 +41,29 @@ python3 -m unittest discover -s scripts/release_control/internal -p 'release_res
 python3 -m unittest discover -s scripts/release_control/internal -p 'release_preflight_test.py' -v
 bash -n scripts/release-preflight-worker.sh
 ```
+
+## Stress-test event window
+
+Rehearsal backend Go output is streamed with `-json` and decoded by
+`release-go-test-events.py`. Every Output string is retained verbatim (including
+verbose test logs, skips and package summaries); stderr remains on stderr.
+Shell pipefail retains a failing Go exit. Invalid event input is drained,
+retained and fails the reader rather than quietly losing diagnostics.
+
+Only the exact API `TestMultiTenant_ConcurrentAPIStress` lifecycle receives
+`RELEASE_GO_TEST_EVENT` records with Go's event Time, a separately labelled
+receipt timestamp and the same allowlisted resource snapshot. Resource time is
+collection time, not the Go event time. Sampling and verbose streaming add
+measurement overhead; they do not reproduce uninstrumented execution. Cached
+results can replay events and are not fresh execution timing: check the package
+summary for `(cached)`. No cache, scheduling, threshold or test-selection policy
+is changed. Release-profile sharded tests are unchanged.
+
+This closes a prospective observability gap, not the historical qualification
+failure. Use only with an independently justified future run; no qualification
+retry is warranted merely to collect these records. Abrupt termination can omit
+terminal events. Cgroup ancestry remains shared context, not per-test CPU usage.
+
+```sh
+python3 -m unittest discover -s scripts/release_control/internal -p 'release_go_test_events_test.py' -v
+```

docs/release-control/v6/internal/subsystems/deployment-installability.md

@@ -27,7 +27,7 @@ Unmapped hierarchies, missing counters and collection failures are unavailable
 evidence, not zero pressure. Collection failure must neither replace a backend
 failure nor prevent its ordinary after-boundary observation.
 
-The backend retains errexit semantics, original exit status, test commands,
+The backend retains errexit semantics, original exit status, test selection,
 concurrency and qualification thresholds. Phase-boundary evidence cannot
 localise contention to a test, establish causation or clear historical adverse
 qualification; abrupt termination may omit the after record. The executed
@@ -5294,3 +5294,15 @@ All 4 tests, 7 Helm Pages retry tests and 47 promotion-policy tests pass locally
 No hosted publication or installed-image qualification is claimed.
 External reference retrieved 2026-09-05: https://helm.sh/docs/topics/charts/#the-appversion-field
 explains that application version is separate from chart version.
+
+### Rehearsal event timing
+
+`scripts/release-go-test-events.py` renders streamed Go JSON Output verbatim and
+adds bounded lifecycle/resource records for the API concurrent stress test.
+The worker uses pipefail; source failures remain failures and malformed event
+input fails closed after draining. Event, receipt and resource collection times
+are distinct. Cached events and collection overhead limit timing inference.
+`scripts/release_control/internal/release_go_test_events_test.py` covers real
+synthetic Go pass/skip/fail output, producer exit retention, malformed input,
+allowlisted targeting and unavailable resource evidence. This is not product
+qualification; see `docs/RELEASE_RESOURCE_EVIDENCE.md`.

scripts/release-go-test-events.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Render streamed go test -json output and retain bounded stress-test timing.
+
+Go's event Time is distinct from receipt time and the resource sample time.
+Samples share the worker cgroup; they do not establish per-test CPU usage or
+causality. Output is rendered verbatim, including package summaries and failures.
+The shell's pipefail retains the Go verdict; malformed input fails this reader.
+"""
+import importlib.util
+import json
+from pathlib import Path
+import sys
+import time
+
+spec = importlib.util.spec_from_file_location(
+    "release_resource_snapshot", Path(__file__).with_name("release-resource-snapshot.py")
+)
+resources = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(resources)
+
+TARGET = "TestMultiTenant_ConcurrentAPIStress"
+PACKAGE = "github.com/rcourtman/pulse-go-rewrite/internal/api"
+ACTIONS = {"run", "pause", "cont", "pass", "fail", "skip"}
+
+
+def render(source, destination):
+    invalid = False
+    for line in source:
+        received = time.time_ns()
+        try:
+            event = json.loads(line)
+            if not isinstance(event, dict):
+                raise ValueError("not an event object")
+            output = event.get("Output", "")
+            if not isinstance(output, str):
+                raise ValueError("non-text output")
+        except (ValueError, TypeError):
+            # Drain the stream rather than masking the producer with SIGPIPE.
+            destination.write(line)
+            invalid = True
+            continue
+        destination.write(output)
+        if (event.get("Package") == PACKAGE and event.get("Test") == TARGET
+                and event.get("Action") in ACTIONS):
+            evidence = {key: event[key] for key in
+                        ("Time", "Action", "Package", "Test", "Elapsed") if key in event}
+            evidence["received_unix_time_ns"] = received
+            try:
+                evidence["resources"] = resources.snapshot()
+            except Exception:
+                evidence["resources"] = {"unavailable": "snapshot collection failed"}
+            destination.write("RELEASE_GO_TEST_EVENT " + json.dumps(evidence, sort_keys=True) + "\n")
+        destination.flush()
+    if invalid:
+        destination.write("RELEASE_GO_TEST_STREAM invalid JSON event input\n")
+        destination.flush()
+    return int(invalid)
+
+
+if __name__ == "__main__":
+    raise SystemExit(render(sys.stdin, sys.stdout))

scripts/release-preflight-worker.sh

@@ -223,7 +223,14 @@ run_backend() (
   rm -rf "$TEST_DATA_DIR"
   mkdir -p "$TEST_DATA_DIR"
   if [ "$PROFILE" = "rehearsal" ]; then
-    phase backend-serial env PULSE_DATA_DIR="$TEST_DATA_DIR" go test -p 1 ./...
+    # -json streams test events instead of waiting for the package buffer.
+    # Keep readable Output and the original verdict via pipefail; instrument
+    # only the known stress-test window, without changing tests or thresholds.
+    backend_serial() {
+      env PULSE_DATA_DIR="$TEST_DATA_DIR" go test -json -p 1 ./... |
+        python3 ./scripts/release-go-test-events.py
+    }
+    phase backend-serial backend_serial
   else
     phase backend-race-sharded ./scripts/run-release-backend-tests.sh \
       --data-root "$TEST_DATA_DIR" \

scripts/release_control/internal/release_go_test_events_test.py

@@ -0,0 +1,91 @@
+import importlib.util
+import io
+import json
+from pathlib import Path
+import subprocess
+import sys
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[3]
+SCRIPT = ROOT / 'scripts/release-go-test-events.py'
+spec = importlib.util.spec_from_file_location('events', SCRIPT)
+events = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(events)
+
+
+class EventsTest(unittest.TestCase):
+    def test_output_and_bounded_evidence(self):
+        data = [dict(Time='2026-09-08T19:00:00Z', Action=action,
+                     Package=events.PACKAGE, Test=events.TARGET, Elapsed=2.0)
+                for action in ('run', 'pause', 'cont', 'pass', 'fail', 'skip')]
+        data += [dict(Action='output', Output='FAIL\tpackage\t2s\n'),
+                 dict(Action='run', Package='other', Test=events.TARGET),
+                 dict(Action='run', Package=events.PACKAGE, Test='TestOther')]
+        output = io.StringIO()
+        with patch.object(events.resources, 'snapshot', return_value={'unix_time_ns': 456}) as sample:
+            self.assertEqual(events.render(io.StringIO(''.join(json.dumps(x)+'\n' for x in data)), output), 0)
+        self.assertEqual(sample.call_count, 6)
+        lines = output.getvalue().splitlines()
+        self.assertEqual(lines[-1], 'FAIL\tpackage\t2s')
+        for line in lines[:-1]:
+            evidence = json.loads(line.removeprefix('RELEASE_GO_TEST_EVENT '))
+            self.assertEqual(evidence['Time'], data[0]['Time'])
+            self.assertGreater(evidence['received_unix_time_ns'], 456)
+            self.assertEqual(evidence['resources'], {'unix_time_ns': 456})
+
+    def test_unavailable_snapshot_does_not_change_verdict(self):
+        event = dict(Action='fail', Package=events.PACKAGE, Test=events.TARGET)
+        output = io.StringIO()
+        with patch.object(events.resources, 'snapshot', side_effect=OSError):
+            self.assertEqual(events.render(io.StringIO(json.dumps(event)+'\n'), output), 0)
+        self.assertIn('snapshot collection failed', output.getvalue())
+
+    def test_invalid_input_is_retained_and_drained(self):
+        output = io.StringIO()
+        self.assertEqual(events.render(io.StringIO('not-json\n[]\n{"Output":"later\\n"}\n'), output), 1)
+        self.assertTrue(output.getvalue().startswith('not-json\n[]\nlater\n'))
+
+    def test_real_go_pass_skip_and_fail(self):
+        # A disposable standard-library-only package, never the product suite.
+        with tempfile.TemporaryDirectory() as raw:
+            root = Path(raw)
+            (root / 'go.mod').write_text('module ' + events.PACKAGE + '\n\ngo 1.26.0\n')
+            (root / 'event_test.go').write_text('''package api
+import ("os"; "testing")
+func TestMultiTenant_ConcurrentAPIStress(t *testing.T) {
+ t.Log("synthetic log")
+ if os.Getenv("FIXTURE_FAIL") == "1" { t.Fatal("synthetic failure") }
+}
+func TestSkipped(t *testing.T) { t.Skip("synthetic skip") }
+''')
+            for fail in ('0', '1'):
+                env = {**os.environ, 'GOWORK': 'off', 'FIXTURE_FAIL': fail}
+                result = subprocess.run(['bash', '-o', 'pipefail', '-c',
+                    'go test -json -p 1 -count=1 . | "$1" "$2"',
+                    'fixture', sys.executable, str(SCRIPT)], cwd=root, env=env,
+                    capture_output=True, text=True, timeout=60)
+                self.assertEqual(result.returncode, int(fail), result.stderr)
+                records = [json.loads(line.removeprefix('RELEASE_GO_TEST_EVENT '))
+                           for line in result.stdout.splitlines()
+                           if line.startswith('RELEASE_GO_TEST_EVENT ')]
+                self.assertEqual([r['Action'] for r in records],
+                                 ['run', 'fail' if fail == '1' else 'pass'])
+                self.assertTrue(all(r.get('Time') for r in records))
+                self.assertIn('synthetic log', result.stdout)
+                self.assertIn('--- SKIP: TestSkipped', result.stdout)
+                self.assertIn('FAIL' if fail == '1' else 'PASS', result.stdout)
+
+    def test_pipeline_preserves_producer_exit(self):
+        for code in (0, 17):
+            result = subprocess.run(['bash', '-o', 'pipefail', '-c',
+                '(printf \'%s\\n\' \'{"Action":"output","Output":"FAIL\\n"}\'; exit "$1") | "$2" "$3"',
+                'fixture', str(code), sys.executable, str(SCRIPT)], capture_output=True, text=True)
+            self.assertEqual(result.returncode, code, result.stderr)
+            self.assertEqual(result.stdout, 'FAIL\n')
+
+
+if __name__ == '__main__':
+    unittest.main()

scripts/release_control/internal/release_preflight_test.py

@@ -312,6 +312,40 @@ echo CONTINUED
                         self.assertIn('backend-serial' if profile == 'rehearsal' else 'backend-race-sharded', result.stdout)
 
 
+    def test_rehearsal_stream_uses_real_decoder_and_preserves_failure(self):
+        worker = (ROOT / 'scripts/release-preflight-worker.sh').read_text()
+        function = re.search(r'^run_backend\(\) \(\n.*?^\)', worker, re.M | re.S).group(0)
+        for code in (0, 17):
+            script = f'''set -euo pipefail
+PROFILE=rehearsal
+TEST_DATA_DIR=unused
+ACTUAL_GO=fixture
+rm() {{ :; }}
+mkdir() {{ :; }}
+python3() {{
+  if [ "$1" = ./scripts/release-resource-snapshot.py ]; then
+    echo "snapshot $2"
+  else
+    command python3 "$@"
+  fi
+}}
+env() {{
+  [ "$*" = 'PULSE_DATA_DIR=unused go test -json -p 1 ./...' ] || return 42
+  printf '%s\\n' '{{"Action":"output","Output":"synthetic verdict\\n"}}'
+  return {code}
+}}
+phase() {{ shift; "$@"; }}
+{function}
+run_backend
+echo CONTINUED
+'''
+            result = subprocess.run(['bash', '-c', script], cwd=ROOT, text=True, capture_output=True)
+            self.assertEqual(result.returncode, code, result.stderr)
+            self.assertIn('synthetic verdict\n', result.stdout)
+            self.assertIn('snapshot after', result.stdout)
+            self.assertIn(f'RELEASE_BACKEND_EXIT {code}', result.stdout)
+            self.assertEqual('CONTINUED' in result.stdout, code == 0)
+
     def test_browser_mount_identity_and_locked_cli_follow_the_docker_daemon(self) -> None:
         worker = (ROOT / "scripts/release-preflight-worker.sh").read_text()
         functions = []