Compare commits

..
Author SHA1 Message Date
praisonai-triage-agent[bot]andMervin Praison b443953ca9 test: verify mixed-scope doc priority survives reload and reopen
Co-authored-by: Mervin Praison <[email protected]>
2026-10-07 09:52:42 +00:00
dajiaohuang 044da7cf02 test: document priority regression contracts 2026-10-03 22:48:43 +08:00
dajiaohuang f55067be38 test: exercise document priority with real agent responses 2026-10-03 22:17:51 +08:00
dajiaohuang 6fd8684b69 refactor: share document scope priority adjustment 2026-10-03 20:44:05 +08:00
Wu Shuwen 874d580160 Preserve documentation priority across creation and reload 2026-10-03 08:40:19 +01:00
701bfba56d fix(ci): tolerate 403 clearing conflict-pending on fork PRs (#5638)
* fix(ci): tolerate 403 when clearing conflict-pending label on fork PRs

Fork pull requests can deny label removal; swallow 403/404 so detect-and-trigger does not fail the required check.

Co-authored-by: Cursor <[email protected]>

* fix(ci): only tolerate 403/404 clearing conflict label on fork PRs

Narrow the swallow to genuine forks (headRepo != baseRepo) so upstream
permission failures surface instead of silently leaving the merge-blocking
claude-conflict-pending label in place.

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: Cursor <[email protected]>
Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Mervin Praison <[email protected]>
2026-10-02 18:38:40 +01:00
b981fcf54a fix: default gateway agent model to detected provider, not OpenAI (#5611)
* fix: default gateway agent model to detected provider, not OpenAI (fixes #5609)

The gateway agent config schema defaulted `model` to the hardcoded
`gpt-4o-mini`, and onboarding wrote that literal into every generated
config, so an operator with only a non-OpenAI credential was handed a
config that could not work.

- AgentConfigSchema.model now defaults to None; new resolved_model()
  (and "auto") resolve to the detected provider's model via the existing
  default_model_for_available_provider() helper (reached through the lazy
  _code_bridge), with gpt-4o-mini only as last-resort fallback.
- Gateway server, bot CLI and the startup preflight route an unset/"auto"
  model through the same detection instead of assuming OpenAI.
- Onboarding YAML leads with the detected model, and the failure hint
  names the detected provider's API key env var instead of OPENAI_API_KEY.

An explicit model: (or --model) always wins; detection only fills the gap.

Co-authored-by: Mervin Praison <[email protected]>

* chore: regenerate gateway.schema.json after rebase onto #5416

Co-authored-by: Cursor <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: Mervin Praison <[email protected]>
Co-authored-by: MervinPraison <[email protected]>
Co-authored-by: Cursor <[email protected]>
2026-10-02 18:29:40 +01:00
praisonai-triage-agent[bot] 3983e45d08 Merge pull request #5417 from MervinPraison/claude/issue-5415-20261001-0914
fix: wire progressive streaming replies for Discord & Slack
2026-10-02 17:29:26 +00:00
praisonai-triage-agent[bot] 88e858adb9 Release v4.7.12 2026-10-02 16:34:56 +00:00
MervinPraison f9ed8ff862 docs: auto-update documentation parity trackers [skip ci] 2026-10-02 16:33:53 +00:00
Wu Shuwen dadac293e3 fix: surface Responses failures before accepting partial output (#5477)
* fix: surface Responses failures before accepting partial output

* fix: close Responses streams on consumer exit

* fix: surface native failed Responses without fallback replay

* test: defer optional HTTPX import until stream test execution

* fix: preserve dictionary Responses output in failure controls

* fix: surface failed native Responses during streaming

* Preserve token usage from dictionary Responses
2026-10-02 17:33:11 +01:00
praisonai-triage-agent[bot] 1ead4131e8 Merge pull request #5626 from MervinPraison/claude/issue-5617-20261002-1135
fix: add CI perf-gate flags to agent benchmark (#5617)
2026-10-02 16:24:59 +00:00
praisonai-triage-agent[bot] 2432197239 Merge pull request #5636 from MervinPraison/claude/issue-5632-20261002-1333
fix: capture re-entrancy sentinel before guarded import in YAML runner
2026-10-02 16:17:09 +00:00
praisonai-triage-agent[bot] 82574988f8 Merge pull request #5630 from MervinPraison/claude/issue-5624-20261002-1138
perf(agents): skip plugin-manager scan in Agent init when no plugins registered
2026-10-02 16:16:55 +00:00
MervinPraison 698aef54b8 docs: auto-update documentation parity trackers [skip ci] 2026-10-02 15:59:46 +00:00
praisonai-triage-agent[bot]andMervin Praison 96e76b535a fix: publish gateway.yaml/bot.yaml JSON Schema for editor + CI validation (#5416)
* fix: publish gateway.yaml/bot.yaml JSON Schema for editor + CI validation (fixes #5414)

Bring the gateway/bot config surface to parity with agents.yaml:
- Add gateway_config_json_schema() deriving a draft-07 schema from the
  canonical Pydantic GatewayConfigSchema (single source of truth).
- Add GATEWAY_SCHEMA_URL / GATEWAY_SCHEMA_HEADER mirroring AGENTS_SCHEMA_*.
- Add `praisonai gateway schema [-o FILE]` CLI command.
- Scaffold the yaml-language-server header into onboarded configs and the
  shipped gateway.yaml sample.
- Publish gateway.schema.json artefact and include it as package-data.

Co-authored-by: MervinPraison <[email protected]>

* fix: align published gateway schema with runtime validation (#5414)

Post-process the Pydantic-derived gateway/bot JSON Schema so the published
artefact faithfully reflects what the gateway actually validates, closing the
author-time/runtime gaps flagged in review:

- Inline GatewayServerSchema/HookSchema (extra="forbid") into gateway/hooks so
  a misspelled server knob (drain_timout) or hook key is caught while authoring,
  not only at startup.
- owner_user_id (top-level + channel) accepts an integer, mirroring the runtime
  str() coercion of unquoted numeric Telegram/Discord ids.
- Reject unknown top-level keys so a typo (auto_enable_from_en) the runtime
  silently ignores is surfaced; config_version (stamped) is allowed and channels
  keep additionalProperties open for plugin-declared keys.
- Allow array env-ref channel fields (allowed_users/allowlist/blocklist) to be a
  single ${VAR} string so scaffolded configs validate pre-substitution.

Regenerate committed gateway.schema.json. Shipped sample + onboard scaffolds now
validate clean; all three findings fixed.

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: MervinPraison <[email protected]>
2026-10-02 16:56:14 +01:00
praisonai-triage-agent[bot]andMervin Praison fdbbcbf52d fix: opt-in per-reply runtime footer for gateway bots (#5437)
* fix: add opt-in per-reply runtime footer for gateway bots (fixes #5428)

Surface a privacy-safe footer (model · latency) on the final reply only,
reusing the already-collected model name and measuring turn latency at the
bot turn site. Off by default via the existing BotConfig.footer knob.

Co-authored-by: MervinPraison <[email protected]>

* fix: exclude runtime footer from voice/TTS and cover slash-command replies

Address reviewer findings on the opt-in gateway-bot footer:

- Voice replies no longer narrate the footer. The footer is now appended to
  the delivered TEXT only (inside _send_response_with_media via a footer_line
  param), never to the pre-footer text handed to TTS/media synthesis. This also
  makes the streaming and non-streaming voice paths behave identically.
- File-based / unknown slash-command agent replies now include the footer,
  routed through the same delivery call with a per-turn latency start.
- Add handler-level delivery tests (opt-in gating, model/latency capture, and
  footer-in-text-not-voice) alongside the existing formatting unit tests.

Scope intentionally kept minimal per AGENTS.md: gateway channel-config wiring
and non-Telegram channel adapters remain out of scope (the PR is a deliberate
Telegram-only opt-in; the shared BotConfig.footer flag is a no-op on channels
that have not wired it, matching other per-channel BotConfig knobs).

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: MervinPraison <[email protected]>
2026-10-02 16:52:55 +01:00
praisonai-triage-agent[bot]andMervin Praison 2c91e0cc4f fix: add --json to gateway status and report real version (#5368)
* fix: add --json to gateway status and report real version (fixes #5363)

- Add --json to `gateway status`, emitting a single machine-readable
  snapshot (version, protocol_version, uptime, sessions, clients,
  per-channel health, degraded owners, delivery back-pressure, config
  revision + drift) built from /health + /info + daemon status.
- Replace the hardcoded "1.0.0" version on GET /info and MCP serverInfo
  with the installed praisonai-bot version via a single helper.
- Add protocol_version (GATEWAY_PROTOCOL_VERSION) to /info.

Co-authored-by: MervinPraison <[email protected]>

* fix: load persisted gateway token for status --json /info + guard version regressions

- Resolve GATEWAY_AUTH_TOKEN from ~/.praisonai/.env when not exported so an
  authenticated gateway's /info answers and status --json reports version
  (greptile P1). Surface version_unavailable="info_unreachable" when /health
  is reachable but /info is not, so a missing version is explicit not silent.
- Strengthen MCP initialize test to assert serverInfo.version == installed
  version and != "1.0.0" (greptile P2), guarding the endpoint from regressing
  back to a hardcoded value.
- Add token-resolution + version-unavailable snapshot tests.

Co-authored-by: Mervin Praison <[email protected]>

---------

Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: MervinPraison <[email protected]>
2026-10-02 16:49:42 +01:00
MervinPraison 27206ed5bd docs: auto-update documentation parity trackers [skip ci] 2026-10-02 15:40:59 +00:00
Wu Shuwen 1fca8e2706 fix(memory): enumerate SQLite records and preserve MCP user filters (#5479)
* fix: enumerate all SQLite memory records without a search cap

* fix: filter SQLite memory users before display limits

* fix: use compatible metadata parsing for SQLite user filters

* fix: keep null memory owners distinct from literal user IDs

* fix(memory): filter explicitly supplied falsey user identifiers

* fix(mcp): preserve explicit memory user filters
2026-10-02 16:23:10 +01:00
MervinPraison ccd5849008 docs: auto-update documentation parity trackers [skip ci] 2026-10-02 14:02:15 +00:00
praisonai-triage-agent[bot]andMervin Praison 6cf57513c9 ci: re-trigger unit suite after flaky native segfault (no code change)
The failed run crashed with exit code 139 (Fatal Python error:
Segmentation fault) inside an unrelated tool-retry test
(tests/unit/tools/test_async_tool_retry_parity.py), with background
threads from posthog/asyncio/heartbeat and C-extensions on the stack.
The test passes in isolation and the crash point does not touch this
PR's plugin-manager change (has_registered_plugins + merge gates), which
is a pure, allocation-free dict read. Empty commit to re-run the full
suite on identical code.

Co-authored-by: Mervin Praison <[email protected]>
2026-10-02 13:53:58 +00:00
Wu Shuwen 8641da4633 fix: retain incomplete Responses outcomes and streamed usage (#5475)
* fix: retain incomplete Responses outcomes and streamed usage

* fix: preserve Responses usage mappings and verify real metrics
2026-10-02 14:53:11 +01:00
praisonai-triage-agent[bot] f7b2ff1be0 Merge pull request #5438 from MervinPraison/claude/issue-5433-20261001-0955
fix: implement live LangSmith trace delivery (PAI-TS-002)
2026-10-02 13:40:37 +00:00
praisonai-triage-agent[bot]anddajiaohuang ebca47a8ff fix: capture re-entrancy sentinel before guarded import in YAML runner (fixes #5632)
The _run_from_file YAML runner assigned _prev_in_run inside the try block,
right after the `from praisonai_code.cli.main import PraisonAI` import. When
that import raised, the broad except re-raised typer.Exit(1) and the finally
block then read the never-assigned _prev_in_run, raising UnboundLocalError and
masking the original diagnostic. Capture the sentinel's prior value before the
guarded block so the finally restore is always safe.

Co-authored-by: dajiaohuang <[email protected]>
2026-10-02 13:35:53 +00:00
Mervin PraisonandCursor 4d604e7716 ci: extend optimised smoke job timeout to 16 minutes (#5634)
Smoke runs were being cancelled at 12 minutes under load; 16 minutes matches headroom used elsewhere without widening test scope.

Co-authored-by: Cursor <[email protected]>
2026-10-02 14:22:07 +01:00
praisonai-triage-agent[bot]anddajiaohuang ce458b2ba5 fix: forward overwrite flag to indexed imported-session save (fixes #5627) (#5631)
Co-authored-by: praisonai-triage-agent[bot] <272766704+praisonai-triage-agent[bot]@users.noreply.github.com>
Co-authored-by: dajiaohuang <[email protected]>
2026-10-02 14:19:42 +01:00
praisonai-triage-agent[bot]andMervin Praison 3e048ab457 fix(session): forward overwrite kwarg in SqliteSessionStore import override
The base DefaultSessionStore._save_imported_session now takes an
overwrite keyword (import no-clobber work) and import_sessions calls it
with overwrite=overwrite, but the SqliteSessionStore override and a test
monkeypatch still used the old positional-only signature, raising
"unexpected keyword argument 'overwrite'" and breaking portable imports.

- sqlite_store.py: match base signature and forward overwrite to super()
- test_sqlite_import_index.py: update delayed_save mock to accept/forward
  overwrite so the concurrent-write test exercises the real code path

Co-authored-by: Mervin Praison <[email protected]>
2026-10-02 12:46:59 +00:00
praisonai-triage-agent[bot] 0a7195dac0 Merge remote-tracking branch 'origin/main' into claude/issue-5617-20261002-1135 2026-10-02 12:44:44 +00:00
Wu Shuwen 8dd23bf1c1 fix: isolate in-memory metadata and read results from caller edits (#5483)
* fix: isolate in-memory metadata and read results from caller edits

* fix: copy memory enumeration outside the write lock
2026-10-02 13:44:06 +01:00
Wu Shuwen 7fb721f3f3 fix(session): distinguish current and completed handoff progress (#5539) 2026-10-02 13:43:55 +01:00
praisonai-triage-agent[bot]andMervin Praison 92add1ef85 test(plugins): cover uncreated-singleton fast path in has_registered_plugins
Adds an isolated regression test that patches the module-global
_default_manager to None and asserts has_registered_plugins() returns
False without constructing the singleton. Closes the greptile P2 gap;
protects the allocation-free hot-path optimisation from regression.

Co-authored-by: Mervin Praison <[email protected]>
2026-10-02 12:41:30 +00:00
praisonai-triage-agent[bot]andMervin Praison 151d74096c fix: keep --json output parseable with --save and honor thresholds iterations/warmup
- Route save_results/update_readme status messages through an injectable
  logger; under --json they go to stderr so stdout stays pure JSON when
  --json and --save are combined (greptile P1).
- Make thresholds.json iterations/warmup authoritative for both the
  measurement and the emitted JSON metadata so configuration is no longer
  silently ignored (greptile P2).

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Mervin Praison <[email protected]>
2026-10-02 12:13:57 +00:00
praisonai-triage-agent[bot]andDhivya-Bharathy 42c410d326 perf(agents): skip plugin-manager scan in Agent init when no plugins registered (fixes #5624)
Add a cheap, allocation-free has_registered_plugins() fast-path so the two
unconditional plugin merge calls in Agent.__init__ (_merge_plugin_tools and
_merge_plugin_subsystems) early-return when no plugins are registered — the
common case. Removes manager construction, lock acquisition and list scans
from the hot init path.

Tool-schema building was already lazy (first run()/chat, cached), so this
targets the remaining per-init overhead profiling surfaced. Benchmark (500
iters, warmed): min ~121us -> ~100us, mean ~135us -> ~111us; function calls
170k -> 134k. No behavioral change: plugins still merge when registered.

Co-authored-by: Dhivya-Bharathy <[email protected]>
2026-10-02 11:46:35 +00:00
praisonai-triage-agent[bot]andDhivya-Bharathy e9a5538a9c fix: add CI perf-gate flags to agent benchmark (fixes #5617)
Add --skip-crewai, --json, and --fail-over-threshold flags to
benchmarks/simple_benchmark.py plus a thresholds.json budget file so CI
can baseline and gate agent instantiation performance regressions.

- --skip-crewai avoids the optional-framework import hang
- --json emits machine-readable results for baseline artifacts
- --fail-over-threshold exits non-zero when PraisonAI init exceeds budget

Co-authored-by: Dhivya-Bharathy <[email protected]>
2026-10-02 11:37:21 +00:00
praisonai-triage-agent[bot]andMervin Praison 07ac1d0547 fix(observability): harden LangSmith adapter per review (#5438)
- Surface failed run deliveries in flush() instead of resolving silently
- Reject plaintext http endpoints to non-loopback hosts (API key leak)
- Export span events into the run payload (extra.events)
- Evict completed runs/ordering state to bound memory in long processes
- Recognise legacy LANGCHAIN_API_KEY in CLI readiness via altEnvKeys
- Tests save/restore credential env; add coverage for all of the above

Co-authored-by: Mervin Praison <[email protected]>
2026-10-01 10:11:39 +00:00
praisonai-triage-agent[bot]andDhivya-Bharathy c3f2676958 fix: implement live LangSmith trace delivery (fixes #5433)
Upgrade the LangSmith observability adapter from an honest stub to a real
deliverer using the existing axios dependency (no new deps). isEnabled is
derived from a live, credential-validated HTTP client, spans map to
LangSmith runs with tool/LLM child runs under the agent root, flush is
batched async, and secrets are read from env only with http(s)-only
endpoint validation. All other external vendors remain honest stubs.

Co-authored-by: Dhivya-Bharathy <[email protected]>
2026-10-01 10:00:14 +00:00
praisonai-triage-agent[bot]andMervin Praison c2b84b84c0 fix: harden Discord/Slack streaming delivery guardrails (#5417)
Address Greptile review findings on the progressive-streaming wiring:
- Slack: defer to the thread-aware send path when a reply must be threaded
  (the root-only DraftStreamer would otherwise strand it in the channel),
  and extract MEDIA: directives so streamed audio is uploaded via Slack's
  file path instead of being edited in as literal text.
- Discord: when a streamed answer exceeds the 2000-char cap, delete the
  placeholder and fall back to the chunked _send_long_message path so long
  replies are split and keep their reply reference.
- Tests: add regression coverage for thread deferral, Slack media upload,
  and the Discord long-reply chunked fallback.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-authored-by: Mervin Praison <[email protected]>
2026-10-01 09:34:36 +00:00
praisonai-triage-agent[bot]andMervinPraison 9bec7d3227 fix: wire progressive streaming replies for Discord & Slack (fixes #5415)
The channel-agnostic DraftStreamer was only instantiated by Telegram, so a
configured streaming.mode: draft|auto was silently inert on Discord and Slack
despite both supporting in-place edits. Reuse the existing engine in their
reply paths via a shared build_streaming_config factory.

Co-authored-by: MervinPraison <[email protected]>
2026-10-01 09:20:28 +00:00
84 changed files with 5793 additions and 623 deletions

No files matched your search

+19 -7
View File
@@ -117,13 +117,25 @@ jobs:
const labels = issue.labels.map((l) => l.name);
if (status !== 'DIRTY' && labels.includes(LABEL)) {
await github.rest.issues.removeLabel({
owner,
repo,
issue_number: prNumber,
name: LABEL,
});
core.info(`Removed ${LABEL} from PR #${prNumber}`);
try {
await github.rest.issues.removeLabel({
owner,
repo,
issue_number: prNumber,
name: LABEL,
});
core.info(`Removed ${LABEL} from PR #${prNumber}`);
} catch (err) {
// Fork PRs may return 403/404 when the token cannot mutate labels on the head repo's issue.
// For upstream PRs a 403/404 is a genuine failure that must surface, otherwise the
// merge-blocking label stays in place while the check reports success.
const isFork = headRepo && headRepo !== baseRepo;
if (isFork && (err.status === 403 || err.status === 404)) {
core.info(`Skip remove ${LABEL} on fork PR #${prNumber}: ${err.status}`);
} else {
throw err;
}
}
}
if (status !== 'DIRTY') return { skipped: true, reason: 'not dirty' };
+1 -1
View File
@@ -42,7 +42,7 @@ jobs:
# ============================================
smoke:
runs-on: ubuntu-latest
timeout-minutes: 12
timeout-minutes: 16
steps:
- uses: actions/checkout@v4
with:
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.7.11" \
"praisonai>=4.7.12" \
"praisonai[chat]" \
"embedchain[github,youtube]"
+1 -1
View File
@@ -20,7 +20,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.7.11" \
"praisonai>=4.7.12" \
"praisonai[ui]" \
"praisonai[chat]" \
"praisonai[realtime]" \
+1 -1
View File
@@ -16,7 +16,7 @@ RUN mkdir -p /root/.praison
# Install Python packages (using latest versions)
RUN pip install --no-cache-dir \
praisonai_tools \
"praisonai>=4.7.11" \
"praisonai>=4.7.12" \
"praisonai[ui]" \
"praisonai[crewai]"
+5 -5
View File
@@ -21,13 +21,13 @@ This report compares **Python SDK feature categories** against **Python document
| ✅ Agent | 22 | 59 | 19076 |
| ✅ Agent-to-Agent (A2A) | 1 | 7 | 2996 |
| ✅ Agent-to-User (A2U) | 1 | 3 | 701 |
| ✅ Approval | 1 | 8 | 3969 |
| ✅ Approval | 1 | 8 | 4028 |
| ✅ Audio | 2 | 12 | 1000 |
| ✅ Auto Generation | 5 | 12 | 3941 |
| ✅ Autonomy | 3 | 7 | 2805 |
| ✅ Bots | 7 | 38 | 15809 |
| ✅ Budget | 1 | 1 | 287 |
| ✅ CLI | 5 | 124 | 39191 |
| ✅ CLI | 5 | 124 | 39199 |
| ✅ Chunking | 2 | 2 | 435 |
| ✅ Citations | 2 | 1 | 202 |
| ✅ Code Execution | 2 | 14 | 5022 |
@@ -43,7 +43,7 @@ This report compares **Python SDK feature categories** against **Python document
| ✅ Failover | 2 | 1 | 537 |
| ✅ Files | 2 | 7 | 2376 |
| ✅ Flow | 1 | 3 | 811 |
| ✅ Gateway | 7 | 97 | 37597 |
| ✅ Gateway | 7 | 97 | 37652 |
| ✅ Guardrails | 5 | 4 | 2642 |
| ✅ Handoffs | 11 | 6 | 2748 |
| ✅ Hooks | 2 | 9 | 5230 |
@@ -59,7 +59,7 @@ This report compares **Python SDK feature categories** against **Python document
| ✅ Output | 3 | 5 | 1289 |
| ✅ Parallel Execution | 3 | 2 | 528 |
| ✅ Planning | 6 | 7 | 1827 |
| ✅ Plugins | 8 | 5 | 4108 |
| ✅ Plugins | 8 | 5 | 4112 |
| ✅ Prompts | 2 | 9 | 2043 |
| ✅ Providers | 1 | 58 | 8751 |
| ✅ Query | 1 | 2 | 736 |
@@ -87,7 +87,7 @@ This report compares **Python SDK feature categories** against **Python document
These docs exist but don't match any implemented feature category:
- ℹ️ Database (44 docs, 8206 lines)
- ℹ️ Database (44 docs, 8292 lines)
- ℹ️ Documents (1 docs, 810 lines)
- ℹ️ Teams (1 docs, 214 lines)
@@ -6,8 +6,13 @@ Compares agent instantiation times across popular AI agent frameworks.
Usage:
python benchmarks/simple_benchmark.py
python benchmarks/simple_benchmark.py --skip-crewai --json
python benchmarks/simple_benchmark.py --json --fail-over-threshold
"""
import json
import os
import sys
import time
from typing import Literal
import argparse
@@ -16,6 +21,14 @@ import argparse
ITERATIONS = 100
WARMUP = 10
THRESHOLDS_PATH = os.path.join(os.path.dirname(__file__), 'thresholds.json')
def load_thresholds(path: str = THRESHOLDS_PATH) -> dict:
"""Load benchmark thresholds from thresholds.json."""
with open(path) as f:
return json.load(f)
def sample_tool(city: Literal['nyc', 'sf']):
"""Sample tool for benchmark testing."""
@@ -36,35 +49,44 @@ def measure_instantiation(create_fn, iterations=ITERATIONS, warmup=WARMUP):
return sum(times) / len(times)
def run_benchmark():
def run_benchmark(skip_crewai: bool = False, quiet: bool = False,
iterations: int = ITERATIONS, warmup: int = WARMUP):
"""Run the benchmark across all available frameworks (without tools)."""
results = {}
print('=' * 60)
print('PraisonAI Agents - Performance Benchmark')
print('=' * 60)
print(f'\nIterations: {ITERATIONS}')
print('Metric: Agent instantiation time (microseconds)\n')
def log(msg: str = ''):
if not quiet:
print(msg)
def measure(create_fn):
return measure_instantiation(create_fn, iterations=iterations, warmup=warmup)
log('=' * 60)
log('PraisonAI Agents - Performance Benchmark')
log('=' * 60)
log(f'\nIterations: {iterations}')
log(f'Warmup: {warmup}')
log('Metric: Agent instantiation time (microseconds)\n')
# PraisonAI (without tools)
print("Testing PraisonAI...")
log("Testing PraisonAI...")
from praisonaiagents import Agent as PraisonAgent
results['PraisonAI'] = measure_instantiation(
results['PraisonAI'] = measure(
lambda: PraisonAgent(name='Test', model='gpt-4o-mini', output="silent")
)
results['PraisonAI (LiteLLM)'] = measure_instantiation(
results['PraisonAI (LiteLLM)'] = measure(
lambda: PraisonAgent(name='Test', model='openai/gpt-4o-mini', output="silent")
)
# Other frameworks for comparison
print("Testing other frameworks...")
log("Testing other frameworks...")
try:
from agno.agent import Agent as AgnoAgent
from agno.models.openai import OpenAIChat
results['Agno'] = measure_instantiation(
results['Agno'] = measure(
lambda: AgnoAgent(model=OpenAIChat(id='gpt-4o-mini'))
)
except ImportError:
@@ -72,7 +94,7 @@ def run_benchmark():
try:
from pydantic_ai import Agent as PydanticAgent
results['PydanticAI'] = measure_instantiation(
results['PydanticAI'] = measure(
lambda: PydanticAgent('openai:gpt-4o-mini')
)
except ImportError:
@@ -80,7 +102,7 @@ def run_benchmark():
try:
from agents import Agent as OpenAIAgent
results['OpenAI Agents SDK'] = measure_instantiation(
results['OpenAI Agents SDK'] = measure(
lambda: OpenAIAgent(name='Test', model='gpt-4o-mini')
)
except ImportError:
@@ -96,47 +118,50 @@ def run_benchmark():
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['LangGraph'] = measure_instantiation(
results['LangGraph'] = measure(
lambda: create_react_agent(model=ChatOpenAI(model='gpt-4o-mini'), tools=[get_weather_lg])
)
except ImportError:
pass
try:
from crewai.agent import Agent as CrewAgent
from crewai.tools import tool as crewai_tool
@crewai_tool("Weather Tool")
def get_weather_crew(city: Literal['nyc', 'sf']):
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['CrewAI'] = measure_instantiation(
lambda: CrewAgent(
role='Weather Agent',
goal='Provide weather info',
backstory='A weather expert',
tools=[get_weather_crew],
verbose=False
if skip_crewai:
log("Skipping CrewAI (--skip-crewai): optional framework import can hang on some platforms.")
else:
try:
from crewai.agent import Agent as CrewAgent
from crewai.tools import tool as crewai_tool
@crewai_tool("Weather Tool")
def get_weather_crew(city: Literal['nyc', 'sf']):
"""Get weather info."""
return 'sunny' if city == 'sf' else 'cloudy'
results['CrewAI'] = measure(
lambda: CrewAgent(
role='Weather Agent',
goal='Provide weather info',
backstory='A weather expert',
tools=[get_weather_crew],
verbose=False
)
)
)
except ImportError:
pass
except ImportError:
pass
# Print results
print('\n' + '=' * 60)
print('RESULTS')
print('=' * 60)
log('\n' + '=' * 60)
log('RESULTS')
log('=' * 60)
baseline = results.get('PraisonAI', 1)
print(f"\n{'Framework':<25} {'Avg Time (μs)':<15} {'Relative':<10}")
print('-' * 50)
log(f"\n{'Framework':<25} {'Avg Time (μs)':<15} {'Relative':<10}")
log('-' * 50)
for name, avg in sorted(results.items(), key=lambda x: x[1]):
ratio = avg / baseline
print(f'{name:<25} {avg:<15.2f} {ratio:.2f}x')
log(f'{name:<25} {avg:<15.2f} {ratio:.2f}x')
print('\n' + '=' * 60)
log('\n' + '=' * 60)
return results
@@ -163,8 +188,13 @@ def get_package_versions():
return versions
def save_results(results: dict, filename: str = 'BENCHMARK_RESULTS.md'):
"""Save benchmark results to a markdown file and update README."""
def save_results(results: dict, filename: str = 'BENCHMARK_RESULTS.md', *,
iterations: int = ITERATIONS, log=print):
"""Save benchmark results to a markdown file and update README.
Status messages are emitted via ``log`` (defaults to ``print``). Pass a
stderr logger when stdout must stay machine-readable (e.g. ``--json``).
"""
import os
import re
from datetime import datetime
@@ -191,7 +221,7 @@ def save_results(results: dict, filename: str = 'BENCHMARK_RESULTS.md'):
with open(filepath, 'w') as f:
f.write('# PraisonAI Agents - Benchmark Results\n\n')
f.write(f'**Generated:** {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}\n')
f.write(f'**Iterations:** {ITERATIONS}\n')
f.write(f'**Iterations:** {iterations}\n')
f.write('**Test:** Agent instantiation (without tools)\n\n')
f.write('## Results\n\n')
f.write('| Framework | Avg Time (μs) | Relative |\n')
@@ -208,17 +238,17 @@ def save_results(results: dict, filename: str = 'BENCHMARK_RESULTS.md'):
f.write('python benchmarks/simple_benchmark.py\n')
f.write('```\n')
print(f'\nResults saved to: {filepath}')
log(f'\nResults saved to: {filepath}')
# Also update the main README.md
readme_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', 'README.md')
if os.path.exists(readme_path):
update_readme(readme_path, table_rows)
update_readme(readme_path, table_rows, log=log)
return filepath
def update_readme(readme_path: str, table_rows: list):
def update_readme(readme_path: str, table_rows: list, *, log=print):
"""Update the performance section in README.md with latest results."""
import re
@@ -239,17 +269,54 @@ def update_readme(readme_path: str, table_rows: list):
with open(readme_path, 'w') as f:
f.write(content)
print(f'README.md updated: {readme_path}')
log(f'README.md updated: {readme_path}')
def main():
parser = argparse.ArgumentParser(description='PraisonAI Agents - Performance Benchmark')
parser.add_argument('--save', action='store_true', help='Save results to file')
parser.add_argument('--skip-crewai', action='store_true',
help='Skip the optional CrewAI benchmark (its import can hang on some platforms)')
parser.add_argument('--json', action='store_true',
help='Emit results as JSON to stdout (for CI baselines/artifacts)')
parser.add_argument('--fail-over-threshold', action='store_true',
help='Exit non-zero if PraisonAI instantiation exceeds the budget in thresholds.json')
args = parser.parse_args()
thresholds = load_thresholds()
iterations = int(thresholds.get('iterations', ITERATIONS))
warmup = int(thresholds.get('warmup', WARMUP))
results = run_benchmark(
skip_crewai=args.skip_crewai,
quiet=args.json,
iterations=iterations,
warmup=warmup,
)
if args.save:
# Keep stdout machine-readable under --json by sending save status to stderr.
save_log = (lambda msg='': print(msg, file=sys.stderr)) if args.json else print
save_results(results, iterations=iterations, log=save_log)
elif not args.json:
print('\nResults not saved (use --save flag to save results to file)')
exit_code = 0
if args.fail_over_threshold:
max_us = thresholds['praisonai_instantiation_us_max']
actual_us = results.get('PraisonAI')
if actual_us is not None and actual_us > max_us:
print(
f"ERROR: PraisonAI instantiation {actual_us:.2f} us > threshold {max_us} us",
file=sys.stderr,
)
exit_code = 1
if args.json:
print(json.dumps({'results': results, 'iterations': iterations, 'warmup': warmup}))
return exit_code
if __name__ == '__main__':
parser = argparse.ArgumentParser(description='PraisonAI Agents - Performance Benchmark')
parser.add_argument('--save', action='store_true', help='Save results to file')
args = parser.parse_args()
results = run_benchmark()
if args.save:
save_results(results)
else:
print('\nResults not saved (use --save flag to save results to file)')
sys.exit(main())
@@ -0,0 +1,5 @@
{
"praisonai_instantiation_us_max": 450,
"iterations": 100,
"warmup": 10
}
@@ -547,6 +547,15 @@ class ToolExecutionMixin:
so disabling or unregistering a plugin revokes its tools from this
agent's discovery and execution paths.
"""
# Hot-path fast exit: when no plugins are registered (the common case)
# skip constructing/scanning the manager entirely on every Agent init.
try:
from ..plugins.manager import has_registered_plugins
if not has_registered_plugins():
return
except Exception: # pragma: no cover - defensive plugin boundary
pass
try:
from ..plugins import get_plugin_manager
@@ -673,6 +682,15 @@ class ToolExecutionMixin:
prevent the Agent from constructing. Zero overhead when no such plugin
is enabled (each collector returns an empty list).
"""
# Hot-path fast exit: when no plugins are registered (the common case)
# skip constructing/scanning the manager entirely on every Agent init.
try:
from ..plugins.manager import has_registered_plugins
if not has_registered_plugins():
return
except Exception: # pragma: no cover - defensive plugin boundary
pass
try:
from ..plugins import get_plugin_manager
@@ -125,6 +125,13 @@ class BotConfig:
# Edit interval for streaming responses in milliseconds (default: 700ms)
stream_edit_interval_ms: int = 700
# Opt-in privacy-safe runtime footer on the final reply (default: False).
# When True, a short trailing line (e.g. ``— gpt-4o · 4.2s``) reporting the
# model and turn latency is appended to the final reply only. Privacy-safe:
# never surfaces tool args, URLs, paths or message content. Unavailable
# fields are omitted rather than shown as placeholders.
footer: bool = False
# Intentional silence support
allow_silence: bool = False
@@ -185,6 +192,7 @@ class BotConfig:
"owner_user_id": "***" if self.owner_user_id else None,
"streaming": self.streaming,
"stream_edit_interval_ms": self.stream_edit_interval_ms,
"footer": self.footer,
"allow_silence": self.allow_silence,
"silence_token": self.silence_token,
"allow_bots": self.allow_bots,
+221 -150
View File
@@ -6465,6 +6465,9 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
finish_reason = getattr(choice, "finish_reason", None)
msg = getattr(choice, "message", None)
refusal = getattr(msg, "refusal", None) if msg is not None else None
if not choices:
from .openai_client import OpenAIClient
finish_reason = OpenAIClient._responses_incomplete_finish_reason(response)
if finish_reason is None and not refusal:
return
from ..agent.run_outcome import classify_finish_reason
@@ -7158,6 +7161,15 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
self._track_token_usage(response, self._response_model_for_tracking(response))
return response
def _raise_responses_failure(self, error) -> None:
"""Surface a failed Responses request without accepting partial output."""
code = error.get("code") if isinstance(error, dict) else getattr(error, "code", None)
message = error.get("message") if isinstance(error, dict) else getattr(error, "message", None)
if self._last_stop_reason == "completed":
self._last_stop_reason = "error"
label = f"Responses API failed ({code})" if code else "Responses API failed"
raise LLMResponseError(f"{label}: {message or 'Provider returned a failed response.'}")
def _extract_from_responses_output(self, response) -> tuple:
"""
Parse a ``ResponsesAPIResponse`` into the same (text, tool_calls,
@@ -7169,11 +7181,16 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
Unlike Chat Completions, text and tool calls are *always* separate
items, so ``content`` is never null when text is present.
"""
status = response.get("status") if isinstance(response, dict) else getattr(response, "status", None)
if status == "failed":
error = response.get("error") if isinstance(response, dict) else getattr(response, "error", None)
self._raise_responses_failure(error)
self._record_finish_reason(response)
response_text = ""
tool_calls: List[Dict[str, Any]] = []
reasoning_content = None
output_items = getattr(response, 'output', None) or []
output_items = (response.get('output') if isinstance(response, dict) else getattr(response, 'output', None)) or []
for item in output_items:
# Handle both dict and object access
if isinstance(item, dict):
@@ -7231,6 +7248,34 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
return response_text, tool_calls if tool_calls else None, reasoning_content
@staticmethod
def _close_responses_stream(stream) -> None:
"""Close a stream or its LiteLLM transport without masking its error."""
try:
close = getattr(stream, "close", None)
if not callable(close):
close = getattr(getattr(stream, "response", None), "close", None)
if callable(close):
close()
except Exception:
logging.debug("Responses stream cleanup failed", exc_info=True)
@staticmethod
async def _aclose_responses_stream(stream) -> None:
"""Await transport cleanup on completion, failure or cancellation."""
try:
target = stream
close = getattr(target, "aclose", None) or getattr(target, "close", None)
if not callable(close):
target = getattr(stream, "response", None)
close = getattr(target, "aclose", None) or getattr(target, "close", None)
if callable(close):
result = close()
if inspect.isawaitable(result):
await result
except Exception:
logging.debug("Responses stream cleanup failed", exc_info=True)
def _stream_responses_api(
self,
params: Dict[str, Any],
@@ -7262,95 +7307,108 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
if _emit:
from ..streaming.events import StreamEvent, StreamEventType
for event in litellm.responses(**params):
# Access event type — handle both object and dict
if isinstance(event, dict):
evt_type = event.get("type", "")
else:
evt_type = getattr(event, "type", "")
# ── Completed — capture the final response for usage accounting ──
if evt_type == "response.completed":
_final_response = (event.get("response") if isinstance(event, dict)
else getattr(event, "response", None))
# ── Text delta ──────────────────────────────────────────
if evt_type == "response.output_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
response_text += delta_text
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=False,
))
# ── Reasoning delta ─────────────────────────────────────
elif evt_type == "response.reasoning_summary_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=True,
))
# ── Function call arguments delta ───────────────────────
elif evt_type == "response.function_call_arguments.delta":
idx = (event.get("output_index", 0) if isinstance(event, dict)
else getattr(event, "output_index", 0))
delta_args = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if idx not in _pending_tools:
item_id = (event.get("item_id", f"tool_{idx}") if isinstance(event, dict)
else getattr(event, "item_id", f"tool_{idx}"))
_pending_tools[idx] = {
"id": item_id,
"name": "",
"arguments": "",
}
_pending_tools[idx]["arguments"] += delta_args
# ── Output item done — finalise tool call metadata ──────
elif evt_type == "response.output_item.done":
response_stream = litellm.responses(**params)
try:
for event in response_stream:
# Access event type — handle both object and dict
if isinstance(event, dict):
item = event.get("item", {})
idx = event.get("output_index", 0)
evt_type = event.get("type", "")
else:
item = getattr(event, "item", None) or {}
idx = getattr(event, "output_index", 0)
evt_type = getattr(event, "type", "")
if isinstance(item, dict):
i_type = item.get("type", "")
else:
i_type = getattr(item, "type", "")
if evt_type == "response.failed":
failed_response = event.get("response") if isinstance(event, dict) else getattr(event, "response", None)
error = failed_response.get("error") if isinstance(failed_response, dict) else getattr(failed_response, "error", None)
self._raise_responses_failure(error)
if evt_type == "error":
self._raise_responses_failure(event)
if i_type == "function_call":
if isinstance(item, dict):
call_id = item.get("call_id", item.get("id", f"tool_{idx}"))
fn_name = item.get("name", "")
fn_args = item.get("arguments", "")
else:
call_id = getattr(item, "call_id", None) or getattr(item, "id", f"tool_{idx}")
fn_name = getattr(item, "name", "")
fn_args = getattr(item, "arguments", "")
# ── Completed — capture the final response for usage accounting ──
if evt_type in ("response.completed", "response.incomplete"):
_final_response = (event.get("response") if isinstance(event, dict)
else getattr(event, "response", None))
self._record_finish_reason(_final_response)
if idx in _pending_tools:
_pending_tools[idx]["id"] = call_id
_pending_tools[idx]["name"] = fn_name
# Prefer the accumulated delta args if available, else use final
if not _pending_tools[idx]["arguments"]:
_pending_tools[idx]["arguments"] = fn_args
else:
# ── Text delta ──────────────────────────────────────────
if evt_type == "response.output_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
response_text += delta_text
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=False,
))
# ── Reasoning delta ─────────────────────────────────────
elif evt_type == "response.reasoning_summary_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=True,
))
# ── Function call arguments delta ───────────────────────
elif evt_type == "response.function_call_arguments.delta":
idx = (event.get("output_index", 0) if isinstance(event, dict)
else getattr(event, "output_index", 0))
delta_args = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if idx not in _pending_tools:
item_id = (event.get("item_id", f"tool_{idx}") if isinstance(event, dict)
else getattr(event, "item_id", f"tool_{idx}"))
_pending_tools[idx] = {
"id": call_id,
"name": fn_name,
"arguments": fn_args,
"id": item_id,
"name": "",
"arguments": "",
}
_pending_tools[idx]["arguments"] += delta_args
# ── Output item done — finalise tool call metadata ──────
elif evt_type == "response.output_item.done":
if isinstance(event, dict):
item = event.get("item", {})
idx = event.get("output_index", 0)
else:
item = getattr(event, "item", None) or {}
idx = getattr(event, "output_index", 0)
if isinstance(item, dict):
i_type = item.get("type", "")
else:
i_type = getattr(item, "type", "")
if i_type == "function_call":
if isinstance(item, dict):
call_id = item.get("call_id", item.get("id", f"tool_{idx}"))
fn_name = item.get("name", "")
fn_args = item.get("arguments", "")
else:
call_id = getattr(item, "call_id", None) or getattr(item, "id", f"tool_{idx}")
fn_name = getattr(item, "name", "")
fn_args = getattr(item, "arguments", "")
if idx in _pending_tools:
_pending_tools[idx]["id"] = call_id
_pending_tools[idx]["name"] = fn_name
# Prefer the accumulated delta args if available, else use final
if not _pending_tools[idx]["arguments"]:
_pending_tools[idx]["arguments"] = fn_args
else:
_pending_tools[idx] = {
"id": call_id,
"name": fn_name,
"arguments": fn_args,
}
finally:
self._close_responses_stream(response_stream)
# ── Build tool_calls list in output_index order ─────────────
tool_calls: List[Dict[str, Any]] = []
@@ -7393,80 +7451,93 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
if _emit:
from ..streaming.events import StreamEvent, StreamEventType
async for event in await litellm.aresponses(**params):
if isinstance(event, dict):
evt_type = event.get("type", "")
else:
evt_type = getattr(event, "type", "")
if evt_type == "response.completed":
_final_response = (event.get("response") if isinstance(event, dict)
else getattr(event, "response", None))
if evt_type == "response.output_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
response_text += delta_text
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=False,
))
elif evt_type == "response.reasoning_summary_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=True,
))
elif evt_type == "response.function_call_arguments.delta":
idx = (event.get("output_index", 0) if isinstance(event, dict)
else getattr(event, "output_index", 0))
delta_args = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if idx not in _pending_tools:
item_id = (event.get("item_id", f"tool_{idx}") if isinstance(event, dict)
else getattr(event, "item_id", f"tool_{idx}"))
_pending_tools[idx] = {"id": item_id, "name": "", "arguments": ""}
_pending_tools[idx]["arguments"] += delta_args
elif evt_type == "response.output_item.done":
response_stream = await litellm.aresponses(**params)
try:
async for event in response_stream:
if isinstance(event, dict):
item = event.get("item", {})
idx = event.get("output_index", 0)
evt_type = event.get("type", "")
else:
item = getattr(event, "item", None) or {}
idx = getattr(event, "output_index", 0)
evt_type = getattr(event, "type", "")
if isinstance(item, dict):
i_type = item.get("type", "")
else:
i_type = getattr(item, "type", "")
if evt_type == "response.failed":
failed_response = event.get("response") if isinstance(event, dict) else getattr(event, "response", None)
error = failed_response.get("error") if isinstance(failed_response, dict) else getattr(failed_response, "error", None)
self._raise_responses_failure(error)
if evt_type == "error":
self._raise_responses_failure(event)
if evt_type in ("response.completed", "response.incomplete"):
_final_response = (event.get("response") if isinstance(event, dict)
else getattr(event, "response", None))
self._record_finish_reason(_final_response)
if evt_type == "response.output_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
response_text += delta_text
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=False,
))
elif evt_type == "response.reasoning_summary_text.delta":
delta_text = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if _emit and delta_text:
stream_callback(StreamEvent(
type=StreamEventType.DELTA_TEXT,
timestamp=time.perf_counter(),
content=delta_text,
is_reasoning=True,
))
elif evt_type == "response.function_call_arguments.delta":
idx = (event.get("output_index", 0) if isinstance(event, dict)
else getattr(event, "output_index", 0))
delta_args = (event.get("delta", "") if isinstance(event, dict)
else getattr(event, "delta", ""))
if idx not in _pending_tools:
item_id = (event.get("item_id", f"tool_{idx}") if isinstance(event, dict)
else getattr(event, "item_id", f"tool_{idx}"))
_pending_tools[idx] = {"id": item_id, "name": "", "arguments": ""}
_pending_tools[idx]["arguments"] += delta_args
elif evt_type == "response.output_item.done":
if isinstance(event, dict):
item = event.get("item", {})
idx = event.get("output_index", 0)
else:
item = getattr(event, "item", None) or {}
idx = getattr(event, "output_index", 0)
if i_type == "function_call":
if isinstance(item, dict):
call_id = item.get("call_id", item.get("id", f"tool_{idx}"))
fn_name = item.get("name", "")
fn_args = item.get("arguments", "")
i_type = item.get("type", "")
else:
call_id = getattr(item, "call_id", None) or getattr(item, "id", f"tool_{idx}")
fn_name = getattr(item, "name", "")
fn_args = getattr(item, "arguments", "")
i_type = getattr(item, "type", "")
if idx in _pending_tools:
_pending_tools[idx]["id"] = call_id
_pending_tools[idx]["name"] = fn_name
if not _pending_tools[idx]["arguments"]:
_pending_tools[idx]["arguments"] = fn_args
else:
_pending_tools[idx] = {"id": call_id, "name": fn_name, "arguments": fn_args}
if i_type == "function_call":
if isinstance(item, dict):
call_id = item.get("call_id", item.get("id", f"tool_{idx}"))
fn_name = item.get("name", "")
fn_args = item.get("arguments", "")
else:
call_id = getattr(item, "call_id", None) or getattr(item, "id", f"tool_{idx}")
fn_name = getattr(item, "name", "")
fn_args = getattr(item, "arguments", "")
if idx in _pending_tools:
_pending_tools[idx]["id"] = call_id
_pending_tools[idx]["name"] = fn_name
if not _pending_tools[idx]["arguments"]:
_pending_tools[idx]["arguments"] = fn_args
else:
_pending_tools[idx] = {"id": call_id, "name": fn_name, "arguments": fn_args}
finally:
await self._aclose_responses_stream(response_stream)
tool_calls: List[Dict[str, Any]] = []
for idx in sorted(_pending_tools.keys()):
@@ -898,16 +898,34 @@ class OpenAIClient:
responses_content.append(part)
return responses_content
@staticmethod
def _responses_incomplete_finish_reason(response) -> Optional[str]:
"""Map known incomplete reasons to existing Chat Completions outcomes."""
status = response.get("status") if isinstance(response, dict) else getattr(response, "status", None)
if status != "incomplete":
return None
details = response.get("incomplete_details") if isinstance(response, dict) else getattr(response, "incomplete_details", None)
reason = details.get("reason") if isinstance(details, dict) else getattr(details, "reason", None)
return {"max_output_tokens": "length", "content_filter": "content_filter"}.get(reason)
def _responses_to_chat_completion(self, response) -> ChatCompletion:
"""
Wrap a Responses API response into a ChatCompletion dataclass
so that all downstream code (tool loop, display, callbacks)
works without modification.
"""
status = response.get("status") if isinstance(response, dict) else getattr(response, "status", None)
if status == "failed":
from .llm import LLMResponseError
error = response.get("error") if isinstance(response, dict) else getattr(response, "error", None)
code = error.get("code") if isinstance(error, dict) else getattr(error, "code", None)
message = error.get("message") if isinstance(error, dict) else getattr(error, "message", None)
label = f"Responses API failed ({code})" if code else "Responses API failed"
raise LLMResponseError(f"{label}: {message or 'Provider returned a failed response.'}")
response_text = ""
tool_calls_list: List[ToolCall] = []
output_items = getattr(response, 'output', None) or []
output_items = (response.get('output') if isinstance(response, dict) else getattr(response, 'output', None)) or []
for item in output_items:
item_type = getattr(item, "type", "") if not isinstance(item, dict) else item.get("type", "")
@@ -942,13 +960,16 @@ class OpenAIClient:
))
# Build usage
raw_usage = getattr(response, 'usage', None)
raw_usage = response.get('usage') if isinstance(response, dict) else getattr(response, 'usage', None)
usage = None
if raw_usage:
def usage_value(name):
return raw_usage.get(name, 0) if isinstance(raw_usage, dict) else getattr(raw_usage, name, 0)
usage = CompletionUsage(
prompt_tokens=getattr(raw_usage, 'input_tokens', 0),
completion_tokens=getattr(raw_usage, 'output_tokens', 0),
total_tokens=getattr(raw_usage, 'total_tokens', 0),
prompt_tokens=usage_value('input_tokens'),
completion_tokens=usage_value('output_tokens'),
total_tokens=usage_value('total_tokens'),
)
message = ChatCompletionMessage(
@@ -957,7 +978,7 @@ class OpenAIClient:
tool_calls=tool_calls_list if tool_calls_list else None,
)
choice = Choice(
finish_reason="tool_calls" if tool_calls_list else "stop",
finish_reason=self._responses_incomplete_finish_reason(response) or ("tool_calls" if tool_calls_list else "stop"),
index=0,
message=message,
)
@@ -1055,6 +1076,7 @@ class OpenAIClient:
Returns:
ChatCompletion object or None if error
"""
raw = None
# Lazy import StreamEvent types only when needed
_emit = emit_events and stream_callback is not None
if _emit:
@@ -1114,6 +1136,9 @@ class OpenAIClient:
except (FileNotFoundError, ValueError):
raise
except Exception as e:
status = raw.get("status") if isinstance(raw, dict) else getattr(raw, "status", None)
if status == "failed":
raise
self.logger.warning(f"Responses API streaming failed, falling back: {e}")
# Fall through to Chat Completions streaming
@@ -1284,6 +1309,9 @@ class OpenAIClient:
return final_response
except Exception as e:
status = raw.get("status") if isinstance(raw, dict) else getattr(raw, "status", None)
if status == "failed":
raise
self.logger.error(f"Error in stream processing: {e}")
return None
@@ -1320,6 +1348,7 @@ class OpenAIClient:
Returns:
ChatCompletion object or None if error
"""
raw = None
# Lazy import StreamEvent types only when needed
_emit = emit_events and stream_callback is not None
if _emit:
@@ -1388,6 +1417,9 @@ class OpenAIClient:
except (FileNotFoundError, ValueError):
raise
except Exception as e:
status = raw.get("status") if isinstance(raw, dict) else getattr(raw, "status", None)
if status == "failed":
raise
self.logger.warning(f"Responses API async streaming failed, falling back: {e}")
# Fall through to Chat Completions streaming
@@ -1550,6 +1582,9 @@ class OpenAIClient:
return final_response
except Exception as e:
status = raw.get("status") if isinstance(raw, dict) else getattr(raw, "status", None)
if status == "failed":
raise
self.logger.error(f"Error in async stream processing: {e}")
return None
@@ -1654,6 +1689,7 @@ class OpenAIClient:
"""
# ── Responses API path (non-streaming only) ────────────────────
if not stream and self._use_responses_api(model):
raw = None
try:
resp_params = self._build_responses_input(
messages, model, temperature, tools,
@@ -1664,6 +1700,9 @@ class OpenAIClient:
except (FileNotFoundError, ValueError):
raise
except Exception as e:
status = raw.get("status") if isinstance(raw, dict) else getattr(raw, "status", None)
if status == "failed":
raise
self.logger.warning(f"Responses API failed, falling back to Chat Completions: {e}")
# Fall through to Chat Completions
@@ -1718,6 +1757,7 @@ class OpenAIClient:
"""
# ── Responses API path (non-streaming only) ────────────────────
if not stream and self._use_responses_api(model):
raw = None
try:
resp_params = self._build_responses_input(
messages, model, temperature, tools,
@@ -1728,6 +1768,9 @@ class OpenAIClient:
except (FileNotFoundError, ValueError):
raise
except Exception as e:
status = raw.get("status") if isinstance(raw, dict) else getattr(raw, "status", None)
if status == "failed":
raise
self.logger.warning(f"Responses API failed, falling back to Chat Completions: {e}")
# Fall through to Chat Completions
@@ -6,6 +6,7 @@ No dependencies required - stores data as plain Python lists.
"""
import threading
from copy import deepcopy
from typing import Any, Dict, List, Optional
@@ -15,6 +16,8 @@ class InMemoryAdapter:
Stores short-term and long-term memories as plain Python lists.
Data is not persisted across process restarts.
Metadata must support deepcopy; writes and reads own independent copies so
caller edits cannot change stored records outside the adapter's lock.
Usage:
adapter = InMemoryAdapter()
@@ -48,7 +51,7 @@ class InMemoryAdapter:
"id": str(self._next_id),
"text": text,
"type": "short",
"metadata": metadata,
"metadata": deepcopy(metadata),
}
self._next_id += 1
self._data.append(entry)
@@ -64,7 +67,7 @@ class InMemoryAdapter:
for e in self._data
if e["type"] == "short" and query.lower() in e["text"].lower()
]
return results[:limit]
return deepcopy(results[:limit])
def store_long_term(
self, text: str, metadata: Optional[Dict[str, Any]] = None, **kwargs
@@ -74,7 +77,7 @@ class InMemoryAdapter:
"id": str(self._next_id),
"text": text,
"type": "long",
"metadata": metadata,
"metadata": deepcopy(metadata),
}
self._next_id += 1
self._data.append(entry)
@@ -90,7 +93,7 @@ class InMemoryAdapter:
for e in self._data
if e["type"] == "long" and query.lower() in e["text"].lower()
]
return results[:limit]
return deepcopy(results[:limit])
def delete_memory(self, memory_id: str, tier: Optional[str] = None, **kwargs) -> bool:
"""Delete a memory by ID. Returns True if an entry was removed.
@@ -123,5 +126,8 @@ class InMemoryAdapter:
def get_all_memories(self, **kwargs) -> List[Dict[str, Any]]:
with self._lock:
# Return defensive copy to prevent external mutation of internal state
return [dict(entry) for entry in self._data]
# Entries are owned on insertion and never mutated afterward.
# Keep their membership snapshot under lock, then copy nested data
# without blocking stores, deletes, or eviction.
snapshot = list(self._data)
return deepcopy(snapshot)
@@ -19,6 +19,16 @@ from ..protocols import MemoryProtocol
logger = get_logger(__name__)
def _metadata_user_id(raw):
"""Match the existing Python JSON reader, including non-finite values."""
try:
metadata = json.loads(raw) if raw else {}
return str(metadata["user_id"]) if isinstance(metadata, dict) and metadata.get("user_id") is not None else None
except (TypeError, ValueError):
return None
class SqliteMemoryAdapter:
"""
SQLite-based memory adapter implementing MemoryProtocol.
@@ -79,6 +89,7 @@ class SqliteMemoryAdapter:
check_same_thread=False,
timeout=10.0
)
self._local.stm_conn.create_function("memory_user_id", 1, _metadata_user_id, deterministic=True)
self._local.stm_conn.execute("""
CREATE TABLE IF NOT EXISTS short_term_memory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -103,6 +114,7 @@ class SqliteMemoryAdapter:
check_same_thread=False,
timeout=10.0
)
self._local.ltm_conn.create_function("memory_user_id", 1, _metadata_user_id, deterministic=True)
self._local.ltm_conn.execute("""
CREATE TABLE IF NOT EXISTS long_term_memory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -143,10 +155,13 @@ class SqliteMemoryAdapter:
) -> List[Dict[str, Any]]:
"""Search short-term memory."""
conn = self._get_stm_conn()
user_id = kwargs.get("user_id")
user_filter = " AND memory_user_id(metadata) = ?" if user_id is not None else ""
params = (f"%{query}%", str(user_id), limit) if user_id is not None else (f"%{query}%", limit)
cursor = conn.execute(
"SELECT id, content, metadata, timestamp FROM short_term_memory "
"WHERE content LIKE ? ORDER BY timestamp DESC LIMIT ?",
(f"%{query}%", limit)
"WHERE content LIKE ?" + user_filter + " ORDER BY timestamp DESC LIMIT ?",
params
)
results = []
@@ -184,10 +199,13 @@ class SqliteMemoryAdapter:
) -> List[Dict[str, Any]]:
"""Search long-term memory."""
conn = self._get_ltm_conn()
user_id = kwargs.get("user_id")
user_filter = " AND memory_user_id(metadata) = ?" if user_id is not None else ""
params = (f"%{query}%", str(user_id), limit) if user_id is not None else (f"%{query}%", limit)
cursor = conn.execute(
"SELECT id, content, metadata, timestamp FROM long_term_memory "
"WHERE content LIKE ? ORDER BY timestamp DESC LIMIT ?",
(f"%{query}%", limit)
"WHERE content LIKE ?" + user_filter + " ORDER BY timestamp DESC LIMIT ?",
params
)
results = []
@@ -252,8 +270,10 @@ class SqliteMemoryAdapter:
def get_all_memories(self, **kwargs) -> List[Dict[str, Any]]:
"""Get all memories from both short-term and long-term."""
short_memories = self.search_short_term("", limit=1000)
long_memories = self.search_long_term("", limit=1000)
# SQLite's negative LIMIT means no upper bound. Enumeration must return
# every row; ordinary searches retain their default/explicit limits.
short_memories = self.search_short_term("", limit=-1)
long_memories = self.search_long_term("", limit=-1)
# Mark memory types
for memory in short_memories:
@@ -298,4 +318,4 @@ class SqliteMemoryAdapter:
continue
with self._connection_lock:
self._all_connections.discard(conn)
setattr(self._local, attr, None)
setattr(self._local, attr, None)
@@ -179,6 +179,11 @@ class DocsManager:
self._docs[key] = doc
self._log(f"Loaded doc: {key}")
@staticmethod
def _effective_priority(priority: int, scope: str) -> int:
"""Apply the same global-document offset on creation and reload."""
return priority - 1000 if scope == "global" else priority
def _load_all_docs(self):
"""Load all docs from global and workspace directories."""
self._docs.clear()
@@ -188,7 +193,7 @@ class DocsManager:
self._load_docs_from_dir(self.global_docs_path, "global")
for key, doc in self._docs.items():
if key.startswith("global:"):
doc.priority = doc.priority - 1000 # Lower priority for global
doc.priority = self._effective_priority(doc.priority, "global")
# 2. Load workspace docs
workspace_docs_dir = self.workspace_path / self.DOCS_DIR_NAME.replace("/", os.sep)
@@ -355,8 +360,8 @@ class DocsManager:
frontmatter_lines = ["---"]
if description:
frontmatter_lines.append(f'description: "{description}"')
if priority != 0:
frontmatter_lines.append(f"priority: {priority}")
# Persist explicit zero instead of reloading the workspace default (100).
frontmatter_lines.append(f"priority: {priority}")
if tags:
tags_str = ", ".join(f'"{t}"' for t in tags)
frontmatter_lines.append(f"tags: [{tags_str}]")
@@ -372,7 +377,7 @@ class DocsManager:
name=name,
content=content,
description=description,
priority=priority,
priority=self._effective_priority(priority, scope),
tags=tags or [],
file_path=str(file_path)
)
@@ -1361,3 +1361,15 @@ def get_plugin_manager() -> PluginManager:
if _default_manager is None:
_default_manager = PluginManager()
return _default_manager
def has_registered_plugins() -> bool:
"""Return whether the global manager has any plugins registered.
Cheap, allocation-free fast path for the hot Agent-init merge calls: when
the singleton has not been created or holds no plugins (the common case),
callers can skip constructing/scanning the manager entirely. Does NOT
create the singleton as a side effect.
"""
manager = _default_manager
return manager is not None and bool(manager._plugins)
@@ -59,17 +59,22 @@ def _progress_section(checkpoint: Optional[Dict[str, Any]]) -> List[str]:
if not isinstance(checkpoint, dict):
return []
completed = checkpoint.get("completed_steps")
total = checkpoint.get("total_steps") or checkpoint.get("step_count")
last = checkpoint.get("last_step") or checkpoint.get("current_step")
if completed is None and last is None:
return []
if total:
head = f"PROGRESS: {completed or 0} of {total} workflow steps done"
else:
head = f"PROGRESS: {completed or 0} workflow steps done"
if last:
head += f"; last completed: {last}"
return [head]
total = checkpoint.get("total_steps")
if total is None:
total = checkpoint.get("step_count")
last = checkpoint.get("last_step")
current = checkpoint.get("current_step")
parts: List[str] = []
if completed is not None:
if total is not None:
parts.append(f"{completed} of {total} workflow steps done")
else:
parts.append(f"{completed} workflow steps done")
if last is not None and last != "":
parts.append(f"last completed: {last}")
if current is not None and current != "":
parts.append(f"current step: {current}")
return ["PROGRESS: " + "; ".join(parts)] if parts else []
def _recent_actions(recap_text: str) -> List[str]:
@@ -216,7 +216,7 @@ class SqliteTranscriptStore(DefaultSessionStore):
except (json.JSONDecodeError, TypeError):
return None
def _write_row(self, session: SessionData, conn=None, *, overwrite: bool = True) -> bool:
def _write_row(self, session: SessionData, conn=None) -> bool:
own_lock = conn is None
if conn is None:
conn = self._connect()
@@ -235,21 +235,17 @@ class SqliteTranscriptStore(DefaultSessionStore):
getattr(session, "updated_at", None),
)
sql = (
("INSERT OR REPLACE" if overwrite else "INSERT OR IGNORE") + " INTO sessions "
"INSERT OR REPLACE INTO sessions "
"(session_id, data, agent_name, gateway_session_id, agent_id, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?)"
)
try:
if own_lock:
with self._db_lock:
cursor = conn.execute(sql, params)
conn.execute(sql, params)
else:
cursor = conn.execute(sql, params)
if not overwrite and cursor.rowcount == 0:
raise FileExistsError(session.session_id)
conn.execute(sql, params)
return True
except FileExistsError:
raise
except Exception as exc:
logger.error("Failed to write session %s: %s", session.session_id, exc)
return False
@@ -292,52 +288,6 @@ class SqliteTranscriptStore(DefaultSessionStore):
self._cache[session.session_id] = session
return True
def _save_imported_session(self, session: SessionData, *, overwrite: bool = True) -> bool:
"""Restore to the database without applying the destination's window."""
session.updated_at = datetime.now(timezone.utc).isoformat()
with self._db_lock:
# INSERT OR IGNORE makes no-clobber atomic across store instances;
# checking session_exists before a replace would still race.
if not self._write_row(session, overwrite=overwrite):
return False
with self._lock:
self._cache[session.session_id] = session
return True
def export_all(self) -> Dict[str, Any]:
"""Export durable database rows, not legacy JSON sidecar files."""
return {
"version": self.PORTABLE_VERSION,
"sessions": [data for data in self._all_rows() if isinstance(data, dict)],
}
def _collect_lineage(
self, session: SessionData, *, exclude: str
) -> List[Dict[str, Any]]:
"""Collect persisted database records from the same conversation chain."""
lineage = self._lineage_key(session.to_dict())
if not lineage:
return []
def record_lineage(raw):
# Preserve the writer's Python JSON semantics (including NaN)
# and the shared lineage precedence, without retaining other rows.
try:
data = json.loads(raw)
return self._lineage_key(data) if isinstance(data, dict) else None
except (json.JSONDecodeError, TypeError, AttributeError):
return None
conn = self._connect()
with self._db_lock:
conn.create_function("portable_session_lineage", 1, record_lineage)
rows = conn.execute(
"SELECT data FROM sessions WHERE session_id != ? "
"AND portable_session_lineage(data) = ? ORDER BY updated_at DESC",
(exclude, lineage),
).fetchall()
return [json.loads(row[0]) for row in rows]
def _modify_session_locked(
self,
session_id: str,
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
description = "Praison AI agents for completing complex tasks with Self Reflection Agents"
readme = "README.md"
requires-python = ">=3.10"
@@ -0,0 +1,52 @@
"""Opt-in real LLM regression for documentation priority across reload.
Run with RUN_REAL_KEY_TESTS=1 and a configured provider key.
"""
import os
import pytest
@pytest.mark.skipif(
os.environ.get("RUN_REAL_KEY_TESTS") != "1",
reason="requires an explicitly enabled real provider",
)
def test_real_agent_uses_same_document_context_after_reload(tmp_path):
"""Effective priority must select the same source for real agent answers."""
from praisonaiagents import Agent
from praisonaiagents.memory.docs_manager import DocsManager
options = dict(
workspace_path=str(tmp_path),
global_docs_path=str(tmp_path / "global"),
)
manager = DocsManager(**options)
manager.create_doc("local", "The project codename is NEBULA-7.", priority=200)
manager.create_doc(
"shared", "The project codename is ORBIT-9.", priority=300, scope="global"
)
manager.create_doc("zero", "Supplementary reference.", priority=0)
original_context = manager.format_docs_for_prompt()
def verify(current):
"""Check effective priorities and the real answer for one manager lifecycle."""
assert current.get_doc("zero").priority == 0
assert current.get_doc("shared").priority == -700
context = current.format_docs_for_prompt()
assert context == original_context
assert "NEBULA-7" in context
assert "ORBIT-9" not in context
reader = Agent(
name="documentation-reader",
instructions="Answer only from this documentation:\n" + context,
)
result = reader.start("What is the project codename? Return only the codename.")
print(result)
assert "NEBULA-7" in str(result)
assert "ORBIT-9" not in str(result)
verify(manager)
manager.reload()
verify(manager)
verify(DocsManager(**options))
@@ -1191,6 +1191,74 @@ class TestPluginTypedSubsystems:
mgr.disable(name)
mgr.unregister(name)
def test_has_registered_plugins_reflects_registration(self):
"""The hot-path gate is False with no plugins and True once registered."""
from praisonaiagents.plugins.plugin import PluginType
from praisonaiagents.plugins import get_plugin_manager
from praisonaiagents.plugins.manager import has_registered_plugins
mgr = get_plugin_manager()
# Clean baseline: no plugins registered on the singleton.
if not mgr._plugins:
assert has_registered_plugins() is False
plugin, name = self._make_plugin(
PluginType.GUARDRAIL, name="gate_probe_test",
as_guardrail=lambda self: object(),
)
mgr.register(plugin)
try:
assert has_registered_plugins() is True
finally:
mgr.disable(name)
mgr.unregister(name)
def test_has_registered_plugins_does_not_create_singleton(self):
"""The gate must stay False and NOT build the singleton when uncreated.
The hot-path optimisation relies on the common case (no plugins ever
registered) being allocation-free. This isolates the
``_default_manager is None`` branch, which the other tests never reach
because they call ``get_plugin_manager()`` first.
"""
from praisonaiagents.plugins import manager as manager_mod
saved = manager_mod._default_manager
manager_mod._default_manager = None
try:
assert manager_mod.has_registered_plugins() is False
# The cheap check must not have constructed the singleton.
assert manager_mod._default_manager is None
finally:
manager_mod._default_manager = saved
def test_registered_plugin_still_merges_despite_fast_path(self):
"""Agent init must still route a registered plugin even with the gate."""
from praisonaiagents.plugins.plugin import PluginType
from praisonaiagents.plugins import get_plugin_manager
from praisonaiagents.plugins.manager import has_registered_plugins
from praisonaiagents import Agent
class _G:
def validate_output(self, content, **kw):
return True, content
plugin, name = self._make_plugin(
PluginType.GUARDRAIL, name="fastpath_merge_test",
as_guardrail=lambda self: _G(),
)
mgr = get_plugin_manager()
mgr.register(plugin)
mgr.enable(name)
try:
# Gate is True, so the merge methods do NOT early-return.
assert has_registered_plugins() is True
agent = Agent(name="t", instructions="x", llm="gpt-4o-mini")
assert agent._tool_result_guardrails
finally:
mgr.disable(name)
mgr.unregister(name)
if __name__ == "__main__":
pytest.main([__file__, "-v"])
@@ -0,0 +1,92 @@
"""Native failed Responses must surface without a second provider request."""
from types import SimpleNamespace
import pytest
from openai.types.responses import ResponseError
from openai.types.chat import ChatCompletionChunk
from praisonaiagents.llm.llm import LLMResponseError
from praisonaiagents.llm.openai_client import OpenAIClient
@pytest.mark.asyncio
@pytest.mark.parametrize('mode', ['sync', 'async', 'sync_stream', 'async_stream'])
@pytest.mark.parametrize('shape', ['sdk', 'dict'])
@pytest.mark.parametrize('profile', ['failed', 'missing_error', 'completed', 'unsupported'])
async def test_native_responses_failure_and_compatibility_fallback(mode, shape, profile, monkeypatch):
monkeypatch.delenv('OPENAI_API_BASE', raising=False)
monkeypatch.delenv('OPENAI_BASE_URL', raising=False)
calls = []
error = ResponseError(code='server_error', message='Provider failed.') if profile == 'failed' else None
fields = {
'status': 'failed' if profile in ('failed', 'missing_error') else 'completed',
'error': error if shape == 'sdk' or error is None else error.model_dump(),
'output': [{'type': 'message', 'content': [{'type': 'output_text', 'text': 'answer'}]}],
'usage': {'input_tokens': 3, 'output_tokens': 2, 'total_tokens': 5},
}
if shape == 'sdk':
fields['usage'] = SimpleNamespace(**fields['usage'])
raw = SimpleNamespace(**fields) if shape == 'sdk' else fields
def respond(**kwargs):
calls.append('responses')
if profile == 'unsupported':
raise NotImplementedError('Responses unsupported')
return raw
async def arespond(**kwargs):
return respond(**kwargs)
fallback = SimpleNamespace(choices=[SimpleNamespace(message=SimpleNamespace(content='fallback'))])
def chat(**kwargs):
calls.append('chat')
if kwargs.get('stream'):
return iter([ChatCompletionChunk(
id='chat-test', object='chat.completion.chunk', created=0, model='gpt-4o-mini',
choices=[{'index': 0, 'delta': {'content': 'fallback'}, 'finish_reason': 'stop'}],
)])
return fallback
async def achat(**kwargs):
result = chat(**kwargs)
if kwargs.get('stream'):
async def chunks():
for chunk in result:
yield chunk
return chunks()
return result
client = OpenAIClient(api_key='sk-test-not-real')
client._sync_client = SimpleNamespace(
responses=SimpleNamespace(create=respond), chat=SimpleNamespace(completions=SimpleNamespace(create=chat)),
)
client._async_client = SimpleNamespace(
responses=SimpleNamespace(create=arespond), chat=SimpleNamespace(completions=SimpleNamespace(create=achat)),
)
async def invoke():
messages = [{'role': 'user', 'content': 'question'}]
if mode == 'sync_stream':
return client.process_stream_response(messages, model='gpt-4o-mini')
if mode == 'async_stream':
return await client.process_stream_response_async(messages, model='gpt-4o-mini')
return client.create_completion(messages) if mode == 'sync' else await client.acreate_completion(messages)
if profile in ('failed', 'missing_error'):
with pytest.raises(LLMResponseError, match='Responses API failed') as caught:
await invoke()
if error:
assert 'server_error' in str(caught.value)
assert 'Provider failed.' in str(caught.value)
assert calls == ['responses']
else:
result = await invoke()
assert result.choices[0].message.content == ('fallback' if profile == 'unsupported' else 'answer')
assert calls == (['responses', 'chat'] if profile == 'unsupported' else ['responses'])
if profile == 'completed':
assert result.usage is not None
assert result.usage.prompt_tokens == 3
assert result.usage.completion_tokens == 2
assert result.usage.total_tokens == 5
@@ -0,0 +1,125 @@
"""Provider failures cannot become successful Responses answers."""
from types import SimpleNamespace
import pytest
from openai.types.responses import ResponseError
from praisonaiagents.llm.llm import LLM, LLMResponseError
@pytest.mark.asyncio
@pytest.mark.parametrize('entry', ['sync', 'async', 'stream', 'astream'])
@pytest.mark.parametrize('shape', ['sdk', 'dict'])
@pytest.mark.parametrize('profile', ['failed', 'missing_error', 'completed'])
async def test_public_responses_failure(entry, shape, profile, monkeypatch):
error = ResponseError(code='server_error', message='Provider failed.') if profile == 'failed' else None
status = 'completed' if profile == 'completed' else 'failed'
fields = {
'status': status,
'error': error if shape == 'sdk' or error is None else error.model_dump(),
'output': [{'type': 'message', 'content': [{'type': 'output_text', 'text': 'partial'}]}],
}
response = SimpleNamespace(**fields) if shape == 'sdk' else fields
llm = LLM(model='gpt-4o-mini')
requests = []
def respond(**kwargs):
requests.append(kwargs)
return response
async def arespond(**kwargs):
return respond(**kwargs)
llm._call_responses_api = respond
llm._call_responses_api_async = arespond
if entry in ('stream', 'astream'):
import litellm
monkeypatch.setattr('praisonaiagents.llm.llm.check_model_request', lambda *args: None)
events = [
{'type': 'response.output_text.delta', 'delta': 'partial'},
{'type': 'response.' + status, 'response': response},
]
if shape == 'sdk':
events = [SimpleNamespace(**event) for event in events]
def stream(**kwargs):
requests.append(kwargs)
return iter(events)
async def astream(**kwargs):
requests.append(kwargs)
async def iterate():
for event in events:
yield event
return iterate()
monkeypatch.setattr(litellm, 'responses', stream)
monkeypatch.setattr(litellm, 'aresponses', astream)
async def invoke():
kwargs = dict(stream=entry in ('stream', 'astream'), verbose=False)
return await llm.get_response_async('question', **kwargs) if entry in ('async', 'astream') else llm.get_response('question', **kwargs)
if profile == 'completed':
assert await invoke() == 'partial'
assert llm._last_stop_reason == 'completed'
else:
with pytest.raises(LLMResponseError, match='Responses API failed') as caught:
await invoke()
assert llm._last_stop_reason == 'error'
if error:
assert 'server_error' in str(caught.value)
assert 'Provider failed.' in str(caught.value)
assert len(requests) == 1
@pytest.mark.asyncio
@pytest.mark.parametrize('mode', ['sync', 'async'])
@pytest.mark.parametrize('shape', ['sdk', 'dict'])
async def test_stream_error_event_prevents_pending_tool_execution(mode, shape, monkeypatch):
import litellm
llm = LLM(model='gpt-4o-mini')
effects = []
requests = []
def lookup() -> str:
"""Return a lookup result."""
effects.append('executed')
return 'found'
events = [
{'type': 'response.output_item.done', 'output_index': 0, 'item': {
'type': 'function_call', 'call_id': 'call_1', 'name': 'lookup', 'arguments': '{}',
}},
{'type': 'error', 'code': 'server_error', 'message': 'Stream failed.'},
]
if shape == 'sdk':
events = [SimpleNamespace(**event) for event in events]
monkeypatch.setattr('praisonaiagents.llm.llm.check_model_request', lambda *args: None)
def stream(**kwargs):
requests.append(kwargs)
return iter(events)
async def astream(**kwargs):
requests.append(kwargs)
async def iterate():
for event in events:
yield event
return iterate()
monkeypatch.setattr(litellm, 'responses', stream)
monkeypatch.setattr(litellm, 'aresponses', astream)
with pytest.raises(LLMResponseError, match='Stream failed.'):
kwargs = dict(stream=True, verbose=False, tools=[lookup], max_iterations=1)
if mode == 'sync':
llm.get_response('question', **kwargs)
else:
await llm.get_response_async('question', **kwargs)
assert effects == []
assert len(requests) == 1
@@ -0,0 +1,115 @@
"""Incomplete Responses retain their terminal reason and streamed usage."""
from types import SimpleNamespace
import pytest
from openai.types.responses.response import IncompleteDetails
from praisonaiagents.llm.llm import LLM
from praisonaiagents.llm.openai_client import OpenAIClient
@pytest.fixture(autouse=True)
def isolate_responses_endpoint(monkeypatch):
monkeypatch.delenv('OPENAI_API_BASE', raising=False)
monkeypatch.delenv('OPENAI_BASE_URL', raising=False)
@pytest.mark.asyncio
@pytest.mark.parametrize('entry', ['llm_sync', 'llm_async', 'llm_stream', 'llm_astream', 'client_sync', 'client_async'])
@pytest.mark.parametrize('shape', ['sdk', 'dict'])
@pytest.mark.parametrize('status,reason,expected_finish,expected_outcome', [
('incomplete', 'max_output_tokens', 'length', 'length_truncated'),
('incomplete', 'content_filter', 'content_filter', 'content_filtered'),
('completed', 'max_output_tokens', 'stop', 'completed'),
])
async def test_public_incomplete_responses(entry, shape, status, reason, expected_finish, expected_outcome, monkeypatch):
details = IncompleteDetails(reason=reason)
response = SimpleNamespace(
status=status, incomplete_details=details if shape == 'sdk' else details.model_dump(),
output=[{'type': 'message', 'content': [{'type': 'output_text', 'text': 'partial'}]}],
usage=(
SimpleNamespace(input_tokens=3, output_tokens=2, total_tokens=5)
if shape == 'sdk' else {'input_tokens': 3, 'output_tokens': 2, 'total_tokens': 5}
),
)
requests = []
def respond(**kwargs):
requests.append(kwargs)
return response
async def arespond(**kwargs):
return respond(**kwargs)
if entry.startswith('llm'):
import litellm
llm = LLM(model='gpt-4o-mini')
monkeypatch.setattr('praisonaiagents.llm.llm.check_model_request', lambda *args: None)
monkeypatch.setattr(litellm, 'responses', respond)
monkeypatch.setattr(litellm, 'aresponses', arespond)
if entry in ('llm_stream', 'llm_astream'):
events = [
{'type': 'response.output_text.delta', 'delta': 'partial'},
{'type': 'response.' + status, 'response': response},
]
if shape == 'sdk':
events = [SimpleNamespace(**event) for event in events]
def stream(**kwargs):
requests.append(kwargs)
return iter(events)
async def astream(**kwargs):
requests.append(kwargs)
async def iterate():
for event in events:
yield event
return iterate()
monkeypatch.setattr(litellm, 'responses', stream)
monkeypatch.setattr(litellm, 'aresponses', astream)
kwargs = dict(stream=entry in ('llm_stream', 'llm_astream'), verbose=False)
answer = await llm.get_response_async('question', **kwargs) if entry in ('llm_async', 'llm_astream') else llm.get_response('question', **kwargs)
assert answer == 'partial'
assert llm._last_stop_reason == expected_outcome
for metrics in (llm.last_token_metrics, llm.session_token_metrics):
assert metrics is not None
assert metrics.input_tokens == 3
assert metrics.output_tokens == 2
assert metrics.total_tokens == 5
else:
client = OpenAIClient(api_key='sk-test-not-real')
client._sync_client = SimpleNamespace(responses=SimpleNamespace(create=respond))
client._async_client = SimpleNamespace(responses=SimpleNamespace(create=arespond))
messages = [{'role': 'user', 'content': 'question'}]
answer = client.create_completion(messages) if entry == 'client_sync' else await client.acreate_completion(messages)
assert answer.choices[0].message.content == 'partial'
assert answer.choices[0].finish_reason == expected_finish
assert answer.usage.prompt_tokens == 3
assert answer.usage.completion_tokens == 2
assert answer.usage.total_tokens == 5
assert len(requests) == 1
@pytest.mark.parametrize('reason', ['max_output_tokens', 'content_filter'])
@pytest.mark.parametrize('existing', ['max_steps', 'cancelled'])
def test_incomplete_response_preserves_stronger_terminal_reason(reason, existing):
llm = LLM(model='gpt-4o-mini')
llm._last_stop_reason = existing
llm._extract_from_responses_output(SimpleNamespace(
status='incomplete', incomplete_details=IncompleteDetails(reason=reason), output=[],
))
assert llm._last_stop_reason == existing
@pytest.mark.parametrize('reason', [None, 'unknown_future_reason'])
def test_unrecognised_incomplete_reason_preserves_existing_behavior(reason):
llm = LLM(model='gpt-4o-mini')
llm._extract_from_responses_output(SimpleNamespace(
status='incomplete', incomplete_details={'reason': reason}, output=[],
))
assert llm._last_stop_reason == 'completed'
@@ -0,0 +1,141 @@
"""Responses streams release their transport on every consumer exit."""
import asyncio
import pytest
from praisonaiagents.llm.llm import LLM, LLMResponseError
@pytest.mark.asyncio
@pytest.mark.parametrize('mode', ['sync', 'async'])
@pytest.mark.parametrize('shape', ['direct', 'litellm'])
@pytest.mark.parametrize('exit_kind', ['completed', 'failed', 'callback', 'cancelled'])
async def test_responses_stream_releases_transport(mode, shape, exit_kind, monkeypatch):
import httpx
import litellm
if mode == 'sync' and exit_kind == 'cancelled':
pytest.skip('Task cancellation is an asynchronous exit')
closed = []
class SyncTransport(httpx.SyncByteStream):
def __iter__(self):
return iter([])
def close(self):
closed.append('closed')
class AsyncTransport(httpx.AsyncByteStream):
async def __aiter__(self):
if False:
yield b''
async def aclose(self):
closed.append('closed')
response = httpx.Response(200, stream=SyncTransport() if mode == 'sync' else AsyncTransport())
event = {'type': 'response.output_text.delta', 'delta': 'answer'}
if exit_kind == 'failed':
event = {'type': 'response.failed', 'response': {'error': {'code': 'server_error', 'message': 'Provider failed.'}}}
class Stream:
def __init__(self):
self.events = iter([event])
if shape == 'litellm':
self.response = response
elif mode == 'sync':
self.close = response.close
else:
self.aclose = response.aclose
def __iter__(self):
return self
def __next__(self):
return next(self.events)
def __aiter__(self):
return self
async def __anext__(self):
if exit_kind == 'cancelled':
raise asyncio.CancelledError()
try:
return next(self.events)
except StopIteration:
raise StopAsyncIteration
stream = Stream()
monkeypatch.setattr('praisonaiagents.llm.llm.check_model_request', lambda *args: None)
monkeypatch.setattr(litellm, 'responses', lambda **kwargs: stream)
async def arespond(**kwargs):
return stream
monkeypatch.setattr(litellm, 'aresponses', arespond)
llm = LLM(model='gpt-4o-mini')
def callback(event):
if exit_kind == 'callback':
raise ValueError('Callback failed.')
async def invoke():
kwargs = dict(stream_callback=callback, emit_events=True)
if mode == 'sync':
return llm._stream_responses_api({'model': llm.model}, **kwargs)
return await llm._stream_responses_api_async({'model': llm.model}, **kwargs)
if exit_kind == 'failed':
with pytest.raises(LLMResponseError, match='Provider failed.'):
await invoke()
elif exit_kind == 'callback':
with pytest.raises(ValueError, match='Callback failed.'):
await invoke()
elif exit_kind == 'cancelled':
with pytest.raises(asyncio.CancelledError):
await invoke()
else:
assert (await invoke())[0] == 'answer'
assert response.is_closed
assert closed == ['closed']
@pytest.mark.asyncio
@pytest.mark.parametrize('mode', ['sync', 'async'])
async def test_cleanup_error_does_not_mask_provider_failure(mode, monkeypatch):
import litellm
event = {'type': 'error', 'message': 'Original failure.'}
def fail_close():
raise ValueError('Cleanup failed.')
async def fail_aclose():
fail_close()
class AsyncStream:
aclose = staticmethod(fail_aclose)
def __aiter__(self):
async def events():
yield event
return events()
monkeypatch.setattr('praisonaiagents.llm.llm.check_model_request', lambda *args: None)
class SyncStream:
close = staticmethod(fail_close)
def __iter__(self):
return iter([event])
monkeypatch.setattr(litellm, 'responses', lambda **kwargs: SyncStream())
async def arespond(**kwargs):
return AsyncStream()
monkeypatch.setattr(litellm, 'aresponses', arespond)
llm = LLM(model='gpt-4o-mini')
with pytest.raises(LLMResponseError, match='Original failure.'):
if mode == 'sync':
llm._stream_responses_api({'model': llm.model})
else:
await llm._stream_responses_api_async({'model': llm.model})
@@ -0,0 +1,36 @@
"""Created documentation must keep the same effective priority after reload."""
import pytest
from praisonaiagents.memory.docs_manager import DocsManager
@pytest.mark.parametrize("scope", ["workspace", "global"])
@pytest.mark.parametrize("priority", [0, -5, 200, 1500])
def test_created_doc_priority_and_selection_survive_reload(tmp_path, scope, priority):
"""Persist explicit priorities and preserve selection across manager lifecycles."""
options = dict(workspace_path=str(tmp_path), global_docs_path=str(tmp_path / "global"))
manager = DocsManager(**options)
manager.create_doc("reference", "Stable documentation content", priority=priority, scope=scope)
before_priority = manager.get_doc("reference").priority
before_selection = [doc.name for doc in manager.get_docs_for_context()]
manager.reload()
reopened = DocsManager(**options)
for current in (manager, reopened):
assert current.get_doc("reference").priority == before_priority
assert [doc.name for doc in current.get_docs_for_context()] == before_selection
expected_priority = priority - 1000 if scope == "global" else priority
assert before_priority == expected_priority
def test_new_global_doc_does_not_temporarily_outrank_workspace_doc(tmp_path):
"""Apply the global priority offset immediately and keep selection across reload."""
options = dict(workspace_path=str(tmp_path), global_docs_path=str(tmp_path / "global"))
manager = DocsManager(**options)
manager.create_doc("local", "local content", priority=200)
manager.create_doc("shared", "global content", priority=300, scope="global")
manager.reload()
reopened = DocsManager(**options)
for current in (manager, reopened):
assert [doc.name for doc in current.get_all_docs()] == ["local", "shared"]
assert [doc.name for doc in current.get_docs_for_context()] == ["local"]
@@ -0,0 +1,83 @@
"""Caller edits must not rewrite previously stored in-memory records."""
import pytest
import threading
from concurrent.futures import ThreadPoolExecutor
import praisonaiagents.memory.adapters.in_memory_adapter as adapter_module
from praisonaiagents.memory.adapters.in_memory_adapter import InMemoryAdapter
@pytest.mark.parametrize('tier', ['short', 'long'])
def test_store_snapshots_nested_metadata(tier):
adapter = InMemoryAdapter()
metadata = {'labels': ['original'], 'details': {'count': 1}}
memory_id = getattr(adapter, f'store_{tier}_term')('original text', metadata)
metadata['labels'].append('caller edit')
metadata['details']['count'] = 2
record = getattr(adapter, f'search_{tier}_term')('original')[0]
assert record['id'] == memory_id
assert record['metadata'] == {'labels': ['original'], 'details': {'count': 1}}
@pytest.mark.parametrize('tier', ['short', 'long'])
@pytest.mark.parametrize('read', ['search', 'all'])
def test_read_results_do_not_alias_stored_records(tier, read):
adapter = InMemoryAdapter()
memory_id = getattr(adapter, f'store_{tier}_term')(
'original text', {'labels': ['original'], 'details': {'count': 1}},
)
search = getattr(adapter, f'search_{tier}_term')
records = search('original') if read == 'search' else adapter.get_all_memories()
records[0]['id'] = 'caller id'
records[0]['text'] = 'caller text'
records[0]['metadata']['labels'].append('caller edit')
records[0]['metadata']['details']['count'] = 2
records.clear()
assert search('original') == [{
'id': memory_id, 'text': 'original text', 'type': tier,
'metadata': {'labels': ['original'], 'details': {'count': 1}},
}]
assert adapter.delete_memory(memory_id, tier=tier)
assert adapter.get_all_memories() == []
@pytest.mark.parametrize('tier', ['short', 'long'])
@pytest.mark.parametrize('metadata', [None, {}])
def test_metadata_defaults_and_search_limits_remain_unchanged(tier, metadata):
adapter = InMemoryAdapter()
write = getattr(adapter, f'store_{tier}_term')
expected_ids = [write('matching text', metadata) for _ in range(6)]
search = getattr(adapter, f'search_{tier}_term')
assert [record['id'] for record in search('matching')] == expected_ids[:5]
assert [record['id'] for record in search('matching', limit=2)] == expected_ids[:2]
assert all(record['metadata'] == metadata for record in adapter.get_all_memories())
def test_enumeration_copy_allows_concurrent_eviction_and_preserves_snapshot(monkeypatch):
"""A slow nested copy must not keep writers behind the collection lock."""
adapter = InMemoryAdapter(max_size=1)
original_id = adapter.store_short_term('original', {'nested': ['original']})
copying = threading.Event()
resume = threading.Event()
real_copy = adapter_module.deepcopy
def paused_copy(value):
if isinstance(value, list):
copying.set()
assert resume.wait(5)
return real_copy(value)
monkeypatch.setattr(adapter_module, 'deepcopy', paused_copy)
with ThreadPoolExecutor(max_workers=2) as executor:
read = executor.submit(adapter.get_all_memories)
try:
assert copying.wait(2)
new_id = executor.submit(adapter.store_long_term, 'new', {'nested': ['new']}).result(timeout=1)
finally:
resume.set()
snapshot = read.result(timeout=5)
assert snapshot == [{'id': original_id, 'text': 'original', 'type': 'short', 'metadata': {'nested': ['original']}}]
snapshot[0]['metadata']['nested'].append('caller edit')
assert adapter.get_all_memories() == [{'id': new_id, 'text': 'new', 'type': 'long', 'metadata': {'nested': ['new']}}]
@@ -0,0 +1,99 @@
"""Enumerating SQLite memories must include rows beyond search-sized caps."""
import pytest
from praisonaiagents.memory.adapters.sqlite_adapter import SqliteMemoryAdapter
from praisonaiagents.memory.memory import Memory
@pytest.mark.parametrize('facade', [False, True])
@pytest.mark.parametrize('short_count,long_count', [(0, 0), (1000, 1), (1, 1000), (1001, 1001)])
def test_get_all_memories_includes_every_stored_row(tmp_path, monkeypatch, facade, short_count, long_count):
monkeypatch.setenv('PRAISONAI_HOME', str(tmp_path / 'home'))
monkeypatch.chdir(tmp_path)
store = Memory(config={'provider': 'sqlite'}) if facade else SqliteMemoryAdapter(
short_db=str(tmp_path / 'short.db'), long_db=str(tmp_path / 'long.db'),
)
adapter = store.memory_adapter if facade else store
expected = set()
try:
for tier, count in [('short', short_count), ('long', long_count)]:
write = getattr(store, f'store_{tier}_term')
for index in range(count):
text = f'{tier} entry {index}'
memory_id = write(text, metadata={'index': index})
expected.add((f'{tier}_term', memory_id, text, index))
records = store.get_all_memories()
assert len(records) == short_count + long_count
assert {
(record['memory_type'], record['id'], record['text'], record['metadata']['index'])
for record in records
} == expected
if facade:
assert all(record['type'] == record['memory_type'] for record in records)
# Enumeration must not widen the public search defaults or explicit limit.
for tier, count in [('short', short_count), ('long', long_count)]:
search = getattr(adapter, f'search_{tier}_term')
assert len(search('')) == min(5, count)
assert len(search('', limit=2)) == min(2, count)
finally:
adapter.close_connections()
if facade:
store.close_connections()
@pytest.mark.parametrize('tier', ['short', 'long'])
@pytest.mark.parametrize('value', [float('nan'), float('inf'), -float('inf')])
def test_filtered_search_accepts_writer_nonfinite_metadata(tmp_path, tier, value):
adapter = SqliteMemoryAdapter(short_db=str(tmp_path / 'short.db'), long_db=str(tmp_path / 'long.db'))
try:
write = getattr(adapter, f'store_{tier}_term')
search = getattr(adapter, f'search_{tier}_term')
write('other record', metadata={'user_id': 'other', 'value': value})
wanted = write('wanted record', metadata={'user_id': 'wanted', 'value': value})
assert [item['id'] for item in search('', user_id='wanted', limit=1)] == [wanted]
assert search('', user_id='absent') == []
finally:
adapter.close_connections()
@pytest.mark.parametrize('tier', ['short', 'long'])
def test_filtered_search_ignores_corrupt_peer_metadata_and_keeps_string_matching(tmp_path, tier):
adapter = SqliteMemoryAdapter(short_db=str(tmp_path / 'short.db'), long_db=str(tmp_path / 'long.db'))
try:
write = getattr(adapter, f'store_{tier}_term')
wanted = write('wanted', metadata={'user_id': True})
conn = getattr(adapter, f'_get_{"stm" if tier == "short" else "ltm"}_conn')()
conn.execute(f'INSERT INTO {tier}_term_memory (content, metadata) VALUES (?, ?)', ('broken peer', '{invalid'))
conn.commit()
search = getattr(adapter, f'search_{tier}_term')
assert [item['id'] for item in search('', user_id='True', limit=1)] == [wanted]
finally:
adapter.close_connections()
@pytest.mark.parametrize('tier', ['short', 'long'])
def test_null_user_is_distinct_from_literal_none_user(tmp_path, tier):
adapter = SqliteMemoryAdapter(short_db=str(tmp_path / 'short.db'), long_db=str(tmp_path / 'long.db'))
try:
write = getattr(adapter, f'store_{tier}_term')
write('unowned', metadata={'user_id': None})
wanted = write('owned', metadata={'user_id': 'None'})
search = getattr(adapter, f'search_{tier}_term')
assert [item['id'] for item in search('', user_id='None')] == [wanted]
finally:
adapter.close_connections()
@pytest.mark.parametrize("tier", ["short_term", "long_term"])
@pytest.mark.parametrize("user_id", [0, False, ""])
def test_falsey_supplied_user_id_still_filters_before_limit(tmp_path, tier, user_id):
adapter = SqliteMemoryAdapter(short_db=str(tmp_path / 'short.db'), long_db=str(tmp_path / 'long.db'))
try:
write = getattr(adapter, f"store_{tier}")
search = getattr(adapter, f"search_{tier}")
write("other", metadata={"user_id": "other"})
wanted = write("wanted", metadata={"user_id": user_id})
assert [item["id"] for item in search("", user_id=user_id, limit=1)] == [wanted]
finally:
adapter.close_connections()
@@ -0,0 +1,29 @@
"""Continuation prompts must distinguish running work from proven completion."""
import pytest
from praisonaiagents.session import build_handoff_prompt
@pytest.mark.parametrize(
"checkpoint,expected,absent",
[
({"current_step": "deploy"}, "PROGRESS: current step: deploy", ["last completed", "steps done"]),
({"completed_steps": 0, "total_steps": 5, "current_step": "deploy"},
"PROGRESS: 0 of 5 workflow steps done; current step: deploy", ["last completed"]),
({"completed_steps": 2, "total_steps": 5, "last_step": "build", "current_step": "deploy"},
"PROGRESS: 2 of 5 workflow steps done; last completed: build; current step: deploy", []),
({"last_step": "build"}, "PROGRESS: last completed: build", ["steps done"]),
({"completed_steps": 0, "total_steps": 0, "step_count": 9},
"PROGRESS: 0 of 0 workflow steps done", ["of 9"]),
({"completed_steps": 2, "step_count": 5, "last_step": "build"},
"PROGRESS: 2 of 5 workflow steps done; last completed: build", []),
],
)
def test_handoff_reports_only_the_supplied_progress(checkpoint, expected, absent):
prompt = build_handoff_prompt(workflow_checkpoint=checkpoint)
assert expected in prompt
for text in absent:
assert text not in prompt
assert prompt.endswith("before acting.")
@@ -1,4 +1,4 @@
"""Default imports must not overwrite a session created after the first check."""
"""Imports must not overwrite a session created after the first check."""
from concurrent.futures import ThreadPoolExecutor
from threading import Event
@@ -6,13 +6,15 @@ from threading import Event
import pytest
from praisonaiagents.session.store import DefaultSessionStore
from praisonaiagents.session.sqlite_store import SqliteSessionStore
@pytest.mark.parametrize("kind", [DefaultSessionStore, SqliteSessionStore])
@pytest.mark.parametrize("overwrite", [False, True])
def test_import_rechecks_new_peer_session_under_write_lock(tmp_path, monkeypatch, overwrite):
def test_import_rechecks_new_peer_session_under_write_lock(tmp_path, monkeypatch, kind, overwrite):
directory = str(tmp_path / "sessions")
importer = DefaultSessionStore(session_dir=directory)
peer = DefaultSessionStore(session_dir=directory)
stores = [kind(session_dir=directory), kind(session_dir=directory)]
importer, peer = stores
ready, completed = Event(), Event()
original = importer._save_imported_session
@@ -35,18 +37,25 @@ def test_import_rechecks_new_peer_session_under_write_lock(tmp_path, monkeypatch
finally:
completed.set()
with ThreadPoolExecutor(max_workers=2) as pool:
future = pool.submit(create_peer)
report = importer.import_sessions(payload, overwrite=overwrite)
future.result(timeout=15)
fresh = DefaultSessionStore(session_dir=directory)
if overwrite:
assert report.imported == 1
assert not report.skipped
assert fresh.get_chat_history("shared") == [{"role": "user", "content": "imported"}]
else:
assert report.imported == 0
assert report.skipped == [{
"session_id": "shared", "reason": "already exists (use overwrite)",
}]
assert fresh.get_chat_history("shared") == [{"role": "user", "content": "peer-created"}]
try:
with ThreadPoolExecutor(max_workers=2) as pool:
future = pool.submit(create_peer)
report = importer.import_sessions(payload, overwrite=overwrite)
future.result(timeout=15)
fresh = kind(session_dir=directory)
stores.append(fresh)
if overwrite:
assert report.imported == 1
assert not report.skipped
assert fresh.get_chat_history("shared") == [{"role": "user", "content": "imported"}]
else:
assert report.imported == 0
assert report.skipped == [{
"session_id": "shared", "reason": "already exists (use overwrite)",
}]
assert fresh.get_chat_history("shared") == [{"role": "user", "content": "peer-created"}]
finally:
for store in stores:
conn = getattr(store, "_conn", None)
if conn is not None:
conn.close()
@@ -111,8 +111,8 @@ def test_write_between_import_save_and_index_keeps_newer_content(make_store, mon
resume = threading.Event()
original = DefaultSessionStore._save_imported_session
def delayed_save(store, session, **kwargs):
result = original(store, session, **kwargs)
def delayed_save(store, session, *, overwrite=True):
result = original(store, session, overwrite=overwrite)
saved.set()
assert resume.wait(5), "concurrent writer did not finish"
return result
@@ -1,179 +0,0 @@
"""Portable session operations use the SQLite transcript persistence backend."""
import json
import tracemalloc
import pytest
from praisonaiagents.session.store import DefaultSessionStore
from praisonaiagents.session.sqlite_transcript_store import SqliteTranscriptStore
@pytest.fixture
def stores(tmp_path):
opened = []
def make(name, backend=SqliteTranscriptStore, **kwargs):
store = backend(session_dir=str(tmp_path / name), **kwargs)
opened.append(store)
return store
yield make
for store in opened:
conn = getattr(store, "_conn", None)
if conn is not None:
conn.close()
def test_export_all_reads_database_not_json_sidecars(stores):
store = stores("source")
assert store.add_message("one", "user", "First")
assert store.add_message("two", "assistant", "Second")
from pathlib import Path
(Path(store.session_dir) / "unrelated.json").write_text(
json.dumps({"session_id": "sidecar", "messages": []}), encoding="utf-8"
)
payload = store.export_all()
assert payload["version"] == store.PORTABLE_VERSION
assert {record["session_id"] for record in payload["sessions"]} == {"one", "two"}
@pytest.mark.parametrize("key", ["lineage_id", "root_session_id", "thread_id"])
def test_single_export_includes_database_lineage(stores, key):
store = stores("source")
for session_id in ("first", "continuation", "unrelated"):
assert store.add_message(session_id, "user", session_id)
assert store.update_session_metadata(session_id, **{key: "shared" if session_id != "unrelated" else "other"})
payload = store.export_session("continuation")
assert {record["session_id"] for record in payload["sessions"]} == {"first", "continuation"}
assert [record["session_id"] for record in store.export_session("continuation", include_lineage=False)["sessions"]] == ["continuation"]
@pytest.mark.parametrize("source_backend", [DefaultSessionStore, SqliteTranscriptStore])
@pytest.mark.parametrize("target_backend", [DefaultSessionStore, SqliteTranscriptStore])
def test_cross_backend_restore_survives_reopen_without_window_truncation(stores, source_backend, target_backend):
source = stores("source", source_backend)
for index in range(6):
assert source.add_message("reference", "user", f"message-{index}")
assert source.set_gateway_info("reference", gateway_session_id="live", agent_id="agent")
payload = source.export_all()
assert len(payload["sessions"]) == 1
target = stores("target", target_backend, active_window=2, retention="truncate")
# An existing SQLite row prevents legacy JSON migration from masking a
# restore incorrectly written to a sidecar file instead of the database.
assert target.add_message("existing", "user", "Existing history")
assert target.import_sessions(payload).imported == 1
assert target.session_exists("reference")
target.invalidate_cache()
reopened = stores("target", target_backend, active_window=2, retention="truncate")
for current in (target, reopened):
assert [message["content"] for message in current.get_chat_history("reference")] == [f"message-{index}" for index in range(6)]
session = current.get_session("reference")
assert session.gateway_session_id is None
assert session.agent_id is None
if target_backend is SqliteTranscriptStore:
from pathlib import Path
assert list(Path(target.session_dir).glob("*.json")) == []
def test_import_sqlite_write_failure_is_reported_and_preserves_existing_row(stores):
source = stores("source", DefaultSessionStore)
assert source.add_message("reference", "user", "New history")
target = stores("target")
assert target.add_message("reference", "user", "Original history")
target._connect().execute("PRAGMA query_only=ON")
report = target.import_sessions(source.export_all(), overwrite=True)
assert report.imported == 0
assert report.skipped_count == 1
assert "write failed" in report.skipped[0]["reason"]
target.invalidate_cache()
assert target.get_chat_history("reference")[0]["content"] == "Original history"
@pytest.mark.parametrize("reset_live_fields", [True, False])
def test_restore_preserves_archived_history_and_tool_calls(stores, reset_live_fields):
tool_calls = [{"id": "call-1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}]
payload = {"sessions": [{
"session_id": "reference",
"gateway_session_id": "live",
"agent_id": "agent",
"metadata": {"gateway_session_id": "live", "agent_id": "agent", "lineage_id": "shared"},
"archived_messages": [{"role": "user", "content": "Archived", "timestamp": 1}],
"messages": [{"role": "assistant", "content": "Tool call", "tool_calls": tool_calls}],
}]}
target = stores("target", active_window=1, retention="truncate")
assert target.add_message("existing", "user", "Existing")
assert target.import_sessions(payload, reset_live_fields=reset_live_fields).imported == 1
record = stores("target").export_session("reference")["sessions"][0]
assert record["messages"][0]["tool_calls"] == tool_calls
assert record["archived_messages"][0]["content"] == "Archived"
assert record["metadata"]["lineage_id"] == "shared"
if reset_live_fields:
assert record.get("gateway_session_id") is None
assert record.get("agent_id") is None
assert "gateway_session_id" not in record["metadata"]
assert "agent_id" not in record["metadata"]
else:
assert record["gateway_session_id"] == "live"
assert record["agent_id"] == "agent"
assert record["metadata"]["gateway_session_id"] == "live"
assert record["metadata"]["agent_id"] == "agent"
def test_sqlite_restore_keeps_import_guards(stores):
source = stores("source", DefaultSessionStore)
assert source.add_message("reference", "user", "Original")
payload = source.export_all()
payload["sessions"].append(dict(payload["sessions"][0]))
target = stores("target")
report = target.import_sessions(payload)
assert report.imported == 1
assert "duplicate" in report.skipped[0]["reason"]
assert target.import_sessions(payload).imported == 0
assert target.import_sessions(payload, overwrite=True).imported == 1
assert target.export_session("missing")["sessions"] == []
payload["version"] = target.PORTABLE_VERSION + 1
assert target.import_sessions(payload, overwrite=True).imported == 0
assert target.get_chat_history("reference")[0]["content"] == "Original"
@pytest.mark.parametrize("overwrite", [False, True])
def test_import_does_not_clobber_peer_created_after_initial_check(stores, monkeypatch, overwrite):
target = stores("target")
peer = stores("target")
original = target._save_imported_session
def create_peer_before_write(*args, **kwargs):
assert peer.add_message("reference", "user", "peer-created")
return original(*args, **kwargs)
monkeypatch.setattr(target, "_save_imported_session", create_peer_before_write)
payload = {"sessions": [{"session_id": "reference", "messages": [{"role": "user", "content": "imported"}]}]}
report = target.import_sessions(payload, overwrite=overwrite)
assert report.imported == int(overwrite)
assert target.get_chat_history("reference")[0]["content"] == ("imported" if overwrite else "peer-created")
if overwrite:
assert target._cache["reference"].messages[0].content == "imported"
else:
assert report.skipped[0]["reason"] == "already exists (use overwrite)"
def test_lineage_export_does_not_materialize_unrelated_transcripts(stores):
store = stores("source")
for sid in ("root", "continuation"):
assert store.add_message(sid, "user", sid)
assert store.update_session_metadata(sid, lineage_id="shared")
for index in range(24):
sid = f"unrelated-{index}"
assert store.add_message(sid, "user", "x" * 256_000)
assert store.update_session_metadata(sid, lineage_id="other", value=float("nan"))
tracemalloc.start()
try:
payload = store.export_session("continuation")
_, peak = tracemalloc.get_traced_memory()
finally:
tracemalloc.stop()
assert {record["session_id"] for record in payload["sessions"]} == {"root", "continuation"}
# The unrelated transcripts total over 6 MB. A single-record parse can
# allocate its input and decoded content, but must not retain all rows.
assert peak < 2_000_000
+10 -6
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10"
resolution-markers = [
"python_full_version >= '3.14'",
@@ -2263,7 +2263,7 @@ wheels = [
[[package]]
name = "mcp"
version = "1.25.0"
version = "1.30.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "anyio" },
@@ -2281,9 +2281,9 @@ dependencies = [
{ name = "typing-inspection" },
{ name = "uvicorn", marker = "sys_platform != 'emscripten'" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d5/2d/649d80a0ecf6a1f82632ca44bec21c0461a9d9fc8934d38cb5b319f2db5e/mcp-1.25.0.tar.gz", hash = "sha256:56310361ebf0364e2d438e5b45f7668cbb124e158bb358333cd06e49e83a6802", size = 605387, upload-time = "2025-12-19T10:19:56.985Z" }
sdist = { url = "https://files.pythonhosted.org/packages/ba/93/0142dc84a666daf8ad51a34268f34c12fd6fda4f3810c4be2504eecc8212/mcp-1.30.0.tar.gz", hash = "sha256:445414625fce5c295faa505bb11bacece661ab6f4028d57c935db57820b7a3e4", size = 680511, upload-time = "2026-09-07T14:34:15.845Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/e2/fc/6dc7659c2ae5ddf280477011f4213a74f806862856b796ef08f028e664bf/mcp-1.25.0-py3-none-any.whl", hash = "sha256:b37c38144a666add0862614cc79ec276e97d72aa8ca26d622818d4e278b9721a", size = 233076, upload-time = "2025-12-19T10:19:55.416Z" },
{ url = "https://files.pythonhosted.org/packages/f5/f4/e58bc33317c92a0203664daaf00bf6f41166cc0149e5d6870a03f7cd004a/mcp-1.30.0-py3-none-any.whl", hash = "sha256:666edb5009503e1047c9d60346a756f94b261f05cc2625f23d41c728ffc484d0", size = 234581, upload-time = "2026-09-07T14:34:14.266Z" },
]
[[package]]
@@ -3123,7 +3123,7 @@ wheels = [
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@@ -3219,6 +3219,9 @@ os = [
{ name = "pyjwt" },
{ name = "uvicorn" },
]
parallel-search = [
{ name = "mcp" },
]
sandbox = [
{ name = "praisonai-sandbox" },
]
@@ -3250,6 +3253,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -3288,7 +3292,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "primp"
+1
View File
@@ -1,3 +1,4 @@
# yaml-language-server: $schema=https://raw.githubusercontent.com/MervinPraison/PraisonAI/main/src/praisonai-bot/praisonai_bot/bots/gateway.schema.json
gateway:
host: 127.0.0.1
port: 8765
+1 -1
View File
@@ -1 +1 @@
__version__ = "0.1.8"
__version__ = "0.1.9"
@@ -23,6 +23,41 @@ from pydantic import (
logger = logging.getLogger(__name__)
#: Stable, published URL for the ``gateway.yaml`` / ``bot.yaml`` JSON Schema,
#: mirroring the ``agents.yaml`` convention (``AGENTS_SCHEMA_URL``). Editors that
#: speak the YAML language server use this via a leading
#: ``# yaml-language-server: $schema=<url>`` header to provide autocomplete,
#: inline validation, and hover docs while authoring the gateway/bot config.
GATEWAY_SCHEMA_URL = (
"https://raw.githubusercontent.com/MervinPraison/PraisonAI/main/"
"src/praisonai-bot/praisonai_bot/bots/gateway.schema.json"
)
#: Leading YAML comment prepended to scaffolded ``gateway.yaml`` / ``bot.yaml``
#: files so editors wire up validation out of the box. A leading comment is
#: ignored by ``yaml.safe_load``, so execution is unaffected.
GATEWAY_SCHEMA_HEADER = f"# yaml-language-server: $schema={GATEWAY_SCHEMA_URL}\n"
def resolve_model_by_provider() -> str:
"""Resolve a default model for whichever provider credential is present.
Reuses the existing detection helper
(``praisonai_code.llm.env.default_model_for_available_provider``) through
the lazy cross-tier bridge so the gateway path leads with the provider the
operator already has (Anthropic/Gemini/Ollama/…) instead of assuming
OpenAI. Falls back to ``gpt-4o-mini`` only when no provider is detected or
the helper is unavailable.
"""
try:
from praisonai_bot._code_bridge import import_code_module
env = import_code_module("praisonai_code.llm.env")
return env.default_model_for_available_provider()
except Exception: # pragma: no cover - detection is best-effort
return "gpt-4o-mini"
def _register_redaction(value: str) -> None:
"""Register a resolved secret value for log redaction (best-effort).
@@ -41,7 +76,12 @@ class AgentConfigSchema(BaseModel):
"""Schema for agent configuration in bot.yaml."""
name: str = "assistant"
instructions: str = ""
model: Optional[str] = "gpt-4o-mini"
# Provider-agnostic by default (Issue #5609). ``None`` / ``"auto"`` resolve
# at load time to the model for whichever provider credential is actually
# present (via ``resolved_model()``), so an operator with only an
# Anthropic/Gemini/Ollama key is not silently handed an OpenAI model. An
# explicit ``model:`` always wins.
model: Optional[str] = None
llm: Optional[str] = None # Alias for model
memory: bool = False
tools: List[str] = Field(default_factory=list)
@@ -58,6 +98,19 @@ class AgentConfigSchema(BaseModel):
self.model = self.llm
return self
def resolved_model(self) -> str:
"""Return the effective model, detecting the provider when unset.
An explicit, non-``"auto"`` ``model`` (or ``llm`` alias) always wins.
When the model is unset or ``"auto"``, resolve it to the detected
provider's default (Issue #5609), with ``gpt-4o-mini`` as the
last-resort fallback handled inside ``resolve_model_by_provider``.
"""
model = self.model
if model and model.strip().lower() != "auto":
return model
return resolve_model_by_provider()
class SessionResetConfigSchema(BaseModel):
"""Schema for session reset policy configuration."""
@@ -1506,6 +1559,184 @@ class GatewayConfigSchema(BaseModel):
BotYamlSchema = GatewayConfigSchema
def _allow_integer_owner_id(node: Any) -> None:
"""Widen an ``owner_user_id`` property to also accept an integer (#5414).
``GatewayConfigSchema``/``ChannelConfigSchema`` type ``owner_user_id`` as
``Optional[str]`` but a ``mode="before"`` validator (``_coerce_owner_user_id``)
accepts an unquoted numeric YAML id (``owner_user_id: 987654321``) by casting
it to ``str``. Pydantic emits the schema from the *field type* only, so the
published schema would reject a config the runtime explicitly accepts. Mirror
the runtime coercion by adding ``integer`` to the allowed types so editor/CI
validation agrees with the gateway.
"""
if not isinstance(node, dict):
return
props = node.get("properties")
if not isinstance(props, dict):
return
owner = props.get("owner_user_id")
if not isinstance(owner, dict):
return
variants = owner.get("anyOf")
if isinstance(variants, list):
has_int = any(
isinstance(v, dict) and v.get("type") == "integer" for v in variants
)
if not has_int:
variants.append({"type": "integer"})
elif owner.get("type") == "string":
owner.pop("type", None)
owner["anyOf"] = [{"type": "string"}, {"type": "integer"}]
def _allow_env_ref_list(node: Any, field_names: tuple) -> None:
"""Allow an array-typed field to also be a single ``${ENV}`` string (#5414).
``${VAR}`` references are expanded *before* validation at load time, so an
operator may author ``allowed_users: ${TELEGRAM_ALLOWED_USERS}`` and have it
expand to a list. Editors validate the *unexpanded* YAML, so the published
schema must accept the string form for these fields too; otherwise it
false-flags a config the runtime happily accepts.
"""
if not isinstance(node, dict):
return
props = node.get("properties")
if not isinstance(props, dict):
return
for name in field_names:
field = props.get(name)
if not isinstance(field, dict):
continue
if field.get("type") == "array":
array_variant = {k: v for k, v in field.items() if k not in ("default", "title")}
new_field: Dict[str, Any] = {
"anyOf": [array_variant, {"type": "string"}]
}
if "default" in field:
new_field["default"] = field["default"]
if "title" in field:
new_field["title"] = field["title"]
props[name] = new_field
elif isinstance(field.get("anyOf"), list):
variants = field["anyOf"]
if not any(
isinstance(v, dict) and v.get("type") == "string" for v in variants
):
variants.append({"type": "string"})
def gateway_config_json_schema() -> Dict[str, Any]:
"""Canonical JSON Schema for ``gateway.yaml`` / ``bot.yaml`` (draft-07).
Derived directly from :class:`GatewayConfigSchema` (Pydantic's
``model_json_schema``) and decorated with the standard ``$schema``/``$id``/
``title`` metadata so the artefact is self-describing and mirrors the
``agents.yaml`` convention (``generate_agents_schema``). This is the single
source of truth surfaced to editors (via the ``# yaml-language-server``
header) and to CI linting, so a mistyped key is flagged at author time
instead of only at gateway startup.
The raw ``model_json_schema()`` is post-processed so the *published* schema
faithfully reflects what the runtime actually validates (#5414):
- ``gateway`` / ``hooks`` are stored as opaque ``Dict``/``List[Dict]`` on the
model for downstream dict-style access, but validated field-by-field at
load time via :class:`GatewayServerSchema` / :class:`HookSchema`
(``extra="forbid"`` for the server block). Inline those typed schemas so a
misspelled server knob (``drain_timout``) or hook key is caught at author
time, not only at startup.
- ``owner_user_id`` (top-level and per-channel) also accepts an integer,
mirroring the runtime ``str(...)`` coercion of unquoted numeric ids.
- Unknown top-level keys are rejected so a typo (``auto_enable_from_en``)
that the runtime silently ignores is surfaced while authoring. Channel
configs keep ``additionalProperties`` open for plugin-declared keys.
"""
base = GatewayConfigSchema.model_json_schema()
defs = base.setdefault("$defs", {})
# Inline the typed gateway-server + hook schemas so the published schema
# carries the same field-by-field constraints the runtime enforces, instead
# of the opaque ``{"type": "object"}`` Pydantic emits for ``Dict[str, Any]``.
server_schema = GatewayServerSchema.model_json_schema(
ref_template="#/$defs/{model}"
)
for name, definition in server_schema.pop("$defs", {}).items():
defs.setdefault(name, definition)
defs["GatewayServerSchema"] = server_schema
hook_schema = HookSchema.model_json_schema(ref_template="#/$defs/{model}")
for name, definition in hook_schema.pop("$defs", {}).items():
defs.setdefault(name, definition)
defs["HookSchema"] = hook_schema
props = base.get("properties", {})
if "gateway" in props:
props["gateway"] = {
"anyOf": [{"$ref": "#/$defs/GatewayServerSchema"}, {"type": "null"}],
"default": None,
"title": props["gateway"].get("title", "Gateway"),
}
if "hooks" in props:
props["hooks"] = {
"anyOf": [
{"type": "array", "items": {"$ref": "#/$defs/HookSchema"}},
{"type": "null"},
],
"default": None,
"title": props["hooks"].get("title", "Hooks"),
}
# Accept unquoted numeric owner ids wherever they appear (top level +
# ChannelConfigSchema), matching the runtime coercion.
_allow_integer_owner_id(base)
channel_def = defs.get("ChannelConfigSchema")
if isinstance(channel_def, dict):
_allow_integer_owner_id(channel_def)
# ``${VAR}`` env references are substituted BEFORE validation at load time
# (``validate_gateway_config``), so a list-typed field may legitimately be
# authored as a single ``${ENV}`` string that expands to a list. The
# scaffolder writes ``allowed_users: ${TELEGRAM_ALLOWED_USERS}``, so an
# editor validating the *authored* (unexpanded) YAML must accept the
# string form too — otherwise the schema false-flags a config the runtime
# accepts. Widen the array-typed env-ref channel fields accordingly.
_allow_env_ref_list(channel_def, ("allowed_users", "allowlist", "blocklist"))
# Reject unknown top-level keys so an author-time typo is caught (the runtime
# silently ignores it, keeping the mistyped opt-out ineffective). Channels
# keep ``additionalProperties`` open for plugin-declared config keys.
#
# ``config_version`` is not a ``GatewayConfigSchema`` field — it is stamped at
# the top level by the doctor/start config-version machinery (and shipped in
# the sample ``gateway.yaml``). Declare it so the stricter schema accepts a
# stamped/migrated config instead of false-flagging it.
base.setdefault("properties", {}).setdefault(
"config_version",
{
"type": "integer",
"title": "Config Version",
"description": (
"Stamped by `praisonai gateway doctor --fix` / `start` to track "
"config migrations. Managed automatically."
),
},
)
base.setdefault("additionalProperties", False)
schema: Dict[str, Any] = {
"$schema": "http://json-schema.org/draft-07/schema#",
"$id": GATEWAY_SCHEMA_URL,
**base,
"title": "PraisonAI Gateway / Bot Configuration",
"description": (
"Schema for gateway.yaml / bot.yaml consumed by the PraisonAI "
"gateway and single-bot runtime (gateway server, agents, channels, "
"routing, schedules, hooks, streaming and delivery)."
),
}
return schema
def validate_gateway_config(raw: Dict[str, Any], apply_env_substitution: bool = True) -> GatewayConfigSchema:
"""Validate a raw YAML dict against the unified gateway config schema.
@@ -105,6 +105,78 @@ def resolve_activity_category(tool_name: Optional[str]) -> str:
return "default"
# Separator between the reply body and the runtime footer line. A blank line
# then an em-dash keeps the footer visually distinct without markup that could
# break on a given platform.
_FOOTER_SEP = "\n\n— "
def build_footer_line(
model: Optional[str] = None,
served_model: Optional[str] = None,
context_pct: Optional[float] = None,
latency_s: Optional[float] = None,
) -> str:
"""Build a privacy-safe, per-reply runtime footer line.
Renders only the fields that are available, e.g. ``gpt-4o · 38% ctx · 4.2s``.
A missing field is omitted rather than shown as a placeholder, so a partial
footer is always well-formed. Only the model name, context-window fill
percentage and turn latency are surfaced — never tool args, URLs, paths or
message content. Returns ``""`` when no field is available.
Args:
model: Configured model id (e.g. ``"gpt-4o"``).
served_model: Deployment that actually served the turn, shown as
``model (served: X)`` when it differs from ``model`` (failover /
routing visibility).
context_pct: Session context-window fill, 0–100.
latency_s: Turn latency in seconds.
Returns:
A single footer line (without the leading separator) or ``""``.
"""
parts: list = []
if model:
model_str = str(model).strip()
if model_str:
served = str(served_model).strip() if served_model else ""
if served and served != model_str:
parts.append(f"{model_str} (served: {served})")
else:
parts.append(model_str)
if context_pct is not None:
try:
pct = float(context_pct)
if pct == pct and pct not in (float("inf"), float("-inf")):
parts.append(f"{int(round(pct))}% ctx")
except (TypeError, ValueError):
pass
if latency_s is not None:
try:
secs = float(latency_s)
if secs == secs and secs >= 0 and secs not in (float("inf"), float("-inf")):
parts.append(f"{secs:.1f}s")
except (TypeError, ValueError):
pass
return " · ".join(parts)
def append_footer(reply: str, footer_line: str) -> str:
"""Append a footer line to a final reply, separated by a blank line.
Returns ``reply`` unchanged when ``footer_line`` is empty. Only ever used on
the final reply — never on partial/streamed frames.
"""
if not footer_line:
return reply
return f"{reply}{_FOOTER_SEP}{footer_line}"
def strip_reasoning_tags(text: str) -> str:
"""Remove ``<think>``/``<reasoning>`` spans from streamed content.
@@ -201,6 +273,25 @@ class StreamingConfig:
)
def build_streaming_config(config: Any) -> Optional["StreamingConfig"]:
"""Derive a :class:`StreamingConfig` from a ``BotConfig``.
Shared by every channel adapter (Telegram, Discord, Slack, …) so a
configured ``streaming`` flag is wired to the same ``DraftStreamer`` engine
identically everywhere, instead of each adapter re-deriving (or silently
ignoring) the knob. Returns ``None`` when streaming is off so the caller's
non-streaming path is unchanged.
"""
if not getattr(config, "streaming", False):
return None
interval_ms = getattr(config, "stream_edit_interval_ms", 700) or 700
return StreamingConfig(
mode=StreamingMode.DRAFT,
min_interval=interval_ms / 1000.0,
min_delta=50,
)
class BotAdapter(Protocol):
"""Protocol for bot adapters that support message editing."""
@@ -45,6 +45,12 @@ from ._commands import (
)
from ._failure import failure_reply_text
from ._session import BotSessionManager
from ._streaming import (
StreamingConfig,
StreamingMode,
DraftStreamer,
build_streaming_config,
)
from ._debounce import InboundDebouncer
from ._ack import AckReactor
from ._unknown_user import UnknownUserHandler, BotContext
@@ -102,7 +108,12 @@ class DiscordBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
# Initialize allow_silence from config
self._allow_silence = getattr(self.config, 'allow_silence', False)
# Initialize streaming config based on BotConfig (shared with Telegram /
# Slack so ``streaming.mode: draft|auto`` is honoured identically across
# channels that support in-place edits — Issue #5415).
self._streaming_config = build_streaming_config(self.config)
self._is_running = False
self._bot_user: Optional[BotUser] = None
self._client = None
@@ -139,6 +150,15 @@ class DiscordBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
self._register_interactive_handlers()
self._bot_context: Optional[BotContext] = None
def configure_streaming(self, config: StreamingConfig) -> None:
"""Configure streaming reply mode.
Args:
config: Streaming configuration. Set mode=StreamingMode.OFF to disable.
"""
self._streaming_config = config
logger.debug("DiscordBot: streaming configured, mode=%s", config.mode)
@property
def is_running(self) -> bool:
return self._is_running
@@ -474,20 +494,95 @@ class DiscordBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
async def _send_agent_response():
text_to_send = await self._debouncer.debounce(user_id, bot_message.text)
channel_id = str(message.channel.id)
# Streaming path: progressively edit a placeholder in
# place using the shared DraftStreamer, mirroring Telegram
# (Issue #5415). Only engaged when streaming is configured.
streaming_enabled = (
self._streaming_config is not None
and self._streaming_config.mode != StreamingMode.OFF
)
if streaming_enabled:
streamer = DraftStreamer(
adapter=self,
channel_id=channel_id,
config=self._streaming_config,
platform="discord",
)
placeholder_message_id = await streamer.start()
try:
response = await self._session.chat(
self._agent, user_id, text_to_send,
chat_id=channel_id,
user_name=str(getattr(message.author, "name", "")),
message_id=str(message.id),
account=getattr(self.config, "account", "default"),
stream_callback=streamer.on_event,
)
send_result = self.fire_message_sending(
channel_id, str(response),
)
if send_result["cancel"]:
try:
await self.delete_message(
channel_id, placeholder_message_id
)
except Exception:
pass
return
final_content = send_result["content"]
# A streamed answer longer than Discord's hard
# cap cannot be delivered by editing the single
# placeholder (both the edit and the send
# fallback 400 on >2000 chars), which would drop
# the whole reply. Fall back to the chunked
# reply path so long answers are split and keep
# the reply reference (Greptile #3 / #5).
_cap = min(self.config.max_message_length, 2000)
if len(final_content) > _cap:
try:
await self.delete_message(
channel_id, placeholder_message_id
)
except Exception:
pass
await self._send_long_message(
message.channel, final_content,
reference=message,
)
else:
await streamer.finalize(final_content)
self.fire_message_sent(channel_id, final_content)
except Exception:
try:
await self.delete_message(
channel_id, placeholder_message_id
)
except Exception:
pass
raise
if ack_ctx:
await self._ack.done(
ack_ctx, react_fn=_discord_react,
unreact_fn=_discord_unreact,
)
return
response = await self._session.chat(
self._agent, user_id, text_to_send,
chat_id=str(message.channel.id),
chat_id=channel_id,
user_name=str(getattr(message.author, "name", "")),
message_id=str(message.id),
account=getattr(self.config, "account", "default"),
)
send_result = self.fire_message_sending(
str(message.channel.id), str(response),
channel_id, str(response),
)
if send_result["cancel"]:
return
await self._send_long_message(message.channel, send_result["content"], reference=message)
self.fire_message_sent(str(message.channel.id), send_result["content"])
self.fire_message_sent(channel_id, send_result["content"])
# Done reaction - show completion
if ack_ctx:
await self._ack.done(ack_ctx, react_fn=_discord_react, unreact_fn=_discord_unreact)
File diff suppressed because it is too large. Load diff
+152 -1
View File
@@ -71,6 +71,12 @@ from ._commands import (
build_command_access_policy,
)
from ._session import BotSessionManager
from ._streaming import (
StreamingConfig,
StreamingMode,
DraftStreamer,
build_streaming_config,
)
from ._debounce import InboundDebouncer
from ._ack import AckReactor
from ._unknown_user import UnknownUserHandler, BotContext
@@ -144,7 +150,12 @@ class SlackBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
# Initialize allow_silence from config
self._allow_silence = getattr(self.config, 'allow_silence', False)
# Initialize streaming config based on BotConfig (shared with Telegram /
# Discord so ``streaming.mode: draft|auto`` is honoured identically
# across channels that support in-place edits — Issue #5415).
self._streaming_config = build_streaming_config(self.config)
self._is_running = False
self._bot_user: Optional[BotUser] = None
self._app = None
@@ -186,6 +197,111 @@ class SlackBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
# Audio capabilities
self._stt_enabled: bool = False
def configure_streaming(self, config: StreamingConfig) -> None:
"""Configure streaming reply mode.
Args:
config: Streaming configuration. Set mode=StreamingMode.OFF to disable.
"""
self._streaming_config = config
logger.debug("SlackBot: streaming configured, mode=%s", config.mode)
def _reply_needs_thread(self, thread_id: str) -> bool:
"""Whether a reply must be posted into a thread rather than the root.
The shared ``DraftStreamer`` posts/edits its placeholder at the channel
root (it only knows ``send_message(channel, text)``), so when a reply
belongs in a thread — the inbound message was itself in a thread, or
``reply_in_thread`` is configured — streaming in place would strand the
answer in the main channel (Greptile #2 / #5417). In that case the
caller falls back to the existing threaded send path.
"""
return bool(thread_id) or bool(getattr(self.config, "reply_in_thread", False))
async def _maybe_stream_reply(
self,
channel_id: str,
user_id: str,
text: str,
*,
thread_id: str = "",
message_id: str = "",
) -> Optional[str]:
"""Progressively stream a reply via DraftStreamer when configured.
Returns the final (hook-processed) response text when the streaming
path handled delivery, or ``None`` when streaming is disabled — or when
the reply must be threaded (the channel-agnostic streamer can only post
at the channel root) — so the caller falls back to its normal
single-message, thread-aware, media-aware send (Issue #5415).
"""
streaming_enabled = (
self._streaming_config is not None
and self._streaming_config.mode != StreamingMode.OFF
)
if not streaming_enabled:
return None
# Defer to the standard send path when the reply must land in a thread:
# the shared streamer posts/edits at the channel root and would
# otherwise strand a threaded answer in the main channel (Greptile #2).
if self._reply_needs_thread(thread_id):
return None
streamer = DraftStreamer(
adapter=self,
channel_id=channel_id,
config=self._streaming_config,
platform="slack",
)
placeholder_message_id = await streamer.start()
try:
response = await self._session.chat(
self._agent, user_id, text,
chat_id=channel_id,
thread_id=thread_id,
message_id=message_id,
account=getattr(self.config, "account", "default"),
stream_callback=streamer.on_event,
)
send_result = self.fire_message_sending(channel_id, str(response))
if send_result["cancel"]:
try:
await self.delete_message(channel_id, placeholder_message_id)
except Exception:
pass
return ""
# Extract any MEDIA: directives before finalizing so an audio
# reply is uploaded via Slack's file path instead of being edited
# into the placeholder as literal "MEDIA:/path" text (Greptile #4).
parsed = split_media_from_output(send_result["content"])
text_content = parsed["text"]
media_urls = parsed.get("media_urls", [])
await streamer.finalize(text_content if text_content else send_result["content"])
for media_path in media_urls:
if not os.path.exists(media_path):
logger.warning(f"Media file not found: {media_path}")
continue
if is_audio_file(media_path) and self._client:
try:
await self._client.files_upload_v2(
channel=channel_id,
file=media_path,
title=os.path.basename(media_path),
)
except Exception as e:
logger.error(f"Failed to upload audio to Slack: {e}")
return send_result["content"]
except Exception:
try:
await self.delete_message(channel_id, placeholder_message_id)
except Exception:
pass
raise
def enable_stt(self, enabled: bool = True) -> None:
"""Enable STT for audio file transcription."""
self._stt_enabled = enabled
@@ -622,6 +738,26 @@ class SlackBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
user_id = event.get("user", "unknown")
logger.info(f"Message received: {text[:100]}...")
text = await self._debouncer.debounce(user_id, text)
# Streaming path: progressively edit a placeholder in place
# using the shared DraftStreamer, mirroring Telegram/Discord
# (Issue #5415). Returns None when streaming is disabled.
streamed = await self._maybe_stream_reply(
str(channel_id) if channel_id else "",
user_id, text,
thread_id=event.get("thread_ts", "") or "",
message_id=event.get("ts", ""),
)
if streamed is not None:
if streamed:
self.fire_message_sent(channel_id, streamed)
if ack_ctx and self._client:
await self._ack.done(
ack_ctx, react_fn=_slack_react,
unreact_fn=_slack_unreact,
)
return
response = await self._session.chat(
self._agent, user_id, text,
chat_id=str(channel_id) if channel_id else "",
@@ -707,6 +843,21 @@ class SlackBot(OutboundResilienceMixin, ChatCommandMixin, MessageHookMixin):
try:
user_id = event.get("user", "unknown")
logger.info(f"@mention received: {text[:100]}...")
# Streaming path: progressively edit a placeholder in place
# using the shared DraftStreamer (Issue #5415). Returns None
# when streaming is disabled so the plain path runs.
streamed = await self._maybe_stream_reply(
str(event.get("channel", "")),
user_id, text,
thread_id=event.get("thread_ts", "") or "",
message_id=event.get("ts", ""),
)
if streamed is not None:
if streamed:
self.fire_message_sent(event.get("channel", ""), streamed)
return
response = await self._session.chat(
self._agent, user_id, text,
chat_id=str(event.get("channel", "")),
@@ -64,7 +64,13 @@ from ._album import (
from ._ack import AckReactor
from ._unknown_user import UnknownUserHandler, BotContext
from ._pairing_ui import PairingUIBuilder, PairingCallbackHandler
from ._streaming import StreamingConfig, StreamingMode, DraftStreamer
from ._streaming import (
StreamingConfig,
StreamingMode,
DraftStreamer,
build_footer_line,
append_footer,
)
from ._rate_limit import RateLimiter
from ._resilience import deliver_with_retry, BackoffPolicy, TELEGRAM_BACKOFF
from ._dlq import OutboundDLQ
@@ -251,6 +257,34 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
"""
self._streaming_config = config
logger.debug("TelegramBot: streaming configured, mode=%s", config.mode)
def _footer_line(self, turn_started_at: Optional[float]) -> str:
"""Build the opt-in runtime footer line for a final reply.
Returns ``""`` (no footer) unless ``config.footer`` is enabled. The line
is privacy-safe: it surfaces only the configured model name and turn
latency, omitting any field that is unavailable. Best-effort — a build
error never blocks the reply.
The line is appended to the *text* of the final reply only (never to
interim streamed frames, and never to the text handed to voice/TTS or
media synthesis), so voice notes don't narrate the footer.
"""
if not getattr(self.config, "footer", False):
return ""
try:
model = None
agent = self._agent
if agent is not None:
llm = getattr(agent, "llm", None)
model = llm if isinstance(llm, str) else None
latency = None
if turn_started_at is not None:
latency = max(0.0, time.monotonic() - turn_started_at)
return build_footer_line(model=model, latency_s=latency)
except Exception as e: # pragma: no cover — never block a reply on footer
logger.debug("footer build skipped: %s", e)
return ""
def enable_stt(self, enabled: bool = True) -> None:
"""Enable STT for voice message transcription."""
@@ -705,6 +739,10 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
message.content = merged.caption
try:
# Mark the turn start so the opt-in runtime footer can report
# latency ("inbound received -> reply delivered"). Cheap and
# unused unless config.footer is enabled.
turn_started_at = time.monotonic()
message_text = await self._debouncer.debounce(user_id, message.content)
# Render any resolved reply/quote context (Issue #5223) as a
@@ -774,8 +812,15 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
text_content = parsed["text"]
media_urls = parsed.get("media_urls", [])
# Finalize with text content (after hook processing and media extraction)
await streamer.finalize(text_content if text_content else send_result["content"])
# Finalize with text content (after hook processing and media extraction).
# Append the opt-in runtime footer to the FINAL reply only
# (never to interim streamed frames, and never to the
# voice-synthesis text below — matching the non-streaming path).
final_text = text_content if text_content else send_result["content"]
final_text = append_footer(
final_text, self._footer_line(turn_started_at)
)
await streamer.finalize(final_text)
# Issue #3623: the streamed reply is delivered as text
# by the streamer, so the outbound voice-reply policy
@@ -879,6 +924,8 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
)
if send_result["cancel"]:
return
# Opt-in runtime footer: appended to the final text reply
# only (never to voice/TTS or media synthesis).
await self._send_response_with_media(
update.message.chat_id,
send_result["content"],
@@ -886,6 +933,7 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
inbound_was_voice=bool(
update.message.voice or update.message.audio
),
footer_line=self._footer_line(turn_started_at),
)
# If the agent attached a portable presentation
# (buttons/menus), render it natively as an inline
@@ -965,6 +1013,10 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
or ""
) if update.message.from_user else ""
try:
# Mark turn start so the opt-in footer can report latency,
# matching the normal message path. Cheap, unused unless
# config.footer is enabled.
cmd_turn_started_at = time.monotonic()
response = await self._session.chat(
self._agent, user_id, prompt,
chat_id=str(update.message.chat_id) if update.message.chat_id else "",
@@ -987,6 +1039,7 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
update.message.chat_id,
send_result["content"],
reply_to=update.message.message_id,
footer_line=self._footer_line(cmd_turn_started_at),
)
try:
if presentation is not None:
@@ -1832,23 +1885,37 @@ class TelegramBot(ChatCommandMixin, MessageHookMixin):
response: str,
reply_to: Optional[int] = None,
inbound_was_voice: bool = False,
footer_line: str = "",
) -> None:
"""Send response, extracting and sending any MEDIA: files."""
"""Send response, extracting and sending any MEDIA: files.
``footer_line`` (opt-in, privacy-safe runtime footer) is appended to the
delivered *text* only — never to the text handed to voice/TTS synthesis —
so a voice note never narrates the model/latency footer and the
streaming and non-streaming paths behave identically.
"""
# Parse response for media
parsed = split_media_from_output(response)
text = parsed["text"]
media_urls = parsed.get("media_urls", [])
audio_as_voice = parsed.get("audio_as_voice", False)
# Send text first if present
# Send text first if present. The footer attaches to the final text
# reply only; the pre-footer ``text`` is what feeds voice synthesis.
if text:
await self._send_long_message(chat_id, text, reply_to=reply_to)
await self._send_long_message(
chat_id, append_footer(text, footer_line), reply_to=reply_to
)
elif footer_line:
# Media-only reply: never emit a footer-only text bubble.
pass
# Issue #3623: outbound voice reply. When the operator has enabled a
# ``voice`` policy, synthesise the agent's plain text and deliver it as
# a native Telegram voice note — the symmetric counterpart to inbound
# STT. The agent authors plain text; voice is a transport concern here.
# Skipped silently (text already sent) when disabled or synthesis fails.
# Footer is intentionally excluded from the synthesised text.
await self._maybe_send_voice_reply(
chat_id, text, inbound_was_voice=inbound_was_voice
)
@@ -570,6 +570,7 @@ def gateway_status(
daemon_only: bool = typer.Option(False, "--daemon-only", help="Show only daemon status"),
deep: bool = typer.Option(False, "--deep", help="Extended diagnostics (health + log tail)"),
probe: bool = typer.Option(False, "--probe", help="Live credential probe per channel"),
json_output: bool = typer.Option(False, "--json", help="Emit a machine-readable status snapshot"),
):
"""Check gateway status and daemon service status.
@@ -579,6 +580,7 @@ def gateway_status(
praisonai gateway status --deep --config bot.yaml
praisonai gateway status --probe --config bot.yaml
praisonai gateway status --daemon-only
praisonai gateway status --json
"""
import os
from ..features.gateway import GatewayHandler
@@ -597,6 +599,17 @@ def gateway_status(
if config is None:
config = _resolve_gateway_config_path(None)
# Machine-readable path (#5363): sibling commands (doctor/test) support
# --json; status did not, so scripts/health-checks/dashboards could not
# consume its state. Emit a single stable snapshot built from /health +
# /info and the daemon status, then return before any human-only text.
if json_output:
import json as _json
snapshot = _build_status_snapshot(host, port)
typer.echo(_json.dumps(snapshot, default=str))
raise typer.Exit(0 if snapshot.get("reachable") else 1)
output = get_output_controller()
# Show daemon status
@@ -873,6 +886,122 @@ def _print_secret_availability(report: dict) -> None:
print()
def _resolve_gateway_auth_token() -> str:
"""Resolve the loopback ``GATEWAY_AUTH_TOKEN`` from env or persisted ``.env``.
``praisonai onboard`` / the daemon persist the auto-generated token to
``~/.praisonai/.env`` rather than exporting it into every shell. Without
reading that file, an authenticated gateway answers ``/health`` (public)
but 401s ``/info`` (authenticated), so ``status --json`` would report
``reachable: true`` while silently omitting ``version`` (#5363, greptile
P1). Prefer an already-exported env value; otherwise fall back to the
persisted file. Returns ``""`` when no token is available.
"""
import os
token = os.environ.get("GATEWAY_AUTH_TOKEN", "").strip()
if token:
return token
env_path = os.environ.get("PRAISONAI_ENV_FILE") or os.path.expanduser(
"~/.praisonai/.env"
)
try:
with open(env_path, encoding="utf-8", errors="replace") as handle:
for raw in handle:
s = raw.strip()
if not s or s.startswith("#") or "=" not in s:
continue
key, _, val = s.partition("=")
if key.strip() == "GATEWAY_AUTH_TOKEN":
return val.strip().strip('"').strip("'")
except OSError:
pass
return ""
def _fetch_gateway_json(host: str, port: int, path: str, timeout: float = 5.0):
"""GET a JSON gateway endpoint over loopback, returning None on failure.
Attaches the loopback-only bearer token (resolved from the environment or
the persisted ``~/.praisonai/.env``) so authenticated gateways still
answer, without leaking the credential onto a remote plaintext hop.
"""
import json
import urllib.request
url = f"http://{host}:{port}{path}"
try:
req = urllib.request.Request(url)
token = _resolve_gateway_auth_token()
if token and host in ("127.0.0.1", "localhost", "::1"):
req.add_header("Authorization", f"Bearer {token}")
with urllib.request.urlopen(req, timeout=timeout) as response:
return json.loads(response.read().decode())
except Exception:
return None
def _build_status_snapshot(host: str, port: int) -> dict:
"""Assemble one machine-readable gateway status snapshot (#5363).
Combines the daemon service state with the rich ``/health`` payload
(uptime, sessions, clients, per-channel supervision, degraded owners,
delivery back-pressure, applied/on-disk config revision + drift) and the
version/protocol_version from ``/info`` so a single ``--json`` call answers
"is my gateway healthy, on the right config, and what version".
"""
from praisonai_bot.daemon import get_daemon_status
snapshot: dict = {"host": host, "port": port, "reachable": False}
try:
daemon_status = get_daemon_status()
snapshot["daemon"] = {
"platform": daemon_status.get("platform", "unknown"),
"installed": bool(daemon_status.get("installed", False)),
"running": bool(daemon_status.get("running", False)),
"pid": daemon_status.get("pid"),
"error": daemon_status.get("error"),
}
except Exception as exc: # noqa: BLE001
snapshot["daemon"] = {"error": str(exc)}
health = _fetch_gateway_json(host, port, "/health")
info = _fetch_gateway_json(host, port, "/info")
if info:
snapshot["version"] = info.get("version")
snapshot["protocol_version"] = info.get("protocol_version")
if health:
snapshot["reachable"] = True
snapshot["status"] = health.get("status")
snapshot["uptime_s"] = health.get("uptime")
snapshot["active_sessions"] = health.get("sessions")
snapshot["clients"] = health.get("clients")
snapshot["agents"] = health.get("agents")
snapshot["channels"] = health.get("channels") or {}
snapshot["degraded"] = health.get("degraded_owners") or []
snapshot["applied_config_revision"] = health.get("applied_config_revision")
snapshot["on_disk_config_revision"] = health.get("on_disk_config_revision")
snapshot["config_drift"] = health.get("config_drift")
snapshot["last_inbound_at"] = health.get("last_inbound_at")
if "pressure" in health:
snapshot["delivery"] = health.get("pressure")
elif "delivery" in health:
snapshot["delivery"] = health.get("delivery")
if snapshot.get("version") is None and health.get("version") is not None:
snapshot["version"] = health.get("version")
# Reachable but /info didn't answer (typically an authenticated gateway
# whose token isn't available to this CLI) — surface it explicitly so
# consumers don't misread a silent missing "version" as "unversioned"
# (#5363, greptile P1).
if info is None and snapshot.get("version") is None:
snapshot["version_unavailable"] = "info_unreachable"
return snapshot
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())
@@ -1854,6 +1983,44 @@ def gateway_send(
raise typer.Exit(1)
@app.command("schema")
def gateway_schema(
output: Optional[str] = typer.Option(
None,
"--output",
"-o",
help="Write the JSON Schema to a file instead of stdout "
"(e.g. gateway.schema.json for editor config / CI linting).",
),
):
"""Emit the gateway.yaml / bot.yaml JSON Schema (for editors / CI).
Derived from the canonical Pydantic ``GatewayConfigSchema`` so a mistyped
key is flagged at author time instead of only at gateway startup. Point your
editor's YAML language server at the output, or lint configs in CI:
praisonai gateway schema # print to stdout
praisonai gateway schema -o gateway.schema.json # write to a file
"""
import json
from praisonai_bot.bots._config_schema import gateway_config_json_schema
schema_json = json.dumps(gateway_config_json_schema(), indent=2)
if output:
from pathlib import Path
out_path = Path(output)
try:
out_path.write_text(schema_json + "\n", encoding="utf-8")
except OSError as exc:
print(f"Failed to write {out_path}: {exc}")
raise typer.Exit(1)
print(f"Wrote gateway JSON Schema to {out_path}")
else:
sys.stdout.write(schema_json + "\n")
hooks_app = typer.Typer(
help="Manage inbound trigger hooks (POST /hooks/<path>) in gateway.yaml",
no_args_is_help=True,
@@ -293,6 +293,9 @@ class BotHandler:
agent_config = validated_config.agent
if agent_config:
agent_config_dict = agent_config.model_dump()
# Resolve an unset/"auto" model to the detected provider's default
# (Issue #5609) so a non-OpenAI operator is not handed gpt-4o-mini.
agent_config_dict["model"] = agent_config.resolved_model()
else:
agent_config_dict = {}
@@ -26,6 +26,44 @@ from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
def _detected_model() -> str:
"""Model for whichever provider credential is present (Issue #5609).
Leads onboarding with the provider the operator already has
(Anthropic/Gemini/Ollama/…) instead of writing ``gpt-4o-mini``
unconditionally. Reuses the gateway/bot config resolver (which itself
bridges to the existing ``default_model_for_available_provider`` helper),
falling back to ``gpt-4o-mini`` only when nothing is detected.
"""
try:
from praisonai_bot.bots._config_schema import resolve_model_by_provider
return resolve_model_by_provider()
except Exception: # pragma: no cover - detection is best-effort
return "gpt-4o-mini"
def _provider_key_hint(model: str) -> str:
"""Name the API-key env var the detected model needs (Issue #5609).
Maps the detected/configured model back to its provider's credential env
var (e.g. ``ANTHROPIC_API_KEY`` for ``anthropic/...``) so the failure hint
points at the key the operator actually needs instead of always
``OPENAI_API_KEY``. Falls back to a generic hint when the provider cannot
be resolved.
"""
try:
from praisonai_bot._code_bridge import import_code_module
env = import_code_module("praisonai_code.llm.env")
key_var, _base = env._provider_from_model(model)
if key_var:
return f"e.g. {key_var}"
except Exception: # pragma: no cover - best-effort hint
pass
return "e.g. OPENAI_API_KEY"
def _is_non_interactive() -> bool:
"""Return True when the wizard must not block on user input.
@@ -280,8 +318,13 @@ def _generate_bot_yaml_multi_channel(channels: Dict[str, Dict], agent_name: str
instructions = agent_instructions or "You are a helpful AI assistant."
instructions = instructions.rstrip('. ')
instructions_escaped = instructions.replace('\\', '\\\\').replace('"', '\\"')
# Lead with the provider the operator already has (Issue #5609).
model = _detected_model()
lines = ["# Generated by `praisonai onboard`. Multi-channel configuration.",
from praisonai_bot.bots._config_schema import GATEWAY_SCHEMA_HEADER
lines = [GATEWAY_SCHEMA_HEADER.rstrip("\n"),
"# Generated by `praisonai onboard`. Multi-channel configuration.",
"# Compatible with both `praisonai bot start` and `praisonai gateway start`.",
"# Each channel can route to different agents.",
""]
@@ -296,7 +339,7 @@ def _generate_bot_yaml_multi_channel(channels: Dict[str, Dict], agent_name: str
lines.append("agent:")
lines.append(f' name: "{agent_name}"')
lines.append(f' instructions: "{instructions_escaped}. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."')
lines.append(' model: "gpt-4o-mini"')
lines.append(f' model: "{model}"')
lines.append("")
# Multi-agent block with agents for each role
@@ -310,7 +353,7 @@ def _generate_bot_yaml_multi_channel(channels: Dict[str, Dict], agent_name: str
lines.append(f' instructions: "{instructions_escaped}. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."')
else:
lines.append(f' instructions: "You are a {role} agent. {instructions_escaped} Use your tools proactively to help with {role}-related tasks."')
lines.append(' model: gpt-4o-mini')
lines.append(f' model: {model}')
lines.append(" memory: false")
lines.append("")
agents_added.add(role)
@@ -379,8 +422,13 @@ def _generate_bot_yaml(platforms: List[str], agent_name: str = "assistant", agen
instructions = agent_instructions or "You are a helpful AI assistant."
instructions = instructions.rstrip('. ') # Remove trailing periods to avoid double periods
instructions_escaped = instructions.replace('\\', '\\\\').replace('"', '\\"') # Escape quotes and backslashes
# Lead with the provider the operator already has (Issue #5609).
model = _detected_model()
lines = ["# Generated by `praisonai onboard`. Works with both",
from praisonai_bot.bots._config_schema import GATEWAY_SCHEMA_HEADER
lines = [GATEWAY_SCHEMA_HEADER.rstrip("\n"),
"# Generated by `praisonai onboard`. Works with both",
"# `praisonai bot start` (single-bot) and `praisonai gateway start`",
"# (multi-bot daemon exposing /health on 8765).",
""]
@@ -395,7 +443,7 @@ def _generate_bot_yaml(platforms: List[str], agent_name: str = "assistant", agen
lines.append("agent:")
lines.append(f' name: "{agent_name}"')
lines.append(f' instructions: "{instructions_escaped}. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."')
lines.append(' model: "gpt-4o-mini"')
lines.append(f' model: "{model}"')
lines.append(" # tools: [] # Uncomment to override auto-injected defaults")
lines.append(" # (search_web, schedule_*, store_memory, search_memory, …)")
lines.append("")
@@ -408,7 +456,7 @@ def _generate_bot_yaml(platforms: List[str], agent_name: str = "assistant", agen
lines.append(f' instructions: "{instructions_escaped}. You can search the web, manage schedules, and remember past conversations. Use your tools proactively."')
else:
lines.append(f' instructions: "You are the {name} agent. {instructions_escaped}. Use your tools proactively."')
lines.append(' model: gpt-4o-mini')
lines.append(f' model: {model}')
lines.append(" memory: false")
lines.append(" # tools: [] # Override auto-injected defaults here")
lines.append("")
@@ -1008,9 +1056,14 @@ class OnboardWizard:
if ok:
print_fn(" [green]✓[/green] Agent replied — model credentials work")
else:
# Name the provider actually detected (Issue #5609) instead of
# hardcoding OPENAI_API_KEY, so the remediation hint matches the
# model onboarding chose (or says none was detected).
model = _detected_model()
key_hint = _provider_key_hint(model)
print_fn(
f" [yellow]⚠[/yellow] Agent did not reply: {detail}\n"
" [dim]Set the model/provider API key (e.g. OPENAI_API_KEY) in "
f" [dim]Set the model/provider API key ({key_hint}) in "
"~/.praisonai/.env, then re-run [cyan]praisonai onboard[/cyan] or "
"[cyan]praisonai gateway test --turn ping[/cyan].[/dim]"
)
@@ -867,10 +867,14 @@ class GatewayApiEndpoints:
)
if method == "initialize":
from .server import _installed_gateway_version
return ok(
{
"protocolVersion": "2024-11-05",
"serverInfo": {"name": "PraisonAI Gateway", "version": "1.0.0"},
"serverInfo": {
"name": "PraisonAI Gateway",
"version": _installed_gateway_version(),
},
"capabilities": {"tools": {}},
}
)
@@ -273,10 +273,18 @@ async def run_turn_test(
# the SAME provider the live gateway will use — otherwise a valid custom
# base_url/api_key config tests the default provider and gives a misleading
# verdict (false abort, or invalid runtime creds passing).
# An unset or ``"auto"`` model resolves to the detected provider's default
# (Issue #5609) so the preflight tests the provider the operator actually
# has instead of assuming OpenAI.
model = acfg.get("model") or acfg.get("llm")
if not model or str(model).strip().lower() == "auto":
from praisonai_bot.bots._config_schema import resolve_model_by_provider
model = resolve_model_by_provider()
agent_kwargs: Dict[str, Any] = {
"name": acfg.get("name", agent_id),
"instructions": acfg.get("instructions", ""),
"llm": acfg.get("model") or acfg.get("llm", "gpt-4o-mini"),
"llm": model,
}
base_url = acfg.get("base_url") or acfg.get("api_base")
if base_url:
@@ -131,6 +131,29 @@ from .unicode_utils import safe_error_message, safe_log_message, extract_root_ca
from .supervisor import ChannelState, ChannelSupervisor
def _installed_gateway_version() -> str:
"""Return the installed ``praisonai-bot`` version, not a hardcoded literal.
Automated version/upgrade checks read this off ``/info`` and the MCP
``serverInfo``; a hardcoded constant (#5363) actively misled them. Prefer
the packaged ``_version.py`` and fall back to installed metadata.
"""
try:
from praisonai_bot._version import __version__ as _v
if _v:
return str(_v)
except Exception:
pass
try:
from importlib.metadata import version as _pkg_version
except ImportError: # pragma: no cover - py<3.8
from importlib_metadata import version as _pkg_version # type: ignore
try:
return _pkg_version("praisonai-bot")
except Exception:
return "0.0.0"
# Per-platform token env-var fallbacks used by the generic channel-launch path
# (Issue #3578). Channels whose credentials live in the environment rather than
# gateway.yaml (email/AgentMail/Linear) resolve their token from the first env
@@ -1696,7 +1719,8 @@ class WebSocketGateway:
api_cfg = getattr(self.config, "api", None)
return JSONResponse({
"name": "PraisonAI Gateway",
"version": "1.0.0",
"version": _installed_gateway_version(),
"protocol_version": GATEWAY_PROTOCOL_VERSION,
"agents": list(self._agents.keys()),
"sessions": len(self._sessions),
"clients": len(self._clients),
@@ -8079,8 +8103,20 @@ class WebSocketGateway:
for agent_id, agent_def in agents_cfg.items():
instructions = agent_def.get("instructions", "")
# G7: Apply provider.model as fallback when agent has no model
model = agent_def.get("model", None) or default_model
# G7: Apply provider.model as fallback when agent has no model.
# An unset or ``"auto"`` model resolves to the detected provider's
# default (Issue #5609) so a non-OpenAI operator is not silently
# handed gpt-4o-mini; an explicit model/default_model still wins.
model = agent_def.get("model", None)
if not model or str(model).strip().lower() == "auto":
if default_model:
model = default_model
else:
from praisonai_bot.bots._config_schema import (
resolve_model_by_provider,
)
model = resolve_model_by_provider()
memory = agent_def.get("memory", False)
# G2: Pass temperature through (optional, SDK uses its own default)
+4 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-bot"
version = "0.1.8"
version = "0.1.9"
description = "Bots, gateway, and channel CLI for PraisonAI — messaging platforms and WebSocket control plane extracted from the praisonai wrapper."
readme = "README.md"
license = "MIT"
@@ -77,3 +77,6 @@ build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["."]
include = ["praisonai_bot*"]
[tool.setuptools.package-data]
"praisonai_bot.bots" = ["*.json"]
@@ -0,0 +1,149 @@
#!/usr/bin/env python3
"""
Tests for the opt-in, privacy-safe per-reply runtime footer (issue #5428).
The footer surfaces only model name, context % and turn latency on the FINAL
reply. Missing fields are omitted (never shown as placeholders), and nothing is
appended when the footer is empty or the feature is disabled.
"""
import asyncio
import types
from praisonai_bot.bots._streaming import append_footer, build_footer_line
def test_full_footer_line():
assert build_footer_line(
model="gpt-4o",
served_model="gpt-4o-2024-08-06",
context_pct=41,
latency_s=3.8,
) == "gpt-4o (served: gpt-4o-2024-08-06) · 41% ctx · 3.8s"
def test_model_and_latency_only():
assert build_footer_line(model="gpt-4o", latency_s=4.2) == "gpt-4o · 4.2s"
def test_served_model_hidden_when_same_as_model():
# No "(served: ...)" when the served deployment matches the configured model.
assert build_footer_line(model="gpt-4o", served_model="gpt-4o") == "gpt-4o"
def test_missing_fields_are_omitted_not_placeheld():
# Only a context % available -> just that field, no "?" placeholders.
assert build_footer_line(context_pct=38) == "38% ctx"
def test_empty_when_no_fields():
assert build_footer_line() == ""
def test_context_pct_rounded_to_int():
assert build_footer_line(context_pct=37.6) == "38% ctx"
def test_invalid_numbers_are_dropped():
assert build_footer_line(model="m", context_pct=float("nan")) == "m"
assert build_footer_line(model="m", latency_s=float("inf")) == "m"
assert build_footer_line(model="m", latency_s=-1) == "m"
def test_blank_model_is_dropped():
assert build_footer_line(model=" ", latency_s=1.0) == "1.0s"
def test_append_footer_adds_separator():
assert (
append_footer("Here is your answer.", "gpt-4o · 4.2s")
== "Here is your answer.\n\n— gpt-4o · 4.2s"
)
def test_append_footer_noop_on_empty_line():
assert append_footer("body", "") == "body"
# ---------------------------------------------------------------------------
# Handler-level delivery behaviour (opt-in wiring + voice exclusion).
#
# TelegramBot carries a heavy optional ``telegram`` dependency, so rather than
# construct the whole bot we bind the two unbound methods under test to a light
# stub that supplies just the attributes they touch. This exercises the real
# code paths (opt-in gating, model/latency capture, footer-in-text-not-voice)
# deterministically and without network or the telegram package.
# ---------------------------------------------------------------------------
from praisonai_bot.bots.telegram import TelegramBot # noqa: E402
def _make_stub(footer_enabled, llm="gpt-4o"):
stub = types.SimpleNamespace()
stub.config = types.SimpleNamespace(footer=footer_enabled)
stub._agent = types.SimpleNamespace(llm=llm)
return stub
def test_footer_line_disabled_returns_empty():
stub = _make_stub(footer_enabled=False)
line = TelegramBot._footer_line(stub, turn_started_at=None)
assert line == ""
def test_footer_line_enabled_includes_model():
stub = _make_stub(footer_enabled=True, llm="gpt-4o")
line = TelegramBot._footer_line(stub, turn_started_at=None)
assert line == "gpt-4o"
def test_footer_line_enabled_includes_model_and_latency():
import time
stub = _make_stub(footer_enabled=True, llm="gpt-4o")
started = time.monotonic() - 2.0
line = TelegramBot._footer_line(stub, turn_started_at=started)
assert line.startswith("gpt-4o · ")
assert line.endswith("s")
def test_footer_line_non_string_llm_omits_model():
stub = _make_stub(footer_enabled=True, llm=object())
line = TelegramBot._footer_line(stub, turn_started_at=None)
assert line == ""
def _run_send(stub, response, footer_line):
sent_text = {}
voiced_text = {}
async def fake_send_long(chat_id, text, reply_to=None):
sent_text["value"] = text
async def fake_voice(chat_id, text, *, inbound_was_voice=False):
voiced_text["value"] = text
stub._send_long_message = fake_send_long
stub._maybe_send_voice_reply = fake_voice
asyncio.run(
TelegramBot._send_response_with_media(
stub, 123, response, reply_to=1, footer_line=footer_line
)
)
return sent_text.get("value"), voiced_text.get("value")
def test_send_appends_footer_to_text_only():
stub = types.SimpleNamespace()
sent, voiced = _run_send(stub, "Here is your answer.", "gpt-4o · 4.2s")
# Footer appears in the delivered text …
assert sent == "Here is your answer.\n\n— gpt-4o · 4.2s"
# … but never in the text handed to voice synthesis.
assert voiced == "Here is your answer."
def test_send_without_footer_leaves_text_unchanged():
stub = types.SimpleNamespace()
sent, voiced = _run_send(stub, "Plain reply.", "")
assert sent == "Plain reply."
assert voiced == "Plain reply."
@@ -0,0 +1,193 @@
#!/usr/bin/env python3
"""Cross-channel progressive streaming wiring (Issue #5415).
The ``DraftStreamer`` engine is channel-agnostic, but it used to be wired only
into Telegram. These tests lock in that Discord and Slack now:
- derive a ``StreamingConfig`` from ``BotConfig.streaming`` via the shared
``build_streaming_config`` factory, and
- advertise ``live_edit`` so ``AUTO`` resolves to ``DRAFT``.
This guards against regressing the fix back to a Telegram-only capability.
"""
import pytest
from praisonaiagents.bots import BotConfig
from praisonai_bot.bots._streaming import (
build_streaming_config,
StreamingMode,
)
from praisonai_bot.bots.discord import DiscordBot
from praisonai_bot.bots.slack import SlackBot
def test_build_streaming_config_off_returns_none():
cfg = BotConfig(token="t", streaming=False)
assert build_streaming_config(cfg) is None
def test_build_streaming_config_draft_when_enabled():
cfg = BotConfig(token="t", streaming=True, stream_edit_interval_ms=700)
sc = build_streaming_config(cfg)
assert sc is not None
assert sc.mode == StreamingMode.DRAFT
assert sc.min_interval == 0.7
def test_discord_streaming_config_built_from_bot_config():
bot = DiscordBot(token="t", config=BotConfig(token="t", streaming=True))
assert bot._streaming_config is not None
assert bot._streaming_config.mode == StreamingMode.DRAFT
# Discord advertises live_edit so AUTO can resolve to DRAFT.
assert bot.capabilities.get("live_edit") is True
def test_discord_streaming_disabled_by_default():
bot = DiscordBot(token="t", config=BotConfig(token="t"))
assert bot._streaming_config is None
def test_slack_streaming_config_built_from_bot_config():
bot = SlackBot(token="t", config=BotConfig(token="t", streaming=True))
assert bot._streaming_config is not None
assert bot._streaming_config.mode == StreamingMode.DRAFT
assert bot.capabilities.get("live_edit") is True
def test_slack_streaming_disabled_by_default():
bot = SlackBot(token="t", config=BotConfig(token="t"))
assert bot._streaming_config is None
def test_configure_streaming_overrides_discord():
bot = DiscordBot(token="t", config=BotConfig(token="t"))
sc = build_streaming_config(BotConfig(token="t", streaming=True))
bot.configure_streaming(sc)
assert bot._streaming_config is sc
def test_configure_streaming_overrides_slack():
bot = SlackBot(token="t", config=BotConfig(token="t"))
sc = build_streaming_config(BotConfig(token="t", streaming=True))
bot.configure_streaming(sc)
assert bot._streaming_config is sc
# --- Streaming delivery guardrails (Greptile review on #5417) ----------------
@pytest.mark.asyncio
async def test_slack_stream_defers_when_threaded():
"""A threaded reply must bypass the root-only streamer (Greptile #2).
Returning ``None`` signals the caller to fall back to the thread-aware
``_send_response_with_media`` path, so a threaded answer is never stranded
in the main channel.
"""
bot = SlackBot(token="t", config=BotConfig(token="t", streaming=True))
result = await bot._maybe_stream_reply(
"C1", "U1", "hi", thread_id="123.456", message_id="123.457"
)
assert result is None
@pytest.mark.asyncio
async def test_slack_stream_defers_when_reply_in_thread_configured():
bot = SlackBot(
token="t",
config=BotConfig(token="t", streaming=True, reply_in_thread=True),
)
result = await bot._maybe_stream_reply("C1", "U1", "hi", message_id="123.457")
assert result is None
@pytest.mark.asyncio
async def test_slack_stream_uploads_media(monkeypatch, tmp_path):
"""A streamed reply carrying a MEDIA: directive uploads the file (Greptile #4)."""
from unittest.mock import AsyncMock
bot = SlackBot(token="t", config=BotConfig(token="t", streaming=True))
audio = tmp_path / "reply.mp3"
audio.write_bytes(b"ID3fake")
# Session returns text + a MEDIA directive; streamer edits are stubbed.
bot._session.chat = AsyncMock(return_value=f"Here you go\nMEDIA:{audio}")
bot.send_message = AsyncMock(return_value={"message_id": "m1"})
bot.edit_message = AsyncMock(return_value=None)
bot.delete_message = AsyncMock(return_value=True)
# Pin the outbound hook seam to a pass-through so a process-global hook
# registered by another test cannot rewrite/strip the MEDIA directive.
bot.fire_message_sending = lambda channel_id, content, **kw: {
"cancel": False,
"content": content,
}
uploads = []
class _Client:
async def files_upload_v2(self, **kwargs):
uploads.append(kwargs)
bot._client = _Client()
result = await bot._maybe_stream_reply("C1", "U1", "play it")
assert result is not None
assert uploads, "expected the MEDIA audio file to be uploaded"
assert uploads[0]["channel"] == "C1"
assert uploads[0]["file"] == str(audio)
@pytest.mark.asyncio
async def test_discord_stream_chunks_long_reply(monkeypatch):
"""A streamed answer over Discord's cap is delivered via the chunked path.
Guards Greptile #3/#5: editing the single placeholder with >2000 chars
would 400 and drop the whole reply, so the finalize path must fall back to
the reference-preserving ``_send_long_message`` chunker.
"""
from unittest.mock import AsyncMock, MagicMock
bot = DiscordBot(token="t", config=BotConfig(token="t", streaming=True))
long_reply = "x" * 5000
bot._session.chat = AsyncMock(return_value=long_reply)
bot.send_message = AsyncMock(return_value={"message_id": "m1"})
bot.edit_message = AsyncMock(return_value=None)
bot.delete_message = AsyncMock(return_value=True)
sent_chunks = []
long_sender = AsyncMock(
side_effect=lambda channel, text, reference=None: sent_chunks.append(text)
)
monkeypatch.setattr(bot, "_send_long_message", long_sender)
from praisonai_bot.bots._streaming import DraftStreamer
streamer = DraftStreamer(
adapter=bot,
channel_id="C1",
config=bot._streaming_config,
platform="discord",
)
await streamer.start()
# Simulate the finalize decision the adapter makes for an over-cap reply.
send_result = bot.fire_message_sending("C1", long_reply)
final_content = send_result["content"]
cap = min(bot.config.max_message_length, 2000)
assert len(final_content) > cap
message = MagicMock()
message.channel = MagicMock()
if len(final_content) > cap:
await bot.delete_message("C1", "m1")
await bot._send_long_message(message.channel, final_content, reference=message)
else: # pragma: no cover - defensive
await streamer.finalize(final_content)
assert sent_chunks, "long reply should have been routed through the chunker"
bot.delete_message.assert_awaited()
@@ -0,0 +1,109 @@
"""Tests for the published gateway.yaml / bot.yaml JSON Schema (issue #5414).
Brings the gateway/bot config surface to parity with ``agents.yaml``:
- ``gateway_config_json_schema()`` emits a draft-07 schema from the canonical
Pydantic ``GatewayConfigSchema``.
- The committed ``gateway.schema.json`` artefact matches the model.
- ``praisonai gateway schema`` prints the schema / writes it to a file.
- Scaffolded ``gateway.yaml``/``bot.yaml`` start with the
``# yaml-language-server`` header (and still parse with ``yaml.safe_load``).
"""
import json
from pathlib import Path
import pytest
import yaml
from praisonai_bot.bots._config_schema import (
GATEWAY_SCHEMA_HEADER,
GATEWAY_SCHEMA_URL,
gateway_config_json_schema,
)
def test_gateway_schema_derived_from_pydantic_model():
schema = gateway_config_json_schema()
assert schema["$schema"] == "http://json-schema.org/draft-07/schema#"
assert schema["$id"] == GATEWAY_SCHEMA_URL
props = schema["properties"]
for key in ("gateway", "agents", "channels", "routing", "schedules"):
assert key in props, f"expected '{key}' in gateway schema properties"
def test_committed_artefact_matches_model():
artefact = (
Path(__file__).resolve().parents[3]
/ "praisonai_bot"
/ "bots"
/ "gateway.schema.json"
)
assert artefact.exists(), "gateway.schema.json artefact must be committed"
committed = json.loads(artefact.read_text(encoding="utf-8"))
assert committed == gateway_config_json_schema(), (
"gateway.schema.json is stale; regenerate via "
"`praisonai gateway schema -o "
"src/praisonai-bot/praisonai_bot/bots/gateway.schema.json`"
)
def test_schema_header_points_at_published_url():
assert GATEWAY_SCHEMA_HEADER.startswith("# yaml-language-server: $schema=")
assert GATEWAY_SCHEMA_URL in GATEWAY_SCHEMA_HEADER
assert GATEWAY_SCHEMA_HEADER.endswith("\n")
def test_shipped_sample_gateway_yaml_has_header():
sample = (
Path(__file__).resolve().parents[3] / "gateway.yaml"
)
if not sample.exists():
pytest.skip("sample gateway.yaml not present")
first_line = sample.read_text(encoding="utf-8").splitlines()[0]
assert first_line.startswith("# yaml-language-server: $schema=")
# Leading comment is ignored by safe_load -> execution unaffected.
loaded = yaml.safe_load(sample.read_text(encoding="utf-8"))
assert "gateway" in loaded
def test_generated_configs_scaffold_header():
from praisonai_bot.cli.features.onboard import (
_generate_bot_yaml,
_generate_bot_yaml_multi_channel,
)
single = _generate_bot_yaml(["telegram"])
assert single.startswith("# yaml-language-server: $schema=")
assert yaml.safe_load(single)["gateway"]["port"] == 8765
multi = _generate_bot_yaml_multi_channel(
{
"telegram": {
"platform": "telegram",
"role": "assistant",
"env_var": "TELEGRAM_BOT_TOKEN",
"token": "",
"info": {"allowed_users_env": "TELEGRAM_ALLOWED_USERS"},
"config": {},
}
}
)
assert multi.startswith("# yaml-language-server: $schema=")
assert "gateway" in yaml.safe_load(multi)
def test_gateway_schema_cli_prints_and_writes(tmp_path):
from typer.testing import CliRunner
from praisonai_bot.cli.commands.gateway import app
runner = CliRunner()
result = runner.invoke(app, ["schema"])
assert result.exit_code == 0, result.output
assert '"$schema"' in result.output
out = tmp_path / "gateway.schema.json"
result2 = runner.invoke(app, ["schema", "-o", str(out)])
assert result2.exit_code == 0, result2.output
assert out.exists()
assert json.loads(out.read_text(encoding="utf-8")) == gateway_config_json_schema()
@@ -0,0 +1,147 @@
"""Tests for the machine-readable ``gateway status --json`` snapshot and the
real (non-hardcoded) version reported by ``/info`` and MCP ``serverInfo`` (#5363).
"""
import json
import pytest
import typer
from praisonai_bot.cli.commands import gateway as gateway_cmd
from praisonai_bot.gateway.server import _installed_gateway_version
def test_installed_version_is_not_hardcoded():
version = _installed_gateway_version()
assert version
assert version != "1.0.0"
def test_build_status_snapshot_merges_health_and_info(monkeypatch):
health = {
"status": "healthy",
"uptime": 3600.0,
"sessions": 3,
"clients": 2,
"agents": 1,
"channels": {"telegram": {"running": True}},
"degraded_owners": [],
"applied_config_revision": "a1b2c3d4e5f6",
"on_disk_config_revision": "a1b2c3d4e5f6",
"config_drift": False,
"pressure": {"outbox_depth": 0, "dlq": 0, "dead_targets": 0},
}
info = {"version": "2.3.4", "protocol_version": 7}
def fake_fetch(host, port, path, timeout=5.0):
return health if path == "/health" else info
monkeypatch.setattr(gateway_cmd, "_fetch_gateway_json", fake_fetch)
monkeypatch.setattr(
"praisonai_bot.daemon.get_daemon_status",
lambda: {"platform": "linux", "installed": True, "running": True, "pid": 42},
)
snap = gateway_cmd._build_status_snapshot("127.0.0.1", 8765)
assert snap["reachable"] is True
assert snap["version"] == "2.3.4"
assert snap["protocol_version"] == 7
assert snap["uptime_s"] == 3600.0
assert snap["active_sessions"] == 3
assert snap["clients"] == 2
assert snap["channels"] == {"telegram": {"running": True}}
assert snap["degraded"] == []
assert snap["applied_config_revision"] == "a1b2c3d4e5f6"
assert snap["config_drift"] is False
assert snap["delivery"] == {"outbox_depth": 0, "dlq": 0, "dead_targets": 0}
assert snap["daemon"]["running"] is True
def test_build_status_snapshot_unreachable(monkeypatch):
monkeypatch.setattr(
gateway_cmd, "_fetch_gateway_json", lambda *a, **k: None
)
monkeypatch.setattr(
"praisonai_bot.daemon.get_daemon_status",
lambda: {"platform": "linux", "installed": False, "running": False},
)
snap = gateway_cmd._build_status_snapshot("127.0.0.1", 8765)
assert snap["reachable"] is False
def test_status_json_emits_snapshot_and_exit_code(monkeypatch, capsys):
monkeypatch.setattr(
gateway_cmd,
"_build_status_snapshot",
lambda host, port: {"reachable": True, "version": "9.9.9"},
)
with pytest.raises(typer.Exit) as excinfo:
gateway_cmd.gateway_status(
host="127.0.0.1", port=8765, config=None,
daemon_only=False, deep=False, probe=False, json_output=True,
)
assert excinfo.value.exit_code == 0
out = capsys.readouterr().out
payload = json.loads(out)
assert payload["reachable"] is True
assert payload["version"] == "9.9.9"
def test_status_json_unreachable_exits_nonzero(monkeypatch, capsys):
monkeypatch.setattr(
gateway_cmd,
"_build_status_snapshot",
lambda host, port: {"reachable": False},
)
with pytest.raises(typer.Exit) as excinfo:
gateway_cmd.gateway_status(
host="127.0.0.1", port=8765, config=None,
daemon_only=False, deep=False, probe=False, json_output=True,
)
assert excinfo.value.exit_code == 1
payload = json.loads(capsys.readouterr().out)
assert payload["reachable"] is False
def test_resolve_auth_token_prefers_env(monkeypatch):
monkeypatch.setenv("GATEWAY_AUTH_TOKEN", "env-token")
assert gateway_cmd._resolve_gateway_auth_token() == "env-token"
def test_resolve_auth_token_falls_back_to_persisted_env_file(monkeypatch, tmp_path):
monkeypatch.delenv("GATEWAY_AUTH_TOKEN", raising=False)
env_file = tmp_path / ".env"
env_file.write_text(
"# comment\nOTHER=1\nGATEWAY_AUTH_TOKEN=\"persisted-token\"\n"
)
monkeypatch.setenv("PRAISONAI_ENV_FILE", str(env_file))
assert gateway_cmd._resolve_gateway_auth_token() == "persisted-token"
def test_resolve_auth_token_missing_returns_empty(monkeypatch, tmp_path):
monkeypatch.delenv("GATEWAY_AUTH_TOKEN", raising=False)
monkeypatch.setenv("PRAISONAI_ENV_FILE", str(tmp_path / "does-not-exist.env"))
assert gateway_cmd._resolve_gateway_auth_token() == ""
def test_snapshot_signals_version_unavailable_when_info_unreachable(monkeypatch):
health = {"status": "healthy", "uptime": 1.0}
def fake_fetch(host, port, path, timeout=5.0):
return health if path == "/health" else None
monkeypatch.setattr(gateway_cmd, "_fetch_gateway_json", fake_fetch)
monkeypatch.setattr(
"praisonai_bot.daemon.get_daemon_status",
lambda: {"platform": "linux", "installed": True, "running": True},
)
snap = gateway_cmd._build_status_snapshot("127.0.0.1", 8765)
assert snap["reachable"] is True
assert snap.get("version") is None
assert snap["version_unavailable"] == "info_unreachable"
@@ -797,7 +797,14 @@ def test_mcp_initialize_and_tools_list():
init = asyncio.run(
ep.mcp_jsonrpc(_FakeReq({"jsonrpc": "2.0", "id": 1, "method": "initialize"}))
)
assert _body(init)["result"]["serverInfo"]["name"] == "PraisonAI Gateway"
server_info = _body(init)["result"]["serverInfo"]
assert server_info["name"] == "PraisonAI Gateway"
# The MCP serverInfo version must be the real installed version, not the
# old hardcoded "1.0.0" that misled upgrade checks (#5363, greptile P2).
from praisonai_bot.gateway.server import _installed_gateway_version
assert server_info["version"] == _installed_gateway_version()
assert server_info["version"] != "1.0.0"
listed = asyncio.run(
ep.mcp_jsonrpc(_FakeReq({"jsonrpc": "2.0", "id": 2, "method": "tools/list"}))
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
resolution-markers = [
"python_full_version >= '3.14'",
@@ -1428,7 +1428,7 @@ wheels = [
[[package]]
name = "praisonai-bot"
version = "0.1.8"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "aiohttp" },
@@ -1510,7 +1510,7 @@ provides-extras = ["gateway", "bot", "bot-whatsapp-web", "all"]
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -1543,6 +1543,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -1581,7 +1582,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "primp"
@@ -1 +1 @@
__version__ = "0.1.8"
__version__ = "0.1.9"
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-browser"
version = "0.1.8"
version = "0.1.9"
description = "Browser automation for PraisonAI — Chrome extension bridge, CDP, and Playwright agents extracted from the praisonai wrapper."
readme = "README.md"
license = {text = "MIT"}
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
[[package]]
@@ -932,7 +932,7 @@ wheels = [
[[package]]
name = "praisonai-browser"
version = "0.1.8"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -976,7 +976,7 @@ provides-extras = ["server", "playwright", "all"]
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -1009,6 +1009,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -1047,7 +1048,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "propcache"
@@ -12,6 +12,6 @@ the remaining terminal-agent modules, with PEP 562 shims left behind at the
old ``praisonai.*`` paths for backward compatibility.
"""
__version__ = "0.1.8"
__version__ = "0.1.9"
__all__ = ["__version__"]
@@ -1899,6 +1899,12 @@ def _run_from_file(
# Note: Credential check already done in run_main() entry point
# Capture the re-entrancy sentinel's prior value *before* the guarded block
# so the ``finally`` restore below can never hit an UnboundLocalError when an
# early statement (e.g. the ``PraisonAI`` import) raises before assignment.
import os as _os_guard
_prev_in_run = _os_guard.environ.get(_IN_MODERN_RUN_ENV)
try:
# Use existing PraisonAI class
from praisonai_code.cli.main import PraisonAI
@@ -1909,8 +1915,6 @@ def _run_from_file(
# delegation loops forever ("maximum recursion depth exceeded"). Setting
# this sentinel tells the legacy ``run`` branch we are already inside a
# modern run so it executes the YAML directly instead of bouncing back.
import os as _os_guard
_prev_in_run = _os_guard.environ.get(_IN_MODERN_RUN_ENV)
_os_guard.environ[_IN_MODERN_RUN_ENV] = "1"
praison = PraisonAI(
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-code"
version = "0.1.8"
version = "0.1.9"
description = "Agentic terminal CLI for PraisonAI — the terminal-native agent (run, chat, code, runtime, cli backends) extracted from the praisonai wrapper."
readme = "README.md"
license = {text = "MIT"}
@@ -249,5 +249,69 @@ def test_modern_run_delegation_preserves_resolved_agent_file(monkeypatch, tmp_pa
assert result == "READY"
def _run_from_file_with_broken_import(monkeypatch, tmp_path):
"""Drive ``_run_from_file`` with the ``PraisonAI`` import forced to fail.
Returns ``exc_info.value.code`` so callers can assert the deliberate
``typer.Exit(1)`` propagates instead of an ``UnboundLocalError`` leaking from
the ``finally`` sentinel restore.
"""
import typer
import praisonai_code.cli.commands.run as run_mod
yaml_path = tmp_path / "agents.yaml"
yaml_path.write_text("framework: praisonai\nroles: {}\n")
# Force the guarded ``from praisonai_code.cli.main import PraisonAI`` to raise
# before ``_prev_in_run`` would historically have been assigned.
import builtins
real_import = builtins.__import__
def _boom(name, *args, **kwargs):
if name == "praisonai_code.cli.main":
raise ImportError("boom: simulated main import failure")
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", _boom)
with pytest.raises(typer.Exit) as exc_info:
run_mod._run_from_file(str(yaml_path))
return exc_info.value.exit_code
def test_run_from_file_import_failure_exits_one_no_prior_sentinel(
monkeypatch, tmp_path
):
"""Issue #5632: a failed ``PraisonAI`` import must surface ``typer.Exit(1)``
(not ``UnboundLocalError`` from the ``finally`` sentinel restore), and must
leave an *absent* sentinel absent."""
monkeypatch.delenv("PRAISONAI_IN_MODERN_RUN", raising=False)
import os
code = _run_from_file_with_broken_import(monkeypatch, tmp_path)
assert code == 1
# The previously-absent sentinel is restored to absent, not left set to "1".
assert "PRAISONAI_IN_MODERN_RUN" not in os.environ
def test_run_from_file_import_failure_exits_one_preserves_sentinel(
monkeypatch, tmp_path
):
"""Issue #5632: when a sentinel already exists, a failed ``PraisonAI`` import
must still exit 1 and leave the prior sentinel value unchanged."""
monkeypatch.setenv("PRAISONAI_IN_MODERN_RUN", "preexisting")
import os
code = _run_from_file_with_broken_import(monkeypatch, tmp_path)
assert code == 1
# The existing sentinel is restored to its original value, not overwritten.
assert os.environ.get("PRAISONAI_IN_MODERN_RUN") == "preexisting"
if __name__ == "__main__":
pytest.main([__file__, "-v"])
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
resolution-markers = [
"python_full_version >= '3.14' and sys_platform == 'win32'",
@@ -1370,7 +1370,7 @@ wheels = [
[[package]]
name = "praisonai-code"
version = "0.1.8"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -1403,7 +1403,7 @@ requires-dist = [
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -1436,6 +1436,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -1474,7 +1475,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "propcache"
@@ -1 +1 @@
__version__ = "0.1.8"
__version__ = "0.1.9"
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-deploy"
version = "0.1.8"
version = "0.1.9"
description = "Deployment for PraisonAI — API, Docker, and cloud (AWS, Azure, GCP) deployments extracted from the praisonai wrapper."
readme = "README.md"
license = {text = "MIT"}
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
[[package]]
@@ -977,7 +977,7 @@ wheels = [
[[package]]
name = "praisonai-deploy"
version = "0.1.8"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -1014,7 +1014,7 @@ provides-extras = ["api", "all"]
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -1047,6 +1047,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -1085,7 +1086,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "propcache"
+1 -1
View File
@@ -1 +1 @@
__version__ = "0.1.10"
__version__ = "0.1.11"
@@ -23,16 +23,28 @@ def register_memory_tools() -> None:
@register_tool("praisonai.memory.show")
def memory_show(user_id: Optional[str] = None) -> str:
"""Show all stored memories (optionally filtered by user)."""
"""Show stored memories; SQLite displays up to 1000 per tier."""
try:
from praisonaiagents.memory import Memory
memory = Memory()
data = memory.get_all_memories()
if user_id:
adapter = getattr(memory, "memory_adapter", None)
if getattr(memory, "provider", None) == "sqlite" and adapter is not None:
# Display retains its previous bounded window. Full enumeration
# is a separate API and must not load the whole store here.
data = []
for tier in ("short", "long"):
for record in getattr(adapter, f"search_{tier}_term")("", limit=1000, user_id=user_id):
data.append({
**record, "memory_type": f"{tier}_term", "type": f"{tier}_term",
})
else:
data = memory.get_all_memories()
if user_id is not None:
data = [
m for m in data
if str((m.get("metadata") or {}).get("user_id")) == str(user_id)
if (m.get("metadata") or {}).get("user_id") is not None
and str((m.get("metadata") or {}).get("user_id")) == str(user_id)
]
return str(data)
except ImportError:
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-mcp"
version = "0.1.10"
version = "0.1.11"
description = "MCP server hosting for PraisonAI — expose agents and tools to Cursor, Claude Desktop, and other MCP clients."
readme = "README.md"
license = {text = "MIT"}
@@ -9,6 +9,7 @@ bind against the real core API and assert no ``Error:`` fallback.
import sys
import types
import ast
import pytest
@@ -72,6 +73,27 @@ def test_memory_show_calls_real_api(fake_memory, memory_tools):
assert "hello" in out
@pytest.mark.parametrize("user_id,expected", [
("", ["empty"]),
("None", ["literal-none"]),
("u1", ["named"]),
(None, ["missing", "null", "empty", "literal-none", "named"]),
])
def test_memory_show_distinguishes_missing_and_explicit_user_ids(fake_memory, memory_tools, monkeypatch, user_id, expected):
import praisonaiagents.memory as module
records = [
{"id": "missing", "metadata": {}},
{"id": "null", "metadata": {"user_id": None}},
{"id": "empty", "metadata": {"user_id": ""}},
{"id": "literal-none", "metadata": {"user_id": "None"}},
{"id": "named", "metadata": {"user_id": "u1"}},
]
monkeypatch.setattr(module.Memory, "get_all_memories", lambda self: records)
actual = ast.literal_eval(_call(memory_tools, "praisonai.memory.show", user_id=user_id))
assert [record["id"] for record in actual] == expected
def test_memory_add_calls_store_short_term(fake_memory, memory_tools):
out = _call(memory_tools, "praisonai.memory.add", content="note")
assert not out.startswith("Error:")
@@ -88,3 +110,40 @@ def test_memory_clear_calls_reset_all(fake_memory, memory_tools):
out = _call(memory_tools, "praisonai.memory.clear")
assert not out.startswith("Error:")
assert fake_memory["reset"] == 1
@pytest.mark.parametrize("user_id,expected", [(None, 2000), ("even", 1000), ("old-user", 2), ("absent", 0)])
def test_sqlite_show_bounds_fetches_without_capping_full_enumeration(tmp_path, monkeypatch, memory_tools, user_id, expected):
import praisonaiagents.memory as module
monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path / "home"))
monkeypatch.chdir(tmp_path)
memory = module.Memory(config={"provider": "sqlite"})
adapter = memory.memory_adapter
queries = []
try:
for tier in ("short", "long"):
for index in range(1001):
getattr(memory, f"store_{tier}_term")(
f"{tier} entry {index}", metadata={"user_id": "old-user" if index == 0 else "even" if index % 2 == 0 else "odd"},
)
# Distinct valid timestamps make the displayed latest-1000 window
# deterministic; SQLite's second-resolution default can tie here.
for tier, conn in (("short", adapter._get_stm_conn()), ("long", adapter._get_ltm_conn())):
conn.execute(f"UPDATE {tier}_term_memory SET timestamp = datetime(1700000000 + id, 'unixepoch')")
conn.commit()
adapter._get_stm_conn().set_trace_callback(queries.append)
adapter._get_ltm_conn().set_trace_callback(queries.append)
monkeypatch.setattr(module, "Memory", lambda: memory)
records = ast.literal_eval(_call(memory_tools, "praisonai.memory.show", user_id=user_id))
assert len(records) == expected
selects = [query for query in queries if query.startswith("SELECT id, content")]
assert len(selects) == 2
assert all("LIMIT 1000" in query for query in selects)
if user_id:
assert all(record["metadata"]["user_id"] == user_id for record in records)
assert {record["type"] for record in records} == ({"short_term", "long_term"} if expected else set())
assert len(memory.get_all_memories()) == 2002
finally:
adapter.close_connections()
memory.close_connections()
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
resolution-markers = [
"python_full_version >= '3.14' and sys_platform == 'win32'",
@@ -1044,7 +1044,7 @@ wheels = [
[[package]]
name = "praisonai-mcp"
version = "0.1.10"
version = "0.1.11"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -1084,7 +1084,7 @@ provides-extras = ["server", "auth", "all"]
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -1117,6 +1117,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -1155,7 +1156,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "propcache"
@@ -1 +1 @@
__version__ = "0.1.8"
__version__ = "0.1.9"
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-sandbox"
version = "0.1.8"
version = "0.1.9"
description = "Where PraisonAI agents run code — Docker, E2B, Modal, Daytona, Fly.io, Tenki, Novita, Sandlock, SSH and subprocess, as both single sandboxes and multi-instance compute providers."
readme = "README.md"
license = {text = "MIT"}
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
resolution-markers = [
"python_full_version >= '3.14'",
@@ -1679,7 +1679,7 @@ wheels = [
[[package]]
name = "praisonai-sandbox"
version = "0.1.8"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -1744,7 +1744,7 @@ provides-extras = ["docker", "e2b", "sandlock", "ssh", "tenki", "flyio", "modal"
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -1777,6 +1777,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -1815,7 +1816,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "propcache"
@@ -1 +1 @@
__version__ = "0.1.8"
__version__ = "0.1.9"
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "praisonai-train"
version = "0.1.8"
version = "0.1.9"
description = "Training for PraisonAI — LLM fine-tuning (Unsloth) and iterative agent training extracted from the praisonai wrapper."
readme = "README.md"
license = "MIT"
+5 -4
View File
@@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 5
requires-python = ">=3.10, <3.15"
resolution-markers = [
"python_full_version >= '3.14' and sys_platform == 'win32'",
@@ -2606,7 +2606,7 @@ wheels = [
[[package]]
name = "praisonai-train"
version = "0.1.8"
version = "0.1.9"
source = { editable = "." }
dependencies = [
{ name = "click" },
@@ -2696,7 +2696,7 @@ provides-extras = ["llm", "vision", "all"]
[[package]]
name = "praisonaiagents"
version = "1.7.9"
version = "1.7.10"
source = { directory = "../praisonai-agents" }
dependencies = [
{ name = "aiohttp" },
@@ -2729,6 +2729,7 @@ requires-dist = [
{ name = "litellm", marker = "extra == 'memory'", specifier = ">=1.83.14" },
{ name = "markitdown", extras = ["all"], marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mcp", marker = "extra == 'mcp'", specifier = ">=1.20.0" },
{ name = "mcp", marker = "extra == 'parallel-search'", specifier = ">=1.30.0,<2" },
{ name = "mem0ai", marker = "extra == 'knowledge'", specifier = ">=0.1.0" },
{ name = "mem0ai", extras = ["graph"], marker = "extra == 'graph'", specifier = ">=0.1.0" },
{ name = "motor", marker = "extra == 'mongodb'", specifier = ">=3.4.0" },
@@ -2767,7 +2768,7 @@ requires-dist = [
{ name = "uvicorn", marker = "extra == 'os'", specifier = ">=0.34.0" },
{ name = "websockets", marker = "extra == 'mcp'", specifier = ">=12.0" },
]
provides-extras = ["sandbox", "mcp", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
provides-extras = ["sandbox", "mcp", "parallel-search", "memory", "knowledge", "graph", "llm", "api", "os", "telemetry", "mongodb", "dakera", "auth", "autonomy", "search", "crawl", "a2ui", "all"]
[[package]]
name = "propcache"
+1 -1
View File
@@ -60,7 +60,7 @@ This report compares **TypeScript/JavaScript SDK feature categories** against **
| ✅ Memory | 24 | 4 | 840 |
| ✅ Middleware | 2 | 1 | 162 |
| ✅ OCR | 7 | 1 | 253 |
| ✅ Observability | 12 | 28 | 2470 |
| ✅ Observability | 12 | 28 | 2692 |
| ✅ Optimizer | 2 | 1 | 162 |
| ✅ Output | 4 | 1 | 220 |
| ✅ Parallel Execution | 12 | 2 | 256 |
@@ -1,21 +1,412 @@
/**
* LangSmith Observability Adapter
*
* DELIVERY IS NOT IMPLEMENTED. This adapter records traces in memory and sends
* nothing to LangSmith. It therefore reports `isEnabled === false` and warns on
* construction; see `./undelivered.ts` for the rationale.
* Delivers traces to LangSmith via its REST Runs API. Like the Langfuse
* adapter, `isEnabled` is derived from a live transport rather than hardcoded:
* it is true only after initialize() has validated an API key and built an
* HTTP client. Until then (and if no key is configured) every span is recorded
* in the in-memory adapter and nowhere else.
*
* NOTE: an earlier version of this adapter dynamically imported the `langsmith`
* SDK in initialize() and constructed a Client. That client was assigned to a
* private field and then never read by any other method, so no run was ever
* created or posted. The import has been removed rather than left in place,
* because it made the adapter look wired up when it was not.
* Transport uses the `axios` dependency already shipped by this package, so no
* new dependency and no vendor SDK is required. Run POST/PATCH requests are
* queued and flushed asynchronously; the hot path never awaits vendor HTTP.
*
* Configuration (secrets are read from the environment only, never logged):
* - LANGSMITH_API_KEY or LANGCHAIN_API_KEY (required to deliver)
* - LANGSMITH_ENDPOINT or LANGCHAIN_ENDPOINT (optional, must be http(s))
* - LANGSMITH_PROJECT or LANGCHAIN_PROJECT (optional project name)
*/
import type { ObservabilityToolConfig } from '../../types';
import { UndeliveredObservabilityAdapter } from './undelivered';
import type {
ObservabilityAdapter,
TraceContext,
SpanContext,
SpanKind,
SpanStatus,
AttributionContext,
ProviderMetadata,
ObservabilityToolConfig
} from '../../types';
import { MemoryObservabilityAdapter } from '../memory';
export class LangSmithObservabilityAdapter extends UndeliveredObservabilityAdapter {
const DEFAULT_ENDPOINT = 'https://api.smith.langchain.com';
type HttpClient = {
post: (url: string, body: unknown, cfg?: unknown) => Promise<unknown>;
patch: (url: string, body: unknown, cfg?: unknown) => Promise<unknown>;
};
/** Minimal shape of a LangSmith run payload we build from spans. */
interface LangSmithRun {
id: string;
trace_id: string;
dotted_order: string;
parent_run_id?: string;
name: string;
run_type: string;
start_time: string;
end_time?: string;
error?: string;
extra?: Record<string, unknown>;
session_name?: string;
}
/** LangSmith run_type values keyed by our SpanKind. */
function toRunType(kind: SpanKind): string {
switch (kind) {
case 'llm':
case 'embedding':
return 'llm';
case 'tool':
return 'tool';
case 'retrieval':
return 'retriever';
case 'agent':
case 'workflow':
return 'chain';
default:
return 'chain';
}
}
function pad(n: number): string {
return n.toString().padStart(3, '0');
}
/** True for http://localhost, 127.0.0.0/8 and ::1 — the only safe plaintext hosts. */
function isLoopbackEndpoint(url: string): boolean {
let host: string;
try {
host = new URL(url).hostname.toLowerCase().replace(/^\[|\]$/g, '');
} catch {
return false;
}
return host === 'localhost' || host === '::1' || /^127\.\d+\.\d+\.\d+$/.test(host);
}
/**
* LangSmith orders sibling runs by a "dotted_order" string built from the run
* start time and id. We keep a monotonic sequence per trace so ordering is
* stable even when spans start within the same millisecond.
*/
function dottedSegment(date: Date, seq: number, id: string): string {
const iso = date.toISOString().replace(/[-:]/g, '').replace('.', '').replace('Z', '');
return `${iso}${pad(seq)}Z${id}`;
}
export class LangSmithObservabilityAdapter implements ObservabilityAdapter {
readonly name = 'langsmith';
/** This adapter has a real delivery implementation. */
readonly delivers: boolean = true;
/**
* True only when a validated HTTP client exists. A missing API key, or a
* non-http(s) endpoint, leaves this false so callers are never told their
* traces are delivered when they are not.
*/
get isEnabled(): boolean {
return this.client != null;
}
private client: HttpClient | null = null;
private endpoint: string = DEFAULT_ENDPOINT;
private project?: string;
private memory: MemoryObservabilityAdapter;
private config: ObservabilityToolConfig;
private runs: Map<string, LangSmithRun> = new Map();
private seqByTrace: Map<string, number> = new Map();
private dottedBySpan: Map<string, string> = new Map();
private queue: Promise<void> = Promise.resolve();
private deliveryErrors = 0;
constructor(config?: ObservabilityToolConfig) {
super('langsmith', config);
this.config = config || { name: 'langsmith' };
this.memory = new MemoryObservabilityAdapter();
}
async initialize(): Promise<void> {
const apiKey =
this.config.apiKey || process.env.LANGSMITH_API_KEY || process.env.LANGCHAIN_API_KEY;
if (!apiKey) {
// No credentials: stay disabled and record to memory only. This is the
// honest "configured but not delivering" state the issue calls out.
this.client = null;
return;
}
const rawEndpoint =
this.config.baseUrl ||
process.env.LANGSMITH_ENDPOINT ||
process.env.LANGCHAIN_ENDPOINT ||
DEFAULT_ENDPOINT;
if (!/^https?:\/\//i.test(rawEndpoint)) {
// SR-003: only http(s) schemes are accepted for self-hosted URLs.
console.warn(
`[OBSERVABILITY] Ignoring LangSmith endpoint "${rawEndpoint}": only http(s) URLs are supported. ` +
'Delivery is disabled; traces are recorded in memory only.'
);
this.client = null;
return;
}
// SR-003: the API key travels in an x-api-key header, so a plaintext http://
// endpoint would leak the credential to any on-path observer. Allow http
// only for loopback hosts (local self-hosted dev); require https otherwise.
if (/^http:\/\//i.test(rawEndpoint) && !isLoopbackEndpoint(rawEndpoint)) {
console.warn(
`[OBSERVABILITY] Refusing LangSmith endpoint "${rawEndpoint}": the API key must not be sent over ` +
'plaintext http to a non-local host. Use https (or an http://localhost endpoint). ' +
'Delivery is disabled; traces are recorded in memory only.'
);
this.client = null;
return;
}
this.endpoint = rawEndpoint.replace(/\/$/, '');
this.project =
this.config.projectId || process.env.LANGSMITH_PROJECT || process.env.LANGCHAIN_PROJECT;
try {
const axiosModule: any = await import('axios' as string);
const axios = axiosModule.default || axiosModule;
const instance = axios.create({
baseURL: this.endpoint,
timeout: 10000,
headers: { 'x-api-key': apiKey, 'Content-Type': 'application/json' }
});
this.client = instance as HttpClient;
} catch {
console.warn(
'[OBSERVABILITY] Failed to construct the LangSmith HTTP client; traces are recorded in memory only.'
);
this.client = null;
}
}
async shutdown(): Promise<void> {
await this.flush();
// Dropping the client makes isEnabled report false again, so a cached
// instance never advertises a torn-down transport as live.
this.client = null;
await this.memory.shutdown();
}
private nextSeq(traceId: string): number {
const n = (this.seqByTrace.get(traceId) || 0) + 1;
this.seqByTrace.set(traceId, n);
return n;
}
private enqueue(work: () => Promise<void>): void {
this.queue = this.queue.then(work).catch(() => {
// A rejected request must not break the chain (so later spans still try),
// but it must not vanish either: count it so flush() can report that not
// everything was delivered. The error is swallowed here only after being
// surfaced by the work function itself.
this.deliveryErrors++;
});
}
/** Drop a run's bookkeeping once it has ended so long processes stay bounded. */
private evict(runId: string): void {
this.runs.delete(runId);
this.dottedBySpan.delete(runId);
}
startTrace(
name: string,
metadata?: Record<string, unknown>,
attribution?: AttributionContext
): TraceContext {
const memoryCtx = this.memory.startTrace(name, metadata, attribution);
const traceId = memoryCtx.traceId;
if (this.client) {
const now = new Date();
const dotted = dottedSegment(now, this.nextSeq(traceId), traceId);
this.dottedBySpan.set(traceId, dotted);
const run: LangSmithRun = {
id: traceId,
trace_id: traceId,
dotted_order: dotted,
name,
run_type: 'chain',
start_time: now.toISOString(),
session_name: this.project,
extra: {
metadata: {
...(metadata || {}),
userId: attribution?.userId,
sessionId: attribution?.sessionId,
agentId: attribution?.agentId
}
}
};
this.runs.set(traceId, run);
this.enqueue(() => this.postRun(run));
}
return {
traceId,
startSpan: (spanName: string, kind: SpanKind, parentId?: string) =>
this.startSpan(traceId, spanName, kind, parentId),
end: (status?: SpanStatus) => this.endTrace(traceId, status)
};
}
endTrace(traceId: string, status?: SpanStatus): void {
this.memory.endTrace(traceId, status);
const run = this.runs.get(traceId);
if (this.client && run) {
const patch: Partial<LangSmithRun> & { id: string } = {
id: run.id,
end_time: new Date().toISOString()
};
if (status === 'failed') patch.error = 'trace failed';
this.enqueue(() => this.patchRun(patch));
this.evict(traceId);
this.seqByTrace.delete(traceId);
}
}
startSpan(traceId: string, name: string, kind: SpanKind, parentId?: string): SpanContext {
const memoryCtx = this.memory.startSpan(traceId, name, kind, parentId);
const spanId = memoryCtx.spanId;
if (this.client) {
const now = new Date();
const parentDotted = this.dottedBySpan.get(parentId || traceId) || '';
const segment = dottedSegment(now, this.nextSeq(traceId), spanId);
const dotted = parentDotted ? `${parentDotted}.${segment}` : segment;
this.dottedBySpan.set(spanId, dotted);
const run: LangSmithRun = {
id: spanId,
trace_id: traceId,
dotted_order: dotted,
parent_run_id: parentId || traceId,
name,
run_type: toRunType(kind),
start_time: now.toISOString(),
session_name: this.project
};
this.runs.set(spanId, run);
this.enqueue(() => this.postRun(run));
}
const self = this;
return {
spanId,
traceId,
addEvent(eventName: string, attributes?: Record<string, unknown>): void {
self.addEvent(spanId, eventName, attributes);
},
setAttributes(attributes: Record<string, unknown>): void {
const run = self.runs.get(spanId);
if (self.client && run) {
self.enqueue(() =>
self.patchRun({ id: spanId, extra: { metadata: attributes } })
);
}
},
setProviderMetadata(metadata: ProviderMetadata): void {
const run = self.runs.get(spanId);
if (self.client && run) {
self.enqueue(() =>
self.patchRun({
id: spanId,
extra: {
metadata: {
provider: metadata.provider,
model: metadata.model,
promptTokens: metadata.promptTokens,
completionTokens: metadata.completionTokens,
totalTokens: metadata.totalTokens,
latencyMs: metadata.latencyMs
}
}
})
);
}
},
recordError(error: Error): void {
self.recordError(spanId, error);
},
end(status?: SpanStatus): void {
self.endSpan(spanId, status);
}
};
}
endSpan(spanId: string, status?: SpanStatus, attributes?: Record<string, unknown>): void {
this.memory.endSpan(spanId, status, attributes);
const run = this.runs.get(spanId);
if (this.client && run) {
const patch: Partial<LangSmithRun> & { id: string } = {
id: spanId,
end_time: new Date().toISOString()
};
if (attributes) patch.extra = { metadata: attributes };
if (status === 'failed') patch.error = 'span failed';
this.enqueue(() => this.patchRun(patch));
this.evict(spanId);
}
}
addEvent(spanId: string, name: string, attributes?: Record<string, unknown>): void {
this.memory.addEvent(spanId, name, attributes);
const run = this.runs.get(spanId);
if (this.client && run) {
// LangSmith has no first-class "event" on a run, so we accumulate events
// on the run payload and patch them into extra.events where they are
// visible alongside the span. Without this the registry's events:true
// claim would be dishonest.
const extra = (run.extra = run.extra || {});
const events = ((extra as any).events = ((extra as any).events as unknown[]) || []);
events.push({ name, time: new Date().toISOString(), attributes: attributes || {} });
const snapshot = events.slice();
this.enqueue(() => this.patchRun({ id: spanId, extra: { events: snapshot } }));
}
}
recordError(spanId: string, error: Error): void {
this.memory.recordError(spanId, error);
const run = this.runs.get(spanId);
if (this.client && run) {
this.enqueue(() => this.patchRun({ id: spanId, error: error.message }));
}
}
async flush(): Promise<void> {
if (!this.client) {
// No client means nothing was ever sent. Resolving quietly would promise
// a delivery that did not happen.
console.warn(
'[OBSERVABILITY] flush() on the "langsmith" adapter delivered nothing: no API key was ' +
'configured, so no client was created. Set LANGSMITH_API_KEY to enable delivery.'
);
return;
}
await this.queue;
if (this.deliveryErrors > 0) {
// Honest degradation: a configured, "enabled" client still may not have
// delivered everything (bad key, network, 4xx/5xx). Surface it rather than
// letting flush() imply a clean delivery.
const failed = this.deliveryErrors;
this.deliveryErrors = 0;
console.warn(
`[OBSERVABILITY] flush() on the "langsmith" adapter completed with ${failed} failed ` +
'run request(s): some traces were NOT delivered to LangSmith. Check the API key, endpoint ' +
'and network connectivity.'
);
}
}
private async postRun(run: LangSmithRun): Promise<void> {
if (!this.client) return;
await this.client.post('/runs', run);
}
private async patchRun(patch: Partial<LangSmithRun> & { id: string }): Promise<void> {
if (!this.client) return;
await this.client.patch(`/runs/${patch.id}`, patch);
}
}
@@ -83,6 +83,10 @@ export async function createObservabilityAdapter(
case 'langfuse':
adapter = await createLangfuseAdapter(config);
break;
case 'langsmith':
adapter = await createLangSmithAdapter(config);
break;
default:
console.warn(`Unknown observability tool: ${name}, using noop adapter`);
@@ -125,7 +129,6 @@ export function clearAdapterCache(): void {
type ExternalAdapterCtor = new (config?: ObservabilityToolConfig) => ObservabilityAdapter;
const UNDELIVERED_ADAPTER_LOADERS: Record<string, () => Promise<ExternalAdapterCtor>> = {
langsmith: async () => (await import('./external/langsmith')).LangSmithObservabilityAdapter,
langwatch: async () => (await import('./external/langwatch')).LangWatchObservabilityAdapter,
arize: async () => (await import('./external/arize')).ArizeObservabilityAdapter,
axiom: async () => (await import('./external/axiom')).AxiomObservabilityAdapter,
@@ -153,3 +156,16 @@ async function createLangfuseAdapter(config?: ObservabilityToolConfig): Promise<
}
}
async function createLangSmithAdapter(config?: ObservabilityToolConfig): Promise<ObservabilityAdapter> {
try {
const { LangSmithObservabilityAdapter } = await import('./external/langsmith');
return new LangSmithObservabilityAdapter(config);
} catch (error) {
// The adapter reports isEnabled true only once initialize() validates an
// API key and builds an HTTP client; whether it delivers is decided there,
// not here. This branch only fires if the local module fails to load.
console.warn('[OBSERVABILITY] Failed to load the langsmith adapter module, falling back to the memory adapter (traces are not delivered).');
return new MemoryObservabilityAdapter();
}
}
+13 -4
View File
@@ -187,6 +187,12 @@ export interface ObservabilityToolInfo {
name: ObservabilityToolName;
package?: string;
envKey: string;
/**
* Additional environment variables that also satisfy this tool's credential
* requirement (e.g. a vendor's legacy key name). Any one being set counts as
* configured for readiness checks.
*/
altEnvKeys?: string[];
description: string;
/**
* Whether traces actually reach this tool's backend. False means the adapter
@@ -220,10 +226,11 @@ export const OBSERVABILITY_TOOLS: Record<ObservabilityToolName, ObservabilityToo
langsmith: {
name: 'langsmith',
package: 'langsmith',
envKey: 'LANGCHAIN_API_KEY',
envKey: 'LANGSMITH_API_KEY',
altEnvKeys: ['LANGCHAIN_API_KEY'],
description: 'LangSmith by LangChain',
delivers: false,
features: { traces: true, spans: true, events: true, errors: true, metrics: false, export: false },
delivers: true,
features: { traces: true, spans: true, events: true, errors: true, metrics: false, export: true },
docsUrl: 'https://docs.smith.langchain.com'
},
langwatch: {
@@ -377,5 +384,7 @@ export function listObservabilityTools(): ObservabilityToolInfo[] {
export function hasObservabilityToolEnvVar(name: ObservabilityToolName): boolean {
const info = OBSERVABILITY_TOOLS[name];
if (!info || !info.envKey) return true; // Built-in tools don't need env vars
return !!process.env[info.envKey];
if (process.env[info.envKey]) return true;
// A vendor's legacy/alternate key name also counts as configured.
return (info.altEnvKeys || []).some(key => !!process.env[key]);
}
@@ -168,9 +168,9 @@ describe('observability registry honesty', () => {
.filter(t => t.delivers)
.map(t => t.name)
.sort();
// Langfuse is the only external integration that actually posts anywhere.
// These are the external integrations that actually post somewhere.
// Adding a name here without implementing delivery re-introduces the bug.
expect(delivering).toEqual(['langfuse']);
expect(delivering).toEqual(['langfuse', 'langsmith']);
});
});
@@ -216,6 +216,167 @@ describe('LangfuseObservabilityAdapter delivery state', () => {
});
});
describe('LangSmithObservabilityAdapter delivery (mock HTTP transport)', () => {
function makeMockClient() {
const posts: any[] = [];
const patches: any[] = [];
const client = {
async post(url: string, body: any) {
posts.push({ url, body });
return { status: 201 };
},
async patch(url: string, body: any) {
patches.push({ url, body });
return { status: 200 };
}
};
return { client, posts, patches };
}
it('is disabled with no API key and warns that flush delivered nothing (OBS-DEL-002)', async () => {
const warnSpy = jest.spyOn(console, 'warn').mockImplementation(() => {});
const savedSmith = process.env.LANGSMITH_API_KEY;
const savedChain = process.env.LANGCHAIN_API_KEY;
try {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
delete process.env.LANGSMITH_API_KEY;
delete process.env.LANGCHAIN_API_KEY;
const adapter = new LangSmithObservabilityAdapter();
await adapter.initialize();
expect(adapter.isEnabled).toBe(false);
expect(adapter.delivers).toBe(true);
await adapter.flush();
const messages = warnSpy.mock.calls.map(c => String(c[0])).join('\n');
expect(messages.toLowerCase()).toContain('delivered nothing');
} finally {
// Restore ambient credentials so later tests in this worker are unaffected.
if (savedSmith === undefined) delete process.env.LANGSMITH_API_KEY;
else process.env.LANGSMITH_API_KEY = savedSmith;
if (savedChain === undefined) delete process.env.LANGCHAIN_API_KEY;
else process.env.LANGCHAIN_API_KEY = savedChain;
warnSpy.mockRestore();
}
});
it('surfaces failed deliveries in flush() instead of resolving silently (OBS-DEL-004)', async () => {
const warnSpy = jest.spyOn(console, 'warn').mockImplementation(() => {});
try {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
const adapter = new LangSmithObservabilityAdapter();
// A client whose POST rejects simulates a bad key / network / 4xx.
(adapter as any).client = {
async post() { throw new Error('401 unauthorized'); },
async patch() { return { status: 200 }; }
};
expect(adapter.isEnabled).toBe(true);
const trace = adapter.startTrace('agent-run', {});
adapter.endTrace(trace.traceId, 'completed');
await adapter.flush();
const messages = warnSpy.mock.calls.map(c => String(c[0])).join('\n').toLowerCase();
expect(messages).toContain('not delivered');
} finally {
warnSpy.mockRestore();
}
});
it('exports span events into the run payload (OBS-DEL-005)', async () => {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
const adapter = new LangSmithObservabilityAdapter();
const { client, patches } = makeMockClient();
(adapter as any).client = client;
const trace = adapter.startTrace('agent-run', {});
const span = adapter.startSpan(trace.traceId, 'search', 'tool');
span.addEvent('cache_hit', { key: 'q1' });
adapter.endSpan(span.spanId, 'completed');
adapter.endTrace(trace.traceId, 'completed');
await adapter.flush();
const eventPatch = patches.find(
(p: any) => p.body.extra && Array.isArray(p.body.extra.events)
);
expect(eventPatch).toBeDefined();
expect(eventPatch.body.extra.events[0].name).toBe('cache_hit');
});
it('refuses a plaintext http endpoint to a non-local host (SR-003)', async () => {
const warnSpy = jest.spyOn(console, 'warn').mockImplementation(() => {});
try {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
const adapter = new LangSmithObservabilityAdapter({
name: 'langsmith',
apiKey: 'test-key',
baseUrl: 'http://traces.example.com'
});
await adapter.initialize();
expect(adapter.isEnabled).toBe(false);
const messages = warnSpy.mock.calls.map(c => String(c[0])).join('\n').toLowerCase();
expect(messages).toContain('plaintext http');
} finally {
warnSpy.mockRestore();
}
});
it('evicts completed runs so memory stays bounded (OBS-DEL-006)', async () => {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
const adapter = new LangSmithObservabilityAdapter();
const { client } = makeMockClient();
(adapter as any).client = client;
const trace = adapter.startTrace('agent-run', {});
const span = adapter.startSpan(trace.traceId, 'search', 'tool');
adapter.endSpan(span.spanId, 'completed');
adapter.endTrace(trace.traceId, 'completed');
await adapter.flush();
expect((adapter as any).runs.size).toBe(0);
expect((adapter as any).dottedBySpan.size).toBe(0);
});
it('posts a run for the agent root and child spans, and patches on end (OBS-DEL-001/003)', async () => {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
const adapter = new LangSmithObservabilityAdapter();
const { client, posts, patches } = makeMockClient();
// Inject a live transport, exactly as the factory would after a valid key.
(adapter as any).client = client;
expect(adapter.isEnabled).toBe(true);
const trace = adapter.startTrace('agent-run', { task: 'demo' });
const toolSpan = adapter.startSpan(trace.traceId, 'search', 'tool');
adapter.endSpan(toolSpan.spanId, 'completed');
adapter.endTrace(trace.traceId, 'completed');
await adapter.flush();
// Root run + one child run posted.
expect(posts.map(p => p.url)).toEqual(['/runs', '/runs']);
const root = posts[0].body;
const child = posts[1].body;
expect(root.run_type).toBe('chain');
expect(child.run_type).toBe('tool');
// Child is parented to the root trace so it nests under the agent span.
expect(child.parent_run_id).toBe(trace.traceId);
expect(child.trace_id).toBe(trace.traceId);
expect(child.dotted_order.startsWith(root.dotted_order)).toBe(true);
// Both runs are closed with an end_time patch.
expect(patches.length).toBe(2);
expect(patches.every((p: any) => typeof p.body.end_time === 'string')).toBe(true);
});
it('rejects a non-http(s) endpoint and stays disabled (SR-003)', async () => {
const warnSpy = jest.spyOn(console, 'warn').mockImplementation(() => {});
try {
const { LangSmithObservabilityAdapter } = require(path.join(EXTERNAL_DIR, 'langsmith'));
const adapter = new LangSmithObservabilityAdapter({
name: 'langsmith',
apiKey: 'test-key',
baseUrl: 'ftp://evil.example'
});
await adapter.initialize();
expect(adapter.isEnabled).toBe(false);
const messages = warnSpy.mock.calls.map(c => String(c[0])).join('\n');
expect(messages.toLowerCase()).toContain('http(s)');
} finally {
warnSpy.mockRestore();
}
});
});
describe('observability CLI reports delivery truthfully', () => {
let logSpy: any;
let warnSpy: any;
+2 -2
View File
@@ -3,8 +3,8 @@ class Praisonai < Formula
desc "AI tools for various AI applications"
homepage "https://github.com/MervinPraison/PraisonAI"
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.7.11.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.7.11.tar.gz | shasum -a 256`.split.first
url "https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.7.12.tar.gz"
sha256 `curl -sL https://github.com/MervinPraison/PraisonAI/archive/refs/tags/v4.7.12.tar.gz | shasum -a 256`.split.first
license "MIT"
depends_on "[email protected]"
+1 -1
View File
@@ -1 +1 @@
__version__ = "4.7.11"
__version__ = "4.7.12"
+8 -8
View File
@@ -12,14 +12,14 @@ dependencies = [
"rich>=13.7",
"markdown>=3.5",
"pyparsing>=3.0.0",
"praisonaiagents>=1.7.9",
"praisonai-code>=0.1.8",
"praisonai-bot>=0.1.8",
"praisonai-train>=0.1.8",
"praisonai-browser>=0.1.8",
"praisonai-mcp>=0.1.10",
"praisonai-sandbox>=0.1.8",
"praisonai-deploy>=0.1.8",
"praisonaiagents>=1.7.10",
"praisonai-code>=0.1.9",
"praisonai-bot>=0.1.9",
"praisonai-train>=0.1.9",
"praisonai-browser>=0.1.9",
"praisonai-mcp>=0.1.11",
"praisonai-sandbox>=0.1.9",
"praisonai-deploy>=0.1.9",
"python-dotenv>=0.19.0",
"litellm>=1.83.14,<2",
"PyYAML>=6.0",