fix: dedupe agent invoke submissions by Idempotency-Key

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:
@@ -492,6 +493,39 @@ async def _safe_close_agent(agent: Any) -> None:
         logger.exception("Agent cleanup failed")
 
 
+_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.
+
+    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 SqliteIdempotencyStore
+
+            base = os.environ.get("PRAISONAI_HOME")
+            home = Path(base) if base else Path.home() / ".praisonai"
+            _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
+
+
 def _supports_async_start(agent: Any) -> bool:
     """Return True if agent.astart is a coroutine function."""
     astart = getattr(agent, "astart", None)
@@ -514,6 +548,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 +564,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 +599,30 @@ async def invoke_agent(
                 detail=f"Agent '{agent_id}' not found"
             )
 
+        import asyncio
+        store = 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()
+                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 +651,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),
@@ -609,11 +672,26 @@ 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 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:
+                    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(

src/praisonai/tests/test_n8n_agent_invoke.py

@@ -746,6 +746,151 @@ 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_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()
+        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_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."""
     try: