可观测性：细分模型候选、合同校正与媒体复核阶段

将模型调用的业务展示区分为初版候选、运行时合同校正和最终媒体证据复核，补充校正次数、触发原因、业务记录类型以及媒体复核范围。

同时放宽对 ExecutorCandidate、Task ValidatorCandidate 和 Stage ValidatorCandidate 的识别条件，使尚未由确定性代码汇总 verdict 的模型候选也能获得正确业务标题。

预处理投影现在会先确认报告内容是映射类型，避免 Markdown 原文被当作结构化记录访问。以上改动仅影响观测展示，不改变 Agent 输入、合同校验或业务执行结果。

新增候选识别、合同校正、媒体复核、校正工具调用和 Markdown 预处理报告的回归测试。

production_build_agents/observability/business_views.py

@@ -325,11 +325,14 @@ def infer_business_record_kind(content: Any) -> str | None:
         return "global_data_plan"
     if {"segments", "timeline"} <= keys:
         return "production_plan"
-    if {"artifact_expectation_bindings", "artifacts", "task_id"} <= keys:
+    if (
+        {"artifact_binding_claims", "artifacts", "task_id"} <= keys
+        or {"artifact_expectation_bindings", "artifacts", "task_id"} <= keys
+    ):
         return "executor_delivery"
-    if {"criterion_results", "verdict", "task_id"} <= keys:
+    if {"criterion_results", "task_id"} <= keys:
         return "task_validation_report"
-    if {"requirement_results", "verdict"} <= keys:
+    if "requirement_results" in keys:
         return "stage_validation_report"
     if {"active_artifacts", "requirement_evaluations", "tasks"} <= keys:
         return "global_data_stage_delivery"

production_build_agents/observability/projections.py

@@ -1020,7 +1020,12 @@ def _compact_node_records(
     if domain == "global_data":
         if phase == "preprocess":
             brief = get("production_brief") or {}
-            report = get("preprocess_report") or {}
+            report_value = get("preprocess_report")
+            report = (
+                report_value
+                if isinstance(report_value, Mapping)
+                else {}
+            )
             output.update(
                 {
                     "帖子类型": brief.get("帖子类型")

production_build_agents/observability/readable_events.py

@@ -101,6 +101,27 @@ _FIELD_TITLES = {
 
 _IMPORTANT_FIELDS = tuple(_FIELD_TITLES)
 _MAX_SUMMARY_CHARS = 900
+_CORRECTION_PREFIX = "上一版未通过"
+_MEDIA_REVIEW_PREFIX = "最终媒体证据自检："
+
+_RESPONSE_SUBJECTS = {
+    "global_data_plan": "Global Data 规划",
+    "production_plan": "Production 规划",
+    "executor_delivery": "Executor 交付",
+    "task_validation_report": "Task 验收结论",
+    "stage_validation_report": "阶段验收结论",
+    "global_data_stage_delivery": "Global Data 阶段交付",
+    "progress_decision": "Production 调度决定",
+}
+
+_CORRECTION_REASON_TITLES = {
+    "missing_binding_evidence": "Artifact Binding 缺少工具调用证据",
+    "undispositioned_source_asset": "存在来源素材尚未纳入交付安排",
+    "source_asset_count_mismatch": "来源素材数量与交付要求不一致",
+    "missing_source_asset_binding": "存在来源素材尚未绑定正式产物",
+    "insufficient_artifact_bindings": "正式产物的验收目标绑定不足",
+    "planner_missing_required_read": "Planner 尚未读取完整必需范围",
+}
 
 
 def tool_title(name: object) -> str:
@@ -266,6 +287,139 @@ def readable_tool_calls(tool_calls: list[dict[str, Any]]) -> list[dict[str, Any]
     ]
 
 
+def _message_text(message: Any) -> str:
+    return _text(getattr(message, "content", None)).strip()
+
+
+def _human_messages(messages: list[Any]) -> list[Any]:
+    return [
+        message
+        for message in messages
+        if getattr(message, "type", None) == "human"
+    ]
+
+
+def _correction_reason_codes(text: str) -> list[str]:
+    codes: list[str] = []
+    for code in re.findall(r"\b([a-z][a-z0-9_]{2,})\s*:", text):
+        if code not in codes:
+            codes.append(code)
+    return codes
+
+
+def _correction_reason(text: str, codes: list[str]) -> str:
+    known = [
+        _CORRECTION_REASON_TITLES[code]
+        for code in codes
+        if code in _CORRECTION_REASON_TITLES
+    ]
+    if known:
+        return "；".join(known)
+    if "未出现在工具或依赖产物中的 URI" in text:
+        return "交付引用了尚未由工具或依赖产物验证的地址"
+    if "格式校验" in text:
+        return "上一版输出格式未通过运行时校验"
+    if "合同校验" in text:
+        return "上一版输出未通过运行时合同校验"
+    return "上一版输出未通过运行时校验"
+
+
+def _reviewed_media_count(text: str, message: Any) -> int:
+    parsed = _parse(text)
+    artifact_uris = (
+        parsed.get("artifact_uris")
+        if isinstance(parsed, Mapping)
+        else None
+    )
+    declared = len(artifact_uris) if isinstance(artifact_uris, list) else 0
+    content = getattr(message, "content", None)
+    attached = (
+        sum(
+            1
+            for block in content
+            if isinstance(block, Mapping)
+            and block.get("type") in {"image", "image_url"}
+        )
+        if isinstance(content, list)
+        else 0
+    )
+    # artifact_uris 是业务产物数量；随消息附带的图片可能包含视频抽帧，
+    # 只能在没有显式产物清单时作为回退，不能把 1 个视频的 N 帧算成 N 个产物。
+    return declared or attached
+
+
+def _turn_projection(
+    messages: list[Any],
+    *,
+    record_kind: str | None,
+) -> dict[str, Any]:
+    humans = _human_messages(messages)
+    last_human = humans[-1] if humans else None
+    last_text = _message_text(last_human) if last_human is not None else ""
+    correction_index = sum(
+        1
+        for message in humans
+        if _message_text(message).startswith(_CORRECTION_PREFIX)
+    )
+    subject = _RESPONSE_SUBJECTS.get(record_kind, "业务答复")
+
+    if last_text.startswith(_CORRECTION_PREFIX):
+        codes = _correction_reason_codes(last_text)
+        reason = _correction_reason(last_text, codes)
+        return {
+            "phase": "contract_correction",
+            "label": f"合同校正 · 第 {correction_index} 次",
+            "tool_prefix": "校正中 · ",
+            "content_prefix": f"校正原因：{reason}\n校正结果：",
+            "payload": {
+                "business_phase": "contract_correction",
+                "attempt": correction_index + 1,
+                "correction_index": correction_index,
+                "trigger": "runtime_contract_validation",
+                "previous_candidate_ok": False,
+                "correction_reason": reason,
+                "correction_reason_codes": codes,
+                "response_kind": record_kind,
+            },
+        }
+    if last_text.startswith(_MEDIA_REVIEW_PREFIX):
+        media_count = _reviewed_media_count(last_text, last_human)
+        scope = (
+            f"{media_count} 个媒体产物"
+            if media_count
+            else "本轮媒体产物"
+        )
+        return {
+            "phase": "media_evidence_review",
+            "label": "媒体证据复核",
+            "tool_prefix": "媒体复核中 · ",
+            "content_prefix": f"复核范围：{scope}\n复核结果：",
+            "payload": {
+                "business_phase": "media_evidence_review",
+                "attempt": correction_index + 1,
+                "trigger": "final_media_evidence_review",
+                "reviewed_media_count": media_count,
+                "response_kind": record_kind,
+            },
+        }
+    return {
+        "phase": "initial_candidate",
+        "label": (
+            f"初版候选 · {subject}"
+            if record_kind is not None
+            else business_response_title(None)
+        ),
+        "tool_prefix": "",
+        "content_prefix": "",
+        "payload": {
+            "business_phase": "initial_candidate",
+            "attempt": 1,
+            "trigger": "initial_request",
+            "response_kind": record_kind,
+        },
+    }
+
+
 class BusinessReadableFlowCollector(FlowCollector):
     """保留 SDK 采集事实，只替换 LLM/Tool 卡片的展示投影。"""
 
@@ -298,6 +452,10 @@ class BusinessReadableFlowCollector(FlowCollector):
             )
             reasoning = _reasoning(message) if message is not None else ""
             input_tokens, output_tokens = _usage(message, response)
+            source = list(self._pending_msgs.get(run_id, None) or [])
+            parsed_text = _parse(text)
+            record_kind = infer_business_record_kind(parsed_text)
+            turn = _turn_projection(source, record_kind=record_kind)
 
             group = self._new_group() if len(tool_calls) >= 2 else ""
             if group:
@@ -309,21 +467,23 @@ class BusinessReadableFlowCollector(FlowCollector):
             readable_calls = readable_tool_calls(tool_calls)
             names = [item["title"] for item in readable_calls]
             if names:
-                label = (
+                action_label = (
                     f"模型决定调用：{names[0]}"
                     if len(names) == 1
                     else f"模型并行调用 {len(names)} 个工具"
                 )
+                label = f"{turn['tool_prefix']}{action_label}"
                 content = readable_value(
                     text,
                     empty_text=f"下一步：{'、'.join(names)}",
                 )
             else:
-                parsed_text = _parse(text)
-                label = business_response_title(parsed_text)
+                label = turn["label"]
                 content = readable_value(text, empty_text="模型已完成本轮判断")
+                if turn["content_prefix"]:
+                    content = f"{turn['content_prefix']}{content}"
 
-            payload: dict[str, Any] = {}
+            payload: dict[str, Any] = dict(turn["payload"])
             if readable_calls:
                 payload["tool_calls"] = readable_calls
             if reasoning:
@@ -345,10 +505,11 @@ class BusinessReadableFlowCollector(FlowCollector):
             if self.run is not None:
                 self.run.account(input_tokens or 0, output_tokens or 0)
 
-            source = self._pending_msgs.pop(run_id, None)
-            if source is not None:
+            original_source = self._pending_msgs.pop(run_id, None)
+            if original_source is not None:
                 mc.record_messages(
-                    list(source) + ([message] if message is not None else []),
+                    list(original_source)
+                    + ([message] if message is not None else []),
                     idx=idx,
                 )
         except Exception:

tests/observability/test_business_projections.py

@@ -383,6 +383,39 @@ def test_prepare_completion_is_a_business_result_not_an_empty_update() -> None:
     assert output["下一阶段"] == "验收 Global Data 阶段"
 
 
+def test_preprocess_markdown_report_does_not_break_business_projection(
+    tmp_path: Path,
+) -> None:
+    brief_path = _record(
+        tmp_path,
+        "production_brief.json",
+        {
+            "post_type": "口播",
+            "core_production_points": ["固定人物形象"],
+            "source_assets": [],
+        },
+    )
+    report_path = tmp_path / "preprocess_report.md"
+    report_path.write_text(
+        "# 预处理报告\n\n全部输入已读取。",
+        encoding="utf-8",
+    )
+    output = project_node_output(
+        "global_data",
+        "preprocess",
+        {"status": "RUNNING", "phase": "PREPROCESS"},
+        {
+            "production_brief_path": brief_path,
+            "preprocess_report_path": str(report_path),
+            "status": "RUNNING",
+            "phase": "PLAN",
+        },
+    )
+
+    assert output["帖子类型"] == "口播"
+    assert output["预处理问题"] == "0 项"
+
+
 def test_failed_planners_explain_business_impact_without_json() -> None:
     global_output = project_node_output(
         "global_data",

tests/observability/test_readable_events.py

@@ -3,13 +3,16 @@ from __future__ import annotations
 import json
 from types import SimpleNamespace
 
-from langchain_core.messages import AIMessage, ToolMessage
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
 
 from production_build_agents.observability.readable_events import (
     BusinessReadableFlowCollector,
     readable_value,
     tool_title,
 )
+from production_build_agents.observability.business_views import (
+    business_response_title,
+)
 
 
 class _ModuleContext:
@@ -88,6 +91,25 @@ def test_fenced_global_data_plan_is_rendered_as_business_logic() -> None:
     assert "```json" not in text
 
 
+def test_validator_candidate_without_verdict_has_business_title() -> None:
+    candidate = {
+        "task_id": "Task3",
+        "criterion_results": [
+            {
+                "criterion_id": "Criterion1",
+                "passed": True,
+                "reason": "媒体证据完整",
+            }
+        ],
+        "summary": "任务满足验收要求",
+    }
+    text = readable_value(json.dumps(candidate, ensure_ascii=False))
+
+    assert business_response_title(candidate) == "模型形成 Task 验收结论"
+    assert "业务摘要：任务满足验收要求" in text
+    assert "criterion_results" not in text
+
+
 def test_tool_names_have_business_titles() -> None:
     assert tool_title("read_production_brief") == "读取 Production Brief"
     assert tool_title("assemble_segment_media") == "组装 Segment 媒体"
@@ -159,12 +181,158 @@ def test_llm_plan_card_has_business_title_and_keeps_raw_message() -> None:
 
     _, type_, event = context.events[0]
     assert type_ == "llm"
-    assert event["label"] == "模型形成 Global Data 规划"
+    assert event["label"] == "初版候选 · Global Data 规划"
     assert event["content"].startswith("规划目标：准备共享素材")
     assert "```json" not in event["content"]
+    assert event["payload"]["business_phase"] == "initial_candidate"
     assert context.messages[0][0][-1] is message
 
 
+def test_contract_correction_is_labeled_before_upload() -> None:
+    run = _Run()
+    context = _ModuleContext()
+    collector = BusinessReadableFlowCollector(run, auto_module=False)
+    collector._frame = lambda *_args: (context, 5)
+    collector._pending_msgs["llm-correction"] = [
+        HumanMessage(content='{"task_id":"Task3"}'),
+        AIMessage(content='{"summary":"初版"}'),
+        HumanMessage(
+            content=(
+                "上一版未通过运行时校验。错误："
+                "missing_binding_evidence: 该 Expectation 必须声明工具证据。"
+            )
+        ),
+    ]
+    corrected = AIMessage(
+        content=json.dumps(
+            {
+                "task_id": "Task3",
+                "artifacts": [{"uri": "/run/image.png"}],
+                "artifact_binding_claims": [
+                    {
+                        "expectation_id": "Expectation1",
+                        "evidence_tool_call_ids": ["call-1"],
+                    }
+                ],
+                "summary": "已补充工具证据",
+            },
+            ensure_ascii=False,
+        ),
+        usage_metadata={
+            "input_tokens": 30,
+            "output_tokens": 8,
+            "total_tokens": 38,
+        },
+    )
+    response = SimpleNamespace(
+        generations=[[SimpleNamespace(message=corrected)]],
+        llm_output={},
+    )
+
+    collector.on_llm_end(response, run_id="llm-correction")
+
+    _, type_, event = context.events[0]
+    assert type_ == "llm"
+    assert event["label"] == "合同校正 · 第 1 次"
+    assert event["content"].startswith(
+        "校正原因：Artifact Binding 缺少工具调用证据\n校正结果："
+    )
+    assert event["payload"]["business_phase"] == "contract_correction"
+    assert event["payload"]["attempt"] == 2
+    assert event["payload"]["previous_candidate_ok"] is False
+    assert event["payload"]["correction_reason_codes"] == [
+        "missing_binding_evidence"
+    ]
+
+
+def test_media_review_is_labeled_and_reports_scope() -> None:
+    run = _Run()
+    context = _ModuleContext()
+    collector = BusinessReadableFlowCollector(run, auto_module=False)
+    collector._frame = lambda *_args: (context, 6)
+    collector._pending_msgs["llm-media-review"] = [
+        HumanMessage(content='{"task_id":"Task3"}'),
+        HumanMessage(
+            content=[
+                {
+                    "type": "text",
+                    "text": (
+                        "最终媒体证据自检："
+                        '{"artifact_uris":["/run/image.png","/run/video.mp4"]}'
+                    ),
+                },
+                {
+                    "type": "image_url",
+                    "image_url": {"url": "data:image/png;base64,AA=="},
+                },
+            ]
+        ),
+    ]
+    reviewed = AIMessage(
+        content=json.dumps(
+            {
+                "task_id": "Task3",
+                "artifacts": [
+                    {"uri": "/run/image.png"},
+                    {"uri": "/run/video.mp4"},
+                ],
+                "artifact_binding_claims": [],
+                "summary": "媒体内容与制作要求一致",
+            },
+            ensure_ascii=False,
+        )
+    )
+    response = SimpleNamespace(
+        generations=[[SimpleNamespace(message=reviewed)]],
+        llm_output={},
+    )
+
+    collector.on_llm_end(response, run_id="llm-media-review")
+
+    _, type_, event = context.events[0]
+    assert type_ == "llm"
+    assert event["label"] == "媒体证据复核"
+    assert event["content"].startswith(
+        "复核范围：2 个媒体产物\n复核结果："
+    )
+    assert event["payload"]["business_phase"] == "media_evidence_review"
+    assert event["payload"]["reviewed_media_count"] == 2
+
+
+def test_correction_tool_call_keeps_action_and_adds_phase() -> None:
+    run = _Run()
+    context = _ModuleContext()
+    collector = BusinessReadableFlowCollector(run, auto_module=False)
+    collector._frame = lambda *_args: (context, 7)
+    collector._pending_msgs["llm-correction-tool"] = [
+        HumanMessage(
+            content="上一版未通过运行时格式校验。错误：缺少字段。"
+        )
+    ]
+    message = AIMessage(
+        content="重新检查图片",
+        tool_calls=[
+            {
+                "id": "call-2",
+                "name": "view_images",
+                "args": {"image_sources": ["/run/image.png"]},
+            }
+        ],
+    )
+    response = SimpleNamespace(
+        generations=[[SimpleNamespace(message=message)]],
+        llm_output={},
+    )
+
+    collector.on_llm_end(response, run_id="llm-correction-tool")
+
+    _, type_, event = context.events[0]
+    assert type_ == "llm"
+    assert event["label"] == "校正中 · 模型决定调用：查看图片"
+    assert event["payload"]["business_phase"] == "contract_correction"
+    assert event["payload"]["tool_calls"][0]["name"] == "view_images"
+
+
 def test_tool_result_card_uses_summary_instead_of_raw_json() -> None:
     context = _ModuleContext()
     collector = BusinessReadableFlowCollector(_Run(), auto_module=False)