CI: corrected markers as their own fixture part; fix and locate the #754 flakes

scripts/conformance/server-parity.mjs

@@ -1,5 +1,5 @@
 #!/usr/bin/env node
-import {mkdtempSync, readFileSync, rmSync} from 'node:fs';
+import {mkdtempSync, readFileSync, rmSync, writeFileSync} from 'node:fs';
 import {createHash} from 'node:crypto';
 import {fileURLToPath} from 'node:url';
 import {basename, dirname, join, resolve} from 'node:path';
@@ -20,6 +20,8 @@ import {checkCorrectedMarkerArtifacts, checkCorrectedMarkerPackages} from './ser
 import {checkCorrectedRangeArtifacts, checkCorrectedRangePackages} from './server-parity/corrected-range-contract.mjs';
 import {checkActivityDeadlinePackages} from './server-parity/activity-deadline-contract.mjs';
 import {checkActivityDeadlineReceiptPackages} from './server-parity/activity-deadline-receipts-contract.mjs';
+import {contractFailureLocation} from './server-parity/failure-location.mjs';
+import {failureExcerpt} from './server-parity/failure-excerpt.mjs';
 
 const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..');
 const {values, positionals} = parseArgs({allowPositionals: true, options: {
@@ -139,7 +141,8 @@ try {
           // its own workflow. Report it and keep recording, so one run shows
           // every difference. A runner-blocked case still stops the record.
           if (item.outcome === 'product-fail') {
-            console.error(`${fixture.id}: ${error.message}`);
+            const location = contractFailureLocation(error);
+            console.error(`${fixture.id}: ${error.message}${location ? `\n  at ${location}` : ''}`);
             continue;
           }
           record.outcome = item.outcome;
@@ -155,6 +158,9 @@ try {
         record.outcome = 'product-fail';
         record.finished_at = new Date().toISOString();
         writeRecord(values.output, record, originals);
+        // The artifact summary keeps this outline; it does not keep the record.
+        writeFileSync(values.output.replace(/\.json$/, '') + '.failure-excerpt.txt',
+          failureExcerpt(record.cases.filter(item => item.outcome !== 'pass')));
         throw new Error(`${failed.length} fixture(s) differ: ${failed.join(', ')}`);
       }
       record.outcome = 'pass';

scripts/conformance/server-parity/artifact-summary.mjs

@@ -12,7 +12,7 @@ function priority(path, failed) {
   if (failed && /(?:^|\/)query-authority\/clients\/(?:php|rust)\/[^/]+\/(?:client\.stderr|observer\.log)$/.test(path)) return 0;
   if (/^(rust|mysql|postgres|comparator|codec-load|artifact-summary)(-execution)?-tests\.txt$/.test(name)) return 0;
   if (/^(comparison|compare)([-.].*)?\.json$/.test(name) || /-comparison\.json$/.test(name)) return 1;
-  if (failed && (/^(?:php|embedded|rust)-record\.txt$/.test(name)
+  if (failed && (/^(?:php|embedded|rust)-record\.txt$/.test(name) || /\.failure-excerpt\.txt$/.test(name)
       || /^sdk-(?:signals|queries)-profile\.log$/.test(name))) return 1;
   if (failed && (/\.stderr$/.test(name) || /(?:stderr|error|failure|bootstrap|install|build).*(?:\.log|\.txt)$/.test(name)
       || /^(?:php|rust|native)(?:[-.].*)?\.log$/.test(name) || /server\.log$/.test(name))) return 2;

scripts/conformance/server-parity/failure-excerpt.mjs

@@ -0,0 +1,52 @@
+// A bounded, value-free outline of every wire request a failed case observed,
+// in observation order, so a flake can be diagnosed from the retained artifact
+// without keeping the record. It names each request's method, path, status,
+// transport flags, worker and poll identity. Request and response bodies,
+// payloads and credentials are never read into it.
+
+const SKIP = new Set(['request', 'response', 'request_body', 'response_body', 'response_bytes_base64',
+  'transport_jsonl', 'stdout', 'stderr', 'fixture']);
+
+function line(path, item) {
+  const fields = [`${path} ${item.method} ${item.path} ${item.status}`];
+  for (const key of ['client_cancelled', 'transport_error', 'request_body_complete', 'upstream_request_started',
+    'response_retry_after']) {
+    if (Object.hasOwn(item, key) && item[key] !== null && item[key] !== false) fields.push(`${key}=${item[key]}`);
+  }
+  const request = item.request && typeof item.request === 'object' ? item.request : {};
+  for (const key of ['worker_id', 'poll_request_id', 'task_queue', 'lease_owner']) {
+    if (typeof request[key] === 'string') fields.push(`${key}=${request[key].slice(0, 120)}`);
+  }
+  return fields.join(' ');
+}
+
+function walk(value, path, out, limit) {
+  if (out.length >= limit || value === null || typeof value !== 'object') return;
+  if (Array.isArray(value)) {
+    value.forEach((item, index) => walk(item, `${path}[${index}]`, out, limit));
+    return;
+  }
+  if (typeof value.method === 'string' && typeof value.path === 'string' && Object.hasOwn(value, 'status')) {
+    out.push(line(path, value));
+  }
+  for (const [key, child] of Object.entries(value)) {
+    if (!SKIP.has(key)) walk(child, `${path}.${key}`, out, limit);
+  }
+}
+
+export function failureExcerpt(cases, {linesPerCase = 300, maxBytes = 60 * 1024} = {}) {
+  let text = '';
+  for (const item of cases) {
+    const lines = [];
+    walk(item.observation, 'observation', lines, linesPerCase + 1);
+    const truncated = lines.length > linesPerCase;
+    const block = [`== ${item.fixture_id} (${item.outcome}): ${String(item.error ?? '').split('\n')[0].slice(0, 300)}`,
+      ...lines.slice(0, linesPerCase), ...(truncated ? [`… more requests omitted after ${linesPerCase}`] : []), ''].join('\n');
+    if (Buffer.byteLength(text + block) > maxBytes) {
+      text += `… further failed cases omitted at ${maxBytes} bytes\n`;
+      break;
+    }
+    text += block;
+  }
+  return text;
+}

scripts/conformance/server-parity/failure-location.mjs

@@ -0,0 +1,10 @@
+// The contract line that refused a fixture, as "scripts/conformance/...:line".
+// A bare "false !== true" is not locatable from a retained log; the stack
+// frame is. Frames carry source paths and line numbers, never payload values.
+export function contractFailureLocation(error) {
+  for (const frame of String(error?.stack ?? '').split('\n').slice(1)) {
+    const match = frame.match(/(scripts\/conformance\/[^\s:()]+\.m?js):(\d+):\d+/);
+    if (match) return `${match[1]}:${match[2]}`;
+  }
+  return null;
+}

scripts/conformance/server-parity/sdk-query-direction-contract.mjs

@@ -35,6 +35,15 @@ const one = (requests, predicate, message) => {
   assert.equal(selected.length, 1, message);
   return selected[0];
 };
+// A wrong status names the request and its observer transport flags, never
+// payload bytes. Status 0 means no upstream status arrived, and the flags say
+// whether the client closed first or the upstream connection failed.
+function wireStatus(item, expected, label) {
+  if (item.status === expected) return;
+  const {method, path, status, client_cancelled, transport_error, request_body_complete, upstream_request_started} = item;
+  assert.fail(`${label}: expected HTTP ${expected}, observed ${status} for ${method} ${path} `
+    + JSON.stringify({client_cancelled, transport_error, request_body_complete, upstream_request_started}));
+}
 function unchanged(before, after, message) {
   assert.deepEqual(after, before, message);
 }
@@ -144,7 +153,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
       assert(!phase.requests.some(item => item.method === 'POST' && item.path === '/api/workflows'));
       const wire = one(phase.requests, item => item.method === 'POST'
         && item.path === '/api/workflows/' + workflowId + '/signal/payload', 'one actual signal stimulus');
-      assert.equal(wire.status, 202);
+      wireStatus(wire, 202, phase.phase + ' signal');
       assert.equal(wire.response.run_id, observation.run_id);
       assert.equal(wire.response.workflow_id, workflowId);
       assert.equal(wire.response.outcome, 'signal_received');
@@ -165,7 +174,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
       unchangedDurableSnapshot(phase.snapshot_before, direction.phases[index - 1].snapshot, 'same acknowledged author snapshot');
       const wire = one(phase.requests, item => item.method === 'POST'
         && item.path === '/api/workflows/' + workflowId + '/query/state', 'one normal instance-scoped query');
-      assert.equal(wire.status, 200);
+      wireStatus(wire, 200, phase.phase + ' query');
       assert.equal(wire.response.workflow_id, workflowId);
       assert.equal(wire.response.run_id, observation.run_id);
       assert.equal(wire.response.query_name, 'state');
@@ -188,7 +197,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
       const claim = one(author.requests, item => item.path === '/api/worker/query-tasks/poll'
         && item.response?.task, 'one actual query worker claim');
       const task = claim.response.task;
-      assert.equal(claim.method, 'POST'); assert.equal(claim.status, 200);
+      assert.equal(claim.method, 'POST'); wireStatus(claim, 200, author.phase + ' query claim');
       assert.equal(claim.request.task_queue, spec.task_queue);
       assert.equal(task.workflow_id, workflowId); assert.equal(task.run_id, observation.run_id);
       assert.equal(task.workflow_type, fixture.workflow_type); assert.equal(task.query_name, 'state');
@@ -211,7 +220,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
       if (wire.request.input) assert.equal(frame(wire.request.input), frame(task.query_arguments));
       const complete = one(author.requests, item => item.method === 'POST'
         && item.path === '/api/worker/query-tasks/' + task.query_task_id + '/complete', 'actual original query completion');
-      assert.equal(complete.status, 200);
+      wireStatus(complete, 200, author.phase + ' query completion');
       assert.equal(complete.request.lease_owner, task.lease_owner);
       assert.equal(complete.request.query_task_attempt, task.query_task_attempt);
       assert.equal(complete.response.query_task_id, task.query_task_id);
@@ -232,7 +241,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
     if (language === 'php') assert.equal(phase.query_completions, 1);
     const register = one(phase.requests, item => item.method === 'POST'
       && item.path === '/api/worker/register', 'one ordinary worker registration');
-    assert.equal(register.status, 201); assert.equal(register.response.registered, true);
+    wireStatus(register, 201, phase.phase + ' registration'); assert.equal(register.response.registered, true);
     assert.equal(register.request.worker_id, phase.worker_id);
     assert.equal(register.request.task_queue, spec.task_queue);
     assert.deepEqual(register.request.supported_workflow_types, [fixture.workflow_type]);
@@ -241,7 +250,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
     const claim = one(phase.requests, item => item.path === '/api/worker/workflow-tasks/poll'
       && item.response?.task, 'one original workflow claim per cold author');
     const task = claim.response.task;
-    assert.equal(claim.status, 200);
+    wireStatus(claim, 200, phase.phase + ' workflow claim');
     for (const key of ['workflow_id', 'run_id', 'workflow_type', 'task_queue']) assert.equal(task[key], observation[key]);
     assert.equal(task.workflow_task_attempt, 1); nonempty(task.task_id); nonempty(task.lease_owner);
     checkWorkerHistory(task.history_events);
@@ -251,7 +260,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
     assert.deepEqual(history(task.history_events), history(observation.events.slice(0, [2, 4, 9][turn])));
     const complete = one(phase.requests, item => item.method === 'POST'
       && item.path === '/api/worker/workflow-tasks/' + task.task_id + '/complete', 'original fenced workflow completion');
-    assert.equal(complete.status, 200); assert.equal(complete.response.recorded, true);
+    wireStatus(complete, 200, phase.phase + ' workflow completion'); assert.equal(complete.response.recorded, true);
     assert.equal(complete.response.task_id, task.task_id); assert.equal(complete.response.run_id, observation.run_id);
     assert.equal(complete.request.lease_owner, task.lease_owner);
     assert.equal(complete.request.workflow_task_attempt, task.workflow_task_attempt);
@@ -264,7 +273,7 @@ export function checkSdkQueryDirection(fixture, observation, workflowId) {
     if (turn === 0) {
       const start = one(phase.requests, item => item.method === 'POST'
         && item.path === '/api/workflows', 'one real observer start');
-      assert.equal(start.status, 201);
+      wireStatus(start, 201, phase.phase + ' start');
       assert.equal(start.request.workflow_id, workflowId); assert.equal(start.response.run_id, observation.run_id);
       assert(phase.requests.indexOf(register) < phase.requests.indexOf(start));
       assert(phase.requests.indexOf(start) < phase.requests.indexOf(claim), 'original first claim after successful start receipt');

scripts/conformance/server-parity/sdk-query-direction-probe.php

@@ -191,6 +191,35 @@ function sdkQueryWorkflowAcknowledgement(array $requests, string $workflowId, st
     return null;
 }
 
+/**
+ * True once the author's own completion of a query task it claimed for this
+ * run has a successful, uncancelled response in its journal.
+ */
+function sdkQueryCompletionAcknowledged(array $requests, string $workflowId, string $runId): bool
+{
+    foreach ($requests as $pollIndex => $poll) {
+        $task = is_array($poll['response'] ?? null) ? ($poll['response']['task'] ?? null) : null;
+        if (($poll['method'] ?? null) !== 'POST' || ($poll['path'] ?? null) !== '/api/worker/query-tasks/poll'
+            || ($poll['status'] ?? null) !== 200 || ! is_array($task)
+            || ($task['workflow_id'] ?? null) !== $workflowId || ($task['run_id'] ?? null) !== $runId
+            || ! is_string($task['query_task_id'] ?? null) || $task['query_task_id'] === '') {
+            continue;
+        }
+        $path = '/api/worker/query-tasks/'.$task['query_task_id'].'/complete';
+        foreach ($requests as $index => $completion) {
+            if ($index > $pollIndex && ($completion['method'] ?? null) === 'POST' && ($completion['path'] ?? null) === $path
+                && ($completion['status'] ?? null) === 200 && ($completion['client_cancelled'] ?? true) === false
+                && ($completion['transport_error'] ?? null) === null
+                && is_array($completion['response'] ?? null)
+                && ($completion['response']['query_task_id'] ?? null) === $task['query_task_id']) {
+                return true;
+            }
+        }
+    }
+
+    return false;
+}
+
 function sdkQueryActorFinish(array &$actor, float $deadline, string $evidence, string $workflowId): array
 {
     while (sdkQueryActorPoll($actor)) {
@@ -307,6 +336,15 @@ function sdkQueryDirectionHttpObservation(array $fixture, array $options): array
             $query = sdkQueryActorFinish($actors[$queryPhase], $deadline, $evidence, $options['workflow-id']);
             $after = sdkDirectionSnapshot($client, $options['workflow-id'], $runId);
             if (($query['client_finished'] ?? false) === true) {
+                // Native wakes the caller as soon as the completion commits, so
+                // the caller can finish before the author's own completion
+                // response arrives. Stopping the author then cancels that request
+                // (published Python runs queries on a thread its stop cancels).
+                // Release the author only after its completion is observed.
+                while (! sdkQueryCompletionAcknowledged(sdkQueryLiveWireRequests($author), $options['workflow-id'], $runId)
+                    && sdkQueryActorPoll($author) && microtime(true) < $deadline) {
+                    usleep(10000);
+                }
                 file_put_contents($author['receipt'], $actors[$queryPhase]['stdout']);
             } else {
                 proc_terminate($author['process']);

tests/Unit/ServerParityArtifactSummaryTest.mjs

@@ -134,3 +134,20 @@ test('failed recorder diagnostic tails survive without retaining raw records or
   summarizeEvidence(source, success, {outcome: 'success'});
   assert.equal(existsSync(join(success, 'php-record.txt')), false);
 });
+
+test('a failed record outline survives a full collection of setup logs and stays out of success summaries', t => {
+  const {source, output, root} = fixture(t);
+  for (let index = 0; index < 140; index++) {
+    writeFileSync(join(source, `completed-${index}-install.txt`), 'installed successfully\n');
+  }
+  mkdirSync(join(source, 'legacy-marker/php'), {recursive: true});
+  writeFileSync(join(source, 'legacy-marker/php/record.failure-excerpt.txt'), '== legacy-one-marker-control (product-fail)\n');
+  writeFileSync(join(source, 'legacy-marker/php/record.json'), 'raw responses');
+  summarizeEvidence(source, output, {outcome: 'failure'});
+  assert.equal(readFileSync(join(output, 'legacy-marker/php/record.failure-excerpt.txt'), 'utf8'),
+    '== legacy-one-marker-control (product-fail)\n');
+  assert.equal(existsSync(join(output, 'legacy-marker/php/record.json')), false);
+  const success = join(root, 'success');
+  summarizeEvidence(source, success, {outcome: 'success'});
+  assert.equal(existsSync(join(success, 'legacy-marker/php/record.failure-excerpt.txt')), false);
+});

tests/Unit/ServerParityFailureExcerptTest.mjs

@@ -0,0 +1,39 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import {failureExcerpt} from '../../scripts/conformance/server-parity/failure-excerpt.mjs';
+
+const poll = (status, extra = {}) => ({method: 'POST', path: '/api/worker/workflow-tasks/poll', status,
+  request: {worker_id: 'w1', poll_request_id: 'p1', task_queue: 'q', secret_input: 'payload-bytes'},
+  response: {reason: 'lock pressure', nested: {method: 'GET', path: '/not-a-request', status: 200}},
+  request_body: '{"secret_input":"payload-bytes"}', response_body: 'body-bytes', client_cancelled: false,
+  transport_error: null, ...extra});
+
+test('a failed case lists its requests in order with identity and transport flags, never bodies', () => {
+  const text = failureExcerpt([{fixture_id: 'legacy-one-marker-control', outcome: 'product-fail',
+    error: 'same original poll recovers\nsecond line', observation: {legacy_marker: {phases: [
+      {role: 'original', requests: [poll(503, {response_retry_after: '2'}), poll(200)]},
+      {role: 'replacement', requests: [{method: 'DELETE', path: '/api/worker/registrations/w1', status: 200,
+        request: null, client_cancelled: true}]}]}}}]);
+  assert.equal(text, [
+    '== legacy-one-marker-control (product-fail): same original poll recovers',
+    'observation.legacy_marker.phases[0].requests[0] POST /api/worker/workflow-tasks/poll 503 response_retry_after=2 worker_id=w1 poll_request_id=p1 task_queue=q',
+    'observation.legacy_marker.phases[0].requests[1] POST /api/worker/workflow-tasks/poll 200 worker_id=w1 poll_request_id=p1 task_queue=q',
+    'observation.legacy_marker.phases[1].requests[0] DELETE /api/worker/registrations/w1 200 client_cancelled=true',
+    ''].join('\n'));
+  assert(!text.includes('payload-bytes') && !text.includes('body-bytes') && !text.includes('/not-a-request'));
+});
+
+test('the outline is bounded per case and in total', () => {
+  const many = Array.from({length: 50}, () => poll(200));
+  const one = failureExcerpt([{fixture_id: 'a', outcome: 'product-fail', observation: {requests: many}}], {linesPerCase: 10});
+  assert.equal(one.split('\n').filter(l => l.startsWith('observation')).length, 10);
+  assert.match(one, /more requests omitted after 10/);
+  const cases = Array.from({length: 20}, (_, i) => ({fixture_id: `case-${i}`, outcome: 'product-fail', observation: {requests: many}}));
+  const all = failureExcerpt(cases, {maxBytes: 8 * 1024});
+  assert(Buffer.byteLength(all) <= 8 * 1024 + 100);
+  assert.match(all, /further failed cases omitted at 8192 bytes/);
+});
+
+test('a case without an observation still names its error', () => {
+  assert.equal(failureExcerpt([{fixture_id: 'x', outcome: 'runner-blocked', error: 'boom'}]), '== x (runner-blocked): boom\n');
+});