Merge remote-tracking branch 'gitea/main' into codex/github-backup-missing-targets-20260627

apps/api/tests/test_ai_agent_autonomous_runtime_control.py

@@ -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

apps/api/tests/test_ai_agent_autonomous_runtime_control_api.py

@@ -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

apps/api/tests/test_iwooos_wazuh_manager_registry_reviewer_validation.py

@@ -6,15 +6,78 @@ from fastapi.testclient import TestClient
 from src.api.v1.iwooos import router
 from src.services.iwooos_wazuh_manager_registry_reviewer_validation import (
     load_latest_iwooos_wazuh_manager_registry_reviewer_validation,
+    validate_iwooos_wazuh_manager_registry_owner_export,
 )
 
 
+EXPECTED_ALIASES = [
+    "managed_core_node_a",
+    "managed_core_node_b",
+    "managed_dev_node_a",
+    "managed_dev_node_b",
+    "managed_control_node_a",
+    "managed_control_node_b",
+]
+
+
 def _client() -> TestClient:
     app = FastAPI()
     app.include_router(router)
     return TestClient(app)
 
 
+def _valid_owner_export() -> dict:
+    return {
+        "owner_role": "資安負責人",
+        "team": "IwoooS",
+        "decision": "accept_readonly_registry_export_for_reviewer_validation",
+        "decision_reason": "Wazuh manager registry 已脫敏，供 reviewer 驗證主機覆蓋與 Dashboard API 修復狀態。",
+        "affected_scope": "iwooos_wazuh_expected_scope_aliases",
+        "collection_method": "owner_export_redacted_summary",
+        "agent_total": 6,
+        "agent_active": 2,
+        "agent_disconnected": 3,
+        "agent_never_connected": 1,
+        "last_seen_window_start": "2026-06-27T15:00:00+08:00",
+        "last_seen_window_end": "2026-06-27T16:00:00+08:00",
+        "registry_collected_at": "2026-06-27T16:05:00+08:00",
+        "registry_export_scope_aliases": EXPECTED_ALIASES,
+        "per_host_registry_matrix": [
+            {
+                "node_alias": alias,
+                "scope_role": "managed_scope",
+                "registry_presence": "present",
+                "agent_status_bucket": "active" if index < 2 else "disconnected",
+                "last_seen_state": "within_owner_window" if index < 2 else "outside_owner_window",
+                "manager_group_ref": f"group-ref-{index}",
+                "agent_id_redacted_ref": f"agent-redacted-ref-{index}",
+                "gap_reason": "none" if index < 2 else "owner_followup_required",
+                "redacted_evidence_ref": f"evidence-ref-{index}",
+            }
+            for index, alias in enumerate(EXPECTED_ALIASES)
+        ],
+        "registry_gap_reason_by_alias": {
+            alias: "none" if index < 2 else "owner_followup_required"
+            for index, alias in enumerate(EXPECTED_ALIASES)
+        },
+        "registry_export_summary_ref": "evidence-ref-registry-summary",
+        "manager_health_ref": "evidence-ref-manager-health",
+        "dashboard_api_status_ref": "evidence-ref-dashboard-api",
+        "dashboard_api_connection_check_status": "ok",
+        "dashboard_api_version_check_status": "ok",
+        "dashboard_index_pattern_statuses": ["alerts_ok", "monitoring_ok", "statistics_ok"],
+        "dashboard_api_degradation_root_cause": "stored_api_connection_repaired_by_owner",
+        "dashboard_api_repair_postcheck_ref": "evidence-ref-dashboard-postcheck",
+        "redacted_evidence_refs": [
+            "evidence-ref-registry-summary",
+            "evidence-ref-dashboard-postcheck",
+        ],
+        "followup_owner": "IwoooS reviewer",
+        "rollback_owner": "IwoooS runtime owner",
+        "postcheck_plan": "post_enable_iwooos_readback_no_raw_payload",
+    }
+
+
 def test_iwooos_wazuh_manager_registry_reviewer_validation_contract_is_waiting_only() -> None:
     payload = load_latest_iwooos_wazuh_manager_registry_reviewer_validation()
 
@@ -94,3 +157,82 @@ def test_iwooos_wazuh_manager_registry_reviewer_validation_api_is_public_safe()
     assert "source_thread_id" not in response.text
     assert "owenhytsai/" not in response.text
     assert "WAZUH_API_PASSWORD" not in response.text
+
+
+def test_iwooos_wazuh_manager_registry_owner_export_validation_accepts_redacted_payload() -> None:
+    payload = validate_iwooos_wazuh_manager_registry_owner_export(_valid_owner_export())
+
+    assert payload["schema_version"] == "iwooos_wazuh_manager_registry_owner_export_validation_result_v1"
+    assert payload["status"] == "accepted_for_readonly_posture_only"
+    assert payload["accepted_for_readonly_posture_only"] is True
+    assert payload["reviewer_validation_passed"] is True
+    assert payload["summary"]["owner_registry_export_received_count"] == 1
+    assert payload["summary"]["owner_registry_export_accepted_count"] == 1
+    assert payload["summary"]["manager_registry_accepted_count"] == 0
+    assert payload["summary"]["runtime_gate_count"] == 0
+    assert payload["boundaries"]["payload_persisted"] is False
+    assert payload["boundaries"]["runtime_execution_authorized"] is False
+    assert payload["boundaries"]["manager_registry_accepted_updated"] is False
+    assert all(slot["received"] is True for slot in payload["evidence_slots"])
+    assert all(slot["accepted"] is True for slot in payload["evidence_slots"])
+
+
+def test_iwooos_wazuh_manager_registry_owner_export_validation_api_does_not_update_global_counters() -> None:
+    client = _client()
+    response = client.post(
+        "/api/v1/iwooos/wazuh-manager-registry-reviewer-validation/validate-owner-export",
+        json=_valid_owner_export(),
+    )
+
+    assert response.status_code == 200
+    result = response.json()
+    assert result["status"] == "accepted_for_readonly_posture_only"
+    assert result["summary"]["owner_registry_export_accepted_count"] == 1
+    assert result["summary"]["manager_registry_accepted_count"] == 0
+    assert result["summary"]["runtime_gate_count"] == 0
+
+    readback = client.get("/api/v1/iwooos/wazuh-manager-registry-reviewer-validation").json()
+    assert readback["summary"]["owner_registry_export_received_count"] == 0
+    assert readback["summary"]["owner_registry_export_accepted_count"] == 0
+    assert readback["summary"]["manager_registry_accepted_count"] == 0
+    assert readback["summary"]["runtime_gate_count"] == 0
+
+
+def test_iwooos_wazuh_manager_registry_owner_export_validation_requests_missing_fields() -> None:
+    candidate = _valid_owner_export()
+    candidate.pop("decision_reason")
+
+    payload = validate_iwooos_wazuh_manager_registry_owner_export(candidate)
+
+    assert payload["status"] == "request_missing_fields"
+    assert payload["accepted_for_readonly_posture_only"] is False
+    assert payload["summary"]["owner_registry_export_received_count"] == 1
+    assert payload["summary"]["owner_registry_export_accepted_count"] == 0
+    assert any("decision_reason" in finding["field_paths"] for finding in payload["validation_findings"])
+
+
+def test_iwooos_wazuh_manager_registry_owner_export_validation_quarantines_sensitive_payload() -> None:
+    candidate = _valid_owner_export()
+    candidate["manager_health_ref"] = "health evidence mentioned 10.250.250.250 by mistake"
+
+    payload = validate_iwooos_wazuh_manager_registry_owner_export(candidate)
+
+    assert payload["status"] == "quarantine_sensitive_payload"
+    assert payload["quarantined"] is True
+    assert payload["summary"]["reviewer_validation_quarantined_count"] == 1
+    assert payload["summary"]["owner_registry_export_accepted_count"] == 0
+    assert "10.250.250.250" not in str(payload)
+    assert any(finding["check_id"] == "RV-07" for finding in payload["validation_findings"])
+
+
+def test_iwooos_wazuh_manager_registry_owner_export_validation_rejects_runtime_action_request() -> None:
+    candidate = _valid_owner_export()
+    candidate["requested_actions"] = ["wazuh_active_response"]
+
+    payload = validate_iwooos_wazuh_manager_registry_owner_export(candidate)
+
+    assert payload["status"] == "reject_runtime_action_request"
+    assert payload["runtime_action_rejected"] is True
+    assert payload["summary"]["owner_registry_export_accepted_count"] == 0
+    assert payload["summary"]["runtime_gate_count"] == 0
+    assert any(finding["check_id"] == "RV-09" for finding in payload["validation_findings"])

apps/api/tests/test_report_generation_service.py

@@ -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 邊界"""

apps/api/tests/test_weekly_report_preview_api.py

@@ -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():