feat: per-run snapshot-based file-change tracking (#257)

backend/openbot/api/schemas.py

@@ -215,6 +215,8 @@ class RunOut(BaseModel):
     total_tokens: int | None = None
     cost_usd: float | None = None
     model_calls: int | None = None
+    changed_files: list[str] | None = None
+    diff: str | None = None
     created_at: datetime
     started_at: datetime | None
     finished_at: datetime | None

backend/openbot/db/migrations/versions/0023_run_snapshots.py

@@ -0,0 +1,22 @@
+"""run snapshots: per-run changed files and diff
+
+Revision ID: 0023
+Revises: 0022
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0023"
+down_revision = "0022"
+branch_labels = depends_on = None
+
+
+def upgrade():
+    op.add_column("runs", sa.Column("changed_files", sa.JSON(), nullable=True))
+    op.add_column("runs", sa.Column("diff", sa.Text(), nullable=True))
+
+
+def downgrade():
+    op.drop_column("runs", "diff")
+    op.drop_column("runs", "changed_files")

backend/openbot/db/models.py

@@ -231,6 +231,10 @@ class Run(Base):
     total_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True)
     cost_usd: Mapped[float | None] = mapped_column(Float, nullable=True)
     model_calls: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # Files the run changed and the unified diff, captured with the snapshot store
+    # (runtime/snapshot.py); None when the run never reached a snapshot point.
+    changed_files: Mapped[list | None] = mapped_column(JSON, nullable=True)
+    diff: Mapped[str | None] = mapped_column(Text, nullable=True)
     created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, nullable=False)
     started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)
     finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True)

backend/openbot/runtime/cli_agents/run.py

@@ -133,7 +133,7 @@ async def session_totals(totals: dict | None, seq: int) -> int:
         seq = await runner._record(run, seq, "message", {"message_id": res.message.id})
     usage = result.usage
     seq = await session_totals(result.totals, seq)
-    await runner._set_status(run.id, "completed", usage=usage)
+    await runner._snapshot_attach_terminal(run.id, thread, "completed", usage=usage)
     log.info("run %s completed in %.1fs via %s: reply=%d chars model_calls=%d prompt_tokens=%d cache_read_tokens=%d "
              "completion_tokens=%d session=%s resumed=%s", run.id, time.monotonic() - started, adapter.binary,
              len(result.text), usage["model_calls"], usage["prompt_tokens"], usage["cache_read_tokens"],

backend/openbot/runtime/runner.py

@@ -54,6 +54,7 @@
 from openbot.runtime.prompt import build_history, build_system_prompt_parts
 from openbot.runtime.providers import builtin_tools, effective_bot_profile
 from openbot.runtime.retry import ModelRetryMiddleware
+from openbot.runtime.snapshot import RunDiff, store_for_run
 from openbot.runtime.tool_failures import ToolFailureMiddleware
 from openbot.runtime.tool_loading import LOAD_TOOLS, DeferredToolsMiddleware
 from openbot.runtime.upstream import (
@@ -423,6 +424,10 @@ def __init__(self, services) -> None:
         self.s = services
         # Each bot's last OpenRouter upstream, so its next call lands on the same warm prompt cache.
         self.upstream_pins = UpstreamPins()
+        # Snapshot before-trees captured at run start, keyed by run id; consumed when the
+        # run ends (runtime/snapshot.py). Kept in memory: captures happen on the same
+        # worker that ends the run, and a crash loses nothing but a diff.
+        self._snapshot_before: dict[str, str] = {}
 
     async def _run_turn_triggers(
         self, provider: str, model: str, system_prompt: SystemMessage | None = None
@@ -462,6 +467,8 @@ async def _set_status(
         error: str | None = None,
         langsmith_run_id: str | None = None,
         usage: dict[str, int | float] | None = None,
+        changed_files: list[str] | None = None,
+        diff: str | None = None,
     ) -> Run | None:
         async with self.s.session_factory() as session:
             run = await session.get(Run, run_id)
@@ -478,6 +485,10 @@ async def _set_status(
             if usage and usage.get("model_calls"):
                 for k in (*USAGE_KEYS, "cost_usd"):
                     setattr(run, k, (getattr(run, k) or 0) + usage[k])
+            if changed_files is not None:
+                run.changed_files = changed_files
+            if diff is not None:
+                run.diff = diff
             if status == "running" and run.started_at is None:
                 run.started_at = utcnow()
             if status in ("completed", "failed", "cancelled"):
@@ -509,6 +520,55 @@ async def _set_status(
                 )
             return run
 
+    async def _snapshot_capture_before(self, run_id: str, thread: Thread) -> None:
+        """Fingerprint the workspace before the run works in it.  Best effort: a failed
+        capture just leaves the run without a diff (runtime/snapshot.py).
+
+        A resumed run re-enters execute(); its original before-tree must survive the
+        pause so the final diff spans every segment of the run."""
+        if run_id in self._snapshot_before:
+            return
+        try:
+            store = store_for_run(self.s.settings, thread)
+            if store is None:
+                return
+            tree = await store.capture()
+            if tree:
+                self._snapshot_before[run_id] = tree
+        except Exception:
+            log.exception("could not capture snapshot before run %s", run_id)
+
+    async def _snapshot_attach_terminal(
+        self, run_id: str, thread: Thread, status: str, **status_kwargs
+    ) -> Run | None:
+        """Record the run's final status, attaching the snapshot diff computed
+        against its before-tree.  The status is written first so a diff failure
+        never leaves a live run stuck in 'running'."""
+        rd = await self._snapshot_store_after(run_id, thread)
+        kwargs = dict(status_kwargs)
+        if rd is not None:
+            kwargs["changed_files"] = rd.changed_files
+            kwargs["diff"] = rd.diff
+        return await self._set_status(run_id, status, **kwargs)
+
+    async def _snapshot_store_after(self, run_id: str, thread: Thread) -> RunDiff | None:
+        """Diff the workspace now against the run's before-tree and return it.
+        None when there was no before-tree (nothing to diff against).  Best effort."""
+        before = self._snapshot_before.pop(run_id, None)
+        if before is None:
+            return None
+        try:
+            store = store_for_run(self.s.settings, thread)
+            if store is None:
+                return None
+            after = await store.capture()
+            if after is None:
+                return None
+            return await store.diff(before, after)
+        except Exception:
+            log.exception("could not capture snapshot after run %s", run_id)
+            return None
+
     async def _reports_so_far(self, run_id: str) -> list[str]:
         async with self.s.session_factory() as session:
             rows = (await session.execute(select(RunEvent.payload).where(RunEvent.run_id == run_id, RunEvent.type == "text")
@@ -1059,6 +1119,7 @@ async def execute(self, run_id: str, resume: Command | None = None) -> None:
             workspace_root = thread_workspace_root(
                 self.s.settings.workspace_root, thread.working_directory
             )
+            await self._snapshot_capture_before(run.id, thread)
             eff_provider, eff_model = effective_bot_profile(bot.bot, self.s.settings)
             triggers = await self._run_turn_triggers(eff_provider, eff_model, system_prompt)
             log.info(
@@ -1164,7 +1225,7 @@ async def execute(self, run_id: str, resume: Command | None = None) -> None:
                     # run "waiting_human" forever -- unresumable, and showing the bot as busy everywhere.
                     err = "ask_human has no human or external participant in this thread to ask"
                     log.warning("run %s waiting_human but undeliverable: %s", run.id, err)
-                    await self._set_status(run.id, "failed", error=err)
+                    await self._snapshot_attach_terminal(run.id, thread, "failed", error=err)
                     await self._record(run, await self._next_seq(run.id), "error", {"error": err})
                     await self._system_message(
                         thread.id, f"@{bot.handle} tried to ask a question, but {err}."
@@ -1185,7 +1246,7 @@ async def execute(self, run_id: str, resume: Command | None = None) -> None:
                         run_id=run.id,
                     )
                 seq = await self._record(run, seq, "message", {"message_id": res.message.id})
-            await self._set_status(run.id, "completed", langsmith_run_id=ls_id)
+            await self._snapshot_attach_terminal(run.id, thread, "completed", langsmith_run_id=ls_id)
             log.info(
                 "run %s completed in %.1fs: reply=%d chars %s langsmith_run_id=%s",
                 run.id,
@@ -1207,15 +1268,15 @@ async def execute(self, run_id: str, resume: Command | None = None) -> None:
         except asyncio.CancelledError:
             log.info("run %s cancelled after %.1fs", run.id, time.monotonic() - started)
             await self._post_interrupted(bot, thread, run, hop, progress)
-            await self._set_status(run.id, "cancelled")
+            await self._snapshot_attach_terminal(run.id, thread, "cancelled")
             await self._system_message(thread.id, f"@{bot.handle} run was cancelled.")
             await self._drop_checkpoint(run.id)
             raise
         except Exception as e:
             log.exception("run %s failed", run.id)
             err = (str(e) if isinstance(e, CliAgentError) else f"{type(e).__name__}: {e}")[:2000]
             # Status first: the bookkeeping below is best-effort and must never leave the run in "running".
-            await self._set_status(run.id, "failed", error=err)
+            await self._snapshot_attach_terminal(run.id, thread, "failed", error=err)
             try:
                 # Re-read the sequence: events recorded inside _stream are not visible to `seq` here.
                 await self._record(run, await self._next_seq(run.id), "error", {"error": err})