diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 6fe07eeaf..9460c8f89 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -20,10 +20,11 @@ from pathlib import Path from typing import Any import structlog -from sqlalchemy import text +from sqlalchemy import select, text from src.core.config import settings from src.db.base import get_db_context +from src.db.models import PlaybookRecord from src.services.ai_automation_runtime_contract import ( AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION, ) @@ -36,6 +37,7 @@ from src.services.awooop_ansible_learning_writeback import ( from src.services.awooop_ansible_post_verifier import ( run_ansible_asset_post_verifier, ) +from src.services.playbook_evolver import TRUST_ARCHIVE_THRESHOLD logger = structlog.get_logger(__name__) @@ -43,6 +45,9 @@ _SAFE_HOST_RE = re.compile(r"^[A-Za-z0-9_.-]+$") _PLAYBOOK_PREFIX = Path("infra/ansible/playbooks") _STDOUT_LIMIT = 20_000 _STDERR_LIMIT = 12_000 +_CONTROLLED_RETRY_MAX_ATTEMPTS = 1 +_CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS = 15 +_CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS = 30 _EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300 _EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800 _EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS = 5 @@ -215,6 +220,181 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]: raise ValueError("no_safe_check_mode_candidate") +def _apply_ansible_playbook_trust_gate( + candidate_input: dict[str, Any], + trust_rows: list[Mapping[str, Any]], +) -> tuple[dict[str, Any], dict[str, Any]]: + """Filter failed PlayBooks before they can claim another runtime apply.""" + + candidates = candidate_input.get("executor_candidates") + if not isinstance(candidates, list): + candidates = [] + trust_by_id = { + str(row.get("playbook_id") or ""): row + for row in trust_rows + if str(row.get("playbook_id") or "") + } + eligible: list[dict[str, Any]] = [] + decisions: list[dict[str, Any]] = [] + for candidate in candidates: + if not isinstance(candidate, dict): + continue + catalog_id = str(candidate.get("catalog_id") or "") + canonical_id = canonical_ansible_playbook_id(catalog_id) + row = trust_by_id.get(canonical_id) + status = str((row or {}).get("status") or "").strip().lower() + trust_score = float((row or {}).get("trust_score") or 0.0) + success_count = int((row or {}).get("success_count") or 0) + failure_count = int((row or {}).get("failure_count") or 0) + retired = status in {"deprecated", "archived"} + learned_failure = bool( + row + and failure_count > success_count + and trust_score < TRUST_ARCHIVE_THRESHOLD + ) + circuit_open = retired or learned_failure + reason = ( + "playbook_status_retired" + if retired + else "playbook_trust_below_archive_threshold" + if learned_failure + else "playbook_trust_allows_controlled_canary" + if row + else "playbook_unobserved_canary_allowed" + ) + decisions.append( + { + "catalog_id": catalog_id, + "canonical_playbook_id": canonical_id, + "trust_record_present": row is not None, + "status": status or None, + "trust_score": round(trust_score, 6) if row else None, + "success_count": success_count, + "failure_count": failure_count, + "review_required": bool((row or {}).get("review_required")), + "circuit_open": circuit_open, + "reason": reason, + } + ) + if not circuit_open: + eligible.append(candidate) + + blocked_count = sum(1 for item in decisions if item["circuit_open"]) + status = ( + "circuit_open" + if decisions and not eligible + else "degraded_candidates_filtered" + if blocked_count + else "closed" + ) + gate = { + "schema_version": "ansible_playbook_trust_claim_gate_v1", + "policy_source": "playbook_evolver.TRUST_ARCHIVE_THRESHOLD", + "trust_archive_threshold": TRUST_ARCHIVE_THRESHOLD, + "status": status, + "candidate_count": len(decisions), + "eligible_candidate_count": len(eligible), + "blocked_candidate_count": blocked_count, + "owner_review_used_as_terminal": False, + "next_controlled_action": ( + "generate_or_verify_playbook_repair_candidate" + if status == "circuit_open" + else "continue_with_highest_ranked_eligible_candidate" + ), + "decisions": decisions, + } + return ( + { + **candidate_input, + "executor_candidates": eligible, + "playbook_trust_gate": gate, + }, + gate, + ) + + +async def _guard_candidate_input_by_playbook_trust( + db: Any, + candidate_input: dict[str, Any], +) -> tuple[dict[str, Any], dict[str, Any]]: + candidates = candidate_input.get("executor_candidates") + canonical_ids = sorted( + { + canonical_ansible_playbook_id(str(candidate.get("catalog_id") or "")) + for candidate in candidates or [] + if isinstance(candidate, dict) and candidate.get("catalog_id") + } + ) + trust_rows: list[Mapping[str, Any]] = [] + if canonical_ids: + result = await db.execute( + select( + PlaybookRecord.playbook_id, + PlaybookRecord.status, + PlaybookRecord.trust_score, + PlaybookRecord.success_count, + PlaybookRecord.failure_count, + PlaybookRecord.review_required, + ).where(PlaybookRecord.playbook_id.in_(canonical_ids)) + ) + trust_rows = list(result.mappings().all()) + return _apply_ansible_playbook_trust_gate(candidate_input, trust_rows) + + +async def _revalidate_claim_playbook_trust( + db: Any, + claim: AnsibleCheckModeClaim, +) -> bool: + candidate_input = { + **claim.input_payload, + "executor_candidates": [ + { + "catalog_id": claim.catalog_id, + "playbook_path": claim.apply_playbook_path, + "check_mode_playbook_path": claim.playbook_path, + "inventory_hosts": list(claim.inventory_hosts), + "risk_level": claim.risk_level, + } + ], + } + _, gate = await _guard_candidate_input_by_playbook_trust( + db, + candidate_input, + ) + claim.input_payload["playbook_trust_gate"] = gate + if gate.get("status") != "circuit_open": + return True + claim.input_payload["controlled_apply_allowed"] = False + claim.input_payload["controlled_apply_blocker"] = ( + "playbook_trust_circuit_open" + ) + await db.execute( + text(""" + UPDATE automation_operation_log + SET input = coalesce(input, '{}'::jsonb) || jsonb_build_object( + 'controlled_apply_allowed', false, + 'controlled_apply_blocker', 'playbook_trust_circuit_open', + 'playbook_trust_gate', CAST(:trust_gate AS jsonb) + ), + dry_run_result = coalesce( + dry_run_result, + '{}'::jsonb + ) || jsonb_build_object( + 'controlled_apply_allowed', false, + 'controlled_apply_blocker', 'playbook_trust_circuit_open', + 'playbook_trust_gate', CAST(:trust_gate AS jsonb) + ) + WHERE op_id = CAST(:op_id AS uuid) + AND operation_type = 'ansible_check_mode_executed' + """), + { + "trust_gate": json.dumps(gate, ensure_ascii=False), + "op_id": claim.op_id, + }, + ) + return False + + def _allowed_controlled_apply_risks() -> set[str]: raw = str(settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS or "") return {item.strip().lower() for item in raw.split(",") if item.strip()} @@ -276,6 +456,7 @@ def build_ansible_check_mode_claim_input( "target_selector": candidate_input.get("target_selector") or {}, "source_truth_diff": candidate_input.get("source_truth_diff") or {}, "risk_policy_decision": candidate_input.get("risk_policy_decision") or {}, + "playbook_trust_gate": candidate_input.get("playbook_trust_gate") or {}, "source_candidate_op_id": source_candidate_op_id, "catalog_id": safe["catalog_id"], "playbook_path": safe["playbook_path"], @@ -1641,34 +1822,14 @@ async def _record_runtime_stage_receipts( return False -async def _record_retry_runtime_stage_receipt( - claim: AnsibleCheckModeClaim, - result: AnsibleRunResult, +async def _append_runtime_stage_receipts_to_apply( *, apply_op_id: str, - replay_op_id: str, + receipts: tuple[dict[str, Any], ...], project_id: str, ) -> bool: - """Append the verified no-write retry terminal to the original apply run.""" - - receipt = _runtime_stage_receipt( - claim, - stage_id="retry_or_rollback", - evidence_ref=f"automation_operation_log:{replay_op_id}:dry_run_result", - detail={ - "failed_apply_op_id": apply_op_id, - "retry_check_mode_op_id": replay_op_id, - "terminal_type": ( - "no_write_replay_passed_waiting_repair_candidate" - if result.returncode == 0 - else "no_write_replay_failed_waiting_playbook_or_transport_repair" - ), - "check_mode_replay_performed": True, - "check_mode_replay_returncode": result.returncode, - "runtime_apply_executed": False, - "rollback_performed": False, - }, - ) + if not receipts: + return False try: async with get_db_context(project_id) as db: updated = await db.execute( @@ -1679,40 +1840,327 @@ async def _record_retry_runtime_stage_receipt( '{runtime_stage_receipts}', coalesce( ( - SELECT jsonb_agg(existing_receipt) - FROM jsonb_array_elements( - coalesce( - apply.input -> 'runtime_stage_receipts', - '[]'::jsonb + SELECT jsonb_agg( + deduplicated.receipt + ORDER BY deduplicated.stage_id + ) + FROM ( + SELECT DISTINCT ON ( + merged.receipt ->> 'stage_id' ) - ) AS existing_receipt - WHERE existing_receipt ->> 'stage_id' - <> 'retry_or_rollback' + merged.receipt, + merged.receipt ->> 'stage_id' AS stage_id + FROM jsonb_array_elements( + coalesce( + apply.input -> 'runtime_stage_receipts', + '[]'::jsonb + ) || CAST(:receipts AS jsonb) + ) WITH ORDINALITY AS merged(receipt, position) + WHERE merged.receipt ->> 'stage_id' IS NOT NULL + ORDER BY + merged.receipt ->> 'stage_id', + merged.position DESC + ) deduplicated ), '[]'::jsonb - ) || jsonb_build_array(CAST(:receipt AS jsonb)), + ), true ) WHERE apply.op_id = CAST(:apply_op_id AS uuid) AND apply.operation_type = 'ansible_apply_executed' """), { - "receipt": json.dumps(receipt, ensure_ascii=False), + "receipts": json.dumps(receipts, ensure_ascii=False), "apply_op_id": apply_op_id, }, ) return bool(updated.rowcount) except Exception as exc: logger.warning( - "ansible_retry_runtime_stage_receipt_write_failed", - automation_run_id=_automation_run_id_for_claim(claim), + "ansible_runtime_terminal_receipt_append_failed", apply_op_id=apply_op_id, - replay_op_id=replay_op_id, - error=str(exc), + error_type=type(exc).__name__, ) return False +async def _record_incident_terminal_disposition( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + project_id: str, + terminal_type: str, + success_terminal: bool, + retry_op_id: str | None = None, + telegram_receipt_acknowledged: bool, +) -> dict[str, Any] | None: + automation_run_id = _automation_run_id_for_claim(claim) + incident_status = "RESOLVED" if success_terminal else "MITIGATING" + safe_next_action = ( + "keep_verified_repair_and_monitor_recurrence" + if success_terminal + else "queue_ai_playbook_or_transport_repair_candidate" + ) + terminal_payload = { + "schema_version": "ansible_incident_terminal_disposition_v1", + "automation_run_id": automation_run_id, + "apply_op_id": apply_op_id, + "retry_op_id": retry_op_id, + "terminal_type": terminal_type, + "incident_status": incident_status, + "incident_resolved": success_terminal, + "no_write_terminal": not success_terminal, + "safe_next_action": safe_next_action, + "telegram_receipt_acknowledged": telegram_receipt_acknowledged, + "raw_log_payload_stored": False, + "secret_value_stored": False, + } + try: + async with get_db_context(project_id) as db: + updated = await db.execute( + text(""" + UPDATE incidents + SET status = CAST(:incident_status AS incidentstatus), + outcome = CAST( + coalesce(CAST(outcome AS jsonb), '{}'::jsonb) + || jsonb_build_object( + 'automation_terminal', + CAST(:terminal_payload AS jsonb) + ) + AS json + ), + resolved_at = CASE + WHEN :success_terminal THEN NOW() + ELSE resolved_at + END, + updated_at = NOW() + WHERE incident_id = :incident_id + AND project_id = :project_id + AND ( + :success_terminal + OR upper(status::text) NOT IN ('RESOLVED', 'CLOSED') + ) + RETURNING + status::text AS incident_status, + updated_at, + resolved_at, + outcome + """), + { + "incident_status": incident_status, + "terminal_payload": json.dumps( + terminal_payload, + ensure_ascii=False, + ), + "success_terminal": success_terminal, + "incident_id": claim.incident_id, + "project_id": project_id, + }, + ) + row = updated.mappings().one_or_none() + if row is None: + return None + outcome = _json_loads(row.get("outcome")) + readback = outcome.get("automation_terminal") + if not isinstance(readback, Mapping): + return None + if ( + readback.get("automation_run_id") != automation_run_id + or readback.get("apply_op_id") != apply_op_id + or readback.get("terminal_type") != terminal_type + ): + return None + return { + **terminal_payload, + "incident_id": claim.incident_id, + "incident_status": str(row.get("incident_status") or ""), + "incident_row_version": ( + row["updated_at"].isoformat() + if row.get("updated_at") is not None + else None + ), + "repository_write_acknowledged": True, + "repository_readback_verified": True, + "durable_write_acknowledged": True, + "writer_source_sha": os.getenv( + "AWOOOI_BUILD_COMMIT_SHA", + "", + ).strip().lower() + or None, + } + except Exception as exc: + logger.warning( + "ansible_incident_terminal_disposition_write_failed", + automation_run_id=automation_run_id, + incident_id=claim.incident_id, + apply_op_id=apply_op_id, + error_type=type(exc).__name__, + ) + return None + + +async def _record_retry_runtime_stage_receipt( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + replay_op_id: str, + project_id: str, +) -> bool: + """Append the verified no-write retry terminal to the original apply run.""" + + automation_run_id = _automation_run_id_for_claim(claim) + terminal_type = ( + "no_write_replay_passed_waiting_repair_candidate" + if result.returncode == 0 + else "no_write_replay_failed_waiting_playbook_or_transport_repair" + ) + retry_idempotency_key = f"ansible-retry:{apply_op_id}:check-mode" + replay_readback_verified = False + telegram_receipt_acknowledged = False + try: + async with get_db_context(project_id) as db: + replay_readback = await db.execute( + text(""" + SELECT status, input, output, dry_run_result + FROM automation_operation_log + WHERE op_id = CAST(:replay_op_id AS uuid) + AND operation_type = 'ansible_execution_skipped' + AND parent_op_id = CAST(:apply_op_id AS uuid) + """), + { + "replay_op_id": replay_op_id, + "apply_op_id": apply_op_id, + }, + ) + replay_row = replay_readback.mappings().one_or_none() + replay_input = _json_loads( + replay_row.get("input") if replay_row else None + ) + replay_output = _json_loads( + replay_row.get("output") if replay_row else None + ) + replay_readback_verified = bool( + replay_row + and replay_row.get("status") + == ("success" if result.returncode == 0 else "failed") + and replay_input.get("automation_run_id") + == automation_run_id + and replay_input.get("retry_of_apply_op_id") == apply_op_id + and replay_output.get("runtime_apply_executed") is False + and replay_output.get("terminal_disposition") == terminal_type + ) + telegram_readback = await db.execute( + text(""" + SELECT EXISTS ( + SELECT 1 + FROM awooop_outbound_message + WHERE project_id = :project_id + AND channel_type = 'telegram' + AND send_status = 'sent' + AND source_envelope #>> '{callback_reply,action}' + = 'controlled_apply_result' + AND COALESCE( + source_envelope ->> 'automation_run_id', + source_envelope #>> + '{callback_reply,automation_run_id}', + source_envelope #>> + '{source_refs,automation_run_ids,0}' + ) = :automation_run_id + ) + """), + { + "project_id": project_id, + "automation_run_id": automation_run_id, + }, + ) + telegram_receipt_acknowledged = telegram_readback.scalar() is True + except Exception as exc: + logger.warning( + "ansible_retry_terminal_readback_failed", + automation_run_id=automation_run_id, + apply_op_id=apply_op_id, + replay_op_id=replay_op_id, + error_type=type(exc).__name__, + ) + + if not replay_readback_verified: + logger.warning( + "ansible_retry_terminal_receipt_not_written_unverified", + automation_run_id=automation_run_id, + apply_op_id=apply_op_id, + replay_op_id=replay_op_id, + ) + return False + + retry_receipt = _runtime_stage_receipt( + claim, + stage_id="retry_or_rollback", + evidence_ref=f"automation_operation_log:{replay_op_id}:dry_run_result", + detail={ + "schema_version": "ansible_controlled_retry_terminal_v2", + "failed_apply_op_id": apply_op_id, + "retry_check_mode_op_id": replay_op_id, + "terminal_type": terminal_type, + "check_mode_replay_performed": True, + "check_mode_replay_returncode": result.returncode, + "runtime_apply_executed": False, + "rollback_performed": False, + "verified_no_write_terminal": replay_readback_verified, + "retry_operation_readback_verified": replay_readback_verified, + "retry_attempt": 1, + "max_retry_attempts": _CONTROLLED_RETRY_MAX_ATTEMPTS, + "retry_idempotency_key": retry_idempotency_key, + "retry_error_backoff_min_seconds": ( + _CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS + ), + "retry_error_backoff_max_seconds": ( + _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS + ), + "bounded_retry": True, + "repair_candidate_required": True, + "safe_next_action": ( + "queue_ai_playbook_or_transport_repair_candidate" + ), + "repository_readback_verified": replay_readback_verified, + "durable_write_acknowledged": replay_readback_verified, + "writer_source_sha": os.getenv( + "AWOOOI_BUILD_COMMIT_SHA", + "", + ).strip().lower() + or None, + "raw_log_payload_stored": False, + "secret_value_stored": False, + }, + ) + incident_detail = await _record_incident_terminal_disposition( + claim, + apply_op_id=apply_op_id, + project_id=project_id, + terminal_type=terminal_type, + success_terminal=False, + retry_op_id=replay_op_id, + telegram_receipt_acknowledged=telegram_receipt_acknowledged, + ) + receipts = [retry_receipt] + if incident_detail is not None: + receipts.append( + _runtime_stage_receipt( + claim, + stage_id="incident_closure", + evidence_ref=( + f"incidents:{claim.incident_id}:automation_terminal" + ), + detail=incident_detail, + ) + ) + return await _append_runtime_stage_receipts_to_apply( + apply_op_id=apply_op_id, + receipts=tuple(receipts), + project_id=project_id, + ) + + async def _record_learning_writeback_receipt( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -2750,6 +3198,18 @@ async def claim_pending_check_modes( for row in rows: source_op_id = str(row["op_id"]) candidate_input = _json_loads(row["input"]) + candidate_input, trust_gate = await _guard_candidate_input_by_playbook_trust( + db, + candidate_input, + ) + if trust_gate.get("status") == "circuit_open": + await _insert_skipped_candidate( + db, + source_candidate_op_id=source_op_id, + candidate_input=candidate_input, + reason="playbook_trust_circuit_open", + ) + continue try: claim_input = build_ansible_check_mode_claim_input( source_candidate_op_id=source_op_id, @@ -3099,6 +3559,9 @@ async def _insert_skipped_candidate( candidate_input: dict[str, Any], reason: str, ) -> None: + trust_gate = candidate_input.get("playbook_trust_gate") + if not isinstance(trust_gate, dict): + trust_gate = {} input_payload = { "automation_run_id": str(source_candidate_op_id), "incident_id": _incident_id_from_payload(candidate_input), @@ -3110,6 +3573,7 @@ async def _insert_skipped_candidate( "apply_enabled": False, "source_candidate_op_id": source_candidate_op_id, "not_used_reason": reason, + "playbook_trust_gate": trust_gate, } await db.execute( text(""" @@ -3138,12 +3602,19 @@ async def _insert_skipped_candidate( "output": json.dumps({ "not_used_reason": reason, "decision_effect": "skipped_before_runtime", + "next_controlled_action": ( + "generate_or_verify_playbook_repair_candidate" + if reason == "playbook_trust_circuit_open" + else "repair_candidate_input_before_retry" + ), + "playbook_trust_gate": trust_gate, }, ensure_ascii=False), "dry_run_result": json.dumps({ "check_mode_executed": False, "apply_executed": False, "skipped": True, "reason": reason, + "playbook_trust_gate": trust_gate, }, ensure_ascii=False), "parent_op_id": source_candidate_op_id, "tags": ["ansible", "check_mode", "skipped", "apply_locked"], @@ -3318,6 +3789,16 @@ async def run_controlled_apply_for_claim( idempotency_key=apply_idempotency_key, ) return None + if not await _revalidate_claim_playbook_trust(db, claim): + logger.warning( + "ansible_controlled_apply_blocked_by_playbook_trust", + op_id=claim.op_id, + source_candidate_op_id=claim.source_candidate_op_id, + incident_id=claim.incident_id, + catalog_id=claim.catalog_id, + blocker="playbook_trust_circuit_open", + ) + return None inserted = await db.execute( text(""" INSERT INTO automation_operation_log ( diff --git a/apps/api/tests/test_awooop_ansible_trust_claim_gate.py b/apps/api/tests/test_awooop_ansible_trust_claim_gate.py new file mode 100644 index 000000000..591a22d2a --- /dev/null +++ b/apps/api/tests/test_awooop_ansible_trust_claim_gate.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import inspect + +from src.services.awooop_ansible_check_mode_service import ( + _apply_ansible_playbook_trust_gate, + _revalidate_claim_playbook_trust, + run_controlled_apply_for_claim, +) + + +def _candidate(catalog_id: str) -> dict[str, object]: + return { + "catalog_id": catalog_id, + "playbook_path": f"infra/ansible/playbooks/{catalog_id.split(':', 1)[1]}.yml", + "inventory_hosts": ["host_188"], + "risk_level": "medium", + } + + +def test_low_trust_failed_playbook_opens_claim_circuit() -> None: + guarded, gate = _apply_ansible_playbook_trust_gate( + {"executor_candidates": [_candidate("ansible:nginx-sync")]}, + [ + { + "playbook_id": "PB-ANSIBLE-NGINX-SYNC", + "status": "approved", + "trust_score": 0.001417, + "success_count": 0, + "failure_count": 24, + "review_required": False, + } + ], + ) + + assert guarded["executor_candidates"] == [] + assert gate["status"] == "circuit_open" + assert gate["blocked_candidate_count"] == 1 + assert gate["decisions"][0]["reason"] == ( + "playbook_trust_below_archive_threshold" + ) + assert gate["next_controlled_action"] == ( + "generate_or_verify_playbook_repair_candidate" + ) + + +def test_low_trust_candidate_is_filtered_without_blocking_healthy_candidate() -> None: + healthy = _candidate("ansible:188-momo-backup-user") + guarded, gate = _apply_ansible_playbook_trust_gate( + { + "executor_candidates": [ + _candidate("ansible:nginx-sync"), + healthy, + ] + }, + [ + { + "playbook_id": "PB-ANSIBLE-NGINX-SYNC", + "status": "approved", + "trust_score": 0.01, + "success_count": 0, + "failure_count": 15, + "review_required": True, + }, + { + "playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER", + "status": "approved", + "trust_score": 0.65, + "success_count": 7, + "failure_count": 1, + "review_required": False, + }, + ], + ) + + assert guarded["executor_candidates"] == [healthy] + assert gate["status"] == "degraded_candidates_filtered" + assert gate["eligible_candidate_count"] == 1 + assert gate["blocked_candidate_count"] == 1 + + +def test_owner_review_flag_does_not_become_low_medium_high_terminal_gate() -> None: + candidate = _candidate("ansible:188-momo-backup-user") + guarded, gate = _apply_ansible_playbook_trust_gate( + {"executor_candidates": [candidate]}, + [ + { + "playbook_id": "PB-ANSIBLE-188-MOMO-BACKUP-USER", + "status": "approved", + "trust_score": 0.4, + "success_count": 2, + "failure_count": 1, + "review_required": True, + } + ], + ) + + assert guarded["executor_candidates"] == [candidate] + assert gate["status"] == "closed" + assert gate["owner_review_used_as_terminal"] is False + + +def test_controlled_apply_revalidates_trust_before_runtime_write() -> None: + apply_source = inspect.getsource(run_controlled_apply_for_claim) + revalidate_source = inspect.getsource(_revalidate_claim_playbook_trust) + + assert "_revalidate_claim_playbook_trust" in apply_source + assert "ansible_controlled_apply_blocked_by_playbook_trust" in apply_source + assert "playbook_trust_circuit_open" in revalidate_source + assert "UPDATE automation_operation_log" in revalidate_source diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 440e4e18d..960992f6a 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -21,6 +21,7 @@ from src.services.awooop_ansible_check_mode_service import ( _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE, AnsibleCheckModeClaim, AnsibleRunResult, + _append_runtime_stage_receipts_to_apply, _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, _claim_from_apply_operation_row, @@ -2310,9 +2311,12 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None: assert "run_controlled_apply_for_claim" not in source receipt_source = inspect.getsource(_record_retry_runtime_stage_receipt) + append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply) assert "_runtime_stage_receipt" in receipt_source assert 'stage_id="retry_or_rollback"' in receipt_source - assert "jsonb_array_elements" in receipt_source + assert "jsonb_array_elements" in append_source + assert "ansible_controlled_retry_terminal_v2" in receipt_source + assert "ansible_retry_terminal_receipt_not_written_unverified" in receipt_source assert "runtime_apply_executed" in receipt_source diff --git a/apps/api/tests/test_telegram_polling_owner_contract.py b/apps/api/tests/test_telegram_polling_owner_contract.py new file mode 100644 index 000000000..cfe4f20bb --- /dev/null +++ b/apps/api/tests/test_telegram_polling_owner_contract.py @@ -0,0 +1,58 @@ +from pathlib import Path + +import yaml + +ROOT = Path(__file__).resolve().parents[3] + + +def _load_deployment(path: str) -> dict: + deployment_documents = list( + yaml.safe_load_all( + (ROOT / path).read_text(encoding="utf-8") + ) + ) + return next( + document + for document in deployment_documents + if document.get("kind") == "Deployment" + and document.get("metadata", {}).get("name") == "awoooi-api" + ) + + +def _api_environment(deployment: dict) -> dict[str, str | None]: + api_container = next( + container + for container in deployment["spec"]["template"]["spec"]["containers"] + if container["name"] == "api" + ) + return {item["name"]: item.get("value") for item in api_container["env"]} + + +def test_production_api_does_not_compete_with_openclaw_polling_owner() -> None: + config = yaml.safe_load( + (ROOT / "k8s/awoooi-prod/04-configmap.yaml").read_text( + encoding="utf-8" + ) + ) + deployment = _load_deployment("k8s/awoooi-prod/06-deployment-api.yaml") + main = (ROOT / "apps/api/src/main.py").read_text(encoding="utf-8") + api_env = _api_environment(deployment) + + assert config["data"]["TELEGRAM_ENABLE_POLLING"] == "false" + assert api_env["TELEGRAM_ENABLE_POLLING"] == "false" + assert "if settings.TELEGRAM_ENABLE_POLLING" in main + assert 'reason="OpenClaw' in main + + +def test_development_api_cannot_emit_legacy_telegram_heartbeats() -> None: + config = yaml.safe_load( + (ROOT / "k8s/awoooi-dev/02-configmap.yaml").read_text(encoding="utf-8") + ) + deployment = _load_deployment("k8s/awoooi-dev/04-deployment-api.yaml") + api_env = _api_environment(deployment) + + assert config["data"]["TELEGRAM_ENABLE_POLLING"] == "false" + assert api_env["TELEGRAM_ENABLE_POLLING"] == "false" + # An explicit env value overrides the same key inherited from envFrom. + # This also stops old images whose heartbeat gate only checks this token. + assert api_env["OPENCLAW_TG_BOT_TOKEN"] == "" diff --git a/k8s/awoooi-dev/04-deployment-api.yaml b/k8s/awoooi-dev/04-deployment-api.yaml index e4db61ec5..e4282e7af 100644 --- a/k8s/awoooi-dev/04-deployment-api.yaml +++ b/k8s/awoooi-dev/04-deployment-api.yaml @@ -39,6 +39,13 @@ spec: - secretRef: name: awoooi-secrets env: + # OpenClaw owns Telegram ingress/outbound monitoring. Keep these + # explicit so legacy dev images cannot inherit the bot token from + # the monolithic awoooi-secrets envFrom and emit heartbeat noise. + - name: TELEGRAM_ENABLE_POLLING + value: "false" + - name: OPENCLAW_TG_BOT_TOKEN + value: "" - name: PROMETHEUS_MULTIPROC_DIR value: "/tmp/awoooi-prometheus-multiproc" volumeMounts: diff --git a/k8s/awoooi-prod/04-configmap.yaml b/k8s/awoooi-prod/04-configmap.yaml index f4b348df7..e18036039 100644 --- a/k8s/awoooi-prod/04-configmap.yaml +++ b/k8s/awoooi-prod/04-configmap.yaml @@ -98,9 +98,10 @@ data: # ============================================================================ SHADOW_MODE_ENABLED: "false" SHADOW_MODE_LOG_ONLY: "false" - # 2026-04-03 ogt: 改為 true — clawbot-v5 已設 STANDBY_MODE，AWOOOI API 接管 Polling - # Phase 22.6: 統帥需要直接在同一 Bot 與 OpenClaw/NemoClaw 雙 AI 對話 - TELEGRAM_ENABLE_POLLING: "true" + # The former API-owned polling rule is superseded_by=global_product_governance_v2. + # 188/OpenClaw owns getUpdates. A second poller receives Telegram 409 and + # can consume commands before the Agent99 bridge records a receipt. + TELEGRAM_ENABLE_POLLING: "false" # ============================================================================ # 2026-04-03 ogt: SRE 戰情室群組三頭政治 (Triumvirate ADR-053) diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index bde9ee69f..cfe931756 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -187,7 +187,8 @@ spec: - name: NEMOTRON_TIMEOUT_SECONDS value: "55" - name: TELEGRAM_ENABLE_POLLING - value: "true" + # 188/OpenClaw is the single getUpdates owner. + value: "false" - name: OLLAMA_URL value: "http://192.168.0.110:11435" # 2026-05-25 Codex: GCP-A via 110 proxy; health cooldown protects noisy offline probes - name: OLLAMA_SECONDARY_URL