feat(ai): 新增 P2-406B receipt owner review

apps/api/tests/test_ai_agent_receipt_readback_owner_review.py

@@ -0,0 +1,314 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from src.services.ai_agent_receipt_readback_owner_review import (
+    load_latest_ai_agent_receipt_readback_owner_review,
+)
+
+
+def test_load_latest_ai_agent_receipt_readback_owner_review_reads_newest_file(tmp_path):
+    older = _snapshot(generated_at="2026-06-17T00:00:00+08:00")
+    newer = _snapshot(generated_at="2026-06-18T00:00:00+08:00")
+    (tmp_path / "ai_agent_receipt_readback_owner_review_2026-06-17.json").write_text(
+        json.dumps(older),
+        encoding="utf-8",
+    )
+    (tmp_path / "ai_agent_receipt_readback_owner_review_2026-06-18.json").write_text(
+        json.dumps(newer),
+        encoding="utf-8",
+    )
+
+    loaded = load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+    assert loaded["generated_at"] == "2026-06-18T00:00:00+08:00"
+    assert loaded["program_status"]["current_task_id"] == "P2-406B"
+    assert loaded["program_status"]["next_task_id"] == "P2-407"
+    assert loaded["receipt_readback_plan"]["receipt_owner"] == "hermes"
+    assert loaded["receipt_readback_plan"]["arbiter"] == "openclaw"
+    assert loaded["receipt_readback_plan"]["replay_owner"] == "nemotron"
+    assert loaded["activation_boundaries"]["telegram_send_enabled"] is False
+    assert loaded["activation_boundaries"]["gateway_queue_write_enabled"] is False
+
+
+def test_ai_agent_receipt_readback_owner_review_requires_read_only_mode(tmp_path):
+    snapshot = _snapshot()
+    snapshot["program_status"]["read_only_mode"] = False
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="read_only_mode"):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def test_ai_agent_receipt_readback_owner_review_blocks_telegram_send(tmp_path):
+    snapshot = _snapshot()
+    snapshot["activation_boundaries"]["telegram_send_enabled"] = True
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="activation boundaries"):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def test_ai_agent_receipt_readback_owner_review_blocks_bot_api_call(tmp_path):
+    snapshot = _snapshot()
+    snapshot["telegram_policy"]["direct_bot_api_allowed"] = True
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="telegram_policy"):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def test_ai_agent_receipt_readback_owner_review_requires_rollup_consistency(tmp_path):
+    snapshot = _snapshot()
+    snapshot["rollups"]["owner_review_gate_count"] = 99
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="rollup counts"):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def test_ai_agent_receipt_readback_owner_review_requires_gate_evidence(tmp_path):
+    snapshot = _snapshot()
+    snapshot["owner_review_gates"][0]["evidence_refs"] = []
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="evidence_refs"):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def test_ai_agent_receipt_readback_owner_review_rejects_private_terms(tmp_path):
+    snapshot = _snapshot()
+    snapshot["owner_review_gates"][0]["next_action"] = "不要顯示工作視窗內容"
+    _write_snapshot(tmp_path, snapshot)
+
+    with pytest.raises(ValueError, match="forbidden public terms"):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def test_ai_agent_receipt_readback_owner_review_fails_when_missing(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        load_latest_ai_agent_receipt_readback_owner_review(tmp_path)
+
+
+def _write_snapshot(tmp_path, payload: dict) -> None:
+    (tmp_path / "ai_agent_receipt_readback_owner_review_2026-06-18.json").write_text(
+        json.dumps(payload),
+        encoding="utf-8",
+    )
+
+
+def _snapshot(*, generated_at: str = "2026-06-18T00:00:00+08:00") -> dict:
+    boundaries = {
+        "owner_review_required_before_canary": True,
+        "read_only_review_allowed": True,
+        "scheduler_enabled": False,
+        "gateway_queue_write_enabled": False,
+        "telegram_send_enabled": False,
+        "bot_api_call_enabled": False,
+        "delivery_receipt_write_enabled": False,
+        "report_receipt_write_enabled": False,
+        "receipt_production_write_enabled": False,
+        "ai_analysis_run_enabled": False,
+        "medium_low_auto_execution_enabled": False,
+        "production_optimization_write_enabled": False,
+        "production_write_enabled": False,
+        "secret_read_enabled": False,
+        "paid_api_enabled": False,
+        "host_write_enabled": False,
+        "kubectl_action_enabled": False,
+        "destructive_operation_enabled": False,
+        "external_lookup_enabled": False,
+    }
+    return {
+        "schema_version": "ai_agent_receipt_readback_owner_review_v1",
+        "generated_at": generated_at,
+        "program_status": {
+            "overall_completion_percent": 100,
+            "current_priority": "P2",
+            "current_task_id": "P2-406B",
+            "next_task_id": "P2-407",
+            "read_only_mode": True,
+            "runtime_authority": "receipt_readback_owner_review_only_no_send_or_write",
+            "status_note": "只讀 owner review。",
+        },
+        "source_refs": ["docs/evaluations/example.json"],
+        "source_readbacks": [
+            _source("status", "ai_agent_report_status_board_v1", "openclaw"),
+            _source("delivery", "ai_agent_report_live_delivery_approval_package_v1", "hermes"),
+            _source("receipt", "ai_agent_telegram_receipt_approval_package_v1", "hermes"),
+            _source("professional", "ai_agent_professional_task_expansion_v1", "nemotron"),
+            _source("drift", "dependency_supply_chain_drift_monitor_v1", "openclaw"),
+            _source("truth", "ai_agent_report_truth_actionability_review_v1", "openclaw"),
+        ],
+        "rollups": {
+            "source_readback_count": 6,
+            "report_cadence_count": 3,
+            "owner_review_gate_count": 6,
+            "receipt_readback_check_count": 8,
+            "drift_candidate_count": 9,
+            "report_truth_blocker_count": 5,
+            "approval_required_count": 6,
+            "blocked_runtime_action_count": 17,
+            "live_write_count": 0,
+            "telegram_send_count": 0,
+            "gateway_queue_write_count": 0,
+            "bot_api_call_count": 0,
+            "production_write_count": 0,
+            "secret_read_count": 0,
+            "paid_api_call_count": 0,
+            "host_write_count": 0,
+            "kubectl_action_count": 0,
+        },
+        "report_owner_review": {
+            "cadences": [
+                _cadence("daily", "hermes"),
+                _cadence("weekly", "openclaw"),
+                _cadence("monthly", "nemotron"),
+            ]
+        },
+        "receipt_readback_plan": {
+            "canonical_room": "AwoooI SRE 戰情室",
+            "canonical_room_env": "SRE_GROUP_CHAT_ID",
+            "gateway_owner": "telegram_ops",
+            "arbiter": "openclaw",
+            "receipt_owner": "hermes",
+            "replay_owner": "nemotron",
+            "dry_run_receipt_only": True,
+            "owner_review_required_before_canary": True,
+            "canary_send_approved": False,
+            "receipt_production_write_enabled": False,
+            "readback_checks": [_check(index) for index in range(8)],
+        },
+        "owner_review_gates": [
+            _gate("daily_report_receipt_owner_review", "hermes"),
+            _gate("weekly_report_truth_owner_review", "openclaw"),
+            _gate("monthly_report_owner_review", "nemotron"),
+            _gate("telegram_canary_receipt_owner_review", "hermes"),
+            _gate("supply_chain_drift_owner_review", "openclaw"),
+            _gate("report_truth_data_chain_owner_review", "openclaw"),
+        ],
+        "drift_monitor_owner_review": {
+            "source_task_id": "P2-004",
+            "drift_candidate_count": 9,
+            "action_required_candidate_count": 9,
+            "stale_source_snapshot_count": 5,
+            "external_lookup_allowed": False,
+            "package_upgrade_allowed": False,
+            "owner_actions": ["prepare_packet"],
+        },
+        "report_truth_owner_review": {
+            "source_task_id": "P2-403J",
+            "all_zero_weekly_report_is_actionable_anomaly": True,
+            "all_zero_weekly_report_confidence": "low_trust_actionable_anomaly",
+            "zero_signal_finding_count": 5,
+            "critical_finding_count": 1,
+            "high_finding_count": 3,
+            "telegram_report_send_allowed": False,
+            "cronjob_change_allowed": False,
+            "freshness_gate_implemented": False,
+            "source_confidence_gate_implemented": False,
+            "actionability_score_implemented": False,
+            "owner_actions": ["keep_anomaly_visible"],
+        },
+        "activation_boundaries": boundaries,
+        "telegram_policy": {
+            "status": "owner_review_only_no_send",
+            "canonical_room": "AwoooI SRE 戰情室",
+            "canonical_room_env": "SRE_GROUP_CHAT_ID",
+            "gateway_queue_write_allowed": False,
+            "direct_bot_api_allowed": False,
+            "telegram_send_allowed": False,
+            "receipt_write_allowed": False,
+        },
+        "agent_roles": [
+            _role("openclaw"),
+            _role("hermes"),
+            _role("nemotron"),
+        ],
+        "operator_decision": {
+            "status": "owner_review_required_before_canary",
+            "approval_packet_required": True,
+            "rollback_plan_required": True,
+            "mute_plan_required": True,
+            "required_fields": ["owner_identity"],
+            "stop_conditions": ["telegram_send_count 非 0"],
+            "next_decision_point": "P2-407",
+        },
+        "next_actions": [
+            {
+                "task_id": "P2-407",
+                "priority": "P2",
+                "summary": "下一步",
+                "gate": "no write",
+            }
+        ],
+    }
+
+
+def _source(readback_id: str, schema: str, owner: str) -> dict:
+    return {
+        "readback_id": readback_id,
+        "source_schema_version": schema,
+        "source_ref": f"docs/evaluations/{readback_id}.json",
+        "endpoint": f"GET /api/v1/agents/{readback_id}",
+        "owner_agent": owner,
+        "status": "readback_ready",
+        "evidence_status": "ready",
+        "key_readback": "zero live write",
+        "next_action": "owner review",
+    }
+
+
+def _cadence(cadence_id: str, owner: str) -> dict:
+    return {
+        "cadence_id": cadence_id,
+        "display_name": cadence_id,
+        "owner_agent": owner,
+        "completion_percent": 100,
+        "delivery_state": "draft_only",
+        "source_confidence": "visible",
+        "live_delivery_count": 0,
+        "review_status": "owner_review_required_before_send",
+        "next_gate": f"{cadence_id}_gate",
+    }
+
+
+def _check(index: int) -> dict:
+    return {
+        "check_id": f"check_{index}",
+        "owner_agent": "hermes",
+        "status": "ready_no_send",
+        "evidence_refs": ["GET /api/v1/agents/example"],
+        "blocked_runtime_action": "telegram_send",
+    }
+
+
+def _gate(gate_id: str, owner: str) -> dict:
+    return {
+        "gate_id": gate_id,
+        "title": gate_id,
+        "owner_agent": owner,
+        "risk_tier": "high",
+        "status": "owner_review_required",
+        "approval_required": True,
+        "required_owner_fields": ["owner_identity"],
+        "acceptance_checks": ["check"],
+        "rejection_reasons": ["missing owner"],
+        "blocked_runtime_actions": ["telegram_send_enabled"],
+        "evidence_refs": ["GET /api/v1/agents/example"],
+        "next_action": "owner review",
+    }
+
+
+def _role(agent_id: str) -> dict:
+    return {
+        "agent_id": agent_id,
+        "role": "role",
+        "specialty": "specialty",
+        "autonomy_level": "L2",
+        "approval_gate": "human",
+        "runtime_write_allowed": False,
+        "outputs": ["packet"],
+    }

apps/api/tests/test_ai_agent_receipt_readback_owner_review_api.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from src.api.v1.agents import router
+
+
+def test_ai_agent_receipt_readback_owner_review_endpoint_returns_committed_snapshot():
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+    client = TestClient(app)
+
+    response = client.get("/api/v1/agents/agent-receipt-readback-owner-review")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["schema_version"] == "ai_agent_receipt_readback_owner_review_v1"
+    assert data["program_status"]["current_task_id"] == "P2-406B"
+    assert data["program_status"]["next_task_id"] == "P2-407"
+    assert data["program_status"]["read_only_mode"] is True
+    assert data["rollups"]["source_readback_count"] == len(data["source_readbacks"]) == 6
+    assert data["rollups"]["report_cadence_count"] == 3
+    assert data["rollups"]["owner_review_gate_count"] == len(data["owner_review_gates"]) == 6
+    assert data["rollups"]["receipt_readback_check_count"] == 8
+    assert data["rollups"]["drift_candidate_count"] == 9
+    assert data["rollups"]["report_truth_blocker_count"] == 5
+    assert data["rollups"]["approval_required_count"] == 6
+    assert data["rollups"]["live_write_count"] == 0
+    assert data["rollups"]["telegram_send_count"] == 0
+    assert data["rollups"]["gateway_queue_write_count"] == 0
+    assert data["rollups"]["bot_api_call_count"] == 0
+    assert data["rollups"]["production_write_count"] == 0
+    assert data["rollups"]["secret_read_count"] == 0
+    assert data["rollups"]["paid_api_call_count"] == 0
+    assert data["rollups"]["host_write_count"] == 0
+    assert data["rollups"]["kubectl_action_count"] == 0
+    assert data["receipt_readback_plan"]["canonical_room"] == "AwoooI SRE 戰情室"
+    assert data["receipt_readback_plan"]["canonical_room_env"] == "SRE_GROUP_CHAT_ID"
+    assert data["receipt_readback_plan"]["arbiter"] == "openclaw"
+    assert data["receipt_readback_plan"]["receipt_owner"] == "hermes"
+    assert data["receipt_readback_plan"]["replay_owner"] == "nemotron"
+    assert data["receipt_readback_plan"]["canary_send_approved"] is False
+    assert data["activation_boundaries"]["gateway_queue_write_enabled"] is False
+    assert data["activation_boundaries"]["telegram_send_enabled"] is False
+    assert data["activation_boundaries"]["bot_api_call_enabled"] is False
+    assert data["activation_boundaries"]["receipt_production_write_enabled"] is False
+    assert data["activation_boundaries"]["production_write_enabled"] is False
+    assert data["activation_boundaries"]["secret_read_enabled"] is False
+    assert data["activation_boundaries"]["paid_api_enabled"] is False
+    assert data["activation_boundaries"]["host_write_enabled"] is False
+    assert data["activation_boundaries"]["kubectl_action_enabled"] is False
+    assert data["telegram_policy"]["telegram_send_allowed"] is False
+    assert data["telegram_policy"]["gateway_queue_write_allowed"] is False
+    assert data["telegram_policy"]["direct_bot_api_allowed"] is False