diff --git a/migrations/versions/20260929090000_a2a_message_idempotency.py b/migrations/versions/20260929090000_a2a_message_idempotency.py new file mode 100644 index 0000000..540c2af --- /dev/null +++ b/migrations/versions/20260929090000_a2a_message_idempotency.py @@ -0,0 +1,45 @@ +"""Add durable idempotency records for CRM A2A message/send requests.""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +revision: str = "20260929090000" +down_revision: Union[str, None] = "26a14ac7025d" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "evo_ai_a2a_idempotency_records", + sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("agent_id", postgresql.UUID(as_uuid=True), nullable=False), + sa.Column("context_id", sa.String(length=255), nullable=False), + sa.Column("idempotency_key", sa.String(length=128), nullable=False), + sa.Column("request_hash", sa.String(length=64), nullable=False), + sa.Column("state", sa.String(length=16), nullable=False), + sa.Column("response", sa.JSON(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.CheckConstraint("state IN ('processing', 'completed')", name="check_a2a_idempotency_state"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "agent_id", + "context_id", + "idempotency_key", + name="uq_a2a_idempotency_agent_context_key", + ), + ) + op.create_index( + "ix_a2a_idempotency_created_at", + "evo_ai_a2a_idempotency_records", + ["created_at"], + ) + + +def downgrade() -> None: + op.drop_index("ix_a2a_idempotency_created_at", table_name="evo_ai_a2a_idempotency_records") + op.drop_table("evo_ai_a2a_idempotency_records") diff --git a/src/api/a2a_routes.py b/src/api/a2a_routes.py index 08521b3..feeb646 100644 --- a/src/api/a2a_routes.py +++ b/src/api/a2a_routes.py @@ -81,6 +81,12 @@ from src.services.tools_service import tools_service from src.services.mcp_server_service import get_mcp_server from src.middleware.permissions import RequirePermission +from src.services.a2a_idempotency import ( + claim_or_replay, + complete as complete_idempotency, + release as release_idempotency, + request_fingerprint, +) logger = setup_logger(__name__) @@ -865,12 +871,9 @@ async def handle_message_send( # Extract message from params message = params.get("message") if not message: - return error_response( - request=request, - code=map_status_to_error_code(status.HTTP_400_BAD_REQUEST), - message="Invalid params", + return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, - details={ + content={ "jsonrpc": "2.0", "id": request_id, "error": { @@ -878,7 +881,7 @@ async def handle_message_send( "message": "Invalid params", "data": {"missing": "message"}, }, - } + }, ) # Extract configuration from params (A2A spec: configuration is optional) @@ -991,13 +994,131 @@ async def handle_message_send( if not text and files: text = "Analyze the provided files" - logger.info(f"📝 Extracted text: {text}") + logger.info(f"📝 Extracted text length: {len(text)}") logger.info(f"📎 Extracted files: {len(files)}") # Generate IDs task_id = str(uuid.uuid4()) context_id = params.get("contextId", str(uuid.uuid4())) + if not isinstance(context_id, str) or len(context_id) > 255: + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_400_BAD_REQUEST), + message="Invalid contextId", + status_code=status.HTTP_400_BAD_REQUEST, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32602, + "message": "contextId must be a string of at most 255 characters", + }}, + ) + message_id = message.get("messageId") + idempotency_key = params.get("idempotencyKey") or message_id + # CRM messages have a stable persisted UUID. Generic A2A callers that omit + # one retain legacy behavior rather than being deduplicated on random IDs. + cacheable = bool(idempotency_key) and not push_notification_config + if idempotency_key and (not isinstance(idempotency_key, str) or len(idempotency_key) > 128): + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_400_BAD_REQUEST), + message="Invalid idempotency key", + status_code=status.HTTP_400_BAD_REQUEST, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32602, + "message": "idempotency key must be a string of at most 128 characters", + }}, + ) + fingerprint = None + owns_idempotency_record = False + + if cacheable: + # Tool execution can depend on metadata and caller identity, not only the + # visible message/file parts. Include every JSON-RPC param in the hash so + # a reused key with altered tool context is rejected rather than replayed. + fingerprint = request_fingerprint(params) + try: + claim_state, cached_response = claim_or_replay( + db, + agent_id=agent_id, + context_id=str(context_id), + idempotency_key=str(idempotency_key), + fingerprint=fingerprint, + ) + except Exception as error: + db.rollback() + logger.error( + "A2A idempotency storage unavailable; refusing unprotected execution " + f"(agent={agent_id}, context={context_id}, error={type(error).__name__})" + ) + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_503_SERVICE_UNAVAILABLE), + message="Request could not be safely deduplicated; retry shortly", + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32000, + "message": "Idempotency service unavailable", + }}, + ) + if claim_state == "replay": + logger.info( + f"A2A duplicate replayed agent={agent_id} context={context_id} " + f"message_id={idempotency_key}" + ) + return JSONResponse( + content={"jsonrpc": "2.0", "id": request_id, "result": cached_response} + ) + if claim_state == "conflict": + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_409_CONFLICT), + message="Idempotency key was already used for a different message", + status_code=status.HTTP_409_CONFLICT, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32602, + "message": "Idempotency key payload mismatch", + }}, + ) + if claim_state == "in_progress": + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_409_CONFLICT), + message="The same message is currently being processed; retry shortly", + status_code=status.HTTP_409_CONFLICT, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32000, + "message": "Request already in progress", + }}, + ) + if claim_state == "expired": + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_409_CONFLICT), + message="Previous processing stopped unexpectedly; operator review is required", + status_code=status.HTTP_409_CONFLICT, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32000, + "message": "Processing lease expired", + }}, + ) + if claim_state != "owner": + logger.error( + f"Unexpected A2A idempotency state {claim_state!r}; refusing execution " + f"(agent={agent_id}, context={context_id})" + ) + return error_response( + request=request, + code=map_status_to_error_code(status.HTTP_503_SERVICE_UNAVAILABLE), + message="Request could not be safely deduplicated; retry shortly", + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + details={"jsonrpc": "2.0", "id": request_id, "error": { + "code": -32000, + "message": "Unknown idempotency claim state", + }}, + ) + owns_idempotency_record = True + + agent_execution_started = False try: # Extract conversation history for context logger.info( @@ -1020,18 +1141,18 @@ async def handle_message_send( # Log detailed combined history for debugging for i, msg in enumerate(combined_history): - logger.info(f" History[{i}]: {msg['role']} - {msg['content'][:50]}...") + logger.debug(f" History[{i}]: {msg['role']} - {len(msg['content'])} chars") # Execute agent with files - the ADK runner will handle session history automatically logger.info( - f"🤖 Executing agent {agent_id} with message: {text} and {len(files)} files" + f"🤖 Executing agent {agent_id} with message length {len(text)} and {len(files)} files" ) logger.info( f"📚 ADK will provide session context automatically ({len(combined_history)} previous messages available)" ) metadata = extract_metadata_from_request(params) - logger.info(f"📋 Extracted metadata: {metadata}") + logger.info(f"📋 Extracted metadata fields: {len(metadata)}") # Extract userId from params (contact_id sent from Rails) # contextId is the conversation UUID, but userId should be the contact UUID @@ -1049,6 +1170,7 @@ async def handle_message_send( # CRM-236: bounded by whoever is waiting. When the caller hangs up the # run is cancelled instead of burning more of the provider's quota. + agent_execution_started = True result = await run_unless_client_disconnects( request, run_agent( @@ -1068,7 +1190,7 @@ async def handle_message_send( ) final_response = result.get("final_response", "No response") - logger.info(f"✅ Agent response: {final_response}") + logger.info(f"✅ Agent response generated ({len(final_response)} chars)") # Log what we're about to send to create_task_response logger.info( @@ -1095,6 +1217,16 @@ async def handle_message_send( current_user_message, ) + if owns_idempotency_record: + complete_idempotency( + db, + agent_id=agent_id, + context_id=str(context_id), + idempotency_key=str(idempotency_key), + fingerprint=fingerprint, + response=task_response, + ) + logger.info( f"📦 Task response created with {len(task_response.get('artifacts', []))} artifacts" ) @@ -1113,6 +1245,21 @@ async def handle_message_send( ) except ClientGoneAway as e: + if owns_idempotency_record and not agent_execution_started: + try: + release_idempotency( + db, + agent_id=agent_id, + context_id=str(context_id), + idempotency_key=str(idempotency_key), + fingerprint=fingerprint, + ) + except Exception as release_error: + db.rollback() + logger.error( + "Could not release cancelled A2A idempotency record " + f"(agent={agent_id}, error={type(release_error).__name__})" + ) # Nobody is waiting for this response — it is logged for the operator # and returned only because the handler must return something. logger.warning(f"⛔ {e}") @@ -1135,17 +1282,29 @@ async def handle_message_send( ) except Exception as e: + if owns_idempotency_record and not agent_execution_started: + try: + release_idempotency( + db, + agent_id=agent_id, + context_id=str(context_id), + idempotency_key=str(idempotency_key), + fingerprint=fingerprint, + ) + except Exception as release_error: + db.rollback() + logger.error( + "Could not release failed A2A idempotency record " + f"(agent={agent_id}, error={type(release_error).__name__})" + ) # CRM-236: a quota exhaustion and a bug of ours both answered "500 Agent # execution failed". Report the provider's condition when we recognise it. provider_failure = classify_provider_error(e) if provider_failure is not None: log_provider_failure(provider_failure, agent_id) - return error_response( - request=request, - code=map_status_to_error_code(provider_failure.http_status), - message=provider_failure.message, + return JSONResponse( status_code=provider_failure.http_status, - details={ + content={ "jsonrpc": "2.0", "id": request_id, "error": { @@ -1163,22 +1322,8 @@ async def handle_message_send( # and the one most likely to carry a credential (`?key=AIza…`). safe = redact_secrets(str(e)) logger.error(f"❌ Agent execution error: {safe}") - return error_response( - request=request, - code=map_status_to_error_code(status.HTTP_500_INTERNAL_SERVER_ERROR), - message="Agent execution failed", - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - details={ - "jsonrpc": "2.0", - "id": request_id, - "error": { - "code": -32603, - "message": "Agent execution failed", - "data": {"error": safe}, - }, - } - ) return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={ "jsonrpc": "2.0", "id": request_id, @@ -1187,7 +1332,7 @@ async def handle_message_send( "message": "Agent execution failed", "data": {"error": safe}, }, - } + }, ) diff --git a/src/models/models.py b/src/models/models.py index 6dfd8bb..38b1cc7 100644 --- a/src/models/models.py +++ b/src/models/models.py @@ -39,6 +39,8 @@ Text, CheckConstraint, Boolean, + Index, + UniqueConstraint, ) from sqlalchemy.dialects.postgresql import UUID from sqlalchemy.sql import func @@ -295,6 +297,34 @@ class SessionMetadata(Base): updated_at = Column(DateTime(timezone=True), onupdate=func.now()) +class A2AIdempotencyRecord(Base): + """Durable response cache keyed by one CRM inbound message and agent.""" + + __tablename__ = "evo_ai_a2a_idempotency_records" + + id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + agent_id = Column(UUID(as_uuid=True), nullable=False) + context_id = Column(String(255), nullable=False) + idempotency_key = Column(String(128), nullable=False) + request_hash = Column(String(64), nullable=False) + state = Column(String(16), nullable=False, default="processing") + response = Column(JSON, nullable=True) + created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False) + + __table_args__ = ( + UniqueConstraint( + "agent_id", + "context_id", + "idempotency_key", + name="uq_a2a_idempotency_agent_context_key", + ), + CheckConstraint("state IN ('processing', 'completed')", name="check_a2a_idempotency_state"), + Index("ix_a2a_idempotency_created_at", "created_at"), + {"info": {"skip_autogenerate": True}}, + ) + + class CustomMCPServer(Base): __tablename__ = "evo_core_custom_mcp_servers" __table_args__ = {"info": {"skip_autogenerate": True}} @@ -374,4 +404,3 @@ class ExecutionMetrics(Base): created_at = Column(DateTime(timezone=True), server_default=func.now()) agent = relationship("Agent", backref="execution_metrics") - diff --git a/src/services/a2a_idempotency.py b/src/services/a2a_idempotency.py new file mode 100644 index 0000000..9636333 --- /dev/null +++ b/src/services/a2a_idempotency.py @@ -0,0 +1,160 @@ +"""Persistent idempotency for A2A requests produced by the CRM message queue.""" + +import hashlib +import json +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional, Tuple + +from sqlalchemy import delete, select +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.orm import Session + +from src.models.models import A2AIdempotencyRecord + +RETENTION = timedelta(hours=24) +PROCESSING_LEASE = timedelta(minutes=5) +MAX_KEY_LENGTH = 128 + + +def request_fingerprint(request_params: Dict[str, Any]) -> str: + """Hash canonical execution inputs; never persist raw payload or PII.""" + encoded = json.dumps( + request_params, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + default=str, + ) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def claim_or_replay( + db: Session, + *, + agent_id: uuid.UUID, + context_id: str, + idempotency_key: str, + fingerprint: str, +) -> Tuple[str, Optional[Dict[str, Any]]]: + """Return ownership/replay state and an optional cached A2A task response.""" + now = datetime.now(timezone.utc) + # Opportunistic bounded cleanup: only a small batch per accepted request. + # Sample cleanup so steady-state request cost remains independent of the + # number of retained records. Expired rows are also replaced below. + if uuid.uuid4().int % 100 == 0: + expired_ids = select(A2AIdempotencyRecord.id).where( + A2AIdempotencyRecord.created_at < now - RETENTION + ).limit(1000) + db.execute( + delete(A2AIdempotencyRecord) + .where(A2AIdempotencyRecord.id.in_(expired_ids)) + .execution_options(synchronize_session=False) + ) + statement = ( + insert(A2AIdempotencyRecord) + .values( + id=uuid.uuid4(), + agent_id=agent_id, + context_id=context_id, + idempotency_key=idempotency_key, + request_hash=fingerprint, + state="processing", + created_at=now, + updated_at=now, + ) + .on_conflict_do_nothing( + constraint="uq_a2a_idempotency_agent_context_key" + ) + .returning(A2AIdempotencyRecord.id) + ) + claimed_id = db.execute(statement).scalar_one_or_none() + db.commit() + if claimed_id is not None: + return "owner", None + + record = db.execute( + select(A2AIdempotencyRecord).where( + A2AIdempotencyRecord.agent_id == agent_id, + A2AIdempotencyRecord.context_id == context_id, + A2AIdempotencyRecord.idempotency_key == idempotency_key, + ) + ).scalar_one_or_none() + if record is None: + return "in_progress", None + if record.request_hash != fingerprint: + return "conflict", None + if record.state == "completed" and record.response is not None: + return "replay", record.response + if record.created_at < now - RETENTION: + db.execute( + delete(A2AIdempotencyRecord).where(A2AIdempotencyRecord.id == record.id) + ) + db.commit() + return claim_or_replay( + db, + agent_id=agent_id, + context_id=context_id, + idempotency_key=idempotency_key, + fingerprint=fingerprint, + ) + + if record.updated_at < now - PROCESSING_LEASE: + # The prior owner may have completed an external tool immediately + # before crashing. Never automatically repeat potentially side-effecting + # agent execution after lease expiry; require explicit operator review. + return "expired", None + + db.rollback() # Close the read transaction before returning a retryable conflict. + + return "in_progress", None + + +def complete( + db: Session, + *, + agent_id: uuid.UUID, + context_id: str, + idempotency_key: str, + fingerprint: str, + response: Dict[str, Any], +) -> None: + updated = db.query(A2AIdempotencyRecord).filter( + A2AIdempotencyRecord.agent_id == agent_id, + A2AIdempotencyRecord.context_id == context_id, + A2AIdempotencyRecord.idempotency_key == idempotency_key, + A2AIdempotencyRecord.request_hash == fingerprint, + A2AIdempotencyRecord.state == "processing", + ).update( + { + A2AIdempotencyRecord.state: "completed", + A2AIdempotencyRecord.response: response, + A2AIdempotencyRecord.updated_at: datetime.now(timezone.utc), + }, + synchronize_session=False, + ) + if updated != 1: + db.rollback() + raise RuntimeError("A2A idempotency claim was lost before completion") + db.commit() + + +def release( + db: Session, + *, + agent_id: uuid.UUID, + context_id: str, + idempotency_key: str, + fingerprint: str, +) -> None: + """Remove an uncompleted claim so transient agent errors may be retried.""" + db.execute( + delete(A2AIdempotencyRecord).where( + A2AIdempotencyRecord.agent_id == agent_id, + A2AIdempotencyRecord.context_id == context_id, + A2AIdempotencyRecord.idempotency_key == idempotency_key, + A2AIdempotencyRecord.request_hash == fingerprint, + A2AIdempotencyRecord.state == "processing", + ) + ) + db.commit() diff --git a/tests/integration/test_response_pipeline.py b/tests/integration/test_response_pipeline.py index bf293b0..64b488c 100644 --- a/tests/integration/test_response_pipeline.py +++ b/tests/integration/test_response_pipeline.py @@ -159,6 +159,8 @@ def test_exception_branch_returns_jsonrpc_envelope(self) -> None: a2a_routes, "extract_conversation_history", new=AsyncMock(return_value=[]) ), patch.object( a2a_routes, "extract_history_from_params", return_value=[] + ), patch.object( + a2a_routes, "claim_or_replay", return_value=("owner", None) ): response = asyncio.run( a2a_routes.handle_message_send( diff --git a/tests/unit/test_a2a_idempotency.py b/tests/unit/test_a2a_idempotency.py new file mode 100644 index 0000000..7620a47 --- /dev/null +++ b/tests/unit/test_a2a_idempotency.py @@ -0,0 +1,180 @@ +import uuid +from datetime import datetime, timedelta, timezone +from importlib.util import module_from_spec, spec_from_file_location +from pathlib import Path +from unittest.mock import MagicMock + +from src.models.models import A2AIdempotencyRecord + +# Loading this leaf module directly avoids src.services.__init__ eagerly +# importing the full ADK runtime for a database-service unit test. +_module_spec = spec_from_file_location( + "a2a_idempotency_under_test", + Path(__file__).parents[2] / "src/services/a2a_idempotency.py", +) +a2a_idempotency = module_from_spec(_module_spec) +_module_spec.loader.exec_module(a2a_idempotency) + + +def _record(*, request_hash="hash", state="processing", created_at=None, updated_at=None, response=None): + return A2AIdempotencyRecord( + id=uuid.uuid4(), + agent_id=uuid.uuid4(), + context_id="conversation-1", + idempotency_key="message-1", + request_hash=request_hash, + state=state, + created_at=created_at or datetime.now(timezone.utc), + updated_at=updated_at or datetime.now(timezone.utc), + response=response, + ) + + +def test_claim_creates_a_persistent_record_without_storing_message_content(): + db = MagicMock() + db.execute.side_effect = [MagicMock(scalar_one_or_none=MagicMock(return_value=uuid.uuid4()))] + message = {"messageId": "message-1", "parts": [{"type": "text", "text": "private text"}]} + + result = a2a_idempotency.claim_or_replay( + db, + agent_id=uuid.uuid4(), + context_id="conversation-1", + idempotency_key="message-1", + fingerprint=a2a_idempotency.request_fingerprint(message), + ) + + assert result == ("owner", None) + insert_statement = db.execute.call_args.args[0] + assert "private text" not in repr(insert_statement.compile().params) + db.commit.assert_called_once() + + +def test_request_fingerprint_covers_tool_context_as_well_as_message_parts(): + common = {"message": {"parts": [{"type": "text", "text": "hello"}]}} + with_context_a = {**common, "metadata": {"conversation_id": "conversation-a"}} + with_context_b = {**common, "metadata": {"conversation_id": "conversation-b"}} + + assert a2a_idempotency.request_fingerprint(with_context_a) != a2a_idempotency.request_fingerprint( + with_context_b + ) + + +def test_claim_replays_completed_response(): + response = {"id": "task-1", "status": {"state": "completed"}} + record = _record(state="completed", response=response) + db = MagicMock() + db.execute.return_value.scalar_one_or_none.side_effect = [None, record] + + claim = a2a_idempotency.claim_or_replay( + db, + agent_id=record.agent_id, + context_id=record.context_id, + idempotency_key=record.idempotency_key, + fingerprint=record.request_hash, + ) + + assert claim == ("replay", response) + + +def test_claim_rejects_reuse_with_different_payload(): + record = _record(request_hash="original-hash") + db = MagicMock() + db.execute.return_value.scalar_one_or_none.side_effect = [None, record] + + claim = a2a_idempotency.claim_or_replay( + db, + agent_id=record.agent_id, + context_id=record.context_id, + idempotency_key=record.idempotency_key, + fingerprint="different-hash", + ) + + assert claim == ("conflict", None) + + +def test_fresh_processing_claim_is_not_taken_over(): + now = datetime.now(timezone.utc) + record = _record(created_at=now, updated_at=now) + db = MagicMock() + db.execute.return_value.scalar_one_or_none.side_effect = [None, record] + + claim = a2a_idempotency.claim_or_replay( + db, + agent_id=record.agent_id, + context_id=record.context_id, + idempotency_key=record.idempotency_key, + fingerprint=record.request_hash, + ) + + assert claim == ("in_progress", None) + db.query.assert_not_called() + + +def test_stale_processing_claim_is_not_retried_automatically(): + now = datetime.now(timezone.utc) + record = _record(updated_at=now - timedelta(minutes=6)) + db = MagicMock() + db.execute.return_value.scalar_one_or_none.side_effect = [None, record] + claim = a2a_idempotency.claim_or_replay( + db, + agent_id=record.agent_id, + context_id=record.context_id, + idempotency_key=record.idempotency_key, + fingerprint=record.request_hash, + ) + + assert claim == ("expired", None) + db.query.assert_not_called() + + +def test_completed_response_is_saved_only_for_matching_processing_claim(): + db = MagicMock() + db.query.return_value.filter.return_value.update.return_value = 1 + a2a_idempotency.complete( + db, + agent_id=uuid.uuid4(), + context_id="conversation-1", + idempotency_key="message-1", + fingerprint="hash", + response={"id": "task-1"}, + ) + + update_values = db.query.return_value.filter.return_value.update.call_args.args[0] + assert update_values[A2AIdempotencyRecord.state] == "completed" + assert update_values[A2AIdempotencyRecord.response] == {"id": "task-1"} + db.commit.assert_called_once() + + +def test_complete_refuses_to_report_success_if_claim_disappeared(): + db = MagicMock() + db.query.return_value.filter.return_value.update.return_value = 0 + + try: + a2a_idempotency.complete( + db, + agent_id=uuid.uuid4(), + context_id="conversation-1", + idempotency_key="message-1", + fingerprint="hash", + response={"id": "task-1"}, + ) + except RuntimeError as error: + assert "claim was lost" in str(error) + else: + raise AssertionError("completion without the processing claim must fail") + db.rollback.assert_called_once() + db.commit.assert_not_called() + + +def test_failure_releases_only_matching_processing_claim(): + db = MagicMock() + a2a_idempotency.release( + db, + agent_id=uuid.uuid4(), + context_id="conversation-1", + idempotency_key="message-1", + fingerprint="hash", + ) + + db.execute.assert_called_once() + db.commit.assert_called_once()