可观测性：投影 Production 正式 Run 记录

扫描 Production Run 内正式 JSON 与文本记录，按内容哈希识别节点新增或变化文件。\n\n提供 State、节点输出和最终结果的完整展开，同时保留脱敏摘要与路径边界检查。

production_build_agents/observability/production_capture.py

@@ -0,0 +1,164 @@
+"""Production 0.7 State 与 Run 记录的完整观测投影。"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+from typing import Any, Mapping
+
+from .full_capture import (
+    collect_state_records,
+    jsonable,
+    load_observation_record,
+)
+
+
+_RUN_TEXT_SUFFIXES = frozenset({".json", ".md", ".mmd", ".txt"})
+
+
+def _inside(path: Path, root: Path) -> bool:
+    try:
+        path.resolve().relative_to(root.resolve())
+    except (OSError, ValueError):
+        return False
+    return True
+
+
+def collect_production_run_records(
+    run_dir: Path,
+    *,
+    seen_hashes: dict[str, str] | None = None,
+    changed_only: bool = False,
+) -> dict[str, Any]:
+    """读取 Run 内正式文本记录。
+
+    ``changed_only`` 为真时只返回本节点新增或变化的文件。
+    """
+
+    root = run_dir.resolve()
+    records: dict[str, Any] = {}
+    if not root.is_dir():
+        return records
+    for path in sorted(root.rglob("*")):
+        if (
+            not path.is_file()
+            or path.suffix.lower() not in _RUN_TEXT_SUFFIXES
+            or not _inside(path, root)
+        ):
+            continue
+        relative = path.resolve().relative_to(root).as_posix()
+        try:
+            digest = hashlib.sha256(path.read_bytes()).hexdigest()
+        except OSError as exc:
+            records[relative] = {
+                "path": str(path.resolve()),
+                "exists": path.is_file(),
+                "read_error": f"{type(exc).__name__}: {exc}",
+            }
+            continue
+        previous = seen_hashes.get(relative) if seen_hashes is not None else None
+        if seen_hashes is not None:
+            seen_hashes[relative] = digest
+        if changed_only and previous == digest:
+            continue
+        record = load_observation_record(path.resolve())
+        if record is not None:
+            records[relative] = {**record, "sha256": digest}
+    return records
+
+
+def summarize_production_state(state: Mapping[str, Any]) -> dict[str, Any]:
+    """脱敏回退只保留运行身份、调度状态、计数和记录存在性。"""
+
+    counter_fields = (
+        "planner_calls",
+        "planner_review_calls",
+        "executor_calls",
+        "segment_validator_calls",
+        "assembly_calls",
+        "production_validator_calls",
+        "replan_count",
+        "max_plan_revisions",
+    )
+    segment_records = state.get("segment_records")
+    segments = (
+        {
+            str(segment_id): {
+                "status": record.get("status"),
+                "active_plan_version": record.get("active_plan_version"),
+                "accepted_plan_version": record.get("accepted_plan_version"),
+            }
+            for segment_id, record in segment_records.items()
+            if isinstance(record, Mapping)
+        }
+        if isinstance(segment_records, Mapping)
+        else {}
+    )
+    return {
+        "run_id": str(state.get("run_id") or "")[:160],
+        "protocol_version": state.get("protocol_version"),
+        "status": state.get("status"),
+        "phase": state.get("phase"),
+        "failure_code": state.get("failure_code"),
+        "replan_scope": state.get("replan_scope"),
+        "current_segment_id": state.get("current_segment_id"),
+        "counters": {
+            field: state.get(field)
+            for field in counter_fields
+            if field in state
+        },
+        "segments": segments,
+        "record_fields_present": sorted(collect_state_records(state)),
+        "event_count": len(state.get("event_log") or []),
+    }
+
+
+def expand_production_state(state: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "state": jsonable(state),
+        "records": collect_state_records(state),
+    }
+
+
+def expand_production_node_output(
+    phase: str,
+    state: Mapping[str, Any],
+    update: Mapping[str, Any] | None,
+    *,
+    run_records: Mapping[str, Any],
+    exception: BaseException | None = None,
+) -> dict[str, Any]:
+    merged = dict(state)
+    if update is not None:
+        merged.update(update)
+    return {
+        "node": phase,
+        "update": jsonable(update or {}),
+        "state_after": jsonable(merged),
+        "referenced_records": collect_state_records(merged),
+        "new_or_changed_run_records": jsonable(run_records),
+        "exception": (
+            {
+                "type": type(exception).__name__,
+                "message": str(exception),
+            }
+            if exception is not None
+            else None
+        ),
+    }
+
+
+def expand_production_result(
+    result: Mapping[str, Any],
+    *,
+    execution_mode: str,
+    graph_invoked: bool,
+    run_records: Mapping[str, Any],
+) -> dict[str, Any]:
+    return {
+        "execution_mode": execution_mode,
+        "graph_invoked": graph_invoked,
+        "result": jsonable(result),
+        "referenced_records": collect_state_records(result),
+        "run_records": jsonable(run_records),
+    }

tests/observability/test_production_capture.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from production_build_agents.observability.production_capture import (
+    collect_production_run_records,
+    expand_production_node_output,
+    summarize_production_state,
+)
+
+
+def test_production_run_record_scan_is_incremental(
+    tmp_path: Path,
+) -> None:
+    run_dir = tmp_path / "production-run"
+    candidate = run_dir / "segment_candidates" / "Segment1.v1.json"
+    candidate.parent.mkdir(parents=True)
+    candidate.write_text(
+        json.dumps(
+            {
+                "summary": "完整 Candidate",
+                "image_url": "https://internal.example/segment.png",
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    (run_dir / "video.mp4").write_bytes(b"\x00\x01")
+    seen: dict[str, str] = {}
+
+    first = collect_production_run_records(
+        run_dir,
+        seen_hashes=seen,
+        changed_only=True,
+    )
+    second = collect_production_run_records(
+        run_dir,
+        seen_hashes=seen,
+        changed_only=True,
+    )
+    candidate.write_text(
+        json.dumps(
+            {"summary": "更新后的完整 Candidate"},
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    third = collect_production_run_records(
+        run_dir,
+        seen_hashes=seen,
+        changed_only=True,
+    )
+
+    key = "segment_candidates/Segment1.v1.json"
+    assert first[key]["content"]["summary"] == "完整 Candidate"
+    assert "video.mp4" not in first
+    assert second == {}
+    assert third[key]["content"]["summary"] == "更新后的完整 Candidate"
+
+
+def test_production_node_output_keeps_segment_and_error_details(
+    tmp_path: Path,
+) -> None:
+    report = tmp_path / "Segment1.v1.json"
+    report.write_text(
+        json.dumps(
+            {"verdict": "FAIL", "reason": "真实验收原因"},
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    state = {
+        "run_id": "round-production",
+        "status": "RUNNING",
+        "current_segment_id": "Segment1",
+    }
+    update = {
+        "status": "FAILED",
+        "error": "真实 Production 错误",
+        "current_segment_validation_report_path": str(report),
+    }
+
+    observed = expand_production_node_output(
+        "validate_segment",
+        state,
+        update,
+        run_records={},
+        exception=RuntimeError("工具结果未知"),
+    )
+
+    assert observed["state_after"]["error"] == "真实 Production 错误"
+    assert observed["exception"]["message"] == "工具结果未知"
+    assert observed["referenced_records"][
+        "current_segment_validation_report_path"
+    ]["content"]["reason"] == "真实验收原因"
+
+
+def test_redacted_production_state_preserves_scheduling_facts() -> None:
+    summary = summarize_production_state(
+        {
+            "run_id": "round-production",
+            "protocol_version": "0.7",
+            "status": "RUNNING",
+            "phase": "EXECUTE_SEGMENT",
+            "current_segment_id": "Segment2",
+            "executor_calls": 3,
+            "event_log": ["正文不会进入摘要"],
+            "segment_records": {
+                "Segment2": {
+                    "status": "ready",
+                    "active_plan_version": 2,
+                    "delivery_path": "/private/Delivery.json",
+                }
+            },
+        }
+    )
+
+    assert summary["phase"] == "EXECUTE_SEGMENT"
+    assert summary["counters"]["executor_calls"] == 3
+    assert summary["segments"]["Segment2"]["status"] == "ready"
+    assert summary["event_count"] == 1
+    assert "/private/Delivery.json" not in json.dumps(summary)