wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into codex/security-governance-review-20260710

Browse Source
This commit is contained in:
ogt
2026-07-10 19:51:35 +08:00
parent bb66db6e57 542fbda109
commit 2a0466703b
7 changed files with 389 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

103
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -11,6 +11,7 @@ from __future__ import annotations
import asyncio
import copy
import json
import time
from collections.abc import Iterable, Mapping
from datetime import UTC, datetime
@@ -34,6 +35,7 @@ from src.services.ai_automation_runtime_contract import (
AI_AUTOMATION_EXECUTION_CORRELATED_STAGES,
AI_AUTOMATION_REQUIRED_LOOP_STAGES,
AI_AUTOMATION_RUNTIME_CONTRACT_SCHEMA_VERSION,
AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION,
)
from src.services.report_generation_service import (
DAILY_REPORT_HOUR_TAIPEI,
@@ -3347,6 +3349,45 @@ def _merge_runtime_operation_rows(
return merged
def _runtime_stage_receipt_map(
rows: Iterable[Mapping[str, Any] | Any],
*,
expected_run_id: str,
) -> dict[str, dict[str, Any]]:
receipts: dict[str, dict[str, Any]] = {}
if not expected_run_id:
return receipts
for raw_row in rows:
row = _row_mapping(raw_row)
if str(row.get("automation_run_id") or "") != expected_run_id:
continue
raw_receipts = row.get("runtime_stage_receipts")
if isinstance(raw_receipts, str):
try:
raw_receipts = json.loads(raw_receipts)
except json.JSONDecodeError:
continue
if not isinstance(raw_receipts, list):
continue
for raw_receipt in raw_receipts:
if not isinstance(raw_receipt, Mapping):
continue
receipt = dict(raw_receipt)
stage_id = str(receipt.get("stage_id") or "")
if (
stage_id not in AI_AUTOMATION_REQUIRED_LOOP_STAGES
or receipt.get("schema_version")
!= AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION
or str(receipt.get("automation_run_id") or "")
!= expected_run_id
or receipt.get("durable_receipt") is not True
or not str(receipt.get("evidence_ref") or "")
):
continue
receipts[stage_id] = receipt
return receipts
def _stage_status(row: Mapping[str, Any] | None, *, fallback_status: str | None = None) -> str:
if row is None:
return fallback_status or "missing"
@@ -3499,6 +3540,10 @@ def _autonomous_execution_loop_ledger(
or (latest_apply or {}).get("automation_run_id")
or ""
)
same_run_stage_receipt_by_id = _runtime_stage_receipt_map(
operation_rows,
expected_run_id=automation_run_id,
)
candidate_present = latest_candidate is not None
check_present = latest_check is not None
apply_present = latest_apply is not None
@@ -3653,6 +3698,11 @@ def _autonomous_execution_loop_ledger(
"run_id_mismatch_stage_ids": run_id_mismatch_stage_ids,
"missing_stage_ids": missing_stage_ids,
"next_executor_action": next_executor_action,
"same_run_stage_receipts": [
same_run_stage_receipt_by_id[stage_id]
for stage_id in AI_AUTOMATION_REQUIRED_LOOP_STAGES
if stage_id in same_run_stage_receipt_by_id
],
"stages": stages,
"safety_contract": {
"writes_on_read": False,
@@ -4920,6 +4970,14 @@ def _attach_runtime_receipt_readback(
for stage in trace_stages
if isinstance(stage, Mapping) and stage.get("stage_id")
}
same_run_stage_receipts = loop_ledger.get("same_run_stage_receipts")
if not isinstance(same_run_stage_receipts, list):
same_run_stage_receipts = []
same_run_stage_receipt_by_id = {
str(receipt.get("stage_id") or ""): receipt
for receipt in same_run_stage_receipts
if isinstance(receipt, Mapping) and receipt.get("stage_id")
}
latest_flow_closed = latest_flow.get("closed") is True
latest_loop_closed = loop_ledger.get("closed") is True
automation_run_id = str(loop_ledger.get("automation_run_id") or "")
@@ -4932,18 +4990,30 @@ def _attach_runtime_receipt_readback(
same_run_proven = stage.get("run_id_matches_expected") is True
evidence_source = "same_run_execution_loop_ledger"
else:
stage = trace_stage_by_id.get(stage_id, {})
evidence_present = stage.get("recorded") is True
# Aggregate trace counts prove that a source exists, not that it
# belongs to the latest AutomationRun. Incident closure is the
# only auxiliary stage currently carrying the same run identity.
same_run_proven = bool(
stage_id == "incident_closure"
and evidence_present
and automation_run_id
and latest_flow_run_id == automation_run_id
)
evidence_source = "aggregate_trace_ledger"
same_run_receipt = same_run_stage_receipt_by_id.get(stage_id, {})
if same_run_receipt:
stage = same_run_receipt
evidence_present = True
same_run_proven = bool(
automation_run_id
and str(same_run_receipt.get("automation_run_id") or "")
== automation_run_id
and same_run_receipt.get("durable_receipt") is True
)
evidence_source = "same_run_runtime_stage_receipt"
else:
stage = trace_stage_by_id.get(stage_id, {})
evidence_present = stage.get("recorded") is True
# Aggregate trace counts prove that a source exists, not that it
# belongs to the latest AutomationRun. Incident closure is the
# only auxiliary stage currently carrying the same run identity.
same_run_proven = bool(
stage_id == "incident_closure"
and evidence_present
and automation_run_id
and latest_flow_run_id == automation_run_id
)
evidence_source = "aggregate_trace_ledger"
completion_eligible = evidence_present and same_run_proven
strict_stage_contracts.append(
{
@@ -4952,6 +5022,11 @@ def _attach_runtime_receipt_readback(
"same_run_correlation_proven": same_run_proven,
"completion_eligible": completion_eligible,
"evidence_source": evidence_source,
"evidence_ref": (
stage.get("evidence_ref")
if evidence_source == "same_run_runtime_stage_receipt"
else None
),
}
)
required_stage_count = len(strict_stage_contracts)
@@ -5948,6 +6023,7 @@ _RUNTIME_OPERATION_LATEST_SQL = """
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
input -> 'runtime_stage_receipts' AS runtime_stage_receipts,
coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode,
duration_ms,
created_at
@@ -6023,6 +6099,7 @@ _RUNTIME_OPERATION_LATEST_DIRECT_SQL = """
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
input -> 'runtime_stage_receipts' AS runtime_stage_receipts,
coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode,
duration_ms,
created_at
@@ -6088,6 +6165,7 @@ _RUNTIME_OPERATION_CHAIN_SQL = """
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
input -> 'runtime_stage_receipts' AS runtime_stage_receipts,
coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode,
duration_ms,
created_at
@@ -6126,6 +6204,7 @@ _RUNTIME_OPERATION_CHAIN_DIRECT_SQL = """
input ->> 'check_mode_op_id' AS check_mode_op_id,
input ->> 'risk_level' AS risk_level,
input ->> 'controlled_apply_allowed' AS controlled_apply_allowed,
input -> 'runtime_stage_receipts' AS runtime_stage_receipts,
coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode,
duration_ms,
created_at

1
apps/api/src/services/ai_automation_runtime_contract.py
View File

@@ -5,6 +5,7 @@ from __future__ import annotations
AI_AUTOMATION_RUNTIME_CONTRACT_SCHEMA_VERSION = (
"ai_automation_runtime_completion_contract_v2"
)
AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION = "ai_automation_stage_receipt_v1"
# Every stage is required before the product may claim an autonomous security
# or operations loop is complete. A source contract or historical aggregate is

210
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -22,6 +22,9 @@ from sqlalchemy import text
from src.core.config import settings
from src.db.base import get_db_context
from src.services.ai_automation_runtime_contract import (
AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION,
)
from src.services.awooop_ansible_audit_service import get_ansible_catalog_item
logger = structlog.get_logger(__name__)
@@ -735,6 +738,179 @@ def _post_apply_action_label(claim: AnsibleCheckModeClaim, *, apply_op_id: str)
)
def _automation_run_id_for_claim(claim: AnsibleCheckModeClaim) -> str:
return str(
claim.input_payload.get("automation_run_id")
or claim.source_candidate_op_id
)
def _runtime_stage_receipt(
claim: AnsibleCheckModeClaim,
*,
stage_id: str,
evidence_ref: str,
detail: dict[str, Any],
derived_from_durable_chain: bool = False,
) -> dict[str, Any]:
return {
"schema_version": AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION,
"stage_id": stage_id,
"automation_run_id": _automation_run_id_for_claim(claim),
"incident_id": claim.incident_id,
"evidence_ref": evidence_ref,
"detail": detail,
"durable_receipt": True,
"derived_from_durable_chain": derived_from_durable_chain,
"raw_log_payload_stored": False,
"secret_value_stored": False,
}
def build_ansible_pre_apply_runtime_stage_receipts(
claim: AnsibleCheckModeClaim,
*,
derived_from_durable_chain: bool = False,
) -> list[dict[str, Any]]:
"""Build receipts proven before controlled apply from the claimed dry-run."""
return [
_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.apply_playbook_path,
},
derived_from_durable_chain=derived_from_durable_chain,
),
_runtime_stage_receipt(
claim,
stage_id="source_truth_diff",
evidence_ref=(
"automation_operation_log:"
f"{claim.op_id}:dry_run_result"
),
detail={
"check_mode_op_id": claim.op_id,
"check_mode": True,
"diff": True,
"check_mode_passed_before_apply": True,
},
derived_from_durable_chain=derived_from_durable_chain,
),
_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": True,
"break_glass_required": False,
"policy": "global_product_governance_v2",
},
derived_from_durable_chain=derived_from_durable_chain,
),
]
def build_ansible_post_apply_runtime_stage_receipts(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
*,
apply_op_id: str,
verifier_ready: bool,
derived_from_durable_chain: bool = False,
) -> list[dict[str, Any]]:
receipts = [
_runtime_stage_receipt(
claim,
stage_id="executor_log_projection",
evidence_ref=f"automation_operation_log:{apply_op_id}:output",
detail={
"apply_op_id": apply_op_id,
"returncode": result.returncode,
"timed_out": result.timed_out,
"duration_ms": result.duration_ms,
"projection_sanitized": True,
},
derived_from_durable_chain=derived_from_durable_chain,
)
]
if result.returncode == 0 and verifier_ready:
receipts.append(
_runtime_stage_receipt(
claim,
stage_id="retry_or_rollback",
evidence_ref=f"incident_evidence:{apply_op_id}:post_execution_state",
detail={
"apply_op_id": apply_op_id,
"terminal_type": "verified_success_no_retry_or_rollback_required",
"bounded_retry_performed": False,
"rollback_performed": False,
"post_apply_verifier_passed": True,
},
derived_from_durable_chain=derived_from_durable_chain,
)
)
return receipts
async def _record_runtime_stage_receipts(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
*,
apply_op_id: str,
verifier_ready: bool,
project_id: str,
derived_from_durable_chain: bool = False,
) -> bool:
receipts = [
*build_ansible_pre_apply_runtime_stage_receipts(
claim,
derived_from_durable_chain=derived_from_durable_chain,
),
*build_ansible_post_apply_runtime_stage_receipts(
claim,
result,
apply_op_id=apply_op_id,
verifier_ready=verifier_ready,
derived_from_durable_chain=derived_from_durable_chain,
),
]
try:
async with get_db_context(project_id) as db:
updated = await db.execute(
text("""
UPDATE automation_operation_log
SET input = jsonb_set(
coalesce(input, '{}'::jsonb),
'{runtime_stage_receipts}',
CAST(:receipts AS jsonb),
true
)
WHERE op_id = CAST(:apply_op_id AS uuid)
AND operation_type = 'ansible_apply_executed'
"""),
{
"receipts": json.dumps(receipts, ensure_ascii=False),
"apply_op_id": apply_op_id,
},
)
return bool(updated.rowcount)
except Exception as exc:
logger.warning(
"ansible_runtime_stage_receipt_write_failed",
incident_id=claim.incident_id,
apply_op_id=apply_op_id,
error=str(exc),
)
return False
async def _record_learning_writeback_receipt(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
@@ -1080,6 +1256,7 @@ async def backfill_missing_auto_repair_execution_receipts_once(
"verification_written": 0,
"learning_written": 0,
"trust_learning_written": 0,
"runtime_stage_receipts_written": 0,
"skipped": 0,
"error": None,
}
@@ -1098,6 +1275,16 @@ async def backfill_missing_auto_repair_execution_receipts_once(
apply.error,
apply.duration_ms,
apply.status,
EXISTS (
SELECT 1
FROM incident_evidence verifier
WHERE verifier.incident_id = coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
)
AND verifier.post_execution_state ->> 'apply_op_id' = apply.op_id::text
AND verifier.verification_result = 'success'
) AS verifier_ready,
apply.created_at
FROM automation_operation_log apply
WHERE apply.operation_type = 'ansible_apply_executed'
@@ -1134,6 +1321,7 @@ async def backfill_missing_auto_repair_execution_receipts_once(
WHERE learning.operation_type = 'ansible_learning_writeback_recorded'
AND learning.parent_op_id = apply.op_id
)
OR NOT (coalesce(apply.input, '{}'::jsonb) ? 'runtime_stage_receipts')
)
ORDER BY apply.created_at DESC
LIMIT :limit
@@ -1169,6 +1357,17 @@ async def backfill_missing_auto_repair_execution_receipts_once(
stats["learning_written"] += 1
if writeback.get("trust_learning"):
stats["trust_learning_written"] += 1
if await _record_runtime_stage_receipts(
claim,
result,
apply_op_id=str(row.get("op_id") or ""),
verifier_ready=bool(
writeback.get("verification") or row.get("verifier_ready")
),
project_id=project_id,
derived_from_durable_chain=True,
):
stats["runtime_stage_receipts_written"] += 1
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning("ansible_auto_repair_execution_receipt_backfill_failed", **stats)
@@ -1853,6 +2052,9 @@ async def run_controlled_apply_for_claim(
"apply_enabled": True,
"approval_required_before_apply": False,
"controlled_apply_allowed": True,
"runtime_stage_receipts": (
build_ansible_pre_apply_runtime_stage_receipts(claim)
),
}, ensure_ascii=False),
"dry_run_result": json.dumps({
"check_mode_executed_before_apply": True,
@@ -1920,6 +2122,13 @@ async def run_controlled_apply_for_claim(
apply_op_id=apply_op_id,
project_id=project_id,
)
runtime_stage_receipts_written = await _record_runtime_stage_receipts(
claim,
result,
apply_op_id=apply_op_id,
verifier_ready=bool(writeback.get("verification")),
project_id=project_id,
)
telegram_receipt_sent = await _send_controlled_apply_telegram_receipt(
claim,
result,
@@ -1940,6 +2149,7 @@ async def run_controlled_apply_for_claim(
auto_repair_receipt_written=receipt_written,
post_apply_verification_written=writeback.get("verification"),
post_apply_learning_written=writeback.get("learning"),
runtime_stage_receipts_written=runtime_stage_receipts_written,
telegram_receipt_sent=telegram_receipt_sent,
)
return result