feat(governance): 新增 owner-approved execution rehearsal
All checks were successful
Code Review / ai-code-review (push) Successful in 13s
CD Pipeline / tests (push) Successful in 1m27s
CD Pipeline / build-and-deploy (push) Successful in 4m45s
CD Pipeline / post-deploy-checks (push) Successful in 26s

This commit is contained in:
Your Name
2026-06-14 01:57:23 +08:00
parent 0a737cf400
commit 1ceaa45829
10 changed files with 1533 additions and 1 deletions

View File

@@ -0,0 +1,96 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from src.services.ai_agent_result_capture_owner_approved_execution_rehearsal import (
load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal,
)
REPO_ROOT = Path(__file__).resolve().parents[3]
FIXTURE = REPO_ROOT / "docs/evaluations/ai_agent_result_capture_owner_approved_execution_rehearsal_2026-06-14.json"
def test_load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal_snapshot() -> None:
data = load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal()
assert data["schema_version"] == "ai_agent_result_capture_owner_approved_execution_rehearsal_v1"
assert data["program_status"]["current_task_id"] == "P2-126"
assert data["program_status"]["next_task_id"] == "P2-127"
assert data["program_status"]["runtime_authority"] == "result_capture_owner_approved_execution_rehearsal_only_no_live_write"
rollups = data["rollups"]
assert rollups["execution_rehearsal_count"] == 5
assert rollups["no_write_apply_check_count"] == 5
assert rollups["post_write_verifier_rehearsal_count"] == 5
assert rollups["rollback_drill_count"] == 5
assert rollups["blocked_live_apply_count"] == 6
assert rollups["operator_action_count"] == 5
assert rollups["critical_blocker_count"] == 5
assert rollups["owner_approval_record_received_count"] == 0
assert rollups["no_write_apply_execution_count"] == 0
assert rollups["post_write_verifier_pass_count"] == 0
assert rollups["rollback_drill_confirmed_count"] == 0
assert rollups["writer_apply_count"] == 0
assert rollups["execution_apply_count"] == 0
assert rollups["receipt_write_count"] == 0
assert rollups["gateway_queue_write_count"] == 0
assert rollups["telegram_send_count"] == 0
assert rollups["bot_api_call_count"] == 0
assert rollups["result_capture_write_count"] == 0
assert rollups["learning_write_count"] == 0
assert rollups["playbook_trust_write_count"] == 0
assert rollups["production_write_count"] == 0
assert {item["no_write_only"] for item in data["execution_rehearsals"]} == {True}
assert {item["live_apply_enabled"] for item in data["execution_rehearsals"]} == {False}
assert {item["live_read_enabled"] for item in data["post_write_verifier_rehearsals"]} == {False}
assert {item["rollback_drill_confirmed"] for item in data["rollback_drills"]} == {False}
def _copy_fixture(tmp_path: Path) -> dict:
source = json.loads(FIXTURE.read_text(encoding="utf-8"))
target = tmp_path / "ai_agent_result_capture_owner_approved_execution_rehearsal_2026-06-14.json"
target.write_text(json.dumps(source, ensure_ascii=False), encoding="utf-8")
return source
def test_result_capture_owner_approved_execution_rehearsal_rejects_live_apply_enabled(tmp_path: Path) -> None:
source = _copy_fixture(tmp_path)
source["execution_rehearsals"][0]["live_apply_enabled"] = True
target = tmp_path / "ai_agent_result_capture_owner_approved_execution_rehearsal_2026-06-14.json"
target.write_text(json.dumps(source, ensure_ascii=False), encoding="utf-8")
with pytest.raises(ValueError, match="must not enable live apply"):
load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal(tmp_path)
def test_result_capture_owner_approved_execution_rehearsal_rejects_truth_write_flag(tmp_path: Path) -> None:
source = _copy_fixture(tmp_path)
source["execution_rehearsal_truth"]["result_capture_write_enabled"] = True
target = tmp_path / "ai_agent_result_capture_owner_approved_execution_rehearsal_2026-06-14.json"
target.write_text(json.dumps(source, ensure_ascii=False), encoding="utf-8")
with pytest.raises(ValueError, match="live execution/send/write flags"):
load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal(tmp_path)
def test_result_capture_owner_approved_execution_rehearsal_rejects_rollup_drift(tmp_path: Path) -> None:
source = _copy_fixture(tmp_path)
source["rollups"]["execution_rehearsal_count"] = 4
target = tmp_path / "ai_agent_result_capture_owner_approved_execution_rehearsal_2026-06-14.json"
target.write_text(json.dumps(source, ensure_ascii=False), encoding="utf-8")
with pytest.raises(ValueError, match="rollup counts mismatch"):
load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal(tmp_path)
def test_result_capture_owner_approved_execution_rehearsal_rejects_forbidden_display_terms(tmp_path: Path) -> None:
source = _copy_fixture(tmp_path)
source["operator_actions"][0]["operator_instruction"] = "work_window_transcript must never show"
target = tmp_path / "ai_agent_result_capture_owner_approved_execution_rehearsal_2026-06-14.json"
target.write_text(json.dumps(source, ensure_ascii=False), encoding="utf-8")
with pytest.raises(ValueError, match="forbidden display terms"):
load_latest_ai_agent_result_capture_owner_approved_execution_rehearsal(tmp_path)

View File

@@ -0,0 +1,40 @@
from __future__ import annotations
import pytest
from httpx import ASGITransport, AsyncClient
from src.main import app
@pytest.mark.asyncio
async def test_get_agent_result_capture_owner_approved_execution_rehearsal_api() -> None:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
response = await client.get("/api/v1/agents/agent-result-capture-owner-approved-execution-rehearsal")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "ai_agent_result_capture_owner_approved_execution_rehearsal_v1"
assert data["program_status"]["current_task_id"] == "P2-126"
assert data["program_status"]["next_task_id"] == "P2-127"
rollups = data["rollups"]
assert rollups["execution_rehearsal_count"] == 5
assert rollups["no_write_apply_check_count"] == 5
assert rollups["post_write_verifier_rehearsal_count"] == 5
assert rollups["rollback_drill_count"] == 5
assert rollups["blocked_live_apply_count"] == 6
assert rollups["operator_action_count"] == 5
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["bot_api_call_count"] == 0
assert rollups["production_write_count"] == 0
assert {item["no_write_only"] for item in data["execution_rehearsals"]} == {True}
assert {item["live_apply_enabled"] for item in data["execution_rehearsals"]} == {False}
assert {item["live_read_enabled"] for item in data["post_write_verifier_rehearsals"]} == {False}