Fix session reset: move JSONL to inbox instead of in-process branching

agent.reset() now moves pi-session.jsonl to staging/inbox/ and restarts
Pi fresh, severing the parentId chain that caused unbounded context growth.

- Replace SessionReader watermark class with parse_session() function
- Add read_sessions ctools command for reflect skill
- Simplify reset_clarvis_session (parallel resets, remove redundant voice disconnect)
- Add STAGING_INBOX/STAGING_DIGESTED constants to paths.py
- Update reflect skill to use read_sessions on inbox files

Author: shepardxia

clarvis/agent/agent.py

@@ -9,11 +9,14 @@
 import json
 import logging
 import signal
+import time
 from collections.abc import AsyncGenerator
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from ..core.paths import STAGING_INBOX
+
 if TYPE_CHECKING:
     from .context import ContextInjector
 
@@ -230,10 +233,18 @@ async def shutdown(self) -> None:
         await self.disconnect()
 
     async def reset(self) -> None:
-        """Reset the Pi session (new conversation, retains JSONL history)."""
-        if not self._connected:
-            return
-        self._send_command({"type": "new_session"})
+        """Reset by moving session file to inbox and restarting Pi fresh."""
+        was_connected = self._connected
+        await self.disconnect()
+
+        if self._session_file.exists():
+            STAGING_INBOX.mkdir(parents=True, exist_ok=True)
+            dest = STAGING_INBOX / f"session_{self._session_key}_{int(time.time())}.jsonl"
+            self._session_file.rename(dest)
+            logger.info("Moved session to %s", dest.name)
+
+        if was_connected:
+            await self.connect()
         if self.context:
             self.context.reset()
 

clarvis/core/commands/agent.py

@@ -33,46 +33,37 @@ def reload_agents(self: CommandHandlers, **kwargs) -> dict:
 
 
 def reset_clarvis_session(self: CommandHandlers, **kw) -> str:
-    """Reset Clarvis agent session (new_session RPC)."""
+    """Reset all agent sessions — moves session files to inbox and restarts."""
     import asyncio
 
-    from ..paths import CLARVIS_HOME, agent_home
-
-    # Flush unreflected session content to inbox before resetting
-    session_reader = self._get_service("session_reader")
-    if session_reader:
-        inbox = CLARVIS_HOME / "staging" / "inbox"
-        for source in ("clarvis", "factoria"):
-            try:
-                session_reader.flush_to_inbox(source, inbox)
-            except Exception as exc:
-                logger.warning("Failed to flush %s session to inbox: %s", source, exc)
+    from ..paths import agent_home
 
     for sid_file in [
         agent_home("clarvis") / "session_id",
         agent_home("factoria") / "session_id",
     ]:
         sid_file.unlink(missing_ok=True)
 
-    # Reset the Clarvis agent session
+    # Reset both agents in parallel (each handles its own file move + restart)
     agents = self._get_service("agents") or {}
-    clarvis_agent = agents.get("clarvis")
-    if clarvis_agent and clarvis_agent.connected:
-        try:
-            asyncio.run_coroutine_threadsafe(clarvis_agent.reset(), self.ctx.loop).result(timeout=30)
-        except Exception as exc:
-            logger.warning("Failed to reset Clarvis agent: %s", exc)
+    if agents:
+
+        async def _reset_all():
+            results = await asyncio.gather(
+                *(a.reset() for a in agents.values()),
+                return_exceptions=True,
+            )
+            for name, result in zip(agents, results):
+                if isinstance(result, Exception):
+                    logger.warning("Failed to reset %s agent: %s", name, result)
 
-    # Disconnect voice orchestrator's agent if active
-    orchestrator = self._get_service("voice")
-    if orchestrator and orchestrator.agent.connected:
-        asyncio.run_coroutine_threadsafe(orchestrator.agent.disconnect(), orchestrator._loop)
+        asyncio.run_coroutine_threadsafe(_reset_all(), self.ctx.loop).result(timeout=30)
 
     return "ok"
 
 
 def reflect_complete(self: CommandHandlers, **kw) -> dict:
-    """Signal that reflect is done — advance watermarks and reset agents."""
+    """Signal that reflect is done — archive inbox and reset agents."""
     import asyncio
 
     daemon = self._get_service("daemon")
@@ -114,10 +105,21 @@ def nudge_agent(self: CommandHandlers, *, reason: str = "timer", **kw) -> dict:
         return {"error": str(exc)}
 
 
+def read_sessions(self: CommandHandlers, *, path: str, **kw) -> dict:
+    """Parse a Pi session JSONL file and return structured messages."""
+    from pathlib import Path
+
+    from ...memory.session_reader import parse_session
+
+    messages = parse_session(Path(path))
+    return {"messages": messages, "count": len(messages)}
+
+
 COMMANDS: dict[str, str] = {
     "reload_agents": "reload_agents",
     "reset_clarvis_session": "reset_clarvis_session",
     "reflect_complete": "reflect_complete",
     "listen": "listen",
     "nudge": "nudge_agent",
+    "read_sessions": "read_sessions",
 }

clarvis/core/paths.py

@@ -1,6 +1,9 @@
 from pathlib import Path
 
 CLARVIS_HOME = Path.home() / ".clarvis"
+STAGING_DIR = CLARVIS_HOME / "staging"
+STAGING_INBOX = STAGING_DIR / "inbox"
+STAGING_DIGESTED = STAGING_DIR / "digested"
 
 
 def agent_home(name: str) -> Path:

clarvis/daemon.py

@@ -22,7 +22,7 @@
 from .core.commands import CommandHandlers
 from .core.context import AppContext
 from .core.ipc import DaemonServer
-from .core.paths import CLARVIS_HOME
+from .core.paths import STAGING_DIGESTED, STAGING_INBOX
 from .core.persistence import json_load_safe
 from .core.scheduler import Scheduler
 from .core.signals import SignalBus
@@ -128,8 +128,6 @@ def __init__(
         self.voice_orchestrator = None
         self.channel_manager = None
         self._chat_bridge = None
-        self.staging_dir = CLARVIS_HOME / "staging"
-        self._session_reader = None
         self._owned_services: list = []  # services with no other refs (prevent GC)
 
         # Deferred — initialized in run()
@@ -700,25 +698,24 @@ async def stop(self) -> None:
                 pass
 
     async def reset_all_agents(self) -> None:
-        """Reset all agent sessions (called after reflect)."""
-        for name, agent in self._agents.items():
-            try:
-                await agent.reset()
+        """Reset all agent sessions in parallel."""
+        results = await asyncio.gather(
+            *(agent.reset() for agent in self._agents.values()),
+            return_exceptions=True,
+        )
+        for name, result in zip(self._agents, results):
+            if isinstance(result, Exception):
+                logger.warning("Failed to reset agent %s: %s", name, result, exc_info=True)
+            else:
                 logger.info("Reset agent session: %s", name)
-            except Exception:
-                logger.warning("Failed to reset agent %s", name, exc_info=True)
 
     async def complete_reflect(self) -> dict:
-        """Finalize reflect: advance watermarks, archive inbox, reset agents."""
-        if self._session_reader:
-            self._session_reader.advance_all()
-        inbox = self.staging_dir / "inbox"
-        if inbox.is_dir():
-            read_dir = self.staging_dir / "digested"
-            read_dir.mkdir(exist_ok=True)
-            for f in inbox.glob("*"):
+        """Finalize reflect: move inbox → digested, then reset agents."""
+        if STAGING_INBOX.is_dir():
+            STAGING_DIGESTED.mkdir(parents=True, exist_ok=True)
+            for f in STAGING_INBOX.glob("*"):
                 if f.is_file():
-                    f.rename(read_dir / f.name)
+                    f.rename(STAGING_DIGESTED / f.name)
         await self.reset_all_agents()
         return {"status": "reflect complete"}
 
@@ -741,18 +738,6 @@ async def run(self) -> None:
         # Initialize display pipeline (scene, socket, rendering)
         self._init_display()
 
-        # Create SessionReader for Pi session files
-        from .core.paths import agent_home
-        from .memory.session_reader import SessionReader
-
-        self._session_reader = SessionReader(
-            sources={
-                "clarvis": agent_home("clarvis") / "pi-session.jsonl",
-                "factoria": agent_home("factoria") / "pi-session.jsonl",
-            },
-            watermark_file=self.staging_dir / "session_watermarks.json",
-        )
-
         self.commands = CommandHandlers(
             ctx=self.ctx,
             session_tracker=self.session_tracker,
@@ -766,7 +751,6 @@ async def run(self) -> None:
                 "timer_service": lambda: self.timer_service,
                 "channel_manager": lambda: self.channel_manager,
                 "daemon": lambda: self,
-                "session_reader": lambda: self._session_reader,
             },
         )
 

clarvis/data/skills/reflect/SKILL.md

@@ -11,10 +11,11 @@ Memory maintenance — extract facts from conversation history and staging inbox
 
 Three sources to check:
 1. **Your current session** — already in your context. Extract facts from what you know.
-2. **Inbox** — check `~/.clarvis/staging/inbox/` for prior session transcripts (flushed automatically on session reset), user-submitted summaries, and staged files.
-3. **Factoria transcript** — read `~/.clarvis/factoria/pi-session.jsonl` for Factoria's conversations since last reflect.
+2. **Inbox sessions** — list `~/.clarvis/staging/inbox/` for `session_*.jsonl` files. Parse each with `ctools read_sessions '{"path": "<file>"}'`.
+3. **Factoria's live session** — parse `~/.clarvis/factoria/pi-session.jsonl` with `read_sessions` (still active, not in inbox).
+4. **Other inbox items** — check for non-session files in inbox (user-submitted summaries from `/remember`, staged markdown files).
 
-If nothing new across all three sources, report "nothing to reflect on" and stop.
+If nothing new across all sources, report "nothing to reflect on" and stop.
 
 ## Phase 2: Extract facts
 
@@ -58,5 +59,5 @@ Guidelines:
 
 ## Phase 5: Complete
 
-1. Call `reflect_complete` to reset all agent sessions
+1. Call `reflect_complete` — moves inbox to `staging/digested/`, resets all agent sessions (current sessions move to inbox and restart fresh)
 2. Report what was processed

clarvis/memory/session_reader.py

@@ -1,27 +1,22 @@
-"""Multi-source JSONL session reader with per-source watermarks.
-
-Reads Pi session files (pi-session.jsonl) incrementally using byte-offset
-watermarks. Each source (e.g. clarvis, factoria) has an independent watermark.
-"""
+"""Parse Pi session JSONL files into structured messages."""
 
 import json
 import logging
-import time
 from pathlib import Path
 
-from clarvis.core.persistence import json_load_safe, json_save_atomic
-
 logger = logging.getLogger(__name__)
 
-# Entry types that are metadata, not conversation content.
-_SKIP_TYPES = {"session", "model_change", "thinking_level_change"}
 
+def parse_session(path: Path) -> list[dict[str, str]]:
+    """Parse a Pi session JSONL file, return [{"role": ..., "text": ...}].
 
-def _parse_pi_messages(raw: str) -> list[dict[str, str]]:
-    """Parse Pi JSONL and extract user/assistant text messages.
-
-    Returns list of {"role": ..., "text": ...} dicts.
+    Extracts user and assistant text messages, skipping metadata entries
+    (session, model_change, system prompts, etc.).
     """
+    try:
+        raw = path.read_text(encoding="utf-8", errors="replace")
+    except FileNotFoundError:
+        return []
     messages = []
     for line in raw.splitlines():
         line = line.strip()
@@ -48,113 +43,3 @@ def _parse_pi_messages(raw: str) -> list[dict[str, str]]:
         if text_parts:
             messages.append({"role": role, "text": "\n".join(text_parts)})
     return messages
-
-
-def _filter_for_inbox(raw: str) -> list[str]:
-    """Filter JSONL lines for inbox dump.
-
-    Keeps all message entries (user, assistant, toolResult — including
-    ambient context, tool calls, thinking blocks). Drops session metadata
-    and system prompts.
-    """
-    kept = []
-    for line in raw.splitlines():
-        line = line.strip()
-        if not line:
-            continue
-        try:
-            entry = json.loads(line)
-        except json.JSONDecodeError:
-            continue
-        if entry.get("type") in _SKIP_TYPES:
-            continue
-        if entry.get("type") == "message":
-            role = entry.get("message", {}).get("role")
-            if role == "system":
-                continue
-        kept.append(line)
-    return kept
-
-
-class SessionReader:
-    """Reads multiple Pi session JSONL files with per-source byte watermarks."""
-
-    def __init__(self, sources: dict[str, Path], watermark_file: Path) -> None:
-        self._sources = {k: Path(v) for k, v in sources.items()}
-        self._watermark_file = Path(watermark_file)
-        self._watermarks: dict[str, int] = json_load_safe(self._watermark_file) or {}
-        self._pending_offsets: dict[str, int] = {}
-
-    def _read_since_watermark(self, source: str) -> tuple[str, int] | None:
-        """Read raw bytes from *source* since its watermark.
-
-        Returns (raw_text, file_size) or None if nothing new.
-        """
-        path = self._sources.get(source)
-        if not path or not path.exists():
-            return None
-        watermark = self._watermarks.get(source, 0)
-        file_size = path.stat().st_size
-        if watermark >= file_size:
-            return None
-        with open(path, "rb") as f:
-            f.seek(watermark)
-            raw = f.read().decode("utf-8", errors="replace")
-        return raw, file_size
-
-    def read_pending(self) -> dict[str, list[dict[str, str]]]:
-        """Read new messages from all sources since their watermarks.
-
-        Returns {source_name: [{"role": ..., "text": ...}, ...]}.
-        """
-        result: dict[str, list[dict[str, str]]] = {}
-        for name in self._sources:
-            chunk = self._read_since_watermark(name)
-            if chunk is None:
-                result[name] = []
-                continue
-            raw, file_size = chunk
-            result[name] = _parse_pi_messages(raw)
-            self._pending_offsets[name] = file_size
-        return result
-
-    def advance(self, source: str) -> None:
-        """Advance watermark for a source after successful processing."""
-        if source in self._pending_offsets:
-            self._watermarks[source] = self._pending_offsets.pop(source)
-            json_save_atomic(self._watermark_file, self._watermarks)
-
-    def advance_all(self) -> None:
-        """Advance watermarks for all sources."""
-        if not self._pending_offsets:
-            return
-        for source in list(self._pending_offsets):
-            self._watermarks[source] = self._pending_offsets.pop(source)
-        json_save_atomic(self._watermark_file, self._watermarks)
-
-    def flush_to_inbox(self, source: str, inbox_dir: Path) -> Path | None:
-        """Dump unreflected content for *source* into inbox and advance watermark.
-
-        Filters out session metadata and system prompts. Returns the
-        path of the written file, or None if there was nothing to flush.
-        """
-        chunk = self._read_since_watermark(source)
-        if chunk is None:
-            return None
-        raw, file_size = chunk
-
-        lines = _filter_for_inbox(raw)
-        if not lines:
-            # Only metadata — advance watermark, nothing to dump.
-            self._watermarks[source] = file_size
-            json_save_atomic(self._watermark_file, self._watermarks)
-            return None
-
-        inbox_dir.mkdir(parents=True, exist_ok=True)
-        out = inbox_dir / f"{source}_{int(time.time())}.jsonl"
-        out.write_text("\n".join(lines) + "\n")
-
-        self._watermarks[source] = file_size
-        json_save_atomic(self._watermark_file, self._watermarks)
-        logger.info("Flushed %d lines from %s to %s", len(lines), source, out.name)
-        return out