feat(governance): 新增 result capture owner promotion review

apps/api/tests/test_ai_agent_result_capture_owner_promotion_review.py

@@ -0,0 +1,118 @@
+import copy
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.services.ai_agent_result_capture_owner_promotion_review import (
+    load_latest_ai_agent_result_capture_owner_promotion_review,
+)
+
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+FIXTURE = REPO_ROOT / "docs/evaluations/ai_agent_result_capture_owner_promotion_review_2026-06-14.json"
+
+
+def test_load_latest_ai_agent_result_capture_owner_promotion_review_snapshot() -> None:
+    data = load_latest_ai_agent_result_capture_owner_promotion_review()
+
+    assert data["schema_version"] == "ai_agent_result_capture_owner_promotion_review_v1"
+    assert data["program_status"]["current_task_id"] == "P2-125"
+    assert data["program_status"]["next_task_id"] == "P2-126"
+    assert data["program_status"]["overall_completion_percent"] == 100
+    assert data["program_status"]["read_only_mode"] is True
+
+    rollups = data["rollups"]
+    assert rollups["owner_promotion_packet_count"] == 5
+    assert rollups["execution_gate_check_count"] == 5
+    assert rollups["rollback_owner_review_count"] == 5
+    assert rollups["blocked_execution_write_count"] == 6
+    assert rollups["operator_action_count"] == 5
+    assert rollups["approval_required_packet_count"] == 2
+    assert rollups["blocked_packet_count"] == 1
+    assert rollups["approval_required_gate_count"] == 2
+    assert rollups["blocked_gate_count"] == 1
+    assert rollups["approval_required_rollback_count"] == 2
+    assert rollups["blocked_rollback_count"] == 1
+    assert rollups["critical_blocker_count"] == 5
+
+    assert {packet["promotion_review_only"] for packet in data["owner_promotion_packets"]} == {True}
+    assert {packet["execution_apply_enabled"] for packet in data["owner_promotion_packets"]} == {False}
+    assert {gate["execution_enabled"] for gate in data["execution_gate_checks"]} == {False}
+    assert {review["rollback_confirmed"] for review in data["rollback_owner_reviews"]} == {False}
+
+    zero_fields = [
+        "owner_promotion_approved_count",
+        "execution_gate_pass_count",
+        "rollback_owner_confirmed_count",
+        "post_write_verifier_ready_count",
+        "writer_apply_count",
+        "execution_apply_count",
+        "receipt_write_count",
+        "reviewer_queue_write_count",
+        "gateway_queue_write_count",
+        "telegram_send_count",
+        "bot_api_call_count",
+        "report_receipt_write_count",
+        "result_capture_write_count",
+        "learning_write_count",
+        "playbook_trust_write_count",
+        "production_write_count",
+        "secret_read_count",
+        "destructive_operation_count",
+    ]
+    for field in zero_fields:
+        assert rollups[field] == 0
+
+
+def test_result_capture_owner_promotion_review_rejects_execution_enabled(tmp_path: Path) -> None:
+    source = json.loads(FIXTURE.read_text(encoding="utf-8"))
+    source["execution_gate_checks"][0]["execution_enabled"] = True
+    target = tmp_path / "ai_agent_result_capture_owner_promotion_review_2026-06-14.json"
+    target.write_text(json.dumps(source), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="must not enable execution"):
+        load_latest_ai_agent_result_capture_owner_promotion_review(tmp_path)
+
+
+def test_result_capture_owner_promotion_review_rejects_truth_write_flag(tmp_path: Path) -> None:
+    source = json.loads(FIXTURE.read_text(encoding="utf-8"))
+    source["owner_promotion_truth"]["result_capture_write_enabled"] = True
+    target = tmp_path / "ai_agent_result_capture_owner_promotion_review_2026-06-14.json"
+    target.write_text(json.dumps(source), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="live execution/send/write flags"):
+        load_latest_ai_agent_result_capture_owner_promotion_review(tmp_path)
+
+
+def test_result_capture_owner_promotion_review_rejects_packet_apply_enabled(tmp_path: Path) -> None:
+    source = json.loads(FIXTURE.read_text(encoding="utf-8"))
+    source["owner_promotion_packets"][0]["execution_apply_enabled"] = True
+    target = tmp_path / "ai_agent_result_capture_owner_promotion_review_2026-06-14.json"
+    target.write_text(json.dumps(source), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="must not enable execution apply"):
+        load_latest_ai_agent_result_capture_owner_promotion_review(tmp_path)
+
+
+def test_result_capture_owner_promotion_review_rejects_rollup_drift(tmp_path: Path) -> None:
+    source = json.loads(FIXTURE.read_text(encoding="utf-8"))
+    source["rollups"]["owner_promotion_packet_count"] = 4
+    target = tmp_path / "ai_agent_result_capture_owner_promotion_review_2026-06-14.json"
+    target.write_text(json.dumps(source), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="rollup counts mismatch"):
+        load_latest_ai_agent_result_capture_owner_promotion_review(tmp_path)
+
+
+def test_result_capture_owner_promotion_review_rejects_forbidden_display_terms(tmp_path: Path) -> None:
+    source = copy.deepcopy(json.loads(FIXTURE.read_text(encoding="utf-8")))
+    source["operator_actions"][0]["operator_instruction"] = "do not expose session_id"
+    target = tmp_path / "ai_agent_result_capture_owner_promotion_review_2026-06-14.json"
+    target.write_text(json.dumps(source), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="forbidden display terms"):
+        load_latest_ai_agent_result_capture_owner_promotion_review(tmp_path)

apps/api/tests/test_ai_agent_result_capture_owner_promotion_review_api.py

@@ -0,0 +1,47 @@
+import os
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")
+
+from src.main import app
+
+
+@pytest.mark.asyncio
+async def test_get_agent_result_capture_owner_promotion_review_api() -> None:
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/api/v1/agents/agent-result-capture-owner-promotion-review")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "ai_agent_result_capture_owner_promotion_review_v1"
+    assert data["program_status"]["current_task_id"] == "P2-125"
+    assert data["program_status"]["next_task_id"] == "P2-126"
+    assert data["program_status"]["overall_completion_percent"] == 100
+
+    rollups = data["rollups"]
+    assert rollups["owner_promotion_packet_count"] == 5
+    assert rollups["execution_gate_check_count"] == 5
+    assert rollups["rollback_owner_review_count"] == 5
+    assert rollups["blocked_execution_write_count"] == 6
+    assert rollups["operator_action_count"] == 5
+    assert rollups["critical_blocker_count"] == 5
+    assert rollups["owner_promotion_approved_count"] == 0
+    assert rollups["execution_gate_pass_count"] == 0
+    assert rollups["rollback_owner_confirmed_count"] == 0
+    assert rollups["writer_apply_count"] == 0
+    assert rollups["execution_apply_count"] == 0
+    assert rollups["receipt_write_count"] == 0
+    assert rollups["result_capture_write_count"] == 0
+    assert rollups["learning_write_count"] == 0
+    assert rollups["playbook_trust_write_count"] == 0
+    assert rollups["gateway_queue_write_count"] == 0
+    assert rollups["telegram_send_count"] == 0
+    assert rollups["production_write_count"] == 0
+
+    assert {packet["promotion_review_only"] for packet in data["owner_promotion_packets"]} == {True}
+    assert {packet["execution_apply_enabled"] for packet in data["owner_promotion_packets"]} == {False}
+    assert {gate["execution_enabled"] for gate in data["execution_gate_checks"]} == {False}
+    assert {review["rollback_confirmed"] for review in data["rollback_owner_reviews"]} == {False}