Merge pull request #1798 from pikasTech/fix/1796-otel-warm-terminal

修正 warm-session 后续轮次的 OTel 终态诊断

scripts/src/platform-infra-observability/apply-status-scripts.ts

@@ -164,6 +164,8 @@ export function compactAgentRunAuthority(value: unknown): Record<string, unknown
   const result: Record<string, unknown> = {
     queried: authority.queried ?? null,
     namespace: authority.namespace ?? null,
+    namespaceSource: authority.namespaceSource ?? null,
+    providerDecisionNamespace: authority.providerDecisionNamespace ?? null,
     ok: authority.ok ?? null,
     commandOk: authority.commandOk ?? null,
     resultOk: authority.resultOk ?? null,
@@ -179,6 +181,7 @@ export function compactAgentRunAuthority(value: unknown): Record<string, unknown
     failureKind: authority.failureKind ?? null,
     latestSeq: authority.latestSeq ?? null,
     eventCount: authority.eventCount ?? null,
+    scopedEventCount: authority.scopedEventCount ?? null,
     fallback: authority.fallback ?? null,
     error: authority.error ?? null,
     warnings: limitArray(authority.warnings, 3),

scripts/src/platform-infra-observability/diagnose-code-agent-script.test.ts

@@ -0,0 +1,215 @@
+import { afterEach, describe, expect, test } from "bun:test";
+import { chmodSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { spawnSync } from "node:child_process";
+import { compactDiagnoseCodeAgentResult } from "./apply-status-scripts";
+import { diagnoseCodeAgentScript } from "./diagnose-code-agent-script";
+import { renderDiagnoseCodeAgentTable } from "./render";
+import type { DiagnoseCodeAgentOptions, ObservabilityConfig, ObservabilityTarget } from "./types";
+
+const temporaryDirectories: string[] = [];
+
+afterEach(() => {
+  for (const directory of temporaryDirectories.splice(0)) rmSync(directory, { force: true, recursive: true });
+});
+
+describe("diagnose-code-agent warm-session terminal classification", () => {
+  test("uses the provider-decision lane and scopes run events by commandId", () => {
+    const diagnosis = runFixture({ authorityAvailable: true, terminalSpan: false });
+    expect(diagnosis.status).toBe(0);
+    expect(diagnosis.payload.agentrun).toMatchObject({
+      terminalStatus: "completed",
+      terminalSource: "agentrun-authority",
+      runnerProviderClassification: "completed",
+      authority: {
+        namespace: "agentrun-v02",
+        namespaceSource: "provider-decision",
+        providerDecisionNamespace: "agentrun-v02",
+        ok: true,
+        terminalStatus: "completed",
+        terminalEventSeq: 10,
+        eventCount: 2,
+        scopedEventCount: 1,
+      },
+    });
+    expect(diagnosis.payload.observabilityGap).toMatchObject({
+      status: "missing-command-terminal-span",
+      complete: false,
+      serviceCoverageComplete: true,
+      terminalSpanSeen: false,
+      authorityQuery: { namespace: "agentrun-v02", namespaceSource: "provider-decision", ok: true },
+    });
+    expect(diagnosis.calls).toContain("-n agentrun-v02 exec deploy/agentrun-mgr");
+    expect(diagnosis.calls).not.toContain("agentrun-v01");
+
+    const compact = compactDiagnoseCodeAgentResult(diagnosis.payload);
+    expect((compact.agentrun as Record<string, any>).authority).toMatchObject({
+      namespace: "agentrun-v02",
+      namespaceSource: "provider-decision",
+      scopedEventCount: 1,
+    });
+    const rendered = renderDiagnoseCodeAgentTable({
+      ok: true,
+      target,
+      options,
+      query: {},
+      result: compact,
+    });
+    expect(rendered.renderedText).toContain("authorityQuery namespace=agentrun-v02 source=provider-decision ok=true commandState=completed terminal=completed");
+  });
+
+  test("reports unknown instead of running when command authority and terminal span are both absent", () => {
+    const diagnosis = runFixture({ authorityAvailable: false, terminalSpan: false });
+    expect(diagnosis.status).toBe(0);
+    expect(diagnosis.payload.agentrun).toMatchObject({
+      terminalStatus: null,
+      runnerProviderClassification: "unknown",
+      authority: { namespace: "agentrun-v02", namespaceSource: "provider-decision", ok: false },
+    });
+    expect(diagnosis.payload.observabilityGap).toMatchObject({
+      status: "authority-query-failed",
+      complete: false,
+      terminalSpanSeen: false,
+    });
+    expect(diagnosis.payload.rootCauseCandidates[0].code).toBe("observability_gap_authority_query_failed");
+  });
+
+  test("keeps a scoped OTel terminal span complete even if the supplemental authority read fails", () => {
+    const diagnosis = runFixture({ authorityAvailable: false, terminalSpan: true });
+    expect(diagnosis.status).toBe(0);
+    expect(diagnosis.payload.agentrun).toMatchObject({
+      terminalStatus: "completed",
+      terminalSource: "otel-span",
+      runnerProviderClassification: "completed",
+    });
+    expect(diagnosis.payload.observabilityGap).toMatchObject({
+      status: "complete",
+      complete: true,
+      terminalSpanSeen: true,
+      terminalSpanCount: 1,
+    });
+  });
+});
+
+function runFixture(input: { authorityAvailable: boolean; terminalSpan: boolean }): {
+  status: number | null;
+  payload: Record<string, any>;
+  calls: string;
+} {
+  const directory = mkdtempSync(join(tmpdir(), "unidesk-otel-diagnose-"));
+  temporaryDirectories.push(directory);
+  writeFileSync(join(directory, "trace.json"), JSON.stringify(traceFixture(input.terminalSpan)));
+  writeFileSync(join(directory, "command.json"), JSON.stringify({ id: "cmd_second", state: "completed" }));
+  writeFileSync(join(directory, "result.json"), JSON.stringify({ ok: true, commandId: "cmd_second", status: "completed", commandState: "completed", terminalStatus: "completed" }));
+  writeFileSync(join(directory, "events.json"), JSON.stringify({
+    items: [
+      { seq: 10, type: "terminal_status", payload: { commandId: "cmd_second", terminalStatus: "completed" } },
+      { seq: 11, type: "terminal_status", payload: { commandId: "cmd_first", terminalStatus: "failed", failureKind: "backend-timeout" } },
+    ],
+  }));
+  const kubectl = join(directory, "kubectl");
+  writeFileSync(kubectl, `#!/bin/sh
+set -u
+echo "$*" >>"$FIXTURE_DIR/calls.log"
+case "$*" in
+  *"/api/traces/"*) cat "$FIXTURE_DIR/trace.json"; exit 0 ;;
+esac
+if [ "$AUTHORITY_AVAILABLE" != "1" ]; then
+  echo "fixture authority unavailable" >&2
+  exit 1
+fi
+case "$*" in
+  *"-n agentrun-v02 exec deploy/agentrun-mgr"*"/commands/cmd_second/result"*) cat "$FIXTURE_DIR/result.json"; exit 0 ;;
+  *"-n agentrun-v02 exec deploy/agentrun-mgr"*"/commands/cmd_second"*) cat "$FIXTURE_DIR/command.json"; exit 0 ;;
+  *"-n agentrun-v02 exec deploy/agentrun-mgr"*"/runs/run_warm/events"*) cat "$FIXTURE_DIR/events.json"; exit 0 ;;
+esac
+echo "unexpected kubectl call: $*" >&2
+exit 1
+`);
+  chmodSync(kubectl, 0o755);
+  const script = diagnoseCodeAgentScript(observability, target, null, options);
+  const result = spawnSync("bash", ["-c", script], {
+    encoding: "utf8",
+    env: {
+      ...process.env,
+      AUTHORITY_AVAILABLE: input.authorityAvailable ? "1" : "0",
+      FIXTURE_DIR: directory,
+      PATH: `${directory}:${process.env.PATH ?? ""}`,
+    },
+    maxBuffer: 4 * 1024 * 1024,
+  });
+  expect(result.stdout).not.toBe("");
+  return {
+    status: result.status,
+    payload: JSON.parse(result.stdout),
+    calls: readFileSync(join(directory, "calls.log"), "utf8"),
+  };
+}
+
+function traceFixture(terminalSpan: boolean): Record<string, unknown> {
+  const common = [
+    attribute("traceId", "trc_warm_second"),
+    attribute("runId", "run_warm"),
+    attribute("commandId", "cmd_second"),
+    attribute("sessionId", "ses_warm"),
+  ];
+  return {
+    batches: [
+      batch("hwlab-cloud-api", [span("provider_decision", [
+        ...common,
+        attribute("agent.chat.provider_profile", "gpt.pika"),
+        attribute("agentRunRunnerNamespace", "agentrun-v02"),
+      ])]),
+      batch("agentrun-manager", [
+        span("command_created", common),
+        ...(terminalSpan ? [span("runner_terminal.completed", [...common, attribute("terminalStatus", "completed"), attribute("eventType", "terminal_status")])] : []),
+      ]),
+      batch("agentrun-runner", [span("codex_stdio.notification", common)]),
+    ],
+  };
+}
+
+function batch(service: string, spans: Array<Record<string, unknown>>): Record<string, unknown> {
+  return {
+    resource: { attributes: [attribute("service.name", service)] },
+    scopeSpans: [{ scope: { name: `${service}.fixture` }, spans }],
+  };
+}
+
+function span(name: string, attributes: Array<Record<string, unknown>>): Record<string, unknown> {
+  return { name, startTimeUnixNano: "1000000", endTimeUnixNano: "2000000", attributes };
+}
+
+function attribute(key: string, value: string): Record<string, unknown> {
+  return { key, value: { stringValue: value } };
+}
+
+const target: ObservabilityTarget = {
+  id: "NC01",
+  route: "NC01:k3s",
+  namespace: "devops-infra",
+  role: "active",
+  enabled: true,
+  createNamespace: false,
+};
+
+const options: DiagnoseCodeAgentOptions = {
+  targetId: "NC01",
+  businessTraceId: "trc_warm_second",
+  traceId: "de11d47d1bab0ca4d3f6259279f3a332",
+  runId: null,
+  commandId: null,
+  sessionId: null,
+  runnerJobId: null,
+  full: false,
+  raw: false,
+  limit: 40,
+  candidateLimit: 20,
+  lookbackMinutes: 60,
+};
+
+const observability = {
+  traceBackend: { serviceName: "tempo" },
+  probes: { traceQueryPathTemplate: "/api/traces/{{traceId}}" },
+} as ObservabilityConfig;

scripts/src/platform-infra-observability/diagnose-code-agent-script.ts

@@ -1057,16 +1057,19 @@ def unwrap_payload(value):
         return result
     return value
 
-def guess_agentrun_namespace(identity):
+def guess_agentrun_namespace(identity, provider_decision):
+    runner_namespace = provider_decision.get("runnerNamespace") if isinstance(provider_decision, dict) else None
+    if isinstance(runner_namespace, str) and re.fullmatch(r"agentrun-v[0-9]+", runner_namespace.strip()):
+        return runner_namespace.strip(), "provider-decision"
     if isinstance(identity, dict):
         for key in ("jobName", "podName", "runnerJobId"):
             text = str(identity.get(key) or "")
             match = re.search(r"\\bagentrun-v[0-9]+\\b", text)
             if match:
-                return match.group(0)
+                return match.group(0), "workload-identity"
     if str(TARGET_ID).upper() == "D601":
-        return "agentrun-v02"
-    return "agentrun-v01"
+        return "agentrun-v02", "target-default"
+    return "agentrun-v01", "target-default"
 
 def agentrun_service_proxy_paths(namespace, api_path):
     return [
@@ -1131,9 +1134,22 @@ def event_items(value):
             return [item for item in items if isinstance(item, dict)]
     return []
 
-def terminal_from_events(events):
+def event_matches_command(item, command_id):
+    if not isinstance(item, dict) or command_id in (None, ""):
+        return False
+    payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
+    return str(first_present(
+        payload.get("commandId"),
+        payload.get("targetCommandId"),
+        item.get("commandId"),
+        item.get("targetCommandId"),
+    ) or "") == str(command_id)
+
+def terminal_from_events(events, command_id):
     latest = None
     for item in events:
+        if not event_matches_command(item, command_id):
+            continue
         payload = item.get("payload") if isinstance(item.get("payload"), dict) else {}
         event_type = item.get("type") or item.get("eventType") or payload.get("type")
         status = first_present(payload.get("terminalStatus"), payload.get("status"), item.get("terminalStatus"), item.get("status"))
@@ -1146,14 +1162,15 @@ def terminal_from_events(events):
             }
     return latest
 
-def classify_terminal_status(status, spans):
+def classify_terminal_status(status, authority):
     if status in ("completed", "succeeded", "success"):
         return "completed"
     if status in ("failed", "error", "timeout", "cancelled", "canceled"):
         return "failed"
     if status == "blocked":
         return "blocked"
-    if any(str(item.get("service") or "") == "agentrun-runner" for item in spans):
+    command_state = authority.get("commandState") if isinstance(authority, dict) and authority.get("ok") else None
+    if command_state in ("pending", "claimed", "running", "processing", "in_progress", "in-progress"):
         return "running"
     return "unknown"
 
@@ -1179,7 +1196,7 @@ def preferred_failure_kind(values):
             return kind
     return kinds[-1]
 
-def agentrun_authority_summary(identity):
+def agentrun_authority_summary(identity, provider_decision):
     run_id = identity.get("runId") if isinstance(identity, dict) else None
     command_id = identity.get("commandId") if isinstance(identity, dict) else None
     if not run_id or not command_id:
@@ -1187,7 +1204,7 @@ def agentrun_authority_summary(identity):
             "queried": False,
             "reason": "missing-run-or-command-id",
         }
-    namespace = guess_agentrun_namespace(identity)
+    namespace, namespace_source = guess_agentrun_namespace(identity, provider_decision)
     run_q = urllib.parse.quote(str(run_id), safe="")
     command_q = urllib.parse.quote(str(command_id), safe="")
     command_response, command_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s" % (run_q, command_q), timeout=8)
@@ -1196,7 +1213,8 @@ def agentrun_authority_summary(identity):
     command_payload = unwrap_payload(command_response.get("json"))
     result_payload = unwrap_payload(result_response.get("json"))
     events = event_items(events_response.get("json"))
-    terminal_event = terminal_from_events(events)
+    scoped_events = [item for item in events if event_matches_command(item, command_id)]
+    terminal_event = terminal_from_events(scoped_events, command_id)
     command_state = first_present(command_payload.get("state"), command_payload.get("status"))
     result_status = first_present(
         result_payload.get("terminalStatus"),
@@ -1205,13 +1223,15 @@ def agentrun_authority_summary(identity):
         result_payload.get("status"),
     )
     terminal_status = first_present(
-        terminal_event.get("status") if isinstance(terminal_event, dict) else None,
         result_status,
+        terminal_event.get("status") if isinstance(terminal_event, dict) else None,
         command_state if command_state in ("completed", "failed", "blocked", "cancelled", "canceled") else None,
     )
     return {
         "queried": True,
         "namespace": namespace,
+        "namespaceSource": namespace_source,
+        "providerDecisionNamespace": provider_decision.get("runnerNamespace") if isinstance(provider_decision, dict) else None,
         "ok": bool(command_response.get("ok") or result_response.get("ok") or events_response.get("ok")),
         "commandOk": command_response.get("ok"),
         "resultOk": result_response.get("ok"),
@@ -1230,6 +1250,7 @@ def agentrun_authority_summary(identity):
         "terminalEventType": terminal_event.get("type") if isinstance(terminal_event, dict) else None,
         "terminalEventSeq": terminal_event.get("seq") if isinstance(terminal_event, dict) else None,
         "eventCount": len(events),
+        "scopedEventCount": len(scoped_events),
         "attempts": {
             "command": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in command_attempts],
             "result": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in result_attempts],
@@ -1762,7 +1783,7 @@ def agentrun_summary(spans, authority=None):
             failure_kinds.append(authority.get("failureKind"))
         terminal_source = "agentrun-authority"
     failure_kind = preferred_failure_kind(failure_kinds)
-    runner_classification = classify_terminal_status(terminal_status, spans)
+    runner_classification = classify_terminal_status(terminal_status, authority)
     return {
         "terminalStatus": terminal_status,
         "failureKind": failure_kind,
@@ -2266,7 +2287,7 @@ http_summary = http_status_summary(spans)
 read_model = hwlab_read_model_summary(spans)
 identity = identity_from_spans(spans)
 provider_decision = provider_decision_summary(spans)
-agentrun_authority = agentrun_authority_summary(identity)
+agentrun_authority = agentrun_authority_summary(identity, provider_decision)
 agentrun = agentrun_summary(spans, agentrun_authority)
 lag = projection_lag_summary(agentrun, read_model)
 candidates = root_cause_candidates(http_summary, agentrun, read_model, lag, error_spans, idle_warning_spans)
@@ -2277,12 +2298,37 @@ service_path = {
 }
 missing_services = [service for service in expected_services if service not in services]
 service_path["complete"] = len(missing_services) == 0
+terminal_status = agentrun.get("terminalStatus")
+terminal_span_count = len(agentrun.get("terminalSpans", []))
+authority_queried = bool(agentrun_authority.get("queried")) if isinstance(agentrun_authority, dict) else False
+authority_ok = bool(agentrun_authority.get("ok")) if isinstance(agentrun_authority, dict) else False
+authority_query_failed = authority_queried and not authority_ok
+if missing_services:
+    observability_gap_status = "missing-service-spans"
+elif terminal_status not in (None, "") and terminal_span_count == 0:
+    observability_gap_status = "missing-command-terminal-span"
+elif terminal_span_count == 0 and authority_query_failed:
+    observability_gap_status = "authority-query-failed"
+elif terminal_span_count == 0 and agentrun.get("runnerProviderClassification") == "unknown":
+    observability_gap_status = "unknown-command-terminal-state"
+else:
+    observability_gap_status = "complete"
 observability_gap = {
-    "status": "complete" if len(missing_services) == 0 else "missing-service-spans",
+    "status": observability_gap_status,
     "expectedServices": expected_services,
     "seenServices": sorted(services),
     "missingServices": missing_services,
-    "complete": len(missing_services) == 0,
+    "serviceCoverageComplete": len(missing_services) == 0,
+    "terminalSpanSeen": terminal_span_count > 0,
+    "terminalSpanCount": terminal_span_count,
+    "authorityQuery": {
+        "queried": authority_queried,
+        "ok": authority_ok,
+        "namespace": agentrun_authority.get("namespace") if isinstance(agentrun_authority, dict) else None,
+        "namespaceSource": agentrun_authority.get("namespaceSource") if isinstance(agentrun_authority, dict) else None,
+        "commandState": agentrun_authority.get("commandState") if isinstance(agentrun_authority, dict) else None,
+    },
+    "complete": observability_gap_status == "complete",
 }
 if missing_services and ("hwlab-cloud-api" in services or identity.get("runId") not in (None, "") or identity.get("commandId") not in (None, "")):
     candidates.insert(0, {
@@ -2292,14 +2338,43 @@ if missing_services and ("hwlab-cloud-api" in services or identity.get("runId")
         "summary": "The business trace is correlated to Code Agent context but is missing expected service spans; do not interpret the missing manager/runner spans as proof that those services were not involved.",
         "evidence": observability_gap,
     })
+elif observability_gap_status == "missing-command-terminal-span":
+    candidates.insert(0, {
+        "code": "observability_gap_missing_command_terminal_span",
+        "label": "command terminal span missing",
+        "confidence": 0.92,
+        "summary": "The command-specific AgentRun authority is terminal, but the scoped business trace has no command terminal span; this is an instrumentation gap, not an active runner.",
+        "evidence": observability_gap,
+    })
+elif observability_gap_status == "authority-query-failed":
+    candidates.insert(0, {
+        "code": "observability_gap_authority_query_failed",
+        "label": "AgentRun authority query failed",
+        "confidence": 0.9,
+        "summary": "The scoped business trace has no command terminal span and the AgentRun authority query failed; terminal state is unknown and must not be reported as running.",
+        "evidence": observability_gap,
+    })
+elif observability_gap_status == "unknown-command-terminal-state":
+    candidates.insert(0, {
+        "code": "observability_gap_unknown_command_terminal_state",
+        "label": "command terminal state unknown",
+        "confidence": 0.82,
+        "summary": "The scoped business trace has no command terminal span and command authority has no active or terminal state; classify it as unknown rather than running.",
+        "evidence": observability_gap,
+    })
 facts = []
-if missing_services:
+if observability_gap_status == "missing-service-spans":
     facts.append("observability gap: missing service spans " + ",".join(missing_services))
+elif observability_gap_status == "missing-command-terminal-span":
+    facts.append("observability gap: missing command terminal span")
+elif observability_gap_status == "authority-query-failed":
+    facts.append("observability gap: AgentRun authority query failed")
+elif observability_gap_status == "unknown-command-terminal-state":
+    facts.append("observability gap: command terminal state unknown")
 if business_scope.get("mode") == "business-trace-scoped" and business_scope.get("crossBusinessTraceIds"):
     facts.append("cross-business-trace spans excluded")
 if http_summary.get("actorForbidden"):
     facts.append("actor forbidden")
-terminal_status = agentrun.get("terminalStatus")
 read_model_turn_status = read_model.get("turnStatus")
 agentrun_read_model_status_conflict = terminal_status_conflicts(terminal_status, read_model_turn_status)
 if agentrun_read_model_status_conflict:

scripts/src/platform-infra-observability/render.ts

@@ -716,6 +716,7 @@ export function renderDiagnoseCodeAgentTable(input: {
   const mapping = asPlainRecord(input.result.mapping);
   const identity = asPlainRecord(input.result.identity);
   const agentrun = asPlainRecord(input.result.agentrun);
+  const agentrunAuthority = asPlainRecord(agentrun?.authority);
   const projectionLag = asPlainRecord(input.result.projectionLag);
   const summary = asPlainRecord(input.result.summary);
   const evidence = asPlainRecord(input.result.evidence);
@@ -786,6 +787,7 @@ export function renderDiagnoseCodeAgentTable(input: {
     "Summary:",
     `  target=${input.target.id} spanCount=${textValue(input.result.spanCount)} servicePath=${joinValues(input.result.servicePath, 60)}`,
     `  agentrun=${textValue(agentrun?.terminalStatus)} failureKind=${textValue(agentrun?.failureKind)} source=${textValue(agentrun?.terminalSource)} authority=${shortenEnd(JSON.stringify((asPlainRecord(agentrun?.authority) ?? {})), 120)}`,
+    `  authorityQuery namespace=${textValue(agentrunAuthority?.namespace)} source=${textValue(agentrunAuthority?.namespaceSource)} ok=${textValue(agentrunAuthority?.ok)} commandState=${textValue(agentrunAuthority?.commandState)} terminal=${textValue(agentrunAuthority?.terminalStatus)}`,
     `  readModel=${shortenEnd(JSON.stringify(input.result.hwlabReadModel ?? null), 140)}`,
   ];
   const next = asPlainRecord(input.result.next);