feat(git): workspace turn checkpointing via git shadow refs for safe rollback

coworker/agent.py

@@ -267,7 +267,11 @@ def build_engine(
     executor = LocalExecutor(cwd=ws) if ws is not None else None
     todo = TodoList()
     context = AgentContext(
-        workspace=ws, executor=executor, todo=todo, roots=root_list or None
+        workspace=ws,
+        executor=executor,
+        todo=todo,
+        roots=root_list or None,
+        session_id=session_id,
     )
 
     registry = ToolRegistry()
@@ -533,6 +537,7 @@ def context_provider() -> str:
         model=model,
         instructions=instructions,
         approver=approver,
+        session_id=session_id or "default",
         # Stop kills the in-flight foreground shell command, not just the loop.
         interrupt_hooks=[executor.interrupt_now] if executor is not None else None,
         max_iterations=(

coworker/agents/base.py

@@ -23,6 +23,7 @@ class AgentContext:
     # When None, tools fall back to the single `workspace` root. Held by reference so runtime
     # add/remove of folders is seen by the file tools built from it.
     roots: Optional[list] = None
+    session_id: Optional[str] = None
 
 
 @dataclass

coworker/catalog.py

@@ -92,7 +92,8 @@ def _files(context: AgentContext) -> list:
 
 def _git(context: AgentContext) -> list:
     ws = str(context.workspace)
-    return [*ai.toolkits.git(root=ws), *git_tools(ws)]  # git_status, git_diff, git_log
+    sid = getattr(context, "session_id", None) or ""
+    return [*ai.toolkits.git(root=ws), *git_tools(ws, session_id=sid)]
 
 
 def _search(context: AgentContext) -> list:

coworker/engine.py

@@ -97,6 +97,7 @@ def __init__(
         model: str,
         instructions: Optional[str] = None,
         approver: Optional[Approver] = None,
+        session_id: str = "default",
         max_iterations: int = 12,
         model_settings: Optional[dict[str, Any]] = None,
         messages: Optional[list[dict[str, Any]]] = None,
@@ -129,6 +130,9 @@ def __init__(
         self.permissions = permissions
         self.model = model
         self.approver = approver or _deny_all
+        self.session_id = session_id or "default"
+        self.turn_index = 0
+        self._turn_checkpoint_created = False
         self.max_iterations = max_iterations
         self.model_settings = dict(model_settings or {})
         self.messages: list[dict[str, Any]] = list(messages or [])
@@ -286,6 +290,25 @@ def queue_steering(
     ) -> None:
         self._steering.append((text, source))
 
+    def revert_turn(self, turn: Optional[int] = None) -> dict[str, Any]:
+        """Revert workspace files to the checkpoint taken before turn `turn`."""
+        from .tools.git import list_checkpoints, restore_checkpoint
+
+        target = turn
+        if target is None or target <= 0:
+            ckpts = list_checkpoints(
+                self.permissions.workspace_root, session_id=self.session_id
+            )
+            if not ckpts:
+                return {
+                    "ok": False,
+                    "error": "No checkpoints available to revert.",
+                }
+            target = ckpts[-1]["turn"]
+        return restore_checkpoint(
+            self.permissions.workspace_root, self.session_id, target
+        )
+
     # -- main loop --------------------------------------------------------------
     async def run(
         self,
@@ -313,6 +336,8 @@ async def run(
             message["_display"] = display
         self.messages.append(message)
         self._cancel.clear()
+        self.turn_index += 1
+        self._turn_checkpoint_created = False
         if self.session_facts is not None:
             self.session_facts.begin_turn()
         # §8.4 retry guard resets per user turn: two reviewer denials in one turn route
@@ -817,6 +842,22 @@ async def _handle_tool_calls(
             if allowed:
                 cleared.append(tool_call)
 
+        if cleared and not self._turn_checkpoint_created:
+            from .risk import WRITE_TOOLS
+
+            if any(tc.name in WRITE_TOOLS for tc in cleared):
+                try:
+                    from .tools.git import create_checkpoint
+
+                    create_checkpoint(
+                        self.permissions.workspace_root,
+                        self.session_id,
+                        self.turn_index,
+                    )
+                    self._turn_checkpoint_created = True
+                except Exception:
+                    pass
+
         concurrent = (
             [tc for tc in cleared if self._parallel_safe(tc)]
             if len(cleared) > 1

coworker/risk.py

@@ -55,6 +55,7 @@ class RiskClass(str, Enum):
 
 _BASE: dict[str, RiskClass] = {
     **{name: RiskClass.WRITE_LOCAL for name in WRITE_TOOLS},
+    "revert_turn": RiskClass.WRITE_LOCAL,
     SHELL_TOOL: RiskClass.EXEC,
     **{name: RiskClass.EGRESS for name in EGRESS_TOOLS},
 }