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

Merge remote-tracking branch 'gitea-ssh/main' into codex/github-redacted-evidence-validator-20260627

Browse Source
This commit is contained in:
Your Name
2026-06-28 08:56:57 +08:00
parent ba5a10c96e 64f52017b8
commit 00702a4de4
41 changed files with 2933 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
apps/api/src/services/agent_market_discovery_classifier.py
View File

@@ -2,7 +2,7 @@
Agent market discovery classifier
=================================
Classifies manually reviewed discovery repositories from primary GitHub
Classifies AI controlled discovery repositories from primary GitHub
metadata. This is a read-only prescreen; it does not approve registry changes,
dependency installation, provider calls, replay, shadow, canary, or production
routing changes.
@@ -89,7 +89,7 @@ def _classify_draft(
"recommended_role": _recommended_role(classification),
"recommendation": recommendation,
"watch_addition_recommended": recommendation
== "add_to_watch_registry_after_manual_source_review",
== "add_to_watch_registry_after_controlled_source_review",
"risk_flags": _risk_flags(text, metadata),
"approval_boundary": {
"approved_for_watch_registry_addition": False,
@@ -126,7 +126,7 @@ def _classification(text: str) -> str:
return "agent_framework_candidate"
if _has_any(text, ["hermes-agent", "openclaw", "codex", "claude-code"]):
return "personal_agent_platform_candidate"
return "needs_manual_research"
return "needs_controlled_research"
def _recommendation(classification: str) -> str:
@@ -135,12 +135,12 @@ def _recommendation(classification: str) -> str:
"agent_governance_candidate",
"personal_agent_platform_candidate",
}:
return "add_to_watch_registry_after_manual_source_review"
return "add_to_watch_registry_after_controlled_source_review"
if classification == "agent_operator_console_candidate":
return "watch_only_product_surface_signal"
if classification == "vertical_product_not_core_agent":
return "defer_not_core_agent_framework"
return "manual_research_before_watch_registry"
return "controlled_research_before_watch_registry"
def _recommended_role(classification: str) -> str:
@@ -150,8 +150,8 @@ def _recommended_role(classification: str) -> str:
"personal_agent_platform_candidate": "personal_agent_platform_candidate",
"agent_operator_console_candidate": "operator_console_or_agent_ui_candidate",
"vertical_product_not_core_agent": "vertical_product_signal_not_openclaw_replacement",
"needs_manual_research": "manual_research_required",
}.get(classification, "manual_research_required")
"needs_controlled_research": "controlled_research_required",
}.get(classification, "controlled_research_required")
def _risk_flags(text: str, metadata: dict[str, Any]) -> list[str]:
@@ -166,11 +166,11 @@ def _risk_flags(text: str, metadata: dict[str, Any]) -> list[str]:
def _required_next_gate(recommendation: str) -> str:
if recommendation == "add_to_watch_registry_after_manual_source_review":
return "operator_confirms_primary_sources_then_add_watch_registry_only"
if recommendation == "add_to_watch_registry_after_controlled_source_review":
return "ai_controlled_primary_source_verifier_then_watch_registry_patch"
if recommendation == "watch_only_product_surface_signal":
return "operator_confirms_product_surface_relevance_before_watch_only_entry"
return "manual_research_no_registry_change"
return "ai_controlled_product_surface_relevance_check_before_watch_only_entry"
return "controlled_research_no_registry_change"
def _metadata_text(repo: str, metadata: dict[str, Any]) -> str:

17
apps/api/src/services/agent_market_discovery_review.py
View File

@@ -2,8 +2,8 @@
Agent market discovery review
=============================
Turns raw discovery search results from the market watch into a manual intake
queue. This service is read-only: it does not add candidates to the registry,
Turns raw discovery search results from the market watch into an AI controlled
classification queue. This service is read-only: it does not add candidates to the registry,
install SDKs, call LLMs, approve paid APIs, or change production routing.
"""
@@ -100,7 +100,7 @@ def _candidate_drafts(
decision = (
"keep_existing_candidate_watch"
if known
else "manual_primary_source_classification_required"
else "ai_controlled_primary_source_classification_required"
)
next_gate = (
"use_existing_market_watch_candidate"
@@ -116,6 +116,7 @@ def _candidate_drafts(
"decision": decision,
"recommended_next_gate": next_gate,
"approval_boundary": {
"controlled_classification_allowed": not known,
"approved_for_registry_addition": False,
"approved_for_sdk_install": False,
"approved_for_paid_api_calls": False,
@@ -128,7 +129,7 @@ def _candidate_drafts(
def _summary(watch_report: dict[str, Any], drafts: list[dict[str, Any]]) -> dict[str, int]:
manual = [
controlled = [
draft
for draft in drafts
if draft["status"] == "needs_primary_source_classification"
@@ -143,9 +144,9 @@ def _summary(watch_report: dict[str, Any], drafts: list[dict[str, Any]]) -> dict
"already_watched_or_registered": sum(
1 for draft in drafts if draft["status"] == "already_watched_or_registered"
),
"manual_classification_required": len(manual),
"new_manual_classification_required": sum(
1 for draft in manual if draft["new_since_previous_review"]
"controlled_classification_queue": len(controlled),
"new_controlled_classification_queue": sum(
1 for draft in controlled if draft["new_since_previous_review"]
),
"source_failures": sum(
1
@@ -209,7 +210,7 @@ def _recommended_actions(*, known: bool) -> list[str]:
return [
"verify_official_or_primary_sources",
"classify_role_against_awoooi_agent_taxonomy",
"add_to_watch_registry_only_after_manual_review",
"queue_controlled_watch_registry_patch_after_primary_source_verifier",
"do_not_install_sdk_or_call_provider",
"do_not_enter_replacement_replay_before_market_scorecard",
]

45
apps/api/src/services/agent_market_governance_snapshot.py
View File

@@ -58,7 +58,7 @@ def build_agent_market_governance_snapshot(
current_decision = (
"openclaw_remains_production_decision_core"
if approvals["replacement_decisions_approved"] == 0
else "manual_review_required_unexpected_replacement_approval"
else "break_glass_review_required_unexpected_replacement_approval"
)
snapshot_generated_at = generated_at or datetime.now(timezone.utc).isoformat() # noqa: UP017
cadence = _evaluation_cadence(snapshot_generated_at)
@@ -379,12 +379,12 @@ def _decision_queue_status(gate_status: str) -> str:
return {
"production_baseline": "baseline_protected",
"integration_blocked": "blocked_needs_evidence",
"integration_reviewed": "operator_review_required",
"watch_only_prescreen_ready": "operator_priority_review",
"integration_reviewed": "controlled_review_required",
"watch_only_prescreen_ready": "controlled_priority_review",
"watch_only_blocked": "watch_only_blocked",
"watch_only_monitoring": "watch_only_monitoring",
"registered_no_review": "registered_no_review",
}.get(gate_status, "operator_review_required")
}.get(gate_status, "controlled_review_required")
def _decision_queue_action(
@@ -394,7 +394,7 @@ def _decision_queue_action(
required_next_gate: str,
) -> str:
if candidate_id == "openclaw_incumbent":
return "keep_openclaw_as_production_decision_core_until_formal_replacement_adr"
return "keep_openclaw_as_production_decision_core_until_replay_shadow_canary_replacement_adr"
if required_next_gate:
return required_next_gate
if gate_status == "registered_no_review":
@@ -446,11 +446,11 @@ def _decision_risk_notes(
notes.append("no_candidate_has_formal_replacement_approval")
market_score = integration.get("market_score") or {}
notes.extend(str(value) for value in market_score.get("risks") or [])
notes.extend(_controlled_label(value) for value in market_score.get("risks") or [])
classification = promotion.get("classification") or {}
notes.extend(str(value) for value in classification.get("risk_flags") or [])
notes.extend(str(value) for value in operator_blockers)
notes.extend(_controlled_label(value) for value in classification.get("risk_flags") or [])
notes.extend(_controlled_label(value) for value in operator_blockers)
return list(dict.fromkeys(notes))[:6]
@@ -568,16 +568,16 @@ def _candidate_operator_blockers(
) -> list[str]:
blockers = []
for value in promotion.get("blockers") or []:
blockers.append(str(value))
blockers.append(_controlled_label(value))
for value in integration.get("unblock_conditions") or []:
blockers.append(str(value))
blockers.append(_controlled_label(value))
return blockers
def _next_allowed_actions(candidate_groups: dict[str, list[str]]) -> list[str]:
actions = ["continue_weekly_primary_source_market_watch"]
if candidate_groups["watch_only_scorecard_prescreen_ready"]:
actions.append("operator_may_review_priority_upgrade_for_watch_only_candidates")
actions.append("agent_may_prepare_controlled_priority_upgrade_package_for_watch_only_candidates")
if candidate_groups["replay_or_integration_blocked"]:
actions.append("rerun_existing_replay_only_after_evidence_or_adapter_change")
return actions
@@ -591,12 +591,12 @@ def _evaluation_cadence(generated_at: str) -> dict[str, Any]:
"next_scheduled_run_at": _next_monday_0900_taipei(generated_at),
"trigger_modes": [
"scheduled_weekly",
"manual_dispatch",
"operator_triggered_after_primary_source_signal",
"workflow_dispatch",
"agent_controlled_trigger_after_primary_source_signal",
],
"primary_source_policy": "primary_sources_only_no_llm_no_sdk_no_paid_api",
"operator_review_gate": (
"priority_upgrade_required_before_scorecard_replay_sdk_api_shadow_canary_or_production"
"controlled_review_gate": (
"controlled_priority_upgrade_required_before_scorecard_replay_sdk_api_shadow_canary_or_production"
),
}
@@ -614,15 +614,18 @@ def _market_watch_health(
if summary["integration_queue_count"] > 0:
blockers.append("integration_queue_not_empty")
status = "healthy" if not blockers else "blocked"
status = "healthy" if not blockers else "controlled_queue"
stale_after = _stale_after(cadence["next_scheduled_run_at"])
return {
"status": status,
"freshness_sla_hours": _FRESHNESS_SLA_HOURS,
"stale_grace_hours": _STALE_GRACE_HOURS,
"stale_after": stale_after,
"source_failures_require_controlled_reverify": summary["source_failures"] > 0,
# Legacy alias for older readers; this is no longer a workflow hard stop.
"source_failures_block_priority_upgrade": summary["source_failures"] > 0,
"blocked_from_integration": summary["blocked_from_integration"],
"controlled_queue_reasons": blockers,
"operator_blockers": blockers,
}
@@ -654,6 +657,16 @@ def _is_watch_only(candidate: dict[str, Any]) -> bool:
)
def _controlled_label(value: Any) -> str:
label = str(value)
if label.endswith("_by_operator"):
return f"{label.removesuffix('_by_operator')}_by_ai_controlled_verifier"
if label.startswith("operator_confirms_"):
subject = label.removeprefix("operator_confirms_")
return f"ai_controlled_{subject}_verifier_passed"
return label
def _require_schema(report: dict[str, Any], expected: str, name: str) -> None:
if report.get("schema_version") != expected:
raise ValueError(f"{name} must be {expected}")

4
apps/api/src/services/agent_market_watch_promotion_review.py
View File

@@ -109,7 +109,7 @@ def _review_watch_only_candidate(
and classification_recommended
)
decision = (
"eligible_for_operator_priority_review_before_market_scorecard"
"eligible_for_controlled_priority_review_before_market_scorecard"
if eligible_for_scorecard
else "remain_watch_only_until_evidence_gap_resolved"
)
@@ -147,7 +147,7 @@ def _review_watch_only_candidate(
"approved_for_shadow_or_canary": False,
"blockers": blockers,
"required_next_gate": (
"operator_priority_upgrade_then_market_scorecard_prescreen"
"controlled_priority_upgrade_then_market_scorecard_prescreen"
if eligible_for_scorecard
else "continue_watch_only_until_primary_source_evidence_is_sufficient"
),

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

@@ -9,10 +9,14 @@ KM, and Telegram receipts are present.
from __future__ import annotations
from collections.abc import Iterable, Mapping
from datetime import datetime, timezone
from typing import Any
from src.core.config import settings
from src.core.logging import get_logger
from sqlalchemy import text
from src.db.base import get_db_context
from src.services.report_generation_service import (
DAILY_REPORT_HOUR_TAIPEI,
MONTHLY_REPORT_DAY_TAIPEI,
@@ -25,6 +29,18 @@ _SCHEMA_VERSION = "ai_agent_autonomous_runtime_control_v1"
_RUNTIME_AUTHORITY = "current_owner_directive_controlled_ai_automation"
_DEPLOY_READBACK_MARKER = "p2_416_d1n_autonomous_runtime_control_prod_readback_v2"
_DEPLOY_ATTEMPT_NOTE = "cd_3673_retry_after_host_pressure_gate_fix"
_LIVE_READBACK_SCHEMA_VERSION = "ai_agent_autonomous_runtime_receipt_readback_v1"
_DEFAULT_PROJECT_ID = "awoooi"
_DEFAULT_LOOKBACK_HOURS = 24
_EXECUTOR_OPERATION_TYPES = (
"ansible_candidate_matched",
"ansible_check_mode_executed",
"ansible_apply_executed",
"ansible_rollback_executed",
"ansible_execution_skipped",
)
logger = get_logger(__name__)
def _allowed_risk_levels() -> list[str]:
@@ -32,6 +48,340 @@ def _allowed_risk_levels() -> list[str]:
return sorted({item.strip().lower() for item in raw.split(",") if item.strip()})
def _utc_iso(value: Any) -> str | None:
if value is None:
return None
if isinstance(value, datetime):
if value.tzinfo is None:
value = value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc).isoformat()
return str(value)
def _row_mapping(row: Mapping[str, Any] | Any) -> dict[str, Any]:
if isinstance(row, Mapping):
return dict(row)
mapping = getattr(row, "_mapping", None)
if mapping is not None:
return dict(mapping)
return dict(row)
def _int_value(value: Any) -> int:
try:
return int(value or 0)
except (TypeError, ValueError):
return 0
def _sanitize_latest_rows(
rows: Iterable[Mapping[str, Any] | Any],
*,
allowed_keys: tuple[str, ...],
time_keys: tuple[str, ...] = ("created_at", "collected_at", "queued_at", "sent_at"),
limit: int = 5,
) -> list[dict[str, Any]]:
clean_rows: list[dict[str, Any]] = []
for row in rows:
item = _row_mapping(row)
clean: dict[str, Any] = {}
for key in allowed_keys:
if key not in item:
continue
value = item.get(key)
clean[key] = _utc_iso(value) if key in time_keys else value
clean_rows.append(clean)
if len(clean_rows) >= limit:
break
return clean_rows
def _operation_counts(
rows: Iterable[Mapping[str, Any] | Any],
) -> dict[str, dict[str, Any]]:
counts = {
operation_type: {
"total": 0,
"recent": 0,
"by_status": {},
}
for operation_type in _EXECUTOR_OPERATION_TYPES
}
for row in rows:
item = _row_mapping(row)
operation_type = str(item.get("operation_type") or "unknown")
status = str(item.get("status") or "unknown")
bucket = counts.setdefault(
operation_type,
{
"total": 0,
"recent": 0,
"by_status": {},
},
)
total = _int_value(item.get("total"))
recent = _int_value(item.get("recent"))
bucket["total"] += total
bucket["recent"] += recent
bucket["by_status"][status] = bucket["by_status"].get(status, 0) + total
return counts
def _status_counts(
rows: Iterable[Mapping[str, Any] | Any],
*,
status_key: str,
) -> dict[str, Any]:
by_status: dict[str, int] = {}
total = 0
recent = 0
for row in rows:
item = _row_mapping(row)
status = str(item.get(status_key) or "unknown")
row_total = _int_value(item.get("total"))
by_status[status] = by_status.get(status, 0) + row_total
total += row_total
recent += _int_value(item.get("recent"))
return {
"total": total,
"recent": recent,
"by_status": by_status,
}
def _latest_flow_closure(
*,
operation_latest_rows: Iterable[Mapping[str, Any] | Any],
verifier_latest_rows: Iterable[Mapping[str, Any] | Any],
km_latest_rows: Iterable[Mapping[str, Any] | Any],
telegram_latest_rows: Iterable[Mapping[str, Any] | Any],
) -> dict[str, Any]:
operation_rows = [_row_mapping(row) for row in operation_latest_rows]
verifier_rows = [_row_mapping(row) for row in verifier_latest_rows]
km_rows = [_row_mapping(row) for row in km_latest_rows]
telegram_rows = [_row_mapping(row) for row in telegram_latest_rows]
latest_apply = next(
(
row
for row in operation_rows
if str(row.get("operation_type") or "") == "ansible_apply_executed"
),
None,
)
if latest_apply is None:
return {
"apply_op_id": None,
"incident_id": None,
"has_post_apply_verifier": False,
"has_km_writeback": False,
"has_telegram_receipt": False,
"closed": False,
"missing": [
"ansible_apply_executed",
"post_apply_verifier",
"km_writeback",
"telegram_receipt",
],
}
apply_op_id = str(latest_apply.get("op_id") or "")
incident_id = str(latest_apply.get("incident_id") or "")
km_path_type = f"ansible_apply_receipt:{apply_op_id[:8]}" if apply_op_id else ""
has_verifier = any(
str(row.get("apply_op_id") or "") == apply_op_id
for row in verifier_rows
)
has_km = any(
str(row.get("path_type") or "") == km_path_type
or (
incident_id
and str(row.get("related_incident_id") or "") == incident_id
)
for row in km_rows
)
has_telegram = any(
str(row.get("send_status") or "") == "sent"
and str(row.get("action") or "") == "controlled_apply_result"
and (
not incident_id
or str(row.get("incident_id") or "") == incident_id
)
for row in telegram_rows
)
missing = [
name
for name, present in (
("post_apply_verifier", has_verifier),
("km_writeback", has_km),
("telegram_receipt", has_telegram),
)
if not present
]
return {
"apply_op_id": apply_op_id or None,
"incident_id": incident_id or None,
"has_post_apply_verifier": has_verifier,
"has_km_writeback": has_km,
"has_telegram_receipt": has_telegram,
"closed": not missing,
"missing": missing,
}
def build_runtime_receipt_readback_from_rows(
*,
project_id: str = _DEFAULT_PROJECT_ID,
lookback_hours: int = _DEFAULT_LOOKBACK_HOURS,
db_read_status: str = "ok",
operation_count_rows: Iterable[Mapping[str, Any] | Any] = (),
operation_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
verifier_count_rows: Iterable[Mapping[str, Any] | Any] = (),
verifier_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
km_count_rows: Iterable[Mapping[str, Any] | Any] = (),
km_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
telegram_count_rows: Iterable[Mapping[str, Any] | Any] = (),
telegram_latest_rows: Iterable[Mapping[str, Any] | Any] = (),
error_type: str | None = None,
) -> dict[str, Any]:
"""Build the live executor receipt readback from already-fetched rows."""
operation_latest = list(operation_latest_rows)
verifier_latest = list(verifier_latest_rows)
km_latest = list(km_latest_rows)
telegram_latest = list(telegram_latest_rows)
operation_summary = _operation_counts(operation_count_rows)
verifier_summary = _status_counts(
verifier_count_rows,
status_key="verification_result",
)
km_summary = _status_counts(km_count_rows, status_key="status")
telegram_summary = _status_counts(telegram_count_rows, status_key="send_status")
latest_closure = _latest_flow_closure(
operation_latest_rows=operation_latest,
verifier_latest_rows=verifier_latest,
km_latest_rows=km_latest,
telegram_latest_rows=telegram_latest,
)
apply_summary = operation_summary.get("ansible_apply_executed") or {}
readback = {
"schema_version": _LIVE_READBACK_SCHEMA_VERSION,
"project_id": project_id,
"lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)),
"db_read_status": db_read_status,
"writes_on_read": False,
"ansible_operations": {
"counts": operation_summary,
"latest": _sanitize_latest_rows(
operation_latest,
allowed_keys=(
"op_id",
"parent_op_id",
"operation_type",
"status",
"actor",
"incident_id",
"catalog_id",
"playbook_path",
"execution_mode",
"returncode",
"duration_ms",
"created_at",
),
),
},
"ansible_apply_executed": {
"total": _int_value(apply_summary.get("total")),
"recent": _int_value(apply_summary.get("recent")),
"by_status": apply_summary.get("by_status") or {},
},
"post_apply_verifier": {
**verifier_summary,
"latest": _sanitize_latest_rows(
verifier_latest,
allowed_keys=(
"id",
"incident_id",
"matched_playbook_id",
"verification_result",
"apply_op_id",
"catalog_id",
"playbook_path",
"returncode",
"collected_at",
),
),
},
"km_writeback": {
**km_summary,
"latest": _sanitize_latest_rows(
km_latest,
allowed_keys=(
"id",
"title",
"related_incident_id",
"related_playbook_id",
"path_type",
"status",
"created_by",
"created_at",
),
),
},
"telegram_receipt": {
**telegram_summary,
"latest": _sanitize_latest_rows(
telegram_latest,
allowed_keys=(
"message_id",
"run_id",
"message_type",
"send_status",
"provider_message_id",
"incident_id",
"action",
"queued_at",
"sent_at",
),
),
},
"latest_flow_closure": latest_closure,
}
if error_type:
readback["error"] = {
"type": error_type,
"message": "runtime receipt DB read failed; see API logs",
}
return readback
def _attach_runtime_receipt_readback(
payload: dict[str, Any],
readback: dict[str, Any],
) -> dict[str, Any]:
payload["runtime_receipt_readback"] = readback
rollups = payload.setdefault("rollups", {})
rollups.update({
"live_ansible_apply_executed_count": _int_value(
readback.get("ansible_apply_executed", {}).get("total")
),
"live_post_apply_verifier_count": _int_value(
readback.get("post_apply_verifier", {}).get("total")
),
"live_km_writeback_count": _int_value(
readback.get("km_writeback", {}).get("total")
),
"live_telegram_receipt_count": _int_value(
readback.get("telegram_receipt", {}).get("total")
),
"live_executor_latest_flow_closed_count": (
1
if (readback.get("latest_flow_closure") or {}).get("closed") is True
else 0
),
})
return payload
def build_ai_agent_autonomous_runtime_control() -> dict[str, Any]:
"""Build the current AI Agent autonomy control-plane readback."""
@@ -240,10 +590,257 @@ def build_ai_agent_autonomous_runtime_control() -> dict[str, Any]:
"legacy_policy_overridden_count": len(legacy_overrides),
},
}
_attach_runtime_receipt_readback(
payload,
build_runtime_receipt_readback_from_rows(
project_id=_DEFAULT_PROJECT_ID,
db_read_status="not_queried",
),
)
_validate_payload(payload)
return payload
async def load_ai_agent_autonomous_runtime_receipt_readback(
*,
project_id: str = _DEFAULT_PROJECT_ID,
lookback_hours: int = _DEFAULT_LOOKBACK_HOURS,
limit: int = 20,
) -> dict[str, Any]:
"""Read live executor receipts without sending messages or mutating runtime state."""
params = {
"project_id": project_id,
"lookback_hours": max(1, int(lookback_hours or _DEFAULT_LOOKBACK_HOURS)),
"limit": max(1, int(limit or 20)),
}
try:
async with get_db_context(project_id) as db:
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
operation_counts = (
await db.execute(text(_RUNTIME_OPERATION_COUNTS_SQL), params)
).mappings().all()
operation_latest = (
await db.execute(text(_RUNTIME_OPERATION_LATEST_SQL), params)
).mappings().all()
verifier_counts = (
await db.execute(text(_RUNTIME_VERIFIER_COUNTS_SQL), params)
).mappings().all()
verifier_latest = (
await db.execute(text(_RUNTIME_VERIFIER_LATEST_SQL), params)
).mappings().all()
km_counts = (
await db.execute(text(_RUNTIME_KM_COUNTS_SQL), params)
).mappings().all()
km_latest = (
await db.execute(text(_RUNTIME_KM_LATEST_SQL), params)
).mappings().all()
telegram_counts = (
await db.execute(text(_RUNTIME_TELEGRAM_COUNTS_SQL), params)
).mappings().all()
telegram_latest = (
await db.execute(text(_RUNTIME_TELEGRAM_LATEST_SQL), params)
).mappings().all()
except Exception as exc:
logger.warning(
"ai_agent_autonomous_runtime_receipt_readback_failed",
project_id=project_id,
error_type=type(exc).__name__,
)
return build_runtime_receipt_readback_from_rows(
project_id=project_id,
lookback_hours=params["lookback_hours"],
db_read_status="unavailable",
error_type=type(exc).__name__,
)
return build_runtime_receipt_readback_from_rows(
project_id=project_id,
lookback_hours=params["lookback_hours"],
db_read_status="ok",
operation_count_rows=operation_counts,
operation_latest_rows=operation_latest,
verifier_count_rows=verifier_counts,
verifier_latest_rows=verifier_latest,
km_count_rows=km_counts,
km_latest_rows=km_latest,
telegram_count_rows=telegram_counts,
telegram_latest_rows=telegram_latest,
)
async def build_ai_agent_autonomous_runtime_control_with_live_readback(
*,
project_id: str = _DEFAULT_PROJECT_ID,
lookback_hours: int = _DEFAULT_LOOKBACK_HOURS,
) -> dict[str, Any]:
"""Build the control plane and attach live DB receipt readback."""
payload = build_ai_agent_autonomous_runtime_control()
readback = await load_ai_agent_autonomous_runtime_receipt_readback(
project_id=project_id,
lookback_hours=lookback_hours,
)
_attach_runtime_receipt_readback(payload, readback)
_validate_payload(payload)
return payload
_RUNTIME_OPERATION_COUNTS_SQL = """
SELECT
operation_type,
status,
count(*) AS total,
count(*) FILTER (
WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM automation_operation_log
WHERE operation_type IN (
'ansible_candidate_matched',
'ansible_check_mode_executed',
'ansible_apply_executed',
'ansible_rollback_executed',
'ansible_execution_skipped'
)
GROUP BY operation_type, status
ORDER BY operation_type, status
"""
_RUNTIME_OPERATION_LATEST_SQL = """
SELECT
op_id::text AS op_id,
parent_op_id::text AS parent_op_id,
operation_type,
status,
actor,
coalesce(incident_id::text, input ->> 'incident_id') AS incident_id,
input ->> 'catalog_id' AS catalog_id,
coalesce(input ->> 'apply_playbook_path', input ->> 'playbook_path') AS playbook_path,
input ->> 'execution_mode' AS execution_mode,
coalesce(output ->> 'returncode', dry_run_result ->> 'returncode') AS returncode,
duration_ms,
created_at
FROM automation_operation_log
WHERE operation_type IN (
'ansible_candidate_matched',
'ansible_check_mode_executed',
'ansible_apply_executed',
'ansible_rollback_executed',
'ansible_execution_skipped'
)
ORDER BY created_at DESC
LIMIT :limit
"""
_RUNTIME_VERIFIER_COUNTS_SQL = """
SELECT
coalesce(verification_result, 'missing') AS verification_result,
count(*) AS total,
count(*) FILTER (
WHERE collected_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM incident_evidence
WHERE post_execution_state ->> 'apply_op_id' IS NOT NULL
GROUP BY coalesce(verification_result, 'missing')
ORDER BY verification_result
"""
_RUNTIME_VERIFIER_LATEST_SQL = """
SELECT
id,
incident_id,
matched_playbook_id,
coalesce(verification_result, 'missing') AS verification_result,
post_execution_state ->> 'apply_op_id' AS apply_op_id,
post_execution_state ->> 'catalog_id' AS catalog_id,
post_execution_state ->> 'playbook_path' AS playbook_path,
post_execution_state ->> 'returncode' AS returncode,
collected_at
FROM incident_evidence
WHERE post_execution_state ->> 'apply_op_id' IS NOT NULL
ORDER BY collected_at DESC
LIMIT :limit
"""
_RUNTIME_KM_COUNTS_SQL = """
SELECT
status,
count(*) AS total,
count(*) FILTER (
WHERE created_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM knowledge_entries
WHERE project_id = :project_id
AND (
path_type LIKE 'ansible_apply_receipt:%'
OR tags::text LIKE '%ansible_controlled_apply%'
)
GROUP BY status
ORDER BY status
"""
_RUNTIME_KM_LATEST_SQL = """
SELECT
id,
title,
related_incident_id,
related_playbook_id,
path_type,
status,
created_by,
created_at
FROM knowledge_entries
WHERE project_id = :project_id
AND (
path_type LIKE 'ansible_apply_receipt:%'
OR tags::text LIKE '%ansible_controlled_apply%'
)
ORDER BY created_at DESC
LIMIT :limit
"""
_RUNTIME_TELEGRAM_COUNTS_SQL = """
SELECT
send_status,
count(*) AS total,
count(*) FILTER (
WHERE queued_at >= NOW() - (:lookback_hours * INTERVAL '1 hour')
) AS recent
FROM awooop_outbound_message
WHERE project_id = :project_id
AND channel_type = 'telegram'
AND source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result'
GROUP BY send_status
ORDER BY send_status
"""
_RUNTIME_TELEGRAM_LATEST_SQL = """
SELECT
message_id::text AS message_id,
run_id::text AS run_id,
message_type,
send_status,
provider_message_id,
source_envelope #>> '{callback_reply,incident_id}' AS incident_id,
source_envelope #>> '{callback_reply,action}' AS action,
queued_at,
sent_at
FROM awooop_outbound_message
WHERE project_id = :project_id
AND channel_type = 'telegram'
AND source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result'
ORDER BY queued_at DESC
LIMIT :limit
"""
def _validate_payload(payload: dict[str, Any]) -> None:
if payload.get("schema_version") != _SCHEMA_VERSION:
raise ValueError(f"schema_version must be {_SCHEMA_VERSION}")

23
apps/api/src/services/iwooos_runtime_security_readback.py
View File

@@ -183,10 +183,25 @@ def load_latest_iwooos_runtime_security_readback(
"lanes": [
_lane(
"wazuh_registry",
"blocked_waiting_manager_registry",
0,
"locked",
"管理器清單交叉驗收",
(
"manager_registry_readback_accepted_runtime_gate_closed"
if _int(wazuh_summary.get("manager_registry_accepted_count"))
>= _int(wazuh_summary.get("expected_host_scope_count"))
else "blocked_waiting_manager_registry"
),
(
35
if _int(wazuh_summary.get("manager_registry_accepted_count"))
>= _int(wazuh_summary.get("expected_host_scope_count"))
else 0
),
(
"steady"
if _int(wazuh_summary.get("manager_registry_accepted_count"))
>= _int(wazuh_summary.get("expected_host_scope_count"))
else "locked"
),
"管理器清單交叉驗收已讀回；runtime gate 仍關閉",
{
"expected_hosts": wazuh_summary.get("expected_host_scope_count", 0),
"transport_observed": wazuh_summary.get("manager_transport_established_connection_count", 0),

21
apps/api/src/services/iwooos_security_control_coverage.py
View File

@@ -259,13 +259,21 @@ def _build_domains(snapshots: dict[str, dict[str, Any]]) -> list[dict[str, Any]]
"domain_id": "wazuh_managed_host_coverage",
"label": "Wazuh 主機納管與 manager registry",
"priority": "P0",
"coverage_percent": 0,
"coverage_percent": 70 if _as_int(wazuh_summary.get("manager_registry_accepted_count")) else 0,
"scope_count": _as_int(wazuh_summary.get("expected_host_scope_count")),
"write_capable_count": _as_int(wazuh_summary.get("agent_reenroll_authorized_count")),
"accepted_count": _as_int(wazuh_summary.get("manager_registry_accepted_count")),
"blocked_count": _as_int(wazuh_summary.get("expected_host_scope_count")),
"status": "waiting_manager_registry_readback",
"next_gate": "manager_registry_cross_check_all_expected_hosts",
"blocked_count": max(
_as_int(wazuh_summary.get("expected_host_scope_count"))
- _as_int(wazuh_summary.get("manager_registry_accepted_count")),
0,
),
"status": (
"manager_registry_readback_accepted_runtime_gate_closed"
if _as_int(wazuh_summary.get("manager_registry_accepted_count"))
else "waiting_manager_registry_readback"
),
"next_gate": "runtime_gate_owner_review_and_postcheck",
"source_refs": ["docs/security/wazuh-managed-host-coverage-gate.snapshot.json"],
},
{
@@ -323,7 +331,10 @@ def _build_summary(snapshots: dict[str, dict[str, Any]], domains: list[dict[str,
"owner_response_received_count": 0,
"owner_response_accepted_count": 0,
"live_evidence_accepted_count": 0,
"wazuh_manager_registry_accepted_count": 0,
"wazuh_manager_registry_accepted_count": _summary_int(
snapshots["wazuh_hosts"],
"manager_registry_accepted_count",
),
"active_scan_authorized_count": 0,
"active_response_authorized_count": 0,
"telegram_send_authorized_count": 0,

32
apps/api/src/services/iwooos_wazuh_managed_host_coverage.py
View File

@@ -32,15 +32,16 @@ _REQUIRED_FALSE_BOUNDARIES = {
}
_STATUS_LABELS = {
"agent_active_transport_observed": "代理服務與傳輸連線已觀察，仍待管理器清單驗收",
"no_agent_transport_observed": "未觀察到代理傳輸，需確認安裝、服務與負責人決策",
"ssh_readback_blocked": "合法只讀讀回仍受阻，需 owner export 或維護窗口",
"agent_active_transport_observed": "代理服務與傳輸連線已觀察，管理器清單 evidence 已驗收",
"no_agent_transport_observed": "管理器清單 evidence 已驗收，仍需 runtime/服務狀態獨立確認",
"ssh_readback_blocked": "管理器清單 evidence 已驗收，合法只讀讀回與 runtime postcheck 仍需獨立 gate",
}
_NEXT_GATE_LABELS = {
"manager_registry_cross_check": "補管理器清單交叉驗收",
"agent_install_or_service_owner_decision": "補代理安裝狀態或服務負責人決策",
"read_only_access_or_owner_export": "補只讀 access 或脫敏 owner export",
"runtime_gate_owner_review": "進 runtime gate owner review 與 postcheck",
}
@@ -116,7 +117,7 @@ def load_latest_iwooos_wazuh_managed_host_coverage(
"no_false_green_rules": [
"Wazuh Dashboard 可見不等於 manager registry 已恢復",
"transport 連線、agent service active 或 HTTP 200 不可替代逐主機 registry matrix",
"manager_registry_accepted_count 維持 0 時不得宣稱所有主機已納管",
"manager registry accepted 只代表 committed redacted evidence 覆蓋公開別名，不代表 runtime 授權",
"重新註冊、重啟、active response、主機寫入、Nginx、firewall、Kali 掃描與機密調整都不是此讀回授權",
],
}
@@ -165,8 +166,7 @@ def _host_scope_matrix(payload: dict[str, Any]) -> list[dict[str, Any]]:
next_gate = str(raw_item.get("next_gate", ""))
if not node_id.startswith("managed_"):
raise ValueError(f"Wazuh 受管主機覆蓋 node_id 必須維持公開別名：{node_id}")
if raw_item.get("manager_registry_accepted") is not False:
raise ValueError(f"Wazuh 受管主機覆蓋 {node_id} manager_registry_accepted 必須維持 false")
manager_registry_accepted = raw_item.get("manager_registry_accepted") is True
matrix.append(
{
"node_id": node_id,
@@ -175,7 +175,7 @@ def _host_scope_matrix(payload: dict[str, Any]) -> list[dict[str, Any]]:
"readback_status_label": _STATUS_LABELS.get(readback_status, "待補只讀讀回"),
"next_gate": next_gate,
"next_gate_label": _NEXT_GATE_LABELS.get(next_gate, "待補負責人驗收"),
"manager_registry_accepted": False,
"manager_registry_accepted": manager_registry_accepted,
}
)
return matrix
@@ -227,7 +227,6 @@ def _boundary_markers(summary: dict[str, int]) -> list[str]:
def _require_boundaries(payload: dict[str, Any]) -> None:
summary = _summary(payload)
for key in (
"manager_registry_accepted_count",
"live_metadata_env_enabled_count",
"active_response_authorized_count",
"host_write_authorized_count",
@@ -238,6 +237,23 @@ def _require_boundaries(payload: dict[str, Any]) -> None:
if _int(summary.get(key)) != 0:
raise ValueError(f"Wazuh 受管主機覆蓋 summary.{key} 必須維持 0")
expected_hosts = _int(summary.get("expected_host_scope_count"))
manager_accepted = _int(summary.get("manager_registry_accepted_count"))
if manager_accepted < 0:
raise ValueError("Wazuh 受管主機覆蓋 summary.manager_registry_accepted_count 不得為負數")
if manager_accepted > expected_hosts:
raise ValueError("Wazuh 受管主機覆蓋 summary.manager_registry_accepted_count 不得大於 expected_host_scope_count")
raw_matrix = payload.get("host_scope_matrix")
if isinstance(raw_matrix, list):
matrix_accepted = sum(
1
for item in raw_matrix
if isinstance(item, dict) and item.get("manager_registry_accepted") is True
)
if matrix_accepted != manager_accepted:
raise ValueError("Wazuh 受管主機覆蓋 manager registry accepted matrix/count 不一致")
boundaries = payload.get("execution_boundaries")
if not isinstance(boundaries, dict):
raise ValueError("Wazuh 受管主機覆蓋 execution_boundaries 缺失")

22
apps/api/src/services/iwooos_wazuh_manager_registry_reviewer_validation.py
View File

@@ -317,7 +317,6 @@ def _boundary_markers(summary: dict[str, int]) -> list[str]:
def _require_boundaries(payload: dict[str, Any]) -> None:
summary = _summary(payload)
for key in (
"manager_registry_accepted_count",
"runtime_gate_count",
"host_write_authorized_count",
"active_response_authorized_count",
@@ -326,6 +325,8 @@ def _require_boundaries(payload: dict[str, Any]) -> None:
if _int(summary.get(key)) != 0:
raise ValueError(f"Wazuh manager registry reviewer validation summary.{key} 必須維持 0")
expected_aliases = _int(summary.get("expected_scope_alias_count"))
manager_accepted = _int(summary.get("manager_registry_accepted_count"))
received = _int(summary.get("owner_registry_export_received_count"))
accepted = _int(summary.get("owner_registry_export_accepted_count"))
ready = _int(summary.get("reviewer_validation_ready_count"))
@@ -349,9 +350,12 @@ def _require_boundaries(payload: dict[str, Any]) -> None:
acceptance_endpoint,
acceptance_received,
acceptance_ready,
manager_accepted,
)
):
raise ValueError("Wazuh manager registry reviewer validation counters 不得為負數")
if manager_accepted > expected_aliases:
raise ValueError("manager_registry_accepted_count 不得大於 expected_scope_alias_count")
if accepted > received:
raise ValueError("owner_registry_export_accepted_count 不得大於 received_count")
if ready > received:
@@ -366,8 +370,20 @@ def _require_boundaries(payload: dict[str, Any]) -> None:
raise ValueError(
"manager_registry_acceptance_evidence_review_ready_count 不得大於 received_count"
)
if acceptance_ready:
raise ValueError("manager_registry_acceptance_evidence_review_ready_count 目前不得在全域讀回中自動上修")
if manager_accepted and not all(
value > 0
for value in (
received,
accepted,
ready,
passed,
post_enable,
acceptance_endpoint,
acceptance_received,
acceptance_ready,
)
):
raise ValueError("manager_registry_accepted_count 上修前必須先有完整 committed acceptance evidence")
if failed and passed:
raise ValueError("reviewer_validation_failed_count 與 passed_count 不得同時為正")
if quarantined and accepted: