Merge pull request #6251 from MervinPraison/claude/issue-6248-20261105

fix: session manifest with write-ahead run-in-flight marker (#6248)

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

@@ -192,6 +192,12 @@ def __getattr__(name: str):
         _module_cache[name] = workspace_id
         return workspace_id
 
+    # Session manifest: plane refs + run-in-flight marker (Issue #6248)
+    if name in ("FileSessionManifest", "ManifestEntry", "SessionManifestProtocol"):
+        from . import manifest as _manifest
+        _module_cache[name] = getattr(_manifest, name)
+        return _module_cache[name]
+
     # Cross-context continuation prompt assembler (session handoff)
     if name == "build_handoff_prompt":
         from .handoff import build_handoff_prompt
@@ -249,6 +255,9 @@ __all__ = [
     "clear_session_context",
     "workspace_id",
     "build_handoff_prompt",
+    "FileSessionManifest",
+    "ManifestEntry",
+    "SessionManifestProtocol",
     "MessageOrigin",
     "wrap_inter_agent",
     # Encryption at rest

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

@@ -0,0 +1,220 @@
+"""
+Session manifest: one durable record per ``session_id`` holding a write-ahead
+"run in flight" marker (Issue #6248), so a later launch can tell that a run on
+this session died before reaching a terminal outcome.
+
+Usage::
+
+    from praisonaiagents.session.manifest import FileSessionManifest
+
+    manifest = FileSessionManifest()
+    entry = manifest.begin_run("my-session", "run-42", run_input="fix the bug")
+    ...  # run the agent
+    manifest.end_run("my-session", "completed", run_id="run-42")
+
+    # Next launch:
+    stale = manifest.interrupted("my-session")   # entry or None
+
+The run input itself is never written to disk: only a SHA-256 digest of it
+is kept (``run_digest``), enough to recognise a retry of the same task without
+storing prompts (which may contain secrets) in plaintext.
+
+A run that keeps dying is bounded by ``max_attempts``: once the budget is
+spent, ``begin_run`` logs the stale run as abandoned and restarts the attempt
+counter instead of counting it up forever.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import tempfile
+import threading
+import time
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Dict, Iterator, Optional, Protocol, runtime_checkable
+
+from .store import FileLock, _safe_session_filename
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_MAX_ATTEMPTS = 3
+
+
+@dataclass
+class ManifestEntry:
+    """A session's last run state."""
+
+    session_id: str
+    run_id: Optional[str] = None
+    run_status: Optional[str] = None
+    run_in_flight: bool = False
+    run_digest: Optional[str] = None
+    attempt: int = 0
+    updated_at: float = 0.0
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "ManifestEntry":
+        return cls(**{k: data[k] for k in cls.__dataclass_fields__ if k in data})
+
+
+def _digest(run_input: Optional[str]) -> Optional[str]:
+    if run_input is None:
+        return None
+    return hashlib.sha256(run_input.encode("utf-8")).hexdigest()
+
+
+@runtime_checkable
+class SessionManifestProtocol(Protocol):
+    """Per-session run-in-flight marker."""
+
+    def get(self, session_id: str) -> Optional[ManifestEntry]: ...
+
+    def begin_run(
+        self,
+        session_id: str,
+        run_id: str,
+        run_input: Optional[str] = None,
+        max_attempts: int = DEFAULT_MAX_ATTEMPTS,
+    ) -> ManifestEntry: ...
+
+    def end_run(
+        self, session_id: str, status: str, run_id: Optional[str] = None
+    ) -> Optional[ManifestEntry]: ...
+
+    def interrupted(self, session_id: str) -> Optional[ManifestEntry]: ...
+
+
+class FileSessionManifest:
+    """Default JSON-file manifest: ``<dir>/<session_id>.json``, atomic writes."""
+
+    def __init__(self, directory: Optional[str] = None):
+        """``directory`` defaults to ``<sessions>/manifests``; callers whose
+        session store is scoped (e.g. per project) should pass the matching
+        directory so equal ids in different scopes never share a manifest."""
+        if directory is None:
+            from ..paths import get_sessions_dir
+
+            directory = str(get_sessions_dir() / "manifests")
+        self._dir = Path(directory)
+        self._lock = threading.Lock()
+
+    def _path(self, session_id: str) -> Path:
+        if not session_id:
+            raise ValueError("session_id is required")
+        # Same injective naming as the conversation store: ids that differ
+        # only in ``:``/``.``/``/`` never share a file, and none escape ``_dir``.
+        return self._dir / f"{_safe_session_filename(session_id)}.json"
+
+    @contextmanager
+    def _locked(self, session_id: str) -> Iterator[None]:
+        """Serialise read-modify-write across threads *and* processes."""
+        with self._lock, FileLock(str(self._path(session_id))):
+            yield
+
+    def _read(self, session_id: str) -> Optional[ManifestEntry]:
+        path = self._path(session_id)
+        try:
+            with open(path, "r", encoding="utf-8") as fh:
+                data = json.load(fh)
+        except FileNotFoundError:
+            return None
+        except (OSError, ValueError) as exc:
+            logger.warning("Unreadable session manifest %s: %s", path, exc)
+            return None
+        if not isinstance(data, dict):
+            return None
+        data["session_id"] = session_id
+        return ManifestEntry.from_dict(data)
+
+    def _write(self, entry: ManifestEntry) -> ManifestEntry:
+        entry.updated_at = time.time()
+        path = self._path(entry.session_id)
+        self._dir.mkdir(parents=True, exist_ok=True)
+        fd, tmp = tempfile.mkstemp(dir=str(self._dir), suffix=".tmp")
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8") as fh:
+                json.dump(asdict(entry), fh, default=str)
+                fh.flush()
+                os.fsync(fh.fileno())
+            os.replace(tmp, path)
+        except BaseException:
+            try:
+                os.unlink(tmp)
+            except OSError:
+                pass
+            raise
+        return entry
+
+    def get(self, session_id: str) -> Optional[ManifestEntry]:
+        with self._lock:
+            return self._read(session_id)
+
+    def begin_run(
+        self,
+        session_id: str,
+        run_id: str,
+        run_input: Optional[str] = None,
+        max_attempts: int = DEFAULT_MAX_ATTEMPTS,
+    ) -> ManifestEntry:
+        """Commit the write-ahead in-flight marker for a run on this session.
+
+        Only a digest of ``run_input`` is persisted, never the text itself.
+        If the previous run never released its marker it was interrupted; a new
+        run with the same ``run_input`` counts as another attempt of it. Once
+        ``max_attempts`` is exceeded the stale run is abandoned and the counter
+        restarts, so a run that keeps dying is not counted up forever.
+        """
+        with self._locked(session_id):
+            entry = self._read(session_id) or ManifestEntry(session_id=session_id)
+            digest = _digest(run_input)
+            if entry.run_in_flight and entry.run_digest == digest:
+                attempt = entry.attempt + 1
+                if attempt > max_attempts:
+                    logger.warning(
+                        "Session %s: run %s abandoned after %d attempts",
+                        session_id, entry.run_id, entry.attempt,
+                    )
+                    attempt = 1
+            else:
+                attempt = 1
+            entry.run_id = run_id
+            entry.run_digest = digest
+            entry.run_status = "running"
+            entry.run_in_flight = True
+            entry.attempt = attempt
+            return self._write(entry)
+
+    def end_run(
+        self, session_id: str, status: str, run_id: Optional[str] = None
+    ) -> Optional[ManifestEntry]:
+        """Release the in-flight marker with a terminal ``status``.
+
+        With ``run_id``, only that run's own marker is released: if another
+        run on the same session has since taken the marker, it is left alone
+        (returns ``None``) so that run's later crash is still reported.
+        """
+        with self._locked(session_id):
+            entry = self._read(session_id)
+            if entry is None or (run_id is not None and entry.run_id != run_id):
+                return None
+            entry.run_in_flight = False
+            entry.run_status = status
+            return self._write(entry)
+
+    def interrupted(self, session_id: str) -> Optional[ManifestEntry]:
+        """Return the entry if its last run never reached a terminal outcome."""
+        entry = self.get(session_id)
+        return entry if entry is not None and entry.run_in_flight else None
+
+
+__all__ = [
+    "DEFAULT_MAX_ATTEMPTS",
+    "FileSessionManifest",
+    "ManifestEntry",
+    "SessionManifestProtocol",
+]

src/praisonai-agents/tests/unit/session/test_session_manifest.py

@@ -0,0 +1,94 @@
+"""Tests for the session manifest (Issue #6248)."""
+
+from praisonaiagents.session import FileSessionManifest, SessionManifestProtocol
+
+
+def test_satisfies_protocol(tmp_path):
+    assert isinstance(FileSessionManifest(str(tmp_path)), SessionManifestProtocol)
+
+
+def test_missing_session_has_no_entry(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    assert m.get("missing") is None and m.interrupted("missing") is None
+
+
+def test_legacy_entry_with_unknown_keys_still_loads(tmp_path):
+    (tmp_path / "s1.json").write_text(
+        '{"session_id": "s1", "refs": {"plan": "p"}, "run_in_flight": true}',
+        encoding="utf-8",
+    )
+    assert FileSessionManifest(str(tmp_path)).interrupted("s1") is not None
+
+
+def test_completed_run_is_not_interrupted(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    m.begin_run("s1", "r1", run_input="task")
+    assert m.interrupted("s1") is not None
+    m.end_run("s1", "completed")
+    assert m.interrupted("s1") is None
+    assert m.get("s1").run_status == "completed"
+
+
+def test_interrupted_run_survives_new_process(tmp_path):
+    FileSessionManifest(str(tmp_path)).begin_run("s1", "r1", run_input="task")
+    # A new process (new instance) sees the surviving write-ahead marker.
+    stale = FileSessionManifest(str(tmp_path)).interrupted("s1")
+    assert stale is not None and stale.run_id == "r1" and stale.run_digest
+
+
+def test_attempt_budget_is_bounded(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    attempts = [m.begin_run("s1", f"r{i}", run_input="task", max_attempts=2).attempt
+                for i in range(4)]
+    # Retries of the same interrupted task count up, then the budget resets.
+    assert attempts == [1, 2, 1, 2]
+
+
+def test_new_task_after_interruption_is_a_fresh_attempt(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    m.begin_run("s1", "r1", run_input="task a")
+    assert m.begin_run("s1", "r2", run_input="task b").attempt == 1
+
+
+def test_unsafe_session_id_stays_in_directory(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    m.begin_run("../../etc/x", "r1")
+    assert all(p.parent == tmp_path for p in tmp_path.iterdir())
+    assert m.get("../../etc/x").run_id == "r1"
+
+
+def test_run_input_is_never_written_in_plaintext(tmp_path):
+    secret = "deploy with token sk-SECRET-12345"
+    m = FileSessionManifest(str(tmp_path))
+    m.begin_run("s1", "r1", run_input=secret)
+    for f in tmp_path.iterdir():
+        assert "sk-SECRET-12345" not in f.read_text(encoding="utf-8")
+    # The digest still recognises a retry of the same task.
+    assert m.begin_run("s1", "r2", run_input=secret).attempt == 2
+
+
+def test_absolute_and_backslash_ids_stay_in_directory(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    for sid in ("/etc/passwd", "..\\..\\x", "..", "C:\\x"):
+        m.begin_run(sid, sid)
+        assert m.get(sid).run_id == sid
+    assert all(p.parent == tmp_path for p in tmp_path.iterdir())
+
+
+def test_ids_differing_only_in_punctuation_stay_separate(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    m.begin_run("a:b", "colon")
+    m.begin_run("a_b", "underscore")
+    assert m.get("a:b").run_id == "colon"
+    assert m.get("a_b").run_id == "underscore"
+
+
+def test_end_run_only_releases_its_own_marker(tmp_path):
+    m = FileSessionManifest(str(tmp_path))
+    m.begin_run("s1", "r1", run_input="first")
+    m.begin_run("s1", "r2", run_input="second")  # second terminal takes over
+    assert m.end_run("s1", "completed", run_id="r1") is None
+    stale = m.interrupted("s1")
+    assert stale is not None and stale.run_id == "r2"
+    m.end_run("s1", "completed", run_id="r2")
+    assert m.interrupted("s1") is None

src/praisonai-code/praisonai_code/cli/commands/run.py

@@ -2393,6 +2393,8 @@ def _run_prompt(
     if allow_local_tools:
         _os.environ[_ALLOW_LOCAL_TOOLS_ENV] = "true"
 
+    _manifest_session = None
+
     try:
         # Handle session continuity first (before any execution mode)
         from praisonai_code.cli.main import PraisonAI
@@ -2442,6 +2444,13 @@ def _run_prompt(
             import uuid
             auto_save_name = session_id or "session-" + str(uuid.uuid4())[:8]
 
+        # Write-ahead "run in flight" marker on the session manifest, released
+        # in ``finally`` below; a run that dies before then is reported as
+        # interrupted on the next --continue/--session (Issue #6248).
+        _manifest_session = _begin_manifest_run(
+            session_id or auto_save_name, prompt, output, quiet=_json_stdout
+        )
+
         # Detect-and-attach: if a warm runtime is running, forward the plain
         # prompt to it (skipping per-invocation cold-start) and fall back to
         # in-process execution otherwise. Only the simple text path attaches;
@@ -2774,6 +2783,7 @@ def _run_prompt(
             output.print_error(str(e))
         raise typer.Exit(1)
     finally:
+        _end_manifest_run(_manifest_session)
         # Restore the prior opt-in state so the grant is strictly per-invocation.
         if _prev_allow_local_tools is None:
             _os.environ.pop(_ALLOW_LOCAL_TOOLS_ENV, None)
@@ -2785,6 +2795,57 @@ def _run_prompt(
             output.mode = _prev_mode
 
 
+def _begin_manifest_run(session_key, prompt, output, *, quiet: bool = False):
+    """Commit the session manifest's in-flight marker for this run.
+
+    If the previous run on this session never released its marker it was
+    interrupted; say so, so the user can re-issue it. Returns the session key
+    a ``(manifest, session_key, run_id)`` handle to release later, or ``None``
+    when there is no session or the manifest is unavailable (manifest problems
+    never block a run).
+    """
+    if not session_key:
+        return None
+    try:
+        import uuid
+        from praisonaiagents.session.manifest import FileSessionManifest
+        from ..utils.project import get_project_sessions_dir
+
+        # Scope manifests like the project session store, so equal session
+        # names in different projects never share a run marker.
+        manifest = FileSessionManifest(str(get_project_sessions_dir() / "manifests"))
+        stale = manifest.interrupted(session_key)
+        if stale is not None and not quiet:
+            # The prompt itself is not stored (only a digest), so it is not echoed.
+            output.print_warning(
+                f"The previous run in this session was interrupted "
+                f"(attempt {stale.attempt}); re-issue it to retry."
+            )
+        run_id = uuid.uuid4().hex[:12]
+        manifest.begin_run(session_key, run_id, run_input=prompt)
+        return manifest, session_key, run_id
+    except Exception:
+        return None
+
+
+def _end_manifest_run(handle) -> None:
+    """Release this run's own in-flight marker unless it was interrupted (Ctrl+C)."""
+    if not handle:
+        return
+    import sys
+
+    exc = sys.exc_info()[1]
+    if isinstance(exc, KeyboardInterrupt):
+        return
+    manifest, session_key, run_id = handle
+    try:
+        manifest.end_run(
+            session_key, "completed" if exc is None else "ended", run_id=run_id
+        )
+    except Exception:
+        pass
+
+
 def _record_session_usage(session_id, model, output, *, suppress_info: bool = False) -> None:
     """Accumulate this run's token/cost usage into the active session and show
     a compact running total footer (Issue #2421).