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

feat(telegram): verify alert context receipts
Some checks failed
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Failing after 1m47s
Details
CD Pipeline / build-and-deploy (push) Has been skipped
Details
CD Pipeline / post-deploy-checks (push) Has been skipped
Details

Browse Source
This commit is contained in:
Your Name
2026-07-02 20:59:59 +08:00
parent ef69616b6b
commit d99cb7b2a5
9 changed files with 901 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
apps/api/src/services/awoooi_priority_work_order_readback.py
View File

@@ -3899,6 +3899,7 @@ def apply_ai_automation_live_closure_readbacks(
executor_readback: dict[str, Any] | None = None,
verifier_readback: dict[str, Any] | None = None,
consumer_readback: dict[str, Any] | None = None,
telegram_alert_context_verifier_readback: dict[str, Any] | None = None,
) -> None:
"""Overlay live AI-loop executor/verifier/consumer readbacks onto receipts."""
@@ -4056,6 +4057,54 @@ def apply_ai_automation_live_closure_readbacks(
"ai_automation_consumer_runtime_target_write_performed"
]
telegram_verifier = _dict(telegram_alert_context_verifier_readback)
telegram_verifier_rollups = _dict(telegram_verifier.get("rollups"))
telegram_verifier_blockers = _strings(telegram_verifier.get("active_blockers"))
if telegram_verifier:
summary["ai_automation_telegram_alert_post_apply_verifier_status"] = str(
telegram_verifier.get("status") or ""
)
summary["ai_automation_telegram_alert_post_apply_verifier_ready"] = bool(
telegram_verifier_rollups.get(
"telegram_alert_learning_context_post_apply_verifier_ready"
)
is True
)
summary[
"ai_automation_telegram_alert_post_apply_verified_context_receipt_count"
] = _int(telegram_verifier_rollups.get("verified_context_receipt_count"))
summary[
"ai_automation_telegram_alert_post_apply_verified_target_count"
] = _int(telegram_verifier_rollups.get("verified_target_count"))
summary[
"ai_automation_telegram_alert_post_apply_verified_ai_agent_context_receipt_count"
] = _int(
telegram_verifier_rollups.get("verified_ai_agent_context_receipt_count")
)
summary[
"ai_automation_telegram_alert_post_apply_verifier_ref_count"
] = _int(telegram_verifier_rollups.get("post_apply_verifier_ref_count"))
summary["ai_automation_telegram_alert_post_apply_verifier_active_blockers"] = (
telegram_verifier_blockers
)
rollups["ai_automation_telegram_alert_post_apply_verifier_ready"] = summary[
"ai_automation_telegram_alert_post_apply_verifier_ready"
]
rollups[
"ai_automation_telegram_alert_post_apply_verified_context_receipt_count"
] = summary[
"ai_automation_telegram_alert_post_apply_verified_context_receipt_count"
]
rollups[
"ai_automation_telegram_alert_post_apply_verified_target_count"
] = summary["ai_automation_telegram_alert_post_apply_verified_target_count"]
rollups[
"ai_automation_telegram_alert_post_apply_verifier_active_blocker_count"
] = len(telegram_verifier_blockers)
state["ai_automation_telegram_alert_post_apply_verifier_ready"] = summary[
"ai_automation_telegram_alert_post_apply_verifier_ready"
]
apply_ai_automation_node_receipts(payload)
@@ -4136,6 +4185,20 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
telegram_alert_ai_agent_context_receipt_count = _int(
summary.get("ai_automation_telegram_alert_ai_agent_context_receipt_count")
)
telegram_alert_post_apply_verifier_ready = bool(
summary.get("ai_automation_telegram_alert_post_apply_verifier_ready") is True
)
telegram_alert_post_apply_verified_context_receipt_count = _int(
summary.get(
"ai_automation_telegram_alert_post_apply_verified_context_receipt_count"
)
)
telegram_alert_post_apply_verified_target_count = _int(
summary.get("ai_automation_telegram_alert_post_apply_verified_target_count")
)
telegram_alert_post_apply_verifier_ref_count = _int(
summary.get("ai_automation_telegram_alert_post_apply_verifier_ref_count")
)
consumer_apply_route = str(summary.get("ai_automation_consumer_apply_route") or "")
normalizer_ready = bool(
queue_fields
@@ -4153,6 +4216,7 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
post_verifier_ready = bool(
receipt_outputs
or verifier_dry_run_ready
or telegram_alert_post_apply_verifier_ready
or consumer_post_apply_verifier_ref_count > 0
)
learning_writeback_ready = bool(
@@ -4290,13 +4354,29 @@ def _apply_ai_automation_node_receipts(payload: dict[str, Any]) -> None:
*receipt_outputs[:4],
f"post_write_verifier_nodes:{verifier_passed_node_count}",
f"consumer_post_apply_verifier_refs:{consumer_post_apply_verifier_ref_count}",
(
"telegram_alert_post_apply_verified_context_receipts:"
f"{telegram_alert_post_apply_verified_context_receipt_count}"
),
(
"telegram_alert_post_apply_verified_targets:"
f"{telegram_alert_post_apply_verified_target_count}"
),
(
"telegram_alert_post_apply_verifier_refs:"
f"{telegram_alert_post_apply_verifier_ref_count}"
),
],
"verifier_result": "post_verifier_receipt_readback"
if receipt_outputs
else (
"verifier_result": (
"telegram_alert_learning_context_post_apply_verified"
if telegram_alert_post_apply_verifier_ready
else "post_verifier_receipt_readback"
if receipt_outputs
else (
"post_write_verifier_dry_run_ready"
if verifier_dry_run_ready
else "post_verifier_required"
)
),
"next_action": "read back route/API/UI/verifier output before closing",
},

88
apps/api/src/services/telegram_alert_ai_automation_matrix.py
View File

@@ -90,6 +90,9 @@ def load_latest_telegram_alert_ai_automation_matrix(
ai_loop_context_present = bool(
source_proof["telegram_alert_ai_loop_consumer_context_readback_present"]
)
post_apply_verifier_present = bool(
source_proof["telegram_alert_post_apply_verifier_readback_present"]
)
direct_gap_count = int(egress_summary["direct_bot_api_call_count"])
direct_gap_file_count = int(egress_summary["direct_bot_api_file_count"])
@@ -316,6 +319,46 @@ def load_latest_telegram_alert_ai_automation_matrix(
"/api/v1/agents/agent-log-controlled-writeback-consumer-readback",
],
},
{
"surface_id": "telegram_ai_loop_post_apply_verifier",
"priority_work_item_id": "CIR-P0-TG-001",
"surface_kind": "ai_loop_post_apply_verifier",
"status": (
"post_apply_verifier_readback_present"
if post_apply_verifier_present
else "post_apply_verifier_readback_missing"
),
"known_item_count": source_proof[
"telegram_alert_post_apply_verifier_readback_count"
],
"db_or_log_receipt": "ready_ai_loop_context_receipt_source",
"ai_route": (
"ready_verified_context_reuse"
if post_apply_verifier_present
else "missing_verified_context_reuse"
),
"controlled_queue": (
"ready_metadata_only_verifier"
if post_apply_verifier_present
else "missing_metadata_only_verifier"
),
"post_verifier": (
"ready_context_receipt_post_apply_verifier"
if post_apply_verifier_present
else "missing_context_receipt_post_apply_verifier"
),
"learning_writeback": (
"ready_verified_context_receipt_for_km_playbook_trust"
if post_apply_verifier_present
else "missing_verified_context_receipt"
),
"manual_default_gap_count": 0,
"direct_send_gap_count": 0,
"evidence_refs": [
"apps/api/src/services/telegram_alert_learning_context_post_apply_verifier.py",
"/api/v1/agents/telegram-alert-learning-context-post-apply-verifier",
],
},
]
next_priority_action = _next_priority_action(
@@ -325,6 +368,7 @@ def load_latest_telegram_alert_ai_automation_matrix(
learning_writeback_refs_present=learning_writeback_refs_present,
learning_registry_present=learning_registry_present,
ai_loop_context_present=ai_loop_context_present,
post_apply_verifier_present=post_apply_verifier_present,
)
next_controlled_actions = _ordered_controlled_actions(next_priority_action)
summary = _build_summary(
@@ -382,9 +426,17 @@ def load_latest_telegram_alert_ai_automation_matrix(
(
(
(
"Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
"Telegram alert registry is now consumable by AI Loop Agent context, "
"so the next P0 step is post-apply verifier receipt."
(
"Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
"Telegram alert AI Loop context has a post-apply verifier readback, "
"so the next P0 step is surfacing verified receipts in Runs / Work Items / Alerts."
)
if post_apply_verifier_present
else (
"Manual/default semantics are only acceptable for critical/break-glass or historical evidence; "
"Telegram alert registry is now consumable by AI Loop Agent context, "
"so the next P0 step is post-apply verifier receipt."
)
)
if ai_loop_context_present
else (
@@ -543,6 +595,10 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
"apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py",
"telegram_alert_ai_loop_consumer_context_readback_v1",
),
"telegram_alert_post_apply_verifier_readback_present": (
"apps/api/src/services/telegram_alert_learning_context_post_apply_verifier.py",
"telegram_alert_learning_context_post_apply_verifier_v1",
),
}
result: dict[str, int | bool] = {}
for key, (relative_path, marker) in source_checks.items():
@@ -561,6 +617,10 @@ def _inspect_source_contract(repo_root: Path) -> dict[str, int | bool]:
result["telegram_alert_ai_loop_consumer_context_readback_count"] = (
1 if present else 0
)
if key == "telegram_alert_post_apply_verifier_readback_present":
result["telegram_alert_post_apply_verifier_readback_count"] = (
1 if present else 0
)
required = [key for key, value in result.items() if key.endswith("_present") and value is not True]
if required:
@@ -588,6 +648,7 @@ def _next_priority_action(
learning_writeback_refs_present: bool,
learning_registry_present: bool,
ai_loop_context_present: bool,
post_apply_verifier_present: bool,
) -> dict[str, Any]:
if api_direct_gap_count > 0:
return {
@@ -649,14 +710,24 @@ def _next_priority_action(
"requires_runtime_send": False,
"post_verifier": "AI Loop Agent context readback references the CIR-P0-TG-001 Telegram alert learning registry.",
}
if not post_apply_verifier_present:
return {
"action_id": "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier",
"scope": "AI Loop Agent Telegram alert consumer context -> post-apply verifier receipt",
"priority": "P0",
"why": "Telegram alert learning registry is readable by AI Loop Agent context; the next closure is verifier receipt and controlled apply reuse.",
"requires_secret": False,
"requires_runtime_send": False,
"post_verifier": "AI Loop post-apply verifier reads Telegram alert context receipts and reports reusable decision context.",
}
return {
"action_id": "verify_telegram_alert_learning_context_receipts_with_ai_loop_post_apply_verifier",
"scope": "AI Loop Agent Telegram alert consumer context -> post-apply verifier receipt",
"action_id": "surface_verified_telegram_alert_context_receipts_in_awooop_runs_work_items_alerts",
"scope": "Verified Telegram alert context receipts -> AwoooP Runs / Work Items / Alerts",
"priority": "P0",
"why": "Telegram alert learning registry is readable by AI Loop Agent context; the next closure is verifier receipt and controlled apply reuse.",
"why": "Post-apply verifier is source-controlled; the next closure is visible AI automation proof across operator surfaces.",
"requires_secret": False,
"requires_runtime_send": False,
"post_verifier": "AI Loop post-apply verifier reads Telegram alert context receipts and reports reusable decision context.",
"post_verifier": "Runs, Work Items, and Alerts surfaces show verified Telegram alert AI Loop receipts without manual-default terminal states.",
}
@@ -725,6 +796,9 @@ def _build_summary(
"telegram_alert_ai_loop_consumer_context_readback_count": source_proof[
"telegram_alert_ai_loop_consumer_context_readback_count"
],
"telegram_alert_post_apply_verifier_readback_count": source_proof[
"telegram_alert_post_apply_verifier_readback_count"
],
"next_priority_action_id": next_priority_action["action_id"],
"next_priority_work_item_id": "CIR-P0-TG-001",
}

350
apps/api/src/services/telegram_alert_learning_context_post_apply_verifier.py Normal file
View File

@@ -0,0 +1,350 @@
"""Telegram alert learning context post-apply verifier readback.
Verifies metadata-only Telegram alert learning context receipts emitted by the
AI Loop consumer readback. This module only reads public-safe receipt metadata;
it does not write KM/RAG/PlayBook, call MCP tools, send Telegram, or touch
runtime infrastructure.
"""
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from src.services.ai_agent_log_controlled_writeback_consumer_readback import (
load_latest_ai_agent_log_controlled_writeback_consumer_readback,
)
SCHEMA_VERSION = "telegram_alert_learning_context_post_apply_verifier_v1"
DEFAULT_PROJECT_ID = "awoooi"
_TARGETS = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
_REQUIRED_TARGET_SELECTOR_FIELDS = (
"project_id",
"run_id",
"message_id",
"target",
"source_ref",
)
_REQUIRED_CHECKS = (
"ai_alert_card_delivery_receipt_present",
"learning_registry_binding_ready",
"metadata_only_raw_payload_absent",
"ai_agent_context_ref_present",
)
async def load_latest_telegram_alert_learning_context_post_apply_verifier(
*,
project_id: str = DEFAULT_PROJECT_ID,
) -> dict[str, Any]:
"""Return post-apply verifier results for Telegram alert context receipts."""
consumer_readback = (
await load_latest_ai_agent_log_controlled_writeback_consumer_readback(
project_id=project_id
)
)
return build_telegram_alert_learning_context_post_apply_verifier(
consumer_readback=consumer_readback,
project_id=project_id,
)
def build_telegram_alert_learning_context_post_apply_verifier(
*,
consumer_readback: Mapping[str, Any],
project_id: str = DEFAULT_PROJECT_ID,
) -> dict[str, Any]:
"""Build a deterministic verifier readback from consumer context receipts."""
consumer = _dict(consumer_readback)
telegram_context = _dict(consumer.get("telegram_alert_learning_context"))
context_receipts = [
_dict(receipt) for receipt in _list(telegram_context.get("context_receipts"))
]
verifier_results = [
_verify_context_receipt(receipt) for receipt in context_receipts
]
target_count = _safe_int(telegram_context.get("target_count")) or len(_TARGETS)
verified_targets = sorted(
{
str(item["target"])
for item in verifier_results
if item["verifier_status"] == "passed"
}
)
verified_ai_agent_count = sum(
1
for item in verifier_results
if item["target"] == "ai_agent" and item["verifier_status"] == "passed"
)
failed_results = [
item for item in verifier_results if item["verifier_status"] != "passed"
]
source_ready = (
telegram_context.get("status") == "ai_loop_agent_context_receipt_ready"
)
ready = bool(
source_ready
and context_receipts
and not failed_results
and len(verified_targets) == target_count
and verified_ai_agent_count > 0
)
active_blockers = _active_blockers(
source_ready=source_ready,
context_receipts=context_receipts,
failed_results=failed_results,
verified_targets=verified_targets,
target_count=target_count,
verified_ai_agent_count=verified_ai_agent_count,
)
return {
"schema_version": SCHEMA_VERSION,
"priority": "P0-TELEGRAM-AI-LOOP-POST-APPLY-VERIFIER",
"priority_work_item_id": "CIR-P0-TG-001",
"project_id": project_id,
"status": (
"telegram_alert_learning_context_post_apply_verified"
if ready
else "blocked_telegram_alert_learning_context_post_apply_verifier"
),
"source_readback": {
"source_endpoint": (
"/api/v1/agents/agent-log-controlled-writeback-consumer-readback"
),
"source_schema_version": consumer.get("schema_version"),
"telegram_context_schema_version": telegram_context.get("schema_version"),
"telegram_context_status": telegram_context.get("status"),
"consumer_context_route": telegram_context.get("consumer_context_route"),
"source_registry_status": telegram_context.get("source_registry_status"),
},
"controlled_verifier": {
"mode": "metadata_only_post_apply_verifier_readback",
"post_apply_verifier_executed": True,
"controlled_apply_allowed_for_low_medium_high": ready,
"owner_review_required_for_low_medium_high": False,
"critical_break_glass_required": True,
"target_selector_verified": ready,
"source_of_truth_diff_verified": ready,
"check_mode_verified": ready,
"rollback_verified": ready,
"raw_payload_absence_verified": ready,
},
"verifier_results": verifier_results,
"rollups": {
"telegram_alert_learning_context_post_apply_verifier_ready": ready,
"source_context_ready": source_ready,
"context_receipt_count": len(context_receipts),
"verified_context_receipt_count": sum(
1 for item in verifier_results if item["verifier_status"] == "passed"
),
"failed_context_receipt_count": len(failed_results),
"target_count": target_count,
"verified_target_count": len(verified_targets),
"verified_targets": verified_targets,
"verified_ai_agent_context_receipt_count": verified_ai_agent_count,
"target_selector_verified_count": sum(
1 for item in verifier_results if item["target_selector_verified"]
),
"source_of_truth_diff_verified_count": sum(
1
for item in verifier_results
if item["source_of_truth_diff_verified"]
),
"check_mode_verified_count": sum(
1 for item in verifier_results if item["check_mode_verified"]
),
"rollback_verified_count": sum(
1 for item in verifier_results if item["rollback_verified"]
),
"post_apply_verifier_ref_count": sum(
len(item["post_apply_verifier_refs"]) for item in verifier_results
),
"metadata_only_receipt_count": sum(
1 for item in verifier_results if item["metadata_only"] is True
),
"raw_payload_included_count": sum(
1 for item in verifier_results if item["raw_payload_included"] is True
),
"runtime_write_performed_count": 0,
"telegram_send_performed_count": 0,
"mcp_tool_call_performed_count": 0,
"secret_value_read_count": 0,
},
"active_blockers": active_blockers,
"next_action": (
"surface_verified_telegram_alert_context_receipts_in_awooop_runs_work_items_alerts"
if ready
else "fix_failed_telegram_alert_context_receipts_then_rerun_post_apply_verifier"
),
"operation_boundaries": {
"metadata_read_performed": True,
"post_apply_verifier_read_performed": True,
"runtime_target_write_performed": False,
"km_write_performed": False,
"rag_index_write_performed": False,
"playbook_trust_write_performed": False,
"mcp_tool_call_performed": False,
"agent_runtime_action_performed": False,
"telegram_send_performed": False,
"workflow_trigger_performed": False,
"raw_payload_included": False,
"secret_value_collection_allowed": False,
"github_api_used": False,
},
}
def _verify_context_receipt(receipt: dict[str, Any]) -> dict[str, Any]:
target = str(receipt.get("target") or "")
target_selector = _dict(receipt.get("target_selector"))
source_diff = _dict(receipt.get("source_of_truth_diff"))
check_mode = _dict(receipt.get("check_mode"))
rollback = _dict(receipt.get("rollback"))
post_apply_verifier = _dict(receipt.get("post_apply_verifier"))
verifier_refs = _strings(post_apply_verifier.get("verifier_refs"))
target_selector_missing = [
key for key in _REQUIRED_TARGET_SELECTOR_FIELDS if not target_selector.get(key)
]
checks = _strings(check_mode.get("checks"))
missing_checks = [check for check in _REQUIRED_CHECKS if check not in checks]
failed_checks: list[str] = []
if target not in _TARGETS:
failed_checks.append("target_in_allowed_set")
if receipt.get("status") != "ready_for_ai_loop_context":
failed_checks.append("receipt_ready_for_ai_loop_context")
if target_selector_missing:
failed_checks.append("target_selector_required_fields_present")
if not _source_diff_verified(source_diff, target):
failed_checks.append("source_of_truth_diff_verified")
if check_mode.get("enabled") is not True or missing_checks:
failed_checks.append("check_mode_required_checks_present")
if rollback.get("required") is not True or not rollback.get("rollback_ref"):
failed_checks.append("rollback_ref_present")
if post_apply_verifier.get("required") is not True or not verifier_refs:
failed_checks.append("post_apply_verifier_refs_present")
if receipt.get("metadata_only") is not True:
failed_checks.append("metadata_only_receipt")
if receipt.get("raw_payload_included") is not False:
failed_checks.append("raw_payload_absent")
if receipt.get("target_write_performed") is not False:
failed_checks.append("target_write_not_performed_by_verifier")
if not receipt.get("ai_agent_context_ref"):
failed_checks.append("ai_agent_context_ref_present")
passed_checks = [
"target_in_allowed_set",
"receipt_ready_for_ai_loop_context",
"target_selector_required_fields_present",
"source_of_truth_diff_verified",
"check_mode_required_checks_present",
"rollback_ref_present",
"post_apply_verifier_refs_present",
"metadata_only_receipt",
"raw_payload_absent",
"target_write_not_performed_by_verifier",
"ai_agent_context_ref_present",
]
passed_checks = [check for check in passed_checks if check not in failed_checks]
return {
"receipt_id": str(receipt.get("receipt_id") or ""),
"source_receipt_id": str(receipt.get("source_receipt_id") or ""),
"source_run_id": str(receipt.get("source_run_id") or ""),
"source_message_id": str(receipt.get("source_message_id") or ""),
"work_item_id": str(receipt.get("work_item_id") or "CIR-P0-TG-001"),
"target": target,
"consumer_surface": str(receipt.get("consumer_surface") or ""),
"verifier_status": "passed" if not failed_checks else "failed",
"passed_checks": passed_checks,
"failed_checks": failed_checks,
"target_selector_verified": not target_selector_missing,
"target_selector_missing_fields": target_selector_missing,
"source_of_truth_diff_verified": _source_diff_verified(source_diff, target),
"check_mode_verified": check_mode.get("enabled") is True and not missing_checks,
"check_mode_missing_checks": missing_checks,
"rollback_verified": (
rollback.get("required") is True and bool(rollback.get("rollback_ref"))
),
"post_apply_verifier_refs": verifier_refs,
"metadata_only": receipt.get("metadata_only") is True,
"raw_payload_included": receipt.get("raw_payload_included") is True,
"runtime_target_write_performed": False,
"telegram_send_performed": False,
"evidence_refs": _unique(
[
str(receipt.get("source_ref") or ""),
str(receipt.get("ai_agent_context_ref") or ""),
*verifier_refs,
]
),
}
def _active_blockers(
*,
source_ready: bool,
context_receipts: list[dict[str, Any]],
failed_results: list[dict[str, Any]],
verified_targets: list[str],
target_count: int,
verified_ai_agent_count: int,
) -> list[str]:
blockers: list[str] = []
if not source_ready:
blockers.append("telegram_alert_learning_context_source_not_ready")
if not context_receipts:
blockers.append("telegram_alert_learning_context_receipts_missing")
if len(verified_targets) < target_count:
blockers.append("telegram_alert_learning_context_verified_target_count_below_target")
if verified_ai_agent_count <= 0:
blockers.append("telegram_alert_learning_context_ai_agent_receipt_not_verified")
for item in failed_results:
receipt_id = item.get("receipt_id") or "unknown-receipt"
blockers.append(f"telegram_alert_context_receipt_verifier_failed:{receipt_id}")
return blockers
def _source_diff_verified(source_diff: dict[str, Any], target: str) -> bool:
return bool(
source_diff.get("current_state") == "telegram_alert_learning_registry_readable"
and source_diff.get("desired_state")
== "ai_loop_agent_context_receipt_available"
and source_diff.get("delta_kind") == f"telegram_alert_{target}_context_binding"
and source_diff.get("raw_payload_included") is False
)
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _list(value: Any) -> list[Any]:
return value if isinstance(value, list) else []
def _strings(value: Any) -> list[str]:
return [str(item) for item in _list(value) if str(item)]
def _unique(values: list[str]) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for value in values:
if value in seen:
continue
seen.add(value)
result.append(value)
return result
def _safe_int(value: Any) -> int:
try:
return int(value)
except (TypeError, ValueError):
return 0