Release v4.6.157

fix: remove dead deprecated _classify_error_and_should_retry_legacy

.github/scripts/release-gate-selftest.js

@@ -36,8 +36,10 @@ assert.strictEqual(dayStart.toISOString(), '2026-07-08T00:00:00.000Z');
       },
     },
   };
+  assert.strictEqual(rg.PATCH_RELEASE_INTERVAL_DAYS, 3);
+
   assert.strictEqual(
-    await rg.hasSuccessfulReleaseToday(mockGithub, 'o', 'r', noonUtc),
+    await rg.hasSuccessfulReleaseWithinDays(mockGithub, 'o', 'r', noonUtc),
     true
   );
 
@@ -46,12 +48,34 @@ assert.strictEqual(dayStart.toISOString(), '2026-07-08T00:00:00.000Z');
     { headSha: 'abc', isCiTrigger: false, bump: 'patch', now: noonUtc },
     null
   );
-  assert.ok(result.reasons.some((r) => r.includes('already released today')));
+  assert.ok(result.reasons.some((r) => r.includes('every 3 days')));
+
+  const oldReleaseGithub = {
+    rest: {
+      actions: {
+        listWorkflowRuns: async () => ({
+          data: {
+            workflow_runs: [
+              {
+                conclusion: 'success',
+                created_at: '2026-07-04T09:00:00Z',
+                status: 'completed',
+              },
+            ],
+          },
+        }),
+      },
+    },
+  };
+  assert.strictEqual(
+    await rg.hasSuccessfulReleaseWithinDays(oldReleaseGithub, 'o', 'r', noonUtc),
+    false
+  );
 
   console.log('ok: bumpPatch');
   console.log('ok: readVersionsFromTree');
   console.log('ok: pypiVersionExists missing version');
   console.log('ok: utcDayStart');
-  console.log('ok: hasSuccessfulReleaseToday');
-  console.log('ok: daily dedupe blocks second release');
+  console.log('ok: hasSuccessfulReleaseWithinDays');
+  console.log('ok: 3-day dedupe blocks second release');
 })();

.github/scripts/release-gate.js

@@ -5,6 +5,8 @@
 const https = require('https');
 
 const PACKAGE_PATHS = ['src/praisonai', 'src/praisonai-agents', 'src/praisonai-code'];
+/** Minimum days between successful patch auto-releases. */
+const PATCH_RELEASE_INTERVAL_DAYS = 3;
 const ACTIVE_RELEASE_STATUSES = new Set([
   'queued', 'in_progress', 'waiting', 'pending', 'requested',
 ]);
@@ -79,8 +81,18 @@ function utcDayStart(now = new Date()) {
   return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate()));
 }
 
-async function hasSuccessfulReleaseToday(github, owner, repo, now = new Date()) {
-  const dayStart = utcDayStart(now);
+function releaseIntervalStart(now = new Date(), intervalDays = PATCH_RELEASE_INTERVAL_DAYS) {
+  return new Date(now.getTime() - intervalDays * 24 * 60 * 60 * 1000);
+}
+
+async function hasSuccessfulReleaseWithinDays(
+  github,
+  owner,
+  repo,
+  now = new Date(),
+  intervalDays = PATCH_RELEASE_INTERVAL_DAYS,
+) {
+  const windowStart = releaseIntervalStart(now, intervalDays);
   const runs = await github.rest.actions.listWorkflowRuns({
     owner,
     repo,
@@ -89,10 +101,15 @@ async function hasSuccessfulReleaseToday(github, owner, repo, now = new Date())
     per_page: 30,
   });
   return runs.data.workflow_runs.some(
-    (r) => r.conclusion === 'success' && new Date(r.created_at) >= dayStart
+    (r) => r.conclusion === 'success' && new Date(r.created_at) >= windowStart
   );
 }
 
+/** @deprecated Use hasSuccessfulReleaseWithinDays — kept for selftests. */
+async function hasSuccessfulReleaseToday(github, owner, repo, now = new Date()) {
+  return hasSuccessfulReleaseWithinDays(github, owner, repo, now, 1);
+}
+
 async function lastGreenCoreTestsSha(github, owner, repo) {
   const runs = await github.rest.actions.listWorkflowRuns({
     owner,
@@ -135,9 +152,11 @@ async function evaluateReleasePreflight(github, owner, repo, options, core) {
   }
 
   const referenceTime = options.now instanceof Date ? options.now : new Date();
-  if (await hasSuccessfulReleaseToday(github, owner, repo, referenceTime)) {
-    const day = referenceTime.toISOString().slice(0, 10);
-    reasons.push(`already released today (UTC ${day}); max one patch release per day`);
+  if (await hasSuccessfulReleaseWithinDays(github, owner, repo, referenceTime)) {
+    reasons.push(
+      `already released within last ${PATCH_RELEASE_INTERVAL_DAYS} days; `
+      + `max one patch release every ${PATCH_RELEASE_INTERVAL_DAYS} days`
+    );
     return out;
   }
 
@@ -222,6 +241,9 @@ module.exports = {
   bumpPatch,
   readVersionsFromTree,
   pypiVersionExists,
+  PATCH_RELEASE_INTERVAL_DAYS,
+  releaseIntervalStart,
+  hasSuccessfulReleaseWithinDays,
   hasSuccessfulReleaseToday,
   utcDayStart,
   evaluateReleasePreflight,

.github/workflows/nightly-release-gate.yml

@@ -1,12 +1,12 @@
 name: Nightly Release Gate
 
 # Preflight before PyPI Release (patch bump).
-# Triggers: Core Tests success on main (workflow_run) or nightly cron at 00:00 UTC.
-# Dedupe: max one successful patch release per UTC day (see release-gate.js).
+# Triggers: Core Tests success on main (workflow_run) or cron every 3 days at 00:00 UTC.
+# Dedupe: max one successful patch release every 3 days (see release-gate.js).
 
 on:
   schedule:
-    - cron: '0 0 * * *'
+    - cron: '0 0 */3 * *'
   workflow_dispatch:
     inputs:
       dry_run:

.github/workflows/pypi-release.yml

@@ -134,7 +134,9 @@ jobs:
         'pypi-auto' || 'pypi'
       }}
     env:
-      UV_PUBLISH_TOKEN: ${{ secrets.PYPI_TOKEN || secrets.PYPI_API_TOKEN }}
+      # PyPI API token auth: username __token__, password = pypi-… token (see pypi.org/help/#invalid-auth)
+      UV_PUBLISH_USERNAME: __token__
+      UV_PUBLISH_PASSWORD: ${{ secrets.PYPI_TOKEN || secrets.PYPI_API_TOKEN }}
       GH_TOKEN: ${{ secrets.GH_TOKEN }}
     steps:
       - name: Checkout main

.github/workflows/test-optimized.yml

@@ -782,6 +782,9 @@ jobs:
           tests/unit/test_decorator_simple.py
           tests/unit/test_db_adapter_tracing_init.py
           ../praisonai-bot/tests/unit/test_bot_wiring.py
+          ../praisonai-bot/tests/unit/bots/test_enable_shell.py
+          ../praisonai-bot/tests/unit/gateway/test_preflight.py
+          ../praisonai-bot/tests/unit/gateway/test_gateway_doctor.py
           ../praisonai-bot/tests/unit/test_bot_history.py
           ../praisonai-bot/tests/unit/test_bot_fixes.py
           tests/unit/test_auto_lazy_loading.py

api.md

@@ -83,6 +83,7 @@ Methods:
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">run_autonomous_async</a>(prompt: str, max_iterations: Optional[int] = None, timeout_seconds: Optional[float] = None, completion_promise: Optional[str] = None, clear_context: bool = False)</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">run_until</a>(prompt: str, criteria: str = '', threshold: float = 8.0, max_iterations: int = 5, mode: str = 'optimize', on_iteration: Optional[Callable[[Any], None]] = None, verbose: bool = False, goal: Optional[str] = None, goal_criteria: Optional[Any] = None, judge_model: Optional[str] = None) -> 'EvaluationLoopResult'</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">run_until_async</a>(prompt: str, criteria: str, threshold: float = 8.0, max_iterations: int = 5, mode: str = 'optimize', on_iteration: Optional[Callable[[Any], None]] = None, verbose: bool = False) -> 'EvaluationLoopResult'</code>
+* <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">set_snapshot_root</a>(project_path: str) -> bool</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">skill_manager</a>() -> Optional[Any]</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">store_memory</a>(content: str, memory_type: str = 'short_term', action: str = 'add', **kwargs: Any) -> None</code>
 * <code title="class Agent">Agent.<a href="./src/praisonai-agents/praisonaiagents/agent/agent.py">stream_emitter</a>() -> Optional[Any]</code>
@@ -568,6 +569,7 @@ Methods:
 * <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">get_prompts</a>() -> List[dict]</code>
 * <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">get_resources</a>() -> List[dict]</code>
 * <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">get_tools</a>() -> List[Callable]</code>
+* <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">list_active_server_names</a>() -> set</code>
 * <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">shutdown</a>()</code>
 * <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">to_openai_tool</a>()</code>
 * <code title="class MCP">MCP.<a href="./src/praisonai-agents/praisonaiagents/mcp/mcp.py">with_tool_prefix</a>(prefix: str) -> 'MCP'</code>
@@ -731,7 +733,7 @@ Methods:
 
 Types:
 ```python
-from praisonaiagents import AgentAppConfig, AgentAppProtocol, AgentFlow, AgentManager, AgentOSConfig, AgentOSProtocol, AgentTeam, AutoApproveBackend, EmbeddingResult, RetryBackoffConfig, __version__, aembedding, aembeddings, embedding, embeddings, get_dimensions
+from praisonaiagents import AgentAppConfig, AgentAppProtocol, AgentFlow, AgentManager, AgentOSConfig, AgentOSProtocol, AgentTeam, AutoApproveBackend, EmbeddingResult, RetryBackoffConfig, RunOutcome, __version__, aembedding, aembeddings, embedding, embeddings, get_dimensions
 ```
 
 Methods:
@@ -799,12 +801,16 @@ Methods:
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">set_state</a>(key: str, value: Any) -> None</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">spawn_sub_agent</a>(agent: Agent, task: Any, completion_callback: Optional[Callable[[SubAgentCompletionEvent], Any]] = None, metadata: Optional[Dict[str, Any]] = None) -> SpawnedSubAgent</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">start</a>(content = None, return_dict = False, output = None, **kwargs)</code>
+* <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">stream_emitter</a>()</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">todo_list</a>()</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">update_plan_step_status</a>(step_id: str, status: str) -> bool</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">update_state</a>(updates: Dict) -> None</code>
 * <code title="class AgentTeam">AgentTeam.<a href="./src/praisonai-agents/praisonaiagents/agents/agents.py">wait_for_completions</a>(timeout: Optional[float] = None, agent_ids: Optional[List[str]] = None) -> List[SubAgentCompletionEvent]</code>
 * <code title="class AutoApproveBackend">AutoApproveBackend.<a href="./src/praisonai-agents/praisonaiagents/approval/backends.py">request_approval</a>(request: ApprovalRequest) -> ApprovalDecision</code>
 * <code title="class AutoApproveBackend">AutoApproveBackend.<a href="./src/praisonai-agents/praisonaiagents/approval/backends.py">request_approval_sync</a>(request: ApprovalRequest) -> ApprovalDecision</code>
+* <code title="class RunOutcome">RunOutcome.<a href="./src/praisonai-agents/praisonaiagents/agent/run_outcome.py">completed</a>(output: Optional[str] = None) -> 'RunOutcome'</code>
+* <code title="class RunOutcome">RunOutcome.<a href="./src/praisonai-agents/praisonaiagents/agent/run_outcome.py">from_exception</a>(exc: BaseException, output: Optional[str] = None) -> 'RunOutcome'</code>
+* <code title="class RunOutcome">RunOutcome.<a href="./src/praisonai-agents/praisonaiagents/agent/run_outcome.py">succeeded</a>() -> bool</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/embedding/embed.py">aembedding</a>(input: Union[str, List[str]], model: str = 'text-embedding-3-small', dimensions: Optional[int] = None, encoding_format: str = 'float', timeout: float = 600.0, api_key: Optional[str] = None, api_base: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs) -> EmbeddingResult</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/embedding/embed.py">aembeddings</a>(input: Union[str, List[str]], model: str = 'text-embedding-3-small', dimensions: Optional[int] = None, encoding_format: str = 'float', timeout: float = 600.0, api_key: Optional[str] = None, api_base: Optional[str] = None, metadata: Optional[Dict[str, Any]] = None, **kwargs) -> EmbeddingResult</code>
 * <code title="function">praisonaiagents.<a href="./src/praisonai-agents/praisonaiagents/embedding/dimensions.py">get_dimensions</a>(model_name: str) -> int</code>

docker/Dockerfile.chat

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.155" \
+    "praisonai>=4.6.157" \
     "praisonai[chat]" \
     "embedchain[github,youtube]"
 

docker/Dockerfile.dev

@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.155" \
+    "praisonai>=4.6.157" \
     "praisonai[ui]" \
     "praisonai[chat]" \
     "praisonai[realtime]" \

docker/Dockerfile.ui

@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
 # Install Python packages (using latest versions)
 RUN pip install --no-cache-dir \
     praisonai_tools \
-    "praisonai>=4.6.155" \
+    "praisonai>=4.6.157" \
     "praisonai[ui]" \
     "praisonai[crewai]"
 

examples/python/api/simple-mcp-server.py

@@ -1,5 +1,15 @@
+import threading
+import time
+
 from praisonaiagents import Agent
 
+
 if __name__ == "__main__":
     agent = Agent(name="TweetAgent", instructions="Create a Tweet based on the topic provided")
-    agent.launch(port=8080, protocol="mcp", path="/mcp")
+    server = threading.Thread(
+        target=lambda: agent.launch(port=8080, protocol="mcp", path="/mcp"),
+        daemon=True,
+    )
+    server.start()
+    time.sleep(2)
+    print("MCP server started on http://127.0.0.1:8080/mcp")

examples/python/concepts/knowledge-reranker-example.py

@@ -150,7 +150,7 @@ def main():
     agent = Agent(
         name="AI Research Assistant",
         instructions="You are an AI research assistant. Use your knowledge base to answer questions about artificial intelligence topics.",
-        knowledge={**rerank_config, "sources": sample_documents},
+        knowledge={"sources": sample_documents, "rerank": True},
         llm="gpt-4o-mini"
     )
     

examples/python/memory/advanced-graph-memory-integration.py

@@ -67,6 +67,7 @@ agents_system = AgentTeam(
 print("Starting graph memory demonstration...")
 result = agents_system.start()
 
-print(f"\nGraph Memory Result: {result[:200]}...")
+result_text = str(result) if result is not None else ""
+print(f"\nGraph Memory Result: {result_text[:200]}...")
 print("\n✅ Graph memory integration complete!")
 print("Agent built knowledge graph and performed relationship-aware queries.")

examples/terminal_bench/test_basic.py

@@ -14,18 +14,18 @@ def test_imports():
     """Test that all required modules can be imported."""
     try:
         from praisonaiagents import Agent
-        print("✅ Agent imported successfully")
+        print("[OK] Agent imported successfully")
         
         from praisonaiagents.tools import execute_command
-        print("✅ execute_command tool imported successfully")
+        print("[OK] execute_command tool imported successfully")
         
         from praisonaiagents.approval import get_approval_registry, AutoApproveBackend
-        print("✅ Approval system imported successfully")
+        print("[OK] Approval system imported successfully")
         
         return True
         
     except ImportError as e:
-        print(f"❌ Import failed: {e}")
+        print(f"[FAIL] Import failed: {ascii(e)}")
         return False
 
 def test_agent_creation():
@@ -40,11 +40,11 @@ def test_agent_creation():
         )
         
         assert agent.name == 'test-agent'
-        print("✅ Agent creation successful")
+        print("[OK] Agent creation successful")
         return True
         
     except Exception as e:
-        print(f"❌ Agent creation failed: {e}")
+        print(f"[FAIL] Agent creation failed: {ascii(e)}")
         return False
 
 def test_approval_system():
@@ -55,11 +55,11 @@ def test_approval_system():
         registry = get_approval_registry()
         registry.set_backend(AutoApproveBackend())
         
-        print("✅ Approval system configuration successful")
+        print("[OK] Approval system configuration successful")
         return True
         
     except Exception as e:
-        print(f"❌ Approval system test failed: {e}")
+        print(f"[FAIL] Approval system test failed: {ascii(e)}")
         return False
 
 def test_external_agent_import():
@@ -80,11 +80,11 @@ def test_external_agent_import():
         
         # Now try to import our agent
         import praisonai_external_agent
-        print("✅ External agent import successful")
+        print("[OK] External agent import successful")
         return True
         
     except Exception as e:
-        print(f"❌ External agent import failed: {e}")
+        print(f"[FAIL] External agent import failed: {ascii(e)}")
         return False
 
 def test_multi_agent_import():
@@ -104,11 +104,11 @@ def test_multi_agent_import():
         sys.modules['harbor.models.agent.context'] = MagicMock()
         
         import multi_agent_example
-        print("✅ Multi-agent example import successful")
+        print("[OK] Multi-agent example import successful")
         return True
         
     except Exception as e:
-        print(f"❌ Multi-agent example import failed: {e}")
+        print(f"[FAIL] Multi-agent example import failed: {ascii(e)}")
         return False
 
 def test_wrapper_agent_import():
@@ -128,11 +128,11 @@ def test_wrapper_agent_import():
         sys.modules['harbor.models.agent.context'] = MagicMock()
         
         import praisonai_wrapper_agent
-        print("✅ Wrapper agent import successful (CLI-based approach)")
+        print("[OK] Wrapper agent import successful (CLI-based approach)")
         return True
         
     except Exception as e:
-        print(f"❌ Wrapper agent import failed: {e}")
+        print(f"[FAIL] Wrapper agent import failed: {ascii(e)}")
         return False
 
 if __name__ == "__main__":
@@ -152,19 +152,19 @@ if __name__ == "__main__":
     total = len(tests)
     
     for test_name, test_func in tests:
-        print(f"\n🧪 Testing {test_name}...")
+        print(f"\n[TEST] {test_name}...")
         try:
             if test_func():
                 passed += 1
         except Exception as e:
-            print(f"❌ Test {test_name} crashed: {e}")
+            print(f"[FAIL] Test {test_name} crashed: {ascii(e)}")
     
     print("\n" + "=" * 55)
     print(f"Results: {passed}/{total} tests passed")
     
     if passed == total:
-        print("🎉 ALL BASIC TESTS PASSED!")
+        print("[OK] ALL BASIC TESTS PASSED!")
         print("The Terminal-Bench integration components are properly structured.")
     else:
-        print("❌ Some tests failed - check output above")
+        print("[FAIL] Some tests failed - check output above")
         sys.exit(1)

examples/tools/external/hackernews/tool.py

@@ -30,9 +30,11 @@ def main():
     else:
         print(f"Found {len(stories)} top stories:")
         for story in stories:
-            print(f"  - {story.get('title', 'N/A')[:50]}...")
+            title = story.get("title") or "N/A"
+            url = story.get("url") or "N/A"
+            print(f"  - {title[:50]}...")
             print(f"    Score: {story.get('score', 'N/A')} | Comments: {story.get('descendants', 'N/A')}")
-            print(f"    URL: {story.get('url', 'N/A')[:60]}...")
+            print(f"    URL: {url[:60]}...")
             print()
     
     print("✅ Hacker News tool working correctly!")

examples/yaml/gateway.yaml

@@ -19,6 +19,9 @@ agents:
 channels:
   telegram:
     token: "${TELEGRAM_BOT_TOKEN}"
+    # allow_shell: true
+    # auto_approve_shell: false
+    # approval_channel: "123456789"   # Telegram chat ID for shell approvals
     routing:
       dm: "personal"
       group: "support"
@@ -26,6 +29,9 @@ channels:
 
   discord:
     token: "${DISCORD_BOT_TOKEN}"
+    # allow_shell: true
+    # auto_approve_shell: false
+    # approval_channel: "9876543210"  # Discord channel ID (or use home_channel)
     routing:
       dm: "personal"
       channel: "support"
@@ -34,6 +40,8 @@ channels:
   slack:
     token: "${SLACK_BOT_TOKEN}"
     app_token: "${SLACK_APP_TOKEN}"
+    allow_shell: true
+    auto_approve_shell: true
     routing:
       dm: "personal"
       channel: "support"

src/praisonai-agents/DOCS_PARITY.md

@@ -18,39 +18,39 @@ This report compares **Python SDK feature categories** against **Python document
 | Category | Features | Docs | Lines |
 |----------|----------|------|-------|
 | ✅ AGUI | 1 | 2 | 420 |
-| ✅ Agent | 21 | 45 | 14056 |
+| ✅ Agent | 21 | 46 | 14663 |
 | ✅ Agent-to-Agent (A2A) | 1 | 7 | 2965 |
 | ✅ Agent-to-User (A2U) | 1 | 3 | 585 |
-| ✅ Approval | 1 | 5 | 2286 |
+| ✅ Approval | 1 | 5 | 2312 |
 | ✅ Audio | 2 | 12 | 1000 |
-| ✅ Auto Generation | 5 | 10 | 2868 |
+| ✅ Auto Generation | 5 | 10 | 2880 |
 | ✅ Autonomy | 3 | 7 | 2707 |
-| ✅ Bots | 7 | 28 | 11165 |
+| ✅ Bots | 7 | 31 | 12353 |
 | ✅ Budget | 1 | 1 | 287 |
-| ✅ CLI | 5 | 119 | 33595 |
+| ✅ CLI | 5 | 119 | 34537 |
 | ✅ Chunking | 2 | 2 | 385 |
 | ✅ Citations | 2 | 1 | 202 |
-| ✅ Code Execution | 2 | 13 | 3907 |
+| ✅ Code Execution | 2 | 13 | 4031 |
 | ✅ Conditions | 1 | 3 | 1179 |
-| ✅ Configuration | 3 | 7 | 2772 |
+| ✅ Configuration | 3 | 7 | 2773 |
 | ✅ Context Management | 16 | 34 | 13204 |
 | ✅ Deep Research | 8 | 2 | 587 |
 | ✅ Display | 6 | 3 | 741 |
 | ✅ Embeddings | 6 | 23 | 2049 |
-| ✅ Evaluation | 1 | 7 | 2902 |
+| ✅ Evaluation | 1 | 7 | 2905 |
 | ✅ Events | 1 | 2 | 749 |
 | ✅ Execution | 3 | 4 | 1528 |
 | ✅ Failover | 2 | 1 | 400 |
-| ✅ Files | 2 | 6 | 1964 |
+| ✅ Files | 2 | 6 | 2066 |
 | ✅ Flow | 1 | 3 | 777 |
-| ✅ Gateway | 7 | 52 | 18862 |
+| ✅ Gateway | 7 | 59 | 21584 |
 | ✅ Guardrails | 4 | 4 | 1673 |
 | ✅ Handoffs | 11 | 6 | 2454 |
-| ✅ Hooks | 2 | 9 | 3985 |
+| ✅ Hooks | 2 | 9 | 4128 |
 | ✅ Image | 1 | 11 | 1255 |
 | ✅ Knowledge | 4 | 13 | 4160 |
-| ✅ LLM | 3 | 14 | 4767 |
-| ✅ Loops | 4 | 4 | 1202 |
+| ✅ LLM | 3 | 14 | 4864 |
+| ✅ Loops | 4 | 5 | 1422 |
 | ✅ MCP | 1 | 58 | 13322 |
 | ✅ Memory | 6 | 17 | 6635 |
 | ✅ OCR | 2 | 1 | 237 |
@@ -59,23 +59,23 @@ This report compares **Python SDK feature categories** against **Python document
 | ✅ Output | 3 | 5 | 1158 |
 | ✅ Parallel Execution | 3 | 2 | 497 |
 | ✅ Planning | 6 | 6 | 1545 |
-| ✅ Plugins | 8 | 3 | 1879 |
-| ✅ Prompts | 2 | 8 | 1726 |
-| ✅ Providers | 1 | 55 | 7459 |
+| ✅ Plugins | 8 | 3 | 1900 |
+| ✅ Prompts | 2 | 9 | 1965 |
+| ✅ Providers | 1 | 55 | 7502 |
 | ✅ Query | 1 | 2 | 736 |
 | ✅ RAG | 5 | 15 | 2936 |
 | ✅ Realtime | 2 | 5 | 668 |
 | ✅ Reflection | 3 | 3 | 680 |
 | ✅ Retrieval | 2 | 5 | 1044 |
 | ✅ Routing | 1 | 2 | 430 |
-| ✅ Sandbox | 5 | 6 | 2700 |
-| ✅ Security | 1 | 3 | 2908 |
-| ✅ Sessions | 4 | 13 | 5286 |
-| ✅ Skills | 6 | 13 | 4946 |
-| ✅ Tasks | 2 | 6 | 2743 |
+| ✅ Sandbox | 5 | 6 | 2766 |
+| ✅ Security | 1 | 3 | 2970 |
+| ✅ Sessions | 4 | 13 | 5302 |
+| ✅ Skills | 6 | 13 | 4998 |
+| ✅ Tasks | 2 | 6 | 2809 |
 | ✅ Telemetry | 1 | 2 | 604 |
-| ✅ Templates | 1 | 8 | 1657 |
-| ✅ Tools | 12 | 132 | 37592 |
+| ✅ Templates | 1 | 8 | 1663 |
+| ✅ Tools | 12 | 133 | 38149 |
 | ✅ Tracing | 3 | 2 | 139 |
 | ✅ Vector Store | 1 | 12 | 1193 |
 | ✅ Video | 2 | 6 | 649 |
@@ -87,7 +87,7 @@ This report compares **Python SDK feature categories** against **Python document
 
 These docs exist but don't match any implemented feature category:
 
-- ℹ️ Database (40 docs, 6184 lines)
+- ℹ️ Database (41 docs, 6478 lines)
 - ℹ️ Documents (1 docs, 787 lines)
 
 ---

src/praisonai-agents/praisonaiagents/__init__.py

@@ -258,6 +258,7 @@ _LAZY_IMPORTS = {
     
     # Agent classes
     'Agent': ('praisonaiagents.agent.agent', 'Agent'),
+    'RunOutcome': ('praisonaiagents.agent.run_outcome', 'RunOutcome'),
     'RetryBackoffConfig': ('praisonaiagents.agent.retry_utils', 'RetryBackoffConfig'),
     'BudgetExceededError': ('praisonaiagents.errors', 'BudgetExceededError'),
     
@@ -838,6 +839,7 @@ __all__ = [
     
     # Core classes - the essentials
     'Agent',
+    'RunOutcome',
     'RetryBackoffConfig',
     'AgentTeam',  # Primary class for multi-agent coordination (v1.0+)
     'AgentManager',  # Silent alias for AgentTeam

src/praisonai-agents/praisonaiagents/_logging.py

@@ -4,7 +4,6 @@ This module consolidates all logging configuration in one place to avoid duplica
 """
 
 import os
-import json
 import logging
 from typing import Any, Dict, List, Optional, Union
 
@@ -196,6 +195,9 @@ class StructuredFormatter(logging.Formatter):
             extra = {k: v for k, v in record.extra_data.items() if k not in self._STANDARD_FIELDS}
             log_data.update(extra)
 
+        # Deferred: json is only needed for opt-in structured logging
+        import json
+
         return json.dumps(log_data)
 
 

src/praisonai-agents/praisonaiagents/agent/__init__.py

@@ -37,6 +37,15 @@ def __getattr__(name):
         from .interrupt import InterruptController
         _lazy_cache[name] = InterruptController
         return InterruptController
+    if name in ('RunOutcome', 'TerminalReason'):
+        from . import run_outcome as _run_outcome_module
+        value = getattr(_run_outcome_module, name)
+        _lazy_cache[name] = value
+        return value
+    if name == 'prompt_prefix_signature':
+        from .prompt_cache import prompt_prefix_signature
+        _lazy_cache[name] = prompt_prefix_signature
+        return prompt_prefix_signature
     
     # Specialized agents - lazy loaded (import rich)
     if name == 'ImageAgent':
@@ -215,6 +224,9 @@ __all__ = [
     'Heartbeat',
     'HeartbeatConfig',
     'InterruptController',
+    'RunOutcome',
+    'TerminalReason',
+    'prompt_prefix_signature',
     'ImageAgent',
     'VideoAgent',
     'VideoConfig',

src/praisonai-agents/praisonaiagents/agent/agent.py

@@ -20,7 +20,7 @@ from .async_memory_mixin import AsyncMemoryMixin
 from .tool_execution import ToolExecutionMixin, BackoffPolicy
 from .chat_handler import ChatHandlerMixin
 from .session_manager import SessionManagerMixin
-from .async_safety import AsyncSafeState
+from .async_safety import AsyncSafeState, DualLock
 # NOTE: UnifiedExecutionMixin is deprecated and unused by any production path
 # (Issue #2644). It is kept in the MRO for backward compatibility during the
 # deprecation cycle and will be removed afterwards.
@@ -2116,6 +2116,12 @@ Your Goal: {self.goal}
         
         # Store tool retry policy for tool execution with exponential backoff
         self._tool_retry_policy = _tool_config.retry_policy if _tool_config else None
+
+        # Whether undeclared tool names may resolve via the process-global @tool
+        # registry. Default False keeps the agent scoped to its own tools=[...].
+        self._allow_global_tools = (
+            _tool_config.allow_global_tools if _tool_config else False
+        )
         
         # Retry configuration with jittered exponential backoff.
         # Default (retry is None) applies RetryBackoffConfig() so the native
@@ -2224,7 +2230,10 @@ Your Goal: {self.goal}
         # Per-turn tool-name buffer feeding the self-improve review policy.
         # Populated in _execute_tool_with_context, reset each chat turn, and
         # read by _trigger_after_agent_hook when tools_used is not passed.
+        # Guarded by a DualLock (like chat_history) so concurrent chat()/achat()
+        # turns on the same Agent instance don't corrupt each other's buffer.
         self._turn_tools_used = []
+        self._turn_tools_lock = DualLock()
 
         # Database persistence (lazy - no imports until used)
         self._db = db
@@ -3192,7 +3201,44 @@ Summary:"""
     # ================================================================
     # Filesystem tracking convenience methods (powered by FileSnapshot)
     # ================================================================
-    
+
+    def set_snapshot_root(self, project_path: str) -> bool:
+        """Root filesystem change-tracking at ``project_path``.
+
+        Bots/gateway resolve file tools against a per-chat ``Workspace`` and
+        attach it *after* construction, but the FileSnapshot backing
+        :meth:`undo`/:meth:`redo`/:meth:`diff` was created at ``__init__`` time
+        rooted at ``os.getcwd()``. Without this, ``/undo`` tracks the wrong
+        directory (never where the tools actually wrote). Call this once after
+        the workspace is known so change tracking follows the tools.
+
+        Rooting at a new directory clears the undo/redo stacks (they belong to
+        the previous root). A no-op when the root is unchanged. Returns ``True``
+        when a snapshot manager is rooted at ``project_path``.
+        """
+        import os
+        new_root = os.path.abspath(str(project_path))
+        current = self._file_snapshot
+        if current is not None and getattr(current, "project_path", None) == new_root:
+            return True
+        try:
+            from ..snapshot import FileSnapshot
+            snapshot_dir = None
+            cfg = getattr(self, "autonomy_config", None)
+            if isinstance(cfg, dict):
+                snapshot_dir = cfg.get("snapshot_dir")
+            self._file_snapshot = FileSnapshot(
+                project_path=new_root,
+                snapshot_dir=snapshot_dir,
+            )
+            with self._snapshot_lock:
+                self._snapshot_stack = []
+                self._redo_stack = []
+            return True
+        except Exception as e:  # pragma: no cover - git may be unavailable
+            logger.debug(f"Re-rooting FileSnapshot failed: {e}")
+            return False
+
     def undo(self) -> bool:
         """Undo the last set of file changes.
         
@@ -6117,6 +6163,12 @@ Answer:"""
                 images=images,
                 system_prompt=system_prompt
             )
+
+            await self._emit_cli_backend_hook(
+                backend=backend,
+                session_id=session_id,
+                result=result,
+            )
             
             # Check for CLI backend errors
             if result is None:
@@ -6145,10 +6197,55 @@ Answer:"""
             return result.content if result else None
             
         except Exception as e:
+            await self._emit_cli_backend_hook(
+                backend=backend,
+                session_id=session_id,
+                result=None,
+                error=str(e),
+            )
             raise RuntimeError(
                 f"CLI backend execution failed for agent={self.display_name!r}: {e}"
             ) from e
-    
+
+    async def _emit_cli_backend_hook(
+        self,
+        *,
+        backend: Any,
+        session_id: Optional[str],
+        result: Any,
+        error: Optional[str] = None,
+    ) -> None:
+        """Emit the CLI_BACKEND_EXECUTE hook (no-op when no listener is registered).
+
+        Fires on both success and failure so subprocess startup errors and
+        timeouts remain observable. Prompt-bearing argv values are redacted at
+        the payload serialization boundary.
+        """
+        from ..hooks.types import HookEvent
+
+        if not self._hook_runner.registry.has_hooks(HookEvent.CLI_BACKEND_EXECUTE):
+            return
+
+        from ..hooks.events import CliBackendExecuteInput
+        from ..cli_backend.debug import backend_label
+
+        metadata = getattr(result, "metadata", None) or {}
+        hook_input = CliBackendExecuteInput(
+            session_id=session_id,
+            cwd=os.getcwd(),
+            event_name=HookEvent.CLI_BACKEND_EXECUTE.value,
+            timestamp=str(time.time()),
+            agent_name=self.display_name,
+            backend=backend_label(backend),
+            command=metadata.get("command"),
+            content=getattr(result, "content", None),
+            error=error if error is not None else getattr(result, "error", None),
+        )
+        await self._hook_runner.execute(
+            HookEvent.CLI_BACKEND_EXECUTE,
+            hook_input,
+        )
+
     # -------------------------------------------------------------------------
     #                       Resource Lifecycle Management
     # -------------------------------------------------------------------------

src/praisonai-agents/praisonaiagents/agent/async_memory_mixin.py

@@ -7,7 +7,6 @@ async capabilities following the AsyncMemoryProtocol.
 """
 
 import asyncio
-import logging
 from typing import List, Dict, Any, Optional, Union
 from praisonaiagents._logging import get_logger
 from ..memory.protocols import AsyncMemoryProtocol

src/praisonai-agents/praisonaiagents/agent/autonomy.py

@@ -34,7 +34,6 @@ Usage:
 from dataclasses import dataclass, field
 from typing import Optional, Dict, Any, List, Set
 from enum import Enum
-import logging
 from praisonaiagents._logging import get_logger
 
 logger = get_logger(__name__)

src/praisonai-agents/praisonaiagents/agent/chat_mixin.py

@@ -625,11 +625,26 @@ Your Goal: {self.goal}"""
                     formatted_tools.extend(openai_tools)
                 elif openai_tools is not None:
                     formatted_tools.append(openai_tools)
-            # Handle callable functions
-            elif callable(tool):
-                tool_def = self._generate_tool_definition(tool.__name__)
+            # Handle BaseTool-style instances (ClarifyTool, etc.)
+            elif getattr(tool, "name", None) and hasattr(tool, "run"):
+                tool_def = self._generate_tool_definition(tool.name)
                 if tool_def:
                     formatted_tools.append(tool_def)
+                else:
+                    logging.warning(
+                        "Could not generate definition for tool: %s", tool.name
+                    )
+            # Handle callable functions
+            elif callable(tool):
+                fname = getattr(tool, "name", None) or getattr(tool, "__name__", None)
+                if not fname:
+                    logging.warning("Tool %r not recognized (no name)", tool)
+                    continue
+                tool_def = self._generate_tool_definition(fname)
+                if tool_def:
+                    formatted_tools.append(tool_def)
+                else:
+                    logging.warning(f"Could not generate definition for tool: {fname}")
             else:
                 logging.warning(f"Tool {tool} not recognized")
         
@@ -2447,8 +2462,7 @@ Your Goal: {self.goal}"""
         """Internal chat implementation (extracted for trace wrapping)."""
         # Reset the per-turn tool buffer so the self-improve review policy only
         # sees tools used in this turn (not during a nested skill-review turn).
-        if not getattr(self, "_in_skill_review", False):
-            self._turn_tools_used = []
+        self._reset_turn_tools()
         # Apply rate limiter if configured (before any LLM call)
         if self._rate_limiter is not None:
             self._rate_limiter.acquire()
@@ -3049,8 +3063,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
         """Internal async chat implementation (extracted for trace wrapping)."""
         # Reset the per-turn tool buffer so the self-improve review policy only
         # sees tools used in this turn (not during a nested skill-review turn).
-        if not getattr(self, "_in_skill_review", False):
-            self._turn_tools_used = []
+        self._reset_turn_tools()
         # C2 - cooperative cancellation: abort early if a pre-set token is given
         _cancel = cancel_token if cancel_token is not None else getattr(self, "interrupt_controller", None)
         if _cancel is not None and getattr(_cancel, "is_set", lambda: False)():
@@ -3718,8 +3731,20 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                         logging.error(f"Failed to parse tool arguments as JSON: {json_error}")
                         arguments = {}
                     
-                    # Find the matching tool
-                    tool = next((t for t in tools if t.__name__ == function_name), None)
+                    # Find the matching tool by comparing every supported identifier:
+                    # __name__ (plain callables), .name (BaseTool instances like
+                    # BrowserBaseTool, or aliased FunctionTools), or the class name.
+                    # Compare all candidates (not short-circuit) so an aliased .name
+                    # that differs from __name__ still resolves and BaseTool
+                    # subclasses without __name__ do not crash dispatch.
+                    tool = next(
+                        (t for t in tools if function_name in (
+                            getattr(t, "__name__", None),
+                            getattr(t, "name", None),
+                            type(t).__name__,
+                        )),
+                        None,
+                    )
                     if not tool:
                         _get_display_functions()['display_error'](f"Tool {function_name} not found")
                         continue
@@ -4419,6 +4444,64 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
             logging.debug(f"[compaction] skipped (non-fatal): {_ce}")
             return False
 
+    def _emit_retry_stream_event(self, *, attempt, max_attempts, delay, reason):
+        """Emit a ``StreamEventType.RETRY`` event during backoff.
+
+        Lets consumers (e.g. the CLI stream-json bridge) render a live
+        "retrying in Ns (attempt k/N)" status instead of appearing hung.
+        Guarded by ``has_callbacks`` so it stays zero-overhead when nothing is
+        listening, and never raises into the retry loop.
+        """
+        emitter = getattr(self, 'stream_emitter', None)
+        if emitter is None or not getattr(emitter, 'has_callbacks', False):
+            return
+        try:
+            from ..streaming.events import StreamEvent, StreamEventType
+            emitter.emit(self._build_retry_stream_event(
+                StreamEvent, StreamEventType,
+                attempt=attempt, max_attempts=max_attempts, delay=delay, reason=reason,
+            ))
+        except Exception as _re:
+            logger.debug(f"Failed to emit RETRY stream event: {_re}")
+
+    async def _aemit_retry_stream_event(self, *, attempt, max_attempts, delay, reason):
+        """Async counterpart of ``_emit_retry_stream_event``.
+
+        Uses ``emit_async`` so consumers registered via the public
+        ``add_async_callback()`` API also receive RETRY events; ``emit`` alone
+        only visits synchronous callbacks. Same zero-overhead guard and
+        never-raise contract as the sync path.
+        """
+        emitter = getattr(self, 'stream_emitter', None)
+        if emitter is None or not getattr(emitter, 'has_callbacks', False):
+            return
+        try:
+            from ..streaming.events import StreamEvent, StreamEventType
+            event = self._build_retry_stream_event(
+                StreamEvent, StreamEventType,
+                attempt=attempt, max_attempts=max_attempts, delay=delay, reason=reason,
+            )
+            emit_async = getattr(emitter, 'emit_async', None)
+            if emit_async is not None:
+                await emit_async(event)
+            else:
+                emitter.emit(event)
+        except Exception as _re:
+            logger.debug(f"Failed to emit RETRY stream event: {_re}")
+
+    def _build_retry_stream_event(self, StreamEvent, StreamEventType, *, attempt, max_attempts, delay, reason):
+        """Construct the RETRY ``StreamEvent`` shared by the sync/async emitters."""
+        return StreamEvent(
+            type=StreamEventType.RETRY,
+            metadata={
+                "attempt": attempt,
+                "max_attempts": max_attempts,
+                "delay": delay,
+                "reason": reason,
+            },
+            agent_id=getattr(self, 'name', None),
+        )
+
     def _chat_completion_with_retry(self, messages, temperature=1.0, tools=None, stream=None, reasoning_steps=False, task_name=None, task_description=None, task_id=None, response_format=None, stream_callback=None, emit_events=True):
         """
         Wrapper for _execute_unified_chat_completion that adds jittered exponential backoff retry logic.
@@ -4480,6 +4563,17 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                 )
                 self._hook_runner.execute_sync(HookEvent.ON_RETRY, retry_input)
                 
+                # Surface the retry as a first-class stream event so CLI/UI
+                # consumers can show a live "retrying in Ns" status instead of
+                # appearing hung. Guarded so it is zero-overhead when nothing
+                # is listening.
+                self._emit_retry_stream_event(
+                    attempt=attempt + 1,
+                    max_attempts=max_attempts,
+                    delay=delay,
+                    reason=str(e),
+                )
+                
                 # Log retry attempt (buffered to avoid spam during transient failures)
                 logger.debug(f"[{self.name}] Retry {attempt + 1}/{max_attempts} after {delay:.1f}s: {str(e)[:100]}")
                 
@@ -4561,6 +4655,16 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                 )
                 await self._hook_runner.execute_async(HookEvent.ON_RETRY, retry_input)
                 
+                # Surface a streaming RETRY event (see sync path for rationale).
+                # Use the async emitter so async-only consumers registered via
+                # add_async_callback() also receive the event.
+                await self._aemit_retry_stream_event(
+                    attempt=attempt + 1,
+                    max_attempts=max_attempts,
+                    delay=delay,
+                    reason=str(e),
+                )
+                
                 # Log retry attempt
                 logger.debug(f"[{self.name}] Async retry {attempt + 1}/{max_attempts} after {delay:.1f}s: {str(e)[:100]}")
                 

src/praisonai-agents/praisonaiagents/agent/cost_persistence.py

@@ -12,7 +12,6 @@ Usage::
 """
 
 import json
-import logging
 from praisonaiagents._logging import get_logger
 import os
 import time

src/praisonai-agents/praisonaiagents/agent/deep_research_agent.py

@@ -34,7 +34,6 @@ Example:
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 import time
 import asyncio

src/praisonai-agents/praisonaiagents/agent/execution_mixin.py

@@ -125,16 +125,30 @@ class ExecutionMixin:
         
         Args:
             prompt: The input prompt to process
-            **kwargs: Additional arguments passed to achat()
+            **kwargs: Additional arguments passed to achat():
+                - return_outcome (bool): If True, return a canonical
+                  ``RunOutcome`` (completed | hard_timeout | cancelled |
+                  aborted | failed) instead of raising on error/timeout/
+                  cancellation. Default: False.
+                - timeout (float): When used with ``return_outcome=True``,
+                  a hard timeout budget for the run.
             
         Returns:
-            The agent's response as a string, or AutonomyResult if autonomy enabled
+            The agent's response as a string, or AutonomyResult if autonomy
+            enabled, or a ``RunOutcome`` when ``return_outcome=True``.
             
         Note:
             If autonomy=True was set on the agent, astart() automatically uses
             the autonomous loop (run_autonomous_async) instead of single-turn chat.
         """
         import sys
+
+        return_outcome = kwargs.pop('return_outcome', False)
+        if return_outcome:
+            outcome_timeout = kwargs.pop('timeout', None)
+            return await self._astart_with_outcome(
+                prompt, outcome_timeout, **kwargs
+            )
         
         # ─────────────────────────────────────────────────────────────────────
         # UNIFIED AUTONOMY API: If autonomy is enabled, route to run_autonomous_async
@@ -196,6 +210,57 @@ class ExecutionMixin:
         kwargs['stream'] = stream_requested
         return await self.achat(prompt, **kwargs)
 
+    async def _astart_with_outcome(self, prompt, timeout=None, **kwargs):
+        """Run astart() and normalise its result/exception into a RunOutcome.
+
+        Optionally enforces a hard run-level timeout budget. The budget is the
+        *authoritative* source of ``hard_timeout``: when our own
+        ``asyncio.wait_for`` fires we return ``hard_timeout`` directly, so a
+        nested operation timeout (which surfaces as a generic
+        ``asyncio.TimeoutError`` from inside the run) is never misreported as a
+        run-budget timeout — it falls through to ``failed`` via
+        ``RunOutcome.from_exception``.
+
+        External ``asyncio.CancelledError`` (e.g. the enclosing task being
+        cancelled during host shutdown) is re-raised, not swallowed, so
+        cooperative cancellation semantics are honoured.
+        """
+        import asyncio
+        from .run_outcome import RunOutcome
+
+        enforce_budget = timeout is not None and timeout > 0
+
+        # Enforce the run budget with an explicit deadline on a task we own, so
+        # that only *our* budget expiry maps to hard_timeout. This deliberately
+        # avoids ``asyncio.wait_for`` catching a bare inner ``TimeoutError``:
+        # a nested operation timeout raised inside the run must fall through to
+        # ``failed`` rather than be misreported as a run-budget ``hard_timeout``.
+        try:
+            if enforce_budget:
+                task = asyncio.ensure_future(self.astart(prompt, **kwargs))
+                try:
+                    done, _ = await asyncio.wait({task}, timeout=timeout)
+                except asyncio.CancelledError:
+                    # Enclosing task cancelled (e.g. host shutdown): cancel the
+                    # child and propagate so cooperative cancellation is honoured.
+                    task.cancel()
+                    raise
+                if task not in done:
+                    # Our budget expired first: authoritative, sticky hard_timeout.
+                    task.cancel()
+                    return RunOutcome(reason="hard_timeout")
+                result = task.result()
+            else:
+                result = await self.astart(prompt, **kwargs)
+        except asyncio.CancelledError:
+            # External cancellation of the awaiting task: honour asyncio
+            # cancellation / host shutdown by propagating instead of converting
+            # it into a benign RunOutcome.
+            raise
+        except Exception as exc:  # noqa: BLE001 - normalised into outcome
+            return RunOutcome.from_exception(exc)
+        return RunOutcome.completed(output=str(result) if result is not None else None)
+
     def run(self, prompt: str, **kwargs: Any) -> Optional[str]:
         """Execute agent silently and return structured result.
         
@@ -208,9 +273,14 @@ class ExecutionMixin:
             **kwargs: Additional arguments:
                 - stream (bool): Force streaming if True. Default: False
                 - output (str): Output preset override (rarely needed)
+                - return_outcome (bool): If True, return a canonical
+                  ``RunOutcome`` describing how the run ended
+                  (completed | hard_timeout | cancelled | aborted | failed)
+                  instead of raising on error. Default: False.
                 
         Returns:
-            The agent's response as a string
+            The agent's response as a string, or a ``RunOutcome`` when
+            ``return_outcome=True``.
             
         Example:
             ```python
@@ -227,8 +297,14 @@ class ExecutionMixin:
             - Background processing
             - API endpoints
         """
+        return_outcome = kwargs.pop('return_outcome', False)
+
         # Check if external managed backend is configured
         if hasattr(self, 'backend') and self.backend is not None:
+            if return_outcome:
+                return self._run_with_outcome(
+                    lambda: self._delegate_to_backend(prompt, **kwargs)
+                )
             return self._delegate_to_backend(prompt, **kwargs)
         
         # Production defaults: no streaming, no display
@@ -242,17 +318,34 @@ class ExecutionMixin:
         
         # Load history context
         self._load_history_context()
-        
-        # Check if planning mode is enabled
-        if self.planning:
-            result = self._start_with_planning(prompt, **kwargs)
-        else:
-            result = self.chat(prompt, **kwargs)
-        
-        # Auto-save session if enabled
-        self._auto_save_session()
-        
-        return result
+
+        def _execute():
+            if self.planning:
+                _result = self._start_with_planning(prompt, **kwargs)
+            else:
+                _result = self.chat(prompt, **kwargs)
+            # Auto-save session if enabled
+            self._auto_save_session()
+            return _result
+
+        if return_outcome:
+            return self._run_with_outcome(_execute)
+
+        return _execute()
+
+    def _run_with_outcome(self, executor):
+        """Run ``executor`` and normalise its result/exception into a RunOutcome.
+
+        Keeps the canonical terminal-outcome logic in one place so callers get
+        a single, closed description of how the run ended instead of inferring
+        it from exception identity.
+        """
+        from .run_outcome import RunOutcome
+        try:
+            result = executor()
+        except BaseException as exc:  # noqa: BLE001 - normalised into outcome
+            return RunOutcome.from_exception(exc)
+        return RunOutcome.completed(output=str(result) if result is not None else None)
 
     def _delegate_to_backend(self, prompt: str, **kwargs) -> Optional[str]:
         """Delegate execution to external managed backend (e.g., ManagedAgentIntegration).
@@ -1146,13 +1239,9 @@ Write the complete compiled report:"""
         # Record the tool name for this turn so the self-improve review policy
         # sees async tool usage too (issue #3037). Mirrors the sync path in
         # _execute_tool_with_context; skipped during a guarded review turn so
-        # the review's own calls are not tracked and cannot recurse.
-        if not getattr(self, "_in_skill_review", False):
-            turn_tools = getattr(self, "_turn_tools_used", None)
-            if turn_tools is None:
-                self._turn_tools_used = []
-                turn_tools = self._turn_tools_used
-            turn_tools.append(function_name)
+        # the review's own calls are not tracked and cannot recurse. Locked so
+        # concurrent turns on the same Agent don't corrupt the buffer (#3307).
+        self._record_turn_tool(function_name)
 
         # Enforce BEFORE_TOOL/AFTER_TOOL security hooks for every async caller,
         # mirroring the sync execute_tool path in tool_execution.py. Without
@@ -1369,11 +1458,22 @@ Write the complete compiled report:"""
                     return policy_result  # Error dict
                 _, arguments = policy_result
 
-            # Try to find the function in the override tools list first, then agent's tools list
+            # Try to find the function in the override tools list first, then agent's tools list.
+            # Resolve by BaseTool/FunctionTool ``.name`` (instances like BrowserBaseTool or
+            # aliased decorated tools), plain callable ``__name__``, or class name so async
+            # dispatch matches the robust sync resolution in _execute_tool_impl.
             func = None
             tools_to_search = tools_override if tools_override is not None else self.tools
+            from ..tools.base import BaseTool
             for tool in tools_to_search:
-                if (callable(tool) and getattr(tool, '__name__', '') == function_name):
+                if isinstance(tool, BaseTool) and getattr(tool, 'name', None) == function_name:
+                    func = tool
+                    break
+                if hasattr(tool, 'name') and getattr(tool, 'name', None) == function_name:
+                    func = tool
+                    break
+                if (callable(tool) and getattr(tool, '__name__', '') == function_name) or \
+                   (inspect.isclass(tool) and tool.__name__ == function_name):
                     func = tool
                     break
             
@@ -1382,15 +1482,18 @@ Write the complete compiled report:"""
                 return {"error": f"Function {function_name} not found in tools"}
 
             try:
-                if inspect.iscoroutinefunction(func):
+                # BaseTool instances (plugin system, e.g. BrowserBaseTool) are not
+                # directly callable — dispatch to their .run() method like the sync path.
+                call_target = func.run if isinstance(func, BaseTool) else func
+                if inspect.iscoroutinefunction(call_target):
                     logging.debug(f"Executing async function: {function_name}")
-                    result = await func(**arguments)
+                    result = await call_target(**arguments)
                 else:
                     logging.debug(f"Executing sync function in executor: {function_name}")
                     loop = asyncio.get_running_loop()
                     from ..trace.context_events import copy_context_to_callable
                     result = await loop.run_in_executor(
-                        None, copy_context_to_callable(lambda: func(**arguments))
+                        None, copy_context_to_callable(lambda: call_target(**arguments))
                     )
                 
                 # Ensure result is JSON serializable
@@ -1474,157 +1577,166 @@ Write the complete compiled report:"""
                 
         should_start = False
         with _server_lock:
-            # Initialize port-specific collections if needed
+            # Initialize port-specific collections if needed (once per port)
             if port not in _registered_agents:
                 _registered_agents[port] = {}
 
-                # Initialize shared FastAPI app if not already created for this port
-                if _shared_apps.get(port) is None:
-                    _shared_apps[port] = FastAPI(
-                        title=f"PraisonAI Agents API (Port {port})",
-                        description="API for interacting with PraisonAI Agents"
+            # Initialize shared FastAPI app if not already created for this port
+            if _shared_apps.get(port) is None:
+                _shared_apps[port] = FastAPI(
+                    title=f"PraisonAI Agents API (Port {port})",
+                    description="API for interacting with PraisonAI Agents"
+                )
+
+                # Add a root endpoint with a welcome message
+                @_shared_apps[port].get("/")
+                async def root():
+                    return {
+                        "message": f"Welcome to PraisonAI Agents API on port {port}. See /docs for usage.",
+                        "endpoints": list(_registered_agents[port].keys())
+                    }
+
+                # Add healthcheck endpoint
+                @_shared_apps[port].get("/health")
+                async def healthcheck():
+                    return {
+                        "status": "ok",
+                        "endpoints": list(_registered_agents[port].keys())
+                    }
+
+            # The path registration below must run on EVERY call, not just when the
+            # port is new, so multiple agents can share a single port.
+
+            # Normalize path to ensure it starts with /
+            if not path.startswith('/'):
+                path = f'/{path}'
+
+            # Check if path is already registered for this port
+            if path in _registered_agents[port]:
+                logging.warning(f"Path '{path}' is already registered on port {port}. Please use a different path.")
+                print(f"⚠️ Warning: Path '{path}' is already registered on port {port}.")
+                # Use a modified path to avoid conflicts
+                original_path = path
+                path = f"{path}_{self.agent_id[:6]}"
+                logging.warning(f"Using '{path}' instead of '{original_path}'")
+                print(f"🔄 Using '{path}' instead")
+
+            # Register the agent to this path
+            _registered_agents[port][path] = self.agent_id
+
+            # Define the endpoint handler
+            @_shared_apps[port].post(path)
+            async def handle_agent_query(request: Request, query_data: Optional[AgentQuery] = None):
+                # Handle both direct JSON with query field and form data
+                if query_data is None:
+                    try:
+                        request_data = await request.json()
+                        if "query" not in request_data:
+                            raise HTTPException(status_code=400, detail="Missing 'query' field in request")
+                        query = request_data["query"]
+                    except Exception:
+                        # Fallback to form data or query params
+                        form_data = await request.form()
+                        if "query" in form_data:
+                            query = form_data["query"]
+                        else:
+                            raise HTTPException(status_code=400, detail="Missing 'query' field in request")
+                else:
+                    query = query_data.query
+
+                try:
+                    # Use async version if available, otherwise use sync version
+                    if asyncio.iscoroutinefunction(self.chat):
+                        response = await self.achat(query, task_name=None, task_description=None, task_id=None)
+                    else:
+                        # Run sync function in a thread to avoid blocking
+                        loop = asyncio.get_running_loop()
+                        response = await loop.run_in_executor(None, lambda p=query: self.chat(p))
+
+                    return {"response": response}
+                except Exception as e:
+                    logging.error(f"Error processing query: {str(e)}", exc_info=True)
+                    return JSONResponse(
+                        status_code=500,
+                        content={"error": f"Error processing query: {str(e)}"}
                     )
 
-                    # Add a root endpoint with a welcome message
-                    @_shared_apps[port].get("/")
-                    async def root():
-                        return {
-                            "message": f"Welcome to PraisonAI Agents API on port {port}. See /docs for usage.",
-                            "endpoints": list(_registered_agents[port].keys())
-                        }
+            # Invalidate the cached OpenAPI schema so routes registered by later
+            # shared-port launch() calls still show up in /openapi.json and /docs.
+            # FastAPI caches app.openapi_schema on first access and does not
+            # regenerate it when new routes are added afterwards.
+            _shared_apps[port].openapi_schema = None
 
-                    # Add healthcheck endpoint
-                    @_shared_apps[port].get("/health")
-                    async def healthcheck():
-                        return {
-                            "status": "ok",
-                            "endpoints": list(_registered_agents[port].keys())
-                        }
+            print(f"🚀 Agent '{self.name}' available at http://{host}:{port}")
 
-                # Normalize path to ensure it starts with /
-                if not path.startswith('/'):
-                    path = f'/{path}'
-
-                # Check if path is already registered for this port
-                if path in _registered_agents[port]:
-                    logging.warning(f"Path '{path}' is already registered on port {port}. Please use a different path.")
-                    print(f"⚠️ Warning: Path '{path}' is already registered on port {port}.")
-                    # Use a modified path to avoid conflicts
-                    original_path = path
-                    path = f"{path}_{self.agent_id[:6]}"
-                    logging.warning(f"Using '{path}' instead of '{original_path}'")
-                    print(f"🔄 Using '{path}' instead")
-
-                # Register the agent to this path
-                _registered_agents[port][path] = self.agent_id
-
-                # Define the endpoint handler
-                @_shared_apps[port].post(path)
-                async def handle_agent_query(request: Request, query_data: Optional[AgentQuery] = None):
-                    # Handle both direct JSON with query field and form data
-                    if query_data is None:
-                        try:
-                            request_data = await request.json()
-                            if "query" not in request_data:
-                                raise HTTPException(status_code=400, detail="Missing 'query' field in request")
-                            query = request_data["query"]
-                        except Exception:
-                            # Fallback to form data or query params
-                            form_data = await request.form()
-                            if "query" in form_data:
-                                query = form_data["query"]
-                            else:
-                                raise HTTPException(status_code=400, detail="Missing 'query' field in request")
-                    else:
-                        query = query_data.query
-
-                    try:
-                        # Use async version if available, otherwise use sync version
-                        if asyncio.iscoroutinefunction(self.chat):
-                            response = await self.achat(query, task_name=None, task_description=None, task_id=None)
-                        else:
-                            # Run sync function in a thread to avoid blocking
-                            loop = asyncio.get_running_loop()
-                            response = await loop.run_in_executor(None, lambda p=query: self.chat(p))
-
-                        return {"response": response}
-                    except Exception as e:
-                        logging.error(f"Error processing query: {str(e)}", exc_info=True)
-                        return JSONResponse(
-                            status_code=500,
-                            content={"error": f"Error processing query: {str(e)}"}
-                        )
-
-                print(f"🚀 Agent '{self.name}' available at http://{host}:{port}")
-
-                # Check and mark server as started atomically to prevent race conditions
-                should_start = not _server_started.get(port, False)
-                if should_start:
-                    _server_started[port] = True
-
-            # Server start/wait outside the lock to avoid holding it during sleep  
+            # Check and mark server as started atomically to prevent race conditions
+            should_start = not _server_started.get(port, False)
             if should_start:
-                # Start the server in a separate thread
-                def run_server():
-                    try:
-                        print(f"✅ FastAPI server started at http://{host}:{port}")
-                        print(f"📚 API documentation available at http://{host}:{port}/docs")
-                        print(f"🔌 Available endpoints: {', '.join(list(_registered_agents[port].keys()))}")
-                        uvicorn.run(_shared_apps[port], host=host, port=port, log_level="debug" if debug else "info")
-                    except Exception as e:
-                        logging.error(f"Error starting server: {str(e)}", exc_info=True)
-                        print(f"❌ Error starting server: {str(e)}")
+                _server_started[port] = True
 
-                # Run server in a background thread
-                server_thread = threading.Thread(target=run_server, daemon=True)
-                server_thread.start()
-
-                # Wait for a moment to allow the server to start and register endpoints
-                self._safe_sleep(0.5)
-            else:
-                # If server is already running, wait a moment to make sure the endpoint is registered
-                self._safe_sleep(0.1)
-                print(f"🔌 Available endpoints on port {port}: {', '.join(list(_registered_agents[port].keys()))}")
-            
-            # Get the stack frame to check if this is the last launch() call in the script
-            import inspect
-            stack = inspect.stack()
-            
-            # If this is called from a Python script (not interactive), try to detect if it's the last launch call
-            if len(stack) > 1 and stack[1].filename.endswith('.py'):
-                caller_frame = stack[1]
-                caller_line = caller_frame.lineno
-                
+        # Server start/wait outside the lock to avoid holding it during sleep
+        if should_start:
+            # Start the server in a separate thread
+            def run_server():
                 try:
-                    # Read the file to check if there are more launch calls after this one
-                    with open(caller_frame.filename, 'r') as f:
-                        lines = f.readlines()
-                    
-                    # Check if there are more launch() calls after the current line
-                    has_more_launches = False
-                    for line_content in lines[caller_line:]: # renamed line to line_content
-                        if '.launch(' in line_content and not line_content.strip().startswith('#'):
-                            has_more_launches = True
-                            break
-                    
-                    # If this is the last launch call, block the main thread
-                    if not has_more_launches:
-                        try:
-                            print("\nAll agents registered for HTTP mode. Press Ctrl+C to stop the servers.")
-                            while True:
-                                self._safe_sleep(1)
-                        except KeyboardInterrupt:
-                            print("\nServers stopped")
+                    print(f"✅ FastAPI server started at http://{host}:{port}")
+                    print(f"📚 API documentation available at http://{host}:{port}/docs")
+                    print(f"🔌 Available endpoints: {', '.join(list(_registered_agents[port].keys()))}")
+                    uvicorn.run(_shared_apps[port], host=host, port=port, log_level="debug" if debug else "info")
                 except Exception as e:
-                    # If something goes wrong with detection, block anyway to be safe
-                    logging.error(f"Error in launch detection: {e}")
+                    logging.error(f"Error starting server: {str(e)}", exc_info=True)
+                    print(f"❌ Error starting server: {str(e)}")
+
+            # Run server in a background thread
+            server_thread = threading.Thread(target=run_server, daemon=True)
+            server_thread.start()
+
+            # Wait for a moment to allow the server to start and register endpoints
+            self._safe_sleep(0.5)
+        else:
+            # If server is already running, wait a moment to make sure the endpoint is registered
+            self._safe_sleep(0.1)
+            print(f"🔌 Available endpoints on port {port}: {', '.join(list(_registered_agents[port].keys()))}")
+
+        # Get the stack frame to check if this is the last launch() call in the script
+        import inspect
+        stack = inspect.stack()
+
+        # If this is called from a Python script (not interactive), try to detect if it's the last launch call
+        if len(stack) > 1 and stack[1].filename.endswith('.py'):
+            caller_frame = stack[1]
+            caller_line = caller_frame.lineno
+
+            try:
+                # Read the file to check if there are more launch calls after this one
+                with open(caller_frame.filename, 'r') as f:
+                    lines = f.readlines()
+
+                # Check if there are more launch() calls after the current line
+                has_more_launches = False
+                for line_content in lines[caller_line:]: # renamed line to line_content
+                    if '.launch(' in line_content and not line_content.strip().startswith('#'):
+                        has_more_launches = True
+                        break
+
+                # If this is the last launch call, block the main thread
+                if not has_more_launches:
                     try:
-                        print("\nKeeping HTTP servers alive. Press Ctrl+C to stop.")
+                        print("\nAll agents registered for HTTP mode. Press Ctrl+C to stop the servers.")
                         while True:
                             self._safe_sleep(1)
                     except KeyboardInterrupt:
                         print("\nServers stopped")
-            return None
+            except Exception as e:
+                # If something goes wrong with detection, block anyway to be safe
+                logging.error(f"Error in launch detection: {e}")
+                try:
+                    print("\nKeeping HTTP servers alive. Press Ctrl+C to stop.")
+                    while True:
+                        self._safe_sleep(1)
+                except KeyboardInterrupt:
+                    print("\nServers stopped")
+        return None
 
     def _launch_mcp_server(self, path: str, port: int, host: str, debug: bool):
         """
@@ -1671,7 +1783,7 @@ Write the complete compiled report:"""
             base_path = (path or "/mcp").rstrip("/") or "/mcp"
             transport = SseServerTransport(f"{base_path}/sse")
             starlette_app = Starlette(
-                routes=[Mount(base_path, mcp.create_app())]
+                routes=[Mount(base_path, mcp.sse_app())]
             )
 
             def run_mcp_server():

src/praisonai-agents/praisonaiagents/agent/handoff.py

@@ -14,9 +14,9 @@ from typing import Optional, Any, Callable, Dict, List, Union, TYPE_CHECKING, Li
 from dataclasses import dataclass, field
 from enum import Enum
 import inspect
-import logging
 from praisonaiagents._logging import get_logger
 import asyncio
+import contextvars
 import threading
 import time
 import json
@@ -163,33 +163,50 @@ from ..errors import (
     HandoffValidationError
 )
 
-# Thread-local storage for tracking handoff chains
-_handoff_context = threading.local()
+# Per-task/per-thread storage for tracking handoff chains.
+#
+# ``contextvars.ContextVar`` isolates the chain per :class:`asyncio.Task`
+# (and per OS thread), unlike ``threading.local()`` which shares one list
+# across every coroutine running on the same event-loop thread. Concurrent
+# async handoffs (``asyncio.gather``, a server handling parallel requests, or
+# any ``max_concurrent`` workflow) would otherwise push/pop into the same list
+# and corrupt each other's cycle/depth state.
+_handoff_chain_var: "contextvars.ContextVar[Optional[List[str]]]" = contextvars.ContextVar(
+    "handoff_chain", default=None
+)
 
 def _get_handoff_chain() -> List[str]:
-    """Get current handoff chain from thread-local storage."""
-    if not hasattr(_handoff_context, 'chain'):
-        _handoff_context.chain = []
-    return _handoff_context.chain
+    """Get current handoff chain for this task/thread."""
+    chain = _handoff_chain_var.get()
+    if chain is None:
+        chain = []
+        _handoff_chain_var.set(chain)
+    return chain
 
 def _get_handoff_depth() -> int:
     """Get current handoff depth."""
     return len(_get_handoff_chain())
 
 def _push_handoff(agent_name: str) -> None:
-    """Push agent to handoff chain."""
-    chain = _get_handoff_chain()
+    """Push agent to handoff chain.
+
+    Uses copy-on-write so a push made inside a child task does not leak back
+    into the parent task's chain once the child completes.
+    """
+    chain = list(_get_handoff_chain())
     chain.append(agent_name)
+    _handoff_chain_var.set(chain)
 
 def _pop_handoff() -> Optional[str]:
     """Pop agent from handoff chain."""
-    chain = _get_handoff_chain()
-    return chain.pop() if chain else None
+    chain = list(_get_handoff_chain())
+    popped = chain.pop() if chain else None
+    _handoff_chain_var.set(chain)
+    return popped
 
 def _clear_handoff_chain() -> None:
     """Clear the handoff chain."""
-    if hasattr(_handoff_context, 'chain'):
-        _handoff_context.chain = []
+    _handoff_chain_var.set([])
 
 @dataclass
 class HandoffInputData:

src/praisonai-agents/praisonaiagents/agent/heartbeat.py

@@ -13,7 +13,6 @@ Usage::
 """
 
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 import threading
 import time

src/praisonai-agents/praisonaiagents/agent/loop_detection.py

@@ -22,7 +22,6 @@ from __future__ import annotations
 
 import hashlib
 import json
-import logging
 from praisonaiagents._logging import get_logger
 import time
 from dataclasses import dataclass, field

src/praisonai-agents/praisonaiagents/agent/message_queue.py

@@ -8,7 +8,6 @@ Thread-safe implementation for concurrent access.
 import threading
 import heapq
 import time
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Optional, Any, List, Tuple
 from dataclasses import dataclass, field

src/praisonai-agents/praisonaiagents/agent/prompt_cache.py

@@ -0,0 +1,138 @@
+"""Prompt-cache-stability contract for the Agent turn (Issue #3352).
+
+Provider prompt caching (Anthropic/OpenAI) only hits when the *leading* bytes
+of a request are byte-identical to the previous request. For a long-lived
+conversation the cached prefix — the system instructions, the serialised tool
+schemas, and the model identity — is the dominant cost and latency lever.
+
+Gateways run many users on a single shared ``Agent`` instance and swap
+``tools``/``llm`` in and out per turn (per-route tool scoping, per-user
+``/model`` override). Those swaps silently change the cached prefix and the
+provider cache misses with zero operator visibility.
+
+``prompt_prefix_signature`` gives callers a single, cheap, deterministic
+fingerprint of exactly the cache-relevant inputs — model, the sorted set of
+tool names, and a fingerprint of the system instructions — and *nothing*
+volatile per-turn (arrival time, routing facts, memory/RAG recall). When the
+signature is unchanged across turns the prefix is guaranteed stable, so the
+provider cache keeps hitting; when it changes, that is the one auditable point
+at which cache warmth is knowingly sacrificed.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Any
+
+
+def _tool_name(tool: Any) -> str:
+    """Best-effort stable name for a tool entry.
+
+    Tools may be plain callables, dicts holding an OpenAI-style function
+    schema, or objects exposing ``__name__``/``name``. The name identifies the
+    schema slot in the serialised tool block.
+    """
+    if isinstance(tool, dict):
+        fn = tool.get("function")
+        if isinstance(fn, dict) and fn.get("name"):
+            return str(fn["name"])
+        for key in ("name", "type"):
+            if tool.get(key):
+                return str(tool[key])
+        return str(sorted(tool.keys()))
+    for attr in ("__name__", "name"):
+        value = getattr(tool, attr, None)
+        if value:
+            return str(value)
+    return str(tool)
+
+
+def _tool_fingerprint(tool: Any) -> str:
+    """Best-effort stable fingerprint of a tool's *provider-visible* schema.
+
+    The serialised tool block sent to the provider is not just the tool name —
+    it is the full function schema (name + description + parameters). A change
+    to a tool's description, parameters, or required fields *without* a rename
+    changes those bytes and therefore invalidates the prompt cache, so the
+    fingerprint must reflect the whole schema, not only the name (Issue #3352
+    review). Falls back to the name when no richer schema is exposed.
+    """
+    name = _tool_name(tool)
+
+    # Dict tools already carry an OpenAI-style function schema — fingerprint the
+    # description + parameters alongside the name so a same-name schema edit is
+    # detected. ``json.dumps(sort_keys=True)`` makes the render deterministic.
+    if isinstance(tool, dict):
+        fn = tool.get("function")
+        schema = fn if isinstance(fn, dict) else tool
+        desc = schema.get("description") if isinstance(schema, dict) else None
+        params = schema.get("parameters") if isinstance(schema, dict) else None
+        try:
+            import json
+
+            return "\x01".join(
+                (
+                    name,
+                    str(desc or ""),
+                    json.dumps(params or {}, sort_keys=True, default=str),
+                )
+            )
+        except Exception:  # pragma: no cover — defensive
+            return name
+
+    # Callable/object tools: fold in the docstring (the description the schema
+    # generator emits) so an edited tool doc flips the signature too.
+    doc = getattr(tool, "__doc__", None)
+    if doc:
+        return f"{name}\x01{doc}"
+    return name
+
+
+def prompt_prefix_signature(agent: Any) -> str:
+    """Return a sha256 signature of an agent's cache-relevant prompt prefix.
+
+    The signature is computed over, and only over:
+
+    * the model identity (``agent.llm``),
+    * the sorted set of tool *schema fingerprints* (name + description +
+      parameters, ``agent.tools``) — so a same-name schema edit is detected,
+    * a fingerprint of the system instructions (``agent.instructions``).
+
+    Volatile per-turn data is intentionally excluded so that an unchanged
+    route/model keeps a byte-identical prefix and the provider cache keeps
+    hitting. Callers compare the signature turn-over-turn: an unchanged value
+    means the cached prefix is reused; a changed value is the single auditable
+    point of prompt-cache invalidation.
+
+    Best-effort and never raises: any attribute access failure degrades the
+    corresponding component to an empty string rather than breaking the turn.
+    """
+    try:
+        llm = getattr(agent, "llm", None)
+        model = llm if isinstance(llm, str) else getattr(llm, "model", None) or str(llm)
+    except Exception:
+        model = ""
+
+    try:
+        tools = getattr(agent, "tools", None) or []
+        tool_fps = sorted(_tool_fingerprint(t) for t in tools)
+    except Exception:
+        tool_fps = []
+
+    try:
+        instructions = getattr(agent, "instructions", None) or ""
+        instructions = str(instructions)
+    except Exception:
+        instructions = ""
+
+    hasher = hashlib.sha256()
+    hasher.update(b"model\x00")
+    hasher.update(str(model).encode("utf-8", "replace"))
+    hasher.update(b"\x00tools\x00")
+    hasher.update("\x00".join(tool_fps).encode("utf-8", "replace"))
+    hasher.update(b"\x00instructions\x00")
+    hasher.update(instructions.encode("utf-8", "replace"))
+    return hasher.hexdigest()
+
+
+__all__ = ["prompt_prefix_signature"]

src/praisonai-agents/praisonaiagents/agent/router_agent.py

@@ -6,7 +6,6 @@ model selection based on task characteristics.
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Dict, List, Optional, Any, Union
 from .agent import Agent

src/praisonai-agents/praisonaiagents/agent/run_outcome.py

@@ -0,0 +1,95 @@
+"""Canonical terminal-outcome contract for agent runs.
+
+A single, closed description of *how* a run ended, produced once by the core
+run and consumed everywhere. Long-running hosts (e.g. gateways/bots) can make
+delivery/retry/DLQ/status decisions from one field instead of inferring the
+terminal state from exception identity.
+
+Precedence is explicit and sticky: a ``hard_timeout`` is never silently
+downgraded by later cleanup. See ``RunOutcome.from_exception``.
+"""
+
+from dataclasses import dataclass
+from typing import Optional
+try:
+    from typing import Literal
+except ImportError:  # pragma: no cover - py<3.8 fallback
+    Literal = None  # type: ignore
+
+if Literal is not None:
+    TerminalReason = Literal[
+        "completed", "hard_timeout", "cancelled", "aborted", "failed"
+    ]
+else:  # pragma: no cover
+    TerminalReason = str  # type: ignore
+
+# Precedence: higher wins and is sticky (a hard timeout is not downgraded).
+_REASON_PRECEDENCE = {
+    "completed": 0,
+    "failed": 1,
+    "aborted": 2,
+    "cancelled": 3,
+    "hard_timeout": 4,
+}
+
+
+@dataclass(frozen=True)
+class RunOutcome:
+    """Closed, canonical description of how an agent run ended.
+
+    Attributes:
+        reason: One of ``completed | hard_timeout | cancelled | aborted | failed``.
+        output: Partial or final text, if any.
+        error: Redacted message when ``reason == "failed"``.
+    """
+
+    reason: "TerminalReason"
+    output: Optional[str] = None
+    error: Optional[str] = None
+
+    @property
+    def succeeded(self) -> bool:
+        """True only when the run completed normally."""
+        return self.reason == "completed"
+
+    @classmethod
+    def completed(cls, output: Optional[str] = None) -> "RunOutcome":
+        return cls(reason="completed", output=output)
+
+    @classmethod
+    def from_exception(
+        cls, exc: BaseException, output: Optional[str] = None
+    ) -> "RunOutcome":
+        """Normalise a raised exception into a canonical terminal outcome.
+
+        Maps well-known exception types to their terminal reason. Anything
+        unrecognised is treated as ``failed`` with a redacted message.
+
+        Note on ``hard_timeout``: the run-level budget is the authoritative
+        source of ``hard_timeout`` and is applied explicitly by the caller
+        (see ``_astart_with_outcome``). A bare ``asyncio.TimeoutError`` reaching
+        this normaliser is a *nested* operation timeout (e.g. a handoff/tool that
+        exhausted its own budget while the run budget remained) and is therefore
+        classified as ``failed`` — not silently promoted to a run-budget
+        ``hard_timeout``, which would drive the wrong host retry/DLQ decision.
+        Real cooperative cancellation is intercepted and re-raised by the run
+        wrapper (see ``_astart_with_outcome``) *before* it reaches this
+        normaliser, so host shutdown is honoured. A supersede/interrupt named
+        error that does surface here is a domain "cancelled" outcome.
+        """
+        import asyncio
+
+        if _name_matches(exc, ("hardtimeout", "runtimeout", "budgettimeout")):
+            return cls(reason="hard_timeout", output=output)
+        if isinstance(exc, asyncio.CancelledError) or _name_matches(
+            exc, ("supersed", "interrupt", "cancelled")
+        ):
+            return cls(reason="cancelled", output=output)
+        if _name_matches(exc, ("abort", "drain", "shutdown")):
+            return cls(reason="aborted", output=output)
+        return cls(reason="failed", output=output, error=str(exc))
+
+
+def _name_matches(exc: BaseException, needles: tuple) -> bool:
+    name = type(exc).__name__.lower()
+    return any(n in name for n in needles)

src/praisonai-agents/praisonaiagents/agent/summarization.py

@@ -5,7 +5,6 @@ Provides automatic summarization when context window fills up.
 Reuses existing telemetry/token_collector for token tracking.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Optional, List, Dict, Any
 

src/praisonai-agents/praisonaiagents/agent/tool_execution.py

@@ -471,6 +471,46 @@ class ToolExecutionMixin:
             self._middleware_manager = manager
         return manager if manager.has_tool_hooks else None
 
+    def _get_turn_tools_lock(self):
+        """Return the DualLock guarding the per-turn tool buffer.
+
+        Falls back to a lazily-created lock so subclasses or objects that
+        predate the lock attribute stay safe. The lock protects
+        ``_turn_tools_used`` from corruption when concurrent chat()/achat()
+        turns run on the same Agent instance (issue #3307).
+        """
+        lock = getattr(self, "_turn_tools_lock", None)
+        if lock is None:
+            from .async_safety import DualLock
+            lock = DualLock()
+            self._turn_tools_lock = lock
+        return lock
+
+    def _reset_turn_tools(self):
+        """Reset the per-turn tool buffer under lock (start of a chat turn)."""
+        if getattr(self, "_in_skill_review", False):
+            return
+        with self._get_turn_tools_lock().sync():
+            self._turn_tools_used = []
+
+    def _record_turn_tool(self, function_name):
+        """Append a tool name to the per-turn buffer under lock."""
+        if getattr(self, "_in_skill_review", False):
+            return
+        with self._get_turn_tools_lock().sync():
+            turn_tools = getattr(self, "_turn_tools_used", None)
+            if turn_tools is None:
+                turn_tools = []
+                self._turn_tools_used = turn_tools
+            turn_tools.append(function_name)
+
+    def _drain_turn_tools(self):
+        """Return a copy of the per-turn buffer and clear it, under lock."""
+        with self._get_turn_tools_lock().sync():
+            tools_used = list(getattr(self, "_turn_tools_used", []) or [])
+            self._turn_tools_used = []
+        return tools_used
+
     def _execute_tool_with_context(self, function_name, arguments, state, tool_call_id=None):
         """Execute tool within injection context, with optional output truncation.
         
@@ -487,13 +527,9 @@ class ToolExecutionMixin:
         
         # Record the tool name for this turn so the self-improve review policy
         # can see what ran (issue #3037). Skipped during a guarded review turn
-        # to avoid tracking the review's own tool calls or recursing.
-        if not getattr(self, "_in_skill_review", False):
-            turn_tools = getattr(self, "_turn_tools_used", None)
-            if turn_tools is None:
-                self._turn_tools_used = []
-                turn_tools = self._turn_tools_used
-            turn_tools.append(function_name)
+        # to avoid tracking the review's own tool calls or recursing. Locked so
+        # concurrent turns on the same Agent don't corrupt the buffer (#3307).
+        self._record_turn_tool(function_name)
 
         # Emit tool call start event (zero overhead when not set)
         _trace_emitter = get_context_emitter()
@@ -1132,8 +1168,9 @@ class ToolExecutionMixin:
         # tools_used, so without this the review policy always sees an empty
         # list and never runs. Consume the buffer so the next turn starts clean.
         if tools_used is None:
-            tools_used = list(getattr(self, "_turn_tools_used", []) or [])
-        self._turn_tools_used = []
+            tools_used = self._drain_turn_tools()
+        else:
+            self._drain_turn_tools()
         # Trigger AFTER_AGENT hook (only build the input if a hook is actually registered)
         from ..hooks import HookEvent
         if self._hook_runner.registry.has_hooks(HookEvent.AFTER_AGENT):
@@ -1189,8 +1226,9 @@ class ToolExecutionMixin:
         # Default tools_used from the per-turn buffer when the caller did not
         # pass it explicitly (issue #3037); mirrors the sync path.
         if tools_used is None:
-            tools_used = list(getattr(self, "_turn_tools_used", []) or [])
-        self._turn_tools_used = []
+            tools_used = self._drain_turn_tools()
+        else:
+            self._drain_turn_tools()
         # Trigger AFTER_AGENT hook (only build the input if a hook is actually registered)
         from ..hooks import HookEvent
         if self._hook_runner.registry.has_hooks(HookEvent.AFTER_AGENT):
@@ -1395,7 +1433,9 @@ class ToolExecutionMixin:
             # only gate the built-in DEFAULT_DANGEROUS_TOOLS and silently skip
             # registry-required tools.
             approval_registry = get_approval_registry()
-            registry_required = approval_registry.is_required(tool_name)
+            registry_required = approval_registry.is_required(
+                tool_name, getattr(self, 'name', None)
+            )
             # Check if tool needs approval based on multiple criteria
             needs_approval = (
                 approve_all 
@@ -1409,7 +1449,7 @@ class ToolExecutionMixin:
                     tool_name=tool_name,
                     arguments=tool_args,
                     risk_level=(
-                        approval_registry.get_risk_level(tool_name)
+                        approval_registry.get_risk_level(tool_name, getattr(self, 'name', None))
                         or DEFAULT_DANGEROUS_TOOLS.get(tool_name, "medium")
                     ),
                     agent_name=getattr(self, 'name', None),
@@ -1474,21 +1514,19 @@ class ToolExecutionMixin:
                     return ApprovalDecision(approved=True, reason="Not a dangerous tool")
         else:
             # No approval backend configured. An explicit PermissionManager
-            # ``ask`` rule must still gate the call, so mark it required in the
-            # approval registry (idempotent) before delegating so the registry
-            # prompts instead of silently allowing.
-            if manager_forces_approval:
-                try:
-                    get_approval_registry().add_requirement(tool_name)
-                except Exception:  # noqa: BLE001
-                    pass
+            # ``ask`` rule must still gate the call. Rather than mutating shared
+            # registry state (which leaked onto other agents when this agent had
+            # no stable name), pass the intent per-call via ``force`` so the
+            # registry prompts for *this* call only without side effects.
             if is_async:
                 return get_approval_registry().approve_async(
                     getattr(self, 'name', None), tool_name, tool_args,
+                    force=manager_forces_approval,
                 )
             else:
                 return get_approval_registry().approve_sync(
                     getattr(self, 'name', None), tool_name, tool_args,
+                    force=manager_forces_approval,
                 )
 
     def _permission_manager_requires_approval(self, function_name) -> bool:
@@ -1602,7 +1640,9 @@ class ToolExecutionMixin:
             if self._permission_manager_requires_approval(function_name):
                 return True
             from ..approval import get_approval_registry
-            if get_approval_registry().is_required(function_name):
+            if get_approval_registry().is_required(
+                function_name, getattr(self, 'name', None)
+            ):
                 return True
             from ..tools import get_registry as get_tool_registry
             if get_tool_registry().get_trust_level(function_name) == "external":
@@ -1839,6 +1879,35 @@ class ToolExecutionMixin:
             raise last_exception
         return {"error": "Maximum retry attempts exceeded"}
 
+    @staticmethod
+    def _remove_circuit_breaker(breaker_name):
+        """Remove a circuit breaker entry from the process-global registry."""
+        try:
+            from ..tools.circuit_breaker import _get_global_registry
+        except Exception:
+            return
+        try:
+            _get_global_registry().remove(breaker_name)
+        except Exception:
+            pass
+
+    def _register_breaker_finalizer(self, breaker_name):
+        """Register a weakref finalizer so this agent's breaker entry is removed
+        the moment the agent is garbage-collected, even if close() is never called.
+        """
+        registered = self.__dict__.setdefault('_breaker_finalizer_names', set())
+        if breaker_name in registered:
+            return
+        registered.add(breaker_name)
+        try:
+            import weakref
+            finalizers = self.__dict__.setdefault('_breaker_finalizers', [])
+            finalizers.append(
+                weakref.finalize(self, self._remove_circuit_breaker, breaker_name)
+            )
+        except Exception:
+            pass
+
     def _execute_tool_with_circuit_breaker_impl(self, function_name, arguments):
         """Execute tool with circuit breaker protection (internal implementation).
         
@@ -1870,7 +1939,13 @@ class ToolExecutionMixin:
                 graceful_degradation=True   # Return error instead of raising exception
             )
             breaker = get_circuit_breaker(breaker_name, config)
-            
+
+            # Ensure the registry entry is removed the moment this Agent is
+            # actually garbage-collected, regardless of whether close()/aclose()
+            # was ever called. This closes the CPython id-reuse window where a
+            # new Agent at the same address could inherit a stale OPEN breaker.
+            self._register_breaker_finalizer(breaker_name)
+
             # Execute tool through circuit breaker with failure detection wrapper
             def _tool_wrapper():
                 result = self._execute_tool_impl(function_name, arguments)
@@ -2067,8 +2142,11 @@ class ToolExecutionMixin:
                 func = tool
                 break
         
-        if func is None:
-            # Check the global tool registry for plugins
+        if func is None and getattr(self, '_allow_global_tools', False):
+            # Check the process-global tool registry for plugins. Gated behind
+            # ToolConfig(allow_global_tools=True) so an agent is scoped strictly
+            # to its declared tools=[...] by default (safe by default) and cannot
+            # execute an @tool-decorated callable it was never given.
             try:
                 from ..tools.registry import get_registry
                 registry = get_registry()
@@ -2079,46 +2157,220 @@ class ToolExecutionMixin:
         if func is None:
             # Tool not found in declared tools or registry — do not fall back to
             # globals() or __main__ as that allows undeclared callables to execute.
-            pass
+            # Cheap, deterministic self-repair: the model often emits a name that
+            # only differs by case/separator (e.g. 'WebSearch' -> 'web_search').
+            # Build a normalised index of the agent's active tools and re-match.
+            def _norm(n):
+                return str(n).lower().replace('_', '').replace('-', '').replace(' ', '')
+
+            normalised = {}
+            for name, tool in self._iter_active_named_tools():
+                normalised.setdefault(_norm(name), []).append((name, tool))
+            match = normalised.get(_norm(function_name))
+            if match and len(match) == 1:
+                matched_name, func = match[0]
+                logging.debug(
+                    f"Self-repaired tool name {function_name!r} -> {matched_name!r}"
+                )
 
         if func:
+            bind_target = func
+            bind_arguments = arguments
             try:
                 # BaseTool instances (plugin system) - call run() method
                 from ..tools.base import BaseTool
                 if isinstance(func, BaseTool):
+                    bind_target = func.run
                     casted_arguments = self._cast_arguments(func.run, arguments)
+                    bind_arguments = casted_arguments
                     return func.run(**casted_arguments)
                 
                 # Langchain: If it's a class with run but not _run, instantiate and call run
                 if inspect.isclass(func) and hasattr(func, 'run') and not hasattr(func, '_run'):
                     instance = func()
+                    bind_target = instance.run
                     run_params = {k: v for k, v in arguments.items() 
                                   if k in inspect.signature(instance.run).parameters 
                                   and k != 'self'}
                     casted_params = self._cast_arguments(instance.run, run_params)
+                    bind_arguments = casted_params
                     return instance.run(**casted_params)
 
                 # CrewAI: If it's a class with an _run method, instantiate and call _run
                 elif inspect.isclass(func) and hasattr(func, '_run'):
                     instance = func()
+                    bind_target = instance._run
                     run_params = {k: v for k, v in arguments.items() 
                                   if k in inspect.signature(instance._run).parameters 
                                   and k != 'self'}
                     casted_params = self._cast_arguments(instance._run, run_params)
+                    bind_arguments = casted_params
                     return instance._run(**casted_params)
 
                 # Otherwise treat as regular function
                 elif callable(func):
+                    bind_target = func
                     casted_arguments = self._cast_arguments(func, arguments)
+                    bind_arguments = casted_arguments
                     return func(**casted_arguments)
             except Exception as e:
                 error_msg = str(e)
                 logging.error(f"Error executing tool {function_name}: {error_msg}")
+                # Only echo the parameter schema when the failure is a genuine
+                # argument-binding error (wrong/missing/extra kwargs). A
+                # TypeError/ValueError raised *inside* a successfully-bound tool
+                # (domain validation) must not be mislabelled as a parameter
+                # problem, or the model would alter valid call arguments instead
+                # of fixing the offending value.
+                if self._is_argument_binding_error(bind_target, bind_arguments):
+                    schema = self._tool_parameter_hint(func)
+                    if schema:
+                        # Fold the parameter names into the error string itself so
+                        # the hint survives conversion to ToolExecutionError (which
+                        # keeps only the message) and actually reaches the model.
+                        error_msg = (
+                            f"{error_msg} Expected parameters for '{function_name}' — "
+                            f"required: {schema['required']}, optional: {schema['optional']}."
+                        )
+                        return {"error": error_msg, "expected_parameters": schema}
                 return {"error": error_msg}
-        
-        error_msg = f"Tool '{function_name}' is not callable"
+
+        # Unresolved: return a corrective, model-readable message so the model can
+        # retry with a valid name instead of repeating the same mistake.
+        available = self._available_active_tool_names()
+        suggestion = None
+        try:
+            import difflib
+            near = difflib.get_close_matches(function_name, available, n=1, cutoff=0.5)
+            suggestion = near[0] if near else None
+        except Exception:
+            pass
+        hint = f" Did you mean '{suggestion}'?" if suggestion else ""
+        error_msg = (
+            f"Tool '{function_name}' not found.{hint} "
+            f"Available tools: {available}"
+        )
         logging.error(error_msg)
-        return {"error": error_msg}
+        return {"error": error_msg, "available_tools": available}
+
+    def _get_tool_display_name(self, tool):
+        """Best-effort display name for an agent tool of any supported kind."""
+        try:
+            from ..tools.base import BaseTool
+            if isinstance(tool, BaseTool):
+                return getattr(tool, 'name', None)
+        except ImportError:
+            pass
+        name = getattr(tool, 'name', None)
+        if isinstance(name, str) and name:
+            return name
+        if callable(tool) or inspect.isclass(tool):
+            return getattr(tool, '__name__', None)
+        return None
+
+    def _iter_active_named_tools(self):
+        """Yield ``(name, tool)`` for every active tool.
+
+        MCP instances are expanded into their contained tools (each MCP tool is
+        an iterable callable with a ``__name__``/``name``) so they participate
+        in name repair and appear in the corrective inventory, rather than only
+        exposing the opaque container.
+        """
+        MCP = None
+        try:
+            from ..mcp.mcp import MCP
+        except ImportError:
+            pass
+
+        def _expand(tool):
+            if MCP is not None and isinstance(tool, MCP):
+                try:
+                    for sub in tool:
+                        sub_name = self._get_tool_display_name(sub)
+                        if sub_name:
+                            yield sub_name, sub
+                except Exception:
+                    pass
+                return
+            name = self._get_tool_display_name(tool)
+            if name:
+                yield name, tool
+
+        tools = self.tools
+        if MCP is not None and isinstance(tools, MCP):
+            yield from _expand(tools)
+            return
+        for tool in tools if isinstance(tools, (list, tuple)) else []:
+            yield from _expand(tool)
+
+    def _available_active_tool_names(self):
+        """Names of the agent's currently active tools, for corrective feedback."""
+        names = [name for name, _tool in self._iter_active_named_tools()]
+        return sorted(set(names))
+
+    def _resolve_callable_signature_target(self, func):
+        """Return the callable whose signature describes ``func``'s arguments."""
+        target = func
+        try:
+            from ..tools.base import BaseTool
+            if isinstance(func, BaseTool):
+                return func.run
+        except ImportError:
+            pass
+        if inspect.isclass(func):
+            run = getattr(func, 'run', None) or getattr(func, '_run', None)
+            if run is not None:
+                target = run
+        return target
+
+    def _is_argument_binding_error(self, func, arguments) -> bool:
+        """True only when ``arguments`` cannot bind to ``func``'s signature.
+
+        Distinguishes a genuine call-boundary failure (wrong/missing/extra
+        kwargs) from a ``TypeError``/``ValueError`` raised *inside* a
+        successfully-bound tool during its own domain logic. Only the former
+        should receive a parameter-schema hint; the latter is a runtime error
+        the model must fix by changing the value, not the parameter names.
+        """
+        target = self._resolve_callable_signature_target(func)
+        try:
+            sig = inspect.signature(target)
+        except (TypeError, ValueError):
+            return False
+        try:
+            sig.bind(**(arguments or {}))
+        except TypeError:
+            return True
+        return False
+
+    def _tool_parameter_hint(self, func):
+        """Return {'required': [...], 'optional': [...]} for a callable tool."""
+        target = func
+        try:
+            from ..tools.base import BaseTool
+            if isinstance(func, BaseTool):
+                target = func.run
+            elif inspect.isclass(func):
+                target = getattr(func, 'run', None) or getattr(func, '_run', None)
+            if target is None or not callable(target):
+                return None
+            sig = inspect.signature(target)
+            required, optional = [], []
+            for pname, param in sig.parameters.items():
+                if pname == 'self' or param.kind in (
+                    inspect.Parameter.VAR_POSITIONAL,
+                    inspect.Parameter.VAR_KEYWORD,
+                ):
+                    continue
+                if param.default is inspect.Parameter.empty:
+                    required.append(pname)
+                else:
+                    optional.append(pname)
+            if not required and not optional:
+                return None
+            return {"required": required, "optional": optional}
+        except (ValueError, TypeError):
+            return None
 
     async def submit_for_approval(self, function_name: str, arguments: Dict[str, Any]) -> str:
         """Fire an approval request in the background without blocking.

src/praisonai-agents/praisonaiagents/agents/agents.py

@@ -330,17 +330,26 @@ async def _execute_with_agent_async(executor_agent, task_prompt, task, tools, st
         )
 
 
-def _build_execution_context(agents_instance, task_id):
+def _build_execution_context(agents_instance, task_id, skip_memory_init=False):
     """
     Build unified execution context for task execution (DRY helper).
     Eliminates duplication between sync/async execution paths.
+
+    Args:
+        skip_memory_init: When True, skip the synchronous ``initialize_memory()``
+            call. The async path (``aexecute_task``) already attempts
+            ``initialize_memory_async()`` beforehand, so this avoids blocking the
+            event loop on a synchronous ``Memory()`` construction (and a
+            duplicate backend attempt) if that async init failed.
     """
     from .protocols import ExecutionContext
     
     task = agents_instance.tasks[task_id]
     
-    # Initialize memory before task execution
-    if not task.memory:
+    # Initialize memory before task execution. The async path (aexecute_task)
+    # already attempts initialize_memory_async() and passes skip_memory_init=True
+    # so we never block the event loop on a synchronous Memory() construction.
+    if not task.memory and not skip_memory_init:
         task.memory = task.initialize_memory()
 
     executor_agent = task.agent
@@ -842,6 +851,11 @@ class AgentTeam(SpawnAnnounceProtocol):
         self._event_bus: Optional[EventBus] = None
         self._spawn_lock = threading.RLock()  # Thread-safe spawn operations (reentrant)
         self._team_id = str(uuid.uuid4())  # Unique team identifier
+        # Aggregate stream emitter (lazy). Fans in member agents' per-step
+        # StreamEventEmitter events, tagging each with the emitting agent's id,
+        # so a single consumer can attribute activity to a specific team member.
+        self.__stream_emitter = None
+        self.__stream_fanin_wired = False
         
         # Check for manager_llm in environment variable if not provided
         self.manager_llm = manager_llm or os.getenv('OPENAI_MODEL_NAME', 'gpt-4o-mini')
@@ -960,6 +974,63 @@ class AgentTeam(SpawnAnnounceProtocol):
         except (ImportError, AttributeError):
             self._telemetry = None
 
+    @property
+    def stream_emitter(self):
+        """Aggregate ``StreamEventEmitter`` fanning in member agents' events.
+
+        Lazily created on first access. On creation it registers a forwarding
+        callback on each member agent's own ``stream_emitter`` so per-step
+        events (tool calls, text/reasoning deltas, retries, errors) surface on a
+        single team-level emitter, each tagged with the emitting agent's
+        ``agent_id``. This gives a single attach point (e.g. the CLI
+        ``--output stream-json`` bridge) parity with single-agent runs.
+
+        Zero-overhead when unused: nothing is wired unless this property is
+        accessed, and forwarding only tags ``agent_id`` when the source event
+        did not already carry one.
+        """
+        if self.__stream_emitter is None:
+            try:
+                from ..streaming.events import StreamEventEmitter
+            except ImportError:
+                return None
+            self.__stream_emitter = StreamEventEmitter()
+        if not self.__stream_fanin_wired:
+            self._wire_stream_fanin()
+        return self.__stream_emitter
+
+    def _wire_stream_fanin(self):
+        """Forward each member agent's stream events onto the team emitter.
+
+        Best-effort and idempotent: an agent whose emitter is unavailable is
+        skipped, and a forwarding callback is attached at most once per team.
+        """
+        team_emitter = self.__stream_emitter
+        if team_emitter is None:
+            return
+        self.__stream_fanin_wired = True
+        for agent in self.agents:
+            member_emitter = getattr(agent, "stream_emitter", None)
+            if member_emitter is None or not hasattr(member_emitter, "add_callback"):
+                continue
+            agent_id = getattr(agent, "agent_id", None) or getattr(agent, "display_name", None)
+            member_emitter.add_callback(self._make_fanin_callback(team_emitter, agent_id))
+
+    @staticmethod
+    def _make_fanin_callback(team_emitter, agent_id):
+        """Build a callback that re-emits an event on the team emitter."""
+        def _forward(event):
+            try:
+                if agent_id is not None and getattr(event, "agent_id", None) is None:
+                    try:
+                        event.agent_id = agent_id
+                    except (AttributeError, TypeError):
+                        pass
+                team_emitter.emit(event)
+            except Exception:
+                logger.debug("AgentTeam stream fan-in forward failed", exc_info=True)
+        return _forward
+
     def add_task(self, task):
         with self._task_id_lock:
             task_id = self.task_id_counter
@@ -1054,8 +1125,17 @@ class AgentTeam(SpawnAnnounceProtocol):
         if task.status == "not started":
             task.status = "in progress"
 
-        # Build execution context using DRY helper
-        context = _build_execution_context(self, task_id)
+        # Initialize memory asynchronously to avoid blocking the event loop on
+        # synchronous Memory() construction. The shared helper's own
+        # `if not task.memory:` guard makes this a safe no-op for the sync path.
+        if not task.memory:
+            await task.initialize_memory_async()
+
+        # Build execution context using DRY helper. skip_memory_init=True prevents
+        # the helper from falling back to the *synchronous* initialize_memory()
+        # (which would block the event loop and duplicate a failed backend attempt)
+        # when the async init above did not populate task.memory.
+        context = _build_execution_context(self, task_id, skip_memory_init=True)
 
         # Execute with agent using DRY helper
         agent_output = await _execute_with_agent_async(
@@ -1448,7 +1528,13 @@ class AgentTeam(SpawnAnnounceProtocol):
                     except Exception as e:
                         logger.error(f"Error executing memory callback for task {task_id}: {e}")
                         logger.exception(e)
-                            
+                        # Respect task failure policies - re-raise if configured
+                        # (mirrors arun_task so sync/async surfaces behave identically)
+                        if hasattr(task, 'fail_on_callback_error') and task.fail_on_callback_error:
+                            raise
+                        if hasattr(task, 'fail_on_memory_error') and task.fail_on_memory_error:
+                            raise
+
                     self.save_output_to_file(task, task_output)
 
                     # Call on_task_complete callback (shared with arun_task)

src/praisonai-agents/praisonaiagents/agents/auto_rag_agent.py

@@ -17,7 +17,6 @@ Usage:
     result = auto_rag.chat("Hello!")  # Skips retrieval
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 import time
 from dataclasses import dataclass, field

src/praisonai-agents/praisonaiagents/agents/delegator.py

@@ -5,7 +5,6 @@ Provides delegation primitives for spawning subagents with scoped permissions
 and context. Inspired by Gemini CLI's delegate-to-agent pattern.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 import asyncio
 import warnings

src/praisonai-agents/praisonaiagents/approval/backends.py

@@ -11,7 +11,6 @@ Provides lightweight backends that ship with the core SDK:
 from __future__ import annotations
 
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Any
 

src/praisonai-agents/praisonaiagents/approval/registry.py

@@ -20,7 +20,6 @@ import asyncio
 import contextvars
 import hashlib
 import json
-import logging
 from praisonaiagents._logging import get_logger
 import os
 from typing import Dict, List, Optional, Set
@@ -91,10 +90,19 @@ class ApprovalRegistry:
         self._global_backend = None  # type: ignore[assignment]
         self._agent_backends: Dict[str, object] = {}
 
-        # Tool requirements (mirrors old APPROVAL_REQUIRED_TOOLS / TOOL_RISK_LEVELS)
+        # Tool requirements (mirrors old APPROVAL_REQUIRED_TOOLS / TOOL_RISK_LEVELS).
+        # These hold process-wide defaults (e.g. DEFAULT_DANGEROUS_TOOLS and any
+        # intentional global registration).
         self._required_tools: Set[str] = set()
         self._risk_levels: Dict[str, str] = {}
 
+        # Per-agent tool requirements. A PermissionManager ``ask`` rule belongs
+        # to a single agent, so it must not leak an approval gate onto unrelated
+        # agents sharing the same process. Mirrors the agent-keyed pattern used
+        # by ``_agent_backends`` / ``_agent_tool_auto_approve``.
+        self._agent_required_tools: Dict[str, Set[str]] = {}
+        self._agent_risk_levels: Dict[tuple[str, str], str] = {}
+
         # Per-agent, per-tool auto-approval (G-A fix)
         self._agent_tool_auto_approve: Dict[tuple[str, str], bool] = {}
 
@@ -153,18 +161,51 @@ class ApprovalRegistry:
 
     # ── Tool requirement management ──────────────────────────────────────
 
-    def add_requirement(self, tool_name: str, risk_level: str = "high") -> None:
-        self._required_tools.add(tool_name)
-        self._risk_levels[tool_name] = risk_level
+    def add_requirement(
+        self,
+        tool_name: str,
+        risk_level: str = "high",
+        agent_name: Optional[str] = None,
+    ) -> None:
+        """Mark *tool_name* as requiring approval.
 
-    def remove_requirement(self, tool_name: str) -> None:
-        self._required_tools.discard(tool_name)
-        self._risk_levels.pop(tool_name, None)
+        When *agent_name* is given the requirement is scoped to that agent only
+        (used for per-agent ``PermissionManager`` ``ask`` rules), so it never
+        forces approval onto other agents in the same process. Omitting
+        *agent_name* keeps the historical process-wide behaviour used for
+        genuinely dangerous tools registered at startup.
+        """
+        if agent_name:
+            self._agent_required_tools.setdefault(agent_name, set()).add(tool_name)
+            self._agent_risk_levels[(agent_name, tool_name)] = risk_level
+        else:
+            self._required_tools.add(tool_name)
+            self._risk_levels[tool_name] = risk_level
 
-    def is_required(self, tool_name: str) -> bool:
+    def remove_requirement(
+        self, tool_name: str, agent_name: Optional[str] = None
+    ) -> None:
+        if agent_name:
+            tools = self._agent_required_tools.get(agent_name)
+            if tools is not None:
+                tools.discard(tool_name)
+            self._agent_risk_levels.pop((agent_name, tool_name), None)
+        else:
+            self._required_tools.discard(tool_name)
+            self._risk_levels.pop(tool_name, None)
+
+    def is_required(self, tool_name: str, agent_name: Optional[str] = None) -> bool:
+        if agent_name and tool_name in self._agent_required_tools.get(agent_name, ()):
+            return True
         return tool_name in self._required_tools
 
-    def get_risk_level(self, tool_name: str) -> Optional[str]:
+    def get_risk_level(
+        self, tool_name: str, agent_name: Optional[str] = None
+    ) -> Optional[str]:
+        if agent_name:
+            level = self._agent_risk_levels.get((agent_name, tool_name))
+            if level is not None:
+                return level
         return self._risk_levels.get(tool_name)
 
     # ── Per-tool auto-approval (G-A fix) ─────────────────────────────────
@@ -195,9 +236,11 @@ class ApprovalRegistry:
         self._approved_context.set(approved)
 
     def is_already_approved(self, tool_name: str, arguments: Optional[Dict] = None) -> bool:
-        if self.get_risk_level(tool_name) == "critical":
-            return False
-        return self._approval_cache_key(tool_name, arguments or {}) in self._approved_context.get(set())
+        # Honour an explicit mark_approved() from the agent approval path even
+        # for critical tools (e.g. execute_command after AutoApproveBackend).
+        if self._approval_cache_key(tool_name, arguments or {}) in self._approved_context.get(set()):
+            return True
+        return False
 
     def _is_session_scoped(
         self, agent_name: Optional[str], tool_name: str, arguments: Optional[Dict]
@@ -344,10 +387,17 @@ class ApprovalRegistry:
         agent_name: Optional[str],
         tool_name: str,
         arguments: Dict,
+        force: bool = False,
     ) -> ApprovalDecision:
-        """Synchronous approval — used by ``Agent._execute_tool_impl``."""
-        # Fast-path: not required
-        if not self.is_required(tool_name):
+        """Synchronous approval — used by ``Agent._execute_tool_impl``.
+
+        ``force`` gates this single call even when the tool is not otherwise
+        registered as requiring approval (e.g. a per-agent ``PermissionManager``
+        ``ask`` rule). It applies only to this call and never mutates shared
+        registry state, so it cannot leak an approval gate onto other agents.
+        """
+        # Fast-path: not required (checks both global and this agent's scope)
+        if not force and not self.is_required(tool_name, agent_name):
             return ApprovalDecision(approved=True, reason="No approval required")
 
         # Already approved in this context
@@ -379,7 +429,7 @@ class ApprovalRegistry:
         request = ApprovalRequest(
             tool_name=tool_name,
             arguments=arguments,
-            risk_level=self._risk_levels.get(tool_name, "medium"),
+            risk_level=self.get_risk_level(tool_name, agent_name) or "medium",
             agent_name=agent_name,
         )
 
@@ -404,10 +454,15 @@ class ApprovalRegistry:
         agent_name: Optional[str],
         tool_name: str,
         arguments: Dict,
+        force: bool = False,
     ) -> ApprovalDecision:
-        """Asynchronous approval — used by async tool execution path."""
-        # Fast-path: not required
-        if not self.is_required(tool_name):
+        """Asynchronous approval — used by async tool execution path.
+
+        See :meth:`approve_sync` for the ``force`` semantics (per-call gate,
+        no shared-state mutation).
+        """
+        # Fast-path: not required (checks both global and this agent's scope)
+        if not force and not self.is_required(tool_name, agent_name):
             return ApprovalDecision(approved=True, reason="No approval required")
 
         if self.is_already_approved(tool_name, arguments):
@@ -435,7 +490,7 @@ class ApprovalRegistry:
         request = ApprovalRequest(
             tool_name=tool_name,
             arguments=arguments,
-            risk_level=self._risk_levels.get(tool_name, "medium"),
+            risk_level=self.get_risk_level(tool_name, agent_name) or "medium",
             agent_name=agent_name,
         )
 

src/praisonai-agents/praisonaiagents/auth/subscription/claude_code.py

@@ -6,20 +6,12 @@ import os
 import re
 import subprocess
 import sys
-import time
-import urllib.parse
-import urllib.request
 from pathlib import Path
 from typing import Any, Dict, Optional
 
-from .protocols import AuthError, SubscriptionAuthProtocol, SubscriptionCredentials
+from .protocols import AuthError, SubscriptionCredentials
 from .registry import register_subscription_provider
 
-_OAUTH_CLIENT_ID = "9d1c250a-e61b-44d9-88ed-5944d1962f5e"
-_TOKEN_ENDPOINTS = (
-    "https://platform.claude.com/v1/oauth/token",
-    "https://console.anthropic.com/v1/oauth/token",
-)
 _CLAUDE_CODE_VERSION_FALLBACK = "2.1.74"
 
 
@@ -97,43 +89,6 @@ def _read_file_credentials() -> Optional[Dict[str, Any]]:
     }
 
 
-def _is_expiring(expires_at_ms: Optional[int], skew_ms: int = 60_000) -> bool:
-    if not expires_at_ms:
-        return False
-    return int(time.time() * 1000) >= (expires_at_ms - skew_ms)
-
-
-def _refresh(refresh_token: str) -> Dict[str, Any]:
-    """Pure refresh — does not touch local files."""
-    if not refresh_token:
-        raise AuthError("no refresh_token; please re-run 'claude /login'")
-    body = urllib.parse.urlencode({
-        "grant_type": "refresh_token",
-        "refresh_token": refresh_token,
-        "client_id": _OAUTH_CLIENT_ID,
-    }).encode()
-    last = None
-    for endpoint in _TOKEN_ENDPOINTS:
-        req = urllib.request.Request(
-            endpoint, data=body, method="POST",
-            headers={
-                "Content-Type": "application/x-www-form-urlencoded",
-                "User-Agent":   f"claude-cli/{_detect_claude_code_version()} (external, cli)",
-            },
-        )
-        try:
-            with urllib.request.urlopen(req, timeout=10) as resp:
-                payload = json.loads(resp.read().decode())
-            return {
-                "accessToken":  payload["access_token"],
-                "refreshToken": payload.get("refresh_token", refresh_token),
-                "expiresAt":    int(time.time() * 1000) + int(payload.get("expires_in", 3600)) * 1000,
-            }
-        except Exception as exc:
-            last = exc
-    raise AuthError(f"Anthropic refresh failed: {last}")
-
-
 class ClaudeCodeAuth:
     """Claude Code OAuth subscription auth."""
 
@@ -159,8 +114,9 @@ class ClaudeCodeAuth:
                 "No Claude Code credentials found. Install Claude Code and run "
                 "'claude /login', or set ANTHROPIC_TOKEN."
             )
-        if _is_expiring(creds["expiresAt"]) and creds.get("refreshToken"):
-            creds.update(_refresh(creds["refreshToken"]))
+        # Read-only: never refresh here. Anthropic rotates refresh tokens and
+        # PraisonAI does not write back to Keychain, which would invalidate
+        # Claude CLI and other tools sharing the same session.
         return SubscriptionCredentials(
             api_key=creds["accessToken"],
             base_url="https://api.anthropic.com",
@@ -171,17 +127,11 @@ class ClaudeCodeAuth:
         )
 
     def refresh(self) -> SubscriptionCredentials:
-        creds = _read_keychain_credentials() or _read_file_credentials()
-        if not creds or not creds.get("refreshToken"):
-            raise AuthError("cannot refresh: no refresh token available")
-        new = _refresh(creds["refreshToken"])
-        return SubscriptionCredentials(
-            api_key=new["accessToken"],
-            base_url="https://api.anthropic.com",
-            headers=self.headers_for("https://api.anthropic.com", ""),
-            auth_scheme="bearer",
-            expires_at_ms=new["expiresAt"],
-            source="claude-code-refreshed",
+        raise AuthError(
+            "PraisonAI does not refresh shared Claude Code OAuth sessions "
+            "(refresh-token rotation would invalidate Claude CLI and other "
+            "tools using the same Keychain entry). Run 'claude /login' or "
+            "let Claude CLI refresh credentials, then retry."
         )
 
     def headers_for(self, base_url: str, model: str) -> Dict[str, str]:
@@ -190,8 +140,7 @@ class ClaudeCodeAuth:
         return {
             "anthropic-beta": ",".join([
                 "interleaved-thinking-2025-05-14",
-                "fine-grained-tool-streaming-2025-05-14", 
-                "context-1m-2025-08-07",
+                "fine-grained-tool-streaming-2025-05-14",
                 "claude-code-20250219",
             ]),
             "user-agent":     f"claude-cli/{_detect_claude_code_version()} (external, cli)",

src/praisonai-agents/praisonaiagents/background/runner.py

@@ -5,7 +5,6 @@ Manages background task execution.
 """
 
 import asyncio
-import logging
 import threading
 from praisonaiagents._logging import get_logger
 from typing import Optional, List, Dict, Any, Callable, Union

src/praisonai-agents/praisonaiagents/background/task.py

@@ -6,7 +6,6 @@ Represents a task running in the background.
 
 import uuid
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 from enum import Enum
 from dataclasses import dataclass, field

src/praisonai-agents/praisonaiagents/bots/__init__.py

@@ -31,6 +31,11 @@ from .protocols import (
     ChannelField,
     ChannelDescriptor,
     WebhookVerifierProtocol,
+    CallbackPayloadStoreProtocol,
+    InMemoryCallbackPayloadStore,
+    GatewayRuntimeSeams,
+    SupportsGatewayRuntime,
+    GatewayAdapterContractError,
 )
 from .base import (
     BasePlatformAdapter,
@@ -118,6 +123,11 @@ __all__ = [
     "ChannelField",
     "ChannelDescriptor",
     "WebhookVerifierProtocol",
+    "CallbackPayloadStoreProtocol",
+    "InMemoryCallbackPayloadStore",
+    "GatewayRuntimeSeams",
+    "SupportsGatewayRuntime",
+    "GatewayAdapterContractError",
     "BasePlatformAdapter",
     "SendResult",
     "InteractiveContext",

src/praisonai-agents/praisonaiagents/bots/config.py

@@ -51,7 +51,7 @@ class BotConfig:
     thread_threshold: int = 500  # Auto-thread responses longer than this (0 = disabled)
     
     # Group message policy
-    group_policy: str = "mention_only"  # respond_all, mention_only, command_only
+    group_policy: str = "mention_only"  # respond_all, mention_only, command_only, observe
     
     # Default safe tools (auto-injected for bots with no tools configured)
     # With workspace scoping, file operations are now safe by construction

src/praisonai-agents/praisonaiagents/bots/interactive.py

@@ -17,10 +17,39 @@ from typing import Any, Awaitable, Callable, Dict, Optional, Tuple, TYPE_CHECKIN
 
 if TYPE_CHECKING:
     from .presentation import PresentationAction
-    from .protocols import BotAdapter, BotMessage
+    from .protocols import BotAdapter, BotMessage, CallbackPayloadStoreProtocol
 
 logger = logging.getLogger(__name__)
 
+# Marker prefixing a stored-reference callback payload. Mirrors
+# ``presentation.CALLBACK_REF_MARKER``; kept local so the inbound handler does
+# not import the (heavier) render module. When an interactive value is too long
+# for the channel callback byte-cap and a callback-payload store is available,
+# the encoder emits ``<namespace>:@<ref>``; the registry resolves the reference
+# back to the exact value the agent authored before routing it.
+_CALLBACK_REF_MARKER = "@"
+
+
+def _extract_ref(value: Optional[str]) -> Optional[str]:
+    """Return the stored-payload reference in ``value``, or ``None``.
+
+    The decoded callback ``value`` is either the whole payload (``reply``) or a
+    ``<action_id>:<value>`` tail (``select``). A stored reference always appears
+    as a trailing ``@<ref>`` segment — i.e. the marker sits at the start of the
+    value or immediately after a ``:`` separator. Requiring that anchoring means
+    an ordinary value that merely contains an ``@`` (e.g. an email address) is
+    never mistaken for a reference.
+    """
+    if not isinstance(value, str) or not value:
+        return None
+    if value.startswith(_CALLBACK_REF_MARKER):
+        return value[len(_CALLBACK_REF_MARKER):] or None
+    sep = f":{_CALLBACK_REF_MARKER}"
+    idx = value.rfind(sep)
+    if idx != -1:
+        return value[idx + len(sep):] or None
+    return None
+
 
 @dataclass
 class InteractiveContext:
@@ -176,11 +205,25 @@ class InteractiveRegistry:
     callback namespaces. Each namespace can have one handler.
     """
     
-    def __init__(self):
-        """Initialize the registry."""
+    def __init__(
+        self,
+        store: Optional["CallbackPayloadStoreProtocol"] = None,
+    ):
+        """Initialize the registry.
+
+        Args:
+            store: Optional :class:`CallbackPayloadStoreProtocol` used to resolve
+                stored-reference callbacks (``<namespace>:@<ref>``) emitted when
+                an interactive value was too long to travel inline on a
+                tight-callback-cap channel. When supplied, a reference is
+                resolved back to the exact value before the handler runs, so
+                long ``reply``/``select`` values round-trip losslessly. When
+                omitted, behaviour is unchanged.
+        """
         self._handlers: Dict[str, InteractiveHandler] = {}
         self._authorizers: Dict[str, "InteractiveAuthorizer"] = {}
         self._fallback_handler: Optional[InteractiveHandler] = None
+        self._store = store
     
     def register(
         self,
@@ -239,7 +282,39 @@ class InteractiveRegistry:
             True if handled, False otherwise
         """
         namespace, payload = decode_callback(context.callback_data)
-        
+
+        # Resolve a stored-reference payload (``<namespace>:@<ref>``) back to the
+        # canonical value the agent authored. Long ``reply``/``select`` values
+        # that overflow the channel callback byte-cap are persisted under a short
+        # reference on render; here we restore the exact value before routing so
+        # the handler receives what the user actually chose. An unknown/expired
+        # reference is dropped (fails closed) rather than routing the opaque
+        # reference token.
+        value = payload.get("value") if isinstance(payload, dict) else None
+        ref = _extract_ref(value)
+        if ref is not None:
+            resolved = None
+            if self._store is not None:
+                try:
+                    resolved = self._store.get(ref)
+                except Exception as e:  # pragma: no cover - defensive
+                    logger.error(f"Callback payload store lookup failed: {e}")
+                    resolved = None
+            if resolved is None:
+                logger.warning(
+                    "Interactive callback referenced a stored value that is "
+                    "unknown or expired; dropping rather than routing an "
+                    "unresolvable reference."
+                )
+                return False
+            # Replace only the trailing ``@<ref>`` segment with the restored
+            # value, preserving any leading ``<action_id>:`` prefix (the
+            # ``select`` payload is ``<action_id>:<value>``) so handlers that
+            # parse the action scope still see it.
+            marked = f"{_CALLBACK_REF_MARKER}{ref}"
+            payload = dict(payload)
+            payload["value"] = value[: len(value) - len(marked)] + resolved
+
         # Try to find a handler for this namespace
         handler = self._handlers.get(namespace)
         
@@ -325,16 +400,27 @@ class InteractiveRegistry:
 _global_registry = InteractiveRegistry()
 
 
-def create_registry() -> InteractiveRegistry:
+def create_registry(
+    store: Optional["CallbackPayloadStoreProtocol"] = None,
+) -> InteractiveRegistry:
     """Create a new interactive registry instance.
     
     Each adapter should create its own registry to avoid conflicts
     when multiple adapters are used in the same process.
-    
+
+    Args:
+        store: Optional :class:`CallbackPayloadStoreProtocol` used to resolve
+            stored-reference callbacks (``<namespace>:@<ref>``). Pass the same
+            store instance that the render side hands to
+            ``adapt_presentation(..., callback_store=...)`` so long
+            ``reply``/``select`` values round-trip losslessly on
+            tight-callback-cap channels (e.g. Telegram). When omitted,
+            behaviour is unchanged.
+
     Returns:
         A new InteractiveRegistry instance
     """
-    return InteractiveRegistry()
+    return InteractiveRegistry(store=store)
 
 
 def get_registry() -> InteractiveRegistry:

src/praisonai-agents/praisonaiagents/bots/presentation.py

@@ -12,14 +12,41 @@ rendering belongs in ``praisonai-bot`` (``praisonai_bot.bots``).
 from __future__ import annotations
 
 import hashlib
+import time
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Dict, List, Literal, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Union
+
+if TYPE_CHECKING:
+    from .protocols import CallbackPayloadStoreProtocol
 
 # Most channels (e.g. Telegram) hard-cap inline callback payloads at 64 bytes.
 # Keep degraded select callbacks within this bound while preserving uniqueness.
 _MAX_CALLBACK_LEN = 64
 
+# Marker prefixing a stored-reference payload. When an interactive value does
+# not fit the channel callback byte-cap and a ``CallbackPayloadStoreProtocol``
+# is available, the canonical value is persisted under a short reference and the
+# callback carries ``@<ref>`` instead of the (unrecoverable) hash. The inbound
+# registry resolves the reference back to the exact value on click.
+CALLBACK_REF_MARKER = "@"
+
+# Default lifetime (seconds) for a persisted callback reference. An interactive
+# menu is a short-lived affordance; a bounded TTL keeps the store from growing
+# without cleanup while comfortably covering a user pondering their choice.
+CALLBACK_REF_TTL = 3600.0
+
+
+def _callback_ref(namespace: str, value: str) -> str:
+    """Build a short, collision-resistant reference for a callback ``value``.
+
+    The reference is derived from both the namespace/action scope and the value
+    so distinct choices stay distinct, and is short enough to always fit within
+    ``_MAX_CALLBACK_LEN`` alongside its prefix.
+    """
+    digest = hashlib.sha256(f"{namespace}\x00{value}".encode("utf-8")).hexdigest()
+    return digest[:16]
+
 
 class ActionType(str, Enum):
     """Types of actions that can be triggered by interactive elements."""
@@ -570,7 +597,11 @@ class PresentationLimits:
         )
 
 
-def _adapt_button(button: PresentationButton, limits: PresentationLimits) -> PresentationButton:
+def _adapt_button(
+    button: PresentationButton,
+    limits: PresentationLimits,
+    store: Optional["CallbackPayloadStoreProtocol"] = None,
+) -> PresentationButton:
     """Return a copy of a button adapted to the given limits.
 
     Truncates the label and, when the channel does not support web apps,
@@ -591,7 +622,7 @@ def _adapt_button(button: PresentationButton, limits: PresentationLimits) -> Pre
         if action_type == ActionType.REPLY.value and action.value is not None:
             action = PresentationAction(
                 type=ActionType.CALLBACK,
-                value=_encode_reply_callback(action.value),
+                value=_encode_reply_callback(action.value, store),
             )
         elif (
             not limits.supports_web_apps
@@ -624,7 +655,10 @@ REPLY_CALLBACK_PREFIX = "reply:"
 REPLY_HASH_MARKER = "#"
 
 
-def _encode_reply_callback(value: str) -> str:
+def _encode_reply_callback(
+    value: str,
+    store: Optional["CallbackPayloadStoreProtocol"] = None,
+) -> str:
     """Build a channel-safe callback payload for a ``reply`` action.
 
     Produces ``reply:<value>`` so the inbound interactive registry can route
@@ -632,33 +666,59 @@ def _encode_reply_callback(value: str) -> str:
     in UTF-8 bytes because channel callback caps (e.g. Telegram's 64-byte cap)
     are byte limits, not character limits.
 
-    When the raw form exceeds ``_MAX_CALLBACK_LEN`` the value is replaced with a
-    short, collision-resistant hash marked with ``#`` (``reply:#<digest>``).
-    Truncating to a prefix was unsafe: two long choices sharing a prefix would
-    collapse to the same payload, and the agent would receive a value it never
-    authored. The hash keeps distinct choices distinct; the marker lets the
-    reply handler recognise that the original value could not be carried inline
-    and avoid routing a lossy value into the turn.
+    When the raw form exceeds ``_MAX_CALLBACK_LEN`` and a *store* is available,
+    the canonical value is persisted under a short reference and the callback
+    carries ``reply:@<ref>``; the inbound handler resolves the reference back to
+    the exact value, so long values round-trip losslessly.
+
+    Without a store the value is replaced with a short, collision-resistant hash
+    marked with ``#`` (``reply:#<digest>``). Truncating to a prefix was unsafe:
+    two long choices sharing a prefix would collapse to the same payload. The
+    marker lets the reply handler recognise that the original value could not be
+    carried and avoid routing a lossy value into the turn.
     """
     raw = f"{REPLY_CALLBACK_PREFIX}{value}"
     if len(raw.encode("utf-8")) <= _MAX_CALLBACK_LEN:
         return raw
+    if store is not None:
+        ref = _callback_ref(REPLY_CALLBACK_PREFIX.rstrip(":"), value)
+        store.put(ref, value, expires_at=time.time() + CALLBACK_REF_TTL)
+        return f"{REPLY_CALLBACK_PREFIX}{CALLBACK_REF_MARKER}{ref}"
     digest = hashlib.sha1(value.encode("utf-8")).hexdigest()[:16]
     return f"{REPLY_CALLBACK_PREFIX}{REPLY_HASH_MARKER}{digest}"
 
 
-def _encode_select_callback(action_id: str, value: str) -> str:
+def _encode_select_callback(
+    action_id: str,
+    value: str,
+    store: Optional["CallbackPayloadStoreProtocol"] = None,
+) -> str:
     """Build a channel-safe callback payload for a degraded select option.
 
     The raw ``select:<action_id>:<value>`` form can exceed channel callback
-    limits (e.g. Telegram's 64-byte cap) and collide when distinct options
-    share a long prefix. When the raw payload fits within ``_MAX_CALLBACK_LEN``
-    it is returned unchanged; otherwise the value is replaced with a short,
-    collision-resistant hash so distinct options stay distinct after truncation.
+    limits (e.g. Telegram's 64-byte cap). When it fits it is returned unchanged.
+
+    When it overflows and a *store* is available, the canonical value is
+    persisted under a short reference and the callback carries
+    ``select:<action_id>:@<ref>``; the registry resolves the reference back to
+    the exact value on click, so long option values round-trip losslessly.
+
+    Without a store the value is replaced with a short, collision-resistant hash
+    so distinct options stay distinct after truncation (but the value cannot be
+    recovered — see the issue this addresses).
     """
     raw = f"select:{action_id}:{value}"
     if len(raw.encode("utf-8")) <= _MAX_CALLBACK_LEN:
         return raw
+    if store is not None:
+        ref = _callback_ref(f"select:{action_id}", value)
+        store.put(ref, value, expires_at=time.time() + CALLBACK_REF_TTL)
+        prefix = f"select:{action_id}:{CALLBACK_REF_MARKER}"
+        if len(prefix.encode("utf-8")) + len(ref) > _MAX_CALLBACK_LEN:
+            # action_id itself is long; hash it so the ref still fits the bound.
+            aid_digest = hashlib.sha1((action_id or "").encode("utf-8")).hexdigest()[:8]
+            prefix = f"select:{aid_digest}:{CALLBACK_REF_MARKER}"
+        return f"{prefix}{ref}"
     digest = hashlib.sha1(value.encode("utf-8")).hexdigest()[:16]
     prefix = f"select:{action_id}:"
     # Reserve room for the digest; trim the action_id prefix if needed.
@@ -670,7 +730,10 @@ def _encode_select_callback(action_id: str, value: str) -> str:
     return f"{prefix[:_MAX_CALLBACK_LEN - len(digest)]}{digest}"
 
 
-def _select_to_buttons(block: PresentationBlock) -> PresentationBlock:
+def _select_to_buttons(
+    block: PresentationBlock,
+    store: Optional["CallbackPayloadStoreProtocol"] = None,
+) -> PresentationBlock:
     """Convert a SELECT block into an equivalent BUTTONS block.
 
     Used when a channel does not support native select menus. Each option
@@ -688,7 +751,7 @@ def _select_to_buttons(block: PresentationBlock) -> PresentationBlock:
                 label=label,
                 action=PresentationAction(
                     type=ActionType.CALLBACK,
-                    value=_encode_select_callback(action_id, option.value),
+                    value=_encode_select_callback(action_id, option.value, store),
                 ),
             )
         )
@@ -698,6 +761,8 @@ def _select_to_buttons(block: PresentationBlock) -> PresentationBlock:
 def adapt_presentation(
     presentation: MessagePresentation,
     limits: PresentationLimits,
+    *,
+    callback_store: Optional["CallbackPayloadStoreProtocol"] = None,
 ) -> MessagePresentation:
     """Return a copy of ``presentation`` guaranteed to satisfy ``limits``.
 
@@ -720,6 +785,12 @@ def adapt_presentation(
     Args:
         presentation: The portable presentation to adapt.
         limits: The target channel's capability limits.
+        callback_store: Optional :class:`CallbackPayloadStoreProtocol`. When a
+            ``reply``/``select`` value overflows the channel callback byte-cap
+            and a store is supplied, the canonical value is persisted under a
+            short reference and the callback carries ``@<ref>`` so the inbound
+            registry can resolve the exact value on click. When omitted the
+            existing (lossy) hash behaviour is preserved for compatibility.
 
     Returns:
         A new ``MessagePresentation`` that is safe to render natively.
@@ -731,7 +802,7 @@ def adapt_presentation(
 
         if block_type == BlockType.SELECT.value and not limits.supports_select:
             # Degrade select -> buttons, then adapt the resulting buttons block
-            block = _select_to_buttons(block)
+            block = _select_to_buttons(block, callback_store)
             block_type = BlockType.BUTTONS.value
 
         if block_type == BlockType.BUTTONS.value and block.buttons:
@@ -754,7 +825,7 @@ def adapt_presentation(
                 kept.sort(key=lambda iv: iv[0])
                 buttons = [b for _, b in kept]
 
-            adapted_buttons = [_adapt_button(b, limits) for b in buttons]
+            adapted_buttons = [_adapt_button(b, limits, callback_store) for b in buttons]
             adapted_blocks.append(
                 PresentationBlock(type=BlockType.BUTTONS, buttons=adapted_buttons)
             )

src/praisonai-agents/praisonaiagents/bots/protocols.py

@@ -302,6 +302,79 @@ class WebhookVerifierProtocol(Protocol):
         ...
 
 
+class GatewayAdapterContractError(TypeError):
+    """Raised when a channel adapter cannot receive the gateway runtime seams.
+
+    The gateway wires four reliability seams into every channel adapter —
+    the cross-platform identity resolver, the delivery router, the admission
+    gate, and the shared per-turn lock map. When the gateway has these seams
+    to inject but the adapter neither implements
+    :class:`SupportsGatewayRuntime` nor exposes a compatible session, the wiring
+    would silently no-op and the adapter would run without admission control,
+    durable delivery routing, or cross-platform turn locking. Failing loudly
+    with this error surfaces the missing contract instead.
+    """
+
+
+@dataclass
+class GatewayRuntimeSeams:
+    """The gateway reliability seams handed to a channel adapter at build time.
+
+    A single, typed carrier for the four runtime seams the gateway injects so
+    an adapter wires them up in one place instead of via four duck-typed
+    private-attribute splices:
+
+    Attributes:
+        identity_resolver: Cross-platform identity resolver; unifies the same
+            human across platforms onto one session key.
+        delivery_router: Backing router for the built-in ``send_message`` tool,
+            so an agent turn can proactively reach the user mid-task.
+        admission_gate: Gateway-wide admission control (concurrency ceiling /
+            fair queue / backpressure) applied to inbound runs.
+        turn_lock_map: Shared per-turn lock map for cross-platform turn
+            serialisation on the resolved session id.
+
+    A seam left ``None`` means the gateway has nothing to inject for it and the
+    adapter should keep whatever it already has.
+    """
+
+    identity_resolver: Optional[Any] = None
+    delivery_router: Optional[Any] = None
+    admission_gate: Optional[Any] = None
+    turn_lock_map: Optional[Any] = None
+
+
+@runtime_checkable
+class SupportsGatewayRuntime(Protocol):
+    """Contract for channel adapters that accept the gateway runtime seams.
+
+    Instead of the gateway reaching into an adapter's private session and
+    splicing ``_identity_resolver`` / ``_delivery_router`` / ``_admission_gate``
+    / ``_locks`` via ``getattr``/``hasattr`` (which silently no-ops on any
+    adapter whose session does not match the built-in shape), an adapter — in
+    tree or ``pip``-installed — implements this one method. The gateway calls it
+    once with a :class:`GatewayRuntimeSeams`, so the same reliability guarantees
+    (admission/backpressure, delivery routing, cross-platform turn locking)
+    hold for *every* adapter.
+
+    Adapters whose ``__init__`` creates a ``BotSessionManager`` get this for
+    free — the session manager implements ``attach_gateway_runtime`` and the
+    adapter can simply delegate to ``self._session.attach_gateway_runtime(...)``.
+    """
+
+    def attach_gateway_runtime(self, runtime: "GatewayRuntimeSeams") -> None:
+        """Wire the gateway runtime seams into this adapter.
+
+        Only seams present (non-``None``) on ``runtime`` should be applied;
+        seams left ``None`` mean the gateway has nothing to inject and the
+        adapter must keep its existing value.
+
+        Args:
+            runtime: The gateway reliability seams to attach.
+        """
+        ...
+
+
 class MessageType(str, Enum):
     """Types of bot messages."""
     
@@ -1274,3 +1347,77 @@ class BotOSProtocol(Protocol):
             The Bot instance, or None if not found.
         """
         ...
+
+
+@runtime_checkable
+class CallbackPayloadStoreProtocol(Protocol):
+    """Protocol for durable, reference-addressable interactive callback values.
+
+    Some channels hard-cap inline callback payloads (e.g. Telegram's 64-byte
+    inline-callback limit). When an interactive ``reply``/``select`` value is
+    too long to travel inline, the framework persists the canonical value under
+    a short, collision-resistant reference and emits that reference in the
+    callback instead. On click the registry resolves the reference back to the
+    exact value — so long option values (URLs, file paths, free-text choices)
+    round-trip losslessly on every channel.
+
+    This mirrors :class:`ApprovalStoreProtocol`: the contract lives in core so
+    any backend and any channel can interoperate; core ships a bounded
+    in-memory default (:class:`InMemoryCallbackPayloadStore`) and heavier
+    durable backends (e.g. SQLite) live in the ``praisonai-bot`` runtime.
+
+    Behaviour is unchanged when no store is configured — values that fit inline
+    are always sent inline, so this is additive and backward compatible.
+    """
+
+    def put(self, ref: str, value: str, *, expires_at: float) -> None:
+        """Persist ``value`` under ``ref`` until ``expires_at`` (epoch seconds).
+
+        Called from the synchronous render path, so this is a plain method.
+        """
+        ...
+
+    def get(self, ref: str) -> Optional[str]:
+        """Return the value stored for ``ref``, or ``None`` if unknown/expired."""
+        ...
+
+
+class InMemoryCallbackPayloadStore:
+    """Bounded, zero-dependency in-memory :class:`CallbackPayloadStoreProtocol`.
+
+    The default store when a channel does not inject a durable one. Entries are
+    kept until their ``expires_at`` and the store is capped at ``max_entries``
+    (oldest inserted evicted first) so a long-running process cannot grow
+    unbounded. This is per-process only; durable, restart-surviving persistence
+    belongs to a runtime-provided backend (as approvals already do).
+    """
+
+    def __init__(self, *, max_entries: int = 4096) -> None:
+        self._max_entries = max(1, int(max_entries))
+        # ref -> (value, expires_at); insertion-ordered for FIFO eviction.
+        self._entries: "Dict[str, tuple[str, float]]" = {}
+
+    def _purge_expired(self, now: float) -> None:
+        expired = [ref for ref, (_, exp) in self._entries.items() if exp <= now]
+        for ref in expired:
+            self._entries.pop(ref, None)
+
+    def put(self, ref: str, value: str, *, expires_at: float) -> None:
+        now = time.time()
+        self._purge_expired(now)
+        # Refresh insertion order on overwrite so it is treated as most-recent.
+        self._entries.pop(ref, None)
+        self._entries[ref] = (value, expires_at)
+        while len(self._entries) > self._max_entries:
+            oldest = next(iter(self._entries))
+            self._entries.pop(oldest, None)
+
+    def get(self, ref: str) -> Optional[str]:
+        entry = self._entries.get(ref)
+        if entry is None:
+            return None
+        value, expires_at = entry
+        if expires_at <= time.time():
+            self._entries.pop(ref, None)
+            return None
+        return value

src/praisonai-agents/praisonaiagents/bus/bus.py

@@ -5,7 +5,6 @@ Provides publish/subscribe functionality for typed events.
 """
 
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 import threading
 from typing import Any, Callable, Dict, List, Optional, Set, Union

src/praisonai-agents/praisonaiagents/checkpoints/service.py

@@ -7,7 +7,6 @@ Implements shadow git repository for file-level checkpointing.
 import os
 import re
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 import shutil
 from typing import Optional, List, Dict, Any, Callable

src/praisonai-agents/praisonaiagents/cli_backend/__init__.py

@@ -25,6 +25,19 @@ def __getattr__(name: str):
     elif name == "CliBackendDelta":
         from .protocols import CliBackendDelta
         return CliBackendDelta
+    elif name == "cli_backend_debug_enabled":
+        raise AttributeError(
+            f"module '{__name__}' has no attribute '{name}'. "
+            "CLI backend logging moved to praisonai-plugins cli_backend_tracer."
+        )
+    elif name == "log_cli_backend_execution":
+        raise AttributeError(
+            f"module '{__name__}' has no attribute '{name}'. "
+            "CLI backend logging moved to praisonai-plugins cli_backend_tracer."
+        )
+    elif name == "backend_label":
+        from .debug import backend_label
+        return backend_label
     else:
         raise AttributeError(f"module '{__name__}' has no attribute '{name}'")
 
@@ -33,5 +46,6 @@ __all__ = [
     "CliBackendConfig", 
     "CliSessionBinding",
     "CliBackendResult",
-    "CliBackendDelta"
+    "CliBackendDelta",
+    "backend_label",
 ]

src/praisonai-agents/praisonaiagents/cli_backend/debug.py

@@ -0,0 +1,42 @@
+"""CLI backend helpers for hooks and observability."""
+
+from __future__ import annotations
+
+from typing import Any
+
+
+def backend_label(backend: Any) -> str:
+    """Human-readable backend identifier for hooks and logs."""
+    config = getattr(backend, "config", None)
+    command = getattr(config, "command", None)
+    if command:
+        return str(command)
+    return type(backend).__name__
+
+
+_PROMPT_FLAGS = frozenset(
+    {"-p", "--prompt", "-i", "--input", "-m", "--message", "--system"}
+)
+
+
+def redact_command(command: Any) -> Any:
+    """Redact prompt/system values from a subprocess argv for safe serialization.
+
+    Keeps the executable and flags visible for verification while masking the
+    value that follows a known prompt-bearing flag, since that value may carry
+    user prompts or system instructions that should not leak into log sinks.
+    Non-list inputs are returned unchanged.
+    """
+    if not isinstance(command, (list, tuple)):
+        return command
+    redacted = []
+    mask_next = False
+    for arg in command:
+        if mask_next:
+            redacted.append("<redacted>")
+            mask_next = False
+            continue
+        redacted.append(arg)
+        if isinstance(arg, str) and arg in _PROMPT_FLAGS:
+            mask_next = True
+    return redacted

src/praisonai-agents/praisonaiagents/compaction/compactor.py

@@ -160,7 +160,8 @@ class ContextCompactor:
                 )
             except Exception:
                 pass
-        return max(1, len(text) // 4)
+        from ..context.tokens import estimate_tokens_heuristic
+        return estimate_tokens_heuristic(text)
 
     def count_message_tokens(self, message: Dict[str, Any]) -> int:
         """Count tokens in a message, including tool_calls payloads."""

src/praisonai-agents/praisonaiagents/conditions/evaluator.py

@@ -8,7 +8,6 @@ The implementation is extracted from workflows.py _evaluate_condition()
 to enable DRY condition handling across the codebase.
 """
 import re
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Dict, Any, List, Optional
 

src/praisonai-agents/praisonaiagents/config/feature_configs.py

@@ -950,7 +950,16 @@ class ExecutionConfig:
         if isinstance(context_compaction, dict):
             from ..context.policy import ContextCompactionPolicy
             context_compaction = ContextCompactionPolicy.from_dict(context_compaction)
-        
+
+        # Handle compaction_strategy restoration (serialised as a string value in to_dict)
+        compaction_strategy = data.get("compaction_strategy")
+        if isinstance(compaction_strategy, str):
+            from ..compaction.strategy import CompactionStrategy
+            try:
+                compaction_strategy = CompactionStrategy(compaction_strategy)
+            except ValueError:
+                compaction_strategy = None
+
         return cls(
             max_iter=data.get("max_iter", 20),
             max_steps=data.get("max_steps", None),
@@ -958,13 +967,17 @@ class ExecutionConfig:
             max_rpm=data.get("max_rpm", None),
             max_execution_time=data.get("max_execution_time", None),
             max_retry_limit=data.get("max_retry_limit", 2),
+            retry_initial_delay=data.get("retry_initial_delay", 1.0),
+            retry_backoff_factor=data.get("retry_backoff_factor", 2.0),
+            retry_jitter=data.get("retry_jitter", 0.1),
             code_execution=data.get("code_execution", False),
             code_mode=data.get("code_mode", "safe"),
-            code_sandbox_mode=data.get("code_sandbox_mode", "docker"),
+            code_sandbox_mode=data.get("code_sandbox_mode", "sandbox"),
             code_tools=data.get("code_tools", False),
             code_tools_allow=data.get("code_tools_allow", None),
             context_compaction=context_compaction,
             max_context_tokens=data.get("max_context_tokens", None),
+            compaction_strategy=compaction_strategy,
             max_budget=data.get("max_budget", None),
             parallel_tool_calls=data.get("parallel_tool_calls", False),
         )
@@ -1076,7 +1089,13 @@ class ToolConfig:
     artifact_retention_days: int = 7  # Days to retain artifacts before garbage collection
     artifact_store: Optional[Any] = None  # Custom artifact store instance
     redact_secrets: bool = True  # Whether to redact secrets from artifacts
-    
+
+    # Allow resolving tool calls via the process-global @tool registry when a
+    # tool name is not in this agent's declared tools=[...]. Default False so an
+    # agent is scoped strictly to its own tools (safe by default); opt in for
+    # plugin-host style dynamic dispatch.
+    allow_global_tools: bool = False
+
     def __post_init__(self) -> None:
         """Validate configuration after initialization."""
         if self.output_limit <= 0:
@@ -1105,6 +1124,7 @@ class ToolConfig:
             "artifact_retention_days": self.artifact_retention_days,
             "artifact_store": self.artifact_store,
             "redact_secrets": self.redact_secrets,
+            "allow_global_tools": self.allow_global_tools,
         }
     
     @classmethod
@@ -1130,6 +1150,7 @@ class ToolConfig:
             artifact_retention_days=data.get("artifact_retention_days", 7),
             artifact_store=data.get("artifact_store"),
             redact_secrets=data.get("redact_secrets", True),
+            allow_global_tools=data.get("allow_global_tools", False),
         )
 
 

src/praisonai-agents/praisonaiagents/config/parse_utils.py

@@ -50,42 +50,6 @@ def detect_url_scheme(value: str) -> Optional[str]:
     return None
 
 
-def parse_url_to_config(
-    url: str,
-    config_class: type,
-    url_schemes: Dict[str, str],
-) -> Any:
-    """
-    Parse a URL string into a config object.
-    
-    Args:
-        url: URL string (e.g., "postgresql://localhost/db")
-        config_class: Config dataclass to instantiate
-        url_schemes: Mapping of URL schemes to backend names
-        
-    Returns:
-        Config instance with backend and url set
-        
-    Raises:
-        ValueError: If URL scheme is not supported
-    """
-    scheme = detect_url_scheme(url)
-    if not scheme:
-        raise ValueError(f"Invalid URL format: {url}")
-    
-    if scheme not in url_schemes:
-        valid_schemes = ", ".join(sorted(url_schemes.keys()))
-        raise ValueError(
-            f"Unsupported URL scheme '{scheme}' in '{url}'. "
-            f"Supported schemes: {valid_schemes}"
-        )
-    
-    backend = url_schemes[scheme]
-    
-    # Create config with backend and URL in config dict
-    return config_class(backend=backend, config={"url": url})
-
-
 def is_path_like(value: str) -> bool:
     """
     Check if a string looks like a file path. O(1) operation.
@@ -136,21 +100,6 @@ def is_path_like(value: str) -> bool:
     return False
 
 
-def is_numeric_string(value: str) -> bool:
-    """
-    Check if a string is numeric. O(1) operation.
-    
-    Args:
-        value: String to check
-        
-    Returns:
-        True if string is numeric
-    """
-    if not isinstance(value, str):
-        return False
-    return value.isdigit()
-
-
 def suggest_similar(value: str, candidates: Iterable[str], max_distance: int = 2) -> Optional[str]:
     """
     Find the most similar string from candidates using Levenshtein distance.
@@ -246,28 +195,6 @@ def make_preset_error(
     return ValueError(" ".join(msg_parts))
 
 
-def make_array_error(
-    param_name: str,
-    value: list,
-    expected_format: str,
-) -> ValueError:
-    """
-    Create a helpful error message for invalid array format.
-    
-    Args:
-        param_name: Name of the parameter
-        value: Invalid array value
-        expected_format: Description of expected format
-        
-    Returns:
-        ValueError with helpful message
-    """
-    return ValueError(
-        f"Invalid {param_name} array format: {value}. "
-        f"Expected: {expected_format}"
-    )
-
-
 def is_policy_string(value: str) -> bool:
     """
     Check if a string is a policy specification. O(1) operation.

src/praisonai-agents/praisonaiagents/context/aggregator.py

@@ -6,7 +6,6 @@ Inspired by CrewAI's ContextualMemory pattern.
 """
 
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass, field
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union

src/praisonai-agents/praisonaiagents/context/fast/async_file_ops.py

@@ -12,7 +12,6 @@ Design principles:
 """
 
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Optional, List
 

src/praisonai-agents/praisonaiagents/context/fast/compressor.py

@@ -12,7 +12,6 @@ Design principles:
 - Preserves most relevant content
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Protocol, Optional, List
 from dataclasses import dataclass

src/praisonai-agents/praisonaiagents/context/fast/context_injector.py

@@ -7,7 +7,6 @@ Injects FastContext results into the main agent with:
 - Precision over recall (avoid context pollution)
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Optional, List, Dict, Any
 from dataclasses import dataclass

src/praisonai-agents/praisonaiagents/context/fast/fast_context.py

@@ -25,7 +25,6 @@ Example usage:
 import os
 import hashlib
 import json
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Optional, List, Dict, Any
 

src/praisonai-agents/praisonaiagents/context/fast/fast_context_agent.py

@@ -10,7 +10,6 @@ This agent is designed for fast context retrieval using:
 
 import os
 import json
-import logging
 from praisonaiagents._logging import get_logger
 from typing import List, Dict, Any, Optional
 

src/praisonai-agents/praisonaiagents/context/fast/index_manager.py

@@ -15,7 +15,6 @@ Design principles:
 import os
 import json
 import hashlib
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass, field, asdict
 from pathlib import Path

src/praisonai-agents/praisonaiagents/context/fast/indexer/file_indexer.py

@@ -12,7 +12,6 @@ import os
 import fnmatch
 import hashlib
 import json
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass
 from typing import Dict, List, Set, Optional, Any

src/praisonai-agents/praisonaiagents/context/fast/indexer/file_watcher.py

@@ -1,318 +0,0 @@
-"""
-File Watcher for Incremental Indexing.
-
-Monitors file system changes and triggers incremental index updates:
-- Watches for file creation, modification, deletion
-- Debounces rapid changes
-- Supports background indexing
-"""
-
-import os
-import time
-import logging
-from praisonaiagents._logging import get_logger
-import threading
-from typing import Dict, Set, Callable, Optional, Any
-from dataclasses import dataclass
-from enum import Enum
-
-logger = get_logger(__name__)
-
-class ChangeType(Enum):
-    """Types of file system changes."""
-    CREATED = "created"
-    MODIFIED = "modified"
-    DELETED = "deleted"
-
-@dataclass
-class FileChange:
-    """Represents a file system change.
-    
-    Attributes:
-        path: Path to the changed file
-        change_type: Type of change
-        timestamp: When the change was detected
-    """
-    path: str
-    change_type: ChangeType
-    timestamp: float
-
-class FileWatcher:
-    """Watches for file system changes and triggers callbacks.
-    
-    Uses polling-based approach for cross-platform compatibility.
-    For production use, consider using watchdog library.
-    
-    Attributes:
-        workspace_path: Root directory to watch
-        extensions: File extensions to watch
-        poll_interval: Seconds between polls
-    """
-    
-    def __init__(
-        self,
-        workspace_path: str,
-        extensions: Optional[Set[str]] = None,
-        poll_interval: float = 1.0,
-        debounce_delay: float = 0.5,
-        on_change: Optional[Callable[[FileChange], None]] = None
-    ):
-        """Initialize file watcher.
-        
-        Args:
-            workspace_path: Root directory to watch
-            extensions: File extensions to watch (None = all)
-            poll_interval: Seconds between polls
-            debounce_delay: Delay before processing changes
-            on_change: Callback for file changes
-        """
-        self.workspace_path = os.path.abspath(workspace_path)
-        self.extensions = extensions or {
-            '.py', '.js', '.ts', '.jsx', '.tsx', '.go', '.rs', '.java',
-            '.c', '.cpp', '.h', '.hpp', '.cs', '.rb', '.php', '.swift'
-        }
-        self.poll_interval = poll_interval
-        self.debounce_delay = debounce_delay
-        self.on_change = on_change
-        
-        # State tracking
-        self._file_mtimes: Dict[str, float] = {}
-        self._pending_changes: Dict[str, FileChange] = {}
-        self._running = False
-        self._thread: Optional[threading.Thread] = None
-        self._lock = threading.Lock()
-    
-    def _scan_files(self) -> Dict[str, float]:
-        """Scan workspace and get file modification times.
-        
-        Returns:
-            Dict mapping file paths to modification times
-        """
-        mtimes = {}
-        
-        for root, dirs, files in os.walk(self.workspace_path):
-            # Skip hidden directories
-            dirs[:] = [d for d in dirs if not d.startswith('.')]
-            
-            for filename in files:
-                ext = os.path.splitext(filename)[1].lower()
-                if ext in self.extensions:
-                    file_path = os.path.join(root, filename)
-                    try:
-                        mtimes[file_path] = os.path.getmtime(file_path)
-                    except OSError:
-                        pass
-        
-        return mtimes
-    
-    def _detect_changes(self, new_mtimes: Dict[str, float]) -> None:
-        """Detect changes between old and new file states.
-        
-        Args:
-            new_mtimes: New file modification times
-        """
-        old_paths = set(self._file_mtimes.keys())
-        new_paths = set(new_mtimes.keys())
-        
-        now = time.time()
-        
-        # Detect created files
-        for path in new_paths - old_paths:
-            self._add_pending_change(FileChange(
-                path=path,
-                change_type=ChangeType.CREATED,
-                timestamp=now
-            ))
-        
-        # Detect deleted files
-        for path in old_paths - new_paths:
-            self._add_pending_change(FileChange(
-                path=path,
-                change_type=ChangeType.DELETED,
-                timestamp=now
-            ))
-        
-        # Detect modified files
-        for path in old_paths & new_paths:
-            if new_mtimes[path] > self._file_mtimes[path]:
-                self._add_pending_change(FileChange(
-                    path=path,
-                    change_type=ChangeType.MODIFIED,
-                    timestamp=now
-                ))
-        
-        self._file_mtimes = new_mtimes
-    
-    def _add_pending_change(self, change: FileChange) -> None:
-        """Add a change to pending queue (with debouncing).
-        
-        Args:
-            change: File change to add
-        """
-        with self._lock:
-            self._pending_changes[change.path] = change
-    
-    def _process_pending_changes(self) -> None:
-        """Process pending changes after debounce delay."""
-        with self._lock:
-            now = time.time()
-            to_process = []
-            
-            for path, change in list(self._pending_changes.items()):
-                if now - change.timestamp >= self.debounce_delay:
-                    to_process.append(change)
-                    del self._pending_changes[path]
-        
-        # Process changes outside lock
-        for change in to_process:
-            if self.on_change:
-                try:
-                    self.on_change(change)
-                except Exception as e:
-                    logger.error(f"Error processing change {change.path}: {e}")
-    
-    def _poll_loop(self) -> None:
-        """Main polling loop."""
-        # Initial scan
-        self._file_mtimes = self._scan_files()
-        
-        while self._running:
-            try:
-                # Scan for changes
-                new_mtimes = self._scan_files()
-                self._detect_changes(new_mtimes)
-                
-                # Process pending changes
-                self._process_pending_changes()
-                
-            except Exception as e:
-                logger.error(f"Error in file watcher: {e}")
-            
-            time.sleep(self.poll_interval)
-    
-    def start(self) -> None:
-        """Start watching for file changes."""
-        if self._running:
-            return
-        
-        self._running = True
-        self._thread = threading.Thread(target=self._poll_loop, daemon=True)
-        self._thread.start()
-        logger.info(f"File watcher started for {self.workspace_path}")
-    
-    def stop(self) -> None:
-        """Stop watching for file changes."""
-        self._running = False
-        if self._thread:
-            self._thread.join(timeout=2.0)
-            self._thread = None
-        logger.info("File watcher stopped")
-    
-    def __enter__(self):
-        self.start()
-        return self
-    
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.stop()
-
-class IncrementalIndexer:
-    """Manages incremental index updates based on file changes.
-    
-    Combines FileIndexer/SymbolIndexer with FileWatcher for
-    automatic index updates when files change.
-    """
-    
-    def __init__(
-        self,
-        workspace_path: str,
-        file_indexer: Optional[Any] = None,
-        symbol_indexer: Optional[Any] = None,
-        poll_interval: float = 2.0
-    ):
-        """Initialize incremental indexer.
-        
-        Args:
-            workspace_path: Root directory to index
-            file_indexer: FileIndexer instance (optional)
-            symbol_indexer: SymbolIndexer instance (optional)
-            poll_interval: Seconds between file system polls
-        """
-        self.workspace_path = os.path.abspath(workspace_path)
-        self.file_indexer = file_indexer
-        self.symbol_indexer = symbol_indexer
-        
-        self._watcher = FileWatcher(
-            workspace_path=workspace_path,
-            poll_interval=poll_interval,
-            on_change=self._handle_change
-        )
-        
-        # Statistics
-        self.changes_processed = 0
-        self.last_change_time: Optional[float] = None
-    
-    def _handle_change(self, change: FileChange) -> None:
-        """Handle a file change.
-        
-        Args:
-            change: File change to process
-        """
-        rel_path = os.path.relpath(change.path, self.workspace_path)
-        logger.debug(f"Processing {change.change_type.value}: {rel_path}")
-        
-        if change.change_type == ChangeType.DELETED:
-            # Remove from indexes
-            if self.file_indexer and hasattr(self.file_indexer, 'files'):
-                self.file_indexer.files.pop(rel_path, None)
-            if self.symbol_indexer and hasattr(self.symbol_indexer, 'symbols'):
-                self.symbol_indexer.symbols.pop(rel_path, None)
-        else:
-            # Re-index the file
-            if self.file_indexer and hasattr(self.file_indexer, 'index_file'):
-                # FileIndexer doesn't have index_file, so we skip
-                pass
-            if self.symbol_indexer and hasattr(self.symbol_indexer, 'index_file'):
-                self.symbol_indexer.index_file(change.path)
-        
-        self.changes_processed += 1
-        self.last_change_time = time.time()
-    
-    def start(self) -> None:
-        """Start incremental indexing."""
-        # Initial full index
-        if self.file_indexer:
-            self.file_indexer.index()
-        if self.symbol_indexer:
-            self.symbol_indexer.index()
-        
-        # Start watching
-        self._watcher.start()
-    
-    def stop(self) -> None:
-        """Stop incremental indexing."""
-        self._watcher.stop()
-    
-    def get_stats(self) -> Dict[str, Any]:
-        """Get indexer statistics.
-        
-        Returns:
-            Dictionary with stats
-        """
-        stats = {
-            "changes_processed": self.changes_processed,
-            "last_change_time": self.last_change_time
-        }
-        
-        if self.file_indexer:
-            stats["file_indexer"] = self.file_indexer.get_stats()
-        if self.symbol_indexer:
-            stats["symbol_indexer"] = self.symbol_indexer.get_stats()
-        
-        return stats
-    
-    def __enter__(self):
-        self.start()
-        return self
-    
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.stop()

src/praisonai-agents/praisonaiagents/context/fast/indexer/symbol_indexer.py

@@ -11,7 +11,6 @@ Extracts code symbols from source files:
 
 import os
 import re
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass, field
 from typing import Dict, List, Set, Optional, Any

src/praisonai-agents/praisonaiagents/context/fast/parallel_executor.py

@@ -11,7 +11,6 @@ Provides async parallel execution of search tools with:
 import asyncio
 from concurrent.futures import ThreadPoolExecutor
 from typing import List, Dict, Any, Optional, Callable
-import logging
 from praisonaiagents._logging import get_logger
 import time
 

src/praisonai-agents/praisonaiagents/context/fast/search_backends.py

@@ -11,7 +11,6 @@ Design principles:
 - No performance impact on default behavior
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Protocol, List, Dict, Any, Optional, runtime_checkable
 

src/praisonai-agents/praisonaiagents/context/fast/search_strategy.py

@@ -1,366 +0,0 @@
-"""
-Search Strategy for Fast Context.
-
-Orchestrates multi-turn search with:
-- Turn 1-3: Exploration (grep, glob, directory listing)
-- Turn 4: Final answer (file list + line ranges)
-- Early termination if sufficient context found
-"""
-
-import logging
-from praisonaiagents._logging import get_logger
-from dataclasses import dataclass, field
-from typing import List, Dict, Any, Optional, Set
-from enum import Enum
-
-from praisonaiagents.context.fast.result import FastContextResult, FileMatch, LineRange
-from praisonaiagents.context.fast.parallel_executor import ToolCallBatch
-
-logger = get_logger(__name__)
-
-class SearchPhase(Enum):
-    """Phases of the search strategy."""
-    DISCOVERY = "discovery"      # Turn 1: Discover structure
-    EXPLORATION = "exploration"  # Turn 2-3: Deep search
-    REFINEMENT = "refinement"    # Turn 4: Final answer
-
-@dataclass
-class SearchState:
-    """Tracks the state of a multi-turn search.
-    
-    Attributes:
-        query: Original search query
-        current_turn: Current turn number (1-indexed)
-        max_turns: Maximum turns allowed
-        phase: Current search phase
-        discovered_files: Files found so far
-        explored_paths: Paths already explored
-        result: Accumulated search result
-        should_terminate: Whether to terminate early
-    """
-    query: str
-    current_turn: int = 0
-    max_turns: int = 4
-    phase: SearchPhase = SearchPhase.DISCOVERY
-    discovered_files: Set[str] = field(default_factory=set)
-    explored_paths: Set[str] = field(default_factory=set)
-    result: FastContextResult = field(default_factory=FastContextResult)
-    should_terminate: bool = False
-    
-    # Thresholds for early termination
-    min_files_for_termination: int = 5
-    max_files_before_refinement: int = 20
-    
-    def advance_turn(self) -> None:
-        """Advance to the next turn and update phase."""
-        self.current_turn += 1
-        
-        if self.current_turn == 1:
-            self.phase = SearchPhase.DISCOVERY
-        elif self.current_turn < self.max_turns:
-            self.phase = SearchPhase.EXPLORATION
-        else:
-            self.phase = SearchPhase.REFINEMENT
-    
-    def check_early_termination(self) -> bool:
-        """Check if search should terminate early.
-        
-        Returns:
-            True if search should terminate
-        """
-        # Terminate if we have enough high-quality results
-        if len(self.discovered_files) >= self.min_files_for_termination:
-            high_relevance = sum(
-                1 for f in self.result.files 
-                if f.relevance_score >= 0.8
-            )
-            if high_relevance >= 3:
-                self.should_terminate = True
-                return True
-        
-        # Terminate if we've explored everything
-        if self.current_turn >= 2 and not self._has_unexplored_paths():
-            self.should_terminate = True
-            return True
-        
-        return False
-    
-    def _has_unexplored_paths(self) -> bool:
-        """Check if there are unexplored paths."""
-        # Simple heuristic: if we found files but haven't explored many paths
-        return len(self.discovered_files) > len(self.explored_paths) * 2
-
-class SearchStrategy:
-    """Orchestrates multi-turn search strategy.
-    
-    Implements the Fast Context search approach:
-    - Turn 1: Discovery - understand codebase structure
-    - Turn 2-3: Exploration - deep search for relevant code
-    - Turn 4: Refinement - finalize results
-    
-    Supports early termination when sufficient context is found.
-    """
-    
-    def __init__(
-        self,
-        workspace_path: str,
-        max_turns: int = 4,
-        max_parallel: int = 8
-    ):
-        """Initialize search strategy.
-        
-        Args:
-            workspace_path: Root directory for searches
-            max_turns: Maximum search turns
-            max_parallel: Maximum parallel tool calls per turn
-        """
-        self.workspace_path = workspace_path
-        self.max_turns = max_turns
-        self.max_parallel = max_parallel
-    
-    def create_state(self, query: str) -> SearchState:
-        """Create a new search state.
-        
-        Args:
-            query: Search query
-            
-        Returns:
-            New SearchState instance
-        """
-        state = SearchState(query=query, max_turns=self.max_turns)
-        state.result.query = query
-        return state
-    
-    def plan_discovery_turn(self, state: SearchState) -> ToolCallBatch:
-        """Plan the discovery turn (Turn 1).
-        
-        Focus on understanding codebase structure:
-        - List root directory
-        - Find common file patterns
-        - Initial grep for query terms
-        
-        Args:
-            state: Current search state
-            
-        Returns:
-            Batch of tool calls to execute
-        """
-        batch = ToolCallBatch(max_size=self.max_parallel)
-        query = state.query
-        
-        # List root directory
-        batch.add("list_directory", dir_path=self.workspace_path, recursive=False)
-        
-        # Find Python files (most common)
-        batch.add("glob_search", search_path=self.workspace_path, pattern="**/*.py", max_results=20)
-        
-        # Initial grep for query terms
-        query_words = query.lower().split()
-        for word in query_words[:3]:  # First 3 words
-            if len(word) >= 3:  # Skip short words
-                batch.add("grep_search", search_path=self.workspace_path, pattern=word, max_results=10)
-                if batch.is_full:
-                    break
-        
-        return batch
-    
-    def plan_exploration_turn(self, state: SearchState) -> ToolCallBatch:
-        """Plan an exploration turn (Turn 2-3).
-        
-        Deep search based on discovery results:
-        - Search in discovered directories
-        - Read promising files
-        - Follow references
-        
-        Args:
-            state: Current search state
-            
-        Returns:
-            Batch of tool calls to execute
-        """
-        batch = ToolCallBatch(max_size=self.max_parallel)
-        
-        # Read top files found so far
-        for file_match in state.result.files[:4]:
-            if file_match.path not in state.explored_paths:
-                batch.add("read_file", filepath=file_match.path, context_lines=5)
-                state.explored_paths.add(file_match.path)
-                if batch.is_full:
-                    break
-        
-        # Search in unexplored directories
-        unexplored_dirs = self._find_unexplored_dirs(state)
-        for dir_path in unexplored_dirs[:2]:
-            batch.add("list_directory", dir_path=dir_path, recursive=True, max_depth=2)
-            if batch.is_full:
-                break
-        
-        # Additional grep searches with refined patterns
-        if not batch.is_full:
-            refined_patterns = self._generate_refined_patterns(state)
-            for pattern in refined_patterns[:2]:
-                batch.add("grep_search", search_path=self.workspace_path, pattern=pattern, max_results=15)
-                if batch.is_full:
-                    break
-        
-        return batch
-    
-    def plan_refinement_turn(self, state: SearchState) -> ToolCallBatch:
-        """Plan the refinement turn (Turn 4).
-        
-        Finalize results:
-        - Read remaining important files
-        - Get specific line ranges
-        
-        Args:
-            state: Current search state
-            
-        Returns:
-            Batch of tool calls to execute
-        """
-        batch = ToolCallBatch(max_size=self.max_parallel)
-        
-        # Read top files that haven't been fully explored
-        for file_match in state.result.files[:self.max_parallel]:
-            if file_match.path not in state.explored_paths:
-                # Read specific line ranges if available
-                if file_match.line_ranges:
-                    for lr in file_match.line_ranges[:2]:
-                        batch.add(
-                            "read_file",
-                            filepath=file_match.path,
-                            start_line=max(1, lr.start - 5),
-                            end_line=lr.end + 5
-                        )
-                        if batch.is_full:
-                            break
-                else:
-                    batch.add("read_file", filepath=file_match.path)
-                
-                state.explored_paths.add(file_match.path)
-                if batch.is_full:
-                    break
-        
-        return batch
-    
-    def plan_next_turn(self, state: SearchState) -> Optional[ToolCallBatch]:
-        """Plan the next turn based on current state.
-        
-        Args:
-            state: Current search state
-            
-        Returns:
-            Batch of tool calls, or None if search should end
-        """
-        state.advance_turn()
-        
-        # Check for early termination
-        if state.should_terminate or state.current_turn > state.max_turns:
-            return None
-        
-        if state.check_early_termination():
-            return None
-        
-        # Plan based on phase
-        if state.phase == SearchPhase.DISCOVERY:
-            return self.plan_discovery_turn(state)
-        elif state.phase == SearchPhase.EXPLORATION:
-            return self.plan_exploration_turn(state)
-        else:  # REFINEMENT
-            return self.plan_refinement_turn(state)
-    
-    def process_results(
-        self,
-        state: SearchState,
-        tool_results: List[Any]
-    ) -> None:
-        """Process results from a turn and update state.
-        
-        Args:
-            state: Current search state
-            tool_results: Results from tool execution
-        """
-        for result in tool_results:
-            if isinstance(result, list):
-                # Grep or glob results
-                for item in result:
-                    if isinstance(item, dict):
-                        path = item.get("path", "")
-                        if path:
-                            state.discovered_files.add(path)
-                            
-                            # Create or update file match
-                            file_match = FileMatch(
-                                path=path,
-                                relevance_score=0.8,
-                                match_count=1
-                            )
-                            
-                            # Add line range if available
-                            line_num = item.get("line_number")
-                            if line_num:
-                                file_match.add_line_range(LineRange(
-                                    start=max(1, line_num - 2),
-                                    end=line_num + 2,
-                                    content=item.get("context"),
-                                    relevance_score=0.8
-                                ))
-                            
-                            state.result.add_file(file_match)
-            
-            elif isinstance(result, dict):
-                # Read file or list directory result
-                if result.get("success"):
-                    path = result.get("path", "")
-                    if path:
-                        state.explored_paths.add(path)
-                    
-                    # Handle directory listing
-                    entries = result.get("entries", [])
-                    for entry in entries:
-                        if not entry.get("is_dir"):
-                            state.discovered_files.add(entry.get("path", ""))
-    
-    def _find_unexplored_dirs(self, state: SearchState) -> List[str]:
-        """Find directories that haven't been explored.
-        
-        Args:
-            state: Current search state
-            
-        Returns:
-            List of unexplored directory paths
-        """
-        import os
-        
-        # Extract unique directories from discovered files
-        dirs = set()
-        for file_path in state.discovered_files:
-            dir_path = os.path.dirname(file_path)
-            if dir_path and dir_path not in state.explored_paths:
-                dirs.add(dir_path)
-        
-        return list(dirs)[:5]  # Limit to 5
-    
-    def _generate_refined_patterns(self, state: SearchState) -> List[str]:
-        """Generate refined search patterns based on results.
-        
-        Args:
-            state: Current search state
-            
-        Returns:
-            List of refined search patterns
-        """
-        patterns = []
-        query_words = state.query.lower().split()
-        
-        # Combine query words
-        if len(query_words) >= 2:
-            patterns.append(f"{query_words[0]}.*{query_words[1]}")
-        
-        # Add common code patterns
-        for word in query_words[:2]:
-            if len(word) >= 3:
-                patterns.append(f"def {word}")
-                patterns.append(f"class {word}")
-        
-        return patterns[:4]

src/praisonai-agents/praisonaiagents/context/fast/search_tools.py

@@ -15,7 +15,6 @@ import re
 import fnmatch
 from pathlib import Path
 from typing import List, Dict, Any, Optional, Set
-import logging
 from praisonaiagents._logging import get_logger
 
 logger = get_logger(__name__)

src/praisonai-agents/praisonaiagents/context/session_tracker.py

@@ -9,7 +9,6 @@ from dataclasses import dataclass, field, asdict
 from typing import Any, Dict, List, Optional
 from datetime import datetime, timezone
 import json
-import logging
 from praisonaiagents._logging import get_logger
 
 logger = get_logger(__name__)

src/praisonai-agents/praisonaiagents/context/store.py

@@ -18,7 +18,6 @@ Zero Performance Impact:
 
 import threading
 import json
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Dict, List, Any, Optional
 from dataclasses import dataclass

src/praisonai-agents/praisonaiagents/escalation/doom_loop.py

@@ -10,7 +10,6 @@ from typing import Any, Dict, List, Optional
 from enum import Enum
 import time
 import hashlib
-import logging
 from praisonaiagents._logging import get_logger
 
 logger = get_logger(__name__)

src/praisonai-agents/praisonaiagents/escalation/observability.py

@@ -5,7 +5,6 @@ Provides lightweight tracing and metrics collection for escalation execution.
 Opt-in only - no overhead when not enabled.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 import time
 from dataclasses import dataclass, field

src/praisonai-agents/praisonaiagents/escalation/pipeline.py

@@ -4,7 +4,6 @@ Escalation Pipeline for PraisonAI Agents.
 Implements progressive escalation from direct response to full autonomous mode.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 import time
 from typing import Any, Dict, List, Optional, Callable

src/praisonai-agents/praisonaiagents/eval/__init__.py

@@ -107,6 +107,12 @@ __all__ = [
     # PromptOptimizer (optimise agent.instructions against an eval, keep best)
     "PromptOptimizer",
     "OptimizeResult",
+    # Trials engine (K isolated attempts per case, pass-rate + frontier)
+    "run_trials",
+    "arun_trials",
+    "TrialScore",
+    "TrialAttempt",
+    "TrialReport",
 ]
 
 _LAZY_IMPORTS = {
@@ -191,6 +197,12 @@ _LAZY_IMPORTS = {
     # PromptOptimizer (optimise agent.instructions against an eval, keep best)
     "PromptOptimizer": ("prompt_optimizer", "PromptOptimizer"),
     "OptimizeResult": ("prompt_optimizer", "OptimizeResult"),
+    # Trials engine (K isolated attempts per case, pass-rate + frontier)
+    "run_trials": ("trials", "run_trials"),
+    "arun_trials": ("trials", "arun_trials"),
+    "TrialScore": ("trials", "TrialScore"),
+    "TrialAttempt": ("trials", "TrialAttempt"),
+    "TrialReport": ("trials", "TrialReport"),
 }
 
 

src/praisonai-agents/praisonaiagents/eval/accuracy.py

@@ -5,7 +5,6 @@ Evaluates agent output accuracy by comparing against expected output using LLM-a
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Callable, Optional, Union, TYPE_CHECKING
 

src/praisonai-agents/praisonaiagents/eval/base.py

@@ -7,7 +7,6 @@ This module provides the abstract base class that all evaluators inherit from.
 from abc import ABC, abstractmethod
 from typing import Any, Dict, Optional, Union, TYPE_CHECKING
 import uuid
-import logging
 from praisonaiagents._logging import get_logger
 
 if TYPE_CHECKING:

src/praisonai-agents/praisonaiagents/eval/criteria.py

@@ -5,7 +5,6 @@ Evaluates agent outputs against custom criteria using LLM-as-judge.
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Callable, Literal, Optional, Union, TYPE_CHECKING
 

src/praisonai-agents/praisonaiagents/eval/grader.py

@@ -9,7 +9,6 @@ that was duplicated across multiple evaluators.
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass, field, asdict
 from datetime import datetime, timezone

src/praisonai-agents/praisonaiagents/eval/judge.py

@@ -15,7 +15,6 @@ Example:
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Type, TYPE_CHECKING

src/praisonai-agents/praisonaiagents/eval/loop.py

@@ -18,7 +18,6 @@ Example:
 
 import time
 import math
-import logging
 from praisonaiagents._logging import get_logger
 from typing import TYPE_CHECKING, Optional, Callable, Any, List
 from dataclasses import dataclass

src/praisonai-agents/praisonaiagents/eval/media.py

@@ -7,7 +7,6 @@ and file-based verification for audio/video.
 """
 
 import os
-import logging
 from praisonaiagents._logging import get_logger
 from dataclasses import dataclass, field
 from typing import Any, Dict, Literal, Optional

src/praisonai-agents/praisonaiagents/eval/package.py

@@ -1,6 +1,6 @@
 """Evaluation package and case definitions."""
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional, Protocol
+from typing import Any, Callable, Dict, List, Optional, Protocol
 
 
 @dataclass
@@ -21,6 +21,7 @@ class EvalCase:
     criteria: List[str] = field(default_factory=list)
     metadata: Dict[str, Any] = field(default_factory=dict)
     timeout_seconds: float = 30.0
+    verify: Optional[Callable[[Any, Any], Any]] = None
     
     def __post_init__(self):
         if not self.name:
@@ -29,7 +30,11 @@ class EvalCase:
             raise ValueError("EvalCase must have an input")
     
     def to_dict(self) -> Dict[str, Any]:
-        """Convert to dictionary."""
+        """Convert to dictionary.
+
+        The ``verify`` callable is not serialisable and is intentionally omitted;
+        it is a live-only scorer supplied in Python.
+        """
         return {
             "name": self.name,
             "input": self.input,
@@ -62,6 +67,7 @@ class EvalResult:
     error: Optional[str] = None
     latency_ms: float = 0.0
     criteria_scores: Dict[str, float] = field(default_factory=dict)
+    record: Optional[Dict[str, Any]] = None  # Optional trajectory (messages, tool calls)
     
     def to_dict(self) -> Dict[str, Any]:
         return {
@@ -72,6 +78,7 @@ class EvalResult:
             "error": self.error,
             "latency_ms": self.latency_ms,
             "criteria_scores": self.criteria_scores,
+            "record": self.record,
         }
     
     @classmethod
@@ -85,6 +92,7 @@ class EvalResult:
             error=data.get("error"),
             latency_ms=data.get("latency_ms", 0.0),
             criteria_scores=data.get("criteria_scores", {}),
+            record=data.get("record"),
         )
 
 

src/praisonai-agents/praisonaiagents/eval/performance.py

@@ -5,7 +5,6 @@ Measures runtime and memory usage of agent executions or functions.
 """
 
 import time
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Callable, Optional, TYPE_CHECKING
 

src/praisonai-agents/praisonaiagents/eval/prompt_optimizer.py

@@ -34,9 +34,10 @@ if TYPE_CHECKING:
 
 logger = get_logger(__name__)
 
-# An eval case is (prompt, expected). ``expected`` may be None when using a
-# Judge with only criteria.
-EvalCase = Tuple[str, Any]
+# An eval pair is (prompt, expected). ``expected`` may be None when using a
+# Judge with only criteria. Named ``EvalPair`` to avoid shadowing the package
+# ``EvalCase`` dataclass (praisonaiagents.eval.package.EvalCase).
+EvalPair = Tuple[str, Any]
 
 
 @dataclass
@@ -84,7 +85,7 @@ class PromptOptimizer:
     def __init__(
         self,
         agent: "Agent",
-        evalset: List[EvalCase],
+        evalset: List[EvalPair],
         *,
         scorer: Optional[Any] = None,
         metric: Optional[Callable[[str, Any], float]] = None,

src/praisonai-agents/praisonaiagents/eval/reliability.py

@@ -4,7 +4,6 @@ Reliability evaluator for PraisonAI Agents.
 Evaluates agent reliability by verifying expected tool calls are made.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Dict, List, Optional, Set, Any, TYPE_CHECKING
 

src/praisonai-agents/praisonaiagents/eval/tokens.py

@@ -18,7 +18,6 @@ Example:
     ...     # Split into chunks
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Dict, Optional, Union
 

src/praisonai-agents/praisonaiagents/eval/trials.py

@@ -0,0 +1,487 @@
+"""Trials engine: K isolated attempts per case, unified scoring, pass-rate + frontier.
+
+This is the missing runner behind :class:`~praisonaiagents.eval.package.EvalRunnerProtocol`.
+It answers "does my agent actually work?" with statistics instead of one sampled run:
+each :class:`EvalCase` is attempted ``k`` independent times, every completed attempt is
+scored through a single :class:`TrialScore` contract, and the report gives per-case
+pass rates plus the *frontier* band (cases with 0 < pass_rate < 1) where an agent is
+capable-but-inconsistent.
+
+Design (lightweight): stdlib only (asyncio, copy, json, time). Reuses the existing
+``Judge`` and the tool-assertion logic (``ReliabilityEvaluator._extract_tool_calls``).
+No new heavy dependencies, no new trace subsystem.
+
+Example:
+    from praisonaiagents.eval import EvalPackage, EvalCase, run_trials
+
+    package = EvalPackage(name="checkout", cases=[
+        EvalCase(name="refund", input="...", expected="...", verify=my_metric_fn),
+        EvalCase(name="lookup", input="...", criteria=["cites the order id"]),
+    ])
+    report = run_trials(agent, package, k=8, concurrency=4)
+    print(report.pass_rates())   # {"refund": 0.75, "lookup": 1.0}
+    print(report.frontier())     # ["refund"]
+    report.save("trials.json")
+"""
+
+import asyncio
+import copy
+import json
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+from praisonaiagents._logging import get_logger
+
+from .package import EvalCase, EvalPackage, EvalResult
+
+logger = get_logger(__name__)
+
+
+@dataclass
+class TrialScore:
+    """Unified per-attempt scoring contract.
+
+    Attributes:
+        value: Numeric score in [0, 1].
+        passed: Whether the attempt is a pass.
+        reason: Optional human-readable explanation.
+    """
+    value: float
+    passed: bool
+    reason: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"value": self.value, "passed": self.passed, "reason": self.reason}
+
+
+def _coerce_score(raw: Any, *, threshold: float = 0.5) -> TrialScore:
+    """Normalise a ``verify`` return value (bool | float | TrialScore) to TrialScore."""
+    if isinstance(raw, TrialScore):
+        return raw
+    if isinstance(raw, bool):
+        return TrialScore(value=1.0 if raw else 0.0, passed=raw)
+    if isinstance(raw, (int, float)):
+        value = float(raw)
+        return TrialScore(value=value, passed=value >= threshold)
+    # Truthiness fallback for anything else.
+    passed = bool(raw)
+    return TrialScore(value=1.0 if passed else 0.0, passed=passed)
+
+
+@dataclass
+class TrialAttempt:
+    """One isolated attempt at a case."""
+    case_name: str
+    attempt: int
+    stop_reason: str  # "completed" | "error" | "timeout"
+    output: Optional[str] = None
+    score: Optional[TrialScore] = None
+    duration_ms: float = 0.0
+    record: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "case_name": self.case_name,
+            "attempt": self.attempt,
+            "stop_reason": self.stop_reason,
+            "output": self.output,
+            "score": self.score.to_dict() if self.score else None,
+            "duration_ms": self.duration_ms,
+            "record": self.record,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "TrialAttempt":
+        score_data = data.get("score")
+        score = TrialScore(**score_data) if score_data else None
+        return cls(
+            case_name=data["case_name"],
+            attempt=data["attempt"],
+            stop_reason=data["stop_reason"],
+            output=data.get("output"),
+            score=score,
+            duration_ms=data.get("duration_ms", 0.0),
+            record=data.get("record"),
+        )
+
+    def to_eval_result(self) -> EvalResult:
+        """Adapt to the existing ``EvalResult`` shape (single-attempt view)."""
+        return EvalResult(
+            case_name=self.case_name,
+            passed=bool(self.score and self.score.passed),
+            score=float(self.score.value) if self.score else 0.0,
+            actual_output=self.output,
+            error=None if self.stop_reason == "completed" else self.stop_reason,
+            latency_ms=self.duration_ms,
+            record=self.record,
+        )
+
+
+@dataclass
+class TrialReport:
+    """Aggregated report from a trials run: cases -> attempts, with pass rates."""
+    package_name: str
+    k: int
+    attempts: Dict[str, List[TrialAttempt]] = field(default_factory=dict)
+
+    def _scored(self, case: str) -> List[TrialAttempt]:
+        return [a for a in self.attempts.get(case, []) if a.score is not None]
+
+    def pass_rates(self) -> Dict[str, float]:
+        """Per-case pass rate over *scored* attempts (unscored excluded)."""
+        rates: Dict[str, float] = {}
+        for case in self.attempts:
+            scored = self._scored(case)
+            if not scored:
+                rates[case] = 0.0
+                continue
+            passed = sum(1 for a in scored if a.score.passed)
+            rates[case] = passed / len(scored)
+        return rates
+
+    def frontier(self) -> List[str]:
+        """Cases with pass rate strictly between 0 and 1 (capable-but-inconsistent)."""
+        return [c for c, r in self.pass_rates().items() if 0.0 < r < 1.0]
+
+    def summary(self) -> Dict[str, Any]:
+        """Stable dict for CI gating."""
+        rates = self.pass_rates()
+        per_case: Dict[str, Any] = {}
+        for case in self.attempts:
+            scored = self._scored(case)
+            n_passed = sum(1 for a in scored if a.score.passed)
+            mean_value = (
+                sum(a.score.value for a in scored) / len(scored) if scored else 0.0
+            )
+            per_case[case] = {
+                "n_attempts": len(self.attempts[case]),
+                "n_scored": len(scored),
+                "n_passed": n_passed,
+                "pass_rate": rates[case],
+                "mean_value": mean_value,
+            }
+        total_scored = sum(len(self._scored(c)) for c in self.attempts)
+        total_passed = sum(
+            1 for c in self.attempts for a in self._scored(c) if a.score.passed
+        )
+        return {
+            "package_name": self.package_name,
+            "k": self.k,
+            "n_cases": len(self.attempts),
+            "total_scored": total_scored,
+            "total_passed": total_passed,
+            "overall_pass_rate": (total_passed / total_scored) if total_scored else 0.0,
+            "frontier": self.frontier(),
+            "cases": per_case,
+        }
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "package_name": self.package_name,
+            "k": self.k,
+            "attempts": {
+                case: [a.to_dict() for a in attempts]
+                for case, attempts in self.attempts.items()
+            },
+            "summary": self.summary(),
+        }
+
+    def save(self, path: str) -> None:
+        """Persist full attempt records (including trajectory) as JSON."""
+        from pathlib import Path
+        p = Path(path)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(json.dumps(self.to_dict(), indent=2, default=str))
+
+    @classmethod
+    def load(cls, path: str) -> "TrialReport":
+        from pathlib import Path
+        data = json.loads(Path(path).read_text())
+        attempts = {
+            case: [TrialAttempt.from_dict(a) for a in rows]
+            for case, rows in data.get("attempts", {}).items()
+        }
+        return cls(
+            package_name=data["package_name"],
+            k=data.get("k", 1),
+            attempts=attempts,
+        )
+
+
+def _isolated_agent_copy(agent: Any) -> Any:
+    """Return a best-effort isolated copy of the agent for one attempt.
+
+    Each attempt should be an independent sample: fresh session/chat history,
+    caches cleared, and memory/knowledge *writes* severed so the caller's stores
+    are never mutated. Falls back to the original agent when copy is unsupported.
+    """
+    try:
+        clone = copy.copy(agent)
+    except Exception:
+        logger.debug("Agent copy failed; running attempts on the original agent")
+        return agent
+
+    # Fresh identity / session so attempts never share state.
+    if hasattr(clone, "agent_id"):
+        try:
+            clone.agent_id = str(uuid.uuid4())
+        except Exception:
+            pass
+    if hasattr(clone, "_session_id"):
+        try:
+            clone._session_id = str(uuid.uuid4())
+        except Exception:
+            pass
+
+    # Independent chat history: rebind to a *new* list so appends during the
+    # attempt never mutate the caller's shared list (a shallow copy shares it).
+    try:
+        clone.chat_history = list(getattr(agent, "chat_history", None) or [])
+    except Exception:
+        pass
+
+    # Give the clone its own cache object rather than clearing the shared one,
+    # so wiping per-attempt state cannot evict the original agent's cache.
+    cache = getattr(clone, "_system_prompt_cache", None)
+    if cache is not None:
+        try:
+            clone._system_prompt_cache = {}
+        except Exception:
+            pass
+
+    # Sever memory/knowledge writes for the attempt (measurement must not mutate stores).
+    for attr in ("memory", "knowledge"):
+        if getattr(clone, attr, None) is not None:
+            try:
+                setattr(clone, attr, None)
+            except Exception:
+                pass
+
+    return clone
+
+
+def _extract_output(response: Any) -> str:
+    return "" if response is None else str(response)
+
+
+def _extract_tool_calls(agent: Any, response: Any) -> List[str]:
+    """Reuse ReliabilityEvaluator's extraction logic for the attempt trajectory."""
+    try:
+        from .reliability import ReliabilityEvaluator
+        evaluator = ReliabilityEvaluator(agent=agent, input_text="")
+        return sorted(evaluator._extract_tool_calls(response))
+    except Exception:
+        return []
+
+
+def _run_agent_once(agent: Any, input_text: str) -> Any:
+    if hasattr(agent, "chat"):
+        return agent.chat(input_text)
+    if hasattr(agent, "start"):
+        return agent.start(input_text)
+    raise ValueError("Agent must have a 'chat' or 'start' method")
+
+
+def _score_attempt(case: EvalCase, output: str, agent: Any, response: Any) -> TrialScore:
+    """Resolve the single scorer contract for a completed attempt.
+
+    Resolution order: ``verify`` callable -> tool assertions (metadata
+    ``expected_tools``) -> ``criteria``/Judge. Runs only on completed attempts.
+    """
+    # 1. Explicit verify callable. A raising verifier is recorded as a failed
+    # score (data), never propagated — one bad metric must not abort the report.
+    if case.verify is not None:
+        try:
+            return _coerce_score(case.verify(output, case.expected))
+        except Exception as e:
+            logger.warning("verify() raised for case %r: %s", case.name, e)
+            return TrialScore(value=0.0, passed=False, reason=f"verify error: {e}")
+
+    # 2. Tool assertions from metadata (deterministic).
+    expected_tools = case.metadata.get("expected_tools") if case.metadata else None
+    if expected_tools:
+        actual = set(_extract_tool_calls(agent, response))
+        missing = [t for t in expected_tools if t not in actual]
+        passed = not missing
+        value = 1.0 - (len(missing) / len(expected_tools))
+        reason = "all expected tools called" if passed else f"missing tools: {missing}"
+        return TrialScore(value=value, passed=passed, reason=reason)
+
+    # 3. Criteria / expected via the existing Judge.
+    if case.criteria or case.expected is not None:
+        try:
+            from .judge import Judge
+            criteria = ", ".join(case.criteria) if case.criteria else None
+            result = Judge().run(
+                output=output,
+                expected=case.expected,
+                criteria=criteria,
+                input=case.input,
+            )
+            value = max(0.0, min(1.0, float(result.score) / 10.0))
+            return TrialScore(value=value, passed=bool(result.passed), reason=result.reasoning)
+        except Exception as e:  # pragma: no cover - defensive
+            logger.warning("Judge scoring failed for case %r: %s", case.name, e)
+            return TrialScore(value=0.0, passed=False, reason=f"judge error: {e}")
+
+    # Nothing to score against: treat a non-empty completion as a pass.
+    passed = bool(output)
+    return TrialScore(value=1.0 if passed else 0.0, passed=passed, reason="no scorer configured")
+
+
+async def _run_single_attempt(
+    agent: Any,
+    case: EvalCase,
+    attempt_index: int,
+    *,
+    capture_record: bool,
+    executor: Any = None,
+) -> TrialAttempt:
+    """Run one isolated attempt; failures are recorded as data, never raised.
+
+    Note on ``timeout``: the attempt runs in a worker thread via
+    :func:`asyncio.to_thread`. On timeout the *awaiting* coroutine returns
+    immediately with ``stop_reason="timeout"``, but Python cannot forcibly kill
+    the underlying thread, so a wedged agent call may keep running in the
+    background until it finishes on its own. The timeout bounds when the report
+    is produced, not necessarily when every side effect stops.
+    """
+    start = time.perf_counter()
+    isolated = _isolated_agent_copy(agent)
+    loop = asyncio.get_running_loop()
+    try:
+        response = await asyncio.wait_for(
+            loop.run_in_executor(executor, _run_agent_once, isolated, case.input),
+            timeout=case.timeout_seconds,
+        )
+    except asyncio.TimeoutError:
+        duration = (time.perf_counter() - start) * 1000.0
+        return TrialAttempt(
+            case_name=case.name, attempt=attempt_index,
+            stop_reason="timeout", duration_ms=duration,
+        )
+    except Exception as e:
+        duration = (time.perf_counter() - start) * 1000.0
+        logger.debug("Attempt %d of %r errored: %s", attempt_index, case.name, e)
+        return TrialAttempt(
+            case_name=case.name, attempt=attempt_index,
+            stop_reason="error", duration_ms=duration,
+            record={"error": str(e)} if capture_record else None,
+        )
+
+    duration = (time.perf_counter() - start) * 1000.0
+    output = _extract_output(response)
+    try:
+        score = _score_attempt(case, output, isolated, response)
+    except Exception as e:  # pragma: no cover - defensive last resort
+        logger.warning("Scoring failed for case %r: %s", case.name, e)
+        score = TrialScore(value=0.0, passed=False, reason=f"scoring error: {e}")
+
+    record = None
+    if capture_record:
+        record = {
+            "input": case.input,
+            "output": output,
+            "tool_calls": _extract_tool_calls(isolated, response),
+        }
+
+    return TrialAttempt(
+        case_name=case.name,
+        attempt=attempt_index,
+        stop_reason="completed",
+        output=output,
+        score=score,
+        duration_ms=duration,
+        record=record,
+    )
+
+
+async def arun_trials(
+    agent: Any,
+    package: EvalPackage,
+    *,
+    k: int = 1,
+    concurrency: int = 1,
+    capture_record: bool = True,
+) -> TrialReport:
+    """Async: run ``k`` isolated attempts per case with bounded concurrency.
+
+    Args:
+        agent: The agent under test (needs ``chat`` or ``start``).
+        package: The :class:`EvalPackage` of cases.
+        k: Independent attempts per case.
+        concurrency: Max attempts running concurrently (semaphore-bounded).
+        capture_record: Store the per-attempt trajectory (input/output/tool_calls).
+
+    Returns:
+        A :class:`TrialReport` with deterministic attempt ordering.
+    """
+    if k < 1:
+        raise ValueError("k must be >= 1")
+    semaphore = asyncio.Semaphore(max(1, concurrency))
+
+    # Dedicated executor sized to the concurrency budget. It is *not* joined on
+    # exit: a timed-out worker thread cannot be killed, and blocking on it would
+    # defeat the per-case timeout. Abandoning it lets ``arun_trials`` return
+    # promptly while the OS reclaims the leaked thread when it eventually ends.
+    import concurrent.futures
+
+    executor = concurrent.futures.ThreadPoolExecutor(
+        max_workers=max(1, concurrency)
+    )
+
+    async def _bounded(case: EvalCase, idx: int) -> TrialAttempt:
+        async with semaphore:
+            return await _run_single_attempt(
+                agent, case, idx, capture_record=capture_record, executor=executor
+            )
+
+    report = TrialReport(package_name=package.name, k=k)
+    try:
+        for case in package.cases:
+            tasks = [_bounded(case, i) for i in range(k)]
+            results = await asyncio.gather(*tasks)
+            # Stable, deterministic assembly order by attempt index.
+            results = sorted(results, key=lambda a: a.attempt)
+            # Cases can legitimately share a name (e.g. same case re-run):
+            # append instead of overwriting so no attempts are silently dropped.
+            existing = report.attempts.setdefault(case.name, [])
+            offset = len(existing)
+            for r in results:
+                r.attempt += offset
+            existing.extend(results)
+    finally:
+        # Do not wait for potentially-wedged workers (see note above).
+        executor.shutdown(wait=False)
+    return report
+
+
+def run_trials(
+    agent: Any,
+    package: EvalPackage,
+    *,
+    k: int = 1,
+    concurrency: int = 1,
+    capture_record: bool = True,
+) -> TrialReport:
+    """Sync wrapper around :func:`arun_trials` (see it for arguments)."""
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        return asyncio.run(
+            arun_trials(agent, package, k=k, concurrency=concurrency,
+                        capture_record=capture_record)
+        )
+    # Already inside an event loop: run on a dedicated loop in a worker thread.
+    import concurrent.futures
+
+    def _runner() -> TrialReport:
+        return asyncio.run(
+            arun_trials(agent, package, k=k, concurrency=concurrency,
+                        capture_record=capture_record)
+        )
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
+        return pool.submit(_runner).result()

src/praisonai-agents/praisonaiagents/eval/utils.py

@@ -3,7 +3,6 @@ Utility functions for PraisonAI Agents evaluation framework.
 """
 
 import json
-import logging
 from praisonaiagents._logging import get_logger
 from pathlib import Path
 from typing import Any, Dict, Optional, Union

src/praisonai-agents/praisonaiagents/gateway/__init__.py

@@ -70,6 +70,10 @@ from .protocols import (
     GatewayConcurrencyPolicyProtocol,
     GatewayConcurrencyPolicy,  # backward-compat alias
     ConcurrencyLimitPolicy,
+    # Gateway resource-pressure admission (Issue #3445)
+    ResourceSample,
+    ResourcePressurePolicyProtocol,
+    MemoryPressurePolicy,
     # Gateway rate-limit admission (Issue #2532)
     RateLimitDecision,
     RateLimitPolicyProtocol,
@@ -137,6 +141,11 @@ from .protocols import (
     is_weak_secret,
     assert_gateway_secret_strong,
 )
+from .liveness import (
+    # Event-loop liveness watchdog (Issue #3385)
+    LoopWatchdogPolicy,
+    LoopWatchdog,
+)
 from .hooks import (
     HookAction,
     HookConfig,
@@ -158,6 +167,10 @@ from .config import (
     DeliveryConfig,
     PollingConfig,
     LivenessConfig,
+    # Hot-reload registry (Issue #3378)
+    HOT_APPLIABLE_KEYS,
+    SupportsHotReload,
+    is_hot_appliable,
 )
 
 # Lazy loading cache
@@ -268,6 +281,10 @@ __all__ = [
     "GatewayConcurrencyPolicyProtocol",
     "GatewayConcurrencyPolicy",
     "ConcurrencyLimitPolicy",
+    # Gateway resource-pressure admission (Issue #3445)
+    "ResourceSample",
+    "ResourcePressurePolicyProtocol",
+    "MemoryPressurePolicy",
     # Gateway rate-limit admission (Issue #2532)
     "RateLimitDecision",
     "RateLimitPolicyProtocol",
@@ -334,6 +351,9 @@ __all__ = [
     "WeakGatewaySecretError",
     "is_weak_secret",
     "assert_gateway_secret_strong",
+    # Event-loop liveness watchdog (Issue #3385)
+    "LoopWatchdogPolicy",
+    "LoopWatchdog",
     # Inbound trigger / webhook contract (Issue #2281)
     "HookAction",
     "HookConfig",
@@ -353,6 +373,10 @@ __all__ = [
     "DeliveryConfig",
     "PollingConfig",
     "LivenessConfig",
+    # Hot-reload registry (Issue #3378)
+    "HOT_APPLIABLE_KEYS",
+    "SupportsHotReload",
+    "is_hot_appliable",
     # Implementations (lazy loaded from praisonai wrapper)
     "WebSocketGateway",
     "GatewaySession",

src/praisonai-agents/praisonaiagents/gateway/config.py

@@ -5,7 +5,53 @@ Provides configuration dataclasses for gateway and session settings.
 """
 
 from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Mapping, Optional, Protocol, Set, runtime_checkable
+
+
+# ---------------------------------------------------------------------------
+# Hot-reload registry (Issue #3378)
+# ---------------------------------------------------------------------------
+#
+# Closed set of dotted config paths that are safe to apply *in place* on a
+# running gateway without restarting channels or agents. Anything not listed
+# here keeps falling through to the existing restart plans, so restart stays
+# the safe default for unknown/structural changes (fail-safe).
+#
+# This is a pure protocol/registry with no heavy imports; the authoritative
+# classification lives in core so every runtime reloads identically, while the
+# wrapper/bot gateway server only implements the in-place ``apply_hot_reload``.
+HOT_APPLIABLE_KEYS: "frozenset[str]" = frozenset({
+    "gateway.logging.level",
+    "gateway.drain_timeout",
+    "gateway.reload_drain_timeout",
+})
+
+
+def is_hot_appliable(path: str) -> bool:
+    """Return whether a dotted config ``path`` can be applied without restart.
+
+    A change is hot-appliable when the path itself is registered, or when it is
+    a leaf *under* a registered key (e.g. ``gateway.logging.level.extra``).
+    Callers should treat every other path as requiring a restart plan.
+    """
+    if path in HOT_APPLIABLE_KEYS:
+        return True
+    return any(path.startswith(key + ".") for key in HOT_APPLIABLE_KEYS)
+
+
+@runtime_checkable
+class SupportsHotReload(Protocol):
+    """Protocol a gateway implements to apply hot-reloadable config in place.
+
+    The gateway calls :meth:`apply_hot_reload` with the subset of changed paths
+    classified as hot-appliable (see :data:`HOT_APPLIABLE_KEYS`) and the newly
+    loaded config, mutating the relevant live subsystems without a restart.
+    """
+
+    def apply_hot_reload(
+        self, paths: Set[str], new_config: Mapping[str, Any]
+    ) -> None:
+        ...
 
 
 @dataclass
@@ -17,6 +63,9 @@ class SessionConfig:
         max_messages: Maximum messages to keep in history (0 = unlimited)
         persist: Whether to persist session state
         persist_path: Path for session persistence
+        store: Persistence backend when ``persist`` is set — ``"sqlite"``
+            (default: transcripts in a WAL SQLite DB with concurrent readers
+            and indexed lookups) or ``"file"`` (legacy per-session JSON files)
         resume_window: How long (seconds) a session stays resumable after disconnect
         max_inbox: Maximum queued messages per session (0 = unlimited, default 256)
         metadata: Additional session metadata
@@ -27,6 +76,7 @@ class SessionConfig:
     max_messages: int = 1000
     persist: bool = False
     persist_path: Optional[str] = None
+    store: str = "sqlite"  # "sqlite" (concurrent, indexed) | "file" (legacy JSON)
     resume_window: int = 86400  # 24 hours default
     max_inbox: int = 256  # Default bounded queue size
     metadata: Dict[str, Any] = field(default_factory=dict)
@@ -44,6 +94,10 @@ class SessionConfig:
             raise ValueError("max_messages must be >= 0")
         if self.resume_window < 0:
             raise ValueError("resume_window must be >= 0")
+        if self.store not in ("sqlite", "file"):
+            raise ValueError(
+                f"Invalid session store {self.store!r}; expected 'sqlite' or 'file'"
+            )
     
     def to_dict(self) -> Dict[str, Any]:
         """Convert to dictionary."""
@@ -52,6 +106,7 @@ class SessionConfig:
             "max_messages": self.max_messages,
             "persist": self.persist,
             "persist_path": self.persist_path,
+            "store": self.store,
             "resume_window": self.resume_window,
             "max_inbox": self.max_inbox,
             "metadata": self.metadata,
@@ -639,6 +694,7 @@ class MultiChannelGatewayConfig:
                     max_messages=sc_data.get("max_messages", 1000),
                     persist=sc_data.get("persist", False),
                     persist_path=sc_data.get("persist_path"),
+                    store=sc_data.get("store", "sqlite"),
                     resume_window=sc_data.get("resume_window", 86400),
                     max_inbox=sc_data.get("max_inbox", 256),
                     metadata=sc_data.get("metadata", {}),

src/praisonai-agents/praisonaiagents/gateway/liveness.py

@@ -0,0 +1,268 @@
+"""Event-loop liveness watchdog for long-lived gateway loops (Issue #3385).
+
+A gateway process can stay *alive* while its asyncio event loop is **wedged** —
+deadlocked, blocked on a synchronous call inside async context, or stuck on an
+``await`` that never returns. When that happens every asyncio-based recovery
+path (drain, scale-to-zero, reconnect supervision, the ``/live`` HTTP probe)
+is structurally unable to fire, because they all run *on* the frozen loop.
+Ordinary process supervisors only restart a process that has **exited**, so a
+wedged gateway becomes a silent zombie that holds its sockets and answers no
+messages until a human kills it.
+
+This module provides a small, dependency-free (pure stdlib) primitive that runs
+on a dedicated OS thread — so it keeps working precisely when the loop does
+not. It periodically probes the loop via ``loop.call_soon_threadsafe`` and
+measures scheduling lag; after *N* consecutive missed probes it declares the
+loop wedged, optionally dumps all-thread stacks via :mod:`faulthandler`, and
+forces a clean hand-back to the supervisor via ``os._exit`` (bypassing
+``Py_FinalizeEx``, which would itself hang joining the stuck threads).
+
+Design principles:
+
+* **Opt-in.** Nothing arms the watchdog unless an embedder calls :meth:`arm`.
+* **Fail-open.** Any error inside the watchdog must never wedge or kill a
+  healthy gateway. Once armed, it only ever *reacts* to a confirmed stall.
+* **No heavy deps.** Only :mod:`threading`, :mod:`faulthandler`, :mod:`os`,
+  :mod:`sys`, and :mod:`time` from the stdlib.
+
+The primitive lives in core so every embedder — not just the CLI gateway — can
+opt into the same liveness/self-recovery contract around any long-lived loop.
+Deployment wiring (systemd ``Type=notify`` / ``WatchdogSec``, CLI flags) is a
+wrapper concern and intentionally not part of this module.
+"""
+
+from __future__ import annotations
+
+import faulthandler
+import math
+import os
+import sys
+import threading
+import time
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Literal, Optional
+
+if TYPE_CHECKING:  # pragma: no cover - typing only
+    import asyncio
+
+# Reuse the restart-intent exit-code protocol (Issue #2437): EX_TEMPFAIL (75)
+# tells the supervisor "transient failure — please restart".
+from .protocols import GATEWAY_RESTART_EXIT_CODE
+
+__all__ = [
+    "LoopWatchdogPolicy",
+    "LoopWatchdog",
+]
+
+
+@dataclass
+class LoopWatchdogPolicy:
+    """Configuration for :class:`LoopWatchdog`.
+
+    Attributes:
+        probe_interval_s: Seconds between liveness probes scheduled onto the
+            loop. Also the cadence at which the watchdog thread wakes.
+        missed_probes_before_wedged: Number of *consecutive* probes that must
+            fail to round-trip within their budget before the loop is declared
+            wedged. With the defaults (5s x 3) a stall of ~15s trips recovery.
+        on_wedge: What to do once a wedge is confirmed. ``"dump_and_exit"``
+            dumps all-thread stacks then calls ``os._exit(exit_code)``;
+            ``"dump_only"`` dumps stacks but leaves the process running (useful
+            for observation/testing).
+        exit_code: Process exit code used when ``on_wedge == "dump_and_exit"``.
+            Defaults to EX_TEMPFAIL (75) so a supervisor restarts the process.
+        dump_file: Optional path to also write the stack dump to (in addition
+            to stderr). ``None`` writes to stderr only.
+    """
+
+    probe_interval_s: float = 5.0
+    missed_probes_before_wedged: int = 3
+    on_wedge: Literal["dump_and_exit", "dump_only"] = "dump_and_exit"
+    exit_code: int = GATEWAY_RESTART_EXIT_CODE
+    dump_file: Optional[str] = None
+
+    def __post_init__(self) -> None:
+        if not math.isfinite(self.probe_interval_s) or self.probe_interval_s <= 0:
+            raise ValueError("probe_interval_s must be a finite value > 0")
+        if self.missed_probes_before_wedged < 1:
+            raise ValueError("missed_probes_before_wedged must be >= 1")
+        if self.on_wedge not in ("dump_and_exit", "dump_only"):
+            raise ValueError(
+                "on_wedge must be 'dump_and_exit' or 'dump_only'"
+            )
+
+    @property
+    def wedge_after_s(self) -> float:
+        """Approximate stall duration (seconds) before a wedge is declared."""
+        return self.probe_interval_s * self.missed_probes_before_wedged
+
+
+class LoopWatchdog:
+    """OS-thread liveness probe for an asyncio loop. Fail-open by design.
+
+    Typical use, wrapping any long-lived loop::
+
+        watchdog = LoopWatchdog()
+        watchdog.arm(loop)
+        try:
+            loop.run_forever()
+        finally:
+            watchdog.disarm()
+
+    The watchdog thread schedules a no-op callback onto ``loop`` every
+    ``probe_interval_s`` seconds via ``loop.call_soon_threadsafe``. A live loop
+    runs the callback promptly, resetting the missed-probe counter. A wedged
+    loop never runs it; after ``missed_probes_before_wedged`` consecutive
+    misses the watchdog reacts per :attr:`LoopWatchdogPolicy.on_wedge`.
+    """
+
+    def __init__(self, policy: Optional[LoopWatchdogPolicy] = None) -> None:
+        self.policy = policy or LoopWatchdogPolicy()
+        self._loop: "Optional[asyncio.AbstractEventLoop]" = None
+        self._thread: Optional[threading.Thread] = None
+        self._stop = threading.Event()
+        # Set by the loop each time a probe runs; compared by the watchdog
+        # thread to detect misses. Guarded by the GIL for these simple stores.
+        self._last_ack = 0.0
+        self._last_probe = 0.0
+        self._wedged = False
+
+    @property
+    def armed(self) -> bool:
+        return self._thread is not None and self._thread.is_alive()
+
+    @property
+    def wedged(self) -> bool:
+        """True once a wedge has been confirmed (mainly for ``dump_only``)."""
+        return self._wedged
+
+    def arm(self, loop: "asyncio.AbstractEventLoop") -> None:
+        """Start watching ``loop`` from a dedicated daemon OS thread.
+
+        Idempotent: calling :meth:`arm` while already armed is a no-op. Any
+        failure to start the thread is swallowed (fail-open) — a watchdog that
+        cannot start must never prevent a healthy gateway from running.
+        """
+        if self.armed:
+            return
+        try:
+            self._loop = loop
+            self._stop.clear()
+            self._wedged = False
+            now = time.monotonic()
+            self._last_ack = now
+            self._last_probe = now
+            self._thread = threading.Thread(
+                target=self._run,
+                name="praisonai-loop-watchdog",
+                daemon=True,
+            )
+            self._thread.start()
+        except Exception:  # pragma: no cover - fail open
+            self._thread = None
+            self._loop = None
+
+    def disarm(self) -> None:
+        """Stop watching. Safe to call multiple times / when not armed."""
+        self._stop.set()
+        thread = self._thread
+        if thread is not None and thread is not threading.current_thread():
+            try:
+                thread.join(timeout=self.policy.probe_interval_s + 1.0)
+            except Exception:  # pragma: no cover - fail open
+                pass
+        self._thread = None
+        self._loop = None
+
+    def _ack(self) -> None:
+        """Callback run *on the loop*; records that the loop is alive."""
+        self._last_ack = time.monotonic()
+
+    def _schedule_probe(self) -> bool:
+        """Schedule an ack onto the loop. Returns False if scheduling fails."""
+        loop = self._loop
+        if loop is None:
+            return False
+        try:
+            loop.call_soon_threadsafe(self._ack)
+            self._last_probe = time.monotonic()
+            return True
+        except RuntimeError:
+            # Loop is closed / not running — treat as not-scheduled but do not
+            # declare a wedge; a closed loop is a normal shutdown, not a hang.
+            return False
+        except Exception:  # pragma: no cover - fail open
+            return False
+
+    def _run(self) -> None:
+        interval = self.policy.probe_interval_s
+        threshold = self.policy.missed_probes_before_wedged
+        missed = 0
+        while not self._stop.is_set():
+            scheduled = self._schedule_probe()
+            # Wait one interval for the loop to run our ack.
+            if self._stop.wait(interval):
+                break
+            if not scheduled:
+                # Could not schedule (loop closed / gone). Reset and keep
+                # failing open rather than tripping on a normal shutdown.
+                missed = 0
+                continue
+            if self._last_ack >= self._last_probe:
+                # Loop ran the ack within the budget: healthy.
+                missed = 0
+                continue
+            # The loop did not run the ack within this interval.
+            missed += 1
+            if missed >= threshold and not self._wedged:
+                self._on_wedge()
+                if self.policy.on_wedge == "dump_only":
+                    # Keep observing but do not re-fire repeatedly.
+                    missed = 0
+
+    def _on_wedge(self) -> None:
+        self._wedged = True
+        try:
+            self._dump_stacks()
+        except Exception:  # pragma: no cover - fail open
+            pass
+        if self.policy.on_wedge == "dump_and_exit":
+            # If disarm() raced in while we were dumping stacks, honour the
+            # intentional shutdown and do not terminate the process — a
+            # deliberate disarm must never trigger a supervisor restart.
+            if self._stop.is_set():
+                return
+            # Bypass Py_FinalizeEx: normal interpreter shutdown would try to
+            # join the stuck loop thread and hang. os._exit hands the process
+            # straight back to the supervisor.
+            try:
+                sys.stderr.flush()
+            except Exception:  # pragma: no cover - fail open
+                pass
+            os._exit(self.policy.exit_code)
+
+    def _dump_stacks(self) -> None:
+        """Dump all-thread stacks to stderr and, if configured, a file."""
+        stall = self.policy.wedge_after_s
+        header = (
+            f"praisonai-loop-watchdog: event loop wedged "
+            f"(no progress for ~{stall:.0f}s); dumping all-thread stacks\n"
+        )
+        try:
+            sys.stderr.write(header)
+        except Exception:  # pragma: no cover - fail open
+            pass
+        try:
+            faulthandler.dump_traceback(file=sys.stderr, all_threads=True)
+        except Exception:  # pragma: no cover - fail open
+            pass
+        dump_file = self.policy.dump_file
+        if dump_file:
+            try:
+                # Line-buffered append; keep the fd open across the dump.
+                with open(dump_file, "a", encoding="utf-8") as fh:
+                    fh.write(header)
+                    fh.flush()
+                    faulthandler.dump_traceback(file=fh, all_threads=True)
+            except Exception:  # pragma: no cover - fail open
+                pass

src/praisonai-agents/praisonaiagents/gateway/protocols.py

@@ -2928,6 +2928,135 @@ class ConcurrencyLimitPolicy:
 GatewayConcurrencyPolicy = GatewayConcurrencyPolicyProtocol
 
 
+# ---------------------------------------------------------------------------
+# Gateway resource-pressure admission (Issue #3445)
+#
+# Admission today is concurrency/CPU-scaled and blind to memory; on a small
+# always-on host a burst of concurrent turns drives RSS up until the OOM
+# killer fires — the failure a $5 box hits first. This adds a pure, import-
+# free decision that maps a resource *sample* (current RSS) onto the existing
+# :class:`AdmissionDecision` so the gate can queue under soft pressure and
+# shed under hard pressure *before* the process is killed. It reuses the
+# admission seam (no new subsystem) and sits beside the concurrency / rate-
+# limit / scale-to-zero policy family. The live sampler (reading
+# ``resource.getrusage`` / optional ``psutil``) and the wiring into the gate
+# live in the wrapper; this contract keeps the decision testable in isolation.
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class ResourceSample:
+    """A point-in-time snapshot of the gateway process's resource usage.
+
+    Attributes:
+        rss_mb: Resident set size in mebibytes, or ``None`` when the platform
+            cannot report it (the policy then admits and self-disables so the
+            monitor never crashes the gateway it protects).
+    """
+
+    rss_mb: Optional[float] = None
+
+
+@runtime_checkable
+class ResourcePressurePolicyProtocol(Protocol):
+    """Protocol for memory/resource-aware admission decisions.
+
+    Pure, import-free decision contract consumed by the wrapper's admission
+    gate. The wrapper samples its own resource usage on a lightweight cadence
+    and hands the policy a :class:`ResourceSample`; the policy returns an
+    :class:`AdmissionDecision` — ``ADMIT`` below the soft threshold, ``QUEUE``
+    to apply backpressure above it, and ``REJECT`` above the hard threshold so
+    the process sheds load before the OOM killer fires. Sampling and
+    enforcement (the ``asyncio.Semaphore`` ceiling and bounded wait queue)
+    live in the wrapper; this keeps the *decision* provable in isolation,
+    symmetric with :class:`GatewayConcurrencyPolicyProtocol`.
+
+    A config-driven default (:class:`MemoryPressurePolicy`) is provided for
+    the common "soft/hard RSS threshold" case.
+    """
+
+    def evaluate(self, sample: ResourceSample) -> AdmissionDecision:
+        """Return an :class:`AdmissionDecision` for the supplied sample."""
+        ...
+
+
+class MemoryPressurePolicy:
+    """Config-driven RSS-threshold resource-pressure policy.
+
+    The default wired by the ``max_rss_mb`` gateway config key
+    (``BotOS(max_rss_mb=...)`` / ``gateway.yaml``) and the
+    ``AdmissionGate(resource_policy=...)`` Python surface. It is intentionally
+    minimal and dependency-free so the decision lives in core and is provable
+    in isolation; the wrapper owns the live sampler and the side effects
+    (queue / shed / busy ack).
+
+    The decision, given a sample's ``rss_mb``:
+
+    * ``ADMIT`` while ``rss_mb < soft_rss_mb`` (or the platform can't report
+      memory, so ``rss_mb is None`` — never block on a missing signal).
+    * ``QUEUE`` (apply backpressure) while ``soft_rss_mb <= rss_mb <
+      hard_rss_mb``.
+    * ``REJECT`` (shed with a busy ack) while ``rss_mb >= hard_rss_mb``.
+
+    A ``hard_rss_mb`` of ``0`` disables pressure-based shedding entirely
+    (every sample admits) — the legacy default when no threshold is set.
+
+    Example::
+
+        MemoryPressurePolicy(soft_rss_mb=400, hard_rss_mb=550)
+    """
+
+    def __init__(
+        self,
+        soft_rss_mb: float = 0.0,
+        hard_rss_mb: float = 0.0,
+    ):
+        try:
+            soft = float(soft_rss_mb or 0.0)
+        except (TypeError, ValueError) as err:
+            raise ValueError(
+                f"soft_rss_mb must be a number, got {soft_rss_mb!r}"
+            ) from err
+        try:
+            hard = float(hard_rss_mb or 0.0)
+        except (TypeError, ValueError) as err:
+            raise ValueError(
+                f"hard_rss_mb must be a number, got {hard_rss_mb!r}"
+            ) from err
+        if soft < 0:
+            raise ValueError(f"soft_rss_mb must be >= 0, got {soft_rss_mb!r}")
+        if hard < 0:
+            raise ValueError(f"hard_rss_mb must be >= 0, got {hard_rss_mb!r}")
+        # A soft threshold above the hard one would queue turns that should be
+        # shed; fail fast rather than silently invert the pressure ladder.
+        if soft and hard and soft > hard:
+            raise ValueError(
+                f"soft_rss_mb ({soft_rss_mb!r}) must be <= "
+                f"hard_rss_mb ({hard_rss_mb!r})"
+            )
+        self.soft_rss_mb = soft
+        self.hard_rss_mb = hard
+
+    @property
+    def enabled(self) -> bool:
+        """Whether pressure-based admission is active (a threshold is set)."""
+        return self.soft_rss_mb > 0 or self.hard_rss_mb > 0
+
+    def evaluate(self, sample: ResourceSample) -> AdmissionDecision:
+        # Disabled, or the platform can't report memory: never block on a
+        # missing/absent signal — admit and let concurrency limits apply.
+        if not self.enabled:
+            return AdmissionDecision.ADMIT
+        rss = getattr(sample, "rss_mb", None)
+        if rss is None:
+            return AdmissionDecision.ADMIT
+        if self.hard_rss_mb and rss >= self.hard_rss_mb:
+            return AdmissionDecision.REJECT
+        if self.soft_rss_mb and rss >= self.soft_rss_mb:
+            return AdmissionDecision.QUEUE
+        return AdmissionDecision.ADMIT
+
+
 # ---------------------------------------------------------------------------
 # Gateway rate-limit admission (Issue #2532)
 #

src/praisonai-agents/praisonaiagents/guardrails/llm_guardrail.py

@@ -5,7 +5,6 @@ This module provides LLM-powered guardrails that can validate task outputs
 using natural language descriptions, similar to CrewAI's implementation.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 from typing import Any, Tuple, Union, Optional, Dict
 from pydantic import BaseModel

src/praisonai-agents/praisonaiagents/hooks/events.py

@@ -360,6 +360,30 @@ class GatewayStopInput(HookInput):
         return base
 
 
+@dataclass
+class CliBackendExecuteInput(HookInput):
+    """Input for CLI_BACKEND_EXECUTE hooks (agent delegated a turn to a CLI backend)."""
+    backend: str = ""
+    command: Optional[List[Any]] = None
+    content: Optional[str] = None
+    error: Optional[str] = None
+    transport: str = "subprocess"
+    praisonai_llm_http: bool = False
+
+    def to_dict(self) -> Dict[str, Any]:
+        from ..cli_backend.debug import redact_command
+        base = super().to_dict()
+        base.update({
+            "backend": self.backend,
+            "command": redact_command(self.command),
+            "content": self.content[:500] if self.content else None,
+            "error": self.error,
+            "transport": self.transport,
+            "praisonai_llm_http": self.praisonai_llm_http,
+        })
+        return base
+
+
 @dataclass
 class ScheduleTriggerInput(HookInput):
     """Input for SCHEDULE_TRIGGER hooks (a scheduled job fired)."""

src/praisonai-agents/praisonaiagents/hooks/registry.py

@@ -4,7 +4,6 @@ Hook Registry for PraisonAI Agents.
 Manages registration and lookup of hooks for different events.
 """
 
-import logging
 import threading
 from praisonaiagents._logging import get_logger
 from typing import Dict, List, Optional, Callable, Union

src/praisonai-agents/praisonaiagents/hooks/runner.py

@@ -9,7 +9,6 @@ import os
 import json
 import time
 import asyncio
-import logging
 from praisonaiagents._logging import get_logger
 import subprocess
 from typing import List, Optional, Dict, Any

src/praisonai-agents/praisonaiagents/hooks/types.py

@@ -36,6 +36,12 @@ class HookEvent(str, Enum):
     # LLM lifecycle
     BEFORE_LLM = "before_llm"
     AFTER_LLM = "after_llm"
+
+    # Prompt-cache stability (advisory): fired when a turn's cached prompt
+    # prefix (model + tool schemas + system-prompt fingerprint) changes from
+    # the previous turn, so a long-lived conversation's provider prompt cache
+    # will miss. Payload carries the old/new signature and a reason.
+    PROMPT_PREFIX_INVALIDATED = "prompt_prefix_invalidated"
     
     # Error handling
     ON_ERROR = "on_error"
@@ -70,6 +76,9 @@ class HookEvent(str, Enum):
     SCHEDULE_REMOVE = "schedule_remove"
     SCHEDULE_TRIGGER = "schedule_trigger"
 
+    # CLI backend delegation (subprocess, not LiteLLM HTTP)
+    CLI_BACKEND_EXECUTE = "cli_backend_execute"
+
     # Background job lifecycle
     JOB_COMPLETED = "job_completed"  # A background job finished (ok or error)
 

src/praisonai-agents/praisonaiagents/knowledge/adapters/mem0_adapter.py

@@ -7,7 +7,6 @@ Handles all normalization to ensure metadata is ALWAYS a dict (never None).
 LAZY IMPORT: mem0 is only imported when this adapter is instantiated.
 """
 
-import logging
 from praisonaiagents._logging import get_logger
 import os
 from typing import Any, Dict, List, Optional