From 0f219b99af0525373cca6f0849d57028cda16496 Mon Sep 17 00:00:00 2001 From: MervinPraison <454862+MervinPraison@users.noreply.github.com> Date: Sat, 24 Oct 2026 10:00:14 +0000 Subject: [PATCH 1/2] fix: dedupe agent invoke submissions by Idempotency-Key (fixes #5977) Co-Authored-By: Claude Opus 5.5 --- src/praisonai/praisonai/api/agent_invoke.py | 66 ++++++++++++++- src/praisonai/tests/test_n8n_agent_invoke.py | 89 ++++++++++++++++++++ 2 files changed, 153 insertions(+), 2 deletions(-) diff --git a/src/praisonai/praisonai/api/agent_invoke.py b/src/praisonai/praisonai/api/agent_invoke.py index 01b1e8e449..2f6cf97f4e 100644 --- a/src/praisonai/praisonai/api/agent_invoke.py +++ b/src/praisonai/praisonai/api/agent_invoke.py @@ -492,6 +492,32 @@ async def _safe_close_agent(agent: Any) -> None: logger.exception("Agent cleanup failed") +_invoke_idempotency_store: Any = None +_invoke_idempotency_lock = threading.Lock() + + +def _get_invoke_idempotency_store() -> Any: + """Lazily build the durable store that dedups keyed invoke submissions. + + Reuses the same restart-safe ``build_idempotency_store`` the gateway and + scheduler delivery paths use, under ``$PRAISONAI_HOME/state`` (falling back + to ``~/.praisonai``), so a client retry carrying the same + ``Idempotency-Key`` never starts a second agent run. + """ + global _invoke_idempotency_store + with _invoke_idempotency_lock: + if _invoke_idempotency_store is None: + from pathlib import Path + from praisonai_bot.bots._idempotency import build_idempotency_store + + base = os.environ.get("PRAISONAI_HOME") + home = Path(base) if base else Path.home() / ".praisonai" + _invoke_idempotency_store = build_idempotency_store( + "sqlite", path=home / "state" / "invoke_idempotency.db" + ) + return _invoke_idempotency_store + + def _supports_async_start(agent: Any) -> bool: """Return True if agent.astart is a coroutine function.""" astart = getattr(agent, "astart", None) @@ -514,6 +540,7 @@ async def invoke_agent( request: AgentInvokeRequest, _: None = Depends(verify_token), registry: AgentRegistry = Depends(_registry_dependency), + idempotency_key: Optional[str] = Header(None, alias="Idempotency-Key"), ) -> Union[AgentInvokeResponse, ErrorResponse]: """ Invoke a PraisonAI agent with a message. @@ -529,8 +556,11 @@ async def invoke_agent( Agent response with result and session information Raises: - HTTPException: If agent not found or execution fails - + HTTPException: If agent not found or execution fails, or 409 when + an ``Idempotency-Key`` was already submitted (in flight or + completed within the retention window) — the agent is not + re-run. A failed run frees its key so the client can retry. + Example n8n HTTP Request Node configuration: ```json { @@ -561,6 +591,28 @@ async def invoke_agent( detail=f"Agent '{agent_id}' not found" ) + import asyncio + store = None + idem_key = f"invoke:{agent_id}:{idempotency_key}" if idempotency_key else None + succeeded = False + if idem_key: + try: + store = _get_invoke_idempotency_store() + admitted = await asyncio.to_thread(store.reserve, idem_key) + except Exception as e: + await _safe_close_agent(agent) + logger.error(f"Idempotency store unavailable: {e}") + raise HTTPException( + status_code=503, + detail="Idempotency store unavailable; retry later", + ) + if not admitted: + await _safe_close_agent(agent) + raise HTTPException( + status_code=409, + detail="A run with this Idempotency-Key was already submitted", + ) + try: # Apply per-request agent_config overrides to the isolated clone, or # reject unsupported keys with a 400 so the caller isn't silently @@ -589,6 +641,7 @@ async def invoke_agent( raise AttributeError(f"Agent {agent_id} must provide start() or async astart()") logger.info(f"Agent {agent_id} invoked successfully") + succeeded = True return AgentInvokeResponse( result=str(result), @@ -614,6 +667,15 @@ async def invoke_agent( # threads, MCP subprocesses). No-op when the agent fell back to the # shared template. Never masks the response. await _safe_close_agent(agent) + if store is not None: + # Commit the key on success so retries are deduplicated; free it + # on failure so the client can retry. Never masks the response. + try: + await asyncio.to_thread( + store.record if succeeded else store.release, idem_key + ) + except Exception: + logger.exception("Idempotency store update failed") @router.get("/agents") async def list_agents( diff --git a/src/praisonai/tests/test_n8n_agent_invoke.py b/src/praisonai/tests/test_n8n_agent_invoke.py index 1aa7c960af..d9c0b2471d 100644 --- a/src/praisonai/tests/test_n8n_agent_invoke.py +++ b/src/praisonai/tests/test_n8n_agent_invoke.py @@ -746,6 +746,95 @@ def test_route_does_not_close_shared_template(self, client, monkeypatch): unregister_agent("route-shared") +@pytest.mark.skipif(not pytest.importorskip("fastapi", minversion="0.68.0"), reason="FastAPI not available") +class TestInvokeIdempotency: + """Idempotency-Key on the invoke route (Issue #5977).""" + + @pytest.fixture + def client(self, tmp_path, monkeypatch): + from fastapi import FastAPI + from fastapi.testclient import TestClient + import praisonai.api.agent_invoke as agent_invoke + + monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path)) + monkeypatch.setattr(agent_invoke, "_invoke_idempotency_store", None) + app = FastAPI() + app.include_router(agent_invoke.router) + return TestClient(app) + + def _register(self, agent_id, side_effect=None): + from praisonai.api.agent_invoke import register_agent + + agent = Mock() + if side_effect is not None: + agent.start.side_effect = side_effect + else: + agent.start.return_value = "done" + register_agent(agent_id, agent) + return agent + + def test_same_key_runs_agent_once(self, client): + from praisonai.api.agent_invoke import unregister_agent + + agent = self._register("idem-once") + try: + headers = {"Idempotency-Key": "k-1"} + first = client.post("/api/v1/agents/idem-once/invoke", + json={"message": "refund"}, headers=headers) + retry = client.post("/api/v1/agents/idem-once/invoke", + json={"message": "refund"}, headers=headers) + assert first.status_code == 200 + assert retry.status_code == 409 + assert agent.start.call_count == 1 + finally: + unregister_agent("idem-once") + + def test_no_key_runs_every_time(self, client): + from praisonai.api.agent_invoke import unregister_agent + + agent = self._register("idem-none") + try: + for _ in range(2): + resp = client.post("/api/v1/agents/idem-none/invoke", + json={"message": "hi"}) + assert resp.status_code == 200 + assert agent.start.call_count == 2 + finally: + unregister_agent("idem-none") + + def test_failed_run_frees_key(self, client): + from praisonai.api.agent_invoke import unregister_agent + + agent = self._register("idem-fail", side_effect=[RuntimeError("boom"), "ok"]) + try: + headers = {"Idempotency-Key": "k-2"} + first = client.post("/api/v1/agents/idem-fail/invoke", + json={"message": "hi"}, headers=headers) + retry = client.post("/api/v1/agents/idem-fail/invoke", + json={"message": "hi"}, headers=headers) + assert first.status_code == 500 + assert retry.status_code == 200 + assert agent.start.call_count == 2 + finally: + unregister_agent("idem-fail") + + def test_key_is_scoped_per_agent(self, client): + from praisonai.api.agent_invoke import unregister_agent + + a = self._register("idem-a") + b = self._register("idem-b") + try: + headers = {"Idempotency-Key": "shared"} + assert client.post("/api/v1/agents/idem-a/invoke", + json={"message": "hi"}, headers=headers).status_code == 200 + assert client.post("/api/v1/agents/idem-b/invoke", + json={"message": "hi"}, headers=headers).status_code == 200 + assert a.start.call_count == 1 and b.start.call_count == 1 + finally: + unregister_agent("idem-a") + unregister_agent("idem-b") + + def test_agent_invoke_smoke_test(): """Smoke test to verify agent invoke module can be imported and used.""" try: From 4280b9baebdf43ac2e3b8ccf3954e583cf2bcb6d Mon Sep 17 00:00:00 2001 From: MervinPraison <454862+MervinPraison@users.noreply.github.com> Date: Sat, 24 Oct 2026 10:07:45 +0000 Subject: [PATCH 2/2] fix: harden invoke idempotency (durable-or-503, no lease expiry mid-run, keep key on cancel, unambiguous key encoding) Co-Authored-By: Claude Opus 5.5 --- src/praisonai/praisonai/api/agent_invoke.py | 34 ++++++++---- src/praisonai/tests/test_n8n_agent_invoke.py | 56 ++++++++++++++++++++ 2 files changed, 81 insertions(+), 9 deletions(-) diff --git a/src/praisonai/praisonai/api/agent_invoke.py b/src/praisonai/praisonai/api/agent_invoke.py index 2f6cf97f4e..22c4abc46a 100644 --- a/src/praisonai/praisonai/api/agent_invoke.py +++ b/src/praisonai/praisonai/api/agent_invoke.py @@ -7,6 +7,7 @@ from typing import Any, Dict, Optional, Union import inspect +import json import logging try: @@ -494,26 +495,33 @@ async def _safe_close_agent(agent: Any) -> None: _invoke_idempotency_store: Any = None _invoke_idempotency_lock = threading.Lock() +_INVOKE_IDEMPOTENCY_TTL_SECONDS = 86_400.0 # 24h retention window def _get_invoke_idempotency_store() -> Any: """Lazily build the durable store that dedups keyed invoke submissions. - Reuses the same restart-safe ``build_idempotency_store`` the gateway and - scheduler delivery paths use, under ``$PRAISONAI_HOME/state`` (falling back - to ``~/.praisonai``), so a client retry carrying the same - ``Idempotency-Key`` never starts a second agent run. + Uses the gateway's restart-safe ``SqliteIdempotencyStore`` directly under + ``$PRAISONAI_HOME/state`` (falling back to ``~/.praisonai``), so a client + retry carrying the same ``Idempotency-Key`` never starts a second agent run. + Built directly rather than via ``build_idempotency_store`` so an unusable + database raises (the route answers 503) instead of silently degrading to a + per-process in-memory store, and so it never touches the gateway's + durability-degraded health facts. The in-flight lease equals the TTL so a + long-running agent never loses its reservation to a retry. """ global _invoke_idempotency_store with _invoke_idempotency_lock: if _invoke_idempotency_store is None: from pathlib import Path - from praisonai_bot.bots._idempotency import build_idempotency_store + from praisonai_bot.bots._idempotency import SqliteIdempotencyStore base = os.environ.get("PRAISONAI_HOME") home = Path(base) if base else Path.home() / ".praisonai" - _invoke_idempotency_store = build_idempotency_store( - "sqlite", path=home / "state" / "invoke_idempotency.db" + _invoke_idempotency_store = SqliteIdempotencyStore( + home / "state" / "invoke_idempotency.db", + ttl_seconds=_INVOKE_IDEMPOTENCY_TTL_SECONDS, + inflight_lease_seconds=_INVOKE_IDEMPOTENCY_TTL_SECONDS, ) return _invoke_idempotency_store @@ -593,8 +601,10 @@ async def invoke_agent( import asyncio store = None - idem_key = f"invoke:{agent_id}:{idempotency_key}" if idempotency_key else None + # JSON-encode the pair so ``:`` in either part can't make two agents' keys collide. + idem_key = json.dumps(["invoke", agent_id, idempotency_key]) if idempotency_key else None succeeded = False + failed = False if idem_key: try: store = _get_invoke_idempotency_store() @@ -662,12 +672,18 @@ async def invoke_agent( status_code=500, detail=f"Agent execution failed: {str(e)}" ) + except Exception: + # Only a run that actually failed frees its key. A cancelled request + # (``CancelledError`` is not an ``Exception``) keeps the reservation, + # since a sync ``start`` keeps running in its worker thread. + failed = True + raise finally: # Release the per-request clone's resources (LLM pools, tool-executor # threads, MCP subprocesses). No-op when the agent fell back to the # shared template. Never masks the response. await _safe_close_agent(agent) - if store is not None: + if store is not None and (succeeded or failed): # Commit the key on success so retries are deduplicated; free it # on failure so the client can retry. Never masks the response. try: diff --git a/src/praisonai/tests/test_n8n_agent_invoke.py b/src/praisonai/tests/test_n8n_agent_invoke.py index d9c0b2471d..0c913b424a 100644 --- a/src/praisonai/tests/test_n8n_agent_invoke.py +++ b/src/praisonai/tests/test_n8n_agent_invoke.py @@ -756,6 +756,8 @@ def client(self, tmp_path, monkeypatch): from fastapi.testclient import TestClient import praisonai.api.agent_invoke as agent_invoke + monkeypatch.setenv("PRAISONAI_CALL_AUTH", "disabled") + monkeypatch.setenv("PRAISONAI_CALL_BIND_HOST", "127.0.0.1") monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path)) monkeypatch.setattr(agent_invoke, "_invoke_idempotency_store", None) app = FastAPI() @@ -834,6 +836,60 @@ def test_key_is_scoped_per_agent(self, client): unregister_agent("idem-a") unregister_agent("idem-b") + def test_separator_in_ids_does_not_collide(self, client): + from praisonai.api.agent_invoke import unregister_agent + + a = self._register("x:y") + b = self._register("x") + try: + assert client.post("/api/v1/agents/x:y/invoke", json={"message": "hi"}, + headers={"Idempotency-Key": "z"}).status_code == 200 + assert client.post("/api/v1/agents/x/invoke", json={"message": "hi"}, + headers={"Idempotency-Key": "y:z"}).status_code == 200 + assert a.start.call_count == 1 and b.start.call_count == 1 + finally: + unregister_agent("x:y") + unregister_agent("x") + + def test_unusable_store_rejects_instead_of_running(self, client, tmp_path, monkeypatch): + from praisonai.api.agent_invoke import unregister_agent + + blocker = tmp_path / "not-a-dir" + blocker.write_text("") + monkeypatch.setenv("PRAISONAI_HOME", str(blocker)) + agent = self._register("idem-503") + try: + resp = client.post("/api/v1/agents/idem-503/invoke", json={"message": "hi"}, + headers={"Idempotency-Key": "k-3"}) + assert resp.status_code == 503 + assert agent.start.call_count == 0 + finally: + unregister_agent("idem-503") + + def test_cancelled_run_keeps_reservation(self, client): + import asyncio + from praisonai.api.agent_invoke import register_agent, unregister_agent + + class _Agent: + calls = 0 + + async def astart(self, message): + _Agent.calls += 1 + raise asyncio.CancelledError() + + register_agent("idem-cancel", _Agent()) + try: + headers = {"Idempotency-Key": "k-4"} + with pytest.raises(BaseException): + client.post("/api/v1/agents/idem-cancel/invoke", + json={"message": "hi"}, headers=headers) + retry = client.post("/api/v1/agents/idem-cancel/invoke", + json={"message": "hi"}, headers=headers) + assert retry.status_code == 409 + assert _Agent.calls == 1 + finally: + unregister_agent("idem-cancel") + def test_agent_invoke_smoke_test(): """Smoke test to verify agent invoke module can be imported and used."""