feat(agents): expose autonomous runtime control
Some checks failed
CD Pipeline / tests (push) Successful in 1m43s
CD Pipeline / post-deploy-checks (push) Has been cancelled
CD Pipeline / build-and-deploy (push) Has been cancelled
Code Review / ai-code-review (push) Has been cancelled
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled

This commit is contained in:
Your Name
2026-06-27 19:31:02 +08:00
parent d359587316
commit ce0c7cbaf8
13 changed files with 1017 additions and 38 deletions

View File

@@ -0,0 +1,63 @@
from src.services.ai_agent_autonomous_runtime_control import (
build_ai_agent_autonomous_runtime_control,
)
def test_ai_agent_autonomous_runtime_control_uses_current_owner_directive():
data = build_ai_agent_autonomous_runtime_control()
assert data["schema_version"] == "ai_agent_autonomous_runtime_control_v1"
assert data["program_status"]["runtime_authority"] == (
"current_owner_directive_controlled_ai_automation"
)
assert data["program_status"]["legacy_no_send_no_live_rules_overridden"] is True
assert data["current_policy"]["low_risk_controlled_apply_allowed"] is True
assert data["current_policy"]["medium_risk_controlled_apply_allowed"] is True
assert data["current_policy"]["high_risk_controlled_apply_allowed"] is True
assert data["current_policy"]["owner_review_required_for_low_medium_high"] is False
assert data["current_policy"]["telegram_gateway_required"] is True
assert data["current_policy"]["direct_bot_api_allowed"] is False
assert data["current_policy"]["post_apply_verifier_required"] is True
assert data["current_policy"]["km_learning_writeback_required"] is True
def test_ai_agent_autonomous_runtime_control_exposes_reports_and_executor_receipts():
data = build_ai_agent_autonomous_runtime_control()
cadences = {item["cadence"]: item for item in data["report_delivery"]["cadences"]}
assert set(cadences) == {"daily", "weekly", "monthly"}
assert {item["telegram_gateway_delivery_enabled"] for item in cadences.values()} == {True}
assert {item["direct_bot_api_allowed"] for item in cadences.values()} == {False}
assert "run_daily_report_loop" in cadences["daily"]["worker"]
assert "run_weekly_report_loop" in cadences["weekly"]["worker"]
assert "run_monthly_report_loop" in cadences["monthly"]["worker"]
operation_types = {
item["operation_type"]
for item in data["controlled_executor"]["operation_receipts"]
}
assert {
"ansible_candidate_matched",
"ansible_check_mode_executed",
"ansible_apply_executed",
"incident_evidence.post_execution_state",
"knowledge_entries",
}.issubset(operation_types)
assert data["rollups"]["automated_risk_tier_count"] == 3
assert data["rollups"]["report_cadence_enabled_count"] == 3
assert data["rollups"]["direct_bot_api_allowed_count"] == 0
assert data["rollups"]["legacy_policy_overridden_count"] >= 4
def test_ai_agent_autonomous_runtime_control_keeps_hard_blockers_and_redaction():
data = build_ai_agent_autonomous_runtime_control()
assert "secret_token_private_key_cookie_session_auth_header_cleartext" in data["hard_blockers"]
assert "drop_truncate_restore_prune_destructive_database_operation" in data["hard_blockers"]
assert "force_push_delete_repo_refs_or_visibility_change" in data["hard_blockers"]
visibility = data["visibility_contract"]
assert visibility["work_window_transcript_display_allowed"] is False
assert visibility["prompt_body_display_allowed"] is False
assert visibility["internal_reasoning_display_allowed"] is False
assert visibility["sensitive_value_display_allowed"] is False
assert visibility["telegram_unredacted_payload_display_allowed"] is False

View File

@@ -0,0 +1,66 @@
from fastapi.testclient import TestClient
from src.main import app
_PUBLIC_FORBIDDEN_TERMS = [
"工作視窗",
"對話內容",
"批准!繼續",
"In app browser",
"My request for Codex",
"browser_context",
"codex_user_message",
"prompt_text",
"raw prompt",
"raw_prompt",
"raw payload",
"raw_payload",
"private reasoning",
"chain_of_thought",
"authorization header",
"authorization_header",
"secret value",
"secret_value",
]
def _collect_strings(value):
if isinstance(value, str):
return [value]
if isinstance(value, list):
strings = []
for item in value:
strings.extend(_collect_strings(item))
return strings
if isinstance(value, dict):
strings = []
for item in value.values():
strings.extend(_collect_strings(item))
return strings
return []
def test_get_ai_agent_autonomous_runtime_control_api():
client = TestClient(app)
response = client.get("/api/v1/agents/agent-autonomous-runtime-control")
assert response.status_code == 200
data = response.json()
assert data["schema_version"] == "ai_agent_autonomous_runtime_control_v1"
assert data["program_status"]["runtime_authority"] == (
"current_owner_directive_controlled_ai_automation"
)
assert data["current_policy"]["owner_review_required_for_low_medium_high"] is False
assert data["report_delivery"]["status"] == "telegram_gateway_delivery_enabled"
assert data["rollups"]["report_cadence_enabled_count"] == 3
def test_get_ai_agent_autonomous_runtime_control_api_redacts_public_terms():
client = TestClient(app)
response = client.get("/api/v1/agents/agent-autonomous-runtime-control")
assert response.status_code == 200
all_text = "\n".join(_collect_strings(response.json()))
for term in _PUBLIC_FORBIDDEN_TERMS:
assert term not in all_text

View File

@@ -31,6 +31,8 @@ from src.services.report_generation_service import (
PostmortemData,
ReportGenerationService,
_seconds_until_next_report,
_seconds_until_next_monthly_report,
_seconds_until_next_weekly_report,
)
from src.services.weekly_report_service import WeeklyReportService
@@ -426,8 +428,8 @@ class TestFormatDailyReport:
assert "全 0 判讀: source_gap_or_no_signal_requires_review" in report
assert "不自動改排程" in report
def test_monthly_preview_contains_no_send_source_health(self):
"""月報 preview 應顯示 no-send 邊界與資產沉澱"""
def test_monthly_report_contains_telegram_gateway_source_health(self):
"""月報應顯示 Telegram Gateway 派送與資產沉澱"""
source_health = {
"rollups": {
"source_ok_count": 2,
@@ -452,19 +454,24 @@ class TestFormatDailyReport:
],
}
svc = ReportGenerationService()
report = svc.format_monthly_report_preview(
report = svc.format_monthly_report(
source_health,
generated_at=datetime(2026, 6, 18, 10, 0, tzinfo=_TZ_TAIPEI),
)
assert "月報 no-send preview" in report
assert "AWOOOI 月報" in report
assert "Owner: Hermes" in report
assert "實發: 0" in report
assert "Telegram Gateway" in report
assert "來源: <code>2/5</code>" in report
assert "resolution_stats" in report
assert "KM: draft_ready 3/4" in report
assert "Verifier: source_health_ready 1/2" in report
assert "不代表已授權發送或自動修復" in report
assert "AI Agent 受控接手" in report
def test_weekly_and_monthly_report_schedule_helpers_return_positive_seconds(self):
assert _seconds_until_next_report() > 0
assert _seconds_until_next_weekly_report() > 0
assert _seconds_until_next_monthly_report() > 0
def test_sre_digest_preview_contains_assets_and_boundaries(self):
"""SRE 戰情室 digest 應收斂缺口、資產與 no-send 邊界"""

View File

@@ -50,7 +50,7 @@ def test_weekly_report_preview_exposes_source_health_no_send_preview():
assert "不自動改排程" in preview
def test_monthly_report_preview_exposes_source_health_no_send_preview():
def test_monthly_report_preview_exposes_source_health_and_gateway_delivery():
client = TestClient(app)
response = client.get("/api/v1/stats/monthly/preview")
@@ -65,11 +65,11 @@ def test_monthly_report_preview_exposes_source_health_no_send_preview():
assert "formatted_preview" in data
preview = data["formatted_preview"]
assert "月報 no-send preview" in preview
assert "AWOOOI 月報" in preview
assert "報表資料源 / 沉澱" in preview
assert f"來源: <code>{data['source_ok_count']}/{data['source_total_count']}</code>" in preview
assert "實發: 0" in preview
assert "不代表已授權發送或自動修復" in preview
assert "Telegram Gateway" in preview
assert "AI Agent 受控接手" in preview
def test_sre_digest_preview_exposes_source_health_no_send_preview():