Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions loopx/chat_action_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -411,6 +411,49 @@ def record_operation_delivery(
self._write(payload)
return proposal

def record_operation_delivery_snapshot(
self,
proposal_id: str,
*,
submitted_card: Mapping[str, Any],
) -> dict[str, Any]:
"""Persist the immutable sent card after verifying its recorded digest."""

safe_card = _safe_json_value(
dict(submitted_card), path="operation.delivery.submitted_card"
)
if not isinstance(safe_card, dict):
raise ValueError("operation submitted card must be an object")
token = _opaque_id(proposal_id, field="proposal_id")
with exclusive_file_lock(
self.path,
agent_id="loopx-chat",
operation="record_operation_delivery_snapshot",
):
payload = self._read()
proposal = payload["proposals"].get(token)
operation = (
proposal.get("operation") if isinstance(proposal, dict) else None
)
delivery = (
operation.get("delivery") if isinstance(operation, dict) else None
)
if not isinstance(delivery, dict):
raise ActionConflictError("operation card delivery was not recorded")
if _canonical_digest(safe_card) != delivery.get("card_digest"):
raise ActionConflictError("operation submitted card digest drifted")
existing = delivery.get("submitted_card")
if existing is not None:
if existing != safe_card:
raise ActionConflictError(
"operation submitted card snapshot already differs"
)
return proposal
delivery["submitted_card"] = safe_card
proposal["updated_at"] = _utc_now()
self._write(payload)
return proposal

def decide_operation(
self,
proposal_id: str,
Expand Down
9 changes: 9 additions & 0 deletions loopx/extensions/lark/event_collector.py
Original file line number Diff line number Diff line change
Expand Up @@ -735,6 +735,15 @@ def inspect_lark_event_collector(
"operation_callback_last_evidence_at": callback_status.get(
"last_verified_callback_at"
),
"operation_callback_last_failure_code": callback_status.get(
"last_failure_code"
),
"operation_callback_last_failure_stage": callback_status.get(
"last_failure_stage"
),
"operation_callback_last_failure_event_shape": callback_status.get(
"last_failure_event_shape"
),
"operation_callback_console_configuration_preflighted": False,
"thread_complete": all(
route["inbox"]["thread_complete"] for route in config["routes"]
Expand Down
136 changes: 133 additions & 3 deletions loopx/extensions/lark/event_collector_runtime.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from __future__ import annotations

import json
import hashlib
import re
import signal
import subprocess
Expand Down Expand Up @@ -31,10 +32,128 @@
APP_ID_PATTERN = re.compile(r"cli_[A-Za-z0-9_-]+")
EVENT_READY_PREFIX = "[event] ready "
EVENT_DIAGNOSTIC_PREFIX = "[event] "
_CALLBACK_FAILURE_CODES = {
"collector Bot application identity is unverified": "collector_app_identity_unverified",
"operation callback event type is unsupported": "callback_event_type_unsupported",
"operation callback must come from a button": "callback_action_not_button",
"operation callback action_value is invalid": "callback_action_value_invalid",
"operation callback action is incomplete": "callback_action_incomplete",
"operation callback action schema is unsupported": "callback_action_schema_unsupported",
"operation callback decision is unsupported": "callback_decision_unsupported",
"operation callback update token is invalid": "callback_update_token_invalid",
"operation callback event_id is invalid": "callback_event_id_invalid",
"operation callback message_id is invalid": "callback_message_id_invalid",
"operation callback chat_id is invalid": "callback_chat_id_invalid",
"operation callback operator_id is invalid": "callback_operator_id_invalid",
"operation callback host is unsupported": "callback_host_unsupported",
"operation callback card content is unavailable": "callback_card_content_unavailable",
"operation callback proposal was not found": "callback_proposal_not_found",
"typed operation proposal is unavailable": "callback_operation_unavailable",
"typed operation envelope is unavailable": "callback_operation_envelope_unavailable",
"operation review plan is unavailable": "callback_review_plan_unavailable",
"operation review frame is unavailable": "callback_review_frame_unavailable",
"operation projection is unavailable": "callback_projection_unavailable",
"operation projection fields are unavailable": "callback_projection_fields_unavailable",
"operation callback timestamp is invalid": "callback_timestamp_invalid",
"operation confirmation has unsupported or missing fields": "callback_confirmation_invalid",
"operation timestamps require a timezone": "callback_timestamp_timezone_missing",
"claimed operation disappeared before dispatch": "callback_claim_disappeared",
"operation executor outcome does not match the consumed claim": "callback_executor_outcome_invalid",
"operation disappeared before result delivery": "callback_operation_disappeared",
"operation card delivery was not recorded": "delivery_not_recorded",
"operation callback digest drifted": "confirmation_digest_drifted",
"recorded operation card digest drifted": "recorded_card_digest_drifted",
"operation callback card content drifted": "callback_card_projection_drifted",
"operation callback app identity drifted": "callback_app_identity_drifted",
"principal is not authorized for this operation": "principal_not_authorized",
"operation callback tenant membership is unverified": "membership_unverified",
"operation callback does not match the delivered request": "delivery_binding_mismatch",
"operation is not awaiting confirmation": "operation_not_awaiting_confirmation",
"operation confirmation arrived after expiry": "operation_expired",
"operation callback result delivery was not verified": "result_delivery_unverified",
}
_CALLBACK_FAILURE_STAGES = {
"_callback_action": "parse_action",
"_callback_timestamp": "validate_timestamp",
"_callback_card_content_matches": "verify_card_content",
"_operator_membership_verified": "verify_operator_membership",
"decide_operation": "claim_operation",
"_execute_claimed_operation": "execute_operation",
"_update_callback_card": "deliver_result",
}
CommandRunner = Callable[..., subprocess.CompletedProcess[str]]
Sleeper = Callable[[float], None]


def _operation_callback_failure_code(exc: BaseException) -> str:
return _CALLBACK_FAILURE_CODES.get(str(exc), "callback_rejected")


def _operation_callback_failure_stage(exc: BaseException) -> str:
"""Return a value-free processing stage for a rejected callback."""

stage = "handle_callback"
traceback = exc.__traceback__
while traceback is not None:
stage = _CALLBACK_FAILURE_STAGES.get(
traceback.tb_frame.f_code.co_name,
stage,
)
traceback = traceback.tb_next
return stage


def _callback_event_shape(payload: Mapping[str, Any]) -> dict[str, object]:

Check failure on line 106 in loopx/extensions/lark/event_collector_runtime.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 19 to the 15 allowed.

See more on https://sonarcloud.io/project/issues?id=huangruiteng_loopx&issues=AaCfY6yFSQ3tuMegsiOj&open=AaCfY6yFSQ3tuMegsiOj&pullRequest=4375
"""Return a value-free diagnostic projection for a rejected callback."""

action_value = payload.get("action_value")
try:
action = (
json.loads(action_value) if isinstance(action_value, str) else action_value
)
except json.JSONDecodeError:
action = None
card_content = payload.get("card_content")
card_shape = "missing"
if isinstance(card_content, str):
if not card_content:
card_shape = "empty"
else:
try:
parsed_card = json.loads(card_content)
except json.JSONDecodeError:
card_shape = "text"
else:
card_shape = (
"json_object" if isinstance(parsed_card, Mapping) else "json_other"
)
elif isinstance(card_content, Mapping):
card_shape = "object"
elif card_content is not None:
card_shape = type(card_content).__name__
timestamp = str(payload.get("timestamp") or "")
return {
"type_supported": payload.get("type") == "card.action.trigger",

Check failure on line 136 in loopx/extensions/lark/event_collector_runtime.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Define a constant instead of duplicating this literal "card.action.trigger" 3 times.

See more on https://sonarcloud.io/project/issues?id=huangruiteng_loopx&issues=AaCfY6yFSQ3tuMegsiOi&open=AaCfY6yFSQ3tuMegsiOi&pullRequest=4375
"action_is_button": payload.get("action_tag") == "button",
"action_is_object": isinstance(action, Mapping),
"action_field_count": len(action) if isinstance(action, Mapping) else 0,
"event_id_valid": bool(
re.fullmatch(r"[A-Za-z0-9._:-]{1,240}", str(payload.get("event_id") or ""))
),
"timestamp_is_digits": timestamp.isdigit(),
"timestamp_digit_count": len(timestamp),
"operator_id_present": bool(payload.get("operator_id")),
"message_id_present": bool(payload.get("message_id")),
"chat_id_present": bool(payload.get("chat_id")),
"host_supported": payload.get("host") == "im_message",
"token_present": bool(payload.get("token")),
"card_content_shape": card_shape,
"shape_digest": hashlib.sha256(
"\0".join(sorted(str(key) for key in payload)).encode()
).hexdigest()[:16],
}


def _run_json(
runner: CommandRunner,
argv: Sequence[str],
Expand Down Expand Up @@ -392,6 +511,9 @@
listener_ready: bool | None = None,
callback_delivery_verified: bool | None = None,
failure_kind: str | None = None,
failure_code: str | None = None,
failure_stage: str | None = None,
failure_event_shape: Mapping[str, object] | None = None,
consumer_returncode: int | None = None,
recovered_result_count_delta: int = 0,
result_delivery_failure_count_delta: int = 0,
Expand Down Expand Up @@ -444,6 +566,13 @@
else prior.get("last_verified_callback_at")
),
"last_failure_kind": failure_kind or prior.get("last_failure_kind"),
"last_failure_code": failure_code or prior.get("last_failure_code"),
"last_failure_stage": failure_stage or prior.get("last_failure_stage"),
"last_failure_event_shape": (
dict(failure_event_shape)
if failure_event_shape is not None
else prior.get("last_failure_event_shape")
),
"consumer_returncode": consumer_returncode,
"updated_at": now,
"private_content_returned": False,
Expand Down Expand Up @@ -722,6 +851,9 @@
listener_active=True,
listener_ready=True,
failure_kind=type(exc).__name__,
failure_code=_operation_callback_failure_code(exc),
failure_stage=_operation_callback_failure_stage(exc),
failure_event_shape=_callback_event_shape(payload),
)
continue
callback_stats["verified"] += 1
Expand Down Expand Up @@ -973,9 +1105,7 @@
result.update(
{
"operation_callback_listener_started": True,
"operation_callback_listener_ready": bool(
callback_stats.get("ready")
),
"operation_callback_listener_ready": bool(callback_stats.get("ready")),
"operation_callback_received_count": callback_stats["received"],
"operation_callback_verified_count": callback_stats["verified"],
"operation_callback_failure_count": callback_stats["failed"],
Expand Down
Loading