fix: handle stale web observe runners (#875)

Co-authored-by: Codex <codex@noreply.local>
2026-06-25 14:28:29 +08:00
parent bc29ed0743
commit 2431fc73b2
4 changed files with 393 additions and 59 deletions
@@ -44,6 +44,7 @@ const errors = applyAnalysisFocus(filterRowsByTimeWindow(await readJsonl(dataFil
 const artifacts = applyAnalysisFocus(filterRowsByTimeWindow(await readJsonl(dataFile("artifacts.jsonl"), { tail: smallJsonlTailLimit }), sampleWindow), analysisFocus);
 const manifest = await readJson(path.join(stateDir, "manifest.json"));
 const heartbeat = await readJson(path.join(stateDir, "heartbeat.json"));
+const commandState = await readCommandState(stateDir);

 await mkdir(analysisDir, { recursive: true, mode: 0o700 });
 const transitions = buildTransitions(samples);
@@ -93,7 +94,8 @@ const runnerErrors = errors.slice(-8).map((item) => {
  };
 });
 const commandFailures = summarizeCommandFailures(control);
-const findings = buildFindings(samples, control, network, errors, sampleMetrics, promptNetwork, runtimeAlerts, pagePerformance, pageProvenance, commandFailures);
+const toolFindings = buildToolFindings({ manifest, heartbeat, commandState });
+const findings = [...toolFindings, ...buildFindings(samples, control, network, errors, sampleMetrics, promptNetwork, runtimeAlerts, pagePerformance, pageProvenance, commandFailures)];
 if (jsonlReadIssues.length > 0) findings.unshift({ id: "jsonl-read-issues", severity: "red", summary: "observer analyzer hit JSONL read/parse issues", count: jsonlReadIssues.length, issues: jsonlReadIssues.slice(0, 20) });
 const recentWindow = buildRecentAnalysisWindow({ samples, control, network, consoleEvents, errors, manifest });
 const commandTimeline = control.filter((item) => item.phase === "completed" || item.phase === "failed").map((item) => ({ ts: item.ts, phase: item.phase, commandId: item.commandId, type: item.type, input: item.input, afterUrl: item.afterUrl }));
@@ -116,6 +118,8 @@ const report = {
  runtimeAlerts,
  runnerErrors,
  commandFailures,
+  commandState,
+  toolFindings,
  findings,
  windows: { recent: recentWindow },
  readIssues: jsonlReadIssues,
@@ -159,6 +163,8 @@ console.log(JSON.stringify({
  runtimeAlerts: recentWindow.runtimeAlerts.summary,
  runnerErrors,
  commandFailures: commandFailures.slice(-8),
+  commandState,
+  toolFindings: toolFindings.slice(0, 8).map((item) => ({ kind: item.id, severity: item.severity, count: item.count ?? null, summary: String(item.summary ?? "").slice(0, 180) })),
  httpErrorGroups: recentWindow.runtimeAlerts.networkHttpErrorsByPath.slice(0, 8).map((item) => ({
    method: item.method ?? null,
    status: item.status ?? null,
@@ -356,6 +362,135 @@ async function readJson(file) {
  try { return JSON.parse(await readFile(file, "utf8")); } catch { return null; }
 }

+async function readCommandState(rootDir) {
+  const buckets = {};
+  for (const bucket of ["pending", "processing", "done", "failed", "abandoned"]) {
+    buckets[bucket] = await readCommandBucket(path.join(rootDir, "commands", bucket), bucket);
+  }
+  return {
+    pendingCount: buckets.pending.count,
+    processingCount: buckets.processing.count,
+    doneCount: buckets.done.count,
+    failedCount: buckets.failed.count,
+    abandonedCount: buckets.abandoned.count,
+    pending: buckets.pending.items,
+    processing: buckets.processing.items,
+    failed: buckets.failed.items.slice(0, 12),
+    abandoned: buckets.abandoned.items.slice(0, 20),
+    summary: {
+      backlogCount: buckets.pending.count + buckets.processing.count,
+      oldestPendingAgeSeconds: buckets.pending.oldestAgeSeconds,
+      oldestProcessingAgeSeconds: buckets.processing.oldestAgeSeconds,
+      valuesRedacted: true
+    },
+    valuesRedacted: true
+  };
+}
+
+async function readCommandBucket(dir, bucket) {
+  let names = [];
+  try {
+    names = (await readdir(dir)).filter((name) => name.endsWith(".json")).sort();
+  } catch (error) {
+    if (!error || error.code !== "ENOENT") jsonlReadIssues.push({ file: path.basename(dir), kind: "command-dir-read-error", code: error && error.code ? String(error.code) : null, errorMessage: limitText(error && error.message ? error.message : String(error), 240) });
+    return { bucket, count: 0, oldestAgeSeconds: null, items: [] };
+  }
+  const items = [];
+  for (const name of names.slice(0, 200)) {
+    const file = path.join(dir, name);
+    const parsed = await readJson(file) || {};
+    let meta = null;
+    try { meta = await stat(file); } catch {}
+    const timestamp = parsed.createdAt || parsed.abandonedAt || parsed.completedAt || parsed.failedAt || (meta ? meta.mtime.toISOString() : null);
+    const timestampMs = Date.parse(String(timestamp || ""));
+    items.push({
+      bucket,
+      id: parsed.id || parsed.commandId || name.replace(/[.]json$/u, ""),
+      type: parsed.type || parsed.command?.type || null,
+      createdAt: parsed.createdAt || null,
+      completedAt: parsed.completedAt || null,
+      failedAt: parsed.failedAt || null,
+      abandonedAt: parsed.abandonedAt || null,
+      reason: parsed.reason || parsed.error?.message || null,
+      ageSeconds: Number.isFinite(timestampMs) ? Math.max(0, Math.round((Date.now() - timestampMs) / 1000)) : null,
+      mtime: meta ? meta.mtime.toISOString() : null,
+      valuesRedacted: true
+    });
+  }
+  const ages = items.map((item) => item.ageSeconds).filter((value) => Number.isFinite(value));
+  return { bucket, count: names.length, oldestAgeSeconds: ages.length > 0 ? Math.max(...ages) : null, items };
+}
+
+function buildToolFindings({ manifest, heartbeat, commandState }) {
+  const findings = [];
+  const diagnostics = heartbeatDiagnostics(manifest, heartbeat);
+  if (diagnostics.heartbeatStale) {
+    findings.push({
+      id: "tool-runner-heartbeat-stale",
+      severity: "red",
+      summary: "web-probe observe runner heartbeat is stale; treat this as observer tooling failure, not Workbench behavior",
+      count: 1,
+      diagnostics,
+      valuesRedacted: true
+    });
+  }
+  if ((commandState?.pendingCount ?? 0) > 0 || (commandState?.processingCount ?? 0) > 0) {
+    findings.push({
+      id: "tool-pending-commands-unconsumed",
+      severity: "red",
+      summary: "web-probe observe has pending/processing control commands that were not consumed by the runner",
+      count: (commandState?.pendingCount ?? 0) + (commandState?.processingCount ?? 0),
+      pending: (commandState?.pending ?? []).slice(0, 12),
+      processing: (commandState?.processing ?? []).slice(0, 12),
+      valuesRedacted: true
+    });
+  }
+  if ((commandState?.abandonedCount ?? 0) > 0) {
+    findings.push({
+      id: "tool-commands-abandoned",
+      severity: "info",
+      summary: "web-probe observe force-stop abandoned queued commands; do not interpret these as Workbench command failures",
+      count: commandState.abandonedCount,
+      commands: (commandState.abandoned ?? []).slice(0, 20),
+      valuesRedacted: true
+    });
+  }
+  if (heartbeat?.forceStop || manifest?.forceStop) {
+    findings.push({
+      id: "tool-runner-force-stopped",
+      severity: "info",
+      summary: "web-probe observe runner was stopped by the CLI outside the command queue",
+      count: 1,
+      forceStop: heartbeat?.forceStop || manifest?.forceStop,
+      valuesRedacted: true
+    });
+  }
+  return findings;
+}
+
+function heartbeatDiagnostics(manifest, heartbeat) {
+  const status = String(heartbeat?.status || manifest?.status || "");
+  const terminal = /^(completed|failed|force-stopped|stopped|abandoned)$/u.test(status);
+  const sampleIntervalMs = Number(manifest?.sampling?.sampleIntervalMs) || 5000;
+  const staleAfterMs = Math.max(60000, sampleIntervalMs * 3);
+  const updatedAt = heartbeat?.updatedAt || heartbeat?.lastSampleAt || null;
+  const updatedMs = Date.parse(String(updatedAt || ""));
+  const ageSeconds = Number.isFinite(updatedMs) ? Math.max(0, Math.round((Date.now() - updatedMs) / 1000)) : null;
+  const heartbeatStale = !terminal && (!Number.isFinite(updatedMs) || Date.now() - updatedMs > staleAfterMs);
+  return {
+    status: status || null,
+    terminal,
+    updatedAt,
+    heartbeatAgeSeconds: ageSeconds,
+    heartbeatStale,
+    heartbeatStaleAfterSeconds: Math.round(staleAfterMs / 1000),
+    sampleSeq: heartbeat?.sampleSeq ?? null,
+    commandSeq: heartbeat?.commandSeq ?? null,
+    activeCommandId: heartbeat?.activeCommandId ?? null,
+    valuesRedacted: true
+  };
+}
+
 function safeArchivePrefix(value) {
  const text = String(value || "").trim();
  if (!text) return "";