Improve lock publish diagnostics and scheduler failure summaries

scripts/agent/run-scheduler-cycle.mjs

@@ -50,6 +50,23 @@ function parseJsonFromOutput(text) {
   return null;
 }
 
+function parseJsonEventsFromOutput(text) {
+  const events = [];
+  for (const line of String(text || '').split(/\r?\n/)) {
+    const trimmed = line.trim();
+    if (!trimmed.startsWith('{')) continue;
+    try {
+      const parsed = JSON.parse(trimmed);
+      if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
+        events.push(parsed);
+      }
+    } catch {
+      // Ignore non-JSON lines.
+    }
+  }
+  return events;
+}
+
 async function runCommand(command, args = [], options = {}) {
   return new Promise((resolve) => {
     const child = spawn(command, args, {
@@ -428,6 +445,7 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr
   const lockCommandArgs = ['run', 'lock:lock', '--', '--agent', selectedAgent, '--cadence', cadence];
   const backoffScheduleMs = [];
   let attempts = 0;
+  const correlationId = idempotencyKey || `${cadence}:${selectedAgent}:${getRunDateKey()}`;
 
   while (true) {
     attempts += 1;
@@ -435,11 +453,13 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr
       env: {
         AGENT_PLATFORM: platform,
         ...(idempotencyKey ? { SCHEDULER_LOCK_IDEMPOTENCY_KEY: idempotencyKey } : {}),
+        SCHEDULER_LOCK_CORRELATION_ID: correlationId,
+        SCHEDULER_LOCK_ATTEMPT_ID: String(attempts),
       },
     });
 
     if (result.code !== 2) {
-      return { result, attempts, backoffScheduleMs };
+      return { result, attempts, backoffScheduleMs, correlationId };
     }
 
     if (attempts > lockRetry.maxRetries) {
@@ -447,6 +467,7 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr
         result,
         attempts,
         backoffScheduleMs,
+        correlationId,
         finalBackendCategory: classifyLockBackendError(`${result.stderr}\n${result.stdout}`),
       };
     }
@@ -463,6 +484,7 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr
       attempt: attempts,
       max_retries: lockRetry.maxRetries,
       delay_ms: delayMs,
+      correlation_id: correlationId,
       selected_agent: selectedAgent,
       cadence,
     }));
@@ -473,6 +495,34 @@ async function acquireLockWithRetry({ selectedAgent, cadence, platform, lockRetr
   }
 }
 
+function summarizeLockFailureReasons(outputText) {
+  const events = parseJsonEventsFromOutput(outputText);
+  const quorumFailure = [...events].reverse().find((entry) => entry.event === 'lock_publish_quorum_failed');
+  if (quorumFailure && quorumFailure.reasonDistribution && typeof quorumFailure.reasonDistribution === 'object') {
+    return {
+      reasonDistribution: quorumFailure.reasonDistribution,
+      attemptId: quorumFailure.attemptId || null,
+      correlationId: quorumFailure.correlationId || null,
+      totalElapsedMs: quorumFailure.totalElapsedMs ?? null,
+    };
+  }
+
+  const distribution = {};
+  for (const entry of events) {
+    if (entry.event !== 'lock_publish_failure') continue;
+    const reason = typeof entry.reason === 'string' && entry.reason.trim()
+      ? entry.reason.trim()
+      : 'unknown';
+    distribution[reason] = (distribution[reason] || 0) + 1;
+  }
+  return {
+    reasonDistribution: distribution,
+    attemptId: null,
+    correlationId: null,
+    totalElapsedMs: null,
+  };
+}
+
 function toYamlScalar(value) {
   const str = String(value ?? '');
   return `'${str.replace(/'/g, "''")}'`;
@@ -622,6 +672,7 @@ async function main() {
 
     if (lockResult.code === 2) {
       const combinedLockOutput = `${lockResult.stderr}\n${lockResult.stdout}`;
+      const diagnosticsSummary = summarizeLockFailureReasons(combinedLockOutput);
       const backendCategory = lockAttempt.finalBackendCategory || classifyLockBackendError(combinedLockOutput);
       const lockCommand = `AGENT_PLATFORM=${platform} npm run lock:lock -- --agent ${selectedAgent} --cadence ${cadence}`;
       const stderrExcerpt = excerptText(lockResult.stderr);
@@ -678,6 +729,10 @@ async function main() {
           failure_class: 'backend_unavailable',
           lock_attempts_total: lockAttempt.attempts,
           lock_backoff_schedule_ms: backoffSchedule || '(none)',
+          lock_correlation_id: diagnosticsSummary.correlationId || lockAttempt.correlationId,
+          lock_attempt_id: diagnosticsSummary.attemptId || String(lockAttempt.attempts),
+          lock_total_retry_timeline_ms: diagnosticsSummary.totalElapsedMs,
+          lock_failure_reason_distribution: JSON.stringify(diagnosticsSummary.reasonDistribution),
           backend_category: backendCategory,
           deferral_attempt_count: deferralAttemptCount,
           deferral_first_failure_timestamp: firstFailureAt,

src/lock-ops.mjs

@@ -69,6 +69,39 @@ function relayListLabel(relays) {
 
 function classifyPublishError(message) {
   const normalized = String(message || '').toLowerCase();
+  if (normalized.includes('publish timed out after') || normalized.includes('publish timeout')) {
+    return 'publish_timeout';
+  }
+  if (
+    normalized.includes('enotfound')
+    || normalized.includes('eai_again')
+    || normalized.includes('getaddrinfo')
+    || (normalized.includes('dns') && normalized.includes('websocket'))
+  ) {
+    return 'dns_resolution';
+  }
+  if (
+    normalized.includes('connect etimedout')
+    || normalized.includes('tcp connect timed out')
+    || normalized.includes('connect timeout')
+  ) {
+    return 'tcp_connect_timeout';
+  }
+  if (
+    normalized.includes('tls')
+    || normalized.includes('ssl')
+    || normalized.includes('certificate')
+    || normalized.includes('handshake')
+  ) {
+    return 'tls_handshake';
+  }
+  if (
+    normalized.includes('websocket')
+    || normalized.includes('bad response')
+    || normalized.includes('unexpected server response')
+  ) {
+    return 'websocket_open_failure';
+  }
   if (normalized.includes('timed out') || normalized.includes('timeout') || normalized.includes('etimedout')) {
     return 'network_timeout';
   }
@@ -89,7 +122,16 @@ function classifyPublishError(message) {
 }
 
 function isTransientPublishCategory(category) {
-  return category === 'network_timeout' || category === 'connection_reset' || category === 'relay_unavailable';
+  return [
+    'publish_timeout',
+    'dns_resolution',
+    'tcp_connect_timeout',
+    'tls_handshake',
+    'websocket_open_failure',
+    'network_timeout',
+    'connection_reset',
+    'relay_unavailable',
+  ].includes(category);
 }
 
 function calculateBackoffDelayMs(attemptNumber, baseMs, capMs, randomFn = Math.random) {
@@ -366,8 +408,12 @@ export async function publishLock(relays, event, deps = {}) {
     randomFn = Math.random,
     telemetryLogger = console.error,
     healthLogger = console.error,
+    diagnostics = {},
   } = deps;
 
+  const correlationId = diagnostics.correlationId || process.env.SCHEDULER_LOCK_CORRELATION_ID || 'none';
+  const attemptId = diagnostics.attemptId || process.env.SCHEDULER_LOCK_ATTEMPT_ID || '1';
+
   const pool = poolFactory();
   const publishTimeoutMs = getPublishTimeoutMsFn();
   const minSuccesses = getMinSuccessfulRelayPublishesFn();
@@ -423,11 +469,32 @@ export async function publishLock(relays, event, deps = {}) {
   try {
     maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:periodic');
     let lastAttemptState = null;
+    const retryTimeline = [];
+    const overallStartedAt = Date.now();
     for (let attemptNumber = 1; attemptNumber <= maxAttempts; attemptNumber += 1) {
       const startedAtMs = Date.now();
       lastAttemptState = await publishOnce();
+      const elapsedMs = Date.now() - startedAtMs;
+      retryTimeline.push({
+        publishAttempt: attemptNumber,
+        successCount: lastAttemptState.successCount,
+        relayAttemptedCount: lastAttemptState.attempted.size,
+        elapsedMs,
+      });
 
       if (lastAttemptState.successCount >= minSuccesses) {
+        telemetryLogger(JSON.stringify({
+          event: 'lock_publish_quorum_met',
+          correlationId,
+          attemptId,
+          publishAttempt: attemptNumber,
+          successCount: lastAttemptState.successCount,
+          relayAttemptedCount: lastAttemptState.attempted.size,
+          requiredSuccesses: minSuccesses,
+          timeoutMs: publishTimeoutMs,
+          retryTimeline,
+          totalElapsedMs: Date.now() - overallStartedAt,
+        }));
         console.error(
           `  Published to ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} relays `
           + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms)`,
@@ -443,13 +510,14 @@ export async function publishLock(relays, event, deps = {}) {
         break;
       }
 
-      const elapsedMs = Date.now() - startedAtMs;
       const nextDelayMs = calculateBackoffDelayMs(attemptNumber, retryBaseDelayMs, retryCapDelayMs, randomFn);
       for (const failure of lastAttemptState.failures) {
         if (!isTransientPublishCategory(failure.category)) continue;
         telemetryLogger(JSON.stringify({
           event: 'lock_publish_retry',
-          attempt: attemptNumber,
+          correlationId,
+          attemptId,
+          publishAttempt: attemptNumber,
           relayUrl: failure.relay,
           errorCategory: failure.category,
           elapsedMs,
@@ -460,13 +528,42 @@ export async function publishLock(relays, event, deps = {}) {
       await sleepFn(nextDelayMs);
     }
 
+    const reasonDistribution = {};
     const failureLines = lastAttemptState.failures
-      .map((result) => `${result.relay} (${result.phase}, category=${result.category}): ${result.message}`);
+      .map((result) => {
+        reasonDistribution[result.category] = (reasonDistribution[result.category] || 0) + 1;
+        telemetryLogger(JSON.stringify({
+          event: 'lock_publish_failure',
+          correlationId,
+          attemptId,
+          relayUrl: result.relay,
+          phase: result.phase,
+          reason: result.category,
+          message: result.message,
+        }));
+        return `${result.relay} (${result.phase}, reason=${result.category}): ${result.message}`;
+      });
+
+    telemetryLogger(JSON.stringify({
+      event: 'lock_publish_quorum_failed',
+      correlationId,
+      attemptId,
+      successCount: lastAttemptState.successCount,
+      relayAttemptedCount: lastAttemptState.attempted.size,
+      requiredSuccesses: minSuccesses,
+      timeoutMs: publishTimeoutMs,
+      attempts: maxAttempts,
+      reasonDistribution,
+      retryTimeline,
+      totalElapsedMs: Date.now() - overallStartedAt,
+    }));
 
     maybeLogHealthSnapshot(allRelays, healthConfig, healthLogger, 'publish:failure', true);
     throw new Error(
       `Failed relay publish quorum in publish phase: ${lastAttemptState.successCount}/${lastAttemptState.attempted.size} successful `
-      + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts})\n  ${failureLines.join('\n  ')}`,
+      + `(required=${minSuccesses}, timeout=${publishTimeoutMs}ms, attempts=${maxAttempts}, attempt_id=${attemptId}, correlation_id=${correlationId}, total_retry_timeline_ms=${Date.now() - overallStartedAt})\n`
+      + `  retry timeline: ${retryTimeline.map((item) => `#${item.publishAttempt}:${item.elapsedMs}ms`).join(', ')}\n`
+      + `  ${failureLines.join('\n  ')}`,
     );
   } finally {
     pool.close(allRelays);

test/lock-ops.test.mjs

@@ -109,30 +109,47 @@ describe('lock-ops', () => {
           telemetry.push(JSON.parse(line));
         },
         healthLogger: () => {},
+        diagnostics: {
+          attemptId: 'attempt-1',
+          correlationId: 'corr-1',
+        },
       });
 
       assert.strictEqual(result, event);
       assert.strictEqual(publishCalls, 2);
       assert.deepStrictEqual(sleepCalls, [250]);
-      assert.strictEqual(telemetry.length, 2);
+
-      assert.deepStrictEqual(telemetry[0], {
+      const retryEvents = telemetry.filter((entry) => entry.event === 'lock_publish_retry');
+      assert.strictEqual(retryEvents.length, 2);
+      assert.deepStrictEqual(retryEvents[0], {
         event: 'lock_publish_retry',
-        attempt: 1,
+        correlationId: 'corr-1',
+        attemptId: 'attempt-1',
+        publishAttempt: 1,
         relayUrl: 'wss://primary-a',
         errorCategory: 'connection_reset',
-        elapsedMs: telemetry[0].elapsedMs,
+        elapsedMs: retryEvents[0].elapsedMs,
         nextDelayMs: 250,
       });
-      assert.deepStrictEqual(telemetry[1], {
+      assert.deepStrictEqual(retryEvents[1], {
         event: 'lock_publish_retry',
-        attempt: 1,
+        correlationId: 'corr-1',
+        attemptId: 'attempt-1',
+        publishAttempt: 1,
         relayUrl: 'wss://primary-b',
         errorCategory: 'connection_reset',
-        elapsedMs: telemetry[1].elapsedMs,
+        elapsedMs: retryEvents[1].elapsedMs,
         nextDelayMs: 250,
       });
-      assert.ok(Number.isInteger(telemetry[0].elapsedMs));
+      assert.ok(Number.isInteger(retryEvents[0].elapsedMs));
-      assert.ok(Number.isInteger(telemetry[1].elapsedMs));
+      assert.ok(Number.isInteger(retryEvents[1].elapsedMs));
+
+      const metEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_met');
+      assert.ok(metEvent);
+      assert.equal(metEvent.correlationId, 'corr-1');
+      assert.equal(metEvent.attemptId, 'attempt-1');
+      assert.equal(metEvent.publishAttempt, 2);
+      assert.ok(Array.isArray(metEvent.retryTimeline));
     });
 
     it('fails after retry budget is exhausted for persistent transient failures', async () => {
@@ -162,7 +179,7 @@ describe('lock-ops', () => {
           telemetryLogger: () => {},
           healthLogger: () => {},
         }),
-        /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=3\)/,
+        /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=3,/,
       );
 
       assert.strictEqual(publishCalls, 3);
@@ -190,17 +207,63 @@ describe('lock-ops', () => {
           sleepFn: async () => {
             throw new Error('should not sleep for permanent failures');
           },
-          telemetryLogger: () => {
+          telemetryLogger: () => {},
-            throw new Error('should not emit retry telemetry for permanent failures');
-          },
           healthLogger: () => {},
         }),
-        /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4\)/,
+        /Failed relay publish quorum in publish phase: 0\/1 successful \(required=1, timeout=200ms, attempts=4,/,
       );
 
       assert.strictEqual(publishCalls, 1);
     });
 
+    it('includes per-relay reason categories and retry metadata in failure diagnostics', async () => {
+      const telemetry = [];
+      const mockPool = {
+        publish: (relays) => relays.map((relay) => {
+          if (relay.includes('dns')) return Promise.reject(new Error('getaddrinfo ENOTFOUND relay.example'));
+          if (relay.includes('tcp')) return Promise.reject(new Error('connect ETIMEDOUT 1.2.3.4:443'));
+          if (relay.includes('tls')) return Promise.reject(new Error('TLS handshake failure'));
+          if (relay.includes('ws')) return Promise.reject(new Error('websocket: unexpected server response'));
+          return Promise.reject(new Error('[publish:primary] Publish timed out after 200ms'));
+        }),
+        close: () => {},
+      };
+
+      await assert.rejects(
+        () => publishLock(['wss://dns', 'wss://tcp', 'wss://tls', 'wss://ws', 'relay-timeout'], { id: 'evt-diag' }, {
+          poolFactory: () => mockPool,
+          getPublishTimeoutMsFn: () => 200,
+          getMinSuccessfulRelayPublishesFn: () => 1,
+          getRelayFallbacksFn: () => [],
+          getMinActiveRelayPoolFn: () => 1,
+          retryAttempts: 1,
+          telemetryLogger: (line) => telemetry.push(JSON.parse(line)),
+          healthLogger: () => {},
+          diagnostics: { attemptId: 'attempt-42', correlationId: 'corr-42' },
+        }),
+      );
+
+      const summaryEvent = telemetry.find((entry) => entry.event === 'lock_publish_quorum_failed');
+      assert.ok(summaryEvent);
+      const failureEvents = telemetry.filter((entry) => entry.event === 'lock_publish_failure');
+      const failureText = failureEvents.map((entry) => `${entry.relayUrl}:${entry.reason}`).join('\n');
+
+      assert.match(failureText, /wss:\/\/dns:dns_resolution/);
+      assert.match(failureText, /wss:\/\/tcp:tcp_connect_timeout/);
+      assert.match(failureText, /wss:\/\/tls:tls_handshake/);
+      assert.match(failureText, /wss:\/\/ws:websocket_open_failure/);
+      assert.match(failureText, /relay-timeout:publish_timeout/);
+
+
+      assert.deepStrictEqual(summaryEvent.reasonDistribution, {
+        dns_resolution: 1,
+        tcp_connect_timeout: 1,
+        tls_handshake: 1,
+        websocket_open_failure: 1,
+        publish_timeout: 1,
+      });
+    });
+
     it('returns success when mixed relay outcomes still satisfy quorum', async () => {
       let publishCalls = 0;
       const mockPool = {
@@ -269,7 +332,6 @@ describe('lock-ops', () => {
 
       const event = { id: 'evt-health' };
 
-      // Warm up failure history so bad relay enters quarantine.
       for (let i = 0; i < 3; i += 1) {
         await publishLock(['wss://bad', 'wss://healthy'], event, {
           poolFactory: () => mockPool,

test/run-scheduler-cycle-memory-policy.test.mjs

@@ -193,6 +193,9 @@ test('records backend failure metadata when lock command exits with code 2', { c
   assert.match(failedBody, /lock_command: 'AGENT_PLATFORM=codex npm run lock:lock -- --agent agent-a --cadence daily'/);
   assert.match(failedBody, /lock_stderr_excerpt: 'publish failed to all relays token=\[REDACTED\] SECRET_KEY=\[REDACTED\]'/);
   assert.match(failedBody, /lock_stdout_excerpt: 'websocket: connection refused'/);
+  assert.match(failedBody, /lock_failure_reason_distribution: '\{\}'/);
+  assert.match(failedBody, /lock_attempt_id: '3'/);
+  assert.match(failedBody, /lock_correlation_id: 'daily:agent-a:/);
   assert.match(failedBody, /Retry AGENT_PLATFORM=codex npm run lock:lock -- --agent agent-a --cadence daily/);
   assert.match(failedBody, /platform: 'codex'/);
 });