Distinguish serving traffic from latest-ready identity

docs/digest_candidates_producer.zh-CN.md

@@ -129,6 +129,8 @@ publisher 遇到既有身份失配时保留首行 `skipped:runtime_target_mismat
 
 Cloud Run 查询 region 明确沿用现有平台 `us-central1`，不从报告或公开 synthetic manifest 推断；独立调用时缺少/非法 `CLOUD_RUN_REGION` 则不查询服务、metadata unavailable。即使报告与当前服务相等，也不能据此更新期望 revision 或批准身份。该诊断不公开配置原值、账户/目标/service、URI、revision 值、金额、异常内容或 gcloud stderr；最新报告不是所有历史报告的覆盖证明，镜像/安装/config/risk 指纹未取得时继续 UNKNOWN。
 
+`serving_traffic` 另行只读投影同一次 metadata 的 `status.traffic`：`status=single/split/absent/invalid`，附 `report_has_positive_traffic`、`configured_has_positive_traffic`、`report_is_only_serving_revision` 三项 nullable 布尔。只有完整 `revisionName`、明确的整数 percent（0–100）及总和 100 的分布可比较；同一 revision 的分量汇总，零流量 tag 不计 serving。缺失/空列表为 absent，畸形或非完整总和为 invalid，缺比较身份保留 null；不根据 latestReady、tag URL 或缺 percent 推断流量。该新增诊断不改变原 projection guard、status/reason 或 exit 门：exit 0 仍不证明报告是唯一 serving revision，也不批准轮换受保护身份；ready 相等、流量事实和独立批准分开验收。
+
 ## 如何注入 QRS（人工，不自动改生产 Environment）
 
 中央文档：QuantRuntimeSettings `docs/digest-candidates-wiring.zh-CN.md`（PR #586）。

scripts/diagnose_account_facts_identity.py

@@ -33,6 +33,46 @@
 )
 
 
+def _traffic_result(status: str = "absent") -> dict[str, Any]:
+    return {
+        "status": status,
+        "report_has_positive_traffic": None,
+        "configured_has_positive_traffic": None,
+        "report_is_only_serving_revision": None,
+    }
+
+
+def _serving_traffic(rows: object, *, report_revision: object, configured_revision: object) -> dict[str, Any]:
+    """Project effective traffic only; latestReady and tagged URLs are not fallbacks."""
+    if rows is None or isinstance(rows, list) and not rows:
+        return _traffic_result()
+    if not isinstance(rows, list):
+        return _traffic_result("invalid")
+    total = 0
+    positive: set[str] = set()
+    for row in rows:
+        if not isinstance(row, Mapping):
+            return _traffic_result("invalid")
+        revision = publisher._single_line_text(row.get("revisionName"))
+        percent = row.get("percent")
+        if not revision or type(percent) is not int or not 0 <= percent <= 100:
+            return _traffic_result("invalid")
+        total += percent
+        if percent > 0:
+            positive.add(revision)
+    if total != 100:
+        return _traffic_result("invalid")
+    result = _traffic_result("single" if len(positive) == 1 else "split")
+    report = publisher._single_line_text(report_revision)
+    configured = publisher._single_line_text(configured_revision)
+    if report:
+        result["report_has_positive_traffic"] = report in positive
+        result["report_is_only_serving_revision"] = positive == {report}
+    if configured:
+        result["configured_has_positive_traffic"] = configured in positive
+    return result
+
+
 def _result(*, source_sha: object = None, reason: str, status: str = "unavailable") -> dict[str, Any]:
     return {
         "schema_version": SCHEMA,
@@ -44,6 +84,7 @@ def _result(*, source_sha: object = None, reason: str, status: str = "unavailabl
         "freshness_status": "unknown",
         "cloud_run_metadata_status": "unavailable",
         "revision_equalities": dict.fromkeys(_EQUALITY_FIELDS),
+        "serving_traffic": _traffic_result(),
     }
 
 
@@ -136,12 +177,15 @@ def diagnose_identity(
             or _equal(metadata.get("name"), expected["expected_service_name"]) is not True
         ):
             result["cloud_run_metadata_status"] = "invalid"
+            result["serving_traffic"] = _traffic_result("invalid")
         else:
             result["cloud_run_metadata_status"] = "read"
             ready = status.get("latestReadyRevisionName")
             equalities["configured_equals_ready"] = _equal(expected["expected_runtime_revision"], ready)
             equalities["report_equals_ready"] = _equal(revision, ready)
             equalities["ready_equals_latest_created"] = _equal(ready, status.get("latestCreatedRevisionName"))
+            result["serving_traffic"] = _serving_traffic(status.get("traffic"),
+                report_revision=revision, configured_revision=expected["expected_runtime_revision"])
     return result
 
 

tests/test_diagnose_account_facts_identity.py

@@ -410,3 +410,147 @@ def forbidden(*args, **kwargs):
     result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, check=False)
     assert result.returncode == 0, result.stderr
     assert result.stdout.strip() == "inert import passed"
+
+
+@pytest.mark.parametrize("scenario,status,report_positive,configured_positive,report_only", [
+    ("report_single", "single", True, True, True),
+    ("configured_single_report_ready_only", "single", False, True, False),
+    ("report_single_configured_old", "single", True, False, True),
+    ("split_both", "split", True, True, False),
+    ("split_neither", "split", False, False, False),
+    ("same_revision_components", "single", True, True, True),
+    ("zero_tag", "single", False, True, False),
+])
+def test_serving_traffic_uses_effective_positive_distribution_without_ready_fallback(
+    monkeypatch, scenario, status, report_positive, configured_positive, report_only,
+):
+    _environment(monkeypatch)
+    expected = diagnostic.resolve_expected()
+    configured = expected["expected_runtime_revision"]
+    report = _report(expected)
+    cloud = _metadata(expected)
+    report_revision = configured
+    if scenario in {"configured_single_report_ready_only", "split_both", "zero_tag"}:
+        report_revision = PRIVATE + "-new-report"
+    if scenario == "report_single_configured_old":
+        report_revision = PRIVATE + "-new-report"
+    report["diagnostics"]["runtime_revision"] = report_revision
+    report["runtime_release_receipt"]["runtime_revision"] = report_revision
+    cloud["status"]["latestReadyRevisionName"] = report_revision
+    cloud["status"]["latestCreatedRevisionName"] = report_revision
+    traffic = {
+        "report_single": [{"revisionName": report_revision, "percent": 100}],
+        "configured_single_report_ready_only": [{"revisionName": configured, "percent": 100}],
+        "report_single_configured_old": [{"revisionName": report_revision, "percent": 100}],
+        "split_both": [{"revisionName": configured, "percent": 25}, {"revisionName": report_revision, "percent": 75}],
+        "split_neither": [{"revisionName": PRIVATE + "-other-1", "percent": 20}, {"revisionName": PRIVATE + "-other-2", "percent": 80}],
+        "same_revision_components": [{"revisionName": configured, "percent": 40}, {"revisionName": configured, "percent": 60}],
+        "zero_tag": [{"revisionName": configured, "percent": 100}, {"revisionName": report_revision, "percent": 0, "tag": PRIVATE, "url": "https://" + PRIVATE}],
+    }[scenario]
+    cloud["status"]["traffic"] = traffic
+    result = _diagnose(report, expected, cloud_run_metadata=cloud)
+    assert result["revision_equalities"]["report_equals_ready"] is True
+    assert result["serving_traffic"] == {
+        "status": status,
+        "report_has_positive_traffic": report_positive,
+        "configured_has_positive_traffic": configured_positive,
+        "report_is_only_serving_revision": report_only,
+    }
+    assert PRIVATE not in json.dumps(result)
+
+
+@pytest.mark.parametrize("traffic", [None, []])
+def test_missing_traffic_keeps_all_serving_comparisons_unknown(monkeypatch, traffic):
+    _environment(monkeypatch)
+    expected = diagnostic.resolve_expected()
+    cloud = _metadata(expected)
+    if traffic is not None:
+        cloud["status"]["traffic"] = traffic
+    result = _diagnose(_report(expected), expected, cloud_run_metadata=cloud)
+    assert result["serving_traffic"] == {
+        "status": "absent", "report_has_positive_traffic": None,
+        "configured_has_positive_traffic": None, "report_is_only_serving_revision": None,
+    }
+
+
+@pytest.mark.parametrize("traffic", [
+    PRIVATE,
+    {"revisionName": PRIVATE, "percent": 100},
+    [PRIVATE],
+    [{"percent": 100}],
+    [{"revisionName": PRIVATE, "percent": None}],
+    [{"revisionName": PRIVATE}],
+    [{"revisionName": PRIVATE, "percent": True}],
+    [{"revisionName": PRIVATE, "percent": "100"}],
+    [{"revisionName": PRIVATE, "percent": 100.0}],
+    [{"revisionName": PRIVATE, "percent": -1}],
+    [{"revisionName": PRIVATE, "percent": 101}],
+    [{"revisionName": PRIVATE, "percent": 99}],
+    [{"revisionName": PRIVATE, "percent": 0, "tag": PRIVATE}],
+    [{"revisionName": PRIVATE, "percent": 75}, {"revisionName": PRIVATE + "-other", "percent": 75}],
+    [{"revisionName": {"nested": PRIVATE}, "percent": 100}],
+    [{"revisionName": "", "percent": 100}],
+    [{"revisionName": PRIVATE + "\nrevision", "percent": 100}],
+])
+def test_invalid_traffic_never_guesses_serving_or_emits_private_values(monkeypatch, traffic):
+    _environment(monkeypatch)
+    expected = diagnostic.resolve_expected()
+    cloud = _metadata(expected)
+    cloud["status"]["traffic"] = traffic
+    result = _diagnose(_report(expected), expected, cloud_run_metadata=cloud)
+    assert result["serving_traffic"] == {
+        "status": "invalid", "report_has_positive_traffic": None,
+        "configured_has_positive_traffic": None, "report_is_only_serving_revision": None,
+    }
+    assert PRIVATE not in json.dumps(result)
+
+
+def test_private_extra_traffic_fields_do_not_change_fixed_projection(monkeypatch):
+    _environment(monkeypatch)
+    expected = diagnostic.resolve_expected()
+    cloud = _metadata(expected)
+    cloud["status"]["traffic"] = [{
+        "revisionName": expected["expected_runtime_revision"], "percent": 100,
+        "private_account": {"nested": PRIVATE}, "url": PRIVATE, "tag": PRIVATE,
+    }]
+    result = _diagnose(_report(expected), expected, cloud_run_metadata=cloud)
+    assert result["serving_traffic"]["status"] == "single"
+    assert result["serving_traffic"]["report_is_only_serving_revision"] is True
+    assert PRIVATE not in json.dumps(result)
+
+
+def test_traffic_with_wrong_service_metadata_cannot_prove_serving(monkeypatch):
+    _environment(monkeypatch)
+    expected = diagnostic.resolve_expected()
+    cloud = _metadata(expected)
+    cloud["metadata"]["name"] = PRIVATE + "-other-service"
+    cloud["status"]["traffic"] = [{"revisionName": expected["expected_runtime_revision"], "percent": 100}]
+    result = _diagnose(_report(expected), expected, cloud_run_metadata=cloud)
+    assert result["cloud_run_metadata_status"] == "invalid"
+    assert result["serving_traffic"] == {
+        "status": "invalid", "report_has_positive_traffic": None,
+        "configured_has_positive_traffic": None, "report_is_only_serving_revision": None,
+    }
+
+
+def test_preflight_includes_unknown_serving_diagnostic_without_cloud_calls(monkeypatch, capsys):
+    _environment(monkeypatch)
+    assert diagnostic.main(["--preflight"]) == 0
+    result = json.loads(capsys.readouterr().out)
+    assert result["serving_traffic"]["status"] == "absent"
+    assert result["serving_traffic"]["report_has_positive_traffic"] is None
+
+
+def test_serving_diagnostic_does_not_change_original_cli_exit_gate(monkeypatch, capsys):
+    _environment(monkeypatch)
+    expected = diagnostic.resolve_expected()
+    cloud = _metadata(expected)
+    cloud["status"]["traffic"] = [{"revisionName": PRIVATE + "-other", "percent": 100}]
+    monkeypatch.setattr(publisher, "_latest_report_uri", lambda **kwargs: expected["expected_report_prefix"] + "/r.json")
+    monkeypatch.setattr(publisher, "_load_gcs_report", lambda *args, **kwargs: _report(expected))
+    monkeypatch.setattr(diagnostic, "_read_cloud_run_metadata", lambda *args: cloud)
+    assert diagnostic.main([], now=NOW) == 0
+    result = json.loads(capsys.readouterr().out)
+    assert result["projection_passed"] is True
+    assert result["serving_traffic"]["report_has_positive_traffic"] is False
+    assert result["serving_traffic"]["report_is_only_serving_revision"] is False