From cad76bf8354d1ce9081ea8fe77a4bf41f349c41d Mon Sep 17 00:00:00 2001 From: ogt Date: Thu, 9 Jul 2026 23:14:52 +0800 Subject: [PATCH] fix(agents): fallback learning loop from consumer receipts --- ..._controlled_writeback_consumer_readback.py | 148 +++++++++++++++++- ...trolled_writeback_consumer_readback_api.py | 51 ++++++ 2 files changed, 198 insertions(+), 1 deletion(-) diff --git a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py index fa60a3509..ded81f716 100644 --- a/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py +++ b/apps/api/src/services/ai_agent_log_controlled_writeback_consumer_readback.py @@ -394,6 +394,12 @@ def _build_consumer_readback( ) -> dict[str, Any]: consumer_receipts = _consumer_receipts_by_dispatch_id(consumer_rows) bindings = [_consumer_binding(row, consumer_receipts) for row in rows] + telegram_alert_learning_context = ( + _telegram_alert_learning_context_with_consumer_fallback( + telegram_alert_learning_context=telegram_alert_learning_context, + bindings=bindings, + ) + ) active_blockers = _active_blockers(bindings) target_rollups = _target_rollups(bindings) runtime_target_write_performed = ( @@ -1038,6 +1044,146 @@ def _telegram_alert_learning_context_unavailable( } +def _telegram_alert_learning_context_with_consumer_fallback( + *, + telegram_alert_learning_context: dict[str, Any], + bindings: list[dict[str, Any]], +) -> dict[str, Any]: + if ( + telegram_alert_learning_context.get("status") + == "ai_loop_agent_context_receipt_ready" + ): + return telegram_alert_learning_context + + ready_bindings = [ + binding + for binding in bindings + if binding.get("status") == "ready_for_consumer_context" + and binding.get("target_write_performed") is True + and binding.get("raw_payload_included") is False + and binding.get("target") in _TARGETS + ] + ready_targets = {str(binding.get("target") or "") for binding in ready_bindings} + if set(_TARGETS) - ready_targets: + return telegram_alert_learning_context + + project_id = str( + telegram_alert_learning_context.get("project_id") or DEFAULT_PROJECT_ID + ) + context_receipts = [ + _telegram_alert_context_receipt_from_consumer_binding( + binding=binding, + project_id=project_id, + ) + for binding in ready_bindings + ] + ai_agent_context_receipt_count = sum( + 1 for receipt in context_receipts if receipt["target"] == "ai_agent" + ) + return { + **telegram_alert_learning_context, + "status": "ai_loop_agent_context_receipt_ready", + "source_registry_status": ( + "fallback_log_controlled_writeback_consumer_context" + ), + "fallback_source_endpoint": ( + "/api/v1/agents/agent-log-controlled-writeback-consumer-readback" + ), + "ready_target_count": len(ready_targets), + "target_count": len(_TARGETS), + "context_receipt_count": len(context_receipts), + "ai_agent_context_receipt_count": ai_agent_context_receipt_count, + "next_action": ( + "verify_log_controlled_writeback_context_receipts_with_ai_loop_post_apply_verifier" + ), + "context_receipts": context_receipts, + "active_blockers": [], + "operation_boundaries": { + **_dict(telegram_alert_learning_context.get("operation_boundaries")), + "metadata_read_performed": True, + "db_or_log_receipt_read_performed": True, + "runtime_target_write_performed": True, + "telegram_send_performed": False, + "raw_payload_included": False, + "secret_value_collection_allowed": False, + "github_api_used": False, + }, + } + + +def _telegram_alert_context_receipt_from_consumer_binding( + *, + binding: Mapping[str, Any], + project_id: str, +) -> dict[str, Any]: + target = str(binding.get("target") or "") + dispatch_receipt_id = str(binding.get("dispatch_receipt_id") or "") + consumer_receipt = _dict(binding.get("target_write_receipt")) + source_receipt_id = str( + consumer_receipt.get("consumer_receipt_id") or dispatch_receipt_id + ) + verifier_refs = _strings(binding.get("post_apply_verifier_refs")) + return { + "receipt_id": ( + "telegram_alert_learning_context_fallback::" + f"{source_receipt_id or target}" + ), + "source_receipt_id": source_receipt_id, + "source_run_id": "", + "source_message_id": "", + "work_item_id": "CIR-P0-TG-001", + "target": target, + "consumer_surface": str(binding.get("consumer_surface") or ""), + "source_ref": dispatch_receipt_id, + "ai_agent_context_ref": ( + f"ai-agent://{project_id}/log-controlled-writeback/" + f"{source_receipt_id or target}" + ), + "event_type": "log_controlled_writeback_consumed", + "lane": "metadata_feedback_controlled_writeback", + "status": "ready_for_ai_loop_context", + "metadata_only": True, + "target_write_performed": True, + "raw_payload_included": False, + "target_selector": { + "project_id": project_id, + "dispatch_receipt_id": dispatch_receipt_id, + "consumer_receipt_id": source_receipt_id, + "target": target, + "source_ref": dispatch_receipt_id, + }, + "source_of_truth_diff": { + "current_state": "telegram_alert_registry_unavailable_or_not_ready", + "desired_state": "ai_loop_agent_context_receipt_available", + "delta_kind": f"log_controlled_writeback_{target}_context_fallback", + "raw_payload_included": False, + }, + "check_mode": { + "enabled": True, + "checks": [ + "consumer_binding_ready", + "consumer_apply_receipt_present", + "metadata_only_raw_payload_absent", + "post_apply_verifier_refs_present", + ], + }, + "rollback": { + "required": True, + "rollback_ref": ( + "rollback://telegram-alert-learning-context-fallback/" + f"{source_receipt_id or target}" + ), + "strategy": ( + "ignore_fallback_context_receipt_and_wait_for_next_alert_card_readback" + ), + }, + "post_apply_verifier": { + "required": True, + "verifier_refs": verifier_refs, + }, + } + + def _result_rows(result: Any) -> list[dict[str, Any]]: mappings = getattr(result, "mappings", None) if callable(mappings): @@ -1074,7 +1220,7 @@ def _dict(value: Any) -> dict[str, Any]: def _json_dict(value: Any) -> dict[str, Any]: if isinstance(value, Mapping): return dict(value) - if isinstance(value, (bytes, bytearray)): + if isinstance(value, bytes | bytearray): value = value.decode("utf-8", errors="replace") if isinstance(value, str) and value: try: diff --git a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py index 7e2e98314..5b40b1d75 100644 --- a/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py +++ b/apps/api/tests/test_ai_agent_log_controlled_writeback_consumer_readback_api.py @@ -304,6 +304,57 @@ async def test_log_controlled_writeback_consumer_loader_uses_direct_fallback( assert payload["operation_boundaries"]["metadata_ledger_read_performed"] is True +@pytest.mark.asyncio +async def test_log_controlled_writeback_consumer_maps_ready_bindings_when_alert_registry_times_out( + monkeypatch, +): + fake_db = _FakeDb(_ledger_rows(), _consumer_receipt_rows()) + monkeypatch.setattr( + consumer_module, + "get_db_context", + lambda project_id: _FakeContext(fake_db), + ) + + async def alert_registry_timeout(**_kwargs): + raise TimeoutError("alert registry unavailable") + + monkeypatch.setattr( + consumer_module, + "list_ai_alert_card_delivery_readback", + alert_registry_timeout, + ) + + payload = await load_latest_ai_agent_log_controlled_writeback_consumer_readback() + + assert payload["status"] == "controlled_writeback_consumer_readback_ready" + assert payload["rollups"]["telegram_alert_learning_context_readback_ready"] is True + assert payload["rollups"]["telegram_alert_learning_context_receipt_count"] == 6 + assert ( + payload["rollups"]["telegram_alert_learning_ai_agent_context_receipt_count"] + == 1 + ) + context = payload["telegram_alert_learning_context"] + assert context["status"] == "ai_loop_agent_context_receipt_ready" + assert context["source_registry_status"] == ( + "fallback_log_controlled_writeback_consumer_context" + ) + assert context["active_blockers"] == [] + assert {receipt["target"] for receipt in context["context_receipts"]} == { + "km", + "rag", + "playbook", + "mcp", + "verifier", + "ai_agent", + } + assert all( + receipt["status"] == "ready_for_ai_loop_context" + for receipt in context["context_receipts"] + ) + assert context["operation_boundaries"]["runtime_target_write_performed"] is True + assert context["operation_boundaries"]["raw_payload_included"] is False + + def test_log_controlled_writeback_consumer_direct_rows_parse_json_text(): dispatch_row = _ledger_rows()[0] consumer_row = _consumer_receipt_rows()[0]