feat(state): Hermes-style human rewind for shadow checkpoints

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Hermes-style human rewind: `Session.list_checkpoints` / `Session.restore_checkpoint`
+  restore the worktree and truncate matching transcript turns. Turn boundaries
+  come from the session (prompt-submit indices), not message inspection.
+  `build_harness` auto-wires `ShadowCheckpointHook` and a shared store under
+  `DreamPaths.checkpoints_dir` (default-on; oversized trees skip via
+  `ShadowCheckpointConfig.max_files`).
 - OpenTelemetry is **default-on**: core deps ship the OTLP SDK; sessions fan
   JSONL traces to OTel (`CompositeTracer`). Endpoint defaults to
   `http://localhost:4318`; override with `OTEL_EXPORTER_OTLP_ENDPOINT`. Opt out

src/dream/_factory.py

@@ -72,6 +72,12 @@
     build_session_skill_registry,
     render_skill_catalogue,
 )
+from dream.state.shadow import (
+    ShadowCheckpointConfig,
+    ShadowCheckpointHook,
+    ShadowCheckpointManager,
+    ShadowCheckpointStore,
+)
 from dream.subagents._async_delegation import AsyncDelegationManager
 from dream.subagents._catalogue import SubagentCatalogue
 from dream.subagents._declaration import SubagentSet
@@ -181,6 +187,8 @@ def build_harness(
     env: Mapping[str, str] | None = None,
     wake_model: str | None = None,
     verify_on_stop: bool = True,
+    shadow_checkpoints: bool = True,
+    shadow_checkpoint_config: ShadowCheckpointConfig | None = None,
 ) -> Harness:
     """Build a Harness whose engine factory produces a real, tool-wired engine.
 
@@ -233,6 +241,13 @@ def build_harness(
     mutating file tools without a subsequent evidence tool (read/grep/glob)
     nudge another turn before seal (capped by ``max_verify_nudges``).
 
+    ``shadow_checkpoints`` (default True) registers Hermes-style pre-mutate
+    filesystem snapshots and exposes :meth:`~dream.session.Session.restore_checkpoint`
+    for operator rewind (FS + transcript). Worktrees over the checkpoint
+    manager's ``max_files`` threshold (10,000 by default) are skipped to keep
+    per-turn overhead bounded; pass a custom ``ShadowCheckpointConfig`` to
+    override that threshold.
+
     ``env`` is consulted only for host resolution — ``DREAM_HOME`` path
     overrides and shell detection for the runtime-info prompt block — and
     defaults to ``os.environ``. Credentials never come from it.
@@ -346,12 +361,19 @@ def build_harness(
     # so a scheduler tick loop knows where to poll, and `paths` carries the
     # env-resolved roots.
     del wake_model  # ponytail: compat no-op — the wake runtime is gone
+    checkpoint_manager: ShadowCheckpointManager | None = None
+    if shadow_checkpoints:
+        checkpoint_manager = ShadowCheckpointManager(
+            store=ShadowCheckpointStore(base_dir=paths.checkpoints_dir),
+            config=shadow_checkpoint_config,
+        )
     config = HarnessConfig(
         working_dir=working_dir,
         task_manager=task_manager,
         delegations=AsyncDelegationManager(),
         cron_registry_path=task_context.cron_registry_path,
         paths=paths,
+        checkpoint_manager=checkpoint_manager,
         # MCP connect + plugin import are async/IO, so they hang off the
         # async-open chokepoint (``Harness._ensure_open``) rather than running
         # in this sync factory. ``None`` when both surfaces are disabled so the
@@ -369,6 +391,10 @@ def build_harness(
         ),
     )
     harness = Harness(config)
+    if checkpoint_manager is not None:
+        harness.register_hook(
+            ShadowCheckpointHook(manager=checkpoint_manager, working_dir=working_dir)
+        )
     if verify_on_stop:
         harness.register_hook(VerifyOnStopHook())
 
@@ -877,4 +903,5 @@ def _build_session_engine(
         initial_context=render_runtime_context(runtime_info),
         delegations=harness.config.delegations,
         prompt_surfaces=prompt_surfaces,
+        checkpoint_manager=harness.config.checkpoint_manager,
     )

src/dream/config/paths.py

@@ -207,6 +207,11 @@ def memory_dir(self) -> Path:
     def skills_dir(self) -> Path:
         return self.home / "skills"
 
+    @property
+    def checkpoints_dir(self) -> Path:
+        """Shared shadow-git store root under ``$DREAM_HOME/checkpoints``."""
+        return self.home / "checkpoints"
+
     # --- the one explicit side effect ---
 
     def ensure(self) -> DreamPaths:

src/dream/engine/_engine.py

@@ -41,6 +41,7 @@
 from dream.tools._registry import ToolRegistry
 
 if TYPE_CHECKING:
+    from dream.state.shadow import ShadowCheckpointManager
     from dream.subagents._async_delegation import AsyncDelegationManager
 
 
@@ -88,6 +89,9 @@ class QueryEngine:
     # Role name from session metadata (planner/generator/evaluator) for STOP hooks.
     role: str | None = None
     delegations: AsyncDelegationManager | None = None
+    # Hermes shadow checkpoints — shared across sessions on the harness; enables
+    # :meth:`Session.list_checkpoints` / :meth:`Session.restore_checkpoint`.
+    checkpoint_manager: ShadowCheckpointManager | None = None
     # Typed request surfaces for `/context` (FailoverStreamer hides streamer attrs).
     prompt_surfaces: PromptSurfaces | None = None
 
@@ -153,6 +157,7 @@ def build_query_engine(
     orientation: OrientationConfig | None = None,
     delegations: AsyncDelegationManager | None = None,
     prompt_surfaces: PromptSurfaces | None = None,
+    checkpoint_manager: ShadowCheckpointManager | None = None,
 ) -> QueryEngine:
     """Wrap a ``ToolRegistry`` in the canonical dispatcher and bind a streamer.
 
@@ -195,6 +200,7 @@ def build_query_engine(
         role=role_raw if isinstance(role_raw, str) else None,
         delegations=delegations,
         prompt_surfaces=prompt_surfaces,
+        checkpoint_manager=checkpoint_manager,
     )
 
 

src/dream/harness.py

@@ -44,6 +44,7 @@
         RunTaskResult,
         SprintGoalProvider,
     )
+    from dream.state.shadow import ShadowCheckpointManager
     from dream.subagents._async_delegation import AsyncDelegationManager
     from dream.tasks import BackgroundTaskManager
 
@@ -89,6 +90,8 @@ class HarnessConfig:
     # than re-resolving and risking divergence.
     paths: DreamPaths | None = None
     session_store: FileSessionStore | None = None
+    # Shared Hermes-style shadow checkpoint manager (FS snaps + human rewind).
+    checkpoint_manager: ShadowCheckpointManager | None = None
     extra: dict[str, Any] = field(default_factory=dict)
     _engine_factory: EngineFactory | None = None
     # Async setup run once before the first session — MCP connect + plugin

src/dream/session.py

@@ -72,6 +72,11 @@
 if TYPE_CHECKING:
     from dream.context import ContextBreakdown
     from dream.engine._engine import QueryEngine
+    from dream.state.shadow import (
+        CheckpointSnapshot,
+        CombinedRestoreResult,
+        ShadowCheckpointManager,
+    )
 
 
 def _zero_cost() -> SessionCostSnapshot:
@@ -165,6 +170,7 @@ def __init__(
         # path is missing. Resumed sessions receive the exact loaded revision.
         self._snapshot_revision = _snapshot_revision
         self._transcript: list[ConversationMessage] = []
+        self._prompt_indices: list[int] = []
         self._cancel_event: asyncio.Event | None = None
         self._closed = False
         # Single-flight guard: ``Session`` keeps per-call cancel state on the
@@ -231,6 +237,80 @@ def transcript(self) -> list[ConversationMessage]:
         """
         return self._transcript
 
+    @property
+    def checkpoint_manager(self) -> ShadowCheckpointManager | None:
+        """Shadow FS checkpoint manager when the bound engine was built with one."""
+        engine = self._engine
+        if engine is None:
+            return None
+        return engine.checkpoint_manager
+
+    def list_checkpoints(self) -> list[CheckpointSnapshot]:
+        """List shadow checkpoints for this session's working directory (newest first)."""
+        engine = self._engine
+        manager = self.checkpoint_manager
+        if engine is None or manager is None:
+            return []
+        return manager.list_for(engine.working_dir)
+
+    def restore_checkpoint(
+        self,
+        commit_sha: str | None = None,
+        *,
+        rewind_turns: int = 1,
+    ) -> CombinedRestoreResult:
+        """Hermes-style human rewind: restore FS and truncate the transcript.
+
+        ``commit_sha=None`` selects the newest checkpoint. Refuses while a
+        ``send`` is in flight so restore cannot race the live turn loop.
+        Turn boundaries come from ``Session`` (prompt submit indices), not
+        from inspecting message roles — engine-generated user messages are
+        not human turns.
+        """
+        from dream.state.shadow import CombinedRestoreResult, RestoreOutcome, RestoreResult
+
+        if self._active:
+            raise RuntimeError("cannot restore checkpoint while a send is in flight")
+        engine = self._engine
+        manager = self.checkpoint_manager
+        if engine is None or manager is None:
+            return CombinedRestoreResult(
+                fs=RestoreResult(
+                    outcome=RestoreOutcome.DISABLED,
+                    detail="no checkpoint manager bound on this session",
+                ),
+                messages=tuple(self._transcript),
+                transcript_removed=0,
+            )
+
+        sha = commit_sha
+        if sha is None:
+            listed = manager.list_for(engine.working_dir)
+            if not listed:
+                return CombinedRestoreResult(
+                    fs=RestoreResult(
+                        outcome=RestoreOutcome.NOT_FOUND,
+                        detail="no checkpoints for working directory",
+                    ),
+                    messages=tuple(self._transcript),
+                    transcript_removed=0,
+                )
+            sha = listed[0].commit_sha
+
+        result = manager.restore_and_rewind(
+            engine.working_dir,
+            commit_sha=sha,
+            messages=self._transcript,
+            prompt_indices=self._prompt_indices,
+            rewind_turns=rewind_turns,
+        )
+        if result.fs.outcome is RestoreOutcome.RESTORED:
+            self._transcript[:] = list(result.messages)
+            self._prompt_indices = [
+                index for index in self._prompt_indices if index < len(self._transcript)
+            ]
+        return result
+
     def _current_cost(self) -> SessionCostSnapshot:
         return cost_snapshot_from_fields(
             SessionCostFields(
@@ -324,6 +404,7 @@ def restore_from_snapshot(self, snapshot: SessionSnapshot) -> None:
         """Replace transcript and cost counters from a saved snapshot."""
         restored = messages_from_records(snapshot.messages)
         self._transcript[:] = sanitize_conversation_messages(restored)
+        self._prompt_indices.clear()
         self.cost.input_tokens = snapshot.cost.input_tokens
         self.cost.output_tokens = snapshot.cost.output_tokens
         self.cost.cache_read_tokens = snapshot.cost.cache_read_tokens
@@ -359,6 +440,7 @@ async def send(self, prompt: str) -> AsyncIterator[Event]:
         resume = list(self._transcript) if self._transcript else None
         user_msg = ConversationMessage(role="user", content=[TextBlock(text=prompt)])
         self._transcript.append(user_msg)
+        self._prompt_indices.append(len(self._transcript) - 1)
 
         config = self._engine.make_session_config()
         if self._has_sent:
@@ -537,6 +619,7 @@ def _apply_compaction(self) -> None:
         carryover = engine.carryover_metadata
         if carryover is not None and carryover.last_compacted_transcript is not None:
             self._transcript[:] = list(carryover.last_compacted_transcript)
+            self._prompt_indices.clear()
             carryover.last_compacted_transcript = None
             return
         compactor = engine.compactor
@@ -558,6 +641,7 @@ def _apply_compaction(self) -> None:
         )
         if result is not None:
             self._transcript[:] = new_transcript
+            self._prompt_indices.clear()
 
     async def cancel(self) -> None:
         """Cancel the in-flight ``send``, if any.

src/dream/state/__init__.py

@@ -5,21 +5,29 @@
 from dream.state.shadow import (
     CheckpointOutcome,
     CheckpointReason,
+    CheckpointSnapshot,
+    CombinedRestoreResult,
     MutatingToolName,
     RestoreOutcome,
+    RestoreResult,
     ShadowCheckpointConfig,
     ShadowCheckpointHook,
     ShadowCheckpointManager,
     ShadowCheckpointStore,
+    rewind_transcript,
 )
 
 __all__ = [
     "CheckpointOutcome",
     "CheckpointReason",
+    "CheckpointSnapshot",
+    "CombinedRestoreResult",
     "MutatingToolName",
     "RestoreOutcome",
+    "RestoreResult",
     "ShadowCheckpointConfig",
     "ShadowCheckpointHook",
     "ShadowCheckpointManager",
     "ShadowCheckpointStore",
+    "rewind_transcript",
 ]

src/dream/state/shadow/__init__.py

@@ -4,11 +4,13 @@
 
 from dream.state.shadow._hook import ShadowCheckpointHook
 from dream.state.shadow._manager import ShadowCheckpointManager
+from dream.state.shadow._rewind import rewind_transcript
 from dream.state.shadow._store import ShadowCheckpointStore
 from dream.state.shadow._types import (
     CheckpointOutcome,
     CheckpointReason,
     CheckpointSnapshot,
+    CombinedRestoreResult,
     EnsureResult,
     MutatingToolName,
     RestoreOutcome,
@@ -20,6 +22,7 @@
     "CheckpointOutcome",
     "CheckpointReason",
     "CheckpointSnapshot",
+    "CombinedRestoreResult",
     "EnsureResult",
     "MutatingToolName",
     "RestoreOutcome",
@@ -28,4 +31,5 @@
     "ShadowCheckpointHook",
     "ShadowCheckpointManager",
     "ShadowCheckpointStore",
+    "rewind_transcript",
 ]

src/dream/state/shadow/_hook.py

@@ -32,9 +32,14 @@ def __init__(
         self._manager = manager
         self._working_dir = working_dir
 
+    @staticmethod
+    def _session_id(payload: Mapping[str, Any]) -> str | None:
+        raw_session_id = payload.get("session_id")
+        return str(raw_session_id) if raw_session_id is not None else None
+
     async def __call__(self, event: HookEvent, payload: Mapping[str, Any]) -> HookResult:
         if event is HookEvent.USER_PROMPT_SUBMIT:
-            self._manager.begin_turn()
+            self._manager.begin_turn(self._session_id(payload))
             return HookResult()
 
         if event is not HookEvent.PRE_TOOL_USE:
@@ -45,7 +50,11 @@ async def __call__(self, event: HookEvent, payload: Mapping[str, Any]) -> HookRe
         if reason is None:
             return HookResult()
 
-        self._manager.ensure(self._working_dir, reason=reason)
+        self._manager.ensure(
+            self._working_dir,
+            reason=reason,
+            session_id=self._session_id(payload),
+        )
         return HookResult()