fix: improve workbench triad diagnostics

2026-07-01 16:15:50 +00:00
parent 113b6809d1
commit 624736f336
4 changed files with 385 additions and 33 deletions
@@ -1450,6 +1450,88 @@ def identity_from_spans(spans):
            identity[key] = preferred if preferred is not None else fallback
    return identity

+def span_business_trace_id(item):
+    attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
+    for key in ("traceId", "workbench.trace_id", "workbench.turn_id"):
+        value = attrs.get(key)
+        if isinstance(value, str) and value.strip().startswith("trc_"):
+            return value.strip()
+    return None
+
+def span_identity_value(item, key):
+    attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
+    value = attrs.get(key)
+    if value in (None, ""):
+        return None
+    return str(value)
+
+def span_matches_scope_identity(item, identity):
+    if not isinstance(identity, dict):
+        return False
+    # Only strong per-turn identities are safe enough for no-trace-id spans.
+    # runId/sessionId/runnerJobId can cover several turns in the same Workbench
+    # session and caused terminal status leakage across business trace ids.
+    for key in ("commandId", "turnId"):
+        scoped = identity.get(key)
+        value = span_identity_value(item, key)
+        if scoped not in (None, "") and value not in (None, ""):
+            return str(scoped) == str(value)
+    return False
+
+def scoped_spans_for_business_trace(spans, business_trace_id):
+    if business_trace_id in (None, ""):
+        return list(spans), {
+            "mode": "unscoped-no-business-trace-id",
+            "requestedBusinessTraceId": None,
+            "totalSpanCount": len(spans),
+            "scopedSpanCount": len(spans),
+            "excludedDifferentTraceSpanCount": 0,
+            "includedNoTraceIdentitySpanCount": 0,
+            "crossBusinessTraceIds": [],
+        }
+    exact = []
+    no_trace = []
+    cross_ids = collections.Counter()
+    for item in spans:
+        trace_id = span_business_trace_id(item)
+        if trace_id == business_trace_id:
+            exact.append(item)
+        elif trace_id:
+            cross_ids[trace_id] += 1
+        else:
+            no_trace.append(item)
+    if not exact:
+        return list(spans), {
+            "mode": "unscoped-no-exact-business-trace-span",
+            "requestedBusinessTraceId": business_trace_id,
+            "totalSpanCount": len(spans),
+            "scopedSpanCount": len(spans),
+            "excludedDifferentTraceSpanCount": 0,
+            "includedNoTraceIdentitySpanCount": 0,
+            "crossBusinessTraceIds": [{"traceId": key, "count": count} for key, count in cross_ids.most_common(8)],
+        }
+    identity = identity_from_spans(exact)
+    exact_ids = set(id(item) for item in exact)
+    scoped = list(exact)
+    included_no_trace = 0
+    for item in no_trace:
+        if id(item) in exact_ids:
+            continue
+        if span_matches_scope_identity(item, identity):
+            scoped.append(item)
+            included_no_trace += 1
+    scoped.sort(key=lambda item: (item.get("_start", 0), item.get("_index", 0)))
+    return scoped, {
+        "mode": "business-trace-scoped",
+        "requestedBusinessTraceId": business_trace_id,
+        "totalSpanCount": len(spans),
+        "scopedSpanCount": len(scoped),
+        "excludedDifferentTraceSpanCount": len(spans) - len(scoped),
+        "includedNoTraceIdentitySpanCount": included_no_trace,
+        "scopeIdentity": {key: identity.get(key) for key in ("commandId", "turnId", "runId", "sessionId") if identity.get(key) not in (None, "")},
+        "crossBusinessTraceIds": [{"traceId": key, "count": count} for key, count in cross_ids.most_common(8)],
+    }
+
 def flatten_trace(trace_body):
    services = set()
    business_trace_ids = set()
@@ -1752,6 +1834,20 @@ def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error
            "evidence": lag_summary.get("reasons", []),
        })
    failure_kind = str(agentrun.get("failureKind") or "")
+    read_model_turn_status = read_model.get("turnStatus")
+    status_conflict = terminal_status_conflicts(agentrun.get("terminalStatus"), read_model_turn_status)
+    if status_conflict:
+        candidates.append({
+            "code": "otel_business_trace_terminal_conflict",
+            "label": "OTel terminal conflict",
+            "confidence": 0.86,
+            "summary": "Scoped AgentRun terminal status conflicts with HWLAB read-model turn status for the requested business trace; prefer the HWLAB read-model status for Workbench RCA and inspect OTel correlation leakage.",
+            "evidence": {
+                "agentrunTerminalStatus": agentrun.get("terminalStatus"),
+                "hwlabReadModelTurnStatus": read_model_turn_status,
+                "turnStatusCounts": read_model.get("turnStatusCounts"),
+            },
+        })
    if failure_kind == "provider-auth-failed":
        candidates.append({
            "code": "provider_auth_failed",
@@ -1765,7 +1861,7 @@ def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error
                "terminalCategory": agentrun.get("terminalCategory"),
            },
        })
-    if agentrun.get("terminalStatus") in ("failed", "error", "timeout", "blocked", "cancelled"):
+    if agentrun.get("terminalStatus") in ("failed", "error", "timeout", "blocked", "cancelled") and not status_conflict:
        candidates.append({
            "code": "agentrun_terminal_failed",
            "label": "AgentRun terminal failed",
@@ -1790,6 +1886,19 @@ def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error
                "runnerProviderClassification": agentrun.get("runnerProviderClassification"),
            },
        })
+    if agentrun.get("terminalStatus") in (None, "") and read_model_turn_status in ("completed", "failed", "error", "timeout", "blocked", "cancelled", "canceled"):
+        candidates.append({
+            "code": "hwlab_read_model_terminal_without_runner_spans",
+            "label": "HWLAB read model terminal",
+            "confidence": 0.7,
+            "summary": "HWLAB read model shows a terminal turn status but scoped AgentRun manager/runner terminal spans are absent; treat missing runner spans as an observability gap, not proof that the turn is still running.",
+            "evidence": {
+                "turnStatus": read_model_turn_status,
+                "turnStatusCounts": read_model.get("turnStatusCounts"),
+                "sessionListReadCount": read_model.get("sessionListReadCount"),
+                "sessionMessageReadCount": read_model.get("sessionMessageReadCount"),
+            },
+        })
    if error_spans and agentrun.get("terminalStatus") == "completed":
        candidates.append({
            "code": "tool_call_failures_recovered",
@@ -1812,6 +1921,28 @@ def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error
    candidates.sort(key=lambda item: item.get("confidence", 0), reverse=True)
    return candidates

+def comparable_terminal_status(value):
+    text = str(value or "").strip().lower().replace("_", "-")
+    if not text:
+        return None
+    if text == "cancelled":
+        return "canceled"
+    if text == "success" or text == "succeeded" or text == "complete":
+        return "completed"
+    if text == "error" or text == "failure":
+        return "failed"
+    return text
+
+def terminal_status_conflicts(left, right):
+    left_status = comparable_terminal_status(left)
+    right_status = comparable_terminal_status(right)
+    if left_status in (None, "") or right_status in (None, ""):
+        return False
+    terminal_values = {"completed", "failed", "timeout", "blocked", "canceled", "stale", "thread-resume-failed", "interrupted", "expired"}
+    if left_status not in terminal_values or right_status not in terminal_values:
+        return False
+    return left_status != right_status
+
 def candidate_score(trace_id, meta, trace_body, trace_rc, trace_err):
    parsed = parse_json(trace_body)
    meta_root_service = str(meta.get("rootServiceName") or "") if isinstance(meta, dict) else ""
@@ -1837,13 +1968,15 @@ def candidate_score(trace_id, meta, trace_body, trace_rc, trace_err):
            "stderrTail": (trace_err or "")[-1000:],
        }, None
    flat = flatten_trace(parsed)
-    spans = flat["spans"]
-    services = set(flat["services"])
+    all_spans = flat["spans"]
+    spans, business_scope = scoped_spans_for_business_trace(all_spans, BUSINESS_TRACE_ID)
+    services = set(str(item.get("service") or "") for item in spans)
    names = [str(item.get("name") or "") for item in spans]
    lowered_names = [name.lower() for name in names]
    identity = identity_from_spans(spans)
    agentrun = agentrun_summary(spans)
-    error_spans = flat["errorSpans"]
+    scoped_span_ids = set(id(item) for item in spans)
+    error_spans = [item for item in flat["errorSpans"] if id(item) in scoped_span_ids]
    score = 0
    reasons = []
    def add(points, reason):
@@ -1902,6 +2035,7 @@ def candidate_score(trace_id, meta, trace_body, trace_rc, trace_err):
        "identity": {key: identity.get(key) for key in ("runId", "commandId", "sessionId", "runnerJobId", "runnerId", "backendProfile") if identity.get(key) not in (None, "")},
        "terminalStatus": agentrun.get("terminalStatus"),
        "errorSpanCount": len(error_spans),
+        "businessTraceScope": business_scope,
        "candidateQuality": candidate_quality,
        "lowConfidence": candidate_quality != "normal",
        "rootTraceName": meta_root_name or None,
@@ -2043,10 +2177,12 @@ if not isinstance(trace_parsed, dict):
    raise SystemExit(1)

 flat = flatten_trace(trace_parsed)
-spans = flat["spans"]
-services = flat["services"]
+all_spans = flat["spans"]
+spans, business_scope = scoped_spans_for_business_trace(all_spans, BUSINESS_TRACE_ID)
+services = sorted(set(str(item.get("service") or "") for item in spans))
 business_trace_ids = flat["businessTraceIds"]
-error_spans = flat["errorSpans"]
+scoped_span_ids = set(id(item) for item in spans)
+error_spans = [item for item in flat["errorSpans"] if id(item) in scoped_span_ids]
 idle_warning_spans = [item for item in spans if str(item.get("name") or "") == "codex_stdio.idle_warning"]
 http_summary = http_status_summary(spans)
 read_model = hwlab_read_model_summary(spans)
@@ -2080,17 +2216,25 @@ if missing_services and ("hwlab-cloud-api" in services or identity.get("runId")
 facts = []
 if missing_services:
    facts.append("observability gap: missing service spans " + ",".join(missing_services))
+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")
-if terminal_status in ("failed", "error", "timeout", "blocked", "cancelled"):
+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:
+    facts.append("OTel AgentRun terminal status conflicts with HWLAB read model")
+if terminal_status in ("failed", "error", "timeout", "blocked", "cancelled") and not agentrun_read_model_status_conflict:
    failure_kind = agentrun.get("failureKind")
    if failure_kind:
        facts.append(f"AgentRun terminal failed ({failure_kind})")
    else:
        facts.append("AgentRun terminal failed")
-if terminal_status == "completed":
+if terminal_status == "completed" and not agentrun_read_model_status_conflict:
    facts.append("AgentRun completed")
+if (terminal_status in (None, "") or agentrun_read_model_status_conflict) and read_model_turn_status in ("completed", "failed", "error", "timeout", "blocked", "cancelled", "canceled"):
+    facts.append("HWLAB read model terminal " + str(read_model_turn_status))
 if lag.get("status") in ("confirmed", "suspected"):
    facts.append("projection/read-model stale")
 if idle_warning_spans and terminal_status in (None, ""):
@@ -2105,8 +2249,11 @@ summary = {
        "runnerProvider": agentrun.get("runnerProviderClassification"),
        "actorForbidden": http_summary.get("actorForbidden"),
        "terminalStatus": terminal_status,
+        "readModelTurnStatus": read_model_turn_status,
+        "agentrunReadModelStatusConflict": agentrun_read_model_status_conflict,
        "failureKind": agentrun.get("failureKind"),
        "observabilityGap": observability_gap.get("status"),
+        "businessTraceScope": business_scope.get("mode"),
    },
 }
 evidence = {
@@ -2136,6 +2283,7 @@ payload = {
    "servicePath": service_path,
    "observabilityGap": observability_gap,
    "businessTraceIds": business_trace_ids[:20],
+    "businessTraceScope": business_scope,
    "identity": identity,
    "agentrun": {
        "terminalStatus": agentrun.get("terminalStatus"),