fix: gather exception isolation, handoff chain guard, atomic memory writes - #3586
fix: gather exception isolation, handoff chain guard, atomic memory writes#3586praisonai-triage-agent[bot] wants to merge 2 commits into
Conversation
…rites (fixes #3585) - agents.py: add _gather_with_isolation so a failing async task no longer orphans its siblings; siblings are awaited to completion and the first exception is re-raised (return_exceptions=True) across all 3 gather sites. - handoff.py: guard finally:_pop_handoff() with a `pushed` flag at all 5 call sites so a _check_safety rejection (which raises before _push_handoff) no longer pops an ancestor's entry and corrupts the cycle/depth guard. - file_memory.py: write JSON to a temp file + fsync + os.replace so a crash mid-write can no longer truncate the persisted memory store. Co-authored-by: MervinPraison <MervinPraison@users.noreply.github.com>
|
@coderabbitai review |
|
/review |
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
✅ Action performedReview finished.
|
MervinPraison
added
pipeline/blocked:ci
📝 WalkthroughWalkthroughThe PR updates handoff-chain cleanup across execution paths, isolates asynchronous task exceptions until sibling tasks settle, and makes ChangesHandoff chain safety
Asynchronous task isolation
Atomic file memory persistence
Estimated code review effort: 3 (Moderate) | ~25 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Greptile SummaryThis PR isolates asynchronous fan-out failures, preserves handoff-chain state when safety checks reject a handoff, and replaces direct memory-file writes with temporary-file replacement. The memory change still needs destination-level serialization to prevent concurrent updates from silently overwriting one another.
Confidence Score: 4/5The PR is not yet safe to merge because concurrent memory updates can still complete successfully while silently discarding one writer's data. Each write now locks a unique temporary inode, while episodic updates perform an unlocked read-modify-write against a shared destination; competing atomic replacements therefore use stale snapshots and the last replacement silently wins. Files Needing Attention: src/praisonai-agents/praisonaiagents/memory/file_memory.py
|
| Filename | Overview |
|---|---|
| src/praisonai-agents/praisonaiagents/agent/handoff.py | Tracks whether each handoff invocation pushed chain state before conditionally popping it during cleanup. |
| src/praisonai-agents/praisonaiagents/agents/agents.py | Adds an exception-isolating gather helper and applies it to workflow and sequential async batches. |
| src/praisonai-agents/praisonaiagents/memory/file_memory.py | Uses unique temporary files and atomic replacement, but locks distinct temporary inodes and therefore does not serialize concurrent updates to one destination. |
Reviews (2): Last reviewed commit: "fix: use unique temp name for atomic mem..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
|
@claude You are the FINAL architecture reviewer. If the branch is under MervinPraison/PraisonAI (not a fork), you are able to make modifications to this branch and push directly. SCOPE: Focus ONLY on Python packages (praisonaiagents, praisonai). Do NOT modify praisonai-rust or praisonai-ts. Read ALL comments above from Gemini, Qodo, CodeRabbit, and Copilot carefully before responding. Phase 1: Review per AGENTS.md
Phase 2: FIX Valid Issues Phase 3: Final Verdict |
MervinPraison
added
pipeline/awaiting-merge-gate
|
Claude finished @MervinPraison's task —— View job Final Architecture Review — PR #3586 ✅
Phase 1 — Review (AGENTS.md)All three fixes are correctness bug fixes in the core SDK, not scope creep. No new params, exports, dependencies, or Agent-class bloat. Zero feature-surface growth. Correctly routed (core SDK behavior, not tools/plugins). Backward-compatible; no hot-path regressions (atomic write already existed; only temp-naming changed).
Phase 2 — Fix applied
Verification (same-process concurrency): 4 threads × 20 writes to one file → all 80 items persisted, no leftover Skipped: nothing. No Phase 3 — Verdict: Approve ✅All three fixes are valid, minimal, backward-compatible core-SDK correctness improvements. The one blocking issue (Greptile P1 temp-name collision) is now resolved and matches SDK convention. Tests green (73 passed, 3 skipped).
Files modified this pass: |
Replace the fixed .tmp.<pid> temp filename with tempfile.mkstemp so concurrent writers in the same process (multiple threads or FileMemory instances writing the same file) cannot clobber each other's temp file, where the first os.replace would remove the temp the sibling still needs. Matches the atomic-write convention used elsewhere in the SDK (storage/base.py, session/store.py). Addresses Greptile P1. Co-authored-by: Mervin Praison <MervinPraison@users.noreply.github.com>
MervinPraison
added
pipeline/blocked:stale-final
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/praisonai-agents/praisonaiagents/agents/agents.py`:
- Around line 1359-1362: Update _gather_with_isolation to preserve and raise the
first exception observed at runtime, rather than selecting the first exception
returned in asyncio.gather input order. Capture failures as each coroutine
completes, while retaining successful results and existing isolation behavior.
In `@src/praisonai-agents/praisonaiagents/memory/file_memory.py`:
- Around line 259-264: Update _write_json so temporary-file cleanup runs in a
finally block covering both successful writes and failures, including TypeError
from json.dump. In that cleanup, call tmp_path.unlink and ignore
FileNotFoundError to handle files already removed by os.replace; retain the
existing error logging and False return behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: edd11601-94c2-4e9a-ba0a-666ad0046809
📒 Files selected for processing (3)
src/praisonai-agents/praisonaiagents/agent/handoff.pysrc/praisonai-agents/praisonaiagents/agents/agents.pysrc/praisonai-agents/praisonaiagents/memory/file_memory.py
| results = await asyncio.gather(*coros, return_exceptions=True) | ||
| for result in results: | ||
| if isinstance(result, BaseException): | ||
| raise result |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
python - <<'PY'
import asyncio
class SlowError(Exception):
pass
class FastError(Exception):
pass
async def slow_failure():
await asyncio.sleep(0.01)
raise SlowError
async def fast_failure():
raise FastError
async def current_helper(coros):
results = await asyncio.gather(*coros, return_exceptions=True)
for result in results:
if isinstance(result, BaseException):
raise result
async def main():
try:
await current_helper([slow_failure(), fast_failure()])
except FastError:
return
except SlowError as exc:
raise SystemExit("FAIL: input-order exception masked FastError") from exc
raise SystemExit("FAIL: no exception propagated")
asyncio.run(main())
PYRepository: MervinPraison/PraisonAI
Length of output: 206
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
# Locate and inspect the helper plus nearby call sites without running repository code.
sed -n '1330,1415p' src/praisonai-agents/praisonaiagents/agents/agents.py
printf '\n--- references to _gather_with_isolation ---\n'
rg -n "_gather_with_isolation|asyncio\\.gather\\(" src/praisonai-agents/praisonaiagents/agents/agents.pyRepository: MervinPraison/PraisonAI
Length of output: 4732
Preserve the first observed exception.
_gather_with_isolation selects the first exception in input order, not the first coroutine exception that raises. If a later coroutine raises before an earlier slower coroutine settles, the earlier exception is raised instead. Store and raise the first exception when a coroutine fails, or document input-order selection with a regression test that covers a fast-failing later coroutine.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/praisonai-agents/praisonaiagents/agents/agents.py` around lines 1359 -
1362, Update _gather_with_isolation to preserve and raise the first exception
observed at runtime, rather than selecting the first exception returned in
asyncio.gather input order. Capture failures as each coroutine completes, while
retaining successful results and existing isolation behavior.
Source: Coding guidelines
| except (IOError, OSError) as e: | ||
| self._log(f"Error writing {filepath}: {e}", logging.ERROR) | ||
| try: | ||
| tmp_path.unlink() | ||
| except OSError: | ||
| pass |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the relevant file and method without running repository code.
wc -l src/praisonai-agents/praisonaiagents/memory/file_memory.py
sed -n '220,285p' src/praisonai-agents/praisonaiagents/memory/file_memory.py
printf '\nRelevant imports/open/os calls:\n'
sed -n '1,80p' src/praisonai-agents/praisonaiagents/memory/file_memory.py
printf '\nStructural context for write_to_file/save functions:\n'
ast-grep outline src/praisonai-agents/praisonaiagents/memory/file_memory.py || true
printf '\nSearch for temporary-file writes/unlinks in file_memory:\n'
rg -n "Tmp|temp|mktemp|tmp_path|unlink|json\.dump|replace|FileNotFoundError" src/praisonai-agents/praisonaiagents/memory/file_memory.pyRepository: MervinPraison/PraisonAI
Length of output: 7771
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Behavioral probe: show what exceptions json.dump can raise from non-json-serializable objects and how the current try/except control flow would handle them.
python3 - <<'PY'
import json, io
buf = io.StringIO()
for data in [[lambda: 1], (lambda: 1,), {lambda: 1: 'x'}]:
try:
json.dump(data, buf, indent=2, ensure_ascii=False)
except Exception as e:
print(type(e).__name__, str(e), 'is instance of (IOError,OSError):', isinstance(e, (IOError, OSError)))
PYRepository: MervinPraison/PraisonAI
Length of output: 466
Clean up the temporary file outside the IOError/OSError handler.
json.dump can fail with TypeError, so _write_json() can return False while the temp file remains on disk. Move tmp_path.unlink() to finally, and catch FileNotFoundError there to avoid leaking after a successful os.replace.
Clean up in finally
os.replace(tmp_path, filepath)
return True
except (IOError, OSError) as e:
self._log(f"Error writing {filepath}: {e}", logging.ERROR)
- try:
- tmp_path.unlink()
- except OSError:
- pass
return False
+finally:
+ try:
+ tmp_path.unlink()
+ except (FileNotFoundError, OSError):
+ pass📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except (IOError, OSError) as e: | |
| self._log(f"Error writing {filepath}: {e}", logging.ERROR) | |
| try: | |
| tmp_path.unlink() | |
| except OSError: | |
| pass | |
| os.replace(tmp_path, filepath) | |
| return True | |
| except (IOError, OSError) as e: | |
| self._log(f"Error writing {filepath}: {e}", logging.ERROR) | |
| return False | |
| finally: | |
| try: | |
| tmp_path.unlink() | |
| except (FileNotFoundError, OSError): | |
| pass |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/praisonai-agents/praisonaiagents/memory/file_memory.py` around lines 259
- 264, Update _write_json so temporary-file cleanup runs in a finally block
covering both successful writes and failures, including TypeError from
json.dump. In that cleanup, call tmp_path.unlink and ignore FileNotFoundError to
handle files already removed by os.replace; retain the existing error logging
and False return behavior.
| if _HAS_FCNTL: | ||
| fcntl.flock(f.fileno(), fcntl.LOCK_EX) |
There was a problem hiding this comment.
If two agents or FileMemory instances using the same user ID add episodic entries concurrently, each writer locks a different temporary inode after reading the same destination state. Both replacements then succeed, and the later replacement silently discards the other entry.
Knowledge Base Used: praisonai-agents Core Library
praisonai-triage-agent
Bot
removed
pipeline/blocked:ci
1 participant
Fixes #3585
Summary
Three independent core-SDK correctness fixes. All are minimal and backward-compatible — no new params, exports, or dependencies.
1. Async fan-out exception isolation (
agents/agents.py)asyncio.gather(*tasks)used the defaultreturn_exceptions=False, so one failing async task propagated immediately and left its siblings running in the background, mutating sharedself.tasksstate out of band. Added_gather_with_isolation()(usesreturn_exceptions=True, awaits all siblings, then re-raises the first exception) and applied it at all 3 gather sites (workflow + sequential flush).2. Handoff safety-chain corruption (
agent/handoff.py)_check_safety()raises before_push_handoff(), butfinally: _pop_handoff()ran unconditionally — so a rejected handoff popped an ancestor's entry, silently eroding the cycle/depth guard. Guarded eachfinallywith apushedflag at all 5 call sites (sync/asyncHandoff.execute+ LLM tool + sync/asyncTypedHandoff).3. Non-atomic memory writes (
memory/file_memory.py)open(path, 'w')truncated the memory file beforeflockcould be acquired, so a crash mid-write left it permanently empty. Now writes to a temp file,fsyncs, andos.replaces atomically into place (mirrors the existingBaseJSONStorepattern).Test plan
tests/test_file_memory.py,tests/unit/test_handoff_unified.py— pass (73 passed, 3 skipped)test_handoff_tool_policy.pyconfirmed unrelated (fail identically on base commit)Generated with Claude Code
Summary by CodeRabbit