diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py
index f3b7f8cd4..d9f8884f0 100644
--- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py
+++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py
@@ -7235,6 +7235,10 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback(
)
_attach_runtime_receipt_readback(payload, readback)
payload["executor_trust_boundary"] = boundary_readback
+ _attach_effective_executor_runtime_switches(
+ payload,
+ boundary_readback=boundary_readback,
+ )
_attach_autonomous_single_writer_readback(
payload,
readback=readback,
@@ -7257,6 +7261,44 @@ async def build_ai_agent_autonomous_runtime_control_with_live_readback(
return payload
+def _attach_effective_executor_runtime_switches(
+ payload: dict[str, Any],
+ *,
+ boundary_readback: Mapping[str, Any],
+) -> None:
+ """Separate the API process flag from the dedicated broker truth."""
+
+ runtime_switches = payload.setdefault("runtime_switches", {})
+ controls = boundary_readback.get("controls")
+ if not isinstance(controls, Mapping):
+ controls = {}
+ dedicated_broker_ready = bool(
+ boundary_readback.get("status") == "verified_ready"
+ and boundary_readback.get("production_boundary_verified") is True
+ and boundary_readback.get("source_sha_matches_deployment") is True
+ and controls.get("broker_ssh_mount_present") is True
+ and controls.get("broker_ssh_egress_allowlisted") is True
+ and controls.get("broker_kubernetes_token_absent") is True
+ )
+ runtime_switches.update(
+ {
+ "ansible_check_mode_worker_scope": "api_process_only",
+ "dedicated_execution_broker_ready": dedicated_broker_ready,
+ "effective_ansible_executor_available": dedicated_broker_ready,
+ "effective_executor_readback_source": (
+ boundary_readback.get("receipt_ref")
+ if dedicated_broker_ready
+ else None
+ ),
+ "effective_executor_verified_source_sha": (
+ boundary_readback.get("verified_source_sha")
+ if dedicated_broker_ready
+ else None
+ ),
+ }
+ )
+
+
def _attach_autonomous_single_writer_readback(
payload: dict[str, Any],
*,
diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py
index 308dda8b4..b28cdfdfd 100644
--- a/apps/api/src/services/awooop_ansible_audit_service.py
+++ b/apps/api/src/services/awooop_ansible_audit_service.py
@@ -178,6 +178,163 @@ def verified_ansible_candidate_terminal_sql(candidate_alias: str) -> str:
= __AUTOMATION_RUN_ID__
)
)
+ OR EXISTS (
+ SELECT 1
+ FROM automation_operation_log failed_check
+ JOIN automation_operation_log replay_check
+ ON replay_check.parent_op_id = failed_check.parent_op_id
+ AND replay_check.operation_type = 'ansible_check_mode_executed'
+ AND replay_check.status IN ('success', 'failed')
+ AND replay_check.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay'
+ AND replay_check.input ->> 'replay_of_check_mode_op_id'
+ = failed_check.op_id::text
+ JOIN automation_operation_log terminal
+ ON terminal.parent_op_id = replay_check.op_id
+ AND terminal.operation_type = 'ansible_execution_skipped'
+ AND terminal.status IN ('success', 'failed')
+ AND terminal.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay_terminal'
+ AND terminal.output ->> 'runtime_apply_executed' = 'false'
+ WHERE failed_check.parent_op_id = __CANDIDATE__.op_id
+ AND failed_check.operation_type = 'ansible_check_mode_executed'
+ AND failed_check.status = 'failed'
+ AND failed_check.input ->> 'automation_run_id'
+ = __AUTOMATION_RUN_ID__
+ AND coalesce(
+ failed_check.incident_id::text,
+ failed_check.input ->> 'incident_id'
+ ) = __INCIDENT_ID__
+ AND replay_check.input ->> 'automation_run_id'
+ = __AUTOMATION_RUN_ID__
+ AND replay_check.input ->> 'replay_of_check_mode_op_id'
+ = failed_check.op_id::text
+ AND coalesce(
+ replay_check.incident_id::text,
+ replay_check.input ->> 'incident_id'
+ ) = __INCIDENT_ID__
+ AND terminal.input ->> 'automation_run_id'
+ = __AUTOMATION_RUN_ID__
+ AND terminal.input ->> 'retry_of_check_mode_op_id'
+ = failed_check.op_id::text
+ AND coalesce(
+ terminal.incident_id::text,
+ terminal.input ->> 'incident_id'
+ ) = __INCIDENT_ID__
+ AND (
+ SELECT COUNT(DISTINCT receipt.value ->> 'stage_id')
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) receipt(value)
+ WHERE receipt.value ->> 'stage_id' IN (
+ 'mcp_context',
+ 'service_log_evidence',
+ 'normalized_asset_identity',
+ 'source_truth_diff',
+ 'risk_policy_decision',
+ 'executor_log_projection',
+ 'retry_or_rollback',
+ 'km_playbook_writeback',
+ 'rag_writeback',
+ 'playbook_trust',
+ 'timeline_projection'
+ )
+ AND receipt.value ->> 'automation_run_id'
+ = __AUTOMATION_RUN_ID__
+ AND receipt.value ->> 'durable_receipt' = 'true'
+ ) = 11
+ AND EXISTS (
+ SELECT 1
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) receipt(value)
+ WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback'
+ AND receipt.value #>>
+ '{detail,retry_terminal_op_id}'
+ = terminal.op_id::text
+ AND receipt.value #>>
+ '{detail,verified_no_write_terminal}' = 'true'
+ AND receipt.value #>>
+ '{detail,runtime_apply_executed}' = 'false'
+ AND receipt.value #>>
+ '{detail,repository_readback_verified}' = 'true'
+ AND receipt.value ->> 'durable_receipt' = 'true'
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) receipt(value)
+ WHERE receipt.value ->> 'stage_id' = 'incident_closure'
+ AND receipt.value #>> '{detail,terminal_op_id}'
+ = terminal.op_id::text
+ AND receipt.value #>> '{detail,no_write_terminal}'
+ = 'true'
+ AND receipt.value #>>
+ '{detail,telegram_receipt_acknowledged}' = 'true'
+ AND receipt.value #>>
+ '{detail,learning_writeback_acknowledged}' = 'true'
+ AND receipt.value #>>
+ '{detail,incident_projection_readback_verified}'
+ = 'true'
+ AND receipt.value #>>
+ '{detail,projection_readback_verified}' = 'true'
+ AND receipt.value #>>
+ '{detail,repository_readback_verified}' = 'true'
+ AND receipt.value ->> 'durable_receipt' = 'true'
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM awooop_outbound_message outbound
+ WHERE outbound.channel_type = 'telegram'
+ AND outbound.send_status = 'sent'
+ AND outbound.provider_message_id IS NOT NULL
+ AND outbound.source_envelope #>>
+ '{callback_reply,action}'
+ = 'controlled_apply_result'
+ AND outbound.source_envelope #>>
+ '{callback_reply,execution_kind}'
+ = 'no_write_replay'
+ AND outbound.source_envelope #>>
+ '{callback_reply,automation_run_id}'
+ = __AUTOMATION_RUN_ID__
+ AND outbound.source_envelope #>>
+ '{callback_reply,incident_id}' = __INCIDENT_ID__
+ AND outbound.source_envelope #>>
+ '{callback_reply,apply_op_id}'
+ = terminal.op_id::text
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM alert_operation_log lifecycle
+ WHERE lifecycle.incident_id = __INCIDENT_ID__
+ AND lifecycle.event_type::text = 'EXECUTION_COMPLETED'
+ AND lifecycle.success IS FALSE
+ AND lifecycle.context ->> 'apply_op_id'
+ = terminal.op_id::text
+ AND lifecycle.context ->> 'automation_run_id'
+ = __AUTOMATION_RUN_ID__
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM alert_operation_log lifecycle
+ WHERE lifecycle.incident_id = __INCIDENT_ID__
+ AND lifecycle.event_type::text = 'TELEGRAM_RESULT_SENT'
+ AND lifecycle.context ->> 'apply_op_id'
+ = terminal.op_id::text
+ AND lifecycle.context ->> 'automation_run_id'
+ = __AUTOMATION_RUN_ID__
+ )
+ )
"""
return (
template.replace("__CANDIDATE__", candidate)
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 789890289..a83fe1465 100644
--- a/apps/api/src/services/awooop_ansible_check_mode_service.py
+++ b/apps/api/src/services/awooop_ansible_check_mode_service.py
@@ -79,6 +79,7 @@ _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS = frozenset(
"timeline_projection",
}
)
+_NO_WRITE_REPLAY_REQUIRED_STAGE_IDS = _VERIFIED_CLOSURE_REQUIRED_STAGE_IDS
FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519")
REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts")
@@ -1982,6 +1983,21 @@ async def _append_runtime_stage_receipts_to_apply(
apply_op_id: str,
receipts: tuple[dict[str, Any], ...],
project_id: str,
+) -> bool:
+ return await _append_runtime_stage_receipts_to_operation(
+ operation_id=apply_op_id,
+ operation_type="ansible_apply_executed",
+ receipts=receipts,
+ project_id=project_id,
+ )
+
+
+async def _append_runtime_stage_receipts_to_operation(
+ *,
+ operation_id: str,
+ operation_type: str,
+ receipts: tuple[dict[str, Any], ...],
+ project_id: str,
) -> bool:
if not receipts:
return False
@@ -2021,19 +2037,21 @@ async def _append_runtime_stage_receipts_to_apply(
),
true
)
- WHERE apply.op_id = CAST(:apply_op_id AS uuid)
- AND apply.operation_type = 'ansible_apply_executed'
+ WHERE apply.op_id = CAST(:operation_id AS uuid)
+ AND apply.operation_type = :operation_type
"""),
{
"receipts": json.dumps(receipts, ensure_ascii=False),
- "apply_op_id": apply_op_id,
+ "operation_id": operation_id,
+ "operation_type": operation_type,
},
)
return bool(updated.rowcount)
except Exception as exc:
logger.warning(
"ansible_runtime_terminal_receipt_append_failed",
- apply_op_id=apply_op_id,
+ apply_op_id=operation_id,
+ operation_type=operation_type,
error_type=type(exc).__name__,
)
return False
@@ -3559,6 +3577,1216 @@ async def _record_stdin_boundary_replay_terminal(
return False
+def _claim_from_stdin_boundary_terminal_row(
+ row: dict[str, Any],
+) -> AnsibleCheckModeClaim | None:
+ replay_input = _json_loads(row.get("replay_input"))
+ base = _claim_from_stale_check_mode_row(
+ {
+ "op_id": row.get("replay_op_id"),
+ "parent_op_id": row.get("source_candidate_op_id"),
+ "incident_id": row.get("incident_id"),
+ "input": replay_input,
+ }
+ )
+ if base is None:
+ return None
+ terminal_input = _json_loads(row.get("terminal_input"))
+ automation_run_id = str(
+ terminal_input.get("automation_run_id")
+ or replay_input.get("automation_run_id")
+ or base.source_candidate_op_id
+ )
+ return replace(
+ base,
+ input_payload={
+ **base.input_payload,
+ **replay_input,
+ "automation_run_id": automation_run_id,
+ "historical_stdin_boundary_replay": True,
+ "controlled_apply_allowed": False,
+ "apply_enabled": False,
+ },
+ )
+
+
+async def _record_no_write_replay_timeline_receipt(
+ claim: AnsibleCheckModeClaim,
+ result: AnsibleRunResult,
+ *,
+ terminal_op_id: str,
+ project_id: str,
+) -> dict[str, Any] | None:
+ automation_run_id = _automation_run_id_for_claim(claim)
+ description = (
+ f"automation_run_id={automation_run_id};"
+ f"terminal_op_id={terminal_op_id};catalog_id={claim.catalog_id};"
+ f"returncode={result.returncode};runtime_apply_executed=false;"
+ "terminal=no_write_stdin_boundary_replay"
+ )
+ try:
+ async with get_db_context(project_id) as db:
+ selected = await db.execute(
+ text("""
+ WITH existing AS (
+ SELECT id
+ FROM timeline_events
+ WHERE incident_id = :incident_id
+ AND actor = 'ansible_stdin_boundary_replay_worker'
+ AND description = :description
+ ORDER BY created_at DESC
+ LIMIT 1
+ ), inserted AS (
+ INSERT INTO timeline_events (
+ id, incident_id, event_type, status, title,
+ description, actor, actor_role, risk_level,
+ created_at
+ )
+ SELECT
+ gen_random_uuid()::text,
+ :incident_id,
+ 'exec',
+ 'error',
+ :title,
+ :description,
+ 'ansible_stdin_boundary_replay_worker',
+ 'ai_agent',
+ :risk_level,
+ NOW()
+ WHERE NOT EXISTS (SELECT 1 FROM existing)
+ RETURNING id
+ )
+ SELECT id FROM inserted
+ UNION ALL
+ SELECT id FROM existing
+ LIMIT 1
+ """),
+ {
+ "incident_id": claim.incident_id,
+ "title": (
+ "AI no-write replay terminal: "
+ f"{claim.catalog_id}"
+ )[:500],
+ "description": description,
+ "risk_level": str(claim.risk_level or "")[:20] or None,
+ },
+ )
+ timeline_event_id = str(selected.scalar() or "")
+ except Exception as exc:
+ logger.warning(
+ "ansible_stdin_boundary_timeline_write_failed",
+ automation_run_id=automation_run_id,
+ terminal_op_id=terminal_op_id,
+ error_type=type(exc).__name__,
+ )
+ return None
+ if not timeline_event_id:
+ return None
+ return _runtime_stage_receipt(
+ claim,
+ stage_id="timeline_projection",
+ evidence_ref=f"timeline_events:{timeline_event_id}",
+ detail={
+ "timeline_event_id": timeline_event_id,
+ "terminal_op_id": terminal_op_id,
+ "returncode": result.returncode,
+ "runtime_apply_executed": False,
+ "projection_status": "repair_required",
+ "repository_readback_verified": True,
+ },
+ derived_from_durable_chain=True,
+ )
+
+
+async def _record_no_write_replay_learning_receipts(
+ claim: AnsibleCheckModeClaim,
+ result: AnsibleRunResult,
+ *,
+ terminal_op_id: str,
+ project_id: str,
+) -> dict[str, Any]:
+ from src.models.knowledge import (
+ EntrySource,
+ EntryStatus,
+ EntryType,
+ KnowledgeEntryCreate,
+ )
+ from src.repositories.knowledge_repository import KnowledgeDBRepository
+
+ automation_run_id = _automation_run_id_for_claim(claim)
+ canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id)
+ path_type = f"ansible_no_write_replay:{terminal_op_id[:8]}"
+ receipts = await _load_pre_decision_context_runtime_stage_receipts(
+ claim,
+ project_id=project_id,
+ derived_from_durable_chain=True,
+ )
+ failed_check_mode_op_id = str(
+ claim.input_payload.get("replay_of_check_mode_op_id") or ""
+ )
+ if failed_check_mode_op_id:
+ receipts.append(
+ _build_stdin_boundary_replay_stage_receipt(
+ claim,
+ result,
+ failed_check_mode_op_id=failed_check_mode_op_id,
+ terminal_op_id=terminal_op_id,
+ )
+ )
+ receipts.extend(
+ [
+ _runtime_stage_receipt(
+ claim,
+ stage_id="normalized_asset_identity",
+ evidence_ref=f"ansible_catalog:{claim.catalog_id}",
+ detail={
+ "asset_kind": "managed_host",
+ "canonical_asset_ids": list(claim.inventory_hosts),
+ "catalog_id": claim.catalog_id,
+ "playbook_path": claim.playbook_path,
+ },
+ derived_from_durable_chain=True,
+ ),
+ _runtime_stage_receipt(
+ claim,
+ stage_id="source_truth_diff",
+ evidence_ref=(
+ f"automation_operation_log:{claim.op_id}:dry_run_result"
+ ),
+ detail={
+ "check_mode_op_id": claim.op_id,
+ "check_mode_replay_performed": True,
+ "check_mode_replay_returncode": result.returncode,
+ "diff": True,
+ "runtime_apply_executed": False,
+ },
+ derived_from_durable_chain=True,
+ ),
+ _runtime_stage_receipt(
+ claim,
+ stage_id="risk_policy_decision",
+ evidence_ref=f"ansible_policy:{claim.catalog_id}",
+ detail={
+ "risk_level": str(claim.risk_level or "").lower(),
+ "controlled_apply_allowed": (
+ str(claim.risk_level or "").lower()
+ in {"low", "medium", "high"}
+ ),
+ "controlled_apply_executed": False,
+ "execution_scope": "historical_no_write_replay",
+ "apply_deferred_reason": (
+ "stdin_boundary_transport_or_playbook_repair_required"
+ ),
+ "decision": (
+ "verified_no_write_terminal_pending_ai_repair"
+ ),
+ "policy": "global_product_governance_v2",
+ },
+ derived_from_durable_chain=True,
+ ),
+ _runtime_stage_receipt(
+ claim,
+ stage_id="executor_log_projection",
+ evidence_ref=(
+ f"automation_operation_log:{terminal_op_id}:output"
+ ),
+ detail={
+ "terminal_op_id": terminal_op_id,
+ "returncode": result.returncode,
+ "duration_ms": result.duration_ms,
+ "runtime_apply_executed": False,
+ "projection_sanitized": True,
+ },
+ derived_from_durable_chain=True,
+ ),
+ ]
+ )
+
+ km_writeback: dict[str, Any] | None = None
+ try:
+ async with get_db_context(project_id) as db:
+ existing = await db.execute(
+ text("""
+ SELECT id
+ FROM knowledge_entries
+ WHERE related_incident_id = :incident_id
+ AND path_type = :path_type
+ LIMIT 1
+ """),
+ {
+ "incident_id": claim.incident_id,
+ "path_type": path_type,
+ },
+ )
+ if existing.scalar() is None:
+ repo = KnowledgeDBRepository(db)
+ await repo.create(
+ KnowledgeEntryCreate(
+ title=f"AI 無寫入重試反模式：{claim.incident_id}",
+ content=(
+ "AI Agent 已完成歷史 Ansible stdin boundary "
+ "check-mode 重試，未執行 runtime apply。\n\n"
+ f"- Automation run: {automation_run_id}\n"
+ f"- Incident: {claim.incident_id}\n"
+ f"- Catalog: {claim.catalog_id}\n"
+ f"- PlayBook: {claim.playbook_path}\n"
+ f"- Terminal operation: {terminal_op_id}\n"
+ f"- Check-mode return code: {result.returncode}\n"
+ "- Runtime apply executed: false\n"
+ "- Next action: AI transport or PlayBook repair\n"
+ ),
+ entry_type=EntryType.ANTI_PATTERN,
+ category="AI自動化/Ansible反模式",
+ tags=[
+ "ai_auto_repair",
+ "ansible_no_write_replay",
+ "stdin_boundary",
+ str(claim.catalog_id or ""),
+ f"automation_run_id:{automation_run_id}",
+ ],
+ source=EntrySource.AI_EXTRACTED,
+ status=EntryStatus.PUBLISHED,
+ related_incident_id=claim.incident_id,
+ related_playbook_id=canonical_playbook_id,
+ path_type=path_type,
+ created_by="ai_agent_ansible_replay_worker",
+ )
+ )
+ readback = await db.execute(
+ text("""
+ SELECT id, related_playbook_id, path_type, status, updated_at
+ FROM knowledge_entries
+ WHERE related_incident_id = :incident_id
+ AND path_type = :path_type
+ LIMIT 1
+ """),
+ {
+ "incident_id": claim.incident_id,
+ "path_type": path_type,
+ },
+ )
+ km_row = readback.mappings().first()
+ if (
+ km_row
+ and str(km_row.get("related_playbook_id") or "")
+ == canonical_playbook_id
+ and str(km_row.get("status") or "").lower() == "published"
+ and km_row.get("updated_at") is not None
+ ):
+ km_writeback = {
+ "schema_version": "ansible_no_write_km_writeback_v1",
+ "knowledge_entry_id": str(km_row["id"]),
+ "canonical_playbook_id": canonical_playbook_id,
+ "path_type": path_type,
+ "status": str(km_row.get("status") or ""),
+ "repository_write_acknowledged": True,
+ "repository_readback_verified": True,
+ "durable_write_acknowledged": True,
+ "raw_log_payload_stored": False,
+ "secret_value_stored": False,
+ }
+ except Exception as exc:
+ logger.warning(
+ "ansible_stdin_boundary_km_writeback_failed",
+ automation_run_id=automation_run_id,
+ terminal_op_id=terminal_op_id,
+ error_type=type(exc).__name__,
+ )
+ if km_writeback is not None:
+ receipts.append(
+ _runtime_stage_receipt(
+ claim,
+ stage_id="km_playbook_writeback",
+ evidence_ref=(
+ "knowledge_entries:"
+ f"{km_writeback['knowledge_entry_id']}:row_version"
+ ),
+ detail=km_writeback,
+ derived_from_durable_chain=True,
+ )
+ )
+
+ trust_writeback = await record_ansible_playbook_trust_writeback(
+ project_id=project_id,
+ automation_run_id=automation_run_id,
+ incident_id=claim.incident_id,
+ incident_db_id=_automation_operation_log_incident_id(
+ claim.incident_id
+ ),
+ catalog_id=claim.catalog_id,
+ playbook_path=claim.playbook_path,
+ apply_op_id=terminal_op_id,
+ verification_result="not_applicable_no_runtime_apply",
+ trust_mutation_performed=False,
+ observation_kind="no_write_replay",
+ )
+ if trust_writeback is not None:
+ receipts.append(
+ _runtime_stage_receipt(
+ claim,
+ stage_id="playbook_trust",
+ evidence_ref=(
+ "playbooks:"
+ f"{trust_writeback['canonical_playbook_id']}:trust"
+ ),
+ detail=trust_writeback,
+ derived_from_durable_chain=True,
+ )
+ )
+
+ rag_writeback = (
+ await ensure_ansible_rag_writeback(
+ project_id=project_id,
+ incident_id=claim.incident_id,
+ apply_op_id=terminal_op_id,
+ path_type=path_type,
+ )
+ if km_writeback is not None
+ else None
+ )
+ if rag_writeback is not None:
+ rag_evidence_ref = (
+ "knowledge_entries:"
+ f"{rag_writeback['knowledge_entry_id']}:embedding"
+ if rag_writeback.get("embedding_persisted") is True
+ else (
+ "rag_chunks:"
+ f"{rag_writeback.get('source_id', '')}:"
+ f"count={rag_writeback.get('rag_chunk_count', 0)}"
+ )
+ )
+ receipts.append(
+ _runtime_stage_receipt(
+ claim,
+ stage_id="rag_writeback",
+ evidence_ref=rag_evidence_ref,
+ detail=rag_writeback,
+ derived_from_durable_chain=True,
+ )
+ )
+
+ timeline_receipt = await _record_no_write_replay_timeline_receipt(
+ claim,
+ result,
+ terminal_op_id=terminal_op_id,
+ project_id=project_id,
+ )
+ if timeline_receipt is not None:
+ receipts.append(timeline_receipt)
+
+ by_stage = {
+ str(receipt.get("stage_id") or ""): receipt
+ for receipt in receipts
+ if isinstance(receipt, Mapping)
+ }
+ stage_ids = set(by_stage)
+ missing_stage_ids = sorted(
+ _NO_WRITE_REPLAY_REQUIRED_STAGE_IDS - stage_ids
+ )
+ return {
+ "ready": not missing_stage_ids,
+ "receipts": tuple(by_stage.values()),
+ "stage_ids": sorted(stage_ids),
+ "missing_stage_ids": missing_stage_ids,
+ "km_writeback": km_writeback is not None,
+ "rag_writeback": rag_writeback is not None,
+ "playbook_trust": trust_writeback is not None,
+ "mcp_context": "mcp_context" in stage_ids,
+ "runtime_apply_executed": False,
+ }
+
+
+async def _load_missing_stdin_boundary_terminal_projection_rows(
+ *,
+ project_id: str,
+ window_hours: int,
+ limit: int,
+) -> list[dict[str, Any]]:
+ async with get_db_context(project_id) as db:
+ await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
+ result = await db.execute(
+ text("""
+ SELECT
+ terminal.op_id::text AS terminal_op_id,
+ terminal.status AS terminal_status,
+ terminal.input AS terminal_input,
+ terminal.output AS terminal_output,
+ terminal.error AS terminal_error,
+ terminal.duration_ms AS terminal_duration_ms,
+ replay.op_id::text AS replay_op_id,
+ candidate.op_id::text AS source_candidate_op_id,
+ replay.input AS replay_input,
+ coalesce(
+ terminal.incident_id::text,
+ terminal.input ->> 'incident_id',
+ replay.incident_id::text,
+ replay.input ->> 'incident_id'
+ ) AS incident_id
+ FROM automation_operation_log terminal
+ JOIN automation_operation_log replay
+ ON replay.op_id = terminal.parent_op_id
+ AND replay.operation_type = 'ansible_check_mode_executed'
+ AND replay.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay'
+ AND replay.input ->> 'automation_run_id'
+ = terminal.input ->> 'automation_run_id'
+ JOIN automation_operation_log failed_check
+ ON failed_check.op_id::text
+ = replay.input ->> 'replay_of_check_mode_op_id'
+ AND failed_check.operation_type
+ = 'ansible_check_mode_executed'
+ AND failed_check.status = 'failed'
+ AND failed_check.parent_op_id = replay.parent_op_id
+ AND failed_check.input ->> 'automation_run_id'
+ = terminal.input ->> 'automation_run_id'
+ JOIN automation_operation_log candidate
+ ON candidate.op_id = replay.parent_op_id
+ AND candidate.operation_type = 'ansible_candidate_matched'
+ AND coalesce(
+ nullif(candidate.input ->> 'automation_run_id', ''),
+ candidate.op_id::text
+ ) = terminal.input ->> 'automation_run_id'
+ JOIN incidents incident
+ ON incident.incident_id = coalesce(
+ terminal.incident_id::text,
+ terminal.input ->> 'incident_id',
+ replay.incident_id::text,
+ replay.input ->> 'incident_id'
+ )
+ AND incident.project_id = :project_id
+ WHERE terminal.operation_type = 'ansible_execution_skipped'
+ AND terminal.status IN ('success', 'failed')
+ AND terminal.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay_terminal'
+ AND terminal.input ->> 'retry_of_check_mode_op_id'
+ = failed_check.op_id::text
+ AND terminal.output ->> 'runtime_apply_executed' = 'false'
+ AND terminal.created_at >= NOW() - (
+ :window_hours * INTERVAL '1 hour'
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) receipt(value)
+ WHERE receipt.value ->> 'stage_id'
+ = 'retry_or_rollback'
+ AND receipt.value ->> 'automation_run_id'
+ = terminal.input ->> 'automation_run_id'
+ AND receipt.value #>>
+ '{detail,verified_no_write_terminal}' = 'true'
+ AND receipt.value #>>
+ '{detail,runtime_apply_executed}' = 'false'
+ AND receipt.value #>>
+ '{detail,repository_readback_verified}' = 'true'
+ AND receipt.value ->> 'durable_receipt' = 'true'
+ )
+ AND NOT EXISTS (
+ SELECT 1
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) receipt(value)
+ WHERE receipt.value ->> 'stage_id' = 'incident_closure'
+ AND receipt.value ->> 'automation_run_id'
+ = terminal.input ->> 'automation_run_id'
+ AND receipt.value #>>
+ '{detail,terminal_op_id}' = terminal.op_id::text
+ AND receipt.value #>>
+ '{detail,projection_readback_verified}' = 'true'
+ AND receipt.value ->> 'durable_receipt' = 'true'
+ )
+ ORDER BY terminal.created_at ASC, terminal.op_id ASC
+ LIMIT :limit
+ """),
+ {
+ "project_id": project_id,
+ "window_hours": max(1, min(window_hours, 7 * 24)),
+ "limit": max(1, min(limit, 5)),
+ },
+ )
+ return [dict(row) for row in result.mappings().all()]
+
+
+async def _stdin_boundary_replay_telegram_receipt_acknowledged(
+ claim: AnsibleCheckModeClaim,
+ *,
+ terminal_op_id: str,
+ project_id: str,
+) -> bool:
+ automation_run_id = _automation_run_id_for_claim(claim)
+ try:
+ async with get_db_context(project_id) as db:
+ result = await db.execute(
+ text("""
+ SELECT EXISTS (
+ SELECT 1
+ FROM awooop_outbound_message outbound
+ WHERE outbound.project_id = :project_id
+ AND outbound.channel_type = 'telegram'
+ AND outbound.send_status = 'sent'
+ AND outbound.provider_message_id IS NOT NULL
+ AND outbound.source_envelope #>>
+ '{callback_reply,action}'
+ = 'controlled_apply_result'
+ AND coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,execution_kind}',
+ outbound.source_envelope ->> 'execution_kind'
+ ) = 'no_write_replay'
+ AND coalesce(
+ outbound.source_envelope
+ ->> 'automation_run_id',
+ outbound.source_envelope #>>
+ '{callback_reply,automation_run_id}',
+ outbound.source_envelope #>>
+ '{source_refs,automation_run_ids,0}'
+ ) = :automation_run_id
+ AND coalesce(
+ outbound.source_envelope #>>
+ '{callback_reply,incident_id}',
+ outbound.source_envelope #>>
+ '{source_refs,incident_ids,0}'
+ ) = :incident_id
+ AND outbound.source_envelope #>>
+ '{callback_reply,apply_op_id}'
+ = :terminal_op_id
+ )
+ """),
+ {
+ "project_id": project_id,
+ "automation_run_id": automation_run_id,
+ "incident_id": claim.incident_id,
+ "terminal_op_id": terminal_op_id,
+ },
+ )
+ return result.scalar() is True
+ except Exception as exc:
+ logger.warning(
+ "ansible_stdin_boundary_telegram_readback_failed",
+ automation_run_id=automation_run_id,
+ terminal_op_id=terminal_op_id,
+ error_type=type(exc).__name__,
+ )
+ return False
+
+
+async def _record_stdin_boundary_incident_terminal_disposition(
+ claim: AnsibleCheckModeClaim,
+ *,
+ terminal_op_id: str,
+ learning_stage_ids: list[str],
+ project_id: str,
+) -> dict[str, Any] | None:
+ automation_run_id = _automation_run_id_for_claim(claim)
+ terminal_payload = {
+ **_incident_terminal_disposition_payload(
+ claim,
+ apply_op_id=terminal_op_id,
+ terminal_type="no_write_stdin_boundary_replay_terminal",
+ success_terminal=False,
+ retry_op_id=claim.op_id,
+ telegram_receipt_acknowledged=True,
+ ),
+ "schema_version": (
+ "ansible_stdin_boundary_incident_terminal_v1"
+ ),
+ "terminal_op_id": terminal_op_id,
+ "proposal_executed": False,
+ "execution_success": None,
+ "learning_stage_ids": learning_stage_ids,
+ "learning_writeback_acknowledged": True,
+ "repository_readback_verified": True,
+ "runtime_apply_executed": False,
+ }
+ try:
+ async with get_db_context(project_id) as db:
+ selected = await db.execute(
+ text("""
+ SELECT status::text AS incident_status, outcome, updated_at
+ FROM incidents
+ WHERE incident_id = :incident_id
+ AND project_id = :project_id
+ LIMIT 1
+ FOR UPDATE
+ """),
+ {
+ "incident_id": claim.incident_id,
+ "project_id": project_id,
+ },
+ )
+ current = selected.mappings().one_or_none()
+ if current is None:
+ return None
+ current_status = str(current.get("incident_status") or "")
+ projection_superseded = current_status.upper() in {
+ "RESOLVED",
+ "CLOSED",
+ }
+ incident_write_performed = False
+ readback = current
+ if not projection_superseded:
+ updated = await db.execute(
+ text("""
+ UPDATE incidents
+ SET status = 'MITIGATING'::incidentstatus,
+ outcome = CAST(
+ coalesce(
+ CASE
+ WHEN outcome IS NULL THEN '{}'::jsonb
+ WHEN jsonb_typeof(CAST(outcome AS jsonb))
+ = 'object'
+ THEN CAST(outcome AS jsonb)
+ ELSE jsonb_build_object(
+ 'legacy_outcome',
+ CAST(outcome AS jsonb)
+ )
+ END,
+ '{}'::jsonb
+ ) || jsonb_build_object(
+ 'automation_terminal',
+ CAST(:terminal_payload AS jsonb),
+ 'proposal_executed',
+ false,
+ 'execution_success',
+ NULL
+ ) AS json
+ ),
+ updated_at = NOW()
+ WHERE incident_id = :incident_id
+ AND project_id = :project_id
+ AND upper(status::text)
+ NOT IN ('RESOLVED', 'CLOSED')
+ RETURNING
+ status::text AS incident_status,
+ outcome,
+ updated_at
+ """),
+ {
+ "terminal_payload": json.dumps(
+ terminal_payload,
+ ensure_ascii=False,
+ ),
+ "incident_id": claim.incident_id,
+ "project_id": project_id,
+ },
+ )
+ readback = updated.mappings().one_or_none()
+ incident_write_performed = readback is not None
+ if readback is None:
+ return None
+ outcome = _json_loads(readback.get("outcome"))
+ terminal_readback = outcome.get("automation_terminal")
+ if not (
+ isinstance(terminal_readback, Mapping)
+ and terminal_readback.get("automation_run_id")
+ == automation_run_id
+ and terminal_readback.get("terminal_op_id")
+ == terminal_op_id
+ and terminal_readback.get("runtime_apply_executed")
+ is False
+ ):
+ return None
+ return {
+ **terminal_payload,
+ "incident_status": str(
+ readback.get("incident_status") or current_status
+ ),
+ "incident_projection_write_performed": (
+ incident_write_performed
+ ),
+ "incident_projection_superseded": projection_superseded,
+ "incident_projection_readback_verified": True,
+ "incident_row_version": (
+ readback["updated_at"].isoformat()
+ if readback.get("updated_at") is not None
+ else None
+ ),
+ }
+ except Exception as exc:
+ logger.warning(
+ "ansible_stdin_boundary_incident_projection_failed",
+ automation_run_id=automation_run_id,
+ terminal_op_id=terminal_op_id,
+ error_type=type(exc).__name__,
+ )
+ return None
+
+
+async def _verify_stdin_boundary_terminal_projection_readback(
+ claim: AnsibleCheckModeClaim,
+ *,
+ terminal_op_id: str,
+ project_id: str,
+ require_projection_flag: bool,
+) -> bool:
+ automation_run_id = _automation_run_id_for_claim(claim)
+ try:
+ async with get_db_context(project_id) as db:
+ await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
+ result = await db.execute(
+ text("""
+ SELECT EXISTS (
+ SELECT 1
+ FROM automation_operation_log terminal
+ JOIN automation_operation_log replay
+ ON replay.op_id = terminal.parent_op_id
+ AND replay.operation_type
+ = 'ansible_check_mode_executed'
+ AND replay.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay'
+ JOIN automation_operation_log failed_check
+ ON failed_check.op_id::text
+ = replay.input
+ ->> 'replay_of_check_mode_op_id'
+ AND failed_check.operation_type
+ = 'ansible_check_mode_executed'
+ AND failed_check.status = 'failed'
+ AND failed_check.parent_op_id = replay.parent_op_id
+ AND failed_check.input ->> 'automation_run_id'
+ = :automation_run_id
+ JOIN automation_operation_log candidate
+ ON candidate.op_id = replay.parent_op_id
+ AND candidate.operation_type
+ = 'ansible_candidate_matched'
+ AND coalesce(
+ nullif(
+ candidate.input
+ ->> 'automation_run_id',
+ ''
+ ),
+ candidate.op_id::text
+ ) = :automation_run_id
+ WHERE terminal.op_id
+ = CAST(:terminal_op_id AS uuid)
+ AND terminal.operation_type
+ = 'ansible_execution_skipped'
+ AND terminal.status IN ('success', 'failed')
+ AND terminal.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay_terminal'
+ AND terminal.input ->> 'automation_run_id'
+ = :automation_run_id
+ AND terminal.input
+ ->> 'retry_of_check_mode_op_id'
+ = failed_check.op_id::text
+ AND replay.input ->> 'automation_run_id'
+ = :automation_run_id
+ AND coalesce(
+ terminal.incident_id::text,
+ terminal.input ->> 'incident_id'
+ ) = :incident_id
+ AND terminal.output ->> 'runtime_apply_executed'
+ = 'false'
+ AND (
+ SELECT COUNT(DISTINCT receipt.value
+ ->> 'stage_id')
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input
+ -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) receipt(value)
+ WHERE receipt.value ->> 'stage_id' IN (
+ 'mcp_context',
+ 'service_log_evidence',
+ 'normalized_asset_identity',
+ 'source_truth_diff',
+ 'risk_policy_decision',
+ 'executor_log_projection',
+ 'retry_or_rollback',
+ 'km_playbook_writeback',
+ 'rag_writeback',
+ 'playbook_trust',
+ 'timeline_projection'
+ )
+ AND receipt.value ->> 'automation_run_id'
+ = :automation_run_id
+ AND receipt.value ->> 'durable_receipt'
+ = 'true'
+ ) = 11
+ AND EXISTS (
+ SELECT 1
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input
+ -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) retry_receipt(value)
+ WHERE retry_receipt.value ->> 'stage_id'
+ = 'retry_or_rollback'
+ AND retry_receipt.value #>>
+ '{detail,verified_no_write_terminal}'
+ = 'true'
+ AND retry_receipt.value #>>
+ '{detail,runtime_apply_executed}'
+ = 'false'
+ AND retry_receipt.value #>>
+ '{detail,repository_readback_verified}'
+ = 'true'
+ AND retry_receipt.value
+ ->> 'durable_receipt' = 'true'
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM jsonb_array_elements(
+ coalesce(
+ terminal.input
+ -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) closure_receipt(value)
+ WHERE closure_receipt.value ->> 'stage_id'
+ = 'incident_closure'
+ AND closure_receipt.value
+ ->> 'automation_run_id'
+ = :automation_run_id
+ AND closure_receipt.value #>>
+ '{detail,terminal_op_id}'
+ = CAST(:terminal_op_id AS text)
+ AND closure_receipt.value #>>
+ '{detail,no_write_terminal}' = 'true'
+ AND closure_receipt.value #>>
+ '{detail,telegram_receipt_acknowledged}'
+ = 'true'
+ AND closure_receipt.value #>>
+ '{detail,learning_writeback_acknowledged}'
+ = 'true'
+ AND closure_receipt.value #>>
+ '{detail,incident_projection_readback_verified}'
+ = 'true'
+ AND closure_receipt.value #>>
+ '{detail,repository_readback_verified}'
+ = 'true'
+ AND (
+ :require_projection_flag IS FALSE
+ OR closure_receipt.value #>>
+ '{detail,projection_readback_verified}'
+ = 'true'
+ )
+ AND closure_receipt.value
+ ->> 'durable_receipt' = 'true'
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM awooop_outbound_message outbound
+ WHERE outbound.project_id = :project_id
+ AND outbound.channel_type = 'telegram'
+ AND outbound.send_status = 'sent'
+ AND outbound.provider_message_id IS NOT NULL
+ AND outbound.source_envelope #>>
+ '{callback_reply,action}'
+ = 'controlled_apply_result'
+ AND outbound.source_envelope #>>
+ '{callback_reply,execution_kind}'
+ = 'no_write_replay'
+ AND outbound.source_envelope #>>
+ '{callback_reply,automation_run_id}'
+ = :automation_run_id
+ AND outbound.source_envelope #>>
+ '{callback_reply,incident_id}'
+ = :incident_id
+ AND outbound.source_envelope #>>
+ '{callback_reply,apply_op_id}'
+ = CAST(:terminal_op_id AS text)
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM alert_operation_log lifecycle
+ WHERE lifecycle.incident_id = :incident_id
+ AND lifecycle.event_type::text
+ = 'EXECUTION_COMPLETED'
+ AND lifecycle.success IS FALSE
+ AND lifecycle.context
+ ->> 'automation_run_id'
+ = :automation_run_id
+ AND lifecycle.context ->> 'apply_op_id'
+ = CAST(:terminal_op_id AS text)
+ )
+ AND EXISTS (
+ SELECT 1
+ FROM alert_operation_log lifecycle
+ WHERE lifecycle.incident_id = :incident_id
+ AND lifecycle.event_type::text
+ = 'TELEGRAM_RESULT_SENT'
+ AND lifecycle.context
+ ->> 'automation_run_id'
+ = :automation_run_id
+ AND lifecycle.context ->> 'apply_op_id'
+ = CAST(:terminal_op_id AS text)
+ )
+ )
+ """),
+ {
+ "terminal_op_id": terminal_op_id,
+ "automation_run_id": automation_run_id,
+ "incident_id": claim.incident_id,
+ "project_id": project_id,
+ "require_projection_flag": require_projection_flag,
+ },
+ )
+ return result.scalar() is True
+ except Exception as exc:
+ logger.warning(
+ "ansible_stdin_boundary_projection_readback_failed",
+ automation_run_id=automation_run_id,
+ terminal_op_id=terminal_op_id,
+ error_type=type(exc).__name__,
+ )
+ return False
+
+
+async def backfill_missing_stdin_boundary_terminal_projections_once(
+ *,
+ project_id: str = "awoooi",
+ window_hours: int = _STDIN_BOUNDARY_REPLAY_WINDOW_HOURS,
+ limit: int = 1,
+) -> dict[str, Any]:
+ """Close no-write replay attempts without re-executing their apply."""
+
+ stats: dict[str, Any] = {
+ "scanned": 0,
+ "written": 0,
+ "learning_written": 0,
+ "telegram_receipt_acknowledged": 0,
+ "incident_receipt_written": 0,
+ "lifecycle_written": 0,
+ "verified": 0,
+ "runtime_apply_executed": False,
+ "blockers": [],
+ "error": None,
+ }
+ try:
+ rows = await _load_missing_stdin_boundary_terminal_projection_rows(
+ project_id=project_id,
+ window_hours=window_hours,
+ limit=limit,
+ )
+ stats["scanned"] = len(rows)
+ for row in rows:
+ claim = _claim_from_stdin_boundary_terminal_row(row)
+ terminal_op_id = str(row.get("terminal_op_id") or "")
+ terminal_output = _json_loads(row.get("terminal_output"))
+ if claim is None or not terminal_op_id:
+ stats["blockers"].append("terminal_claim_reconstruction_failed")
+ continue
+ returncode = _int_from_value(
+ terminal_output.get("check_mode_replay_returncode"),
+ default=(
+ 0
+ if str(row.get("terminal_status") or "") == "success"
+ else 1
+ ),
+ )
+ replay_result = AnsibleRunResult(
+ returncode=returncode,
+ stdout="",
+ stderr=str(row.get("terminal_error") or ""),
+ duration_ms=_int_from_value(
+ row.get("terminal_duration_ms"),
+ default=0,
+ ),
+ post_verifier_passed=False,
+ )
+ learning = await _record_no_write_replay_learning_receipts(
+ claim,
+ replay_result,
+ terminal_op_id=terminal_op_id,
+ project_id=project_id,
+ )
+ if learning.get("ready") is not True:
+ missing = ",".join(learning.get("missing_stage_ids") or [])
+ stats["blockers"].append(
+ f"learning_receipts_missing:{missing}"[:300]
+ )
+ continue
+ learning_written = (
+ await _append_runtime_stage_receipts_to_operation(
+ operation_id=terminal_op_id,
+ operation_type="ansible_execution_skipped",
+ receipts=tuple(learning.get("receipts") or ()),
+ project_id=project_id,
+ )
+ )
+ stats["learning_written"] += int(learning_written)
+ if not learning_written:
+ stats["blockers"].append("learning_receipt_append_failed")
+ continue
+
+ telegram_acknowledged = (
+ await _stdin_boundary_replay_telegram_receipt_acknowledged(
+ claim,
+ terminal_op_id=terminal_op_id,
+ project_id=project_id,
+ )
+ )
+ if not telegram_acknowledged:
+ receipt_sent = await _send_controlled_apply_telegram_receipt(
+ claim,
+ replay_result,
+ apply_op_id=terminal_op_id,
+ writeback={
+ "verification_result": "failed",
+ "verification": True,
+ "learning": True,
+ },
+ project_id=project_id,
+ execution_kind="no_write_replay",
+ )
+ if receipt_sent:
+ telegram_acknowledged = (
+ await _stdin_boundary_replay_telegram_receipt_acknowledged(
+ claim,
+ terminal_op_id=terminal_op_id,
+ project_id=project_id,
+ )
+ )
+ if not telegram_acknowledged:
+ stats["blockers"].append("telegram_receipt_pending")
+ continue
+ stats["telegram_receipt_acknowledged"] += 1
+
+ execution_lifecycle = await _append_alert_lifecycle_receipt(
+ claim,
+ "EXECUTION_COMPLETED",
+ apply_op_id=terminal_op_id,
+ success=False,
+ action_detail=(
+ "stdin_boundary_replay_verified_no_write_terminal"
+ ),
+ project_id=project_id,
+ error_message="ai_transport_or_playbook_repair_required",
+ post_verifier_passed=False,
+ )
+ telegram_lifecycle = await _append_alert_lifecycle_receipt(
+ claim,
+ "TELEGRAM_RESULT_SENT",
+ apply_op_id=terminal_op_id,
+ success=False,
+ action_detail=(
+ "stdin_boundary_no_write_result_receipt_sent"
+ ),
+ project_id=project_id,
+ post_verifier_passed=False,
+ )
+ if not execution_lifecycle or not telegram_lifecycle:
+ stats["blockers"].append("lifecycle_projection_pending")
+ continue
+ stats["lifecycle_written"] += 1
+
+ incident_detail = (
+ await _record_stdin_boundary_incident_terminal_disposition(
+ claim,
+ terminal_op_id=terminal_op_id,
+ learning_stage_ids=list(learning["stage_ids"]),
+ project_id=project_id,
+ )
+ )
+ if incident_detail is None:
+ stats["blockers"].append("incident_projection_pending")
+ continue
+ closure_detail = {
+ **incident_detail,
+ "terminal_op_id": terminal_op_id,
+ "retry_check_mode_op_id": claim.op_id,
+ "check_mode_replay_returncode": replay_result.returncode,
+ "runtime_apply_executed": False,
+ "verified_no_write_terminal": True,
+ "telegram_receipt_acknowledged": True,
+ "learning_writeback_acknowledged": True,
+ "required_learning_stage_ids": sorted(
+ _NO_WRITE_REPLAY_REQUIRED_STAGE_IDS
+ ),
+ "projection_readback_verified": False,
+ "repository_readback_verified": True,
+ "safe_next_action": (
+ "queue_ai_transport_or_playbook_repair_candidate"
+ ),
+ }
+ closure_receipt = _runtime_stage_receipt(
+ claim,
+ stage_id="incident_closure",
+ evidence_ref=(
+ f"automation_operation_log:{terminal_op_id}:"
+ "no_write_terminal"
+ ),
+ detail=closure_detail,
+ derived_from_durable_chain=True,
+ )
+ incident_receipt_written = (
+ await _append_runtime_stage_receipts_to_operation(
+ operation_id=terminal_op_id,
+ operation_type="ansible_execution_skipped",
+ receipts=(closure_receipt,),
+ project_id=project_id,
+ )
+ )
+ stats["incident_receipt_written"] += int(
+ incident_receipt_written
+ )
+ if not incident_receipt_written:
+ stats["blockers"].append("incident_receipt_append_failed")
+ continue
+ verified = await _verify_stdin_boundary_terminal_projection_readback(
+ claim,
+ terminal_op_id=terminal_op_id,
+ project_id=project_id,
+ require_projection_flag=False,
+ )
+ if not verified:
+ stats["blockers"].append("projection_readback_pending")
+ continue
+ final_receipt = _runtime_stage_receipt(
+ claim,
+ stage_id="incident_closure",
+ evidence_ref=(
+ f"automation_operation_log:{terminal_op_id}:"
+ "no_write_terminal"
+ ),
+ detail={
+ **closure_detail,
+ "projection_readback_verified": True,
+ },
+ derived_from_durable_chain=True,
+ )
+ finalized = await _append_runtime_stage_receipts_to_operation(
+ operation_id=terminal_op_id,
+ operation_type="ansible_execution_skipped",
+ receipts=(final_receipt,),
+ project_id=project_id,
+ )
+ if not finalized:
+ stats["blockers"].append("projection_finalization_pending")
+ continue
+ final_verified = (
+ await _verify_stdin_boundary_terminal_projection_readback(
+ claim,
+ terminal_op_id=terminal_op_id,
+ project_id=project_id,
+ require_projection_flag=True,
+ )
+ )
+ stats["verified"] += int(final_verified)
+ stats["written"] += int(final_verified)
+ if not final_verified:
+ stats["blockers"].append("final_projection_readback_pending")
+ except Exception as exc:
+ stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
+ logger.warning(
+ "ansible_stdin_boundary_projection_backfill_failed",
+ project_id=project_id,
+ **stats,
+ )
+ stats["blockers"] = sorted(set(stats["blockers"]))
+ return stats
+
+
async def _record_learning_writeback_receipt(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
@@ -4032,6 +5260,7 @@ async def _send_controlled_apply_telegram_receipt(
apply_op_id: str,
writeback: dict[str, Any],
project_id: str,
+ execution_kind: str = "controlled_apply",
) -> bool:
try:
from src.services.telegram_gateway import get_telegram_gateway
@@ -4052,6 +5281,7 @@ async def _send_controlled_apply_telegram_receipt(
duration_ms=result.duration_ms,
verifier_written=bool(writeback.get("verification")),
learning_written=bool(writeback.get("learning")),
+ execution_kind=execution_kind,
project_id=project_id,
)
return bool(
@@ -6710,6 +7940,51 @@ async def run_pending_check_modes_once(
effective_timeout_seconds = (
timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS
)
+ stdin_projection_stats = (
+ await backfill_missing_stdin_boundary_terminal_projections_once(
+ project_id=project_id,
+ window_hours=_STDIN_BOUNDARY_REPLAY_WINDOW_HOURS,
+ limit=max(1, min(int(limit), 5)),
+ )
+ )
+ stdin_projection_summary = {
+ "stdin_terminal_projection_scanned": int(
+ stdin_projection_stats.get("scanned") or 0
+ ),
+ "stdin_terminal_projection_written": int(
+ stdin_projection_stats.get("written") or 0
+ ),
+ "stdin_terminal_projection_learning_written": int(
+ stdin_projection_stats.get("learning_written") or 0
+ ),
+ "stdin_terminal_projection_telegram_acknowledged": int(
+ stdin_projection_stats.get("telegram_receipt_acknowledged") or 0
+ ),
+ "stdin_terminal_projection_incident_receipt_written": int(
+ stdin_projection_stats.get("incident_receipt_written") or 0
+ ),
+ "stdin_terminal_projection_verified": int(
+ stdin_projection_stats.get("verified") or 0
+ ),
+ "stdin_terminal_projection_blockers": [
+ str(item)[:300]
+ for item in stdin_projection_stats.get("blockers") or []
+ ],
+ "stdin_terminal_projection_error": (
+ str(stdin_projection_stats.get("error") or "")[:500] or None
+ ),
+ "stdin_terminal_projection_runtime_apply_executed": False,
+ }
+ if (
+ stdin_projection_summary["stdin_terminal_projection_scanned"]
+ or stdin_projection_summary["stdin_terminal_projection_blockers"]
+ or stdin_projection_summary["stdin_terminal_projection_error"]
+ ):
+ logger.info(
+ "ansible_stdin_boundary_projection_priority_tick",
+ project_id=project_id,
+ **stdin_projection_summary,
+ )
receipt_stats = await backfill_missing_auto_repair_execution_receipts_once(
project_id=project_id,
window_hours=24,
@@ -6824,6 +8099,7 @@ async def run_pending_check_modes_once(
"failed": 0,
"capability_expired": expired_capability_count,
"blockers": blockers,
+ **stdin_projection_summary,
**receipt_backfill_summary,
**failed_apply_retry_summary,
}
@@ -6836,6 +8112,7 @@ async def run_pending_check_modes_once(
"failed": 0,
"capability_expired": expired_capability_count,
"blockers": transport_blockers,
+ **stdin_projection_summary,
**receipt_backfill_summary,
**failed_apply_retry_summary,
}
@@ -7061,6 +8338,7 @@ async def run_pending_check_modes_once(
"capability_expired": expired_capability_count,
"capability_revoke_failed": capability_revoke_failed,
"catalog_replay_error": catalog_replay_error,
+ **stdin_projection_summary,
**receipt_backfill_summary,
**failed_apply_retry_summary,
"blockers": [],
diff --git a/apps/api/src/services/awooop_ansible_learning_writeback.py b/apps/api/src/services/awooop_ansible_learning_writeback.py
index c712f54d4..2e529d048 100644
--- a/apps/api/src/services/awooop_ansible_learning_writeback.py
+++ b/apps/api/src/services/awooop_ansible_learning_writeback.py
@@ -104,9 +104,15 @@ def _playbook_trust_readback(
playbook: PlaybookRecord,
identity: dict[str, Any],
verification_result: str,
+ trust_mutation_performed: bool = True,
+ observation_kind: str = "controlled_apply",
) -> dict[str, Any]:
return {
- "schema_version": "ansible_playbook_trust_writeback_v2",
+ "schema_version": (
+ "ansible_playbook_trust_writeback_v2"
+ if trust_mutation_performed
+ else "ansible_playbook_no_write_observation_v1"
+ ),
"identity_schema_version": identity["schema_version"],
"identity_fingerprint": identity["identity_fingerprint"],
"canonical_playbook_id": playbook.playbook_id,
@@ -125,8 +131,11 @@ def _playbook_trust_readback(
),
"trust_score": round(float(playbook.trust_score or 0.0), 6),
"verification_result": verification_result,
+ "observation_kind": observation_kind,
"learning_recorded": True,
- "trust_updated": True,
+ "trust_observation_recorded": True,
+ "trust_updated": trust_mutation_performed,
+ "trust_mutation_performed": trust_mutation_performed,
"repository_write_acknowledged": True,
"repository_readback_verified": True,
"operation_receipt_readback_verified": True,
@@ -146,10 +155,17 @@ def _existing_trust_readback_matches(
*,
playbook: PlaybookRecord,
identity: dict[str, Any],
+ trust_mutation_performed: bool = True,
+ observation_kind: str = "controlled_apply",
) -> bool:
+ expected_schema_version = (
+ "ansible_playbook_trust_writeback_v2"
+ if trust_mutation_performed
+ else "ansible_playbook_no_write_observation_v1"
+ )
return bool(
existing_trust.get("schema_version")
- == "ansible_playbook_trust_writeback_v2"
+ == expected_schema_version
and existing_trust.get("canonical_playbook_id")
== identity["canonical_playbook_id"]
and existing_trust.get("identity_fingerprint")
@@ -161,7 +177,22 @@ def _existing_trust_readback_matches(
and existing_trust.get("repository_readback_verified") is True
and existing_trust.get("operation_receipt_readback_verified") is True
and existing_trust.get("learning_recorded") is True
- and existing_trust.get("trust_updated") is True
+ and existing_trust.get("trust_updated") is trust_mutation_performed
+ and (
+ existing_trust.get("trust_mutation_performed")
+ is trust_mutation_performed
+ or (
+ trust_mutation_performed
+ and existing_trust.get("trust_mutation_performed") is None
+ )
+ )
+ and (
+ existing_trust.get("observation_kind") == observation_kind
+ or (
+ observation_kind == "controlled_apply"
+ and existing_trust.get("observation_kind") is None
+ )
+ )
)
@@ -175,8 +206,10 @@ async def record_ansible_playbook_trust_writeback(
playbook_path: str,
apply_op_id: str,
verification_result: str,
+ trust_mutation_performed: bool = True,
+ observation_kind: str = "controlled_apply",
) -> dict[str, Any] | None:
- """Atomically update canonical PlayBook trust and its learning receipt."""
+ """Persist a PlayBook trust mutation or a no-write observation receipt."""
catalog = get_ansible_catalog_item(catalog_id)
identity = resolve_ansible_playbook_identity(catalog_id)
@@ -237,6 +270,8 @@ async def record_ansible_playbook_trust_writeback(
existing_trust,
playbook=playbook,
identity=identity,
+ trust_mutation_performed=trust_mutation_performed,
+ observation_kind=observation_kind,
):
return existing_trust
logger.warning(
@@ -309,26 +344,29 @@ async def record_ansible_playbook_trust_writeback(
db.add(playbook)
await db.flush()
- source_incident_ids = list(playbook.source_incident_ids or [])
- if incident_id and incident_id not in source_incident_ids:
- source_incident_ids.append(incident_id)
- playbook.source_incident_ids = source_incident_ids
+ if trust_mutation_performed:
+ source_incident_ids = list(playbook.source_incident_ids or [])
+ if incident_id and incident_id not in source_incident_ids:
+ source_incident_ids.append(incident_id)
+ playbook.source_incident_ids = source_incident_ids
- if success:
- playbook.success_count = int(playbook.success_count or 0) + 1
- playbook.trust_score = 0.9 * float(playbook.trust_score) + 0.1
- else:
- playbook.failure_count = int(playbook.failure_count or 0) + 1
- playbook.trust_score = 0.8 * float(playbook.trust_score)
- playbook.trust_score = max(0.0, min(1.0, playbook.trust_score))
- playbook.last_used_at = now
- playbook.updated_at = now
+ if success:
+ playbook.success_count = int(playbook.success_count or 0) + 1
+ playbook.trust_score = 0.9 * float(playbook.trust_score) + 0.1
+ else:
+ playbook.failure_count = int(playbook.failure_count or 0) + 1
+ playbook.trust_score = 0.8 * float(playbook.trust_score)
+ playbook.trust_score = max(0.0, min(1.0, playbook.trust_score))
+ playbook.last_used_at = now
+ playbook.updated_at = now
await db.flush()
await db.refresh(playbook)
trust_writeback = _playbook_trust_readback(
playbook=playbook,
identity=identity,
verification_result=verification_result,
+ trust_mutation_performed=trust_mutation_performed,
+ observation_kind=observation_kind,
)
input_payload = {
"schema_version": "ansible_learning_writeback_receipt_v3",
@@ -341,6 +379,8 @@ async def record_ansible_playbook_trust_writeback(
"playbook_path": playbook_path,
"apply_op_id": apply_op_id,
"verification_result": verification_result,
+ "observation_kind": observation_kind,
+ "trust_mutation_performed": trust_mutation_performed,
"learning_repository": "playbooks",
"writer_source_sha": trust_writeback["writer_source_sha"],
"stores_raw_logs": False,
@@ -348,7 +388,9 @@ async def record_ansible_playbook_trust_writeback(
}
output_payload = {
"learning_recorded": True,
- "trust_updated": True,
+ "trust_observation_recorded": True,
+ "trust_updated": trust_mutation_performed,
+ "trust_mutation_performed": trust_mutation_performed,
"success": success,
"repository_readback_verified": True,
"playbook_trust_writeback": trust_writeback,
@@ -360,7 +402,7 @@ async def record_ansible_playbook_trust_writeback(
input, output, dry_run_result, parent_op_id, tags
) VALUES (
'ansible_learning_writeback_recorded',
- 'ansible_controlled_apply_worker',
+ :actor,
'success',
:incident_db_id,
CAST(:input AS jsonb),
@@ -375,9 +417,14 @@ async def record_ansible_playbook_trust_writeback(
"input": json.dumps(input_payload, ensure_ascii=False),
"output": json.dumps(output_payload, ensure_ascii=False),
"parent_op_id": apply_op_id,
+ "actor": (
+ "ansible_controlled_apply_worker"
+ if trust_mutation_performed
+ else "ansible_no_write_replay_worker"
+ ),
"tags": [
"ansible",
- "controlled_apply",
+ observation_kind,
"learning_writeback",
"playbook_trust",
"ai_agent_auto_execution",
@@ -409,7 +456,11 @@ async def record_ansible_playbook_trust_writeback(
operation_row
and operation_row.get("status") == "success"
and operation_output.get("learning_recorded") is True
- and operation_output.get("trust_updated") is True
+ and operation_output.get("trust_observation_recorded") is True
+ and operation_output.get("trust_updated")
+ is trust_mutation_performed
+ and operation_output.get("trust_mutation_performed")
+ is trust_mutation_performed
and operation_output.get("repository_readback_verified")
is True
and operation_trust.get("playbook_row_fingerprint")
@@ -435,10 +486,11 @@ async def ensure_ansible_rag_writeback(
project_id: str,
incident_id: str,
apply_op_id: str,
+ path_type: str | None = None,
) -> dict[str, Any] | None:
"""Persist and read back the KM embedding used by semantic retrieval."""
- path_type = f"ansible_apply_receipt:{apply_op_id[:8]}"
+ effective_path_type = path_type or f"ansible_apply_receipt:{apply_op_id[:8]}"
try:
async with get_db_context(project_id) as db:
selected = await db.execute(
@@ -453,7 +505,10 @@ async def ensure_ansible_rag_writeback(
AND path_type = :path_type
LIMIT 1
"""),
- {"incident_id": incident_id, "path_type": path_type},
+ {
+ "incident_id": incident_id,
+ "path_type": effective_path_type,
+ },
)
row = selected.mappings().first()
if not row:
diff --git a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
index 659981f14..10f21e35e 100644
--- a/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
+++ b/apps/api/src/services/telegram_alert_monitoring_coverage_readback.py
@@ -88,9 +88,24 @@ _RUNTIME_LIFECYCLE_SQL = """
)
ELSE NULL
END
- AND apply.operation_type
- = 'ansible_apply_executed'
- AND apply.status = 'failed'
+ AND (
+ (
+ apply.operation_type
+ = 'ansible_apply_executed'
+ AND apply.status = 'failed'
+ )
+ OR (
+ apply.operation_type
+ = 'ansible_execution_skipped'
+ AND apply.status
+ IN ('success', 'failed')
+ AND apply.input ->> 'execution_mode'
+ = 'stdin_boundary_check_mode_replay_terminal'
+ AND apply.output #>>
+ '{runtime_apply_executed}'
+ = 'false'
+ )
+ )
AND apply.input ->> 'automation_run_id'
= COALESCE(
NULLIF(
@@ -148,7 +163,59 @@ _RUNTIME_LIFECYCLE_SQL = """
AND closure_receipt.value #>>
'{detail,repository_readback_verified}'
= 'true'
+ AND (
+ apply.operation_type
+ <> 'ansible_execution_skipped'
+ OR (
+ closure_receipt.value #>>
+ '{detail,learning_writeback_acknowledged}'
+ = 'true'
+ AND closure_receipt.value #>>
+ '{detail,incident_projection_readback_verified}'
+ = 'true'
+ AND closure_receipt.value #>>
+ '{detail,projection_readback_verified}'
+ = 'true'
+ )
+ )
)
+ AND (
+ apply.operation_type
+ <> 'ansible_execution_skipped'
+ OR (
+ SELECT COUNT(DISTINCT
+ stage_receipt.value
+ ->> 'stage_id')
+ FROM jsonb_array_elements(
+ COALESCE(
+ apply.input
+ -> 'runtime_stage_receipts',
+ '[]'::jsonb
+ )
+ ) stage_receipt(value)
+ WHERE stage_receipt.value
+ ->> 'stage_id' IN (
+ 'mcp_context',
+ 'service_log_evidence',
+ 'normalized_asset_identity',
+ 'source_truth_diff',
+ 'risk_policy_decision',
+ 'executor_log_projection',
+ 'retry_or_rollback',
+ 'km_playbook_writeback',
+ 'rag_writeback',
+ 'playbook_trust',
+ 'timeline_projection'
+ )
+ AND stage_receipt.value
+ ->> 'automation_run_id'
+ = apply.input
+ ->> 'automation_run_id'
+ AND stage_receipt.value
+ ->> 'durable_receipt'
+ = 'true'
+ ) = 11
+ )
)
AND EXISTS (
SELECT 1
diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py
index 246be013e..855a068b8 100644
--- a/apps/api/src/services/telegram_gateway.py
+++ b/apps/api/src/services/telegram_gateway.py
@@ -10086,6 +10086,7 @@ class TelegramGateway:
duration_ms: int,
verifier_written: bool,
learning_written: bool,
+ execution_kind: str = "controlled_apply",
project_id: str = "awoooi",
) -> dict:
"""Send and mirror the AI Agent controlled-apply result receipt."""
@@ -10098,16 +10099,29 @@ class TelegramGateway:
)
return {}
- success = verification_result == "success" and returncode == 0
+ no_write_replay = execution_kind == "no_write_replay"
+ success = (
+ not no_write_replay
+ and verification_result == "success"
+ and returncode == 0
+ )
next_step = (
"monitor_for_regression"
if success
- else "queue_ai_rollback_or_playbook_repair"
+ else (
+ "queue_ai_transport_or_playbook_repair"
+ if no_write_replay
+ else "queue_ai_rollback_or_playbook_repair"
+ )
)
title = (
"CONTROLLED APPLY RESULT｜AI Agent 受控執行完成"
if success
- else "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復"
+ else (
+ "NO-WRITE RETRY RESULT｜AI Agent 無寫入重試待修復"
+ if no_write_replay
+ else "CONTROLLED APPLY RESULT｜AI Agent 受控執行待修復"
+ )
)
truth_chain: dict[str, object] | None = None
try:
@@ -10147,11 +10161,13 @@ class TelegramGateway:
if source_extra is not None:
source_extra["outbound_message_type"] = "final" if success else "error"
source_extra["automation_run_id"] = automation_run_id
+ source_extra["execution_kind"] = execution_kind
callback_reply = source_extra.get("callback_reply")
if isinstance(callback_reply, dict):
callback_reply["automation_run_id"] = automation_run_id
callback_reply["apply_op_id"] = apply_op_id
callback_reply["project_id"] = project_id or "awoooi"
+ callback_reply["execution_kind"] = execution_kind
source_refs = source_extra.get("source_refs")
if isinstance(source_refs, dict):
source_refs["automation_run_ids"] = [automation_run_id]
@@ -10160,7 +10176,10 @@ class TelegramGateway:
f"Run: {html.escape(str(automation_run_id or '')[:8])}",
f"Incident: {html.escape(str(incident_id))}",
f"Catalog: {html.escape(str(catalog_id or '--'))}",
- f"Apply op: {html.escape(str(apply_op_id or '')[:8])}",
+ (
+ f"{'Terminal' if no_write_replay else 'Apply'} op: "
+ f"{html.escape(str(apply_op_id or '')[:8])}"
+ ),
f"PlayBook: {html.escape(str(playbook_path or '--'))}",
(
"Result: "
@@ -10173,6 +10192,11 @@ class TelegramGateway:
f"{html.escape(_bool_code(verifier_written))} / KM "
f"{html.escape(_bool_code(learning_written))}"
),
+ (
+ "Runtime apply: 0"
+ if no_write_replay
+ else "Runtime apply: 1"
+ ),
f"Next: {html.escape(next_step)}",
f"Runs: {html.escape(incident_runs_url(incident_id, project_id=project_id or 'awoooi'))}",
]
diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py
index 02d9156c8..e2baa3fc1 100644
--- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py
+++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py
@@ -25,6 +25,52 @@ from src.services.ai_automation_runtime_contract import (
)
+def test_runtime_switches_distinguish_api_process_from_dedicated_broker() -> None:
+ payload = {
+ "runtime_switches": {
+ "ansible_check_mode_worker_enabled": False,
+ }
+ }
+ boundary = {
+ "status": "verified_ready",
+ "production_boundary_verified": True,
+ "source_sha_matches_deployment": True,
+ "receipt_ref": (
+ "configmap:awoooi-prod/awoooi-executor-boundary-verification"
+ ),
+ "verified_source_sha": "a" * 40,
+ "controls": {
+ "broker_ssh_mount_present": True,
+ "broker_ssh_egress_allowlisted": True,
+ "broker_kubernetes_token_absent": True,
+ },
+ }
+
+ runtime_control_module._attach_effective_executor_runtime_switches(
+ payload,
+ boundary_readback=boundary,
+ )
+
+ switches = payload["runtime_switches"]
+ assert switches["ansible_check_mode_worker_enabled"] is False
+ assert switches["ansible_check_mode_worker_scope"] == "api_process_only"
+ assert switches["dedicated_execution_broker_ready"] is True
+ assert switches["effective_ansible_executor_available"] is True
+ assert switches["effective_executor_readback_source"] == (
+ boundary["receipt_ref"]
+ )
+ assert switches["effective_executor_verified_source_sha"] == "a" * 40
+
+ boundary["source_sha_matches_deployment"] = False
+ runtime_control_module._attach_effective_executor_runtime_switches(
+ payload,
+ boundary_readback=boundary,
+ )
+ assert switches["dedicated_execution_broker_ready"] is False
+ assert switches["effective_ansible_executor_available"] is False
+ assert switches["effective_executor_readback_source"] is None
+
+
def _closed_retry_terminal_row() -> dict[str, object]:
run_id = "retry-run-1"
apply_op_id = "94925d5e-7d73-4ed2-82e2-a3f2d3026270"
diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py
index fe5eedd99..afc14222f 100644
--- a/apps/api/tests/test_ansible_verified_closure.py
+++ b/apps/api/tests/test_ansible_verified_closure.py
@@ -600,6 +600,12 @@ def test_terminal_candidate_contract_requires_verified_four_node_chain() -> None
assert "apply.parent_op_id = check_mode.op_id" in predicate
assert "replay.parent_op_id = apply.op_id" in predicate
assert "replay.status IN ('success', 'failed')" in predicate
+ assert (
+ "replay_check.parent_op_id = failed_check.parent_op_id"
+ in predicate
+ )
+ assert "replay_of_check_mode_op_id" in predicate
+ assert "failed_check.parent_op_id = candidate.op_id" in predicate
assert "verified_no_write_terminal" in predicate
assert "repository_readback_verified" in predicate
assert "{detail,execution_success}" in predicate
diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py
index 72a4b36f5..fbf23630e 100644
--- a/apps/api/tests/test_awooop_truth_chain_service.py
+++ b/apps/api/tests/test_awooop_truth_chain_service.py
@@ -25,6 +25,7 @@ from src.services.awooop_ansible_check_mode_service import (
AnsibleRunResult,
_append_alert_lifecycle_receipt,
_append_runtime_stage_receipts_to_apply,
+ _append_runtime_stage_receipts_to_operation,
_automation_operation_log_incident_id,
_build_auto_repair_execution_receipt,
_build_retry_runtime_stage_receipt,
@@ -33,21 +34,26 @@ from src.services.awooop_ansible_check_mode_service import (
_execution_capability_timeout_seconds,
_expire_stale_ansible_execution_capabilities,
_issue_ansible_execution_capability,
+ _load_missing_stdin_boundary_terminal_projection_rows,
_load_open_failed_apply_retry_row,
_load_pre_decision_context_runtime_stage_receipts,
_post_apply_km_path_type,
_post_apply_verification_result,
_record_auto_repair_execution_receipt,
_record_learning_writeback_receipt,
+ _record_no_write_replay_learning_receipts,
_record_retry_runtime_stage_receipt,
_record_runtime_stage_receipts,
+ _record_stdin_boundary_incident_terminal_disposition,
_record_stdin_boundary_replay_terminal,
_record_timeline_projection_receipt,
_resolve_execution_capability_operation_type,
_revoke_ansible_execution_capability,
_run_ansible_command,
_send_controlled_apply_telegram_receipt,
+ _verify_stdin_boundary_terminal_projection_readback,
backfill_missing_auto_repair_execution_receipts_once,
+ backfill_missing_stdin_boundary_terminal_projections_once,
build_ansible_apply_command,
build_ansible_check_mode_claim_input,
build_ansible_check_mode_command,
@@ -159,6 +165,13 @@ def test_ansible_audit_keeps_external_incident_id_in_json_not_bigint_column() ->
assert "candidate.input ->> 'incident_id'" in decision_source
assert "verified_ansible_candidate_terminal_sql" in decision_source
assert "ansible_execution_skipped" in terminal_source
+ assert "stdin_boundary_check_mode_replay_terminal" in terminal_source
+ assert "learning_writeback_acknowledged" in terminal_source
+ assert "incident_projection_readback_verified" in terminal_source
+ assert "projection_readback_verified" in terminal_source
+ assert "TELEGRAM_RESULT_SENT" in terminal_source
+ assert "no_write_replay" in terminal_source
+ assert "= 11" in terminal_source
assert "operation_type, actor, status, incident_id" in claim_source
assert "incident_db_id" in decision_source
assert "incident_db_id" in claim_source
@@ -2089,6 +2102,226 @@ def test_failed_stdin_boundary_checks_replay_once_without_apply() -> None:
assert "stdin_boundary_replay_passed_no_write_terminal" in run_source
+def test_stdin_boundary_terminal_projection_closes_only_from_durable_receipts(
+) -> None:
+ backfill_source = inspect.getsource(
+ backfill_missing_stdin_boundary_terminal_projections_once
+ )
+ learning_source = inspect.getsource(
+ _record_no_write_replay_learning_receipts
+ )
+ incident_source = inspect.getsource(
+ _record_stdin_boundary_incident_terminal_disposition
+ )
+ loader_source = inspect.getsource(
+ _load_missing_stdin_boundary_terminal_projection_rows
+ )
+ verifier_source = inspect.getsource(
+ _verify_stdin_boundary_terminal_projection_readback
+ )
+
+ assert "run_controlled_apply_for_claim" not in backfill_source
+ assert 'execution_kind="no_write_replay"' in backfill_source
+ assert "runtime_apply_executed" in backfill_source
+ assert "projection_readback_verified" in backfill_source
+ assert "EntryType.ANTI_PATTERN" in learning_source
+ assert "EntryStatus.PUBLISHED" in learning_source
+ assert "ensure_ansible_rag_writeback" in learning_source
+ assert "record_ansible_playbook_trust_writeback" in learning_source
+ assert 'verification_result="not_applicable_no_runtime_apply"' in learning_source
+ assert "trust_mutation_performed=False" in learning_source
+ assert 'observation_kind="no_write_replay"' in learning_source
+ assert "mcp_context" in learning_source
+ assert "controlled_apply_executed" in learning_source
+ assert "historical_no_write_replay" in learning_source
+ assert "verified_no_write_terminal_pending_ai_repair" in learning_source
+ assert "incident_projection_superseded" in incident_source
+ for source in (loader_source, verifier_source):
+ assert "JOIN automation_operation_log failed_check" in source
+ assert "candidate.op_id = replay.parent_op_id" in source
+ assert "failed_check.parent_op_id = replay.parent_op_id" in source
+ assert "replay_of_check_mode_op_id" in source
+ assert "retry_of_check_mode_op_id" in source
+ assert "TELEGRAM_RESULT_SENT" in verifier_source
+ assert "km_playbook_writeback" in verifier_source
+ assert "playbook_trust" in verifier_source
+ assert "runtime_apply_executed" in verifier_source
+ assert verifier_source.count(
+ "CAST(:terminal_op_id AS text)"
+ ) == 4
+
+
+@pytest.mark.asyncio
+async def test_stdin_boundary_terminal_projection_runs_no_write_pipeline(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ from src.services import awooop_ansible_check_mode_service as service
+
+ claim = AnsibleCheckModeClaim(
+ op_id="00000000-0000-0000-0000-000000000322",
+ source_candidate_op_id="00000000-0000-0000-0000-000000000321",
+ incident_id="INC-STDIN-PROJECTION",
+ catalog_id="ansible:188-momo-backup-user",
+ playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
+ apply_playbook_path=(
+ "infra/ansible/playbooks/188-momo-backup-user.yml"
+ ),
+ inventory_hosts=("host_188",),
+ risk_level="low",
+ input_payload={
+ "automation_run_id": "00000000-0000-0000-0000-000000000321",
+ "historical_stdin_boundary_replay": True,
+ "controlled_apply_allowed": False,
+ "apply_enabled": False,
+ },
+ )
+ terminal_op_id = "00000000-0000-0000-0000-000000000323"
+ calls: list[tuple[str, object]] = []
+ telegram_readbacks = iter((False, True))
+
+ async def load_rows(**_kwargs):
+ return [
+ {
+ "terminal_op_id": terminal_op_id,
+ "terminal_status": "failed",
+ "terminal_output": {
+ "check_mode_replay_returncode": 2,
+ "runtime_apply_executed": False,
+ },
+ "terminal_error": "stdin_boundary_repair_required",
+ "terminal_duration_ms": 71,
+ }
+ ]
+
+ async def learning_receipts(*_args, **_kwargs):
+ calls.append(("learning", _kwargs["terminal_op_id"]))
+ return {
+ "ready": True,
+ "receipts": ({"stage_id": "mcp_context"},),
+ "stage_ids": sorted(service._NO_WRITE_REPLAY_REQUIRED_STAGE_IDS),
+ "missing_stage_ids": [],
+ }
+
+ async def append_receipts(**kwargs):
+ calls.append(
+ (
+ "append",
+ (
+ kwargs["operation_id"],
+ kwargs["operation_type"],
+ tuple(
+ item["stage_id"] for item in kwargs["receipts"]
+ ),
+ ),
+ )
+ )
+ return True
+
+ async def telegram_readback(*_args, **_kwargs):
+ calls.append(("telegram_readback", _kwargs["terminal_op_id"]))
+ return next(telegram_readbacks)
+
+ async def send_telegram(*_args, **kwargs):
+ calls.append(("telegram_send", kwargs["execution_kind"]))
+ return True
+
+ async def append_lifecycle(_claim, event_type, **kwargs):
+ calls.append(
+ (
+ "lifecycle",
+ (event_type, kwargs["apply_op_id"], kwargs["success"]),
+ )
+ )
+ return True
+
+ async def incident_projection(*_args, **_kwargs):
+ calls.append(("incident_projection", _kwargs["terminal_op_id"]))
+ return {
+ "no_write_terminal": True,
+ "incident_projection_readback_verified": True,
+ "repository_readback_verified": True,
+ }
+
+ verifier_flags: list[bool] = []
+
+ async def verify_projection(*_args, **kwargs):
+ verifier_flags.append(kwargs["require_projection_flag"])
+ return True
+
+ monkeypatch.setattr(
+ service,
+ "_load_missing_stdin_boundary_terminal_projection_rows",
+ load_rows,
+ )
+ monkeypatch.setattr(
+ service,
+ "_claim_from_stdin_boundary_terminal_row",
+ lambda _row: claim,
+ )
+ monkeypatch.setattr(
+ service,
+ "_record_no_write_replay_learning_receipts",
+ learning_receipts,
+ )
+ monkeypatch.setattr(
+ service,
+ "_append_runtime_stage_receipts_to_operation",
+ append_receipts,
+ )
+ monkeypatch.setattr(
+ service,
+ "_stdin_boundary_replay_telegram_receipt_acknowledged",
+ telegram_readback,
+ )
+ monkeypatch.setattr(
+ service,
+ "_send_controlled_apply_telegram_receipt",
+ send_telegram,
+ )
+ monkeypatch.setattr(
+ service,
+ "_append_alert_lifecycle_receipt",
+ append_lifecycle,
+ )
+ monkeypatch.setattr(
+ service,
+ "_record_stdin_boundary_incident_terminal_disposition",
+ incident_projection,
+ )
+ monkeypatch.setattr(
+ service,
+ "_verify_stdin_boundary_terminal_projection_readback",
+ verify_projection,
+ )
+
+ result = await backfill_missing_stdin_boundary_terminal_projections_once(
+ project_id="awoooi",
+ limit=1,
+ )
+
+ assert result["scanned"] == 1
+ assert result["written"] == 1
+ assert result["verified"] == 1
+ assert result["runtime_apply_executed"] is False
+ assert result["blockers"] == []
+ assert ("telegram_send", "no_write_replay") in calls
+ assert (
+ "lifecycle",
+ ("EXECUTION_COMPLETED", terminal_op_id, False),
+ ) in calls
+ assert (
+ "lifecycle",
+ ("TELEGRAM_RESULT_SENT", terminal_op_id, False),
+ ) in calls
+ append_calls = [value for name, value in calls if name == "append"]
+ assert len(append_calls) == 3
+ assert all(item[0] == terminal_op_id for item in append_calls)
+ assert all(item[1] == "ansible_execution_skipped" for item in append_calls)
+ assert append_calls[1][2] == ("incident_closure",)
+ assert append_calls[2][2] == ("incident_closure",)
+ assert verifier_flags == [False, True]
+
+
@pytest.mark.asyncio
async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims(
monkeypatch: pytest.MonkeyPatch,
@@ -2443,10 +2676,51 @@ def test_ansible_learning_writeback_uses_canonical_durable_service() -> None:
assert '"operation_receipt_readback_verified": True' in receipt_source
assert "operation_readback" in durable_source
assert 'operation_output.get("learning_recorded") is True' in durable_source
- assert 'operation_output.get("trust_updated") is True' in durable_source
+ assert "trust_mutation_performed: bool = True" in durable_source
+ assert 'operation_output.get("trust_updated")' in durable_source
+ assert "is trust_mutation_performed" in durable_source
assert '"durable_write_acknowledged": True' in receipt_source
+def test_no_write_playbook_observation_does_not_claim_trust_mutation() -> None:
+ playbook = SimpleNamespace(
+ playbook_id="PB-ANSIBLE-188-MOMO-BACKUP-USER",
+ project_id="awoooi",
+ version=3,
+ success_count=7,
+ failure_count=2,
+ trust_score=0.81,
+ updated_at=datetime(2026, 7, 14, tzinfo=UTC),
+ )
+ identity = {
+ "schema_version": "ansible_playbook_identity_v1",
+ "identity_fingerprint": "a" * 64,
+ "canonical_playbook_id": playbook.playbook_id,
+ "catalog_id": "ansible:188-momo-backup-user",
+ "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
+ }
+
+ receipt = _playbook_trust_readback(
+ playbook=playbook,
+ identity=identity,
+ verification_result="not_applicable_no_runtime_apply",
+ trust_mutation_performed=False,
+ observation_kind="no_write_replay",
+ )
+
+ assert receipt["schema_version"] == (
+ "ansible_playbook_no_write_observation_v1"
+ )
+ assert receipt["observation_kind"] == "no_write_replay"
+ assert receipt["learning_recorded"] is True
+ assert receipt["trust_observation_recorded"] is True
+ assert receipt["trust_updated"] is False
+ assert receipt["trust_mutation_performed"] is False
+ assert receipt["success_count"] == 7
+ assert receipt["failure_count"] == 2
+ assert receipt["trust_score"] == 0.81
+
+
def test_ansible_catalog_ids_map_to_stable_playbook_primary_keys() -> None:
expected = {
"ansible:110-devops": "PB-ANSIBLE-110-DEVOPS",
@@ -2492,6 +2766,9 @@ def test_ansible_rag_writeback_requires_embedding_readback() -> None:
source = inspect.getsource(ensure_ansible_rag_writeback)
fallback_source = inspect.getsource(_ensure_rag_chunk_fallback)
+ assert "path_type: str | None = None" in source
+ assert "path_type or f\"ansible_apply_receipt:" in source
+ assert '"path_type": effective_path_type' in source
assert "ensure_entry_embedding" in source
assert "embedding IS NOT NULL" in source
assert "durable_write_acknowledged" in source
@@ -2652,11 +2929,16 @@ def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
receipt_builder_source = inspect.getsource(
_build_retry_runtime_stage_receipt
)
- append_source = inspect.getsource(_append_runtime_stage_receipts_to_apply)
+ append_source = inspect.getsource(
+ _append_runtime_stage_receipts_to_operation
+ )
assert "_build_retry_runtime_stage_receipt" in receipt_source
assert "_runtime_stage_receipt" in receipt_builder_source
assert 'stage_id="retry_or_rollback"' in receipt_builder_source
assert "jsonb_array_elements" in append_source
+ assert "_append_runtime_stage_receipts_to_operation" in inspect.getsource(
+ _append_runtime_stage_receipts_to_apply
+ )
assert "ansible_controlled_retry_terminal_v2" in receipt_builder_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_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py
index 4356a0108..dc14e22fd 100644
--- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py
+++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py
@@ -61,6 +61,7 @@ def test_worker_manifest_has_no_execution_transport_or_write_loop() -> None:
for item in deployment["spec"]["template"]["spec"]["containers"][0]["env"]
}
assert worker_env["ENABLE_SECURITY_CONTROL_PLANE_MAINTENANCE_WORKER"] == "true"
+ assert worker_env["AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS"] == "168"
def test_execution_broker_is_only_workload_with_ssh_transport() -> None:
diff --git a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
index df562da36..533af7de7 100644
--- a/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
+++ b/apps/api/tests/test_telegram_alert_monitoring_coverage_readback_api.py
@@ -805,13 +805,32 @@ def test_runtime_lifecycle_sql_accepts_only_durable_failed_terminals() -> None:
assert "completed_terminal_count" in sql
assert "apply.status = 'failed'" in sql
+ assert "stdin_boundary_check_mode_replay_terminal" in sql
+ assert "= 'ansible_execution_skipped'" in sql
assert "= 'retry_or_rollback'" in sql
assert "'{detail,runtime_apply_executed}'" in sql
assert "'{detail,verified_no_write_terminal}'" in sql
assert "= 'incident_closure'" in sql
assert "'{detail,no_write_terminal}'" in sql
assert "'{detail,telegram_receipt_acknowledged}'" in sql
+ assert "'{detail,learning_writeback_acknowledged}'" in sql
assert "'{detail,repository_readback_verified}'" in sql
+ assert "'{detail,incident_projection_readback_verified}'" in sql
+ assert "'{detail,projection_readback_verified}'" in sql
+ for stage_id in (
+ "mcp_context",
+ "service_log_evidence",
+ "normalized_asset_identity",
+ "source_truth_diff",
+ "risk_policy_decision",
+ "executor_log_projection",
+ "retry_or_rollback",
+ "km_playbook_writeback",
+ "rag_writeback",
+ "playbook_trust",
+ "timeline_projection",
+ ):
+ assert stage_id in sql
assert "telegram_result.event_type" in sql
assert "= 'TELEGRAM_RESULT_SENT'" in sql
assert "completed_success_count" not in sql
diff --git a/apps/api/tests/test_telegram_message_templates.py b/apps/api/tests/test_telegram_message_templates.py
index 6b6a6fc4c..194af0064 100644
--- a/apps/api/tests/test_telegram_message_templates.py
+++ b/apps/api/tests/test_telegram_message_templates.py
@@ -1281,6 +1281,79 @@ async def test_controlled_apply_result_receipt_marks_callback_reply_evidence(mon
assert snapshot["execution"]["ansible"]["controlled_apply"] is True
+@pytest.mark.asyncio
+async def test_no_write_replay_receipt_never_claims_runtime_apply(monkeypatch) -> None:
+ gateway = TelegramGateway()
+ sent_requests = []
+
+ async def fake_send_request(method, payload):
+ sent_requests.append((method, payload))
+ return {"ok": True, "result": {"message_id": 12346}}
+
+ async def fake_fetch_truth_chain(*, source_id, project_id):
+ assert source_id == "INC-NO-WRITE-001"
+ assert project_id == "awoooi"
+ return {
+ "truth_status": {
+ "current_stage": "execution_skipped",
+ "stage_status": "failed",
+ "needs_human": False,
+ "blockers": ["stdin_boundary_repair_required"],
+ },
+ "automation_quality": {
+ "verdict": "ai_repair_queued",
+ "facts": {},
+ "blockers": ["stdin_boundary_repair_required"],
+ },
+ "execution": {"automation_operation_log": [], "ansible": {}},
+ }
+
+ async def fake_fetch_km_completion_summary(**_kwargs):
+ return {}
+
+ monkeypatch.setattr(TelegramGateway, "alert_chat_id", property(lambda _self: "chat"))
+ monkeypatch.setattr(gateway, "_send_request", fake_send_request)
+ monkeypatch.setattr(
+ "src.services.awooop_truth_chain_service.fetch_truth_chain",
+ fake_fetch_truth_chain,
+ )
+ monkeypatch.setattr(
+ "src.services.telegram_gateway._fetch_km_stale_completion_summary_for_incident",
+ fake_fetch_km_completion_summary,
+ )
+
+ result = await gateway.send_controlled_apply_result_receipt(
+ automation_run_id="00000000-0000-0000-0000-000000000002",
+ incident_id="INC-NO-WRITE-001",
+ catalog_id="ansible:188-momo-backup-user",
+ apply_op_id="83b7a95c-3652-4c0d-bb4c-729e500acedb",
+ playbook_path="infra/ansible/playbooks/188-momo-backup-user.yml",
+ verification_result="success",
+ returncode=0,
+ duration_ms=721,
+ verifier_written=True,
+ learning_written=True,
+ execution_kind="no_write_replay",
+ project_id="awoooi",
+ )
+
+ assert result["ok"] is True
+ assert len(sent_requests) == 1
+ method, payload = sent_requests[0]
+ assert method == "sendMessage"
+ assert "NO-WRITE RETRY RESULT" in payload["text"]
+ assert "AI Agent 無寫入重試待修復" in payload["text"]
+ assert "Terminal op:" in payload["text"]
+ assert "Runtime apply: 0" in payload["text"]
+ assert "queue_ai_transport_or_playbook_repair" in payload["text"]
+ assert "AI Agent 受控執行完成" not in payload["text"]
+ source_extra = payload["_awooop_source_envelope_extra"]
+ assert source_extra["execution_kind"] == "no_write_replay"
+ assert source_extra["outbound_message_type"] == "error"
+ assert source_extra["callback_reply"]["action"] == "controlled_apply_result"
+ assert source_extra["callback_reply"]["execution_kind"] == "no_write_replay"
+
+
def test_callback_reply_awooop_status_chain_snapshot_marks_manual_gate() -> None:
"""Callback evidence 要保存當下 AwoooP 狀態鏈，不只保存 live query 結果。"""
snapshot = telegram_gateway_module._callback_reply_awooop_status_chain_snapshot(
diff --git a/k8s/awoooi-prod/08-deployment-worker.yaml b/k8s/awoooi-prod/08-deployment-worker.yaml
index ae77f0c9d..7bf9869bf 100644
--- a/k8s/awoooi-prod/08-deployment-worker.yaml
+++ b/k8s/awoooi-prod/08-deployment-worker.yaml
@@ -183,7 +183,7 @@ spec:
- name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT
value: "2"
- name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS
- value: "24"
+ value: "168"
- name: AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS
value: "10"
- name: ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER