Compare commits

..
Author SHA1 Message Date
praisonai-triage-agent[bot]andMervin Praison 85ba558a3b fix: real MCP readiness check in example + drop unused SseServerTransport
- simple-mcp-server.py: replace blind 2s sleep with TCP readiness poll so
  the smoke test genuinely verifies the server bound and exits non-zero on
  failure (addresses Greptile P1: daemon exited claiming false success)
- execution_mixin.py: remove now-unused SseServerTransport import/instance
  after switching to FastMCP.sse_app() (dead code flagged by Greptile)

Co-authored-by: Mervin Praison <[email protected]>
2026-07-24 08:57:52 +00:00
Dhivya-Bharathy 1cfb2bc26f fix: resolve 4 remaining E2E example failures after batch-3 merge 2026-07-24 14:25:40 +05:30
89 changed files with 1084 additions and 7216 deletions

No files matched your search

-3
View File
@@ -782,9 +782,6 @@ 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
+30 -1
View File
@@ -1,5 +1,34 @@
import socket
import sys
import threading
import time
from praisonaiagents import Agent
def _wait_until_ready(host: str, port: int, timeout: float = 15.0) -> bool:
"""Poll the MCP server port until it accepts connections or times out."""
deadline = time.time() + timeout
while time.time() < deadline:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(1.0)
if sock.connect_ex((host, port)) == 0:
return True
time.sleep(0.25)
return False
if __name__ == "__main__":
host, port = "127.0.0.1", 8080
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=port, protocol="mcp", path="/mcp"),
daemon=True,
)
server.start()
if _wait_until_ready(host, port):
print(f"MCP server started on http://{host}:{port}/mcp")
else:
print(f"MCP server failed to start on http://{host}:{port}/mcp", file=sys.stderr)
sys.exit(1)
@@ -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"
)
@@ -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.")
+4 -2
View File
@@ -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!")
+13 -13
View File
@@ -18,38 +18,38 @@ This report compares **Python SDK feature categories** against **Python document
| Category | Features | Docs | Lines |
|----------|----------|------|-------|
| ✅ AGUI | 1 | 2 | 420 |
| ✅ Agent | 21 | 46 | 14657 |
| ✅ Agent | 21 | 45 | 14186 |
| ✅ Agent-to-Agent (A2A) | 1 | 7 | 2965 |
| ✅ Agent-to-User (A2U) | 1 | 3 | 585 |
| ✅ Approval | 1 | 5 | 2312 |
| ✅ Approval | 1 | 5 | 2286 |
| ✅ Audio | 2 | 12 | 1000 |
| ✅ Auto Generation | 5 | 10 | 2880 |
| ✅ Auto Generation | 5 | 10 | 2868 |
| ✅ Autonomy | 3 | 7 | 2707 |
| ✅ Bots | 7 | 29 | 11795 |
| ✅ Bots | 7 | 28 | 11301 |
| ✅ Budget | 1 | 1 | 287 |
| ✅ CLI | 5 | 119 | 34087 |
| ✅ CLI | 5 | 119 | 33852 |
| ✅ Chunking | 2 | 2 | 385 |
| ✅ Citations | 2 | 1 | 202 |
| ✅ Code Execution | 2 | 13 | 4031 |
| ✅ Code Execution | 2 | 13 | 3970 |
| ✅ Conditions | 1 | 3 | 1179 |
| ✅ Configuration | 3 | 7 | 2772 |
| ✅ Context Management | 16 | 34 | 13204 |
| ✅ Deep Research | 8 | 2 | 587 |
| ✅ Display | 6 | 3 | 741 |
| ✅ Embeddings | 6 | 23 | 2049 |
| ✅ Evaluation | 1 | 7 | 2905 |
| ✅ Evaluation | 1 | 7 | 2902 |
| ✅ Events | 1 | 2 | 749 |
| ✅ Execution | 3 | 4 | 1528 |
| ✅ Failover | 2 | 1 | 400 |
| ✅ Files | 2 | 6 | 1964 |
| ✅ Flow | 1 | 3 | 777 |
| ✅ Gateway | 7 | 56 | 20195 |
| ✅ Gateway | 7 | 54 | 19347 |
| ✅ Guardrails | 4 | 4 | 1673 |
| ✅ Handoffs | 11 | 6 | 2454 |
| ✅ Hooks | 2 | 9 | 4128 |
| ✅ Hooks | 2 | 9 | 4124 |
| ✅ Image | 1 | 11 | 1255 |
| ✅ Knowledge | 4 | 13 | 4160 |
| ✅ LLM | 3 | 14 | 4864 |
| ✅ LLM | 3 | 14 | 4830 |
| ✅ Loops | 4 | 4 | 1202 |
| ✅ MCP | 1 | 58 | 13322 |
| ✅ Memory | 6 | 17 | 6635 |
@@ -60,7 +60,7 @@ This report compares **Python SDK feature categories** against **Python document
| ✅ Parallel Execution | 3 | 2 | 497 |
| ✅ Planning | 6 | 6 | 1545 |
| ✅ Plugins | 8 | 3 | 1900 |
| ✅ Prompts | 2 | 9 | 1913 |
| ✅ Prompts | 2 | 9 | 1910 |
| ✅ Providers | 1 | 55 | 7502 |
| ✅ Query | 1 | 2 | 736 |
| ✅ RAG | 5 | 15 | 2936 |
@@ -69,13 +69,13 @@ This report compares **Python SDK feature categories** against **Python document
| ✅ Retrieval | 2 | 5 | 1044 |
| ✅ Routing | 1 | 2 | 430 |
| ✅ Sandbox | 5 | 6 | 2700 |
| ✅ Security | 1 | 3 | 2970 |
| ✅ Security | 1 | 3 | 2966 |
| ✅ Sessions | 4 | 13 | 5286 |
| ✅ Skills | 6 | 13 | 4946 |
| ✅ Tasks | 2 | 6 | 2743 |
| ✅ Telemetry | 1 | 2 | 604 |
| ✅ Templates | 1 | 8 | 1657 |
| ✅ Tools | 12 | 133 | 37934 |
| ✅ Tools | 12 | 133 | 37869 |
| ✅ Tracing | 3 | 2 | 139 |
| ✅ Vector Store | 1 | 12 | 1193 |
| ✅ Video | 2 | 6 | 649 |
@@ -2116,12 +2116,6 @@ 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
@@ -625,26 +625,11 @@ Your Goal: {self.goal}"""
formatted_tools.extend(openai_tools)
elif openai_tools is not None:
formatted_tools.append(openai_tools)
# 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)
tool_def = self._generate_tool_definition(tool.__name__)
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")
@@ -1638,7 +1638,6 @@ Write the complete compiled report:"""
try:
import uvicorn
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Mount
import threading
@@ -1669,9 +1668,8 @@ Write the complete compiled report:"""
# Create and run MCP server
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():
@@ -1839,35 +1839,6 @@ 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).
@@ -1899,13 +1870,7 @@ 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)
@@ -2102,11 +2067,8 @@ class ToolExecutionMixin:
func = tool
break
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.
if func is None:
# Check the global tool registry for plugins
try:
from ..tools.registry import get_registry
registry = get_registry()
@@ -1448,13 +1448,7 @@ 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)
@@ -195,11 +195,9 @@ class ApprovalRegistry:
self._approved_context.set(approved)
def is_already_approved(self, tool_name: str, arguments: Optional[Dict] = None) -> bool:
# 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
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())
def _is_session_scoped(
self, agent_name: Optional[str], tool_name: str, arguments: Optional[Dict]
@@ -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, observe
group_policy: str = "mention_only" # respond_all, mention_only, command_only
# Default safe tools (auto-injected for bots with no tools configured)
# With workspace scoping, file operations are now safe by construction
@@ -1076,13 +1076,7 @@ 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:
@@ -1111,7 +1105,6 @@ 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
@@ -1137,7 +1130,6 @@ 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),
)
@@ -107,12 +107,6 @@ __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 = {
@@ -197,12 +191,6 @@ _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"),
}
@@ -1,6 +1,6 @@
"""Evaluation package and case definitions."""
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional, Protocol
from typing import Any, Dict, List, Optional, Protocol
@dataclass
@@ -21,7 +21,6 @@ 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:
@@ -30,11 +29,7 @@ class EvalCase:
raise ValueError("EvalCase must have an input")
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary.
The ``verify`` callable is not serialisable and is intentionally omitted;
it is a live-only scorer supplied in Python.
"""
"""Convert to dictionary."""
return {
"name": self.name,
"input": self.input,
@@ -67,7 +62,6 @@ 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 {
@@ -78,7 +72,6 @@ class EvalResult:
"error": self.error,
"latency_ms": self.latency_ms,
"criteria_scores": self.criteria_scores,
"record": self.record,
}
@classmethod
@@ -92,7 +85,6 @@ class EvalResult:
error=data.get("error"),
latency_ms=data.get("latency_ms", 0.0),
criteria_scores=data.get("criteria_scores", {}),
record=data.get("record"),
)
@@ -34,10 +34,9 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
# 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]
# An eval case is (prompt, expected). ``expected`` may be None when using a
# Judge with only criteria.
EvalCase = Tuple[str, Any]
@dataclass
@@ -85,7 +84,7 @@ class PromptOptimizer:
def __init__(
self,
agent: "Agent",
evalset: List[EvalPair],
evalset: List[EvalCase],
*,
scorer: Optional[Any] = None,
metric: Optional[Callable[[str, Any], float]] = None,
@@ -1,487 +0,0 @@
"""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()
@@ -137,11 +137,6 @@ 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,
@@ -163,10 +158,6 @@ from .config import (
DeliveryConfig,
PollingConfig,
LivenessConfig,
# Hot-reload registry (Issue #3378)
HOT_APPLIABLE_KEYS,
SupportsHotReload,
is_hot_appliable,
)
# Lazy loading cache
@@ -343,9 +334,6 @@ __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",
@@ -365,10 +353,6 @@ __all__ = [
"DeliveryConfig",
"PollingConfig",
"LivenessConfig",
# Hot-reload registry (Issue #3378)
"HOT_APPLIABLE_KEYS",
"SupportsHotReload",
"is_hot_appliable",
# Implementations (lazy loaded from praisonai wrapper)
"WebSocketGateway",
"GatewaySession",
@@ -5,53 +5,7 @@ Provides configuration dataclasses for gateway and session settings.
"""
from dataclasses import dataclass, field
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:
...
from typing import Any, Dict, List, Optional
@dataclass
@@ -1,268 +0,0 @@
"""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
@@ -559,44 +559,32 @@ class TestAgentIntegration:
get_registry().clear()
def test_agent_execute_tool_from_registry(self):
"""Agent should find tools in registry when allow_global_tools is opted in."""
"""Agent should find tools in registry during execute_tool."""
from praisonaiagents import Agent, BaseTool, register_tool, get_registry
from praisonaiagents.config.feature_configs import ToolConfig
# Clear registry first
get_registry().clear()
class PowerTool(BaseTool):
name = "power"
description = "Raise to power"
def run(self, base: int, exp: int) -> int:
return base ** exp
register_tool(PowerTool())
# Agent that opts into the global registry via allow_global_tools=True.
# Agent without tools - should still find from registry
agent = Agent(
name="TestAgent",
role="Tester",
goal="Test tools",
tools=[],
tool_config=ToolConfig(allow_global_tools=True),
tools=[]
)
result = agent.execute_tool("power", {"base": 2, "exp": 3})
assert result == 8
# Default agent (no opt-in) must NOT reach the global registry.
safe_agent = Agent(
name="SafeAgent",
role="Tester",
goal="Test isolation",
tools=[],
)
with pytest.raises(Exception):
safe_agent.execute_tool("power", {"base": 2, "exp": 3})
# Clean up
get_registry().clear()
@@ -1,268 +0,0 @@
"""Unit tests for the trials engine (eval/trials.py)."""
import os
import tempfile
from praisonaiagents.eval import (
EvalCase,
EvalPackage,
run_trials,
TrialScore,
TrialAttempt,
TrialReport,
)
from praisonaiagents.eval.trials import _coerce_score, _score_attempt
class FakeAgent:
"""Minimal agent stub with a ``chat`` method and chat_history."""
def __init__(self, response="ok", agent_id="fake"):
self._response = response
self.agent_id = agent_id
self.chat_history = []
self.memory = None
self.knowledge = None
def chat(self, prompt, **kwargs):
self.chat_history.append({"role": "user", "content": prompt})
return self._response
class MemoryWritingAgent(FakeAgent):
"""Agent whose chat writes to a shared memory list (to test isolation)."""
def __init__(self, memory_sink):
super().__init__(response="done")
self._sink = memory_sink
def chat(self, prompt, **kwargs):
self._sink.append(prompt)
return self._response
def test_coerce_score_shapes():
assert _coerce_score(True) == TrialScore(value=1.0, passed=True)
assert _coerce_score(False) == TrialScore(value=0.0, passed=False)
assert _coerce_score(0.9).passed is True
assert _coerce_score(0.1).passed is False
ts = TrialScore(value=0.5, passed=True, reason="r")
assert _coerce_score(ts) is ts
def test_run_trials_k_attempts_per_case():
agent = FakeAgent(response="hello")
pkg = EvalPackage(name="p", cases=[
EvalCase(name="c1", input="a", verify=lambda o, e: True),
EvalCase(name="c2", input="b", verify=lambda o, e: True),
])
report = run_trials(agent, pkg, k=3, concurrency=2)
assert isinstance(report, TrialReport)
assert len(report.attempts["c1"]) == 3
assert len(report.attempts["c2"]) == 3
# Deterministic attempt ordering.
assert [a.attempt for a in report.attempts["c1"]] == [0, 1, 2]
def test_verify_callable_bool_float_score():
agent = FakeAgent(response="x")
pkg = EvalPackage(name="p", cases=[
EvalCase(name="b", input="i", verify=lambda o, e: True),
EvalCase(name="f", input="i", verify=lambda o, e: 0.8),
EvalCase(name="s", input="i",
verify=lambda o, e: TrialScore(value=0.3, passed=False)),
])
report = run_trials(agent, pkg, k=1)
assert report.attempts["b"][0].score.passed is True
assert report.attempts["f"][0].score.value == 0.8
assert report.attempts["s"][0].score.passed is False
def test_attempt_isolation_memory_untouched():
sink = []
agent = MemoryWritingAgent(memory_sink=sink)
# The isolated copy severs `memory`; caller's real memory attr stays intact.
pkg = EvalPackage(name="p", cases=[
EvalCase(name="c", input="write me", verify=lambda o, e: True),
])
run_trials(agent, pkg, k=2)
# Original agent's memory attr is untouched (still None on the original).
assert agent.memory is None
def test_attempts_do_not_share_session_id():
agent = FakeAgent(response="ok", agent_id="orig")
pkg = EvalPackage(name="p", cases=[
EvalCase(name="c", input="i", verify=lambda o, e: True),
])
run_trials(agent, pkg, k=2)
# Original agent id is not mutated by the isolated copies.
assert agent.agent_id == "orig"
def test_unscored_excluded_from_stats():
def slow_verify(o, e):
return True
agent = FakeAgent(response="ok")
# Force a timeout: agent sleeps longer than timeout_seconds.
class SlowAgent(FakeAgent):
def chat(self, prompt, **kwargs):
import time
time.sleep(0.3)
return "late"
pkg = EvalPackage(name="p", cases=[
EvalCase(name="c", input="i", timeout_seconds=0.05,
verify=slow_verify),
])
report = run_trials(SlowAgent(), pkg, k=2)
attempts = report.attempts["c"]
assert all(a.stop_reason == "timeout" for a in attempts)
assert all(a.score is None for a in attempts)
# Unscored attempts excluded -> pass_rate falls back to 0.0, no crash.
assert report.pass_rates()["c"] == 0.0
summary = report.summary()
assert summary["cases"]["c"]["n_scored"] == 0
def test_frontier_selection():
report = TrialReport(package_name="p", k=4)
report.attempts["all_pass"] = [
TrialAttempt("all_pass", i, "completed",
score=TrialScore(1.0, True)) for i in range(4)
]
report.attempts["all_fail"] = [
TrialAttempt("all_fail", i, "completed",
score=TrialScore(0.0, False)) for i in range(4)
]
report.attempts["frontier"] = [
TrialAttempt("frontier", i, "completed",
score=TrialScore(1.0 if i < 2 else 0.0, i < 2))
for i in range(4)
]
assert report.frontier() == ["frontier"]
rates = report.pass_rates()
assert rates["all_pass"] == 1.0
assert rates["all_fail"] == 0.0
assert rates["frontier"] == 0.5
def test_report_save_load_roundtrip():
agent = FakeAgent(response="hi")
pkg = EvalPackage(name="rt", cases=[
EvalCase(name="c", input="i", verify=lambda o, e: 0.9),
])
report = run_trials(agent, pkg, k=2, capture_record=True)
with tempfile.TemporaryDirectory() as d:
path = os.path.join(d, "trials.json")
report.save(path)
loaded = TrialReport.load(path)
assert loaded.package_name == "rt"
assert loaded.k == 2
assert len(loaded.attempts["c"]) == 2
assert loaded.attempts["c"][0].score.value == 0.9
# Record survives.
assert loaded.attempts["c"][0].record is not None
assert loaded.attempts["c"][0].record["output"] == "hi"
def test_score_attempt_no_scorer_passes_on_nonempty():
case = EvalCase(name="c", input="i")
agent = FakeAgent()
score = _score_attempt(case, "some output", agent, "some output")
assert score.passed is True
empty = _score_attempt(case, "", agent, "")
assert empty.passed is False
def test_original_chat_history_not_mutated():
# A shallow copy must not share the caller's chat_history list; running
# attempts should never append to (or clear) the original agent's history.
agent = FakeAgent(response="ok")
agent.chat_history.append({"role": "user", "content": "pre-existing"})
pkg = EvalPackage(name="p", cases=[
EvalCase(name="c", input="i", verify=lambda o, e: True),
])
run_trials(agent, pkg, k=3)
assert agent.chat_history == [{"role": "user", "content": "pre-existing"}]
def test_verifier_exception_does_not_abort_report():
# A raising verify() must be recorded as a failed score, not propagated.
def boom(o, e):
raise RuntimeError("metric blew up")
agent = FakeAgent(response="ok")
pkg = EvalPackage(name="p", cases=[
EvalCase(name="bad", input="i", verify=boom),
EvalCase(name="good", input="i", verify=lambda o, e: True),
])
report = run_trials(agent, pkg, k=2)
assert len(report.attempts["bad"]) == 2
assert all(a.score is not None and a.score.passed is False
for a in report.attempts["bad"])
assert all("verify error" in a.score.reason for a in report.attempts["bad"])
# Other cases still run and pass.
assert report.pass_rates()["good"] == 1.0
def test_duplicate_case_names_are_merged_not_overwritten():
agent = FakeAgent(response="ok")
pkg = EvalPackage(name="p", cases=[
EvalCase(name="dup", input="a", verify=lambda o, e: True),
EvalCase(name="dup", input="b", verify=lambda o, e: False),
])
report = run_trials(agent, pkg, k=2)
# Both same-named cases' attempts are retained (2 + 2), not overwritten.
assert len(report.attempts["dup"]) == 4
# Re-numbered attempt indices are unique and contiguous.
assert sorted(a.attempt for a in report.attempts["dup"]) == [0, 1, 2, 3]
assert report.pass_rates()["dup"] == 0.5
def test_timeout_bounds_runtime():
# A wedged agent must not make run_trials hang; the report returns promptly
# even though the leaked worker thread keeps running in the background.
import threading
import time as _time
release = threading.Event()
class WedgedAgent(FakeAgent):
def chat(self, prompt, **kwargs):
release.wait(timeout=5.0)
return "eventually"
pkg = EvalPackage(name="p", cases=[
EvalCase(name="c", input="i", timeout_seconds=0.05,
verify=lambda o, e: True),
])
start = _time.perf_counter()
report = run_trials(WedgedAgent(), pkg, k=1)
elapsed = _time.perf_counter() - start
release.set() # let the leaked worker finish
assert elapsed < 2.0
assert report.attempts["c"][0].stop_reason == "timeout"
def test_tool_assertion_scoring():
class ToolAgent(FakeAgent):
def chat(self, prompt, **kwargs):
self.chat_history.append({
"role": "assistant",
"tool_calls": [{"function": {"name": "search"}}],
})
return "done"
case = EvalCase(name="c", input="i", metadata={"expected_tools": ["search"]})
agent = ToolAgent()
resp = agent.chat("i")
score = _score_attempt(case, "done", agent, resp)
assert score.passed is True
case_missing = EvalCase(name="c", input="i",
metadata={"expected_tools": ["missing_tool"]})
agent2 = ToolAgent()
resp2 = agent2.chat("i")
score2 = _score_attempt(case_missing, "done", agent2, resp2)
assert score2.passed is False
@@ -210,14 +210,6 @@ class TestApprovalRegistry:
reg.mark_approved("write_file")
assert reg.is_already_approved("write_file")
def test_critical_tool_honours_mark_approved(self):
from praisonaiagents.approval.registry import ApprovalRegistry
reg = ApprovalRegistry()
reg.add_requirement("execute_command", "critical")
assert not reg.is_already_approved("execute_command")
reg.mark_approved("execute_command")
assert reg.is_already_approved("execute_command")
def test_already_approved_skips_backend(self):
"""Once approved, no backend call needed."""
from praisonaiagents.approval.registry import ApprovalRegistry
@@ -1,177 +0,0 @@
"""Unit tests for the event-loop liveness watchdog (Issue #3385).
Covers the pure, core-side primitive: policy validation, arm/disarm lifecycle,
that a healthy loop is never tripped, and that a wedged loop is detected. The
``dump_and_exit`` path (``os._exit``) is exercised via ``dump_only`` so tests
do not kill the interpreter.
"""
import asyncio
import os
import threading
import time
import pytest
from praisonaiagents.gateway import LoopWatchdog, LoopWatchdogPolicy
from praisonaiagents.gateway.protocols import GATEWAY_RESTART_EXIT_CODE
def test_policy_defaults():
p = LoopWatchdogPolicy()
assert p.probe_interval_s == 5.0
assert p.missed_probes_before_wedged == 3
assert p.on_wedge == "dump_and_exit"
assert p.exit_code == GATEWAY_RESTART_EXIT_CODE
assert p.wedge_after_s == 15.0
def test_policy_validation():
with pytest.raises(ValueError):
LoopWatchdogPolicy(probe_interval_s=0)
with pytest.raises(ValueError):
LoopWatchdogPolicy(missed_probes_before_wedged=0)
with pytest.raises(ValueError):
LoopWatchdogPolicy(on_wedge="boom")
def test_policy_rejects_non_finite_interval():
"""NaN / inf probe intervals must be rejected (Event.wait/Thread.join)."""
with pytest.raises(ValueError):
LoopWatchdogPolicy(probe_interval_s=float("nan"))
with pytest.raises(ValueError):
LoopWatchdogPolicy(probe_interval_s=float("inf"))
def test_disarm_when_not_armed_is_safe():
wd = LoopWatchdog()
assert wd.armed is False
wd.disarm() # should not raise
assert wd.armed is False
def _run_loop_for(loop, seconds):
def _stop():
loop.call_later(seconds, loop.stop)
loop.call_soon(_stop)
loop.run_forever()
def test_healthy_loop_not_tripped():
"""A responsive loop must never be declared wedged."""
loop = asyncio.new_event_loop()
policy = LoopWatchdogPolicy(
probe_interval_s=0.02,
missed_probes_before_wedged=3,
on_wedge="dump_only",
)
wd = LoopWatchdog(policy)
wd.arm(loop)
assert wd.armed is True
try:
_run_loop_for(loop, 0.4)
finally:
wd.disarm()
loop.close()
assert wd.wedged is False
assert wd.armed is False
def test_wedged_loop_detected():
"""A loop blocked inside a sync call is detected (dump_only, no exit)."""
loop = asyncio.new_event_loop()
policy = LoopWatchdogPolicy(
probe_interval_s=0.02,
missed_probes_before_wedged=2,
on_wedge="dump_only",
)
wd = LoopWatchdog(policy)
def _wedge():
# Block the loop thread synchronously to simulate a hang.
time.sleep(0.5)
wd.arm(loop)
try:
loop.call_soon(_wedge)
_run_loop_for(loop, 0.7)
finally:
wd.disarm()
loop.close()
assert wd.wedged is True
def test_wedge_writes_dump_file(tmp_path):
dump = tmp_path / "wedge.txt"
loop = asyncio.new_event_loop()
policy = LoopWatchdogPolicy(
probe_interval_s=0.02,
missed_probes_before_wedged=2,
on_wedge="dump_only",
dump_file=str(dump),
)
wd = LoopWatchdog(policy)
def _wedge():
time.sleep(0.5)
wd.arm(loop)
try:
loop.call_soon(_wedge)
_run_loop_for(loop, 0.7)
finally:
wd.disarm()
loop.close()
assert wd.wedged is True
assert dump.exists()
assert "wedged" in dump.read_text()
def test_arm_is_idempotent():
loop = asyncio.new_event_loop()
wd = LoopWatchdog(LoopWatchdogPolicy(probe_interval_s=0.05, on_wedge="dump_only"))
wd.arm(loop)
first = wd._thread
wd.arm(loop) # no-op
assert wd._thread is first
wd.disarm()
loop.close()
def test_disarm_suppresses_in_flight_exit():
"""A disarm racing an in-flight wedge must not call os._exit."""
wd = LoopWatchdog(
LoopWatchdogPolicy(
probe_interval_s=0.02,
missed_probes_before_wedged=1,
on_wedge="dump_and_exit",
)
)
# Simulate disarm() having set the stop flag while the worker is mid-wedge.
wd._stop.set()
exited = []
original_exit = os._exit
os._exit = lambda code: exited.append(code)
try:
wd._on_wedge() # must observe _stop and return without exiting
finally:
os._exit = original_exit
assert exited == []
assert wd.wedged is True
def test_closed_loop_does_not_trip():
"""Scheduling onto a closed loop is a normal shutdown, not a wedge."""
loop = asyncio.new_event_loop()
loop.close()
policy = LoopWatchdogPolicy(
probe_interval_s=0.02,
missed_probes_before_wedged=2,
on_wedge="dump_only",
)
wd = LoopWatchdog(policy)
wd.arm(loop)
time.sleep(0.2)
wd.disarm()
assert wd.wedged is False
@@ -291,7 +291,7 @@ class ChannelConfigSchema(BaseModel):
token: Union[str, Dict[str, Any]] = ""
app_token: Optional[Union[str, Dict[str, Any]]] = None # For Slack Socket Mode
mode: str = "poll" # poll | ws | webhook | hybrid
group_policy: str = "mention_only" # respond_all, mention_only, command_only, observe (record unmentioned msgs as context)
group_policy: str = "mention_only" # Default to secure: respond_all, mention_only, command_only
allow_silence: bool = False # Allow agent to return NO_REPLY to stay silent
silence_token: Optional[str] = None # Custom silence token (defaults to NO_REPLY)
allowlist: List[str] = Field(default_factory=list)
@@ -422,7 +422,7 @@ class ChannelConfigSchema(BaseModel):
@field_validator("group_policy")
@classmethod
def validate_group_policy(cls, v: str) -> str:
allowed = {"respond_all", "mention_only", "command_only", "observe"}
allowed = {"respond_all", "mention_only", "command_only"}
if v not in allowed:
raise ValueError(
f"Invalid group_policy '{v}'. Must be one of: {', '.join(sorted(allowed))}"
@@ -349,27 +349,6 @@ def _channel_token(config: Optional[Any], ch_cfg: Dict[str, Any]) -> Optional[st
return str(token) if token else None
def _sync_approval_registry(agent: Any) -> None:
"""Mirror ``agent._approval_backend`` onto the approval registry.
Tool functions decorated with ``@require_approval`` consult the registry
(often with ``agent_name=None``), so the agent backend alone is not enough
for bot/gateway shell paths.
"""
backend = getattr(agent, "_approval_backend", None)
if backend is None:
return
try:
from praisonaiagents.approval import get_approval_registry
reg = get_approval_registry()
agent_name = getattr(agent, "name", None)
if agent_name:
reg.set_backend(backend, agent_name=agent_name)
except ImportError:
logger.warning("Approval registry unavailable — shell approval may prompt in CLI")
def _wire_shell_approval_backend(
agent: Any,
*,
@@ -571,16 +550,15 @@ def enable_shell_tools(
agent._approval_backend = AutoApproveBackend()
except ImportError:
logger.warning("AutoApproveBackend unavailable for allow_shell")
else:
_wire_shell_approval_backend(
agent,
channel_type=channel_type,
config=config,
ch_cfg=ch_cfg,
allowed_approvers=_parse_shell_approvers(ch_cfg, channel_type),
)
return agent
_sync_approval_registry(agent)
_wire_shell_approval_backend(
agent,
channel_type=channel_type,
config=config,
ch_cfg=ch_cfg,
allowed_approvers=_parse_shell_approvers(ch_cfg, channel_type),
)
logger.info(
"Shell tools enabled for agent %r on channel %r (auto_approve_shell=%s)",
@@ -30,29 +30,6 @@ logger = logging.getLogger(__name__)
# Constant for permanent error detection
PERMANENT_ERROR_PREFIX = "Permanent error:"
# Prefix prepended to a crash-recovered outbound message that is re-sent without
# positive reconciliation. Such a re-send may be a duplicate, so the copy the
# recipient receives is labelled honestly rather than delivered silently.
RECOVERED_PREFIX = (
"\u267b\ufe0f Recovered reply \u2014 the gateway restarted during "
"delivery, so this may be a duplicate.\n\n"
)
def _annotate_recovered_payload(entry: Any, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Prefix a recovered entry's textual content with ``RECOVERED_PREFIX``.
Only string ``content`` is labelled; non-text payloads are returned
unchanged so a media/structured send is never corrupted. Returns a new dict
so the persisted payload is untouched.
"""
content = payload.get("content")
if not isinstance(content, str) or content.startswith(RECOVERED_PREFIX):
return payload
annotated = dict(payload)
annotated["content"] = f"{RECOVERED_PREFIX}{content}"
return annotated
class _TransientDeliveryError(Exception):
"""Recoverable delivery failure carrying the underlying error text.
@@ -483,18 +460,12 @@ class DurableDelivery:
platform: str = "",
backoff: Optional[BackoffPolicy] = None,
max_attempts: int = 3,
mark_recovered: bool = False,
):
self.outbox = outbox
self.adapter = adapter
self.platform = platform
self.backoff = backoff or BackoffPolicy()
self.max_attempts = max_attempts
# When True, a crash-recovered entry re-sent without positive
# reconciliation is labelled as a possible duplicate (honest
# at-least-once) rather than re-delivered silently. Off by default to
# preserve the historic behaviour.
self.mark_recovered = mark_recovered
async def send(
self,
@@ -748,12 +719,8 @@ class DurableDelivery:
return success
annotator = _annotate_recovered_payload if self.mark_recovered else None
return await self.outbox.drain(
sender,
limit=limit,
reconciler=self._build_reconciler(),
recovery_annotator=annotator,
sender, limit=limit, reconciler=self._build_reconciler()
)
@@ -66,7 +66,6 @@ class DurableAdapterMixin:
max_size: int = 50_000,
ttl_seconds: int = 7 * 86400,
ordering: Literal["strict", "best_effort"] = "best_effort",
mark_recovered: bool = False,
) -> None:
"""Set up durable outbound delivery.
@@ -82,11 +81,6 @@ class DurableAdapterMixin:
FIFO so a later same-conversation message can never overtake an
earlier undelivered one. The ``reliability="production"`` preset
resolves to ``"strict"`` via ``resolve_reliability``.
mark_recovered: When True, a crash-recovered message re-sent without
positive reconciliation is prefixed with a visible
"possible duplicate after restart" marker instead of being
re-delivered silently. Defaults to False for backward
compatibility.
"""
self.outbox: Optional[OutboundQueue] = None
self.durable_delivery: Optional[DurableDelivery] = None
@@ -106,7 +100,6 @@ class DurableAdapterMixin:
adapter=self, # Adapter must implement MessageSender protocol
platform=platform,
max_attempts=max_attempts,
mark_recovered=mark_recovered,
)
logger.info(
+12 -105
View File
@@ -19,10 +19,7 @@ Design constraints (per PraisonAI principles):
adapters that declare ``PlatformCapabilities.reconciles_unknown_send`` can),
it is asked whether the prior attempt actually landed; if so the entry is
marked ``sent`` without re-dispatch. Without a reconciler, ``recovered``
entries fall back to at-least-once re-send. Such an unreconciled re-send may
be a duplicate, so ``drain`` accepts an optional ``recovery_annotator`` that
labels that copy (e.g. a short "possible duplicate after restart" prefix)
instead of re-sending it silently.
entries fall back to at-least-once re-send (current behaviour).
Storage schema::
@@ -324,34 +321,12 @@ class OutboundQueue:
key: str,
error: str,
permanent: bool = False,
*,
keep_recovered: bool = False,
) -> bool:
"""Mark a message as failed.
Args:
key: Tracking key for the entry.
error: Error text to persist for retry/backoff decisions.
permanent: Mark a permanent failure that is never retried.
keep_recovered: When True, a transient (non-permanent) failure of a
crash-``recovered`` entry preserves its ``recovered`` status
instead of demoting it to ``failed``. This keeps the
"possible duplicate after restart" semantics sticky across
retries, so a ``recovery_annotator`` still labels the copy on
the next drain. Ignored for permanent failures.
"""
"""Mark a message as failed."""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None, self._sync_mark_failed, key, error, permanent, keep_recovered
)
return await loop.run_in_executor(None, self._sync_mark_failed, key, error, permanent)
def _sync_mark_failed(
self,
key: str,
error: str,
permanent: bool,
keep_recovered: bool = False,
) -> bool:
def _sync_mark_failed(self, key: str, error: str, permanent: bool) -> bool:
"""Synchronous version of mark_failed for thread pool execution."""
entry_id = self._extract_id_from_key(key)
status = 'permanent_failure' if permanent else 'failed'
@@ -361,26 +336,11 @@ class OutboundQueue:
# already recorded 'sent' for this row; clobbering it with 'failed'
# here would re-queue an entry that already reached the channel and
# produce a duplicate. Only in-flight states may transition to failed.
#
# Keep a crash-recovered entry in the 'recovered' state on a
# transient failure (keep_recovered) so its "possible duplicate"
# semantics survive the retry: otherwise the row would flip to
# 'failed', the recovered-only annotator would be skipped on the
# next drain, and a later successful retry would deliver an
# unlabelled duplicate.
if keep_recovered and not permanent:
cur = conn.execute("""
UPDATE outbound_queue
SET status = 'recovered', error = ?, last_attempt = ?,
attempts = attempts + 1
WHERE id = ? AND status IN ('sending', 'recovered')
""", (error, time.time(), entry_id))
else:
cur = conn.execute("""
UPDATE outbound_queue
SET status = ?, error = ?, last_attempt = ?, attempts = attempts + 1
WHERE id = ? AND status IN ('pending', 'sending', 'recovered', 'failed')
""", (status, error, time.time(), entry_id))
cur = conn.execute("""
UPDATE outbound_queue
SET status = ?, error = ?, last_attempt = ?, attempts = attempts + 1
WHERE id = ? AND status IN ('pending', 'sending', 'recovered', 'failed')
""", (status, error, time.time(), entry_id))
conn.commit()
# Release active claim
@@ -394,9 +354,6 @@ class OutboundQueue:
limit: Optional[int] = None,
*,
reconciler: Optional[Callable[[OutboundEntry], Awaitable[bool]]] = None,
recovery_annotator: Optional[
Callable[[OutboundEntry, Dict[str, Any]], Dict[str, Any]]
] = None,
) -> Tuple[int, int]:
"""Process pending messages.
@@ -411,15 +368,6 @@ class OutboundQueue:
fresh ``pending``/``failed`` entries are sent normally. If no
reconciler is supplied, recovered entries are re-sent
(at-least-once).
recovery_annotator: Optional sync function applied ONLY on the
unreconciled ``recovered`` branch — i.e. a mid-send-crash entry
whose delivery outcome is unknown and which is about to be
re-dispatched at-least-once. Given ``(entry, payload)`` it
returns a (possibly new) payload to send, so the copy the
recipient receives can be visibly labelled a possible duplicate
produced by a gateway restart. Never applied to fresh
``pending``/``failed`` entries or to entries confirmed by the
reconciler.
Returns:
Tuple of (succeeded, failed) counts
@@ -480,53 +428,17 @@ class OutboundQueue:
f"falling back to re-send"
)
# Whether this entry entered the drain as a crash-recovered one.
# Captured before any status write so a transient re-send failure can
# keep it 'recovered' (sticky "possible duplicate" semantics) rather
# than demoting it to 'failed' and dropping the recovery label on the
# next retry.
was_recovered = entry.status == "recovered"
try:
# Attempt delivery
payload = json.loads(entry.payload)
# Honest at-least-once: a recovered entry re-dispatched without
# positive reconciliation may be a duplicate, so let the caller
# visibly label it. Applied only on this unreconciled recovered
# branch — never to fresh pending/failed or reconciled entries.
if was_recovered and recovery_annotator is not None:
try:
# Hand the annotator its own fresh copy of the payload so
# a partial in-place mutation followed by a raise cannot
# leak a half-labelled payload into the send. Only adopt
# the result once it completes and returns a dict;
# otherwise fall back to the original unlabelled payload.
annotated = recovery_annotator(entry, json.loads(entry.payload))
if not isinstance(annotated, dict):
raise TypeError(
"recovery_annotator must return a dict, got "
f"{type(annotated).__name__}"
)
payload = annotated
except Exception as e: # pragma: no cover - defensive
logger.warning(
f"recovery_annotator failed for {key}: {e}; "
f"sending unlabelled"
)
success = await sender(entry.target, payload)
if success:
await self.mark_sent(key)
succeeded += 1
else:
# Transient failure, will retry. Keep a recovered entry
# 'recovered' so the duplicate label survives the retry.
await self.mark_failed(
key,
"Delivery returned false",
permanent=False,
keep_recovered=was_recovered,
)
# Transient failure, will retry
await self.mark_failed(key, "Delivery returned false", permanent=False)
failed += 1
except Exception as e:
@@ -541,12 +453,7 @@ class OutboundQueue:
mandated = server_retry_after(e)
if mandated is not None and "retry_after" not in error_text.lower():
error_text = f"{error_text} [retry_after: {mandated:g}]"
await self.mark_failed(
key,
error_text,
permanent=permanent,
keep_recovered=was_recovered,
)
await self.mark_failed(key, error_text, permanent=permanent)
failed += 1
if permanent:
@@ -103,29 +103,6 @@ _TELEGRAM_RECOVERABLE: Set[str] = {
"webhook",
}
# Auth/credential rejection patterns. A token that is revoked, rotated, wrong,
# or expired is NOT transient (retrying an invalid token forever is pointless)
# and NOT a generic fatal (it self-heals the moment the credential is fixed).
# It is its own first-class outcome so the gateway can surface it as a
# redacted, operator-actionable degraded state and auto-recover on repair.
_CREDENTIAL_PATTERNS: Set[str] = {
"unauthorized",
"invalid token",
"invalid_auth",
"token_revoked",
"token revoked",
"not_authed",
"authentication failed",
"authentication error",
"invalid credentials",
"invalid api key",
"invalid_api_key",
"expired token",
"token expired",
"access token is invalid",
"unauthenticated",
}
def _parse_http_date_seconds(value: str) -> Optional[float]:
"""Parse an HTTP-date Retry-After value into seconds from now.
@@ -258,45 +235,6 @@ def is_recoverable_error(err: BaseException, platform: str = "") -> bool:
return False
def is_credential_error(err: BaseException, platform: str = "") -> bool:
"""Check if an error is an auth/credential rejection (revoked/expired token).
A runtime 401/403 (or the platform text equivalent — "Unauthorized",
"invalid token", Slack's ``invalid_auth``/``token_revoked``) means the
channel's credential is invalid *right now*. This is a distinct outcome from
both transient/recoverable errors (network blips, 5xx, rate limits) and
generic fatal errors: it should stop the pointless reconnect loop against a
known-bad credential and surface as a redacted, operator-actionable degraded
state that auto-recovers when the credential is repaired.
Args:
err: The exception to classify.
platform: Optional platform name (reserved for platform-specific tuning).
Returns:
True if the error signals an invalid/rejected credential.
"""
if err is None:
return False
# HTTP 401 Unauthorized / 403 Forbidden are the canonical auth-rejection
# status codes across platforms (Telegram exposes ``error_code``).
status = getattr(err, "status", None)
if status is None:
status = getattr(err, "status_code", None)
if status is None:
status = getattr(err, "error_code", None)
if isinstance(status, int) and status in (401, 403):
return True
msg = str(err).lower()
for pattern in _CREDENTIAL_PATTERNS:
if pattern in msg:
return True
return False
# Patterns that confirm a *whole target* is permanently unreachable: the bot
# was kicked/blocked, the chat/group/channel no longer exists. These differ from
# transient errors (handled by is_recoverable_error) and from thread-/message-
@@ -1599,65 +1599,6 @@ class BotSessionManager:
"""
return self._last_presentation.pop(self._storage_key(user_id), None)
def record_passive(
self,
user_id: str,
content: str,
sender: str = "",
**route: str,
) -> bool:
"""Append a group message as passive context without running the agent.
Issue #3380: under the ``observe`` group policy a message that does not
address the bot is recorded into the session transcript as an ordinary
``user`` turn (optionally attributed to its sender for multi-party
chats) so that when the bot is next mentioned it can see the preceding
conversation. No agent run is dispatched. Appends directly under a
per-key threading lock so it is safe to call from any sync/async
context (including inline on the inbound handler's own loop thread);
errors are swallowed.
Optional ``chat_id``/``thread_id``/``account``/``chat_type`` route kwargs
are threaded through so that with ``session_scope='per_chat'`` the
passive entry lands on the same shared group key that a subsequent
addressed turn reads from (Issue #2376 routing); without them behaviour
is unchanged (per_user, keyed by the sender's session).
"""
if not content:
return False
text = self._attribute(content, sender) if sender else content
entry = {
"role": "user",
"content": text,
"timestamp": datetime.now(timezone.utc).isoformat(),
"passive": True,
}
# A single fire-and-forget append. Unlike the outbound mirror (called
# from other threads / cron), ``record_passive`` runs inline on the
# inbound handler's own event-loop thread, so the cross-thread
# ``run_coroutine_threadsafe`` bridge would dead-lock waiting on the
# very loop that is calling it. Append directly under a per-key
# threading lock (no asyncio lock touched) so it is safe from any
# context and never blocks the loop.
try:
storage_key = self._storage_key(user_id, **route)
sync_locks = getattr(self, "_user_sync_locks", None)
if sync_locks is None:
self._user_sync_locks = sync_locks = {}
lock = sync_locks.get(storage_key)
if lock is None:
import threading
lock = sync_locks[storage_key] = threading.Lock()
with lock:
history = list(self._load_history(user_id, **route))
history.append(entry)
self._save_history(user_id, history, **route)
self._last_active[storage_key] = time.monotonic()
return True
except Exception as e: # pragma: no cover — defensive, never break inbound
logger.warning("record_passive failed: %s", e)
return False
def reset(self, user_id: str, **route: str) -> bool:
"""Clear a session's history. Returns True if it existed.
+1 -13
View File
@@ -637,19 +637,7 @@ class SlackBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
user_allowed = await UnknownUserHandler.handle(bot_message, self._bot_context)
if not user_allowed:
return
# Gateway routing (shell opt-in, per-route agents) runs via on_message
# handlers — mirror the regular message path so @mentions are not stuck
# on a stale default agent missing execute_command.
for handler in self._message_handlers:
try:
if asyncio.iscoroutinefunction(handler):
await handler(bot_message)
else:
handler(bot_message)
except Exception as e:
logger.error(f"Message handler error: {e}")
text = event.get("text", "")
if self._bot_user:
text = text.replace(f"<@{self._bot_user.user_id}>", "").strip()
@@ -2112,38 +2112,6 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
return None
def _record_passive_group_message(bot: "TelegramBot", message) -> None:
"""Record an unmentioned group message as passive session context.
Issue #3380: used by the ``observe`` group policy so that unmentioned
messages are retained in the transcript (without triggering an agent run)
and are visible to the agent when it is next addressed. Best-effort — a
missing session manager or any failure is swallowed so inbound handling is
never broken by observation.
"""
session_mgr = getattr(bot, "_session", None)
if session_mgr is None or not hasattr(session_mgr, "record_passive"):
return
try:
content = message.get_text() if hasattr(message, "get_text") else str(message.content)
user_id = message.sender.user_id if message.sender else ""
sender = (message.sender.display_name or user_id) if message.sender else user_id
# Thread the same routing fields an addressed turn uses so that with
# session_scope="per_chat" the passive entry lands on the shared group
# key the next mentioned run reads from (Issue #3380 / #2376). With the
# default per_user scope these are simply ignored.
session_mgr.record_passive(
user_id,
content,
sender=sender,
chat_id=str(message.channel.channel_id) if message.channel and message.channel.channel_id else "",
thread_id=str(message.thread_id) if getattr(message, "thread_id", None) else "",
account=getattr(bot.config, "account", "default"),
)
except Exception as e: # pragma: no cover — defensive
logger.debug(f"Failed to record passive group message: {e}")
async def process_inbound_telegram_message(
update, # Telegram Update
bot: TelegramBot,
@@ -2237,7 +2205,7 @@ async def process_inbound_telegram_message(
if message.message_type != MessageType.COMMAND:
logger.debug(f"Message dropped: non-command in command_only group {channel_id}")
return None
elif group_policy in ("mention_only", "observe"):
elif group_policy == "mention_only":
# Check if bot was mentioned in the message
bot_username = bot._bot_user.username.lower() if bot._bot_user and bot._bot_user.username else ""
mention_handle = f"@{bot_username}" if bot_username else ""
@@ -2246,15 +2214,7 @@ async def process_inbound_telegram_message(
) or message.message_type == MessageType.COMMAND # Commands are always allowed
if not bot_mentioned:
# Issue #3380: under ``observe`` an unmentioned group message is
# recorded into the session transcript as passive context (no
# agent run) so the bot has memory of the conversation when it is
# next addressed. ``mention_only`` still drops it outright.
if group_policy == "observe":
_record_passive_group_message(bot, message)
logger.debug(f"Message observed (no run) in group {channel_id}")
else:
logger.debug(f"Message dropped: bot not mentioned in group {channel_id}")
logger.debug(f"Message dropped: bot not mentioned in group {channel_id}")
return None
elif group_policy == "respond_all":
# Allow all group messages
@@ -209,10 +209,9 @@ def gateway_restart(
agents: Optional[str] = typer.Option(
None, "--agents", help="Path to agent configuration file (for direct relaunch)"
),
drain_timeout: Optional[float] = typer.Option(
None, "--drain-timeout",
help="Seconds to wait for in-flight agent turns to finish before "
"relaunch (default: the persisted start value, else 10)",
drain_timeout: float = typer.Option(
10.0, "--drain-timeout",
help="Seconds to wait for in-flight agent turns to finish before relaunch",
),
):
"""Gracefully drain in-flight turns, then relaunch the gateway.
@@ -224,14 +223,10 @@ def gateway_restart(
launch arguments. Otherwise it drains the running PID and relaunches
directly (#3161).
The direct (non-service) relaunch replays the CLI-only runtime flags the
Note: the direct (non-service) relaunch cannot recover CLI-only flags the
original process was started with (e.g. ``--openai-api``,
``--reliability``, ``--max-concurrent-runs``) from the persisted start-flags
artefact written at ``start`` time, so a restart faithfully reproduces the
running gateway instead of silently reverting to defaults (#3349). Flags
passed explicitly to ``restart`` still win over the persisted values; an
omitted ``--drain-timeout`` replays the persisted drain window rather than
forcing a fixed default.
``--max-concurrent-runs``); put production settings in ``gateway.yaml`` (or
install as a service) so a restart preserves them.
Examples:
praisonai gateway restart
@@ -240,7 +235,7 @@ def gateway_restart(
"""
import os
from praisonai_bot.daemon import restart_daemon, get_daemon_status
from ..features.gateway import GatewayHandler, load_start_flags
from ..features.gateway import GatewayHandler
from ..output.console import get_output_controller
if port is None:
@@ -268,44 +263,21 @@ def gateway_restart(
"falling back to direct drain + relaunch."
)
# Replay the CLI-only runtime flags the original process was started with so
# the restart reproduces the exact posture (durable delivery, concurrency
# ceiling, OpenAI-compat surface, lifecycle) instead of silently reverting
# to defaults (#3349). Flags passed explicitly to ``restart`` (config /
# agents / drain_timeout) still win over the persisted values.
persisted = load_start_flags(host, port)
start_kwargs = dict(persisted)
if config is not None:
start_kwargs["config_file"] = config
if agents is not None:
start_kwargs["agent_file"] = agents
# An omitted --drain-timeout (None) must replay the persisted start value,
# not clobber it with Typer's old fixed 10s default — otherwise every
# restart silently shortens a production drain window (#3349). An explicit
# value still wins over the persisted one.
if drain_timeout is not None:
start_kwargs["drain_timeout"] = drain_timeout
# Effective window for draining the OLD process before relaunch: explicit
# flag > persisted start value > 10s fallback, so a long configured drain is
# not cut off by a fixed wait before force-kill (#3161).
effective_drain = drain_timeout
if effective_drain is None:
effective_drain = persisted.get("drain_timeout")
if effective_drain is None:
effective_drain = 10.0
# Direct path: gracefully stop the running gateway (honouring drain), then
# start a fresh instance in the foreground.
# start a fresh instance in the foreground. The requested drain window is
# applied to the OLD process too, so a long --drain-timeout is not cut off
# by a fixed 10s wait before force-kill (#3161).
handler = GatewayHandler()
handler.stop(host=host, port=port, force=False, drain_timeout=effective_drain)
handler.stop(host=host, port=port, force=False, drain_timeout=drain_timeout)
if persisted:
output.print_info(
"Replaying persisted start flags: "
+ ", ".join(sorted(persisted.keys()))
)
output.print_info("Relaunching gateway...")
handler.start(host=host, port=port, **start_kwargs)
handler.start(
host=host,
port=port,
agent_file=agents,
config_file=config,
drain_timeout=drain_timeout,
)
@app.command("status")
@@ -429,15 +401,26 @@ def _check_gateway_secret_strength(config_path: str):
return str(WeakGatewaySecretError(field="gateway.auth_token"))
from praisonai_bot.gateway.preflight import ( # noqa: E402 — re-exported for tests/CLI
apply_probe_ca_bundle as _apply_probe_ca_bundle,
check_gateway_running as _check_gateway_running,
probe_channels as _probe_channels,
probe_results_to_dict as _probe_results_to_dict,
resolve_env_token as _resolve_env_token,
run_shell_readiness_check as _run_shell_readiness_check,
run_turn_test as _run_gateway_turn_test,
)
def _resolve_env_token(value):
"""Resolve a credential input to its value for probing.
Handles ``${VAR}`` placeholders and the additive secret-reference form
``{source: file|env|exec, id: ...}`` (Issue #3102). Resolved values are
registered for log redaction. Plain strings pass through unchanged.
"""
import os
if isinstance(value, dict) and "source" in value and "id" in value:
try:
from praisonaiagents.secrets import resolve_secret
result = resolve_secret(value)
return result.value or ""
except Exception: # pragma: no cover — defensive
return ""
if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
return os.environ.get(value[2:-1], "")
return value
def _secret_availability(value) -> str:
@@ -493,6 +476,43 @@ def _is_ssl_error(result) -> bool:
)
def _apply_probe_ca_bundle() -> None:
"""Point the probe HTTP client at a custom CA bundle if configured.
Honors ``PRAISONAI_SSL_CA_BUNDLE`` (preferred), ``REQUESTS_CA_BUNDLE`` and
``SSL_CERT_FILE`` so enterprise users behind an SSL-inspecting proxy can
supply their corporate CA and have the preflight probe trust it — mirroring
what the runtime adapter's SSL stack already honors (#2845). Setting
``SSL_CERT_FILE`` is what Python's default ``ssl`` context (used by
``aiohttp``) reads at load time.
"""
import os
preferred = os.environ.get("PRAISONAI_SSL_CA_BUNDLE")
ca_bundle = (
preferred
or os.environ.get("REQUESTS_CA_BUNDLE")
or os.environ.get("SSL_CERT_FILE")
)
if not ca_bundle:
return
if not os.path.exists(ca_bundle):
print(
f"Warning: CA bundle path '{ca_bundle}' does not exist — "
"SSL_CERT_FILE / REQUESTS_CA_BUNDLE not updated for probe."
)
return
if preferred and preferred == ca_bundle:
# Explicit PraisonAI override wins over any pre-existing values.
os.environ["SSL_CERT_FILE"] = ca_bundle
os.environ["REQUESTS_CA_BUNDLE"] = ca_bundle
else:
os.environ.setdefault("SSL_CERT_FILE", ca_bundle)
os.environ.setdefault("REQUESTS_CA_BUNDLE", ca_bundle)
def _load_channels(config: str) -> dict:
"""Load the ``channels`` mapping from a gateway.yaml file (or exit)."""
import os
@@ -508,6 +528,65 @@ def _load_channels(config: str) -> dict:
return cfg.get("channels", {})
async def _probe_channels(channels: dict, timeout: float = 15.0) -> dict:
"""Build a lightweight Bot per channel and probe its credentials.
Probing builds the adapter lazily and calls the platform identity API
(Telegram getMe, Slack auth.test, …) without starting message
processing. No agent is required. Returns ``{name: ProbeResult}``.
Each probe is bounded by ``timeout`` (seconds) so one stuck adapter
cannot hang the whole pre-flight; a timeout is reported as a failure.
Loads ``~/.praisonai/.env`` first so ``${VAR}`` tokens stored there
(e.g. by ``praisonai onboard``) resolve — mirroring what
``GatewayHandler.start()`` does at runtime, so every credential check
(doctor / channels --probe / start --preflight) uses the same
token-resolution behavior (#2426).
"""
import asyncio as _asyncio
# Load env-file BEFORE resolving ${VAR} tokens so all probe paths
# (doctor, channels --probe, start --preflight) match runtime behavior.
try:
from ..features.gateway import _load_praisonai_env_file
_load_praisonai_env_file()
except Exception: # pragma: no cover — defensive
pass
# Honor a custom CA bundle so the probe's HTTP client trusts a corporate
# proxy/MITM certificate the same way the runtime adapter does (#2845).
_apply_probe_ca_bundle()
from praisonai_bot.bots import Bot
from praisonaiagents.bots import ProbeResult
async def _probe_one(name: str, ch_cfg: dict):
platform = ch_cfg.get("platform", name)
token = _resolve_env_token(ch_cfg.get("token", ""))
extras = {
k: _resolve_env_token(v)
for k, v in ch_cfg.items()
if k not in ("platform", "token")
}
try:
bot = Bot(platform, token=token, **extras)
return name, await _asyncio.wait_for(bot.probe(), timeout=timeout)
except _asyncio.TimeoutError:
return name, ProbeResult(
ok=False,
platform=platform,
error=f"probe timed out after {timeout:g}s",
)
except Exception as e: # pragma: no cover — defensive
return name, ProbeResult(ok=False, platform=platform, error=str(e))
results = await _asyncio.gather(
*(_probe_one(name, ch_cfg or {}) for name, ch_cfg in channels.items())
)
return dict(results)
def _compute_secret_availability(channels: dict) -> dict:
"""Per-channel credential availability without revealing values (#3102).
@@ -541,6 +620,14 @@ def _print_secret_availability(report: dict) -> None:
print()
def _probe_results_to_dict(results: dict) -> dict:
"""JSON-serializable per-channel probe payload."""
return {
name: r.to_dict() if hasattr(r, "to_dict") else vars(r)
for name, r in results.items()
}
def _render_probe_results(results: dict, json_output: bool = False) -> bool:
"""Print per-channel probe verdicts. Returns True if all channels passed."""
all_ok = all(getattr(r, "ok", False) for r in results.values())
@@ -576,16 +663,6 @@ def _render_probe_results(results: dict, json_output: bool = False) -> bool:
def gateway_doctor(
config: str = typer.Option("gateway.yaml", "--config", "-c", help="Path to gateway.yaml"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
channel: Optional[str] = typer.Option(
None,
"--channel",
help="Channel name for --turn (default: first configured channel)",
),
turn: Optional[str] = typer.Option(
None,
"--turn",
help="Run one live inbound agent turn offline (requires LLM API key)",
),
):
"""Validate every configured channel's credentials (pre-flight check).
@@ -593,26 +670,28 @@ def gateway_doctor(
(Telegram getMe, Slack auth.test, Discord identify, WhatsApp token check)
without starting message processing. Exits non-zero if any channel fails.
Optional ``--turn`` runs an offline inbound agent turn via
``BotSessionManager.chat`` (including ``allow_shell`` setup). It does
**not** exercise Slack Bolt/socket handlers or @mention routing.
Examples:
praisonai gateway doctor
praisonai gateway doctor --config my-gateway.yaml --json
praisonai gateway doctor --config gateway.yaml --channel slack --turn "Say OK"
"""
import asyncio
import json
# Inspect the gateway's OWN auth_token for known-weak/placeholder values
# (Issue #3259) — channel probes never cover this. Fail closed on an
# external bind, warn on loopback (consistent with runtime posture).
gateway_secret_error = _check_gateway_secret_strength(config)
channels = _load_channels(config)
channels = _load_channels(config)
if not channels:
payload: dict = {"probes": {}}
if gateway_secret_error:
payload["gateway_auth_token"] = "weak"
if json_output:
# Always emit a single parseable JSON document, even on the
# weak/missing gateway-secret failure path, so automation can
# distinguish an auth failure from an empty/interrupted run.
import json
payload: dict = {"probes": {}}
if gateway_secret_error:
payload["gateway_auth_token"] = "weak"
print(json.dumps(payload, indent=2))
else:
print("No channels configured.")
@@ -622,193 +701,33 @@ def gateway_doctor(
raise typer.Exit(1)
raise typer.Exit(0)
# Credential source availability, computed WITHOUT printing any value
# (Issue #3102): operators can validate file/env secret wiring even when
# the network probe cannot run.
availability = _compute_secret_availability(channels)
results = asyncio.run(_probe_channels(channels))
all_ok = all(getattr(r, "ok", False) for r in results.values())
turn_gate_ok = all_ok
if channel and channel in results:
turn_gate_ok = getattr(results[channel], "ok", False)
payload: dict = {"probes": _probe_results_to_dict(results)}
if availability:
payload["secrets"] = availability
if gateway_secret_error:
payload["gateway_auth_token"] = "weak"
if json_output:
# Emit a SINGLE JSON document so consumers using json.load(s) can parse
# the whole output (previously the availability and probe blocks were
# printed as two separate top-level documents — invalid JSON).
import json
if not json_output:
payload = {"probes": _probe_results_to_dict(results)}
if availability:
payload["secrets"] = availability
if gateway_secret_error:
payload["gateway_auth_token"] = "weak"
print(json.dumps(payload, indent=2))
else:
_print_secret_availability(availability)
_render_probe_results(results, json_output=False)
if gateway_secret_error:
print(gateway_secret_error)
if gateway_secret_error:
if json_output:
print(json.dumps(payload, indent=2))
raise typer.Exit(1)
probe_blocks_turn = not all_ok and not (turn and channel and turn_gate_ok)
if probe_blocks_turn and not turn:
if json_output:
print(json.dumps(payload, indent=2))
raise typer.Exit(1)
if turn:
target = channel or (next(iter(channels.keys())) if channels else None)
if not target:
err = "--turn requires at least one configured channel"
payload["turn"] = {"channel": None, "ok": False, "response": err}
if json_output:
print(json.dumps(payload, indent=2))
else:
print(f"Error: {err}")
raise typer.Exit(1)
if not turn_gate_ok:
err = f"channel '{target}' probe failed — cannot run --turn"
payload["turn"] = {"channel": target, "ok": False, "response": err}
if json_output:
print(json.dumps(payload, indent=2))
else:
print(f"Error: {err}")
raise typer.Exit(1)
ok, message = asyncio.run(_run_gateway_turn_test(config, target, turn))
payload["turn"] = {"channel": target, "ok": ok, "response": message}
if json_output:
print(json.dumps(payload, indent=2))
else:
print(f"\nTurn test ({target}): {'OK' if ok else 'FAIL'}")
print(message if ok else f"Error: {message}")
if not ok or not all_ok:
raise typer.Exit(1)
return
if json_output:
print(json.dumps(payload, indent=2))
if not all_ok:
raise typer.Exit(1)
@app.command("test")
def gateway_test(
config: str = typer.Option("gateway.yaml", "--config", "-c", help="Path to gateway.yaml"),
json_output: bool = typer.Option(False, "--json", help="Output JSON"),
channel: Optional[str] = typer.Option(
None,
"--channel",
help="Channel name for --turn (default: first configured channel)",
),
turn: Optional[str] = typer.Option(
None,
"--turn",
help="Run one live inbound agent turn offline (requires LLM API key)",
),
check_running: bool = typer.Option(
False,
"--check-running",
help="Verify the gateway REST /info endpoint is reachable",
),
):
"""One-shot gateway readiness check (probes + shell wiring + optional turn).
Recommended onboarding path before ``gateway start``. Combines credential
probes, offline shell wiring validation, and an optional offline agent turn.
``--turn`` uses ``BotSessionManager.chat`` only — it does not prove live
Slack @mention delivery. After starting, confirm ``@mention received`` in
gateway logs.
Examples:
praisonai gateway test --config bot.yaml
praisonai gateway test --config bot.yaml --channel slack --turn "Say OK"
praisonai gateway test --config bot.yaml --check-running
"""
import asyncio
import json
gateway_secret_error = _check_gateway_secret_strength(config)
channels = _load_channels(config)
payload: dict = {}
if gateway_secret_error:
payload["gateway_auth_token"] = "weak"
if not channels:
payload.setdefault("probes", {})
if json_output:
print(json.dumps(payload, indent=2))
else:
print("No channels configured.")
if gateway_secret_error:
print(gateway_secret_error)
raise typer.Exit(1 if gateway_secret_error else 0)
availability = _compute_secret_availability(channels)
results = asyncio.run(_probe_channels(channels))
all_ok = all(getattr(r, "ok", False) for r in results.values())
turn_gate_ok = all_ok
if channel and channel in results:
turn_gate_ok = getattr(results[channel], "ok", False)
payload["probes"] = _probe_results_to_dict(results)
if availability:
payload["secrets"] = availability
shell_result = _run_shell_readiness_check(config)
payload["shell"] = {
"ok": shell_result.ok,
"message": shell_result.message,
"issues": shell_result.issues,
}
if not json_output:
_print_secret_availability(availability)
_render_probe_results(results, json_output=False)
shell_mark = "✓" if shell_result.ok else "✗"
print(f"shell wiring {shell_mark} {shell_result.message}")
if shell_result.issues:
for issue in shell_result.issues:
print(f" - {issue}")
if gateway_secret_error:
print(gateway_secret_error)
failed = bool(gateway_secret_error) or not all_ok or not shell_result.ok
if check_running:
running_ok, running_msg = _check_gateway_running(config)
payload["running"] = {"ok": running_ok, "message": running_msg}
if not json_output:
mark = "✓" if running_ok else "✗"
print(f"gateway up {mark} {running_msg}")
failed = failed or not running_ok
if turn:
target = channel or (next(iter(channels.keys())) if channels else None)
if not target:
err = "--turn requires at least one configured channel"
payload["turn"] = {"channel": None, "ok": False, "response": err}
if json_output:
print(json.dumps(payload, indent=2))
else:
print(f"Error: {err}")
raise typer.Exit(1)
if not turn_gate_ok:
err = f"channel '{target}' probe failed — cannot run --turn"
payload["turn"] = {"channel": target, "ok": False, "response": err}
if json_output:
print(json.dumps(payload, indent=2))
else:
print(f"Error: {err}")
raise typer.Exit(1)
ok, message = asyncio.run(_run_gateway_turn_test(config, target, turn))
payload["turn"] = {"channel": target, "ok": ok, "response": message}
if not json_output:
print(f"\nTurn test ({target}): {'OK' if ok else 'FAIL'}")
print(message if ok else f"Error: {message}")
failed = failed or not ok
if json_output:
print(json.dumps(payload, indent=2))
if failed:
if not all_ok or gateway_secret_error:
raise typer.Exit(1)
@@ -1404,7 +1323,6 @@ Manage the gateway server: praisonai gateway <command>
[green]stop[/green] Stop a running gateway instance
[green]status[/green] Check gateway and daemon status
[green]doctor[/green] Validate channel credentials (pre-flight check)
[green]test[/green] One-shot readiness (probes + shell + optional turn)
[green]channels[/green] List channels from gateway.yaml (use --probe to check creds)
[green]send[/green] Send a test message to a channel
[green]hooks[/green] Manage inbound trigger hooks (add | list | remove)
@@ -100,68 +100,6 @@ def _load_praisonai_env_file() -> Dict[str, str]:
return loaded
# Runtime start-flag keys that materially change gateway behaviour and only
# exist on ``start`` (durability, concurrency ceiling, OpenAI-compat surface,
# lifecycle). Persisting them lets a direct ``restart`` replay the exact
# posture the process was started with, instead of silently reverting to
# defaults (#3349).
_START_FLAG_KEYS = (
"agent_file", "config_file", "drain_timeout", "max_concurrent_runs",
"queue_depth", "overflow_policy", "reliability", "openai_api", "mcp",
"identity_store", "scale_to_zero", "idle_minutes", "drain_marker",
)
def _start_flags_path(host: str, port: int) -> Path:
"""Path to the persisted start-flags file, keyed by the bound host+port.
Keying by host:port lets multiple gateways on one machine each keep their
own launch posture, matching how the PID lock is keyed (#3349).
"""
home = Path(os.environ.get("PRAISONAI_HOME") or (Path.home() / ".praisonai"))
safe_host = str(host).replace(":", "_")
return home / f"gateway.start.{safe_host}.{port}.json"
def _persist_start_flags(host: str, port: int, flags: Dict) -> None:
"""Persist the CLI-only runtime start flags so ``restart`` can replay them.
Only non-``None`` values are stored (``None`` means "fall back to YAML"),
so a restart reproduces the running process faithfully. Best-effort: a
write failure never blocks start (#3349).
"""
import json
path = _start_flags_path(host, port)
payload = {k: v for k, v in flags.items() if v is not None}
try:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2, sort_keys=True))
except OSError as exc: # pragma: no cover — advisory only
logger.warning("Could not persist gateway start flags to %s: %s", path, exc)
def load_start_flags(host: str, port: int) -> Dict:
"""Load the persisted start flags for a gateway bound to host:port.
Returns an empty dict when no artefact exists (or it is unreadable), so
a first-ever ``restart`` behaves exactly as before (#3349).
"""
import json
path = _start_flags_path(host, port)
if not path.exists():
return {}
try:
data = json.loads(path.read_text())
except (OSError, ValueError) as exc: # pragma: no cover — advisory only
logger.warning("Could not read gateway start flags from %s: %s", path, exc)
return {}
if not isinstance(data, dict):
return {}
return {k: v for k, v in data.items() if k in _START_FLAG_KEYS}
class GatewayHandler:
"""Handler for gateway CLI commands."""
@@ -241,35 +179,6 @@ class GatewayHandler:
# Load ~/.praisonai/.env BEFORE any config parsing or ${VAR}
# substitution — daemons don't inherit shell env.
_load_praisonai_env_file()
# Snapshot the CLI-only runtime flags this process was launched with so a
# later direct ``gateway restart`` can replay the exact posture (durable
# delivery, concurrency ceiling, OpenAI-compat surface, lifecycle)
# instead of silently reverting to defaults (#3349). The snapshot is only
# WRITTEN once startup validation passes and the gateway is about to bind
# (``_commit_start_flags`` below): a start attempt that fails import /
# config / agent-file validation must NOT clobber the running gateway's
# saved posture, or its next restart would replay the rejected attempt.
start_flags = {
"agent_file": agent_file,
"config_file": config_file,
"drain_timeout": drain_timeout,
"max_concurrent_runs": max_concurrent_runs,
"queue_depth": queue_depth,
"overflow_policy": overflow_policy,
"reliability": reliability,
"openai_api": openai_api,
"mcp": mcp,
"identity_store": identity_store,
"scale_to_zero": scale_to_zero,
"idle_minutes": idle_minutes,
"drain_marker": drain_marker,
}
def _commit_start_flags() -> None:
# Best-effort — never blocks start (#3349).
_persist_start_flags(host, port, start_flags)
logger.info(
"Gateway starting (host=%s port=%s config=%s agents=%s)",
host, port, config_file or "-", agent_file or "-",
@@ -337,10 +246,6 @@ class GatewayHandler:
if drain_marker is not None:
self._gateway._drain_marker_override = drain_marker
print(f"Loading gateway config from {config_file}")
# Config wiring validated; the gateway is about to bind. Persist the
# launch posture now so a restart can replay it, but only after the
# attempt has cleared validation (#3349).
_commit_start_flags()
try:
asyncio.run(self._gateway.start_with_config(config_file))
except KeyboardInterrupt:
@@ -422,10 +327,6 @@ class GatewayHandler:
print(f"Starting gateway on ws://{host}:{port}")
print("Press Ctrl+C to stop")
# Admission/agent wiring validated; the gateway is about to bind.
# Persist the launch posture now so a restart can replay it, but only
# after the attempt has cleared validation (#3349).
_commit_start_flags()
try:
asyncio.run(self._gateway.start())
except KeyboardInterrupt:
@@ -481,68 +481,14 @@ class KanbanDispatcher:
logger.warning(f"Error preparing worktree for task {task.id}: {e}")
return None
def _worktree_outstanding_work(self, path: str, branch: str) -> Optional[str]:
"""Detect work a worktree still holds that removal would destroy.
Lossless-only guard (Refinement 2): a dispatcher must never be the thing
that silently destroys work a worker produced but failed to integrate.
Returns a human-readable reason when the worktree is **dirty** (has
uncommitted changes) or its branch has commits **not yet in base**
(ahead-count > 0); returns ``None`` when the worktree is safe to remove.
Any error inspecting the tree is treated as "outstanding" so we err
toward preservation, never toward destruction.
"""
try:
status = self._run_git("status", "--porcelain", cwd=path)
if status.returncode != 0:
return f"could not read worktree status: {status.stderr.strip()}"
if status.stdout.strip():
return "worktree has uncommitted changes"
base = self._base_branch()
ahead = self._run_git(
"rev-list", "--count", f"{base}..{branch}", cwd=path
)
if ahead.returncode != 0:
return f"could not compute ahead-count: {ahead.stderr.strip()}"
count = ahead.stdout.strip()
if count and count != "0":
return f"branch has {count} commit(s) not in {base}"
return None
except Exception as e:
return f"error inspecting worktree: {e}"
def _remove_worktree(self, path: str, branch: Optional[str] = None,
*, force: bool = False) -> Optional[str]:
"""Tear down a task worktree, losslessly by default.
Unless ``force`` is set, refuse to remove a worktree that still holds
outstanding work (dirty tree or unmerged commits): keep it and return
the preservation reason so the caller can record it. Removal with
outstanding work is an explicit force operation only. Returns ``None``
only when the worktree was actually removed; otherwise returns a reason
string (preservation guard, failed ``git worktree remove``, or
exception) so the caller keeps tracking it instead of orphaning it.
"""
if not force and branch is not None:
reason = self._worktree_outstanding_work(path, branch)
if reason is not None:
logger.warning(
"Preserving worktree %s (not removing): %s", path, reason,
)
return reason
def _remove_worktree(self, path: str):
"""Tear down a task worktree (best-effort)."""
try:
result = self._run_git("worktree", "remove", "--force", path)
if result.returncode != 0:
# Removal failed: the worktree is still on disk. Surface the
# reason so the caller does not drop tracking and leave an
# orphaned worktree/branch behind.
detail = result.stderr.strip() or f"exit code {result.returncode}"
logger.warning(f"worktree remove failed for {path}: {detail}")
return f"worktree removal failed: {detail}"
logger.debug(f"worktree remove failed for {path}: {result.stderr.strip()}")
except Exception as e:
logger.warning(f"Error removing worktree {path}: {e}")
return f"worktree removal error: {e}"
return None
logger.debug(f"Error removing worktree {path}: {e}")
def _commit_worktree_changes(self, path: str, branch: str) -> None:
"""Commit any uncommitted edits the worker left in its worktree.
@@ -645,22 +591,9 @@ class KanbanDispatcher:
logger.warning(f"Task {task_id} blocked: merge conflict in {files}")
return True
# Clean merge: tear down the worktree losslessly. A non-None result
# means the worktree still exists on disk -- either the guard preserved
# outstanding work (dirty/unmerged) or the removal itself failed. In
# both cases keep the tracking entry and record the reason rather than
# dropping it and orphaning the worktree.
preserved = self._remove_worktree(path, branch)
if preserved is not None:
try:
store.add_comment(
task_id, self.worker_id,
f"worktree_preserved at {path}: {preserved}",
)
except Exception:
pass
else:
self._worktrees.pop(task_id, None)
# Clean merge: tear down the worktree.
self._remove_worktree(path)
self._worktrees.pop(task_id, None)
return False
async def _spawn_worker(self, task: Any, store: Any) -> bool:
@@ -1,305 +0,0 @@
"""Gateway preflight helpers — probes, shell wiring, and offline turn tests.
Shared by ``praisonai gateway doctor``, ``praisonai gateway test``, and
``praisonai doctor bots`` so probe/turn logic lives in one place (not the CLI
module).
"""
from __future__ import annotations
import asyncio
import os
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple
@dataclass
class ShellReadinessResult:
"""Outcome of offline shell wiring validation."""
ok: bool
message: str
issues: List[str] = field(default_factory=list)
def load_channels_mapping(config_path: str) -> dict:
"""Load the ``channels`` mapping from a gateway/bot YAML file."""
import yaml
with open(config_path) as f:
cfg = yaml.safe_load(f) or {}
return cfg.get("channels") or {}
def resolve_gateway_endpoint(config_path: str) -> Tuple[str, int]:
"""Resolve ``(host, port)`` from config and ``GATEWAY_PORT`` env."""
import yaml
with open(config_path) as f:
cfg = yaml.safe_load(f) or {}
gateway_cfg = cfg.get("gateway") or {}
host = str(gateway_cfg.get("host") or "127.0.0.1")
raw_port = gateway_cfg.get("port")
if raw_port is None:
raw_port = os.environ.get("GATEWAY_PORT", "8765")
try:
port = int(raw_port)
except (TypeError, ValueError):
port = 8765
return host, port
def probe_results_to_dict(results: dict) -> dict:
"""JSON-serializable per-channel probe payload."""
return {
name: r.to_dict() if hasattr(r, "to_dict") else vars(r)
for name, r in results.items()
}
def resolve_env_token(value):
"""Resolve a credential input to its value for probing."""
if isinstance(value, dict) and "source" in value and "id" in value:
try:
from praisonaiagents.secrets import resolve_secret
result = resolve_secret(value)
return result.value or ""
except Exception: # pragma: no cover — defensive
return ""
if isinstance(value, str) and value.startswith("${") and value.endswith("}"):
return os.environ.get(value[2:-1], "")
return value
def apply_probe_ca_bundle() -> None:
"""Point the probe HTTP client at a custom CA bundle if configured."""
preferred = os.environ.get("PRAISONAI_SSL_CA_BUNDLE")
ca_bundle = (
preferred
or os.environ.get("REQUESTS_CA_BUNDLE")
or os.environ.get("SSL_CERT_FILE")
)
if not ca_bundle:
return
if not os.path.exists(ca_bundle):
print(
f"Warning: CA bundle path '{ca_bundle}' does not exist — "
"SSL_CERT_FILE / REQUESTS_CA_BUNDLE not updated for probe."
)
return
if preferred and preferred == ca_bundle:
os.environ["SSL_CERT_FILE"] = ca_bundle
os.environ["REQUESTS_CA_BUNDLE"] = ca_bundle
else:
os.environ.setdefault("SSL_CERT_FILE", ca_bundle)
os.environ.setdefault("REQUESTS_CA_BUNDLE", ca_bundle)
async def probe_channels(channels: dict, timeout: float = 15.0) -> dict:
"""Probe each channel's credentials without starting message processing."""
try:
from praisonai_bot.cli.features.gateway import _load_praisonai_env_file
_load_praisonai_env_file()
except Exception: # pragma: no cover — defensive
pass
apply_probe_ca_bundle()
from praisonai_bot.bots import Bot
from praisonaiagents.bots import ProbeResult
async def _probe_one(name: str, ch_cfg: dict):
platform = ch_cfg.get("platform", name)
token = resolve_env_token(ch_cfg.get("token", ""))
extras = {
k: resolve_env_token(v)
for k, v in ch_cfg.items()
if k not in ("platform", "token")
}
try:
bot = Bot(platform, token=token, **extras)
return name, await asyncio.wait_for(bot.probe(), timeout=timeout)
except asyncio.TimeoutError:
return name, ProbeResult(
ok=False,
platform=platform,
error=f"probe timed out after {timeout:g}s",
)
except Exception as e: # pragma: no cover — defensive
return name, ProbeResult(ok=False, platform=platform, error=str(e))
results = await asyncio.gather(
*(_probe_one(name, ch_cfg or {}) for name, ch_cfg in channels.items())
)
return dict(results)
async def probe_channels_from_config(
config_path: str,
channel_filter: Optional[str] = None,
timeout: float = 15.0,
) -> dict:
"""Load config and probe channels, optionally filtering to one channel."""
channels = load_channels_mapping(config_path)
if channel_filter:
if channel_filter not in channels:
raise ValueError(f"Channel '{channel_filter}' not found in config")
channels = {channel_filter: channels[channel_filter]}
return await probe_channels(channels, timeout=timeout)
def run_shell_readiness_check(config_path: str) -> ShellReadinessResult:
"""Offline validation of ``allow_shell`` wiring (no LLM, no network)."""
import yaml
from praisonaiagents import Agent
from praisonaiagents.approval import get_approval_registry
from praisonaiagents.bots.config import BotConfig
from praisonai_bot.bots._defaults import apply_bot_smart_defaults, enable_shell_tools
with open(config_path) as f:
cfg = yaml.safe_load(f) or {}
channels = cfg.get("channels") or {}
agents_cfg = cfg.get("agents") or {}
shell_channels = [
name for name, ch in channels.items() if (ch or {}).get("allow_shell")
]
if not shell_channels:
return ShellReadinessResult(
ok=True,
message="No channels with allow_shell: true",
)
issues: List[str] = []
bot_config = BotConfig()
for channel_name in shell_channels:
ch_cfg = dict(channels[channel_name] or {})
ch_cfg.setdefault("platform", channel_name)
platform = str(ch_cfg.get("platform") or channel_name).lower()
routing = ch_cfg.get("routing") or cfg.get("routing") or {}
agent_id = routing.get("default") or next(iter(agents_cfg), None)
if not agent_id or agent_id not in agents_cfg:
issues.append(f"{channel_name}: no agent configured for shell routing")
continue
acfg = agents_cfg[agent_id] or {}
agent = Agent(
name=acfg.get("name", agent_id),
instructions=acfg.get("instructions", ""),
)
agent = apply_bot_smart_defaults(agent, bot_config)
agent = enable_shell_tools(agent, bot_config, ch_cfg, channel_type=platform)
tool_names = {
getattr(t, "name", None) or getattr(t, "__name__", "")
for t in (getattr(agent, "tools", None) or [])
}
if "execute_command" not in tool_names:
issues.append(f"{channel_name}: execute_command not in agent tools")
if getattr(agent, "_approval_backend", None) is None:
issues.append(f"{channel_name}: no approval backend on agent")
reg = get_approval_registry()
agent_name = getattr(agent, "name", None)
if agent_name and reg.get_backend(agent_name=agent_name) is None:
issues.append(f"{channel_name}: approval registry not synced for agent")
if issues:
return ShellReadinessResult(
ok=False,
message=f"Shell wiring issues on {len(issues)} check(s)",
issues=issues,
)
return ShellReadinessResult(
ok=True,
message=f"Shell wiring OK for {len(shell_channels)} channel(s)",
)
async def run_turn_test(
config_path: str,
channel_name: str,
prompt: str,
) -> Tuple[bool, str]:
"""Simulate one inbound agent turn offline via ``BotSessionManager.chat``.
Does **not** exercise Slack Bolt/socket handlers or @mention routing — only
the BotSessionManager path after manual shell setup.
"""
import yaml
from praisonaiagents import Agent
from praisonaiagents.bots.config import BotConfig
from praisonai_bot.bots._defaults import apply_bot_smart_defaults, enable_shell_tools
from praisonai_bot.bots._session import BotSessionManager
with open(config_path) as f:
cfg = yaml.safe_load(f) or {}
channels = cfg.get("channels") or {}
if channel_name not in channels:
return False, f"Channel '{channel_name}' not found in config"
ch_cfg = dict(channels[channel_name] or {})
ch_cfg.setdefault("platform", channel_name)
agents_cfg = cfg.get("agents") or {}
routing = ch_cfg.get("routing") or cfg.get("routing") or {}
agent_id = routing.get("default")
if not agent_id:
agent_id = next(iter(agents_cfg), None)
if not agent_id or agent_id not in agents_cfg:
return False, f"No agent configured for channel '{channel_name}'"
acfg = agents_cfg[agent_id] or {}
agent = Agent(
name=acfg.get("name", agent_id),
instructions=acfg.get("instructions", ""),
llm=acfg.get("model") or acfg.get("llm", "gpt-4o-mini"),
)
bot_config = BotConfig()
agent = apply_bot_smart_defaults(agent, bot_config)
platform = str(ch_cfg.get("platform") or channel_name).lower()
if ch_cfg.get("allow_shell"):
agent = enable_shell_tools(agent, bot_config, ch_cfg, channel_type=platform)
mgr = BotSessionManager(platform=platform)
try:
result = await mgr.chat(
agent,
"gateway-doctor-test",
prompt,
chat_id="gateway-doctor-test",
)
except Exception as exc:
return False, str(exc)
text = str(result or "").strip()
if not text:
return False, "Agent returned an empty response"
return True, text[:500]
def check_gateway_running(config_path: str, timeout: float = 5.0) -> Tuple[bool, str]:
"""Probe the gateway REST ``/info`` endpoint (advisory)."""
import urllib.error
import urllib.request
host, port = resolve_gateway_endpoint(config_path)
url = f"http://{host}:{port}/info"
try:
with urllib.request.urlopen(url, timeout=timeout) as resp:
if resp.status == 200:
return True, f"Gateway reachable at {url}"
return False, f"Gateway returned HTTP {resp.status} at {url}"
except urllib.error.URLError as exc:
return False, f"Gateway not reachable at {url}: {exc.reason}"
except Exception as exc: # pragma: no cover — defensive
return False, str(exc)
@@ -50,7 +50,7 @@ from praisonaiagents.session.store import DefaultSessionStore
logger = logging.getLogger(__name__)
from .unicode_utils import safe_error_message, safe_log_message, extract_root_cause_from_error
from .supervisor import ChannelState, ChannelSupervisor
from .supervisor import ChannelSupervisor
# WebSocket close code for a slow-consumer eviction. 1013 ("Try Again Later")
@@ -339,11 +339,6 @@ class GatewaySession:
# Stepper & Concurrency logic
_inbox: asyncio.Queue = field(default_factory=asyncio.Queue)
_is_executing: bool = False
# Issue #3379: for channel-originated sessions (e.g. a Telegram user), the
# ``"channel:target"`` address to proactively notify if an in-flight turn is
# interrupted by a gateway restart. ``None`` for direct-client sessions,
# which resume on reconnect and need no server-initiated channel notice.
_channel_target: Optional[str] = None
@property
def session_id(self) -> str:
@@ -356,11 +351,6 @@ class GatewaySession:
@property
def client_id(self) -> Optional[str]:
return self._client_id
@property
def channel_target(self) -> Optional[str]:
"""``"channel:target"`` to notify on restart, or ``None`` (Issue #3379)."""
return self._channel_target
@property
def is_active(self) -> bool:
@@ -497,7 +487,6 @@ class GatewaySession:
"events": [e.to_dict() for e in self._events[-100:]], # Keep last 100 events
"pending_inbox": pending_inbox,
"is_executing": self._is_executing,
"channel_target": self._channel_target,
}
@classmethod
@@ -545,9 +534,6 @@ class GatewaySession:
# Restore execution state
session._is_executing = data.get("is_executing", False)
# Issue #3379: restore the channel origin so a boot-time resume can
# notify the originating channel about an interrupted turn.
session._channel_target = data.get("channel_target")
return session
@@ -565,15 +551,6 @@ class GatewaySession:
"""Mark the session as currently executing an agent workflow."""
self._is_executing = status
def set_channel_target(self, channel_target: Optional[str]) -> None:
"""Record the ``"channel:target"`` origin for restart notification.
Issue #3379: channel-originated sessions call this so an interrupted
in-flight turn can be re-driven and the channel proactively notified on
the next boot, independent of any client reconnect.
"""
self._channel_target = channel_target
class ReloadAction(Enum):
"""Actions that can be taken during config reload."""
@@ -1997,15 +1974,7 @@ class WebSocketGateway:
# Start session cleanup task if persistence is enabled
if self._session_store:
await self._start_session_cleanup()
# Issue #3379: re-drive any turn that was in-flight when a previous
# process restarted, and proactively notify the originating channel.
# Mirrors the durable approval rehydrate above; a no-op without a store.
try:
await self._resume_interrupted_turns()
except Exception:
logger.exception("Failed to resume interrupted turns on boot")
logger.info(f"Gateway started on ws://{self._host}:{self._port}")
try:
@@ -4087,7 +4056,7 @@ class WebSocketGateway:
# Get supervision state
sup_status = supervision_status.get(name)
if sup_status:
entry = {
channel_status[name] = {
"platform": platform,
"running": running,
"supervision": {
@@ -4099,16 +4068,6 @@ class WebSocketGateway:
"manual_pause": sup_status.manual_pause,
}
}
# Issue #3348: a channel whose credential was rejected at
# runtime (revoked/rotated/expired token) is surfaced as the
# same degraded, redacted "credential unavailable" state the
# gateway already reports at boot (#3159) — so an operator sees
# *why* the channel is down and that it auto-recovers on repair,
# instead of a bare FAILED/reconnect with no explanation.
if sup_status.state == ChannelState.CREDENTIAL_UNAVAILABLE:
entry["status"] = "degraded"
entry["reason"] = "credential unavailable"
channel_status[name] = entry
else:
channel_status[name] = {
"platform": platform,
@@ -4497,194 +4456,6 @@ class WebSocketGateway:
self._cleanup_task = asyncio.create_task(_cleanup())
logger.info("Session cleanup task started (interval=1h)")
# ── Restart continuation (Issue #3379) ───────────────────────────
def _load_persisted_session_data(self, session_id: str) -> Optional[Dict[str, Any]]:
"""Return the latest persisted ``session_data`` snapshot for a session.
Mirrors the snapshot lookup in :meth:`create_session`: the newest
``system`` message carrying a ``session_data`` metadata blob wins.
Returns ``None`` when unavailable.
"""
if not self._session_store:
return None
try:
session_obj = self._session_store.get_session(session_id)
except Exception:
return None
session_data = None
for msg in getattr(session_obj, "messages", []) or []:
meta = getattr(msg, "metadata", None) or {}
if getattr(msg, "role", None) == "system" and "session_data" in meta:
session_data = meta["session_data"]
return session_data
@staticmethod
def _channel_target_for(data: Dict[str, Any]) -> Optional[str]:
"""Resolve the ``"channel:target"`` origin for a persisted session.
Issue #3379: prefer the explicit ``channel_target`` set via
:meth:`GatewaySession.set_channel_target`. When it is absent (e.g. a
session persisted before the origin was recorded, or one whose ingress
never called the setter), fall back to the already-persisted
``client_id`` when it *itself* encodes a ``"channel:target"`` origin —
the same convention the hook/scheduled delivery path (``deliver_to``)
uses. This keeps the field a live consumer of existing state rather than
dead code, without adding a new ingress parameter. Returns ``None`` for
direct-client sessions (no channel origin to notify).
"""
explicit = data.get("channel_target")
if explicit:
return explicit
client_id = data.get("client_id")
if isinstance(client_id, str) and ":" in client_id:
channel, target = (p.strip() for p in client_id.split(":", 1))
if channel and target:
return f"{channel}:{target}"
return None
async def _resume_interrupted_turns(self) -> int:
"""Re-drive turns interrupted by a restart and notify their channels.
Issue #3379: on boot — before serving new traffic — scan persisted
sessions for an in-flight turn (``is_executing`` or a non-empty
``pending_inbox``) that a previous process lost. For each such session
that carries a ``channel_target`` origin, deliver an exactly-once,
idempotent "interrupted — resuming" notice through the existing durable
outbox path (:meth:`_deliver_hook_reply`, whose idempotency key survives
restart) and re-drive the turn when the agent is available.
Idempotency is keyed on ``(session_id, event_cursor)`` — the ``run_epoch``
of the interrupted turn — so a boot that repeats (crash loop) notifies at
most once per interruption. A no-op without a durable session store.
"""
store = self._session_store
if not store:
return 0
lister = getattr(store, "list_sessions", None)
if not callable(lister):
return 0
# The store's ``list_sessions`` defaults to the 50 most-recent sessions
# (see DefaultSessionStore). A boot-time continuation scan must not
# silently drop interrupted turns beyond that window, so request a large
# explicit cap. Fall back to the no-arg form for stores whose signature
# does not accept ``limit``.
try:
summaries = lister(limit=1_000_000)
except TypeError:
try:
summaries = lister()
except Exception:
logger.exception(
"Failed to list sessions for restart continuation"
)
return 0
except Exception:
logger.exception("Failed to list sessions for restart continuation")
return 0
resumed = 0
for summary in summaries or []:
sid = summary.get("session_id") if isinstance(summary, dict) else summary
if not sid:
continue
data = self._load_persisted_session_data(sid)
if not data:
continue
interrupted = bool(data.get("is_executing")) or bool(
data.get("pending_inbox")
)
if not interrupted:
continue
channel_target = self._channel_target_for(data)
if not channel_target:
# Direct-client session: it resumes on reconnect (existing
# inbox replay). No server-initiated channel notice to emit.
continue
# Exactly-once notice keyed on the interrupted turn's run epoch
# (event_cursor). ``_deliver_hook_reply`` already folds this key into
# the durable outbox so a repeated boot does not re-notify.
run_epoch = data.get("event_cursor", 0)
notice = (
"I was interrupted by a restart - resuming your request. "
"If you don't get a reply shortly, please resend."
)
try:
await self._deliver_restart_notice(
channel_target, notice, sid, run_epoch,
)
except Exception:
logger.exception(
"Failed to deliver restart notice for session %s", sid,
)
# Best-effort re-drive: rehydrate the session and restart its queue
# if the agent is still registered. Where re-drive is unsafe (no
# agent), the notice above already asked the user to resend.
try:
agent_id = data.get("agent_id")
agent = self._agents.get(agent_id) if agent_id else None
if agent is not None:
session = GatewaySession.from_dict(
data, self.config.session_config.max_messages,
)
session._is_active = True
self._sessions[sid] = session
if not session._inbox.empty():
if not session._is_executing:
session.mark_executing(True)
asyncio.create_task(
self._run_session_queue(session, agent, session.client_id)
)
except Exception:
logger.exception("Failed to re-drive interrupted session %s", sid)
resumed += 1
if resumed:
logger.info("Resumed %d interrupted turn(s) on boot", resumed)
return resumed
async def _deliver_restart_notice(
self, channel_target: str, text: str, session_id: str, run_epoch: int,
) -> bool:
"""Deliver a restart-continuation notice exactly-once to a channel.
Reuses the durable delivery router with an idempotency key scoped to
``(session_id, run_epoch)`` so a repeated boot (crash loop) notifies the
originating channel at most once per interruption.
"""
if ":" not in channel_target:
logger.warning(
"channel_target '%s' must be 'channel:target'; skipping notice",
channel_target,
)
return False
channel, target = [p.strip() for p in channel_target.split(":", 1)]
idem = f"restart:{session_id}:{run_epoch}"
router = self.delivery_router
if router is not None:
return await router.deliver(
f"{channel}:{target}", text, idempotency_key=idem,
)
# Fallback: router unavailable — best-effort bare send (no dedup).
bot = self.get_channel_bot(channel)
if bot is None:
for name, b in self._channel_bots.items():
if name.lower() == channel.lower():
bot = b
break
if bot is None:
logger.warning("No channel bot '%s' for restart notice", channel)
return False
try:
await bot.send_message(target, text)
return True
except Exception as e: # noqa: BLE001
logger.error("Restart notice to %s:%s failed: %s", channel, target, e)
return False
# ── Multi-bot lifecycle ───────────────────────────────────────────
@staticmethod
@@ -5596,13 +5367,9 @@ class WebSocketGateway:
# type — so SlackApproval wiring in enable_shell_tools keys correctly.
platform = str(ch_cfg.get("platform") or channel_name).lower()
try:
from praisonai_bot.bots._defaults import (
apply_bot_smart_defaults,
enable_shell_tools,
)
from praisonai_bot.bots._defaults import enable_shell_tools
clone = agent.clone_for_channel() if hasattr(agent, "clone_for_channel") else agent
clone = apply_bot_smart_defaults(clone, config)
enabled = enable_shell_tools(
clone, config, ch_cfg, channel_type=platform
)
@@ -6142,8 +5909,6 @@ class WebSocketGateway:
Returns:
ReloadPlan with actions to take
"""
from praisonaiagents.gateway.config import is_hot_appliable
plan = ReloadPlan()
for path in changed_paths:
@@ -6151,15 +5916,7 @@ class WebSocketGateway:
if not parts:
continue
# Issue #3378: a closed set of paths (core registry) is safe to
# apply in place without restarting channels or agents. Everything
# else keeps falling through to the restart plans below, so restart
# stays the fail-safe default for unknown/structural changes.
if is_hot_appliable(path):
plan.hot_reload_paths.add(path)
continue
# Top-level section changes
if parts[0] == "agents":
if len(parts) == 1:
@@ -6196,82 +5953,7 @@ class WebSocketGateway:
plan.requires_full_restart()
return plan
def apply_hot_reload(
self, paths: Set[str], new_config: Dict[str, Any]
) -> None:
"""Apply hot-reloadable config paths in place (Issue #3378).
Implements the core ``SupportsHotReload`` protocol: mutate the running
gateway for the closed set of paths classified as hot-appliable (see
``praisonaiagents.gateway.config.HOT_APPLIABLE_KEYS``) without
restarting channels or agents. Unknown paths never reach here — they
fall through to the restart plans in :meth:`_build_reload_plan`.
Best-effort per key: a failure applying one key is logged and does not
abort the others or the surrounding reload.
"""
gw_cfg = new_config.get("gateway", {}) or {}
# Sentinel distinguishing "malformed value -> keep live state" from an
# explicit disable (None/absent). Assigning None on a malformed value
# would silently drop a previously-configured live drain window, so a
# bad hot-reload edit must be a no-op for that key (fail-safe).
_KEEP = object()
def _coerce_timeout(value: Any) -> Any:
"""Coerce a YAML/env timeout to a finite non-negative float.
Returns ``None`` for an explicit disable (value is ``None``) and the
``_KEEP`` sentinel for a malformed value so the caller preserves the
current live timeout instead of clearing it.
"""
if value is None:
return None
try:
import math as _math
coerced = float(value)
if not _math.isfinite(coerced) or coerced < 0:
raise ValueError
return coerced
except (TypeError, ValueError):
logger.warning(
"Invalid hot-reload timeout %r; keeping current value %s",
value,
self._reload_drain_timeout,
)
return _KEEP
for path in sorted(paths):
try:
if path == "gateway.logging.level" or path.startswith(
"gateway.logging.level."
):
level = (gw_cfg.get("logging") or {}).get("level")
if level is not None:
logging.getLogger("praisonai_bot").setLevel(level)
logger.info("Hot-applied logging level: %s", level)
elif path == "gateway.drain_timeout":
coerced = _coerce_timeout(gw_cfg.get("drain_timeout"))
if coerced is not _KEEP:
self._reload_drain_timeout = coerced
logger.info(
"Hot-applied drain_timeout: %s",
self._reload_drain_timeout,
)
elif path == "gateway.reload_drain_timeout":
coerced = _coerce_timeout(gw_cfg.get("reload_drain_timeout"))
if coerced is not _KEEP:
self._reload_drain_timeout = coerced
logger.info(
"Hot-applied reload_drain_timeout: %s",
self._reload_drain_timeout,
)
except Exception as e: # pragma: no cover - defensive
logger.warning("Failed to hot-apply %s: %s", path, e)
async def _restart_channel(
self,
channel_name: str,
@@ -6656,9 +6338,9 @@ class WebSocketGateway:
drain_timeout=self._reload_drain_timeout,
)
# Issue #3378: apply hot-reloadable paths in place — no restart.
# Apply hot-reload paths (future enhancement)
if plan.hot_reload_paths:
self.apply_hot_reload(plan.hot_reload_paths, new_cfg)
logger.info(f"Hot-reload paths (no-op for now): {plan.hot_reload_paths}")
# Surface a concise summary of what the reload did (Issue #2533).
if plan.full_restart:
@@ -21,7 +21,6 @@ from ..bots._resilience import (
ConnectionMonitor,
is_recoverable_error,
is_conflict_error,
is_credential_error,
sleep_with_abort,
)
from .health_monitor import ChannelHealthMonitor, HealthMonitorConfig
@@ -35,12 +34,6 @@ class ChannelState(Enum):
FAILED = "failed"
PAUSED = "paused"
STOPPED = "stopped"
# Issue #3348: the channel's credential was rejected at runtime (revoked,
# rotated, or expired token → 401/403). Distinct from FAILED: it is not
# terminal — the supervisor stops hammering the invalid token and waits for
# the credential to change (a reconnect/hot-reload), then auto-recovers
# without a full process restart. Surfaced as a degraded, redacted state.
CREDENTIAL_UNAVAILABLE = "credential-unavailable"
@dataclass
@@ -281,44 +274,7 @@ class ChannelSupervisor:
status.next_retry_at = None
logger.error(f"Channel '{name}' failed with conflict error: {e}")
break
elif is_credential_error(e, monitor.platform):
# Issue #3348: the credential was rejected at runtime (401/403,
# revoked/rotated/expired token). Do NOT hammer the invalid
# token in a tight reconnect loop and do NOT go terminal
# FAILED (which would require a full restart even after the
# token is fixed). Enter a first-class, redacted degraded
# state and wait for the credential to change — a reconnect()
# or config hot-reload sets the abort signal, wakes this loop,
# and the channel auto-recovers without a process restart. The
# reason is redacted: it never includes the token, only that
# the credential is unavailable.
status.state = ChannelState.CREDENTIAL_UNAVAILABLE
status.last_error = "credential unavailable"
status.last_error_time = time.time()
status.next_retry_at = None
logger.error(
f"Channel '{name}' credential rejected "
f"(auth failure); pausing until the credential changes. "
f"Fix/rotate the token and reconnect to auto-recover."
)
await abort_signal.wait() # Wait for reconnect/hot-reload.
abort_signal.clear()
# Re-source the credential onto the *same* bot instance
# before restarting from the parked state. Without this, a
# reconnect() would restart the bot still holding the
# rejected token and bounce straight back to
# CREDENTIAL_UNAVAILABLE. A config hot-reload swaps in a
# freshly-built bot (new token) via _start_single_channel, so
# this hook only matters for an out-of-band repair (e.g. a
# rotated env var) driving a bare reconnect(). Duck-typed and
# opt-in: bots that expose refresh_credentials() get a chance
# to re-read their token; all others are a no-op and rely on
# start() rebuilding the adapter (which re-resolves env-var
# tokens) or on hot-reload.
await self._refresh_credentials(name, bot)
continue
elif not is_recoverable:
# Non-recoverable error - treat as fatal
status.state = ChannelState.FAILED
@@ -344,12 +300,8 @@ class ChannelSupervisor:
# Aborted - check if paused or reconnect requested
continue
# Cleanup - don't overwrite terminal failure or degraded credential
# states (a credential-unavailable channel stays queryable as degraded).
if status.state not in (
ChannelState.FAILED,
ChannelState.CREDENTIAL_UNAVAILABLE,
):
# Cleanup - don't overwrite terminal failure states
if status.state != ChannelState.FAILED:
status.state = ChannelState.STOPPED
logger.info(f"Supervision ended for channel '{name}'")
@@ -368,33 +320,6 @@ class ChannelSupervisor:
# Unregister from health monitor
self._health_monitor.unregister_channel(name)
async def _refresh_credentials(self, name: str, bot: Any) -> None:
"""Give a parked bot a chance to re-source its credential before restart.
Issue #3348: when a channel wakes from ``CREDENTIAL_UNAVAILABLE`` (an
operator repaired the token and called ``reconnect()``), the *same* bot
instance is about to be restarted. If the bot exposes an opt-in
``refresh_credentials()`` it is invoked here so the repaired token is
picked up without a full restart. Duck-typed and best-effort: a bot
without the hook is a no-op (base ``Bot.start()`` already rebuilds its
adapter and re-resolves env-var tokens), and any error is swallowed so a
buggy hook cannot wedge supervision — the restart still proceeds and, if
the credential is still bad, the channel simply re-parks.
"""
refresh = getattr(bot, "refresh_credentials", None)
if not callable(refresh):
return
try:
result = refresh()
if asyncio.iscoroutine(result):
await result
logger.info(f"Channel '{name}' credential re-sourced before restart")
except Exception as exc:
logger.warning(
f"Channel '{name}' credential refresh hook failed "
f"(continuing with restart): {exc}"
)
async def _get_channel_health(self, name: str, bot: Any) -> Any:
"""Get health status for a channel.
@@ -212,59 +212,6 @@ class SQLiteKanbanStore:
(event_id, task_id, event_type, json.dumps(data))
)
@staticmethod
def _is_git_repo(path: Any) -> bool:
"""True when ``path`` points inside a git working tree.
Used purely to decide worktree auto-upgrade from *linkage* to a repo,
never from task content. Any error resolving the path is treated as
"not a repo" so a bad value can never fail the create.
"""
if not path or not isinstance(path, str):
return False
try:
import subprocess
result = subprocess.run(
["git", "-C", path, "rev-parse", "--is-inside-work-tree"],
capture_output=True, text=True,
)
return result.returncode == 0 and result.stdout.strip() == "true"
except Exception:
return False
def _resolve_workspace_kind(self, task_data: Dict[str, Any]) -> str:
"""Resolve the effective ``workspace_kind`` for a new task.
Structural auto-upgrade (Refinement 1): when a task is *linked to a git
repo* (``repo_path`` on the task, or ``repo_path`` in its metadata) and
the caller left ``workspace_kind`` unset/``"default"``, upgrade it to
``"worktree"`` so parallel workers get isolated checkouts without each
card author remembering a flag. Isolation intent is carried by linkage,
not by inspecting the task body.
Escape hatches: an explicit ``workspace_kind`` in ``task_data`` is
always respected; a board/task level ``auto_worktree: false`` (top-level
or in metadata) disables the upgrade entirely.
"""
explicit = task_data.get('workspace_kind')
# An explicitly-provided kind is always respected (escape hatch):
# only the *unset* case is eligible for structural auto-upgrade, so a
# caller that deliberately asks for "default" keeps shared-cwd.
if 'workspace_kind' in task_data and explicit is not None:
return explicit
metadata = task_data.get('metadata') or {}
auto = task_data.get('auto_worktree')
if auto is None and isinstance(metadata, dict):
auto = metadata.get('auto_worktree')
if auto is False:
return 'default'
repo_path = task_data.get('repo_path')
if not repo_path and isinstance(metadata, dict):
repo_path = metadata.get('repo_path')
if self._is_git_repo(repo_path):
return 'worktree'
return explicit or 'default'
def create_task(self, task_data: Dict[str, Any], *, idempotency_key: Optional[str] = None) -> Task:
"""Create a new task.
@@ -293,23 +240,6 @@ class SQLiteKanbanStore:
if max_retries < 1:
max_retries = None
# Resolve workspace_kind with structural auto-upgrade for repo-linked
# tasks (Refinement 1). Derive and persist the per-task branch up-front
# when isolated so the dispatcher just consumes it.
workspace_kind = self._resolve_workspace_kind(task_data)
branch = task_data.get('branch')
if workspace_kind == 'worktree' and not branch:
branch = f"kanban/{task_id}"
# Preserve the repo linkage that drove the upgrade: carry a top-level
# ``repo_path`` into metadata so the isolating repository is never
# discarded between create and dispatch. Metadata already present wins,
# so an explicit metadata.repo_path is left untouched.
metadata = dict(task_data.get('metadata') or {})
top_repo_path = task_data.get('repo_path')
if top_repo_path and not metadata.get('repo_path'):
metadata['repo_path'] = top_repo_path
task = Task(
id=task_id,
title=task_data['title'],
@@ -319,10 +249,10 @@ class SQLiteKanbanStore:
priority=task_data.get('priority', 0),
tenant=tenant,
board=board,
workspace_kind=workspace_kind,
branch=branch,
workspace_kind=task_data.get('workspace_kind', 'default'),
branch=task_data.get('branch'),
worktree_path=task_data.get('worktree_path'),
metadata=metadata,
metadata=task_data.get('metadata', {}),
max_retries=max_retries,
created_at=now,
updated_at=now
@@ -1,154 +0,0 @@
"""Tests for runtime credential/auth failure as a first-class state (Issue #3348).
A channel whose credential is rejected at runtime (revoked/rotated/expired
token -> 401/403) must:
* be classified distinctly from transient and generic-fatal errors;
* enter a named, redacted ``CREDENTIAL_UNAVAILABLE`` degraded state instead of
the terminal ``FAILED`` state (no full-restart requirement);
* stop hammering the invalid token in a tight reconnect loop; and
* auto-recover on ``reconnect()`` (credential repaired) without a restart.
"""
from __future__ import annotations
import asyncio
import pytest
from praisonai_bot.bots._resilience import is_credential_error, is_recoverable_error
from praisonai_bot.gateway.supervisor import ChannelState, ChannelSupervisor
class _AuthError(Exception):
"""Auth rejection carrying an HTTP-style status code."""
def __init__(self, message: str, status: int) -> None:
super().__init__(message)
self.status = status
def test_classifier_401_403_are_credential_not_recoverable():
assert is_credential_error(_AuthError("nope", 401)) is True
assert is_credential_error(_AuthError("nope", 403)) is True
# A credential rejection must NOT be treated as transient/recoverable.
assert is_recoverable_error(_AuthError("nope", 401)) is False
def test_classifier_platform_text_equivalents():
assert is_credential_error(Exception("invalid_auth"))
assert is_credential_error(Exception("Slack error: token_revoked"))
assert is_credential_error(Exception("401 Unauthorized"))
assert is_credential_error(Exception("The access token is invalid"))
def test_classifier_ignores_transient():
assert is_credential_error(Exception("connection reset by peer")) is False
assert is_credential_error(ConnectionError("timed out")) is False
def test_runtime_credential_rejection_enters_degraded_state():
"""A 401 at runtime -> CREDENTIAL_UNAVAILABLE (not FAILED), then recovers."""
recovered = asyncio.Event()
calls = {"n": 0}
async def start_fn(name, bot):
calls["n"] += 1
if calls["n"] == 1:
# First boot: credential rejected.
raise _AuthError("401 Unauthorized: invalid token", 401)
# After the operator fixes the token and reconnects: hold (running).
recovered.set()
await asyncio.Event().wait()
sup = ChannelSupervisor()
async def scenario():
task = asyncio.create_task(sup.run("slack", object(), start_fn))
# Wait until the supervisor parks in the credential-unavailable state.
for _ in range(200):
st = sup.get_status("slack")
if st.state == ChannelState.CREDENTIAL_UNAVAILABLE:
break
await asyncio.sleep(0.01)
st = sup.get_status("slack")
assert st.state == ChannelState.CREDENTIAL_UNAVAILABLE
# Redacted: never leaks the token / raw error text.
assert st.last_error == "credential unavailable"
assert "token" not in (st.last_error or "").lower()
# Operator repairs the credential and forces a reconnect: auto-recover.
assert sup.reconnect("slack") is True
await asyncio.wait_for(recovered.wait(), timeout=2.0)
assert sup.get_status("slack").state == ChannelState.RUNNING
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(asyncio.wait_for(scenario(), timeout=5.0))
def test_reconnect_resources_credential_via_refresh_hook():
"""Issue #3348: a bot exposing refresh_credentials() re-sources its token
on wake so the *same* instance recovers after an out-of-band repair — it
does not restart still holding the rejected credential."""
recovered = asyncio.Event()
class _Bot:
platform = "slack"
def __init__(self):
self.token = "bad"
self.refreshed = 0
def refresh_credentials(self):
# Operator repaired the credential out-of-band (e.g. rotated env).
self.token = "good"
self.refreshed += 1
bot = _Bot()
async def start_fn(name, b):
if b.token != "good":
raise _AuthError("401 Unauthorized: invalid token", 401)
recovered.set()
await asyncio.Event().wait()
sup = ChannelSupervisor()
async def scenario():
task = asyncio.create_task(sup.run("slack", bot, start_fn))
for _ in range(200):
if sup.get_status("slack").state == ChannelState.CREDENTIAL_UNAVAILABLE:
break
await asyncio.sleep(0.01)
assert sup.get_status("slack").state == ChannelState.CREDENTIAL_UNAVAILABLE
assert sup.reconnect("slack") is True
await asyncio.wait_for(recovered.wait(), timeout=2.0)
assert bot.refreshed == 1
assert sup.get_status("slack").state == ChannelState.RUNNING
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(asyncio.wait_for(scenario(), timeout=5.0))
def test_generic_fatal_still_failed():
"""A non-auth, non-recoverable error stays terminal FAILED (unchanged)."""
async def start_fn(name, bot):
raise ValueError("totally unexpected programming error")
sup = ChannelSupervisor()
asyncio.run(asyncio.wait_for(sup.run("slack", object(), start_fn), timeout=5.0))
assert sup.get_status("slack").state == ChannelState.FAILED
@@ -295,54 +295,6 @@ def test_manual_approval_replaces_stale_auto_backend(mock_execute_command, monke
assert decision.approved is False
@patch("praisonaiagents.tools.execute_command", create=True)
def test_enable_shell_syncs_approval_registry(mock_execute_command):
mock_execute_command.name = "execute_command"
agent = MagicMock()
agent.name = "assistant"
agent.tools = []
agent._perm_deny = frozenset({"execute_command"})
with patch("praisonaiagents.tools.execute_command", mock_execute_command):
enable_shell_tools(
agent,
config=BotConfig(),
ch_cfg={"allow_shell": True, "auto_approve_shell": True},
channel_type="slack",
)
from praisonaiagents.approval import get_approval_registry
from praisonaiagents.approval.backends import AutoApproveBackend
reg = get_approval_registry()
assert isinstance(reg.get_backend("assistant"), AutoApproveBackend)
@patch("praisonaiagents.tools.execute_command", create=True)
def test_enable_shell_critical_tool_honours_agent_approval(mock_execute_command):
"""Agent-level approval must flow through to @require_approval(critical)."""
from praisonaiagents import Agent
from praisonaiagents.approval import get_approval_registry, mark_approved
from praisonaiagents.tools import execute_command
mock_execute_command.name = "execute_command"
agent = Agent(name="assistant", tools=[])
enable_shell_tools(
agent,
config=BotConfig(),
ch_cfg={"allow_shell": True, "auto_approve_shell": True},
channel_type="slack",
)
decision = agent._resolve_approval_decision(
"execute_command", {"command": "uname -a"}, is_async=False
)
assert decision.approved is True
mark_approved("execute_command")
assert get_approval_registry().is_already_approved("execute_command")
assert execute_command in (agent.tools or [])
@patch("praisonaiagents.tools.execute_command", create=True)
def test_manual_approval_without_backend_denies_shell(mock_execute_command, monkeypatch):
"""auto_approve_shell=false with no pre-existing backend also fails closed."""
@@ -189,247 +189,6 @@ def test_reconciler_failure_falls_back_to_resend(tmp_path):
asyncio.run(run())
def test_recovery_annotator_labels_unreconciled_resend(tmp_path):
"""An unreconciled recovered entry is labelled by recovery_annotator."""
async def run():
path = tmp_path / "outbox.sqlite"
q1 = OutboundQueue(path=str(path))
await _seed_in_flight(q1)
q2 = OutboundQueue(path=str(path))
sent = []
async def sender(target, payload):
sent.append(payload)
return True
def annotator(entry, payload):
assert entry.idempotency_key == "msg-1"
out = dict(payload)
out["content"] = "[recovered] " + payload.get("content", "")
return out
succeeded, failed = await q2.drain(sender, recovery_annotator=annotator)
assert (succeeded, failed) == (1, 0)
assert len(sent) == 1
assert sent[0]["content"] == "[recovered] hi"
asyncio.run(run())
def test_recovery_annotator_skipped_when_reconciled(tmp_path):
"""A reconciled entry is not re-sent, so the annotator never fires."""
async def run():
path = tmp_path / "outbox.sqlite"
q1 = OutboundQueue(path=str(path))
await _seed_in_flight(q1)
q2 = OutboundQueue(path=str(path))
sent = []
annotated = []
async def sender(target, payload):
sent.append(payload)
return True
async def reconciler(entry):
return True # confirmed already delivered
def annotator(entry, payload):
annotated.append(entry.idempotency_key)
return payload
succeeded, failed = await q2.drain(
sender, reconciler=reconciler, recovery_annotator=annotator
)
assert (succeeded, failed) == (1, 0)
assert sent == [] # no re-dispatch
assert annotated == [] # annotator never consulted
asyncio.run(run())
def test_recovery_annotator_not_applied_to_fresh_entries(tmp_path):
"""Fresh pending entries are sent verbatim, never annotated."""
async def run():
q = _new_queue(tmp_path)
await q.enqueue("msg-fresh", "telegram:123", {"content": "hi"})
sent = []
def annotator(entry, payload):
out = dict(payload)
out["content"] = "LABELLED " + payload.get("content", "")
return out
async def sender(target, payload):
sent.append(payload)
return True
succeeded, failed = await q.drain(sender, recovery_annotator=annotator)
assert (succeeded, failed) == (1, 0)
assert sent[0]["content"] == "hi" # untouched
asyncio.run(run())
def test_durable_delivery_marks_recovered_resend(tmp_path):
"""DurableDelivery(mark_recovered=True) prefixes an unreconciled re-send."""
from praisonai_bot.bots._delivery import DurableDelivery, RECOVERED_PREFIX
async def run():
path = tmp_path / "outbox.sqlite"
q1 = OutboundQueue(path=str(path))
# Enqueue via DurableDelivery-style payload and force in-flight.
key = await q1.enqueue(
"msg-1",
"telegram:123",
{"content": "hello", "kwargs": {}, "idempotency_key": "msg-1"},
)
entry_id = int(key.split(":")[-1])
with q1._lock, closing(q1._connect()) as conn:
conn.execute(
"UPDATE outbound_queue SET status='sending' WHERE id=?",
(entry_id,),
)
conn.commit()
q2 = OutboundQueue(path=str(path))
received = []
class _Adapter:
async def send_message(self, channel_id, content, **kwargs):
received.append(content)
return True
delivery = DurableDelivery(
q2, _Adapter(), platform="telegram", mark_recovered=True
)
succeeded, failed = await delivery.drain_pending()
assert (succeeded, failed) == (1, 0)
assert len(received) == 1
assert received[0].startswith(RECOVERED_PREFIX)
assert received[0].endswith("hello")
asyncio.run(run())
def test_recovery_annotator_mutation_then_raise_sends_unlabelled(tmp_path):
"""An annotator that mutates its dict then raises must not leak a partial label.
The annotator receives its own fresh copy, so the send falls back to the
original unlabelled payload rather than a half-mutated one.
"""
async def run():
path = tmp_path / "outbox.sqlite"
q1 = OutboundQueue(path=str(path))
await _seed_in_flight(q1)
q2 = OutboundQueue(path=str(path))
sent = []
async def sender(target, payload):
sent.append(payload)
return True
def annotator(entry, payload):
payload["content"] = "MUTATED " + payload.get("content", "")
raise RuntimeError("annotator blew up after mutating")
succeeded, failed = await q2.drain(sender, recovery_annotator=annotator)
assert (succeeded, failed) == (1, 0)
assert len(sent) == 1
assert sent[0]["content"] == "hi" # original, not the mutated copy
asyncio.run(run())
def test_recovery_annotator_non_dict_return_sends_unlabelled(tmp_path):
"""An annotator returning a non-dict is rejected; the original is sent."""
async def run():
path = tmp_path / "outbox.sqlite"
q1 = OutboundQueue(path=str(path))
await _seed_in_flight(q1)
q2 = OutboundQueue(path=str(path))
sent = []
async def sender(target, payload):
sent.append(payload)
return True
def annotator(entry, payload):
return "not a dict"
succeeded, failed = await q2.drain(sender, recovery_annotator=annotator)
assert (succeeded, failed) == (1, 0)
assert sent[0]["content"] == "hi"
asyncio.run(run())
def test_recovered_label_survives_transient_resend_failure(tmp_path):
"""A recovered send that fails transiently stays labelled on the next retry.
First drain: send raises a recoverable error -> the entry must remain
'recovered' (not demoted to 'failed'). Second drain: send succeeds and the
recovery_annotator still labels the copy, so no unlabelled duplicate escapes.
"""
async def run():
path = tmp_path / "outbox.sqlite"
q1 = OutboundQueue(path=str(path))
await _seed_in_flight(q1)
q2 = OutboundQueue(path=str(path))
sent = []
attempts = {"n": 0}
async def sender(target, payload):
attempts["n"] += 1
if attempts["n"] == 1:
raise ConnectionError("transient network blip")
sent.append(payload)
return True
def annotator(entry, payload):
out = dict(payload)
out["content"] = "[recovered] " + payload.get("content", "")
return out
# First drain: transient failure keeps the entry recovered.
s1, f1 = await q2.drain(sender, recovery_annotator=annotator)
assert (s1, f1) == (0, 1)
assert sent == []
assert _read_status(q2) == "recovered"
# Clear the backoff gate so the retry is eligible immediately.
with q2._lock, closing(q2._connect()) as conn:
conn.execute(
"UPDATE outbound_queue SET last_attempt=NULL WHERE idempotency_key='msg-1'"
)
conn.commit()
# Second drain: succeeds AND is still labelled.
s2, f2 = await q2.drain(sender, recovery_annotator=annotator)
assert (s2, f2) == (1, 0)
assert len(sent) == 1
assert sent[0]["content"] == "[recovered] hi"
asyncio.run(run())
def test_status_for_reports_entry_state(tmp_path):
"""``status_for`` returns the current status, or None for an unknown key."""
async def run():
@@ -1,267 +0,0 @@
"""Gateway restart replays the CLI-only start flags (#3349).
A direct (non-service) ``gateway restart`` previously dropped the runtime
flags the process was started with (``--openai-api``, ``--reliability``,
``--max-concurrent-runs``, ...), silently reverting production settings to
defaults. ``start`` now persists those flags to ``~/.praisonai/`` keyed by
host:port, and ``restart`` replays them faithfully. These tests cover the
persist -> load round-trip and the restart replay wiring.
"""
import pytest
from praisonai_bot.cli.features.gateway import (
_persist_start_flags,
_start_flags_path,
load_start_flags,
)
def test_load_returns_empty_when_never_started(tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
assert load_start_flags("127.0.0.1", 8765) == {}
def test_persist_and_load_round_trip(tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
_persist_start_flags(
"127.0.0.1", 8765,
{
"openai_api": True,
"reliability": "production",
"max_concurrent_runs": 8,
"queue_depth": 32,
# None means "fall back to YAML" and must NOT be persisted.
"config_file": None,
},
)
loaded = load_start_flags("127.0.0.1", 8765)
assert loaded == {
"openai_api": True,
"reliability": "production",
"max_concurrent_runs": 8,
"queue_depth": 32,
}
assert "config_file" not in loaded
def test_flags_are_keyed_by_host_and_port(tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
_persist_start_flags("127.0.0.1", 8765, {"reliability": "production"})
_persist_start_flags("127.0.0.1", 9000, {"reliability": "off"})
assert load_start_flags("127.0.0.1", 8765) == {"reliability": "production"}
assert load_start_flags("127.0.0.1", 9000) == {"reliability": "off"}
# A gateway on a different port that was never started has no flags.
assert load_start_flags("127.0.0.1", 1234) == {}
def test_unknown_keys_are_ignored_on_load(tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
path = _start_flags_path("127.0.0.1", 8765)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text('{"reliability": "production", "totally_unknown": "x"}')
loaded = load_start_flags("127.0.0.1", 8765)
assert loaded == {"reliability": "production"}
def test_corrupt_file_loads_as_empty(tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
path = _start_flags_path("127.0.0.1", 8765)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("{not valid json")
assert load_start_flags("127.0.0.1", 8765) == {}
def test_restart_replays_persisted_flags(tmp_path, monkeypatch):
"""restart loads persisted flags and forwards them to start()."""
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
from praisonai_bot.cli.commands import gateway as gw_cmd
# A production start persisted these flags.
_persist_start_flags(
"127.0.0.1", 8765,
{
"openai_api": True,
"reliability": "production",
"max_concurrent_runs": 8,
},
)
# No installed daemon -> take the direct relaunch path. restart() imports
# these names inside the function, so patch them at their source modules.
import praisonai_bot.daemon as daemon_mod
import praisonai_bot.cli.features.gateway as feat_mod
monkeypatch.setattr(
daemon_mod, "get_daemon_status", lambda: {"installed": False}
)
captured = {}
class _FakeHandler:
def stop(self, *a, **k):
captured["stopped"] = True
def start(self, *a, **k):
captured["start_kwargs"] = k
return 0
monkeypatch.setattr(feat_mod, "GatewayHandler", _FakeHandler)
gw_cmd.gateway_restart(host="127.0.0.1", port=8765, drain_timeout=10.0)
kwargs = captured["start_kwargs"]
assert kwargs["openai_api"] is True
assert kwargs["reliability"] == "production"
assert kwargs["max_concurrent_runs"] == 8
# Explicit restart drain_timeout is applied.
assert kwargs["drain_timeout"] == 10.0
def test_restart_explicit_config_overrides_persisted(tmp_path, monkeypatch):
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
from praisonai_bot.cli.commands import gateway as gw_cmd
_persist_start_flags(
"127.0.0.1", 8765,
{"config_file": "old.yaml", "reliability": "production"},
)
import praisonai_bot.daemon as daemon_mod
import praisonai_bot.cli.features.gateway as feat_mod
monkeypatch.setattr(
daemon_mod, "get_daemon_status", lambda: {"installed": False}
)
captured = {}
class _FakeHandler:
def stop(self, *a, **k):
pass
def start(self, *a, **k):
captured["start_kwargs"] = k
return 0
monkeypatch.setattr(feat_mod, "GatewayHandler", _FakeHandler)
gw_cmd.gateway_restart(
host="127.0.0.1", port=8765, config="new.yaml", drain_timeout=10.0
)
kwargs = captured["start_kwargs"]
# Explicit --config wins over the persisted one.
assert kwargs["config_file"] == "new.yaml"
# Other persisted flags are still replayed.
assert kwargs["reliability"] == "production"
def test_restart_omitted_drain_timeout_replays_persisted(tmp_path, monkeypatch):
"""An omitted --drain-timeout must replay the persisted value, not 10s (#3349).
Regression for the silent-shortening hazard: Typer's None default must fall
back to the persisted start value for both the OLD-process drain and the
relaunch, so a production drain window survives a restart.
"""
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
from praisonai_bot.cli.commands import gateway as gw_cmd
_persist_start_flags("127.0.0.1", 8765, {"drain_timeout": 90.0})
import praisonai_bot.daemon as daemon_mod
import praisonai_bot.cli.features.gateway as feat_mod
monkeypatch.setattr(
daemon_mod, "get_daemon_status", lambda: {"installed": False}
)
captured = {}
class _FakeHandler:
def stop(self, *a, **k):
captured["stop_kwargs"] = k
def start(self, *a, **k):
captured["start_kwargs"] = k
return 0
monkeypatch.setattr(feat_mod, "GatewayHandler", _FakeHandler)
# drain_timeout omitted -> None (Typer default).
gw_cmd.gateway_restart(host="127.0.0.1", port=8765, drain_timeout=None)
# The relaunch replays the persisted 90s window, not a fixed 10s.
assert captured["start_kwargs"]["drain_timeout"] == 90.0
# The OLD process is drained with the same persisted window.
assert captured["stop_kwargs"]["drain_timeout"] == 90.0
def test_restart_explicit_drain_timeout_overrides_persisted(tmp_path, monkeypatch):
"""An explicit --drain-timeout still wins over the persisted value (#3349)."""
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
from praisonai_bot.cli.commands import gateway as gw_cmd
_persist_start_flags("127.0.0.1", 8765, {"drain_timeout": 90.0})
import praisonai_bot.daemon as daemon_mod
import praisonai_bot.cli.features.gateway as feat_mod
monkeypatch.setattr(
daemon_mod, "get_daemon_status", lambda: {"installed": False}
)
captured = {}
class _FakeHandler:
def stop(self, *a, **k):
captured["stop_kwargs"] = k
def start(self, *a, **k):
captured["start_kwargs"] = k
return 0
monkeypatch.setattr(feat_mod, "GatewayHandler", _FakeHandler)
gw_cmd.gateway_restart(host="127.0.0.1", port=8765, drain_timeout=5.0)
assert captured["start_kwargs"]["drain_timeout"] == 5.0
assert captured["stop_kwargs"]["drain_timeout"] == 5.0
def test_restart_default_drain_timeout_when_none_persisted(tmp_path, monkeypatch):
"""With no persisted drain value and no flag, the OLD process drains at 10s."""
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
from praisonai_bot.cli.commands import gateway as gw_cmd
_persist_start_flags("127.0.0.1", 8765, {"reliability": "production"})
import praisonai_bot.daemon as daemon_mod
import praisonai_bot.cli.features.gateway as feat_mod
monkeypatch.setattr(
daemon_mod, "get_daemon_status", lambda: {"installed": False}
)
captured = {}
class _FakeHandler:
def stop(self, *a, **k):
captured["stop_kwargs"] = k
def start(self, *a, **k):
captured["start_kwargs"] = k
return 0
monkeypatch.setattr(feat_mod, "GatewayHandler", _FakeHandler)
gw_cmd.gateway_restart(host="127.0.0.1", port=8765, drain_timeout=None)
# No persisted drain window -> OLD process falls back to the 10s default.
assert captured["stop_kwargs"]["drain_timeout"] == 10.0
# ...and no drain_timeout is forced into the relaunch (falls back to YAML).
assert "drain_timeout" not in captured["start_kwargs"]
@@ -373,101 +373,3 @@ def test_start_no_preflight_skips_probe(monkeypatch, tmp_path):
result = runner.invoke(app, ["start", "--config", str(cfg), "--no-preflight"])
assert result.exit_code == 0
assert started.get("called") is True
def test_doctor_json_single_document_with_turn(monkeypatch, tmp_path):
"""--json with --turn must emit one parseable document (#gateway-readiness)."""
typer_testing = pytest.importorskip("typer.testing")
import json
async def all_ok_probe(self):
return ProbeResult(ok=True, platform=self._platform, bot_username="bot")
monkeypatch.setattr(Bot, "probe", all_ok_probe)
monkeypatch.setattr(
"praisonai_bot.cli.commands.gateway._check_gateway_secret_strength",
lambda _cfg: None,
)
async def fake_turn(config_path, channel_name, prompt):
return True, "turn-ok"
monkeypatch.setattr(
"praisonai_bot.cli.commands.gateway._run_gateway_turn_test",
fake_turn,
)
cfg = tmp_path / "gateway.yaml"
cfg.write_text("channels:\n slack:\n platform: slack\n token: s\n")
from praisonai_bot.cli.commands.gateway import app
runner = typer_testing.CliRunner()
result = runner.invoke(
app,
["doctor", "--config", str(cfg), "--json", "--channel", "slack", "--turn", "hi"],
)
assert result.exit_code == 0, result.stdout + result.stderr
payload = json.loads(result.stdout.strip())
assert "probes" in payload
assert payload["turn"]["ok"] is True
assert payload["turn"]["response"] == "turn-ok"
def test_doctor_turn_runs_when_other_channel_fails(monkeypatch, tmp_path):
"""--channel slack --turn runs when slack ok even if telegram fails."""
typer_testing = pytest.importorskip("typer.testing")
async def mixed_probe(self):
if self._platform == "slack":
return ProbeResult(ok=True, platform="slack", bot_username="slackbot")
return ProbeResult(ok=False, platform=self._platform, error="bad")
monkeypatch.setattr(Bot, "probe", mixed_probe)
async def fake_turn(config_path, channel_name, prompt):
return True, "slack-turn"
monkeypatch.setattr(
"praisonai_bot.cli.commands.gateway._run_gateway_turn_test",
fake_turn,
)
cfg = tmp_path / "gateway.yaml"
cfg.write_text(
"channels:\n"
" telegram:\n platform: telegram\n token: t\n"
" slack:\n platform: slack\n token: s\n"
)
from praisonai_bot.cli.commands.gateway import app
runner = typer_testing.CliRunner()
result = runner.invoke(
app,
["doctor", "--config", str(cfg), "--channel", "slack", "--turn", "hi"],
)
assert result.exit_code == 1 # telegram still failed overall
assert "Turn test (slack): OK" in result.stdout
assert "slack-turn" in result.stdout
def test_gateway_test_command(monkeypatch, tmp_path):
typer_testing = pytest.importorskip("typer.testing")
async def all_ok_probe(self):
return ProbeResult(ok=True, platform=self._platform, bot_username="bot")
monkeypatch.setattr(Bot, "probe", all_ok_probe)
cfg = tmp_path / "gateway.yaml"
cfg.write_text("channels:\n slack:\n platform: slack\n token: s\n")
from praisonai_bot.cli.commands.gateway import app
runner = typer_testing.CliRunner()
result = runner.invoke(app, ["test", "--config", str(cfg)])
assert result.exit_code == 0
assert "shell wiring" in result.stdout
assert "slack" in result.stdout
@@ -256,168 +256,6 @@ def test_integration_exception_blocks_task(git_repo):
assert ("t_e", "blocked") in store.moves
def test_dirty_worktree_preserved_on_cleanup(git_repo):
"""A worktree left dirty after integration is preserved, not deleted."""
d = _dispatcher_in(git_repo)
store = _FakeStore()
path = d._prepare_worktree(_FakeTask("t_dirty", "worktree"), store)
d._worktrees = {"t_dirty": (path, "kanban/t_dirty")}
(git_repo / ".wt" / "t_dirty" / "new.txt").write_text("committed\n")
_git(path, "add", "-A")
_git(path, "commit", "-m", "add new")
# Simulate a post-integration dirty tree by patching removal-time status:
# leave an uncommitted edit in the worktree just before cleanup.
original_integrate = d._try_integrate
def _integrate(branch):
ok, files = original_integrate(branch)
# Introduce an uncommitted change after the merge, before removal.
(git_repo / ".wt" / "t_dirty" / "leftover.txt").write_text("dirty\n")
return ok, files
d._try_integrate = _integrate
conflicted = d._integrate_worktree("t_dirty", store)
assert conflicted is False
# Worktree preserved because it is dirty; a preservation comment recorded.
assert os.path.exists(path)
assert any("worktree_preserved" in c[2] for c in store.comments)
def test_unpushed_commits_block_removal(git_repo):
"""Removal is refused when the branch has commits not in base."""
d = _dispatcher_in(git_repo)
store = _FakeStore()
path = d._prepare_worktree(_FakeTask("t_ahead", "worktree"), store)
branch = "kanban/t_ahead"
(git_repo / ".wt" / "t_ahead" / "a.txt").write_text("a\n")
_git(path, "add", "-A")
_git(path, "commit", "-m", "unmerged work")
# The branch is 1 commit ahead of base (never integrated) -> preserve.
reason = d._remove_worktree(path, branch)
assert reason is not None
assert "not in" in reason
assert os.path.exists(path)
def test_lossless_removal_of_clean_worktree(git_repo):
"""A fully-integrated, clean worktree is removed (no false preservation)."""
d = _dispatcher_in(git_repo)
store = _FakeStore()
path = d._prepare_worktree(_FakeTask("t_clean", "worktree"), store)
d._worktrees = {"t_clean": (path, "kanban/t_clean")}
(git_repo / ".wt" / "t_clean" / "c.txt").write_text("c\n")
_git(path, "add", "-A")
_git(path, "commit", "-m", "add c")
conflicted = d._integrate_worktree("t_clean", store)
assert conflicted is False
# Branch merged into base -> nothing outstanding -> removed.
assert not os.path.exists(path)
assert "t_clean" not in d._worktrees
assert not any("worktree_preserved" in c[2] for c in store.comments)
def test_force_removal_ignores_outstanding_work(git_repo):
"""force=True removes even a worktree with unmerged commits (explicit)."""
d = _dispatcher_in(git_repo)
store = _FakeStore()
path = d._prepare_worktree(_FakeTask("t_force", "worktree"), store)
(git_repo / ".wt" / "t_force" / "f.txt").write_text("f\n")
_git(path, "add", "-A")
_git(path, "commit", "-m", "unmerged")
reason = d._remove_worktree(path, "kanban/t_force", force=True)
assert reason is None
assert not os.path.exists(path)
def test_failed_removal_returns_reason(git_repo):
"""A nonzero `git worktree remove` surfaces a reason (not a silent None)."""
d = _dispatcher_in(git_repo)
path = d._prepare_worktree(_FakeTask("t_rmfail", "worktree"), _FakeStore())
original = d._run_git
def _run_git(*args, cwd=None):
# Fail only the removal; let status/ahead checks pass so we reach it.
if args[:2] == ("worktree", "remove"):
class _R:
returncode = 1
stdout = ""
stderr = "worktree is locked"
return _R()
return original(*args, cwd=cwd)
d._run_git = _run_git
reason = d._remove_worktree(path, "kanban/t_rmfail")
# A failed removal must NOT report success (None); it returns a reason so
# the caller keeps tracking the still-on-disk worktree.
assert reason is not None
assert "failed" in reason
assert os.path.exists(path)
def test_failed_removal_keeps_tracking_and_comments(git_repo):
"""A failed removal after a clean merge preserves tracking + records it."""
d = _dispatcher_in(git_repo)
store = _FakeStore()
path = d._prepare_worktree(_FakeTask("t_track", "worktree"), store)
d._worktrees = {"t_track": (path, "kanban/t_track")}
(git_repo / ".wt" / "t_track" / "c.txt").write_text("c\n")
_git(path, "add", "-A")
_git(path, "commit", "-m", "add c")
original = d._run_git
def _run_git(*args, cwd=None):
if args[:2] == ("worktree", "remove"):
class _R:
returncode = 1
stdout = ""
stderr = "worktree is locked"
return _R()
return original(*args, cwd=cwd)
d._run_git = _run_git
conflicted = d._integrate_worktree("t_track", store)
assert conflicted is False
# Removal failed => entry must NOT be dropped (no orphaned worktree).
assert "t_track" in d._worktrees
assert any("worktree_preserved" in c[2] for c in store.comments)
def test_removal_exception_returns_reason(git_repo):
"""An exception during removal is surfaced, not swallowed as success."""
d = _dispatcher_in(git_repo)
path = d._prepare_worktree(_FakeTask("t_exc", "worktree"), _FakeStore())
original = d._run_git
def _run_git(*args, cwd=None):
if args[:2] == ("worktree", "remove"):
raise RuntimeError("git exploded")
return original(*args, cwd=cwd)
d._run_git = _run_git
reason = d._remove_worktree(path, "kanban/t_exc")
assert reason is not None
assert "error" in reason
def test_unsafe_task_id_refused(git_repo):
"""Traversal / invalid-ref ids do not create a worktree (no fail-open)."""
d = _dispatcher_in(git_repo)
@@ -1,101 +0,0 @@
"""Tests for praisonai_bot.gateway.preflight helpers."""
import asyncio
from unittest.mock import AsyncMock, patch
import pytest
from praisonaiagents.bots import ProbeResult
def test_run_shell_readiness_no_shell_channels(tmp_path):
cfg = tmp_path / "bot.yaml"
cfg.write_text(
"channels:\n"
" slack:\n platform: slack\n token: x\n"
"agents:\n assistant:\n instructions: hi\n"
)
from praisonai_bot.gateway.preflight import run_shell_readiness_check
result = run_shell_readiness_check(str(cfg))
assert result.ok is True
assert "No channels with allow_shell" in result.message
def test_run_shell_readiness_wires_execute_command(tmp_path):
cfg = tmp_path / "bot.yaml"
cfg.write_text(
"channels:\n"
" slack:\n"
" platform: slack\n"
" token: x\n"
" allow_shell: true\n"
" auto_approve_shell: true\n"
"agents:\n"
" assistant:\n"
" name: assistant\n"
" instructions: hi\n"
"routing:\n default: assistant\n"
)
from praisonai_bot.gateway.preflight import run_shell_readiness_check
result = run_shell_readiness_check(str(cfg))
assert result.ok is True
assert result.issues == []
def test_probe_channels_from_config_filters_channel(tmp_path, monkeypatch):
cfg = tmp_path / "bot.yaml"
cfg.write_text(
"channels:\n"
" telegram:\n platform: telegram\n token: t\n"
" slack:\n platform: slack\n token: s\n"
)
async def fake_probe(channels, timeout=15.0):
return {name: ProbeResult(ok=True, platform=name, bot_username="bot") for name in channels}
monkeypatch.setattr(
"praisonai_bot.gateway.preflight.probe_channels",
fake_probe,
)
from praisonai_bot.gateway.preflight import probe_channels_from_config
results = asyncio.run(probe_channels_from_config(str(cfg), channel_filter="slack"))
assert set(results) == {"slack"}
def test_check_gateway_running_unreachable(tmp_path):
cfg = tmp_path / "bot.yaml"
cfg.write_text("gateway:\n host: 127.0.0.1\n port: 59999\n")
from praisonai_bot.gateway.preflight import check_gateway_running
ok, msg = check_gateway_running(str(cfg), timeout=1.0)
assert ok is False
assert "59999" in msg
def test_run_turn_test_mocked(tmp_path, monkeypatch):
cfg = tmp_path / "bot.yaml"
cfg.write_text(
"channels:\n"
" slack:\n"
" platform: slack\n"
" token: x\n"
"agents:\n"
" assistant:\n"
" name: assistant\n"
" instructions: hi\n"
"routing:\n default: assistant\n"
)
mock_chat = AsyncMock(return_value="OK from test")
monkeypatch.setattr(
"praisonai_bot.bots._session.BotSessionManager.chat",
mock_chat,
)
from praisonai_bot.gateway.preflight import run_turn_test
ok, message = asyncio.run(run_turn_test(str(cfg), "slack", "Say OK"))
assert ok is True
assert message == "OK from test"
@@ -139,78 +139,6 @@ def test_reload_failure_recorded_via_locked(tmp_path):
assert gw._config_path == str(bad)
# ── Hot-reload apply (Issue #3378) ─────────────────────────────────────────
def test_hot_appliable_paths_classified_as_hot_not_restart():
"""Hot-appliable keys go to hot_reload_paths, not a restart plan."""
gw = WebSocketGateway()
plan = gw._build_reload_plan(
{"gateway.logging.level", "channels.telegram.enabled", "gateway.drain_timeout"}
)
assert plan.hot_reload_paths == {"gateway.logging.level", "gateway.drain_timeout"}
assert "telegram" in plan.restart_channels
assert not plan.full_restart
def test_unknown_gateway_key_still_full_restart():
"""A gateway key not on the hot list stays fail-safe (full restart)."""
gw = WebSocketGateway()
plan = gw._build_reload_plan({"gateway.some_unknown_knob"})
assert plan.full_restart
assert not plan.hot_reload_paths
def test_apply_hot_reload_mutates_live_state():
"""apply_hot_reload applies logging level and drain timeouts in place."""
import logging
gw = WebSocketGateway()
gw._reload_drain_timeout = None
new_cfg = {
"gateway": {
"logging": {"level": "DEBUG"},
"reload_drain_timeout": 7,
}
}
gw.apply_hot_reload(
{"gateway.logging.level", "gateway.reload_drain_timeout"}, new_cfg
)
assert logging.getLogger("praisonai_bot").level == logging.DEBUG
assert gw._reload_drain_timeout == 7.0
def test_apply_hot_reload_invalid_timeout_preserves_live_value():
"""A malformed timeout is ignored and keeps the current live value.
Regression for the P1: assigning ``None`` on a bad edit would silently drop
a previously-configured drain window, so subsequent channel reloads would
skip their drain. A malformed hot-reload edit must be a no-op for that key.
"""
gw = WebSocketGateway()
gw._reload_drain_timeout = 5.0
gw.apply_hot_reload(
{"gateway.drain_timeout"}, {"gateway": {"drain_timeout": "oops"}}
)
assert gw._reload_drain_timeout == 5.0
def test_apply_hot_reload_explicit_none_disables_timeout():
"""An explicit None/absent value is an intentional disable (-> None)."""
gw = WebSocketGateway()
gw._reload_drain_timeout = 5.0
gw.apply_hot_reload(
{"gateway.drain_timeout"}, {"gateway": {"drain_timeout": None}}
)
assert gw._reload_drain_timeout is None
def test_gateway_conforms_to_supports_hot_reload_protocol():
"""The gateway satisfies the core SupportsHotReload protocol."""
from praisonaiagents.gateway.config import SupportsHotReload
assert isinstance(WebSocketGateway(), SupportsHotReload)
if __name__ == "__main__":
import pytest
@@ -1,240 +0,0 @@
#!/usr/bin/env python3
"""Tests for restart continuation of interrupted in-flight turns (Issue #3379).
Verifies that a turn left in-flight when the gateway restarts is re-driven on
the next boot and that the originating channel is proactively (and idempotently)
notified — independent of any client reconnect.
"""
import asyncio
import pytest
from praisonai_bot.gateway.server import GatewaySession, WebSocketGateway
def test_channel_target_round_trips_through_serialization():
"""``channel_target`` survives to_dict/from_dict so boot resume can notify."""
session = GatewaySession(
_session_id="s1",
_agent_id="agent-1",
_client_id=None,
)
session.set_channel_target("telegram:12345")
session.mark_executing(True)
data = session.to_dict()
assert data["channel_target"] == "telegram:12345"
restored = GatewaySession.from_dict(data)
assert restored.channel_target == "telegram:12345"
assert restored._is_executing is True
def test_channel_target_defaults_none_for_legacy_data():
"""Sessions persisted before this field deserialize with no target."""
legacy = {
"session_id": "s2",
"agent_id": "agent-1",
"is_executing": True,
"pending_inbox": [],
}
restored = GatewaySession.from_dict(legacy)
assert restored.channel_target is None
class _FakeStore:
"""Minimal session store exposing the two boot-scan methods we use."""
def __init__(self, sessions):
# sessions: {session_id: session_data_dict}
self._sessions = sessions
def list_sessions(self, limit: int = 50):
# Mirror DefaultSessionStore: cap the returned window at ``limit`` so a
# boot scan that fails to raise the cap would miss older sessions.
return [{"session_id": sid} for sid in self._sessions][:limit]
def get_session(self, session_id):
data = self._sessions.get(session_id)
class _Msg:
def __init__(self, role, metadata):
self.role = role
self.metadata = metadata
class _SessionObj:
def __init__(self, messages):
self.messages = messages
if data is None:
return _SessionObj([])
return _SessionObj([_Msg("system", {"session_data": data})])
def _make_gateway_with_store(store):
gateway = WebSocketGateway(port=0)
gateway._session_store = store
return gateway
def test_resume_notifies_channel_for_interrupted_turn():
"""An interrupted channel turn triggers exactly one idempotent notice."""
session = GatewaySession(_session_id="s3", _agent_id="agent-1")
session.set_channel_target("telegram:999")
session.mark_executing(True)
data = session.to_dict()
gateway = _make_gateway_with_store(_FakeStore({"s3": data}))
sent = []
async def _fake_notice(channel_target, text, session_id, run_epoch):
sent.append((channel_target, text, session_id, run_epoch))
return True
gateway._deliver_restart_notice = _fake_notice
resumed = asyncio.run(gateway._resume_interrupted_turns())
assert resumed == 1
assert len(sent) == 1
assert sent[0][0] == "telegram:999"
assert sent[0][2] == "s3"
def test_resume_skips_sessions_without_channel_target():
"""Direct-client interrupted sessions are not channel-notified."""
session = GatewaySession(_session_id="s4", _agent_id="agent-1")
session.mark_executing(True) # interrupted but no channel origin
data = session.to_dict()
gateway = _make_gateway_with_store(_FakeStore({"s4": data}))
sent = []
async def _fake_notice(*args):
sent.append(args)
return True
gateway._deliver_restart_notice = _fake_notice
resumed = asyncio.run(gateway._resume_interrupted_turns())
assert resumed == 0
assert sent == []
def test_resume_skips_completed_sessions():
"""A session with no in-flight work is left untouched."""
session = GatewaySession(_session_id="s5", _agent_id="agent-1")
session.set_channel_target("telegram:1")
session.mark_executing(False) # not executing, empty inbox
data = session.to_dict()
gateway = _make_gateway_with_store(_FakeStore({"s5": data}))
sent = []
async def _fake_notice(*args):
sent.append(args)
return True
gateway._deliver_restart_notice = _fake_notice
resumed = asyncio.run(gateway._resume_interrupted_turns())
assert resumed == 0
assert sent == []
def test_resume_is_noop_without_store():
"""No durable store => no-op, no crash."""
gateway = WebSocketGateway(port=0)
gateway._session_store = None
resumed = asyncio.run(gateway._resume_interrupted_turns())
assert resumed == 0
def test_channel_target_derived_from_client_id_origin():
"""A legacy session whose ``client_id`` encodes ``channel:target`` notifies.
The explicit ``channel_target`` may be absent (session persisted before the
setter was wired, or ingress never called it). The boot scan falls back to
the ``client_id`` origin so the field is a live consumer of existing state.
"""
session = GatewaySession(
_session_id="s6", _agent_id="agent-1", _client_id="telegram:777",
)
session.mark_executing(True)
data = session.to_dict()
assert data.get("channel_target") is None # no explicit target set
gateway = _make_gateway_with_store(_FakeStore({"s6": data}))
sent = []
async def _fake_notice(channel_target, text, session_id, run_epoch):
sent.append((channel_target, session_id))
return True
gateway._deliver_restart_notice = _fake_notice
resumed = asyncio.run(gateway._resume_interrupted_turns())
assert resumed == 1
assert sent == [("telegram:777", "s6")]
def test_boot_scan_covers_more_than_default_fifty_sessions():
"""The boot scan must not silently stop at the store's 50-session default."""
sessions = {}
for i in range(75):
s = GatewaySession(_session_id=f"s{i}", _agent_id="agent-1")
s.set_channel_target(f"telegram:{i}")
s.mark_executing(True)
sessions[f"s{i}"] = s.to_dict()
gateway = _make_gateway_with_store(_FakeStore(sessions))
seen = []
async def _fake_notice(channel_target, text, session_id, run_epoch):
seen.append(session_id)
return True
gateway._deliver_restart_notice = _fake_notice
resumed = asyncio.run(gateway._resume_interrupted_turns())
assert resumed == 75
assert len(seen) == 75
def test_restart_notice_idempotency_key_scopes_session_and_epoch():
"""The notice idempotency key is keyed on (session_id, run_epoch)."""
gateway = WebSocketGateway(port=0)
captured = {}
class _FakeRouter:
async def deliver(self, route, text, idempotency_key=None):
captured["route"] = route
captured["idem"] = idempotency_key
return True
gateway._delivery_router = _FakeRouter()
delivered = asyncio.run(
gateway._deliver_restart_notice("telegram:42", "hi", "sess-7", 3)
)
assert delivered is True
assert captured["route"] == "telegram:42"
assert captured["idem"] == "restart:sess-7:3"
if __name__ == "__main__": # pragma: no cover
import sys
sys.exit(pytest.main([__file__, "-v"]))
@@ -161,82 +161,6 @@ async def test_group_policy_mention_enforcement():
assert command_message is not None, "Commands should always pass in groups"
@pytest.mark.asyncio
async def test_group_policy_observe_records_passive_context():
"""Issue #3380: ``observe`` records unmentioned group messages as context.
Under ``observe`` an unmentioned group message must NOT trigger a run
(returns None) but must be recorded to the session as passive context so
the bot has memory when it is next addressed. A mention still runs.
"""
bot = create_test_bot(group_policy="observe", unknown_user_policy="allow")
bot._bot_user.username = "Test_Bot"
# Unmentioned group message: dropped from dispatch but recorded passively.
no_mention_update = create_mock_telegram_update(
chat_type="group", text="we just decided to ship on Friday"
)
no_mention_message = await process_inbound_telegram_message(no_mention_update, bot)
assert no_mention_message is None, "observe must not trigger a run on no mention"
assert bot._session.record_passive.called, (
"observe must record unmentioned group messages as passive context"
)
args, kwargs = bot._session.record_passive.call_args
assert "we just decided to ship on Friday" in (args[1] if len(args) > 1 else kwargs.get("content", ""))
# Issue #3380: the passive entry must carry the same routing an addressed
# turn uses so that with session_scope="per_chat" it lands on the shared
# group key the next mentioned run reads from (Greptile P1 / CodeRabbit).
assert kwargs.get("chat_id") == "-100123456789", (
"passive recording must thread chat_id so per_chat sessions see it"
)
# A mention still passes through to a real run.
bot._session.record_passive.reset_mock()
mention_update = create_mock_telegram_update(
chat_type="group", text="@test_bot summarise what we just decided"
)
mention_message = await process_inbound_telegram_message(mention_update, bot)
assert mention_message is not None, "observe must still run when the bot is mentioned"
assert not bot._session.record_passive.called, (
"a mentioned message runs; it is not recorded as passive-only context"
)
@pytest.mark.asyncio
async def test_record_passive_per_chat_key_visible_to_addressed_turn():
"""Issue #3380 (Greptile P1 / CodeRabbit): passive context must use the
per_chat group key, not the sender's per_user key.
With ``session_scope="per_chat"`` an addressed turn reads the shared
``(platform, account, chat_id, thread_id)`` key. A passive record threaded
with the same routing must persist under that identical key so the bot sees
the preceding conversation when next mentioned — and must NOT leak into the
sender's separate per_user history.
"""
from praisonai_bot.bots._session import BotSessionManager
mgr = BotSessionManager(platform="telegram", session_scope="per_chat")
route = {"account": "default", "chat_id": "-100999", "thread_id": ""}
ok = mgr.record_passive("sender42", "we ship on Friday", sender="Ann", **route)
assert ok is True
# The addressed turn resolves this shared group key.
group_key = mgr._storage_key("sender42", **route)
history = mgr._load_history("sender42", **route)
assert any(
e.get("passive") and "we ship on Friday" in e.get("content", "")
for e in history
), "passive entry must be visible under the shared per_chat group key"
# It must NOT have leaked into the sender's separate per_user history.
per_user_key = mgr._storage_key("sender42")
assert per_user_key != group_key, "per_chat key must differ from per_user key"
assert not mgr._histories.get(per_user_key), (
"passive entry must not leak into the sender's per_user history"
)
@pytest.mark.asyncio
async def test_dm_messages_bypass_group_policies():
"""Test that DM messages bypass group-specific policies."""
@@ -1,113 +0,0 @@
"""Tests for repo-linked worktree auto-upgrade in SQLiteKanbanStore.create_task.
Refinement 1: a task linked to a git repo auto-upgrades workspace_kind to
'worktree' unless the caller explicitly set a kind or disabled auto_worktree.
"""
import subprocess
import pytest
from praisonai_bot.kanban.sqlite_store import SQLiteKanbanStore
def _git(cwd, *args):
subprocess.run(["git", *args], cwd=cwd, check=True,
capture_output=True, text=True)
@pytest.fixture
def git_repo(tmp_path):
repo = tmp_path / "repo"
repo.mkdir()
_git(repo, "init", "-b", "main")
_git(repo, "config", "user.email", "[email protected]")
_git(repo, "config", "user.name", "Test")
(repo / "f.txt").write_text("x\n")
_git(repo, "add", "-A")
_git(repo, "commit", "-m", "init")
return repo
@pytest.fixture
def store(tmp_path, monkeypatch):
db = tmp_path / "kanban.db"
monkeypatch.setenv("PRAISONAI_KANBAN_DB", str(db))
return SQLiteKanbanStore()
def test_repo_linked_task_auto_upgrades_to_worktree(store, git_repo):
"""A task linked to a git repo (unset kind) becomes 'worktree'."""
task = store.create_task({"title": "t", "repo_path": str(git_repo)})
assert task.workspace_kind == "worktree"
# Branch is derived and persisted up-front for the dispatcher to consume.
assert task.branch == f"kanban/{task.id}"
def test_repo_linked_via_metadata_auto_upgrades(store, git_repo):
"""repo_path carried in metadata also triggers the auto-upgrade."""
task = store.create_task(
{"title": "t", "metadata": {"repo_path": str(git_repo)}}
)
assert task.workspace_kind == "worktree"
def test_explicit_default_kind_respected(store, git_repo):
"""An explicit workspace_kind='default' is never auto-upgraded."""
task = store.create_task(
{"title": "t", "repo_path": str(git_repo), "workspace_kind": "default"}
)
# Explicit default must be respected -> stays default, no branch derived.
assert task.workspace_kind == "default"
assert task.branch is None
def test_auto_worktree_false_disables_upgrade(store, git_repo):
"""Board/task-level auto_worktree=false disables the upgrade entirely."""
task = store.create_task(
{"title": "t", "repo_path": str(git_repo), "auto_worktree": False}
)
assert task.workspace_kind == "default"
def test_non_repo_path_stays_default(store, tmp_path):
"""A path that is not a git repo does not trigger the upgrade."""
plain = tmp_path / "plain"
plain.mkdir()
task = store.create_task({"title": "t", "repo_path": str(plain)})
assert task.workspace_kind == "default"
def test_no_repo_link_stays_default(store):
"""No repo linkage at all keeps today's shared-cwd default."""
task = store.create_task({"title": "t"})
assert task.workspace_kind == "default"
assert task.branch is None
def test_explicit_worktree_kind_still_derives_branch(store):
"""Explicit workspace_kind='worktree' (no repo) still derives a branch."""
task = store.create_task({"title": "t", "workspace_kind": "worktree"})
assert task.workspace_kind == "worktree"
assert task.branch == f"kanban/{task.id}"
def test_repo_path_preserved_in_metadata(store, git_repo):
"""The repo linkage that drove the upgrade is retained in metadata."""
task = store.create_task({"title": "t", "repo_path": str(git_repo)})
# Linkage must not be discarded: it is persisted so a consumer can locate
# the isolating repository between create and dispatch.
assert task.metadata.get("repo_path") == str(git_repo)
# Re-read from the store to confirm it round-trips through persistence.
reloaded = store.get_task(task.id)
assert reloaded.metadata.get("repo_path") == str(git_repo)
def test_explicit_metadata_repo_path_not_overwritten(store, git_repo):
"""An explicit metadata.repo_path wins over the top-level repo_path."""
task = store.create_task({
"title": "t",
"repo_path": str(git_repo),
"metadata": {"repo_path": "/explicit/path"},
})
assert task.metadata.get("repo_path") == "/explicit/path"
@@ -8,9 +8,8 @@ agent commands; wrapper imports fail with a clear install hint.
from __future__ import annotations
import importlib
from pathlib import Path
from types import ModuleType
from typing import Any, Optional, TypeVar
from typing import Any, TypeVar
T = TypeVar("T")
@@ -24,22 +23,6 @@ def wrapper_available() -> bool:
return importlib.util.find_spec("praisonai") is not None
def wrapper_package_path() -> Optional[Path]:
"""Return the on-disk directory of the installed ``praisonai`` wrapper.
Resolves the wrapper package location without importing it, so callers can
reach bundled wrapper assets (e.g. ``ui_*/default_app.py`` presets) that live
beside the wrapper rather than in ``praisonai_code``. Returns ``None`` when the
wrapper is not installed so callers can guide the user to ``pip install``.
"""
import importlib.util
spec = importlib.util.find_spec("praisonai")
if spec is None or not spec.submodule_search_locations:
return None
return Path(spec.submodule_search_locations[0])
def import_wrapper_module(name: str) -> ModuleType:
"""Import ``praisonai.*`` or raise with an install hint."""
if not name.startswith("praisonai"):
@@ -69,152 +69,6 @@ def list(
raise typer.Exit(1)
@app.command()
def create(
name: Optional[str] = typer.Argument(None, help="Agent name (file stem for .praisonai/agents/<name>.md)"),
describe: Optional[str] = typer.Option(None, "--describe", "-d", help="One-line description of what the agent should do"),
model: Optional[str] = typer.Option(None, "--model", "-m", help="Model to use (defaults to your detected provider)"),
permission: Optional[str] = typer.Option(None, "--permission", "-p", help="Permission preset: read-only, review, full"),
global_: bool = typer.Option(False, "--global", help="Write to user-global ~/.praisonai/agents/ instead of the project"),
force: bool = typer.Option(False, "--force", "-f", help="Overwrite an existing agent definition"),
yes: bool = typer.Option(False, "--yes", "-y", help="Non-interactive: accept defaults, skip prompts"),
):
"""Author a new custom agent definition (interactive or scriptable).
Interactive form prompts for the missing pieces; the non-interactive form
(``--describe ... --permission ... --yes``) is scriptable for CI. Drafts a
system prompt via the LLM when available and always writes a valid
``.praisonai/agents/<name>.md`` that ``praisonai run --agent <name>`` uses.
"""
output = get_output_controller()
try:
from praisonai_code.cli.features.agent_scaffold import (
DEFAULT_PERMISSION,
PERMISSION_PRESETS,
draft_system_prompt,
resolve_agents_dir,
validate_agent_name,
write_agent_definition,
)
from praisonai_code.cli.features.custom_definitions import CustomDefinitionsDiscovery
interactive = not yes
try:
from rich.prompt import Prompt
except Exception:
Prompt = None
# Name.
if not name:
if interactive and Prompt is not None:
name = Prompt.ask("Agent name").strip()
if not name:
output.print_error("An agent name is required (pass it as an argument or run interactively).")
raise typer.Exit(1)
# Validate early so path-unsafe names fail fast (before any LLM drafting).
try:
name = validate_agent_name(name)
except ValueError as exc:
output.print_error(str(exc))
raise typer.Exit(1)
# Description.
description = describe
if not description:
if interactive and Prompt is not None:
description = Prompt.ask("Describe what this agent does").strip()
if not description:
description = f"A helpful {name} agent."
# Permission preset.
if not permission:
if interactive and Prompt is not None:
permission = Prompt.ask(
"Permission preset",
choices=list(PERMISSION_PRESETS.keys()),
default=DEFAULT_PERMISSION,
)
else:
permission = DEFAULT_PERMISSION
if permission not in PERMISSION_PRESETS:
output.print_error(
f"Unknown permission preset '{permission}'. "
f"Valid presets: {', '.join(PERMISSION_PRESETS)}"
)
raise typer.Exit(1)
# Model (default via the shared resolver, mirroring init/run/setup).
if not model:
try:
from ..configuration.model_resolver import resolve_default_model
model = resolve_default_model(None, persist=False, notify=False)
except Exception:
model = None
if interactive and Prompt is not None:
model = Prompt.ask("Model", default=model or "").strip() or None
role = name.replace("-", " ").replace("_", " ").title()
goal = description
# Draft the system prompt; degrade to an editable stub on any failure.
body = draft_system_prompt(description, role, model)
if body is None:
output.print_warning("Could not draft a system prompt via the LLM; wrote an editable stub instead.")
agents_dir = resolve_agents_dir(global_)
try:
path = write_agent_definition(
name=name,
description=description,
role=role,
goal=goal,
model=model,
permission=permission,
agents_dir=agents_dir,
body=body,
force=force,
)
except FileExistsError as exc:
output.print_error(
f"Agent already exists: {exc}. Use --force to overwrite."
)
raise typer.Exit(1)
# Post-write validation: re-parse the exact file we just wrote (not a
# precedence lookup) so a global write is never validated against a
# same-named project definition that ``run`` would actually resolve.
discovery = CustomDefinitionsDiscovery()
source = "user" if global_ else "project"
reloaded = discovery._load_agent(path, source=source)
if reloaded is None:
output.print_warning(
f"Wrote {path}, but it could not be re-parsed — check the frontmatter."
)
else:
output.print_success(f"Created {path}")
# If a higher-precedence definition shadows this name, ``run`` would pick
# that one instead — warn so the reported next step is never misleading.
effective = discovery.get_agent(name)
if effective is not None and effective.path != path:
output.print_warning(
f"Another '{name}' agent takes precedence for 'run': {effective.path}. "
"Rename this agent or remove the other to use it directly."
)
output.print_info("You can now run:")
output.print_info(f' praisonai run --agent {name} "..."')
except typer.Exit:
raise
except Exception as e:
output.print_error(str(e))
raise typer.Exit(1)
@app.command()
def show(
name: str = typer.Argument(help="Agent name to inspect"),
@@ -16,13 +16,6 @@ Tools dropped in `.praisonai/tools/*.py` execute local code, so loading them is
opt-in: pass `--allow-local-tools` (or set `PRAISONAI_ALLOW_LOCAL_TOOLS=true`)
on `run` to enable them. Without the opt-in, `run` prints a one-line hint when
tool files are present so the enable step is never a silent no-op.
With ``--generate`` (and a provider credential present), init additionally runs
a short analysis agent that inspects the repository (top-level tree, detected
manifests, README head) and writes a concise, repository-tailored ``AGENTS.md``
at the repo root — immediately discovered by the rules loader on the next run.
Generation is non-destructive (respects ``--force``) and falls back to the
static scaffold when no credential is available or generation fails.
"""
from pathlib import Path
@@ -127,88 +120,6 @@ def _write(path: Path, content: str, force: bool) -> bool:
return True
def _prescan_repo(root: Path) -> str:
"""Build a cheap, token-bounded snapshot of the repository for analysis.
Captures the top-level tree, any detected manifest/CI files, and the head of
the README so the analysis agent has high-signal context without reading the
whole tree. Kept intentionally small to bound tokens and cost.
"""
lines: list[str] = []
try:
entries = sorted(
p.name + ("/" if p.is_dir() else "")
for p in root.iterdir()
if not p.name.startswith(".git")
)
except OSError:
entries = []
if entries:
lines.append("Top-level entries:")
lines.extend(f" {e}" for e in entries[:60])
manifests = (
"pyproject.toml", "setup.py", "setup.cfg", "requirements.txt",
"package.json", "Cargo.toml", "go.mod", "pom.xml", "build.gradle",
"Makefile", "Dockerfile", "docker-compose.yml", "tox.ini",
)
found = [m for m in manifests if (root / m).exists()]
if found:
lines.append("")
lines.append(f"Detected manifests: {', '.join(found)}")
for readme in ("README.md", "README.rst", "README.txt", "README"):
rp = root / readme
if rp.exists():
try:
head = rp.read_text(encoding="utf-8", errors="replace")[:1500]
except OSError:
break
lines.append("")
lines.append(f"{readme} (head):")
lines.append(head)
break
return "\n".join(lines)
def _generate_agents_md(root: Path, model: str) -> str:
"""Run a read-only analysis agent that produces a repo-tailored AGENTS.md.
Reuses the existing Agent runtime with the current provider credential. Feeds
it a cheap pre-scan so it can focus on high-signal, project-specific context.
Raises on any failure so the caller can fall back to the static scaffold.
"""
from praisonaiagents import Agent
prescan = _prescan_repo(root)
prompt = (
"Analyse the repository described below and write a concise AGENTS.md "
"capturing only high-signal, project-specific context an AI coding agent "
"needs on first contact: how to build, test and run; key directories; "
"conventions; and notable constraints or gotchas. Omit generic "
"boilerplate. Output ONLY the markdown body, no code fences.\n\n"
f"Repository snapshot:\n{prescan}"
)
agent = Agent(
instructions=(
"You produce concise, accurate, project-specific AGENTS.md files. "
"Prefer concrete commands and paths over generic advice."
),
llm=model,
)
# Use run() (always silent, non-streaming) rather than start() — start()
# auto-enables streaming in a TTY and returns a generator, whose repr would
# be written instead of the generated markdown.
result = agent.run(prompt)
text = (result or "").strip()
if not text:
raise ValueError("empty generation result")
return text if text.endswith("\n") else text + "\n"
@app.callback(invoke_without_command=True)
def init(
ctx: typer.Context,
@@ -223,13 +134,6 @@ def init(
"-f",
help="Overwrite existing files",
),
generate: bool = typer.Option(
False,
"--generate",
"-g",
help="Analyse the repository and generate a tailored AGENTS.md "
"(requires a provider credential; falls back to the static scaffold)",
),
) -> None:
"""Scaffold the .praisonai/ project convention (config + starter agent + command)."""
# Allow subcommands (none currently) to run without scaffolding.
@@ -285,43 +189,6 @@ def init(
for path in skipped:
output.print_warning(f"Skipped (already exists, use --force): {path}")
# Optional agent-driven generation of a repository-tailored AGENTS.md.
# Non-destructive (respects --force) and degrades gracefully: when no
# provider credential is present or generation fails, the static scaffold
# above remains the guaranteed result. The file lands at the repo root so
# rules_manager discovers it on the next run. This is always the current
# repository (git root / cwd), never ~/AGENTS.md — even with --global,
# which only changes where the static scaffold is written.
if generate:
repo_root = get_git_root() or Path.cwd()
agents_path = repo_root / "AGENTS.md"
if agents_path.exists() and not force:
output.print_warning(
f"Skipped generation (already exists, use --force): {agents_path}"
)
elif not provider_detected:
output.print_warning(
"Skipped --generate: no provider credential detected. "
"Falling back to the static scaffold above."
)
else:
try:
content = _generate_agents_md(repo_root, scaffold_model)
except Exception as exc: # noqa: BLE001 - fall back, never fail init
output.print_warning(
f"Generation failed ({exc}); kept the static scaffold above."
)
else:
try:
agents_path.write_text(content, encoding="utf-8")
except OSError as exc:
output.print_error(
f"Failed to write {agents_path}: {exc.strerror or exc}",
remediation="Check write permissions and free disk space.",
)
raise typer.Exit(code=1)
output.print_success(f"Generated {agents_path}")
if written:
output.print_success(f"Initialised {base}")
if provider_detected:
@@ -21,48 +21,6 @@ _FRAMEWORK_HELP = "Framework: praisonai, crewai, autogen"
_ALLOW_LOCAL_TOOLS_ENV = "PRAISONAI_ALLOW_LOCAL_TOOLS"
def _run_succeeded(result: Any) -> bool:
"""Whether an agent run result represents a genuine success.
The public ``Agent.start``/``run`` surface collapses failures (swallowed
LLM/auth error, guardrail block, tool failure, ``max_iter`` without
completion) to a falsy result (``None``/empty), while a real answer is a
non-empty string. Treat a falsy result as failure so ``praisonai run`` can
exit non-zero and emit a machine-readable outcome instead of reporting
success unconditionally.
"""
if result is None:
return False
if isinstance(result, str):
return bool(result.strip())
return True
def _report_run_failure(output: Any) -> None:
"""Report an agent-run failure and exit non-zero.
Mirrors the existing arg-error exit convention (``typer.Exit(1)``) but for a
*run* failure: emits a machine-readable outcome under ``--output json`` and
prints a human-facing error, so CI/scripts can branch on the exit code and
the JSON ``status`` instead of scraping stderr.
"""
message = "Run failed: the agent did not produce a result."
output.emit_result(
message=message,
data={"status": "failed", "result": None},
)
output.emit_error(message=message, data={"status": "failed"})
output.print_error(
message,
code="run_failed",
remediation=(
"Re-run with --verbose to see the underlying error, "
"or check credentials with: praisonai setup"
),
)
raise typer.Exit(1)
def _is_yaml_file(target: Optional[str]) -> bool:
"""Return True when ``target`` is an existing YAML file path.
@@ -697,11 +655,6 @@ def _try_attach_runtime(
# Runtime went away mid-flight; fall back to in-process execution.
return False
# The warm runtime handled the request; apply the same failure contract as
# the in-process paths so an empty result exits non-zero instead of silently
# reporting success.
if not _run_succeeded(result):
_report_run_failure(output)
output.emit_result(
message="Prompt completed",
data={"result": str(result) if result else None},
@@ -1399,8 +1352,6 @@ def _run_from_file(
result = praison.run()
_record_session_usage(session_id or auto_save_name, model, output)
if not _run_succeeded(result):
_report_run_failure(output)
output.emit_result(
message="Run completed",
data={"result": str(result) if result else None}
@@ -1410,11 +1361,6 @@ def _run_from_file(
if not output.is_json_mode:
output.print_success("Run completed")
except typer.Exit:
# A deliberate non-zero exit (e.g. a classified run failure) must
# propagate unchanged; the broad handler below is only for unexpected
# errors and would otherwise re-report the same failure.
raise
except Exception as e:
output.emit_error(message=str(e))
output.print_error(str(e))
@@ -1616,12 +1562,9 @@ def _run_prompt(
finally:
detach_bridge(agent, bridge)
succeeded = _run_succeeded(result)
if bridge is not None:
bridge.emit_run_result(result, ok=succeeded)
bridge.emit_run_result(result, ok=True)
_record_session_usage(session_id or auto_save_name, model, output)
if not succeeded:
_report_run_failure(output)
output.emit_result(
message="Prompt completed",
data={"result": str(result) if result else None}
@@ -1689,8 +1632,6 @@ def _run_prompt(
result = praison.handle_direct_prompt(prompt)
_record_session_usage(session_id or auto_save_name, model, output)
if not _run_succeeded(result):
_report_run_failure(output)
output.emit_result(
message="Prompt completed",
data={"result": str(result) if result else None}
@@ -1849,12 +1790,6 @@ def _run_from_file_profiled(
# Print profiling report
profiler.print_report()
# Honour the same failure contract as the non-profiled YAML path: an empty
# agent result is a run failure and must exit non-zero (reported after the
# profiling output so the profile is still shown).
if not _run_succeeded(result):
_report_run_failure(get_output_controller())
def _wire_subagent_delegation(
agent_config: Dict[str, Any],
@@ -2059,12 +1994,9 @@ def _run_custom_agent(
finally:
detach_bridge(agent, bridge)
succeeded = _run_succeeded(result)
if bridge is not None:
bridge.emit_run_result(result, ok=succeeded)
bridge.emit_run_result(result, ok=True)
_record_session_usage(session_id or auto_save_name, model, output)
if not succeeded:
_report_run_failure(output)
output.emit_result(
message="Agent completed",
data={"result": str(result) if result else None}
@@ -2184,9 +2116,3 @@ def _run_prompt_profiled(
# Print profiling report
profiler.print_report()
# Honour the same failure contract as the non-profiled paths: an empty
# agent result is a run failure and must exit non-zero (the report is
# emitted after the profiling output so the profile is still shown).
if not _run_succeeded(response):
_report_run_failure(get_output_controller())
@@ -21,25 +21,6 @@ app = typer.Typer(
)
def _resolve_bundled_default_app(default_app_name: str) -> Optional[Path]:
"""Locate a bundled ``ui_*/default_app.py`` inside the praisonai wrapper.
The five bundled UI presets ship in the ``praisonai`` wrapper package
(``praisonai/<name>/default_app.py``), not in ``praisonai_code`` where this
loader now lives after the CLI extraction. Cross-tier wrapper access is
routed through the lazy ``_wrapper_bridge`` per ARCHITECTURE.md §2 rather
than traversing the wrapper package directly. Returns ``None`` when the
wrapper is not installed so the caller can guide the user to
``pip install "praisonai[ui]"``.
"""
from praisonai_code._wrapper_bridge import wrapper_package_path
wrapper_path = wrapper_package_path()
if wrapper_path is None:
return None
return wrapper_path / default_app_name / "default_app.py"
def _launch_aiui_app(
app_dir: str,
default_app_name: str,
@@ -73,18 +54,11 @@ def _launch_aiui_app(
# Ensure default app exists
if not default_app.exists():
ui_dir.mkdir(parents=True, exist_ok=True)
# Bundled ui_*/default_app.py assets ship in the praisonai wrapper,
# not in praisonai_code — resolve against the installed wrapper.
bundled = _resolve_bundled_default_app(default_app_name)
if bundled is None or not bundled.exists():
print(
"\033[91mERROR: Bundled default_app.py not found. "
'Install with:\n pip install "praisonai[ui]"\033[0m'
)
bundled = Path(__file__).parent.parent.parent / default_app_name / "default_app.py"
if not bundled.exists():
print(f"\033[91mERROR: Bundled default_app.py not found at {bundled}\033[0m")
sys.exit(1)
default_app.write_text(
bundled.read_text(encoding="utf-8"), encoding="utf-8"
)
default_app.write_text(bundled.read_text())
print(f" ✓ Created default {ui_name} config: {default_app}")
resolved = default_app
@@ -555,17 +555,13 @@ class ConfigResolver:
# Collect paths from current directory upward
while current != current.parent:
# Never treat the user's home directory as a project directory.
if home is not None and current == home:
break
search_paths.append(current)
if git_root and current == git_root:
break
break # Stop at git root if found
if home is not None and current == home:
break # Do not walk past home into shared parents
current = current.parent
# Search for config files
for search_dir in search_paths:
for config_name in self.PROJECT_CONFIG_NAMES:
@@ -576,6 +572,7 @@ class ConfigResolver:
data["_source"] = str(config_path)
self._validate(data, str(config_path))
return data
return None
def _managed_source_spec(self) -> Dict[str, Any]:
@@ -1,185 +0,0 @@
"""
Interactive, LLM-assisted authoring for custom agent definitions.
Backs ``praisonai agent create`` (see ``commands/agent.py``): turns a one-line
description into a valid, permission-scoped ``.praisonai/agents/<name>.md`` that
``CustomDefinitionsDiscovery._load_agent`` re-parses losslessly.
Design notes (kept lightweight, no new runtime surface):
- Permission presets are the existing ``mode`` shorthands already understood by
``resolve_permission_config`` (``MODE_RULES``): ``full``/``read-only``/
``review``. The chosen preset is written to the ``mode`` frontmatter key, so
the authored agent inherits the identical permission grammar the runtime uses
— no parallel permission vocabulary is introduced here.
- The system-prompt draft reuses the plain ``Agent`` generation path already
proven by ``init --generate`` (``commands/init.py``). Generation is guarded:
any failure degrades to an editable stub so file creation never blocks.
"""
from pathlib import Path
from typing import Optional
import yaml
from .custom_definitions import MODE_RULES
# User-facing permission presets → the ``mode`` frontmatter value they map to.
# Every value is a key of MODE_RULES so it resolves through the runtime's
# existing permission grammar unchanged. Ordered for stable interactive display.
PERMISSION_PRESETS = {
"read-only": "read-only",
"review": "review",
"full": "full",
}
DEFAULT_PERMISSION = "full"
def validate_agent_name(name: str) -> str:
"""Return a sanitised agent name or raise ``ValueError``.
The name becomes a file stem under ``.praisonai/agents/``, so it must be a
single path-free token. Rejecting separators and traversal components keeps
every write contained within the chosen agents directory (no ``../`` escape,
no absolute paths).
"""
candidate = (name or "").strip()
if not candidate:
raise ValueError("Agent name must not be empty.")
if candidate in {".", ".."} or any(sep in candidate for sep in ("/", "\\")):
raise ValueError(
f"Invalid agent name {name!r}: use a simple name without path "
"separators (e.g. 'code-reviewer')."
)
if Path(candidate).name != candidate:
raise ValueError(
f"Invalid agent name {name!r}: use a simple name without path "
"separators (e.g. 'code-reviewer')."
)
return candidate
def resolve_agents_dir(global_: bool) -> Path:
"""Return the target ``.praisonai/agents/`` directory (project or global)."""
from ..utils.project import get_git_root
if global_:
base = Path.home() / ".praisonai"
else:
base = (get_git_root() or Path.cwd()) / ".praisonai"
return base / "agents"
def draft_system_prompt(description: str, role: str, model: Optional[str]) -> Optional[str]:
"""Draft a system prompt from a one-line description via the Agent runtime.
Reuses the same plain ``Agent.run`` path as ``init --generate``. Returns the
drafted prompt text, or ``None`` on any failure so the caller can fall back
to an editable stub (generation must never block file creation).
"""
try:
from praisonaiagents import Agent
prompt = (
"Write a concise, high-quality system prompt for a custom AI agent.\n"
f"The agent's role: {role}\n"
f"The agent's purpose (from the author): {description}\n\n"
"Output ONLY the system prompt body as plain markdown — no code "
"fences, no preamble, no frontmatter. Address the agent in the "
"second person, state its responsibilities, and note when it should "
"ask for clarification instead of guessing."
)
agent = Agent(
instructions=(
"You write clear, focused system prompts for AI agents. "
"Prefer concrete guidance over generic filler."
),
llm=model,
)
result = agent.run(prompt)
text = (result or "").strip()
return text or None
except Exception:
return None
def _stub_system_prompt(description: str) -> str:
"""Editable fallback body used when LLM drafting is unavailable."""
return (
f"You are an agent whose purpose is: {description}\n\n"
"Describe how you should behave here. Be specific about your "
"responsibilities, and ask for clarification instead of guessing."
)
def render_agent_markdown(
*,
description: str,
role: str,
goal: str,
model: Optional[str],
permission: str,
body: str,
) -> str:
"""Render a ``.praisonai/agents/<name>.md`` document (frontmatter + body).
The frontmatter is serialised with ``yaml.safe_dump`` so it round-trips
through ``CustomDefinitionsDiscovery._parse_markdown_frontmatter`` cleanly.
The permission preset is emitted as ``mode`` (a MODE_RULES key), which the
runtime resolves via ``resolve_permission_config`` — no parallel grammar.
"""
frontmatter: dict = {}
if model:
frontmatter["model"] = model
frontmatter["role"] = role
frontmatter["goal"] = goal
mode_value = PERMISSION_PRESETS.get(permission, permission)
if mode_value not in MODE_RULES:
raise ValueError(
f"Unknown permission preset: {permission!r}. "
f"Valid presets: {sorted(PERMISSION_PRESETS)}"
)
# ``full`` carries no restrictions; omit it to keep the file minimal while
# still round-tripping (absence of ``mode`` == full toolset).
if mode_value != "full":
frontmatter["mode"] = mode_value
fm_text = yaml.safe_dump(frontmatter, sort_keys=False, default_flow_style=False)
body_text = body.strip()
return f"---\n{fm_text}---\n{body_text}\n"
def write_agent_definition(
*,
name: str,
description: str,
role: str,
goal: str,
model: Optional[str],
permission: str,
agents_dir: Path,
body: Optional[str] = None,
force: bool = False,
) -> Path:
"""Write ``<agents_dir>/<name>.md`` and return its path.
Raises ``FileExistsError`` when the target exists and ``force`` is False, so
the caller can surface a clear message without overwriting user work.
"""
safe_name = validate_agent_name(name)
target = agents_dir / f"{safe_name}.md"
if target.exists() and not force:
raise FileExistsError(str(target))
content = render_agent_markdown(
description=description,
role=role,
goal=goal,
model=model,
permission=permission,
body=body if body is not None else _stub_system_prompt(description),
)
agents_dir.mkdir(parents=True, exist_ok=True)
target.write_text(content, encoding="utf-8")
return target
@@ -21,52 +21,24 @@ from pathlib import Path
logger = logging.getLogger(__name__)
_RUN_SYNC_TIMEOUT = float(os.environ.get("PRAISONAI_RUN_SYNC_TIMEOUT", "300"))
def _run_sync(coro, *, timeout: Optional[float] = _RUN_SYNC_TIMEOUT):
def _run_sync(coro):
"""Run a coroutine to completion from sync code without a wrapper dependency.
``praisonai-code`` is a Tier-2 package and must not import the ``praisonai``
wrapper at the hot path (C7 gate / ARCHITECTURE §2). This bridge keeps the
union of both historical guarantees:
- **Running-loop safety:** when a loop is already running (e.g. called from
within async agent execution) we offload to a worker thread so we never
try to nest ``asyncio.run`` on a live loop.
- **Timeout:** a stuck coroutine is bounded by ``timeout`` seconds (honoured
the same way whether or not a loop is already running) so a hung tool call
cannot block the agent indefinitely.
wrapper at the hot path (C7 gate / ARCHITECTURE §2). When no event loop is
running we use ``asyncio.run``; when one is already running (e.g. called from
within async agent execution) we offload to a worker thread so we never try
to nest ``asyncio.run`` on a live loop.
"""
import concurrent.futures
try:
asyncio.get_running_loop()
except RuntimeError:
running_loop = False
else:
running_loop = True
if not running_loop and timeout is None:
return asyncio.run(coro)
# Offload to a worker thread so the timeout can be enforced via
# ``Future.result`` even when a loop is already running (never nest
# ``asyncio.run`` on a live loop). We deliberately avoid the executor's
# context manager: on ``__exit__`` it calls ``shutdown(wait=True)``, which
# would re-block on a coroutine that is still hung — defeating the timeout.
pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
future = pool.submit(asyncio.run, coro)
try:
return future.result(timeout=timeout)
except concurrent.futures.TimeoutError:
future.cancel()
raise
finally:
# Return promptly: do not wait for a worker that may still be running
# the (now abandoned) coroutine. The daemon-like thread is reclaimed on
# interpreter exit; on success it has already finished by this point.
pool.shutdown(wait=False)
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(asyncio.run, coro).result()
def _sanitize_filepath(filepath: str, workspace: Optional[str] = None) -> str:
@@ -410,193 +410,4 @@ def check_gateway_env_substitution(config: DoctorConfig) -> CheckResult:
message=f"Error checking env vars: {str(e)[:100]}",
remediation="Check config file access",
duration_ms=(time.time() - start) * 1000,
)
def _find_gateway_config_path(config: DoctorConfig) -> Optional[str]:
"""Resolve gateway/bot config path, honouring ``--file`` first."""
config_paths: List[str] = []
if getattr(config, "config_file", None):
config_paths.append(config.config_file)
from praisonai_code.cli._paths import resolve_bot_config_path
config_paths.extend([
resolve_bot_config_path("gateway.yaml"),
resolve_bot_config_path("bot.yaml"),
"gateway.yaml",
"bot.yaml",
])
for path in config_paths:
if path and os.path.exists(path):
return path
return None
@register_check(
id="gateway_shell_readiness",
title="Gateway Shell Readiness",
description="Validate allow_shell wiring offline (no LLM)",
category=CheckCategory.BOTS,
severity=CheckSeverity.HIGH,
)
def check_gateway_shell_readiness(config: DoctorConfig) -> CheckResult:
"""Offline shell wiring check for channels with ``allow_shell: true``."""
start = time.time()
skipped = skip_if_no_wrapper(
"gateway_shell_readiness", "Gateway Shell Readiness", start=start
)
if skipped:
return skipped
skipped = skip_if_no_bot_package(
"gateway_shell_readiness", "Gateway Shell Readiness", start=start
)
if skipped:
return skipped
config_path = _find_gateway_config_path(config)
if not config_path:
return CheckResult(
id="gateway_shell_readiness",
title="Gateway Shell Readiness",
category=CheckCategory.BOTS,
status=CheckStatus.WARN,
message="No gateway/bot config found",
remediation="Run 'praisonai onboard' or pass --file /path/to/bot.yaml",
duration_ms=(time.time() - start) * 1000,
)
try:
from praisonai_code._bot_bridge import import_bot_module
preflight = import_bot_module("praisonai_bot.gateway.preflight")
result = preflight.run_shell_readiness_check(config_path)
except Exception as exc:
return CheckResult(
id="gateway_shell_readiness",
title="Gateway Shell Readiness",
category=CheckCategory.BOTS,
status=CheckStatus.ERROR,
message=f"Shell readiness check failed: {str(exc)[:100]}",
duration_ms=(time.time() - start) * 1000,
)
if result.ok:
return CheckResult(
id="gateway_shell_readiness",
title="Gateway Shell Readiness",
category=CheckCategory.BOTS,
status=CheckStatus.PASS,
message=result.message,
duration_ms=(time.time() - start) * 1000,
)
return CheckResult(
id="gateway_shell_readiness",
title="Gateway Shell Readiness",
category=CheckCategory.BOTS,
status=CheckStatus.FAIL,
message=result.message,
details="\n".join(result.issues) if result.issues else None,
remediation="Fix allow_shell / auto_approve_shell settings in bot.yaml",
duration_ms=(time.time() - start) * 1000,
)
@register_check(
id="gateway_channel_probe",
title="Gateway Channel Probe",
description="Live credential probe for configured channels",
category=CheckCategory.BOTS,
severity=CheckSeverity.HIGH,
requires_deep=True,
)
def check_gateway_channel_probe(config: DoctorConfig) -> CheckResult:
"""Live platform credential probe (``auth.test``, ``getMe``, etc.)."""
import asyncio
start = time.time()
skipped = skip_if_no_wrapper(
"gateway_channel_probe", "Gateway Channel Probe", start=start
)
if skipped:
return skipped
skipped = skip_if_no_bot_package(
"gateway_channel_probe", "Gateway Channel Probe", start=start
)
if skipped:
return skipped
config_path = _find_gateway_config_path(config)
if not config_path:
return CheckResult(
id="gateway_channel_probe",
title="Gateway Channel Probe",
category=CheckCategory.BOTS,
status=CheckStatus.SKIP,
message="No gateway/bot config found",
duration_ms=(time.time() - start) * 1000,
)
try:
from praisonai_code._bot_bridge import import_bot_module
preflight = import_bot_module("praisonai_bot.gateway.preflight")
channels = preflight.load_channels_mapping(config_path)
if not channels:
return CheckResult(
id="gateway_channel_probe",
title="Gateway Channel Probe",
category=CheckCategory.BOTS,
status=CheckStatus.SKIP,
message="No channels configured",
duration_ms=(time.time() - start) * 1000,
)
results = asyncio.run(preflight.probe_channels(channels))
except Exception as exc:
return CheckResult(
id="gateway_channel_probe",
title="Gateway Channel Probe",
category=CheckCategory.BOTS,
status=CheckStatus.ERROR,
message=f"Probe failed: {str(exc)[:100]}",
duration_ms=(time.time() - start) * 1000,
)
lines = []
failed = []
for name, probe in results.items():
if getattr(probe, "ok", False):
identity = getattr(probe, "bot_username", None) or ""
detail = f"@{identity}" if identity else getattr(probe, "platform", name)
lines.append(f"{name}: OK ({detail})")
else:
err = getattr(probe, "error", None) or "unknown error"
lines.append(f"{name}: FAIL ({err})")
failed.append(name)
if failed:
return CheckResult(
id="gateway_channel_probe",
title="Gateway Channel Probe",
category=CheckCategory.BOTS,
status=CheckStatus.FAIL,
message=f"{len(failed)} channel probe(s) failed",
details="\n".join(lines),
remediation=(
"Fix channel tokens or run: "
f"praisonai gateway test --config {config_path}"
),
duration_ms=(time.time() - start) * 1000,
)
return CheckResult(
id="gateway_channel_probe",
title="Gateway Channel Probe",
category=CheckCategory.BOTS,
status=CheckStatus.PASS,
message=f"All {len(results)} channel probe(s) passed",
details="\n".join(lines),
duration_ms=(time.time() - start) * 1000,
)
)
@@ -124,12 +124,8 @@ class ToolConfig:
"""Create config from environment variables."""
config = cls() # __post_init__ already applied PRAISON_TOOLS_DISABLE
# Check for workspace. The `praisonai code --workspace` flag writes
# PRAISONAI_WORKSPACE (code.py), so honour that canonical name first and
# fall back to the older PRAISON_WORKSPACE for backward compatibility.
workspace = os.environ.get("PRAISONAI_WORKSPACE") or os.environ.get(
"PRAISON_WORKSPACE", ""
)
# Check for workspace
workspace = os.environ.get("PRAISON_WORKSPACE", "")
if workspace:
config.workspace = workspace
@@ -72,8 +72,7 @@ class InteractiveConfig:
"""Create config from environment variables."""
return cls(
model=os.environ.get("PRAISON_MODEL"),
workspace=os.environ.get("PRAISONAI_WORKSPACE")
or os.environ.get("PRAISON_WORKSPACE", os.getcwd()),
workspace=os.environ.get("PRAISON_WORKSPACE", os.getcwd()),
approval_mode=os.environ.get("PRAISON_APPROVAL_MODE", "prompt"),
enable_acp=os.environ.get("PRAISON_DISABLE_ACP", "").lower() != "true",
enable_lsp=os.environ.get("PRAISON_DISABLE_LSP", "").lower() != "true",
@@ -1,62 +0,0 @@
"""Doctor checks for gateway shell readiness and channel probes."""
import asyncio
from unittest.mock import MagicMock, patch
import pytest
from praisonai_code.cli.features.doctor.models import DoctorConfig
from praisonai_code.cli.features.doctor.checks import gateway_checks
def test_shell_readiness_pass(tmp_path, monkeypatch):
cfg = tmp_path / "bot.yaml"
cfg.write_text("channels:\n slack:\n allow_shell: true\n")
mock_result = MagicMock(ok=True, message="Shell wiring OK", issues=[])
def fake_import(name):
mod = MagicMock()
mod.run_shell_readiness_check.return_value = mock_result
return mod
monkeypatch.setattr(gateway_checks, "skip_if_no_wrapper", lambda *a, **k: None)
monkeypatch.setattr(gateway_checks, "skip_if_no_bot_package", lambda *a, **k: None)
monkeypatch.setattr(
"praisonai_code._bot_bridge.import_bot_module",
fake_import,
)
result = gateway_checks.check_gateway_shell_readiness(
DoctorConfig(config_file=str(cfg))
)
assert result.status.value == "pass"
def test_channel_probe_deep_mock(tmp_path, monkeypatch):
cfg = tmp_path / "bot.yaml"
cfg.write_text(
"channels:\n telegram:\n platform: telegram\n token: t\n"
)
from praisonaiagents.bots import ProbeResult
async def fake_probe(channels, timeout=15.0):
return {"telegram": ProbeResult(ok=True, platform="telegram", bot_username="bot")}
mod = MagicMock()
mod.load_channels_mapping.return_value = {"telegram": {"platform": "telegram", "token": "t"}}
mod.probe_channels = fake_probe
monkeypatch.setattr(gateway_checks, "skip_if_no_wrapper", lambda *a, **k: None)
monkeypatch.setattr(gateway_checks, "skip_if_no_bot_package", lambda *a, **k: None)
monkeypatch.setattr(
"praisonai_code._bot_bridge.import_bot_module",
lambda name: mod,
)
result = gateway_checks.check_gateway_channel_probe(
DoctorConfig(config_file=str(cfg))
)
assert result.status.value == "pass"
assert "telegram" in (result.details or "")
@@ -1,193 +0,0 @@
"""Tests for ``praisonai agent create`` (issue #3376).
Covers the golden-file round-trip (write → discover → assert frontmatter/prompt)
and the non-interactive CLI form, with LLM drafting stubbed so tests never call
a provider.
"""
import os
from pathlib import Path
import pytest
from typer.testing import CliRunner
from praisonai_code.cli.commands.agent import app
from praisonai_code.cli.features import agent_scaffold
from praisonai_code.cli.features.agent_scaffold import (
render_agent_markdown,
validate_agent_name,
write_agent_definition,
)
from praisonai_code.cli.features.custom_definitions import (
CustomDefinitionsDiscovery,
resolve_permission_config,
)
def test_render_roundtrips_through_discovery(tmp_path):
agents_dir = tmp_path / ".praisonai" / "agents"
path = write_agent_definition(
name="reviewer",
description="Review code and suggest improvements",
role="Reviewer",
goal="Review code and suggest improvements",
model="gpt-4o-mini",
permission="read-only",
agents_dir=agents_dir,
body="You are a careful code reviewer.",
)
discovery = CustomDefinitionsDiscovery()
agent = discovery._load_agent(path, source="project")
assert agent is not None
assert agent.name == "reviewer"
assert agent.model == "gpt-4o-mini"
assert agent.role == "Reviewer"
assert agent.mode == "read-only"
assert "careful code reviewer" in (agent.system_prompt or "")
# The mode preset must resolve to a real, restrictive permission config.
resolved = resolve_permission_config(agent.permission, agent.mode)
assert resolved is not None
assert resolved.get("edit:*") == "deny"
def test_full_preset_omits_mode(tmp_path):
content = render_agent_markdown(
description="do things",
role="Doer",
goal="do things",
model=None,
permission="full",
body="You do things.",
)
assert "mode:" not in content
# Must still start with valid frontmatter.
assert content.startswith("---\n")
def test_unknown_permission_rejected():
with pytest.raises(ValueError):
render_agent_markdown(
description="x",
role="X",
goal="x",
model=None,
permission="bogus",
body="body",
)
def test_cli_create_non_interactive(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
# Stub LLM drafting so the test is deterministic and offline.
monkeypatch.setattr(
agent_scaffold,
"draft_system_prompt",
lambda description, role, model: "You are a helpful test agent.",
)
result = CliRunner().invoke(
app,
[
"create",
"helper",
"--describe",
"Assist with tasks",
"--model",
"gpt-4o-mini",
"--permission",
"read-only",
"--yes",
],
)
assert result.exit_code == 0, result.output
written = tmp_path / ".praisonai" / "agents" / "helper.md"
assert written.exists()
text = written.read_text()
assert "mode: read-only" in text
assert "helpful test agent" in text
@pytest.mark.parametrize(
"bad_name",
["../evil", "../../etc/passwd", "a/b", "a\\b", "/abs", ".", "..", " "],
)
def test_write_rejects_path_unsafe_names(tmp_path, bad_name):
agents_dir = tmp_path / ".praisonai" / "agents"
with pytest.raises(ValueError):
write_agent_definition(
name=bad_name,
description="d",
role="R",
goal="g",
model=None,
permission="full",
agents_dir=agents_dir,
body="body",
)
def test_validate_agent_name_accepts_simple_names():
assert validate_agent_name("code-reviewer") == "code-reviewer"
assert validate_agent_name(" helper ") == "helper"
def test_cli_create_rejects_traversal_name(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
agent_scaffold,
"draft_system_prompt",
lambda description, role, model: "stub",
)
result = CliRunner().invoke(
app,
["create", "../escape", "--describe", "d", "--permission", "full", "--yes"],
)
assert result.exit_code == 1
assert not (tmp_path.parent / "escape.md").exists()
def test_cli_create_global_warns_on_project_shadow(tmp_path, monkeypatch):
"""A --global write shadowed by a same-named project agent must warn."""
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path / "home"))
monkeypatch.setattr(
agent_scaffold,
"draft_system_prompt",
lambda description, role, model: "stub",
)
# Pre-existing project agent with the same name.
project_dir = tmp_path / ".praisonai" / "agents"
project_dir.mkdir(parents=True)
(project_dir / "shared.md").write_text("---\nrole: Proj\n---\nproject body\n")
result = CliRunner().invoke(
app,
["create", "shared", "--describe", "d", "--permission", "full",
"--global", "--yes"],
)
assert result.exit_code == 0, result.output
# The global file must actually be written.
assert (tmp_path / "home" / ".praisonai" / "agents" / "shared.md").exists()
# And the user must be warned that the project definition takes precedence.
assert "precedence" in result.output.lower()
def test_cli_create_refuses_overwrite(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
monkeypatch.setattr(
agent_scaffold,
"draft_system_prompt",
lambda description, role, model: "stub",
)
args = ["create", "dup", "--describe", "d", "--permission", "full", "--yes"]
first = CliRunner().invoke(app, args)
assert first.exit_code == 0, first.output
second = CliRunner().invoke(app, args)
assert second.exit_code == 1
assert "already exists" in second.output.lower()
@@ -85,17 +85,10 @@ def test_resolve_with_provenance_env_overrides_project(tmp_path, monkeypatch):
assert prov["agent.model"]["layer"] == "environment"
def test_resolve_with_provenance_defaults_present(tmp_path, monkeypatch):
monkeypatch.setattr(
ConfigResolver,
"_load_global_config",
lambda self: None,
)
def test_resolve_with_provenance_defaults_present(tmp_path):
resolver = ConfigResolver(cwd=tmp_path)
prov = resolver.resolve_with_provenance()
# With no config files, values come from built-in defaults.
assert prov["telemetry"]["layer"] == "defaults"
@@ -1,146 +0,0 @@
"""Tests for `praisonai init --generate` repository-tailored AGENTS.md.
Covers the agent-driven onboarding path added on top of the static scaffold:
- With a provider credential, generation writes an AGENTS.md at the repo root
containing the agent's analysis (mentioning detected build/test context).
- With no credential, `--generate` degrades to today's static scaffold and
writes no AGENTS.md.
- An existing AGENTS.md is never overwritten without `--force`.
- Generation failures never break init (static scaffold remains).
"""
from pathlib import Path
from unittest.mock import patch
from praisonai_code.cli.commands import init as init_mod
def _run_init(tmp_path: Path, **kwargs) -> None:
"""Invoke the init callback with a stub context, scoped to tmp_path."""
class _Ctx:
invoked_subcommand = None
with patch.object(init_mod, "get_git_root", return_value=tmp_path):
init_mod.init(_Ctx(), global_=kwargs.get("global_", False),
force=kwargs.get("force", False),
generate=kwargs.get("generate", False))
class TestPrescan:
def test_captures_manifests_and_readme(self, tmp_path):
(tmp_path / "pyproject.toml").write_text("[project]\n")
(tmp_path / "README.md").write_text("# MyProj\nRun tests with pytest.")
(tmp_path / "src").mkdir()
snapshot = init_mod._prescan_repo(tmp_path)
assert "pyproject.toml" in snapshot
assert "src/" in snapshot
assert "Run tests with pytest." in snapshot
class TestGenerateWiring:
def test_generates_agents_md_with_credential(self, tmp_path):
(tmp_path / "pyproject.toml").write_text("[project]\n")
generated = "# Agents\nBuild: pip install -e .\nTest: pytest\n"
with patch.object(init_mod, "_any_provider_credential", return_value=True), \
patch.object(init_mod, "_generate_agents_md", return_value=generated) as gen:
_run_init(tmp_path, generate=True)
agents = tmp_path / "AGENTS.md"
assert agents.exists()
assert "pytest" in agents.read_text()
gen.assert_called_once()
def test_no_credential_falls_back_to_static_scaffold(self, tmp_path):
with patch.object(init_mod, "_any_provider_credential", return_value=False), \
patch.object(init_mod, "_generate_agents_md") as gen:
_run_init(tmp_path, generate=True)
# Static scaffold still produced, but no AGENTS.md and no agent call.
assert (tmp_path / ".praisonai" / "config.yaml").exists()
assert not (tmp_path / "AGENTS.md").exists()
gen.assert_not_called()
def test_existing_agents_md_not_overwritten_without_force(self, tmp_path):
agents = tmp_path / "AGENTS.md"
agents.write_text("# Existing rules\n")
with patch.object(init_mod, "_any_provider_credential", return_value=True), \
patch.object(init_mod, "_generate_agents_md") as gen:
_run_init(tmp_path, generate=True)
assert agents.read_text() == "# Existing rules\n"
gen.assert_not_called()
def test_existing_agents_md_overwritten_with_force(self, tmp_path):
agents = tmp_path / "AGENTS.md"
agents.write_text("# Existing rules\n")
with patch.object(init_mod, "_any_provider_credential", return_value=True), \
patch.object(init_mod, "_generate_agents_md", return_value="# New\n"):
_run_init(tmp_path, generate=True, force=True)
assert agents.read_text() == "# New\n"
def test_generation_failure_does_not_break_init(self, tmp_path):
with patch.object(init_mod, "_any_provider_credential", return_value=True), \
patch.object(init_mod, "_generate_agents_md", side_effect=RuntimeError("boom")):
_run_init(tmp_path, generate=True)
# Init still succeeds with the static scaffold; no AGENTS.md written.
assert (tmp_path / ".praisonai" / "config.yaml").exists()
assert not (tmp_path / "AGENTS.md").exists()
def test_no_generate_flag_never_writes_agents_md(self, tmp_path):
with patch.object(init_mod, "_any_provider_credential", return_value=True), \
patch.object(init_mod, "_generate_agents_md") as gen:
_run_init(tmp_path, generate=False)
assert not (tmp_path / "AGENTS.md").exists()
gen.assert_not_called()
def test_global_writes_agents_md_to_repo_root_not_home(self, tmp_path, monkeypatch):
# --global changes only the static scaffold location; the generated
# AGENTS.md must still land at the repo root, never ~/AGENTS.md.
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("HOME", str(home))
captured = {}
def _fake_generate(root, model):
captured["root"] = root
return "# Repo agents\n"
with patch.object(init_mod, "_any_provider_credential", return_value=True), \
patch.object(init_mod, "_generate_agents_md", side_effect=_fake_generate):
_run_init(tmp_path, generate=True, global_=True)
assert (tmp_path / "AGENTS.md").exists()
assert not (home / "AGENTS.md").exists()
assert captured["root"] == tmp_path
class TestGenerateHelper:
def test_uses_run_and_returns_string_not_generator(self, tmp_path):
# _generate_agents_md must call agent.run() (silent, returns str), not
# start() which can return a streaming generator whose repr would be
# written to AGENTS.md.
class _FakeAgent:
def __init__(self, *args, **kwargs):
pass
def run(self, prompt):
return "# Generated\nBuild: make\n"
def start(self, prompt): # pragma: no cover - must not be used
raise AssertionError("start() must not be used")
with patch("praisonaiagents.Agent", _FakeAgent):
out = init_mod._generate_agents_md(tmp_path, "gpt-4o-mini")
assert out == "# Generated\nBuild: make\n"
@@ -1,115 +0,0 @@
"""Tests for run-outcome exit semantics on `praisonai run` (issue #3344).
`run.py` must exit non-zero and emit a machine-readable failure object when an
agent run does not produce a result (swallowed LLM/auth error, guardrail block,
tool failure, or `max_iter` without completion), instead of always reporting
success and exiting 0. A genuine (non-empty) result still exits 0 with unchanged
text output.
"""
import pytest
import typer
from praisonai_code.cli.commands import run as run_cmd
class _RecordingOutput:
"""Minimal output controller capturing failure-reporting calls."""
def __init__(self):
self.results = []
self.errors = []
self.printed_errors = []
self.is_json_mode = False
def emit_result(self, message=None, data=None):
self.results.append((message, data))
def emit_error(self, message=None, data=None):
self.errors.append((message, data))
def print_error(self, message, code=None, remediation=None):
self.printed_errors.append((message, code, remediation))
@pytest.mark.parametrize(
"result,expected",
[
("A real answer", True),
(" spaced answer ", True),
({"data": 1}, True),
(None, False),
("", False),
(" ", False),
],
)
def test_run_succeeded_classification(result, expected):
assert run_cmd._run_succeeded(result) is expected
def test_report_run_failure_exits_nonzero_and_emits_status():
output = _RecordingOutput()
with pytest.raises(typer.Exit) as exc:
run_cmd._report_run_failure(output)
assert exc.value.exit_code == 1
# Machine-readable failure object for --output json consumers.
assert output.results, "expected a result event"
_, result_data = output.results[-1]
assert result_data.get("status") == "failed"
assert output.errors, "expected an error event"
_, error_data = output.errors[-1]
assert error_data.get("status") == "failed"
# Human-facing error with a code and an actionable remediation.
assert output.printed_errors, "expected a printed error"
_, code, remediation = output.printed_errors[-1]
assert code == "run_failed"
assert remediation
def test_try_attach_runtime_exits_nonzero_on_empty_runtime_result(monkeypatch):
"""A warm-runtime run returning an empty result must fail, not exit 0.
Regression guard: the warm-runtime attach path previously reported success
unconditionally, so a swallowed failure on the runtime still exited 0 and
broke CI/scripted failure detection.
"""
output = _RecordingOutput()
monkeypatch.setattr(run_cmd, "get_output_controller", lambda: output)
class _Descriptor:
pass
class _RuntimeUnavailable(Exception):
pass
class _RuntimeClient:
def __init__(self, descriptor):
pass
def run(self, prompt, model=None, session_id=None):
return None
import sys
import types
fake_runtime = types.ModuleType("praisonai_code.runtime")
fake_runtime.get_runtime_descriptor = lambda require_compatible=True: _Descriptor()
fake_runtime.RuntimeClient = _RuntimeClient
fake_runtime.RuntimeUnavailable = _RuntimeUnavailable
monkeypatch.setitem(sys.modules, "praisonai_code.runtime", fake_runtime)
with pytest.raises(typer.Exit) as exc:
run_cmd._try_attach_runtime(
"hello",
model=None,
output_mode=None,
session_id=None,
)
assert exc.value.exit_code == 1
assert output.printed_errors, "expected a printed failure"
assert output.printed_errors[-1][1] == "run_failed"
@@ -1,43 +0,0 @@
from pathlib import Path
from unittest.mock import patch
from praisonai_code.cli.commands.ui import _launch_aiui_app
def test_launch_aiui_copies_default_app_using_utf8(tmp_path, monkeypatch):
"""Regression test for Windows UnicodeDecodeError when copying
bundled default_app.py.
"""
# Create a fake bundled UI app containing UTF-8 characters.
bundled = tmp_path / "default_app.py"
expected = '"""Hello — café 😀 中文"""\n'
bundled.write_text(expected, encoding="utf-8")
# Redirect ~/.praisonai to our temporary directory.
monkeypatch.setattr(Path, "home", lambda: tmp_path)
# Pretend the bundled file exists.
with patch(
"praisonai_code.cli.commands.ui._resolve_bundled_default_app",
return_value=bundled,
), patch(
"importlib.util.find_spec",
return_value=object(),
), patch(
"subprocess.run"
):
_launch_aiui_app(
app_dir="ui",
default_app_name="ui_chat",
port=8081,
host="127.0.0.1",
app_file=None,
reload=False,
ui_name="Chat",
)
created = tmp_path / ".praisonai" / "ui" / "app.py"
assert created.exists()
assert created.read_text(encoding="utf-8") == expected
@@ -1,6 +1,6 @@
{
"version": "1.5.87",
"lastUpdated": "2026-07-25",
"lastUpdated": "2026-07-23",
"generatedBy": "praisonai._dev.parity.generator",
"sourceOfTruth": "Python SDK (praisonaiagents)",
"status": "PARITY_ACHIEVED",
-42
View File
@@ -117,45 +117,3 @@ praisonai-train validate data/tamil.jsonl --out data/clean.jsonl
Add a language/domain by registering a `Recipe`, or a new QC rule by registering a
`RowCheck` (see `praisonai_train/data/`), and they show up automatically.
## Verify → export → train → re-verify (learn from real agent runs)
Beyond synthetic generation, you can fine-tune on the agent's *own* verified
behaviour. Given a trials report (K scored attempts per case with captured
trajectories), `from-trials` keeps the passing attempts and writes a
trainer-ready dataset — closing the loop from verification to data generation.
```bash
# 1) export the passing attempts as a ShareGPT dataset (+ provenance sidecar)
praisonai-train data from-trials trials.json -o data/train.jsonl
# 2) fine-tune with the existing trainer, unchanged
praisonai-train llm --dataset data/train.jsonl
# 3) re-run the trials on the fine-tuned model and compare pass-rate to measure gain
```
```python
from praisonai_train.data import export_trials
summary = export_trials(
report, "data/train.jsonl",
only_passed=True, # rejection sampling on the verifier (default)
frontier_only=True, # skip saturated / zero-pass cases (default)
)
print(summary) # written / skipped_failed / skipped_tool_runs / skipped_no_text / ...
```
Selection defaults: unscored attempts are never candidates; `only_passed` keeps
verifier-passed attempts; `frontier_only` restricts to cases with
`0 < pass_rate < 1` (saturated cases add near-duplicates of mastered behaviour,
zero-pass cases have nothing to export); tool-using runs are excluded by default
(`--all`, `--include-saturated` relax these). A `{out}.jsonl.meta.json` sidecar
maps every line to its case id, attempt index and score, and emission order is
deterministic → reproducible files. Add `--qc` to run rows through the same QC
filter as `validate`, or `--format alpaca` for instruction/input/output.
> **Honest selection:** `only_passed` is *rejection sampling* — it amplifies
> behaviour the agent already produces and inherits any bias in the scorer/judge
> that decided a run "passed". It cannot teach behaviour the agent never
> exhibited; treat it as reinforcing verified wins, not as an oracle.
@@ -198,46 +198,6 @@ def dedup_data(
typer.echo(f" wrote {kept} unique rows -> {out_path}")
@app.command("from-trials")
def from_trials(
report: str = typer.Argument(..., help="Serialised trials report JSON"),
out: Optional[str] = typer.Option(None, "--out", "-o", help="Output JSONL"),
all_attempts: bool = typer.Option(False, "--all", help="Include failed/unscored attempts"),
include_saturated: bool = typer.Option(
False, "--include-saturated", help="Include all-pass / zero-pass cases"),
fmt: str = typer.Option("messages", "--format", help="messages (ShareGPT) | alpaca"),
qc: bool = typer.Option(False, "--qc", help="Run rows through the QC filter"),
):
"""Export passing agent trials to a trainer-ready SFT dataset (+ provenance).
Rejection sampling on the verifier: keeps verifier-passed attempts from
frontier cases (0 < pass_rate < 1) and writes ShareGPT ``conversations`` the
existing trainer consumes directly. A ``{out}.meta.json`` sidecar traces every
line to its case id, attempt index and score.
Example: praisonai-train data from-trials trials.json -o data/train.jsonl
praisonai-train llm --dataset data/train.jsonl
"""
from praisonai_train.data import export_trials
out_path = out or f"{Path(report).stem}.jsonl"
summary = export_trials(
report, out_path,
only_passed=not all_attempts,
frontier_only=not include_saturated,
format=fmt,
qc=qc,
)
typer.echo(f"\n─── from-trials: {report} ───")
typer.echo(f" {summary}")
typer.echo(f" wrote {summary.written} rows -> {out_path}")
typer.echo(f" provenance -> {out_path}.meta.json")
if summary.written == 0:
typer.echo("error: no rows exported — check --all/--include-saturated or "
"whether the report has passing, tool-free attempts", err=True)
raise typer.Exit(1)
@app.command("validate")
def validate_data(
dataset: Optional[str] = typer.Argument(None, help="Dataset JSONL to validate"),
@@ -8,7 +8,6 @@ from praisonai_train.data import checks as _checks # noqa: F401 registers chec
from praisonai_train.data import recipes as _recipes # noqa: F401 registers recipes
from praisonai_train.data.benchmark import BenchResult, benchmark_deployments
from praisonai_train.data.dedup import MinHashLSH, global_dedup, near_dedup
from praisonai_train.data.export_trials import ExportSummary, export_trials
from praisonai_train.data.generate import azure_sponsorship_guard, generate_dataset
from praisonai_train.data.intent import translation_intent
from praisonai_train.data.qc import filter_rows, score
@@ -16,12 +15,10 @@ from praisonai_train.data.registry import checks, recipes
__all__ = [
"BenchResult",
"ExportSummary",
"MinHashLSH",
"azure_sponsorship_guard",
"benchmark_deployments",
"checks",
"export_trials",
"filter_rows",
"generate_dataset",
"global_dedup",
@@ -1,341 +0,0 @@
"""Export a trials report into a supervised fine-tuning (SFT) dataset.
Closes the loop from *verification* to *data generation*: take a trials report
(K scored attempts per case with captured trajectories), keep the attempts that
passed the verifier, and write them as a dataset the existing trainer consumes
directly — turning "run the agent K times, keep what passed, fine-tune on it,
re-run to measure gain" into a supported pipeline instead of a hand-rolled
script.
The report's *serialised JSON* is the contract with the (sibling) core trials
engine, so the loader is duck-typed over plain dicts: any report exposing
``package`` and ``cases`` (each with ``case_id`` and ``attempts``) works, whether
it comes from the core engine or an interim adapter.
Selection discipline (each skip is counted, never silent):
* attempts with ``score is None`` are never candidates (unscored → cannot judge);
* ``only_passed`` (default) keeps only verifier-passed attempts — this is
rejection sampling: it amplifies existing behaviour and inherits any judge
bias in the scorer;
* ``frontier_only`` (default) restricts to cases with ``0 < pass_rate < 1`` —
saturated (all-pass) cases add K near-duplicates of mastered behaviour and
zero-pass cases have nothing to export;
* tool-using runs are excluded by default (``skipped_tool_runs``): exporting only
the final text of a run that used tools would train the model to answer
*without* the tools it actually needed.
Output is one JSON object per line in ShareGPT shape
(``{"conversations": [{"role", "content"}, ...]}``) — exactly what the trainer's
``process_dataset`` standardises — or Alpaca (``instruction``/``input``/``output``)
via ``format="alpaca"``. A ``{path}.meta.json`` provenance sidecar maps every
emitted line back to the case id, attempt index and score that produced it, and
records the selection options, so any dataset row is traceable to its exact run.
Emission order is deterministic (case order, then attempt order) → reproducible
files.
"""
from __future__ import annotations
import json
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Optional
@dataclass
class ExportSummary:
"""Counters returned by :func:`export_trials` — every attempt is accounted for."""
written: int = 0
skipped_unscored: int = 0
skipped_failed: int = 0
skipped_saturated: int = 0
skipped_tool_runs: int = 0
skipped_no_text: int = 0
def to_dict(self) -> dict[str, int]:
return asdict(self)
def __str__(self) -> str: # concise, CLI-friendly
return (
f"written={self.written} "
f"skipped_failed={self.skipped_failed} "
f"skipped_tool_runs={self.skipped_tool_runs} "
f"skipped_no_text={self.skipped_no_text} "
f"skipped_unscored={self.skipped_unscored} "
f"skipped_saturated={self.skipped_saturated}"
)
@dataclass
class _Line:
"""One emitted dataset row plus the provenance needed to trace it."""
row: dict[str, Any]
case_id: str
attempt_index: int
score: Any
# ── report access (duck-typed over dict or object) ────────────────────────────
def _get(obj: Any, key: str, default: Any = None) -> Any:
if isinstance(obj, dict):
return obj.get(key, default)
return getattr(obj, key, default)
def _load_report(report: Any) -> Any:
"""Accept a report object, a dict, or a path to a serialised JSON report."""
if isinstance(report, (str, Path)):
return json.loads(Path(report).read_text())
return report
def _pass_threshold(attempt: Any) -> Optional[float]:
"""The verifier's passing cutoff for this attempt, if the report carries one.
Looked up on the attempt (``pass_threshold`` / ``threshold`` / ``min_score``)
so a score-only report can still be judged against the exact bar the verifier
used, instead of the naive ``score > 0``.
"""
for key in ("pass_threshold", "threshold", "min_score"):
val = _get(attempt, key)
if val is not None:
try:
return float(val)
except (TypeError, ValueError):
continue
return None
def _passed(attempt: Any) -> bool:
"""A verifier-passed attempt.
Prefers an explicit ``passed`` flag. Otherwise compares the numeric score
against the report's own passing threshold when one is present
(``pass_threshold``/``threshold``/``min_score``); only when no threshold is
carried does it fall back to treating a positive score as a pass, matching the
"keep what passed" rejection-sampling contract for bare score-only reports.
"""
passed = _get(attempt, "passed")
if passed is not None:
return bool(passed)
score = _get(attempt, "score")
try:
score_f = float(score)
except (TypeError, ValueError):
return False
threshold = _pass_threshold(attempt)
if threshold is not None:
return score_f >= threshold
return score_f > 0
def _used_tools(attempt: Any) -> bool:
tool_calls = _get(attempt, "tool_calls")
if tool_calls:
return True
for msg in _get(attempt, "messages", []) or []:
if _get(msg, "tool_calls") or _get(msg, "role") == "tool":
return True
return False
def _conversation_from(attempt: Any) -> Optional[list[dict[str, str]]]:
"""Build ShareGPT turns from an attempt's captured messages.
History is excluded (only the attempt's own messages), the conversation is
truncated at the final assistant message, and the assistant content is the
raw generated text — never a re-serialisation of parsed output.
"""
messages = _get(attempt, "messages")
turns: list[dict[str, str]] = []
if messages:
for msg in messages:
if _get(msg, "history"):
continue
role = _get(msg, "role")
content = _get(msg, "content")
if role is None or content is None:
continue
turns.append({"role": str(role), "content": str(content)})
# Truncate at the final assistant message: drop any trailing non-assistant
# turns so the row always ends on the target the model should learn.
last_assistant = max(
(i for i, t in enumerate(turns) if t["role"] == "assistant"),
default=None,
)
if last_assistant is None:
return None
turns = turns[: last_assistant + 1]
return turns
# Fallback for reports that store input/output rather than messages.
output = _get(attempt, "output")
if output is None:
return None
user = _get(attempt, "input") or _get(attempt, "input_text")
system = _get(attempt, "system")
if system:
turns.append({"role": "system", "content": str(system)})
if user:
turns.append({"role": "user", "content": str(user)})
turns.append({"role": "assistant", "content": str(output)})
return turns
def _to_alpaca(turns: list[dict[str, str]]) -> dict[str, str]:
"""Flatten ShareGPT turns to instruction/input/output (Alpaca)."""
system = "".join(t["content"] for t in turns if t["role"] == "system")
users = [t["content"] for t in turns if t["role"] == "user"]
output = next(
(t["content"] for t in reversed(turns) if t["role"] == "assistant"), ""
)
if system:
# System text is the instruction; every user turn is the input so the
# (often sole) user prompt is never dropped.
instruction, input_users = system, users
else:
# No system: first user turn is the instruction, the rest is the input.
instruction, input_users = (users[0] if users else ""), users[1:]
return {
"instruction": instruction,
"input": "\n".join(input_users),
"output": output,
}
def _classify(attempt: Any, only_passed: bool, summary: ExportSummary):
"""Ordered filter; returns turns to emit or ``None`` (and bumps a counter)."""
if _get(attempt, "score") is None:
# Unscored can't be judged as passing, so they're only ever candidates
# under ``--all`` (only_passed=False), which explicitly opts into them.
if only_passed:
summary.skipped_unscored += 1
return None
elif only_passed and not _passed(attempt):
summary.skipped_failed += 1
return None
if _used_tools(attempt):
summary.skipped_tool_runs += 1
return None
turns = _conversation_from(attempt)
if not turns:
summary.skipped_no_text += 1
return None
return turns
def _pass_rate(attempts: list[Any]) -> Optional[float]:
scored = [a for a in attempts if _get(a, "score") is not None]
if not scored:
return None
return sum(1 for a in scored if _passed(a)) / len(scored)
def export_trials(
report: Any,
path: str | Path,
*,
only_passed: bool = True,
frontier_only: bool = True,
format: str = "messages",
qc: bool = False,
qc_cfg: Optional[dict] = None,
) -> ExportSummary:
"""Export passing trial attempts to an SFT dataset JSONL + provenance sidecar.
Args:
report: a report object, a dict, or a path to serialised report JSON.
path: destination ``.jsonl`` (a ``{path}.meta.json`` sidecar is written
alongside).
only_passed: keep only verifier-passed attempts (rejection sampling).
frontier_only: keep only cases with ``0 < pass_rate < 1``.
format: ``"messages"`` (ShareGPT ``conversations``) or ``"alpaca"``.
qc: run emitted rows through ``data.qc.filter_rows`` before writing.
qc_cfg: config forwarded to the QC filter.
Returns:
:class:`ExportSummary` with per-reason skip counters.
"""
if format not in ("messages", "alpaca"):
raise ValueError(f"format must be 'messages' or 'alpaca', got {format!r}")
report = _load_report(report)
summary = ExportSummary()
lines: list[_Line] = []
for case in _get(report, "cases", []) or []:
attempts = list(_get(case, "attempts", []) or [])
case_id = str(_get(case, "case_id", _get(case, "id", "")))
if frontier_only:
rate = _pass_rate(attempts)
if rate is None or not (0.0 < rate < 1.0):
summary.skipped_saturated += len(attempts)
continue
for idx, attempt in enumerate(attempts):
turns = _classify(attempt, only_passed, summary)
if turns is None:
continue
row = {"conversations": turns} if format == "messages" else _to_alpaca(turns)
lines.append(
_Line(row=row, case_id=case_id, attempt_index=idx,
score=_get(attempt, "score"))
)
if qc:
kept_rows = filter_rows_for_export([ln.row for ln in lines], qc_cfg, format)
kept_ids = {id(r) for r in kept_rows}
lines = [ln for ln in lines if id(ln.row) in kept_ids]
out_path = Path(path)
out_path.parent.mkdir(parents=True, exist_ok=True)
with open(out_path, "w") as fh:
for ln in lines:
fh.write(json.dumps(ln.row, ensure_ascii=False) + "\n")
summary.written = len(lines)
meta = {
"package": _get(report, "package", _get(report, "package_name")),
"source_report": _get(report, "source", _get(report, "report_id")),
"options": {
"only_passed": only_passed,
"frontier_only": frontier_only,
"format": format,
"qc": qc,
},
"summary": summary.to_dict(),
"lines": [
{"case_id": ln.case_id, "attempt_index": ln.attempt_index, "score": ln.score}
for ln in lines
],
}
Path(f"{out_path}.meta.json").write_text(
json.dumps(meta, ensure_ascii=False, indent=2)
)
return summary
def filter_rows_for_export(rows: list[dict], cfg: Optional[dict], format: str) -> list[dict]:
"""Run rows through the existing QC filter, adapting ShareGPT to its fields.
``data.qc`` operates on ``instruction``/``input``/``output`` rows, so ShareGPT
rows are mapped to a QC view keyed back to the original row identity.
"""
from praisonai_train.data.qc import filter_rows
if format == "alpaca":
kept = filter_rows(rows, cfg)
kept_set = {id(r) for r in kept}
return [r for r in rows if id(r) in kept_set]
views: list[dict] = []
view_to_row: dict[int, dict] = {}
for r in rows:
view = _to_alpaca(r["conversations"])
views.append(view)
view_to_row[id(view)] = r
kept_views = filter_rows(views, cfg)
kept_view_ids = {id(v) for v in kept_views}
return [view_to_row[id(v)] for v in views if id(v) in kept_view_ids]
@@ -1,265 +0,0 @@
"""Tests for the trials→SFT exporter (``data/export_trials.py``).
The report is duck-typed over plain dicts (the serialised contract with the
sibling core trials engine), so these fixtures build reports as dicts.
"""
import json
import pytest
from praisonai_train.data import ExportSummary, export_trials
def _attempt(role_text, score, *, passed=None, tools=False):
"""A minimal attempt record: system+user+assistant messages, plus a score."""
messages = [
{"role": "system", "content": "You are helpful."},
{"role": "user", "content": "2+2?"},
{"role": "assistant", "content": role_text},
]
a = {"messages": messages, "score": score}
if passed is not None:
a["passed"] = passed
if tools:
a["tool_calls"] = [{"name": "calc", "args": {}}]
return a
def _report(cases, package="demo"):
return {"package": package, "source": "trials.json", "cases": cases}
def _read_jsonl(path):
return [json.loads(ln) for ln in path.read_text().splitlines() if ln.strip()]
# ── selection: only_passed (default) ──────────────────────────────────────────
def test_export_only_passed_default(tmp_path):
"""Failed and unscored attempts are skipped with the right counters."""
case = {
"case_id": "c1",
"attempts": [
_attempt("4", score=1.0, passed=True), # kept
_attempt("5", score=0.0, passed=False), # skipped_failed
_attempt("?", score=None), # skipped_unscored
],
}
out = tmp_path / "train.jsonl"
summary = export_trials(_report([case]), out, frontier_only=False)
assert isinstance(summary, ExportSummary)
assert summary.written == 1
assert summary.skipped_failed == 1
assert summary.skipped_unscored == 1
rows = _read_jsonl(out)
assert rows[0]["conversations"][-1]["content"] == "4"
# ── selection: frontier_only (default) ────────────────────────────────────────
def test_frontier_only_default(tmp_path):
"""A saturated (all-pass) case contributes zero rows by default."""
saturated = {
"case_id": "sat",
"attempts": [
_attempt("4", score=1.0, passed=True),
_attempt("4", score=1.0, passed=True),
],
}
out = tmp_path / "train.jsonl"
summary = export_trials(_report([saturated]), out) # frontier_only default True
assert summary.written == 0
assert summary.skipped_saturated == 2
# With --include-saturated (frontier_only=False) it now exports.
out2 = tmp_path / "train2.jsonl"
summary2 = export_trials(_report([saturated]), out2, frontier_only=False)
assert summary2.written == 2
# ── selection: tool-using runs excluded ───────────────────────────────────────
def test_tool_runs_excluded_and_counted(tmp_path):
case = {
"case_id": "c1",
"attempts": [
_attempt("4", score=1.0, passed=True), # kept
_attempt("4", score=1.0, passed=True, tools=True), # skipped_tool_runs
],
}
out = tmp_path / "train.jsonl"
summary = export_trials(_report([case]), out, frontier_only=False)
assert summary.written == 1
assert summary.skipped_tool_runs == 1
# ── shape: ShareGPT conversations ─────────────────────────────────────────────
def test_conversation_shape_sharegpt(tmp_path):
"""Rows are ShareGPT ``conversations`` with role/content dicts, truncated at
the final assistant turn."""
case = {
"case_id": "c1",
"attempts": [
_attempt("4", score=1.0, passed=True),
_attempt("5", score=0.0, passed=False),
],
}
out = tmp_path / "train.jsonl"
export_trials(_report([case]), out, frontier_only=False)
rows = _read_jsonl(out)
convo = rows[0]["conversations"]
assert [t["role"] for t in convo] == ["system", "user", "assistant"]
assert all({"role", "content"} <= set(t) for t in convo)
assert convo[-1]["role"] == "assistant"
def test_trailing_non_assistant_truncated(tmp_path):
"""Turns after the final assistant message are dropped."""
a = _attempt("4", score=1.0, passed=True)
a["messages"].append({"role": "user", "content": "thanks"})
out = tmp_path / "train.jsonl"
export_trials(_report([{"case_id": "c1", "attempts": [a,
_attempt("x", score=0.0, passed=False)]}]),
out, frontier_only=False)
convo = _read_jsonl(out)[0]["conversations"]
assert convo[-1]["role"] == "assistant"
# ── provenance sidecar ────────────────────────────────────────────────────────
def test_provenance_sidecar(tmp_path):
"""Every emitted line maps to case id + attempt index + score."""
case = {
"case_id": "c1",
"attempts": [
_attempt("4", score=0.9, passed=True),
_attempt("5", score=0.0, passed=False),
],
}
out = tmp_path / "train.jsonl"
export_trials(_report([case]), out, frontier_only=False)
meta = json.loads((tmp_path / "train.jsonl.meta.json").read_text())
assert meta["package"] == "demo"
assert meta["options"]["only_passed"] is True
assert len(meta["lines"]) == 1
line = meta["lines"][0]
assert line["case_id"] == "c1"
assert line["attempt_index"] == 0
assert line["score"] == 0.9
assert meta["summary"]["written"] == 1
# ── determinism ───────────────────────────────────────────────────────────────
def test_deterministic_output(tmp_path):
"""Same report → byte-identical JSONL (case order, then attempt order)."""
cases = [
{"case_id": "c1", "attempts": [
_attempt("a", score=1.0, passed=True),
_attempt("b", score=0.0, passed=False)]},
{"case_id": "c2", "attempts": [
_attempt("c", score=1.0, passed=True),
_attempt("d", score=0.0, passed=False)]},
]
o1 = tmp_path / "a.jsonl"
o2 = tmp_path / "b.jsonl"
export_trials(_report(cases), o1, frontier_only=False)
export_trials(_report(cases), o2, frontier_only=False)
assert o1.read_bytes() == o2.read_bytes()
# ── --all includes unscored attempts ─────────────────────────────────────────
def test_all_includes_unscored(tmp_path):
"""With only_passed=False (--all), unscored attempts are exported, not dropped."""
case = {
"case_id": "c1",
"attempts": [
_attempt("4", score=1.0, passed=True), # kept
_attempt("?", score=None), # kept under --all
],
}
out = tmp_path / "train.jsonl"
summary = export_trials(
_report([case]), out, only_passed=False, frontier_only=False)
assert summary.written == 2
assert summary.skipped_unscored == 0
# ── alpaca fallback ───────────────────────────────────────────────────────────
def test_alpaca_format(tmp_path):
case = {"case_id": "c1", "attempts": [
_attempt("4", score=1.0, passed=True),
_attempt("x", score=0.0, passed=False)]}
out = tmp_path / "train.jsonl"
export_trials(_report([case]), out, frontier_only=False, format="alpaca")
row = _read_jsonl(out)[0]
assert set(row) == {"instruction", "input", "output"}
assert row["output"] == "4"
def test_alpaca_keeps_user_prompt_with_system(tmp_path):
"""A system+user conversation keeps the user prompt in Alpaca ``input``."""
case = {"case_id": "c1", "attempts": [
_attempt("4", score=1.0, passed=True),
_attempt("x", score=0.0, passed=False)]}
out = tmp_path / "train.jsonl"
export_trials(_report([case]), out, frontier_only=False, format="alpaca")
row = _read_jsonl(out)[0]
assert row["instruction"] == "You are helpful."
assert row["input"] == "2+2?" # the sole user prompt is never dropped
assert row["output"] == "4"
def test_alpaca_no_system_uses_first_user_as_instruction(tmp_path):
"""Without a system turn, first user is instruction and the rest is input."""
a = {"messages": [
{"role": "user", "content": "translate"},
{"role": "user", "content": "hello"},
{"role": "assistant", "content": "bonjour"},
], "score": 1.0, "passed": True}
out = tmp_path / "train.jsonl"
export_trials(_report([{"case_id": "c1", "attempts": [
a, _attempt("x", score=0.0, passed=False)]}]),
out, frontier_only=False, format="alpaca")
row = _read_jsonl(out)[0]
assert row["instruction"] == "translate"
assert row["input"] == "hello"
assert row["output"] == "bonjour"
# ── score threshold (no explicit passed flag) ─────────────────────────────────
def test_score_below_threshold_is_failed(tmp_path):
"""A positive score below the report's threshold counts as a failure."""
case = {"case_id": "c1", "attempts": [
{"messages": _attempt("4", 1.0)["messages"], "score": 0.9, "pass_threshold": 0.5},
{"messages": _attempt("5", 0.0)["messages"], "score": 0.3, "pass_threshold": 0.5},
]}
out = tmp_path / "train.jsonl"
summary = export_trials(_report([case]), out, frontier_only=False)
assert summary.written == 1 # only the 0.9 passes the 0.5 bar
assert summary.skipped_failed == 1
assert _read_jsonl(out)[0]["conversations"][-1]["content"] == "4"
def test_invalid_format_raises(tmp_path):
with pytest.raises(ValueError):
export_trials(_report([]), tmp_path / "x.jsonl", format="bogus")
# ── loads from a path ─────────────────────────────────────────────────────────
def test_loads_report_from_path(tmp_path):
case = {"case_id": "c1", "attempts": [
_attempt("4", score=1.0, passed=True),
_attempt("x", score=0.0, passed=False)]}
report_path = tmp_path / "trials.json"
report_path.write_text(json.dumps(_report([case])))
out = tmp_path / "train.jsonl"
summary = export_trials(str(report_path), out, frontier_only=False)
assert summary.written == 1
+1 -1
View File
@@ -1,6 +1,6 @@
{
"version": "1.5.87",
"lastUpdated": "2026-07-25",
"lastUpdated": "2026-07-23",
"generatedBy": "praisonai._dev.parity.generator",
"sourceOfTruth": "Python SDK (praisonaiagents)",
"summary": {
+1 -1
View File
@@ -1,6 +1,6 @@
# Feature Parity Tracker
> **Version:** 1.5.87 | **Last Updated:** 2026-07-25
> **Version:** 1.5.87 | **Last Updated:** 2026-07-23
> **Source of Truth:** Python SDK (praisonaiagents)
## Summary
+2 -7
View File
@@ -7,7 +7,6 @@ It creates a FastAPI-based web service for deploying agents.
from typing import Any, Dict, List, Optional, Union
import asyncio
import os
from praisonaiagents import AgentOSConfig, AgentOSProtocol
@@ -197,13 +196,9 @@ class AgentOS:
else:
raise HTTPException(status_code=400, detail="No agents available")
# Call the agent without blocking the event loop: prefer the async
# twin, otherwise offload the sync call to a worker thread.
# Call the agent
try:
if hasattr(agent, "achat") and callable(getattr(agent, "achat")):
response = await agent.achat(request.message)
else:
response = await asyncio.to_thread(agent.chat, request.message)
response = agent.chat(request.message)
return ChatResponse(
response=str(response),
agent_name=getattr(agent, 'name', 'unknown'),
+18 -55
View File
@@ -14,7 +14,6 @@ from typing import Any, Awaitable, Dict, List, Optional, Type, TypeVar
import os
import json
import asyncio
import contextvars
import threading
from praisonai._logging import get_logger
@@ -536,13 +535,6 @@ class BaseAutoGenerator:
if registered is not None:
return registered
# Preserve the original LiteLLM failure so that, if the OpenAI fallback
# also fails, the user sees BOTH tracebacks (via ``raise ... from``)
# instead of only the second one. Otherwise the real cause (auth/region
# error, a provider-only model, malformed response_format) survives only
# in a WARNING log line the caller likely never sees.
litellm_error: Optional[BaseException] = None
# Try LiteLLM first (preferred - supports 100+ providers)
if is_available("litellm"):
try:
@@ -575,7 +567,6 @@ class BaseAutoGenerator:
"LiteLLM structured completion failed (%s); "
"falling back to OpenAI SDK.", e
)
litellm_error = e
# Fallback to the core-owned OpenAI client (uses beta.chat.completions.parse).
# Delegating to praisonaiagents.llm.openai_client.OpenAIClient keeps the
@@ -586,28 +577,21 @@ class BaseAutoGenerator:
# The async path awaits the core coroutine directly. The sync path
# offloads the blocking core call via asyncio.to_thread so it never
# blocks the bridge event loop this coroutine runs on.
try:
if is_async:
return await client.aparse_structured_output(
messages=messages,
response_format=response_model,
model=model_name,
**kwargs,
)
return await asyncio.to_thread(
lambda: client.parse_structured_output(
messages=messages,
response_format=response_model,
model=model_name,
**kwargs,
)
if is_async:
return await client.aparse_structured_output(
messages=messages,
response_format=response_model,
model=model_name,
**kwargs,
)
except Exception as openai_error:
# Chain the LiteLLM traceback so users see BOTH provider
# failures, not just the OpenAI one that masked the real cause.
if litellm_error is not None:
raise openai_error from litellm_error
raise
return await asyncio.to_thread(
lambda: client.parse_structured_output(
messages=messages,
response_format=response_model,
model=model_name,
**kwargs,
)
)
# Neither available - raise helpful error
raise ImportError(
@@ -657,12 +641,8 @@ class BaseAutoGenerator:
def _runner() -> None:
try:
# The runner thread has no running loop, so run_sync() is legal
# here and dispatches to the shared background loop (one loop for
# the whole process) instead of standing up a fresh event loop
# per call — preserving LiteLLM/HTTPX per-loop connection pools.
result.append(
run_sync(
asyncio.run(
self._completion_impl(
response_model, messages, is_async=False, **kwargs
)
@@ -671,14 +651,7 @@ class BaseAutoGenerator:
except BaseException as e: # noqa: BLE001 - re-raised on caller thread
error.append(e)
# Run the worker under a copy of the caller's context so run_sync()
# resolves the SAME AsyncBridge the caller would (a scoped_bridge()
# override, else the shared default). A bare Thread does not inherit
# contextvars, which would otherwise leak this fallback onto the global
# bridge and bypass a caller's per-session loop isolation.
thread = threading.Thread(
target=contextvars.copy_context().run, args=(_runner,),
name="praisonai-sync-completion")
thread = threading.Thread(target=_runner, name="praisonai-sync-completion")
thread.start()
thread.join()
if error:
@@ -749,21 +722,11 @@ class BaseAutoGenerator:
def _runner() -> None:
try:
# The runner thread has no running loop, so run_sync() is legal
# here and dispatches to the shared background loop (one loop for
# the whole process) instead of standing up a fresh event loop
# per call — preserving LiteLLM/HTTPX per-loop connection pools.
result.append(run_sync(coro))
result.append(asyncio.run(coro))
except BaseException as e: # noqa: BLE001 - re-raised on caller thread
error.append(e)
# Run under a copy of the caller's context so run_sync() resolves the
# same AsyncBridge (scoped_bridge() override, else shared default); a
# bare Thread does not inherit contextvars and would otherwise bypass a
# caller's per-session loop isolation.
thread = threading.Thread(
target=contextvars.copy_context().run, args=(_runner,),
name="praisonai-sync-generate")
thread = threading.Thread(target=_runner, name="praisonai-sync-generate")
thread.start()
thread.join()
if error:
+3 -23
View File
@@ -16,12 +16,8 @@ Two guarantees matter:
executed twice, producing duplicate class/enum objects that break
``isinstance`` / equality checks across the old and new import paths.
2. **Laziness** — a package's *own* modules are not force-loaded before the
package itself is aliased; the CLI relies on lazy imports for start-up
performance. Guarantee #1 wins where the two conflict: when an old dotted
path is first aliased, its already-moved subtree is pre-registered (see
:func:`_register_submodules`) so a combined ``from old.sub import X`` cannot
re-enter the finder mid-import and double-load ``sub``.
2. **Laziness** — submodules are only imported on demand (the CLI relies on
lazy imports for start-up performance), so nothing is eagerly imported here.
The shim package at the old path keeps its *own* ``__path__`` (the now nearly
empty old directory). Because that directory no longer contains the moved
@@ -83,28 +79,12 @@ def _register_submodules(old_name: str, new_name: str, module) -> None:
create duplicate class/enum objects). Modules that fail to import eagerly
(e.g. optional heavy dependencies not installed) are skipped and resolved
lazily by :class:`_AliasFinder` on first successful use.
Eager pre-registration is required for correctness, not just startup speed:
if the alias for ``old_name.sub`` is *absent* when a combined statement such
as ``from old_name.sub import Symbol`` runs, CPython imports the parent
shim, then resolves ``.sub`` through :class:`_AliasFinder.find_spec`, which
calls ``import_module`` re-entrantly *inside* the parent's in-progress
import. Under the import lock that re-entrancy loads the moved file twice,
producing two distinct module objects (duplicate classes) and breaking
module identity (guarantee #1). Pre-stamping ``sys.modules[old_name.sub]``
here means ``_find_and_load`` returns the single canonical object directly
and never re-enters the finder.
"""
new_prefix = new_name + "."
old_prefix = old_name + "."
# Alias whatever is already imported first (e.g. the package __init__).
for mod_name, mod in list(sys.modules.items()):
# ``sys.modules`` can legitimately hold ``None`` placeholders (failed
# optional imports). Aliasing one would poison the old dotted name so
# that ``import old_name.sub`` raises "None in sys.modules" even though
# ``_AliasFinder`` could still resolve it on demand.
if mod is None:
continue
if mod_name == new_name or mod_name.startswith(new_prefix):
old_equiv = old_name + mod_name[len(new_name):]
sys.modules.setdefault(old_equiv, mod)
@@ -1,25 +1,603 @@
"""
Agent-Centric Tools for PraisonAI Interactive Mode (compatibility shim).
Agent-Centric Tools for PraisonAI Interactive Mode.
The canonical implementation now lives in
``praisonai_code.cli.features.agent_tools`` as part of the C0–C6 migration.
This module is a thin re-export so that both historical import spellings —
- ``from praisonai.cli.features.agent_tools import create_agent_centric_tools``
(submodule import), and
- ``from praisonai.cli.features import create_agent_centric_tools``
(attribute access delegated by ``cli/features/__init__.py``)
— resolve to the *same* single implementation, eliminating the earlier
duplication/drift (e.g. the ACP ``cwd`` forwarding fix).
These tools route file operations and code intelligence through LSP/ACP,
making the Agent the central orchestrator for all actions.
"""
from praisonai_code.cli.features.agent_tools import ( # noqa: F401
create_agent_centric_tools,
get_tool_descriptions,
)
import asyncio
import json
import logging
from typing import Callable, Dict, List, Optional, TYPE_CHECKING
__all__ = [
"create_agent_centric_tools",
"get_tool_descriptions",
]
if TYPE_CHECKING:
from .interactive_runtime import InteractiveRuntime
from .code_intelligence import CodeIntelligenceRouter
from .action_orchestrator import ActionOrchestrator
import os
from pathlib import Path
logger = logging.getLogger(__name__)
def _sanitize_filepath(filepath: str, workspace: Optional[str] = None) -> str:
"""Validate and sanitize a filepath against injection attacks.
Raises ValueError if the path is unsafe.
"""
# Reject null bytes (common injection technique)
if '\x00' in filepath:
raise ValueError("Null bytes are not allowed in file paths")
# Reject obvious shell meta-characters in file names
if any(ch in filepath for ch in ('|', ';', '`', '$', '&', '\n', '\r')):
raise ValueError(f"Unsafe characters in filepath: {filepath!r}")
# Reject absolute paths — files must be relative to workspace
if os.path.isabs(filepath):
raise ValueError(f"Absolute paths are not allowed: {filepath!r}")
# Normalize and reject path traversal
normalized = os.path.normpath(filepath)
if '..' in normalized.split(os.sep):
raise ValueError(f"Path traversal detected in: {filepath!r}")
# If we have a workspace, ensure the resolved path stays within it
if workspace:
resolved = Path(os.path.realpath(os.path.join(workspace, normalized)))
ws_resolved = Path(os.path.realpath(workspace))
try:
resolved.relative_to(ws_resolved)
except ValueError:
raise ValueError(
f"Path {filepath!r} resolves outside workspace {workspace!r}"
)
return normalized
def _sanitize_command(command: str) -> str:
"""Basic command sanitization — reject common injection vectors.
The command still goes through ACP approval flow, but this prevents
the most egregious prompt-injection-to-shell-injection chains.
Raises ValueError if dangerous patterns are detected.
"""
if '\x00' in command:
raise ValueError("Null bytes are not allowed in commands")
# Reject command chaining operators that indicate injection
DANGEROUS_PATTERNS = [
'$(', '`', # Command substitution
'&&', '||', # Command chaining
'>>', '>', # Output redirection
'|', ';', '&', # Pipe and separators
'\n', '\r' # Line breaks
]
for pattern in DANGEROUS_PATTERNS:
if pattern in command:
raise ValueError(
f"Potentially unsafe command pattern detected: {pattern!r} "
f"in command: {command!r}. Use separate commands instead."
)
return command
def create_agent_centric_tools(
runtime: "InteractiveRuntime",
router: "CodeIntelligenceRouter" = None,
orchestrator: "ActionOrchestrator" = None
) -> List[Callable]:
"""
Create tools that route through LSP/ACP for agent-centric architecture.
Args:
runtime: The InteractiveRuntime instance
router: Optional CodeIntelligenceRouter (created if not provided)
orchestrator: Optional ActionOrchestrator (created if not provided)
Returns:
List of tool functions for the Agent
"""
from .code_intelligence import CodeIntelligenceRouter
from .action_orchestrator import ActionOrchestrator
if router is None:
router = CodeIntelligenceRouter(runtime)
if orchestrator is None:
orchestrator = ActionOrchestrator(runtime)
# Helper to run async functions synchronously using the shared bridge
from praisonai._async_bridge import run_sync
# =========================================================================
# ACP-Powered File Tools
# =========================================================================
def acp_create_file(filepath: str, content: str) -> str:
"""
Create a file through ACP with plan/approve/apply/verify flow.
This tool routes file creation through the ActionOrchestrator,
ensuring proper tracking, approval, and verification.
Args:
filepath: Path to the file to create (relative to workspace)
content: Content to write to the file
Returns:
JSON string with result including plan status and verification
"""
async def _create():
# Validate filepath
try:
safe_path = _sanitize_filepath(
filepath,
workspace=getattr(runtime.config, 'workspace', None)
)
except ValueError as e:
return json.dumps({"success": False, "error": str(e)})
# Build a detailed prompt for the orchestrator
prompt = f"create file {safe_path}"
# Create plan
result = await orchestrator.create_plan(prompt)
if not result.success:
return json.dumps({
"success": False,
"error": result.error,
"read_only": result.read_only_blocked
})
plan = result.plan
# Update the plan step with actual content
if plan.steps:
plan.steps[0].params["content"] = content
# Approve based on runtime config
auto_approve = runtime.config.approval_mode == "auto"
approved = await orchestrator.approve_plan(plan, auto=auto_approve)
if not approved and runtime.config.approval_mode == "manual":
return json.dumps({
"success": False,
"error": "Manual approval required",
"plan": plan.to_dict(),
"requires_approval": True
})
# Apply the plan
result = await orchestrator.apply_plan(plan, force=auto_approve)
if not result.success:
return json.dumps({
"success": False,
"error": result.error,
"plan": plan.to_dict()
})
# Verify
result = await orchestrator.verify_plan(plan)
return json.dumps({
"success": result.success,
"file_created": filepath,
"plan_id": plan.id,
"status": plan.status.value,
"verified": result.success
})
return run_sync(_create())
def acp_edit_file(filepath: str, new_content: str) -> str:
"""
Edit a file through ACP with plan/approve/apply/verify flow.
Args:
filepath: Path to the file to edit (relative to workspace)
new_content: New content for the file
Returns:
JSON string with result including plan status
"""
async def _edit():
try:
safe_path = _sanitize_filepath(
filepath,
workspace=getattr(runtime.config, 'workspace', None)
)
except ValueError as e:
return json.dumps({"success": False, "error": str(e)})
prompt = f"edit file {safe_path}"
result = await orchestrator.create_plan(prompt)
if not result.success:
return json.dumps({
"success": False,
"error": result.error,
"read_only": result.read_only_blocked
})
plan = result.plan
# Update step with new content
if plan.steps:
plan.steps[0].params["new_content"] = new_content
auto_approve = runtime.config.approval_mode == "auto"
approved = await orchestrator.approve_plan(plan, auto=auto_approve)
if not approved and runtime.config.approval_mode == "manual":
return json.dumps({
"success": False,
"error": "Manual approval required",
"requires_approval": True
})
result = await orchestrator.apply_plan(plan, force=auto_approve)
return json.dumps({
"success": result.success,
"file_edited": filepath,
"plan_id": plan.id,
"error": result.error
})
return run_sync(_edit())
def acp_delete_file(filepath: str) -> str:
"""
Delete a file through ACP with plan/approve/apply/verify flow.
Args:
filepath: Path to the file to delete
Returns:
JSON string with result
"""
async def _delete():
try:
safe_path = _sanitize_filepath(
filepath,
workspace=getattr(runtime.config, 'workspace', None)
)
except ValueError as e:
return json.dumps({"success": False, "error": str(e)})
prompt = f"delete file {safe_path}"
result = await orchestrator.create_plan(prompt)
if not result.success:
return json.dumps({
"success": False,
"error": result.error
})
plan = result.plan
# Delete requires explicit approval even in auto mode
auto_approve = runtime.config.approval_mode == "auto"
approved = await orchestrator.approve_plan(plan, auto=auto_approve)
if not approved:
return json.dumps({
"success": False,
"error": "Delete requires approval",
"requires_approval": True
})
result = await orchestrator.apply_plan(plan)
result = await orchestrator.verify_plan(plan)
return json.dumps({
"success": result.success,
"file_deleted": filepath,
"verified": result.success
})
return run_sync(_delete())
def acp_execute_command(command: str, cwd: str = None) -> str:
"""
Execute a shell command through ACP with tracking.
Args:
command: The command to execute
cwd: Working directory (optional)
Returns:
JSON string with command output
"""
async def _execute():
try:
safe_cmd = _sanitize_command(command)
except ValueError as e:
return json.dumps({"success": False, "error": str(e)})
prompt = f"run command: {safe_cmd}"
result = await orchestrator.create_plan(prompt)
if not result.success:
return json.dumps({
"success": False,
"error": result.error
})
plan = result.plan
# Commands require approval
auto_approve = runtime.config.approval_mode == "auto"
approved = await orchestrator.approve_plan(plan, auto=auto_approve)
if not approved:
return json.dumps({
"success": False,
"error": "Command requires approval",
"requires_approval": True
})
result = await orchestrator.apply_plan(plan)
# Extract command result
if plan.steps and plan.steps[0].result:
cmd_result = plan.steps[0].result
return json.dumps({
"success": result.success,
"command": command,
"stdout": cmd_result.get("stdout", ""),
"stderr": cmd_result.get("stderr", ""),
"returncode": cmd_result.get("returncode", -1)
})
return json.dumps({
"success": result.success,
"error": result.error
})
return run_sync(_execute())
# =========================================================================
# LSP-Powered Code Intelligence Tools
# =========================================================================
def lsp_list_symbols(file_path: str) -> str:
"""
List all symbols (functions, classes, methods) in a file using LSP.
Falls back to regex-based extraction if LSP is unavailable.
Args:
file_path: Path to the file to analyze
Returns:
JSON string with list of symbols and their locations
"""
async def _list():
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(
f"list all functions and classes in {file_path}",
file_path=file_path
),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP list_symbols timed out (10s)", "success": False})
return run_sync(_list())
def lsp_find_definition(symbol: str, file_path: str = None) -> str:
"""
Find where a symbol is defined using LSP.
Args:
symbol: The symbol name to find
file_path: Optional file path for context
Returns:
JSON string with definition location(s)
"""
async def _find():
query = f"where is {symbol} defined"
if file_path:
query += f" in {file_path}"
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(query, file_path=file_path),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP find_definition timed out (10s)", "success": False})
return run_sync(_find())
def lsp_find_references(symbol: str, file_path: str = None) -> str:
"""
Find all references to a symbol using LSP.
Args:
symbol: The symbol name to find references for
file_path: Optional file path for context
Returns:
JSON string with reference locations
"""
async def _find():
query = f"find all references to {symbol}"
if file_path:
query += f" in {file_path}"
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(query, file_path=file_path),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP find_references timed out (10s)", "success": False})
return run_sync(_find())
def lsp_get_diagnostics(file_path: str = None) -> str:
"""
Get diagnostics (errors, warnings) for a file using LSP.
Args:
file_path: Path to the file (optional, gets all if not specified)
Returns:
JSON string with diagnostic information
"""
async def _get():
query = "show all errors and warnings"
if file_path:
query += f" in {file_path}"
# Add 10s timeout to prevent LSP from blocking autonomy
try:
result = await asyncio.wait_for(
router.handle_query(query, file_path=file_path),
timeout=10.0
)
return json.dumps(result.to_dict())
except asyncio.TimeoutError:
return json.dumps({"error": "LSP diagnostics timed out (10s)", "success": False})
return run_sync(_get())
# =========================================================================
# Basic File Tools (for read-only operations)
# =========================================================================
def read_file(filepath: str) -> str:
"""
Read content from a file.
This is a read-only operation that doesn't require ACP.
Args:
filepath: Path to the file to read
Returns:
File content or error message
"""
from pathlib import Path
try:
path = Path(filepath)
if not path.is_absolute():
path = Path(runtime.config.workspace) / filepath
# SECURITY: Ensure resolved path stays within workspace
resolved = path.resolve()
ws_resolved = Path(runtime.config.workspace).resolve()
try:
resolved.relative_to(ws_resolved)
except ValueError:
return json.dumps({"error": f"Path escapes workspace: {filepath}"})
if not resolved.exists():
return json.dumps({"error": f"File not found: {filepath}"})
content = resolved.read_text()
# Track in trace if enabled
if runtime._trace:
runtime._trace.add_entry(
category="file",
action="read",
params={"file": str(path)},
result={"size": len(content)}
)
return content
except Exception as e:
return json.dumps({"error": str(e)})
def list_files(directory: str = ".", pattern: str = "*") -> str:
"""
List files in a directory.
Args:
directory: Directory to list (relative to workspace)
pattern: Glob pattern to filter files
Returns:
JSON string with list of files
"""
from pathlib import Path
try:
path = Path(directory)
if not path.is_absolute():
path = Path(runtime.config.workspace) / directory
# SECURITY: Ensure resolved path stays within workspace
resolved = path.resolve()
ws_resolved = Path(runtime.config.workspace).resolve()
try:
resolved.relative_to(ws_resolved)
except ValueError:
return json.dumps({"error": f"Directory escapes workspace: {directory}"})
if not resolved.exists():
return json.dumps({"error": f"Directory not found: {directory}"})
files = []
for f in path.glob(pattern):
files.append({
"name": f.name,
"path": str(f.relative_to(runtime.config.workspace)),
"is_dir": f.is_dir(),
"size": f.stat().st_size if f.is_file() else None
})
return json.dumps({"files": files, "count": len(files)})
except Exception as e:
return json.dumps({"error": str(e)})
# Return all tools
return [
# ACP-powered file tools
acp_create_file,
acp_edit_file,
acp_delete_file,
acp_execute_command,
# LSP-powered code intelligence
lsp_list_symbols,
lsp_find_definition,
lsp_find_references,
lsp_get_diagnostics,
# Basic read-only tools
read_file,
list_files,
]
def get_tool_descriptions() -> Dict[str, str]:
"""Get descriptions of all agent-centric tools."""
return {
"acp_create_file": "Create a file through ACP with plan/approve/apply/verify",
"acp_edit_file": "Edit a file through ACP with tracking",
"acp_delete_file": "Delete a file through ACP (requires approval)",
"acp_execute_command": "Execute a shell command through ACP",
"lsp_list_symbols": "List symbols in a file using LSP",
"lsp_find_definition": "Find where a symbol is defined",
"lsp_find_references": "Find all references to a symbol",
"lsp_get_diagnostics": "Get errors and warnings for a file",
"read_file": "Read content from a file (read-only)",
"list_files": "List files in a directory",
}
@@ -45,33 +45,6 @@ def _start_interactive_mode(self, args):
# Set interactive mode flag
self._interactive_mode = True
# Initialize persistent session FIRST so the exact session that drives
# the conversation is also the one used for workspace binding below —
# resolving it once avoids a second, independent last-session lookup that
# could race a concurrent CLI process and bind tools to a different dir.
from .session import get_session_store
session_store = get_session_store()
resume_session_id = getattr(args, 'resume_session', None)
if resume_session_id == 'last':
unified_session = session_store.get_last_session()
if not unified_session:
unified_session = session_store.get_or_create()
elif resume_session_id:
unified_session = session_store.get_or_create(resume_session_id)
else:
unified_session = session_store.get_or_create()
# Session↔directory binding: when resuming a session, root the tools at
# the directory the session started in (persisted UnifiedSession.workspace)
# so `--continue`/`--session` resume INTO that directory instead of the
# unrelated cwd. Only applies on resume and only when the user has not
# already pinned a workspace via --workspace (PRAISONAI_WORKSPACE). If the
# recorded directory no longer exists, warn and fall back to cwd. Pass the
# already-resolved session so the binder and conversation agree exactly.
if not os.environ.get("PRAISONAI_WORKSPACE"):
_bind_resume_workspace(args, console, session=unified_session)
# Load interactive tools
tools_list = _load_interactive_tools(self)
@@ -117,8 +90,20 @@ def _start_interactive_mode(self, args):
# Check for verbose mode
verbose_mode = getattr(args, 'verbose', False) if hasattr(args, 'verbose') else False
# (Persistent session already resolved above, before workspace binding,
# so the binder and the conversation share the exact same session.)
# Initialize persistent session
from .session import get_session_store
session_store = get_session_store()
# Check for --resume flag or get/create session
resume_session_id = getattr(args, 'resume_session', None)
if resume_session_id == 'last':
unified_session = session_store.get_last_session()
if not unified_session:
unified_session = session_store.get_or_create()
elif resume_session_id:
unified_session = session_store.get_or_create(resume_session_id)
else:
unified_session = session_store.get_or_create()
# Set model from args or session
current_model = getattr(args, 'llm', None) or unified_session.current_model or os.environ.get('OPENAI_MODEL_NAME', 'gpt-4o-mini')
@@ -590,60 +575,6 @@ def _start_interactive_mode(self, args):
traceback.print_exc()
sys.exit(1)
def _bind_resume_workspace(args, console=None, session=None):
"""Root a resumed session at the directory it was created in.
Reads back the persisted ``UnifiedSession.workspace`` for a
``--continue``/``--session`` resume and exports it as ``PRAISONAI_WORKSPACE``
so the tool loader (and everything downstream) operates in that directory.
No-op when not resuming.
``session`` may be passed in when the caller has already resolved the exact
``UnifiedSession`` that will drive the conversation, so the binder and the
conversation are guaranteed to agree on the same session (avoids a second,
independent last-session lookup that could race a concurrent CLI process).
When not provided, it is resolved here as a best-effort fallback.
If the recorded directory is gone, warn and pin ``PRAISONAI_WORKSPACE`` to
cwd. Pinning (rather than leaving it unset) is important: an unset canonical
var would let the legacy ``PRAISON_WORKSPACE`` leak through in the tool
loader, contradicting the "using current directory" warning.
"""
resume_session_id = getattr(args, 'resume_session', None) if args is not None else None
if not resume_session_id:
return
if session is None:
try:
try:
from .session import get_session_store
except ImportError:
from praisonai.cli.session import get_session_store
store = get_session_store()
if resume_session_id == 'last':
session = store.get_last_session()
else:
session = store.get_or_create(resume_session_id)
except Exception:
return
workspace = getattr(session, 'workspace', None) if session else None
if not workspace:
return
if not os.path.isdir(workspace):
if console is not None:
try:
console.print(
f"[yellow]⚠ Session workspace no longer exists: {workspace} — "
f"using current directory instead.[/yellow]"
)
except Exception:
pass
# Pin cwd so a stray legacy PRAISON_WORKSPACE cannot override the
# advertised current-directory fallback in the tool loader.
os.environ["PRAISONAI_WORKSPACE"] = os.getcwd()
return
os.environ["PRAISONAI_WORKSPACE"] = workspace
def _load_interactive_tools(self):
"""
Load tools for interactive mode using the canonical provider.
@@ -665,25 +596,15 @@ def _load_interactive_tools(self):
if getattr(self.args, 'no_lsp', False):
disable_groups.append('lsp')
# Resolve the workspace root the agent's tools operate on.
# Order: explicit PRAISONAI_WORKSPACE (set by `code --workspace` or by
# resume-into-session-directory) → legacy PRAISON_WORKSPACE → cwd. Previously
# this hardcoded os.getcwd() and then overwrote config.workspace with it,
# which silently discarded the --workspace flag (the flag was dead).
workspace = (
os.environ.get("PRAISONAI_WORKSPACE")
or os.environ.get("PRAISON_WORKSPACE")
or os.getcwd()
)
# Get workspace
workspace = os.getcwd()
try:
from ..features.interactive_tools import get_interactive_tools, ToolConfig
# Create config. from_env() already honours the workspace env vars; only
# override when we resolved a concrete workspace so the two paths agree.
# Create config
config = ToolConfig.from_env()
if workspace:
config.workspace = workspace
config.workspace = workspace
# Apply CLI overrides
if 'acp' in disable_groups:
@@ -138,32 +138,17 @@ class FrameworkAdapterRegistry(PluginRegistry[FrameworkAdapter]):
_REQUIRED_KW = {"tools_dict", "agent_callback", "task_callback", "cli_config"}
def _accepts_required(fn) -> Optional[str]:
params = inspect.signature(fn).parameters.values()
# A **kwargs catch-all accepts every required keyword by definition,
# so entry-point plugins that forward **kwargs to a delegate (the
# advertised extension surface) validate instead of being silently
# dropped from pick_default()/list_available_frameworks().
if any(p.kind == inspect.Parameter.VAR_KEYWORD for p in params):
return None
named = {
p.name for p in params
if p.kind in (inspect.Parameter.KEYWORD_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD)
}
missing = _REQUIRED_KW - named
return f"missing keyword parameters {sorted(missing)}" if missing else None
for method_name in ("run", "arun"):
fn = getattr(cls, method_name, None)
if fn is None:
continue # arun is optional; sync-only adapters keep working
err = _accepts_required(fn)
if err:
raise TypeError(
f"FrameworkAdapter {name!r}.{method_name} does not implement "
f"the protocol: {err}"
)
sig = inspect.signature(cls.run)
kw_only = {
p.name for p in sig.parameters.values()
if p.kind in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
}
missing = _REQUIRED_KW - kw_only
if missing:
raise TypeError(
f"FrameworkAdapter {name!r} does not implement the protocol: "
f"missing keyword-only parameters {sorted(missing)}"
)
self._validated_classes.add(cls)
def create(self, name: str, *args, **kwargs):
@@ -34,9 +34,6 @@ from typing import AsyncIterator, Dict, Any, Optional, List
from .base import BaseCLIIntegration
_UNSET = object()
class CodexCLIIntegration(BaseCLIIntegration):
"""
Integration with OpenAI's Codex CLI.
@@ -103,34 +100,17 @@ class CodexCLIIntegration(BaseCLIIntegration):
"""Return the CLI command name."""
return "codex"
def _build_command(
self,
task: str,
*,
json_output: Optional[bool] = None,
output_schema: Optional[str] = None,
output_file: Any = _UNSET,
**options,
) -> List[str]:
def _build_command(self, task: str, **options) -> List[str]:
"""
Build the Codex CLI command.
Args:
task: The task to execute
json_output: JSON output override (defaults to self.json_output)
output_schema: Output schema override (defaults to self.output_schema)
output_file: Output file override. Omit to use self.output_file;
pass ``None`` explicitly to disable the instance default and
parse stdout instead.
**options: Additional options
Returns:
List of command arguments
"""
json_output = self.json_output if json_output is None else json_output
output_schema = self.output_schema if output_schema is None else output_schema
output_file = self.output_file if output_file is _UNSET else output_file
cmd = ["codex", "exec", "--skip-git-repo-check"]
# Add working directory
@@ -151,16 +131,16 @@ class CodexCLIIntegration(BaseCLIIntegration):
cmd.extend(["--sandbox", self.sandbox])
# Add JSON output flag if enabled
if json_output:
if self.json_output:
cmd.append("--json")
# Add output schema if specified
if output_schema:
cmd.extend(["--output-schema", output_schema])
if self.output_schema:
cmd.extend(["--output-schema", self.output_schema])
# Add output file if specified
if output_file:
cmd.extend(["-o", output_file])
if self.output_file:
cmd.extend(["-o", self.output_file])
# Add provider if specified
if self.provider:
@@ -233,18 +213,22 @@ class CodexCLIIntegration(BaseCLIIntegration):
Yields:
dict: Parsed JSON events from the CLI
"""
# Ensure JSON output is enabled for streaming (per-call override, no
# instance mutation); drop caller json_output so the forced value wins.
options.pop("json_output", None)
cmd = self._build_command(prompt, json_output=True, **options)
# Ensure JSON output is enabled for streaming
original_json = self.json_output
self.json_output = True
async for line in self.stream_async(cmd):
if line.strip():
try:
event = json.loads(line)
yield event
except json.JSONDecodeError:
yield {"type": "text", "content": line}
try:
cmd = self._build_command(prompt, **options)
async for line in self.stream_async(cmd):
if line.strip():
try:
event = json.loads(line)
yield event
except json.JSONDecodeError:
yield {"type": "text", "content": line}
finally:
self.json_output = original_json
async def execute_with_schema(
self,
@@ -263,26 +247,30 @@ class CodexCLIIntegration(BaseCLIIntegration):
Returns:
dict: Parsed structured output
"""
# Per-call overrides threaded through _build_command, no instance mutation.
# Pass output_path (or explicit None) so an unset path parses stdout
# rather than silently writing to the instance-default output file.
cmd = self._build_command(
prompt,
output_schema=schema_path,
output_file=output_path,
)
output = await self.execute_async(cmd)
original_schema = self.output_schema
original_output = self.output_file
# If output file was specified, read from it
if output_path and os.path.exists(output_path):
with open(output_path, 'r') as f:
return json.load(f)
self.output_schema = schema_path
if output_path:
self.output_file = output_path
# Otherwise parse the output
try:
return json.loads(output)
except json.JSONDecodeError:
return {"result": output}
cmd = self._build_command(prompt)
output = await self.execute_async(cmd)
# If output file was specified, read from it
if output_path and os.path.exists(output_path):
with open(output_path, 'r') as f:
return json.load(f)
# Otherwise parse the output
try:
return json.loads(output)
except json.JSONDecodeError:
return {"result": output}
finally:
self.output_schema = original_schema
self.output_file = original_output
def get_env(self) -> Dict[str, str]:
"""Get environment variables for CLI execution."""
@@ -81,29 +81,17 @@ class CursorCLIIntegration(BaseCLIIntegration):
"""Return the CLI command name."""
return "cursor-agent"
def _build_command(
self,
prompt: str,
*,
output_format: Optional[str] = None,
stream_partial: Optional[bool] = None,
**options,
) -> List[str]:
def _build_command(self, prompt: str, **options) -> List[str]:
"""
Build the Cursor CLI command.
Args:
prompt: The prompt to send
output_format: Output format override (defaults to self.output_format)
stream_partial: Stream-partial override (defaults to self.stream_partial)
**options: Additional options
Returns:
List of command arguments
"""
output_format = output_format or self.output_format
stream_partial = self.stream_partial if stream_partial is None else stream_partial
cmd = ["cursor-agent"]
# Add print mode flag
@@ -118,10 +106,10 @@ class CursorCLIIntegration(BaseCLIIntegration):
cmd.extend(["-m", self.model])
# Add output format
cmd.extend(["--output-format", output_format])
cmd.extend(["--output-format", self.output_format])
# Add stream partial flag if enabled
if stream_partial:
if self.stream_partial:
cmd.append("--stream-partial-output")
# Add resume session if specified
@@ -172,21 +160,26 @@ class CursorCLIIntegration(BaseCLIIntegration):
Yields:
dict: Parsed JSON events from the CLI
"""
# Use stream-json format for streaming (per-call override, no instance
# mutation); drop caller keys so the forced values win.
options.pop("output_format", None)
options.pop("stream_partial", None)
cmd = self._build_command(
prompt, output_format="stream-json", stream_partial=True, **options
)
# Use stream-json format for streaming
original_format = self.output_format
original_partial = self.stream_partial
async for line in self.stream_async(cmd):
if line.strip():
try:
event = json.loads(line)
yield event
except json.JSONDecodeError:
yield {"type": "text", "content": line}
self.output_format = "stream-json"
self.stream_partial = True
try:
cmd = self._build_command(prompt, **options)
async for line in self.stream_async(cmd):
if line.strip():
try:
event = json.loads(line)
yield event
except json.JSONDecodeError:
yield {"type": "text", "content": line}
finally:
self.output_format = original_format
self.stream_partial = original_partial
def get_env(self) -> Dict[str, str]:
"""Get environment variables for CLI execution."""
@@ -86,19 +86,12 @@ class GeminiCLIIntegration(BaseCLIIntegration):
"""Return the CLI command name."""
return "gemini"
def _build_command(
self,
prompt: str,
*,
output_format: Optional[str] = None,
**options,
) -> List[str]:
def _build_command(self, prompt: str, **options) -> List[str]:
"""
Build the Gemini CLI command.
Args:
prompt: The prompt to send
output_format: Output format override (defaults to self.output_format)
**options: Additional options
Returns:
@@ -110,7 +103,7 @@ class GeminiCLIIntegration(BaseCLIIntegration):
cmd.extend(["-m", self.model])
# Add output format
cmd.extend(["--output-format", output_format or self.output_format])
cmd.extend(["--output-format", self.output_format])
# Add include directories if specified
if self.include_directories:
@@ -167,20 +160,23 @@ class GeminiCLIIntegration(BaseCLIIntegration):
Returns:
Tuple[str, dict]: (result, stats) where stats contains usage information
"""
# Ensure JSON format for stats (per-call override, no instance mutation).
# Drop any caller-supplied output_format so the forced value wins without
# raising a duplicate-keyword TypeError.
options.pop("output_format", None)
cmd = self._build_command(prompt, output_format="json", **options)
output = await self.execute_async(cmd)
# Ensure JSON format for stats
original_format = self.output_format
self.output_format = "json"
try:
data = json.loads(output)
response = data.get("response", str(data))
stats = data.get("stats")
return response, stats
except json.JSONDecodeError:
return output, None
cmd = self._build_command(prompt, **options)
output = await self.execute_async(cmd)
try:
data = json.loads(output)
response = data.get("response", str(data))
stats = data.get("stats")
return response, stats
except json.JSONDecodeError:
return output, None
finally:
self.output_format = original_format
async def stream(self, prompt: str, **options) -> AsyncIterator[Dict[str, Any]]:
"""
@@ -193,18 +189,22 @@ class GeminiCLIIntegration(BaseCLIIntegration):
Yields:
dict: Parsed JSON events from the CLI
"""
# Use stream-json format for streaming (per-call override, no instance
# mutation); drop caller output_format so the forced value wins.
options.pop("output_format", None)
cmd = self._build_command(prompt, output_format="stream-json", **options)
# Use stream-json format for streaming
original_format = self.output_format
self.output_format = "stream-json"
async for line in self.stream_async(cmd):
if line.strip():
try:
event = json.loads(line)
yield event
except json.JSONDecodeError:
yield {"type": "text", "content": line}
try:
cmd = self._build_command(prompt, **options)
async for line in self.stream_async(cmd):
if line.strip():
try:
event = json.loads(line)
yield event
except json.JSONDecodeError:
yield {"type": "text", "content": line}
finally:
self.output_format = original_format
def get_last_stats(self) -> Optional[Dict[str, Any]]:
"""
+11 -25
View File
@@ -6,24 +6,12 @@ import warnings
import gc
from unittest.mock import Mock, patch
# NOTE: pytest 9.1.1+ forbids declaring `pytest_plugins` in a non-top-level
# conftest.py. We register the local safety plugins here instead, which is the
# supported mechanism for nested conftest files. This preserves the test gating
# and network guard behaviour without triggering the collection error.
def pytest_configure(config):
"""Register local safety plugins for the PraisonAI test suite.
pytest-asyncio is loaded automatically via its entry point. The gating and
network-guard plugins are registered explicitly (they previously lived in a
module-level `pytest_plugins` tuple which is no longer allowed here).
"""
for plugin_name in (
"tests._pytest_plugins.test_gating",
"tests._pytest_plugins.network_guard",
):
if not config.pluginmanager.has_plugin(plugin_name):
config.pluginmanager.import_plugin(plugin_name)
# Register pytest plugins
pytest_plugins = (
'pytest_asyncio',
'tests._pytest_plugins.test_gating',
'tests._pytest_plugins.network_guard',
)
# Suppress aiohttp unclosed session warnings during tests
warnings.filterwarnings("ignore", message="Unclosed client session")
@@ -56,16 +44,14 @@ def cleanup_async_resources():
# Force garbage collection to clean up any lingering async resources
gc.collect()
# Add the source paths to sys.path for imports.
# Layout: <root>/src/praisonai/tests/conftest.py, with the core SDK living at
# <root>/src/praisonai-agents. From this file, '..', '..' reaches <root>/src.
# Add the source paths to sys.path for imports
# praisonai-agents package (core SDK)
_agents_path = os.path.join(os.path.dirname(__file__), '..', '..', 'praisonai-agents')
if os.path.isdir(_agents_path) and _agents_path not in sys.path:
if _agents_path not in sys.path:
sys.path.insert(0, _agents_path)
# praisonai wrapper package: '..' reaches <root>/src/praisonai.
# praisonai wrapper package
_wrapper_path = os.path.join(os.path.dirname(__file__), '..')
if os.path.isdir(_wrapper_path) and _wrapper_path not in sys.path:
if _wrapper_path not in sys.path:
sys.path.insert(0, _wrapper_path)
@pytest.fixture
@@ -115,33 +115,13 @@ class TestToolConfig:
assert config.enable_basic is True
def test_config_from_env_workspace(self):
"""Test config from env with workspace override (legacy var)."""
"""Test config from env with workspace override."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(os.environ, {"PRAISON_WORKSPACE": "/custom/path"}):
config = ToolConfig.from_env()
assert config.workspace == "/custom/path"
def test_config_from_env_workspace_canonical(self):
"""PRAISONAI_WORKSPACE (written by `code --workspace`) is honoured."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(os.environ, {"PRAISONAI_WORKSPACE": "/canonical/path"}, clear=False):
os.environ.pop("PRAISON_WORKSPACE", None)
config = ToolConfig.from_env()
assert config.workspace == "/canonical/path"
def test_config_from_env_workspace_prefers_canonical(self):
"""When both env names are set, the canonical PRAISONAI_ wins."""
from praisonai.cli.features.interactive_tools import ToolConfig
with patch.dict(
os.environ,
{"PRAISONAI_WORKSPACE": "/canonical", "PRAISON_WORKSPACE": "/legacy"},
):
config = ToolConfig.from_env()
assert config.workspace == "/canonical"
class TestResolveToolGroups:
"""Tests for resolve_tool_groups function."""
@@ -1,171 +0,0 @@
"""
Regression tests for `praisonai code` workspace rooting and session binding.
Covers:
- The `--workspace` flag actually re-roots the interactive tool loader
(previously dead: env-name mismatch + unconditional os.getcwd() override).
- `--continue`/`--session` resume re-binds the tools to the directory the
session was created in (persisted UnifiedSession.workspace), with a graceful
cwd fallback when that directory no longer exists.
"""
import argparse
import os
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
class TestWorkspaceFlagRootsTools:
"""Bug 1: `--workspace <dir>` must re-root the tool loader."""
def test_load_interactive_tools_honours_workspace_env(self, tmp_path):
from praisonai.cli.legacy import interactive_legacy
captured = {}
def _fake_get_tools(config=None, disable=None):
captured["workspace"] = config.workspace
return []
target = tmp_path / "ws"
target.mkdir()
self_stub = SimpleNamespace(args=argparse.Namespace(no_acp=False, no_lsp=False))
with patch.dict(os.environ, {"PRAISONAI_WORKSPACE": str(target)}, clear=False):
os.environ.pop("PRAISON_WORKSPACE", None)
with patch(
"praisonai.cli.features.interactive_tools.get_interactive_tools",
_fake_get_tools,
):
interactive_legacy._load_interactive_tools(self_stub)
assert captured["workspace"] == str(target)
def test_load_interactive_tools_defaults_to_cwd(self):
from praisonai.cli.legacy import interactive_legacy
captured = {}
def _fake_get_tools(config=None, disable=None):
captured["workspace"] = config.workspace
return []
self_stub = SimpleNamespace(args=argparse.Namespace(no_acp=False, no_lsp=False))
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("PRAISONAI_WORKSPACE", None)
os.environ.pop("PRAISON_WORKSPACE", None)
with patch(
"praisonai.cli.features.interactive_tools.get_interactive_tools",
_fake_get_tools,
):
interactive_legacy._load_interactive_tools(self_stub)
assert captured["workspace"] == os.getcwd()
class TestSessionResumeRebindsDirectory:
"""Session↔directory binding on --continue/--session resume."""
def _store_with(self, workspace):
session = SimpleNamespace(workspace=workspace)
store = MagicMock()
store.get_last_session.return_value = session
store.get_or_create.return_value = session
return store
def test_resume_binds_existing_directory(self, tmp_path):
from praisonai.cli.legacy import interactive_legacy
ws = tmp_path / "session_dir"
ws.mkdir()
store = self._store_with(str(ws))
args = argparse.Namespace(resume_session="last")
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("PRAISONAI_WORKSPACE", None)
with patch(
"praisonai.cli.session.get_session_store",
return_value=store,
):
interactive_legacy._bind_resume_workspace(args, console=None)
assert os.environ.get("PRAISONAI_WORKSPACE") == str(ws)
def test_resume_missing_directory_falls_back_to_cwd(self, tmp_path):
from praisonai.cli.legacy import interactive_legacy
missing = tmp_path / "gone"
store = self._store_with(str(missing))
args = argparse.Namespace(resume_session="abc123")
console = MagicMock()
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("PRAISONAI_WORKSPACE", None)
with patch(
"praisonai.cli.session.get_session_store",
return_value=store,
):
interactive_legacy._bind_resume_workspace(args, console=console)
# Canonical var is pinned to cwd so a stray legacy PRAISON_WORKSPACE
# cannot override the advertised current-directory fallback.
assert os.environ.get("PRAISONAI_WORKSPACE") == os.getcwd()
console.print.assert_called_once()
def test_resume_missing_directory_pins_cwd_over_legacy_var(self, tmp_path):
"""P1: missing saved dir + legacy PRAISON_WORKSPACE set must still
resolve to cwd in the tool loader, not the legacy directory."""
from praisonai.cli.legacy import interactive_legacy
missing = tmp_path / "gone"
legacy = tmp_path / "legacy_ws"
legacy.mkdir()
store = self._store_with(str(missing))
args = argparse.Namespace(resume_session="abc123")
with patch.dict(
os.environ, {"PRAISON_WORKSPACE": str(legacy)}, clear=False
):
os.environ.pop("PRAISONAI_WORKSPACE", None)
with patch(
"praisonai.cli.session.get_session_store",
return_value=store,
):
interactive_legacy._bind_resume_workspace(args, console=None)
assert os.environ.get("PRAISONAI_WORKSPACE") == os.getcwd()
def test_resume_uses_provided_session_without_relookup(self, tmp_path):
"""P1: when the caller passes the already-resolved session, the binder
must not perform a second store lookup (avoids racing a concurrent CLI)."""
from praisonai.cli.legacy import interactive_legacy
ws = tmp_path / "provided"
ws.mkdir()
session = SimpleNamespace(workspace=str(ws))
args = argparse.Namespace(resume_session="last")
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("PRAISONAI_WORKSPACE", None)
with patch(
"praisonai.cli.session.get_session_store"
) as mock_store_getter:
interactive_legacy._bind_resume_workspace(
args, console=None, session=session
)
mock_store_getter.assert_not_called()
assert os.environ.get("PRAISONAI_WORKSPACE") == str(ws)
def test_no_resume_is_noop(self):
from praisonai.cli.legacy import interactive_legacy
args = argparse.Namespace(resume_session=None)
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("PRAISONAI_WORKSPACE", None)
interactive_legacy._bind_resume_workspace(args, console=None)
assert "PRAISONAI_WORKSPACE" not in os.environ
if __name__ == "__main__":
pytest.main([__file__, "-v"])