feat(planning): plans as first-class, replayable artifacts (#623)

coworker/audit.py

@@ -67,6 +67,7 @@ def __init__(self, db_path: str | Path) -> None:
             # layer further out.
             ("cache_read", "INTEGER DEFAULT 0"),
             ("cache_write", "INTEGER DEFAULT 0"),
+            ("plan_id", "TEXT"),
         ):
             try:
                 self._conn.execute(
@@ -87,8 +88,8 @@ def append(self, event: dict[str, Any]) -> None:
             self._conn.execute(
                 """
                 INSERT INTO audit_events
-                    (session_id, agent, workspace, connector, tool, stage, status, approval, args, result_preview, reason, resource, call_id, tokens_in, tokens_out, cache_read, cache_write)
-                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    (session_id, agent, workspace, connector, tool, stage, status, approval, args, result_preview, reason, resource, call_id, tokens_in, tokens_out, cache_read, cache_write, plan_id)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 """,
                 (
                     event.get("session_id") or "",
@@ -108,6 +109,7 @@ def append(self, event: dict[str, Any]) -> None:
                     int(event.get("tokens_out") or 0),
                     int(event.get("cache_read") or 0),
                     int(event.get("cache_write") or 0),
+                    str(event.get("plan_id") or ""),
                 ),
             )
             self._conn.commit()
@@ -156,6 +158,7 @@ def list(
         session_id: Optional[str] = None,
         connector: Optional[str] = None,
         tool: Optional[str] = None,
+        plan_id: Optional[str] = None,
     ) -> list[dict[str, Any]]:
         where = []
         params: list[Any] = []
@@ -168,6 +171,9 @@ def list(
         if tool:
             where.append("tool = ?")
             params.append(tool)
+        if plan_id:
+            where.append("plan_id = ?")
+            params.append(plan_id)
         sql = "SELECT * FROM audit_events"
         if where:
             sql += " WHERE " + " AND ".join(where)

coworker/conversations.py

@@ -112,6 +112,7 @@ def __init__(self, base_dir: str | Path) -> None:
             "ALTER TABLE sessions ADD COLUMN compaction TEXT",
             "ALTER TABLE sessions ADD COLUMN team TEXT",
             "ALTER TABLE sessions ADD COLUMN bindings TEXT",
+            "ALTER TABLE sessions ADD COLUMN plan TEXT",
         ):
             try:
                 self._conn.execute(ddl)
@@ -344,13 +345,14 @@ def save(self, record: SessionRecord, touch: bool = True) -> None:
             title = record.title or title_from(record.messages)
             self._conn.execute(
                 """
-                INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, extra_roots, grants, compaction, team, bindings, updated_at)
-                VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
+                INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, extra_roots, grants, compaction, team, bindings, plan, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
                 ON CONFLICT(session_id) DO UPDATE SET
                     workspace = excluded.workspace, model = excluded.model, mode = excluded.mode,
                     title = COALESCE(sessions.title, excluded.title), agent = excluded.agent,
                     n_msgs = excluded.n_msgs, messages = NULL, extra_roots = excluded.extra_roots,
                     grants = excluded.grants, compaction = excluded.compaction,
+                    plan = CASE WHEN excluded.plan != '{}' THEN excluded.plan ELSE sessions.plan END,
                     updated_at = CASE WHEN ? THEN CURRENT_TIMESTAMP ELSE sessions.updated_at END
                 """,
                 (
@@ -366,6 +368,7 @@ def save(self, record: SessionRecord, touch: bool = True) -> None:
                     json.dumps(record.compaction or {}),
                     json.dumps(record.team or {}),
                     json.dumps(record.bindings or {}),
+                    json.dumps(record.plan or {}),
                     touch,
                 ),
             )
@@ -416,6 +419,7 @@ def load(self, session_id: str) -> Optional[SessionRecord]:
             bindings=_load_grants(
                 row["bindings"] if "bindings" in row.keys() else None
             ),
+            plan=_load_grants(row["plan"] if "plan" in row.keys() else None),
         )
 
     def set_team(self, session_id: str, team: dict) -> None:
@@ -430,6 +434,24 @@ def set_team(self, session_id: str, team: dict) -> None:
             )
             self._conn.commit()
 
+    def set_plan(self, session_id: str, plan: dict) -> None:
+        """Persist an approved plan artifact for a session (#623)."""
+        with self._lock:
+            cur = self._conn.execute(
+                "UPDATE sessions SET plan = ? WHERE session_id = ?",
+                (json.dumps(plan or {}), session_id),
+            )
+            if cur.rowcount == 0:
+                self._conn.execute(
+                    """
+                    INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, plan, updated_at)
+                    VALUES (?, '', '', 'interactive', '', 'code', 0, NULL, ?, CURRENT_TIMESTAMP)
+                    ON CONFLICT(session_id) DO UPDATE SET plan = excluded.plan
+                    """,
+                    (session_id, json.dumps(plan or {})),
+                )
+            self._conn.commit()
+
     def names(self):
         """The project-names alias table, riding this store's connection."""
         from .projects import ProjectNames
@@ -485,7 +507,9 @@ def list(self, *, workspace: Optional[str] = None) -> list[SessionRecord]:
                 archived=bool(r["archived"]),
                 origin=r["origin"],
                 origin_label=r["origin_label"],
+                grants=_load_grants(r["grants"] if "grants" in r.keys() else None),
                 team=_load_grants(r["team"] if "team" in r.keys() else None),
+                plan=_load_grants(r["plan"] if "plan" in r.keys() else None),
             )
             for r in rows
         ]

coworker/engine.py

@@ -1703,6 +1703,8 @@ async def _handle_plan_proposal(self, tool_call: ToolCall) -> AsyncIterator[Even
                 self.permissions.mode = Mode(str(result.get("mode", "interactive")))
             except ValueError:
                 self.permissions.mode = Mode.INTERACTIVE
+            if result.get("plan_id"):
+                self.audit_context["plan_id"] = str(result["plan_id"])
             result = {
                 **result,
                 "mode": self.permissions.mode.value,

coworker/server/app.py

@@ -768,6 +768,42 @@ def session_artifact_reveal(session_id: str, body: dict) -> dict[str, Any]:
             session_id, str(body.get("path", "")), str(body.get("mode", "reveal"))
         )
 
+    # Replayable plan artifacts (#623)
+    @app.get("/v1/sessions/{session_id}/plan")
+    def session_plan(session_id: str) -> dict[str, Any]:
+        plan = manager.get_session_plan(session_id)
+        if not plan:
+            return JSONResponse({"error": f"no plan found for session {session_id}"}, status_code=404)
+        return plan
+
+    @app.post("/v1/sessions/{session_id}/plan/replay")
+    def session_plan_replay(session_id: str, body: Optional[dict] = None) -> dict[str, Any]:
+        body = body or {}
+        try:
+            return manager.replay_plan(
+                session_id=session_id,
+                plan_id=body.get("plan_id"),
+                workspace=body.get("workspace"),
+            )
+        except ValueError as err:
+            return JSONResponse({"error": str(err)}, status_code=404)
+
+    @app.get("/v1/plans")
+    def list_plans() -> list[dict[str, Any]]:
+        return manager.list_plans()
+
+    @app.post("/v1/plans/replay")
+    def plans_replay(body: Optional[dict] = None) -> dict[str, Any]:
+        body = body or {}
+        try:
+            return manager.replay_plan(
+                session_id=body.get("session_id"),
+                plan_id=body.get("plan_id"),
+                workspace=body.get("workspace"),
+            )
+        except ValueError as err:
+            return JSONResponse({"error": str(err)}, status_code=404)
+
     # Agent teams (OPE-96): the session's board (workspace-keyed space) + journal
     # overview. Mutations act as the USER — the human side of the gates.
     @app.get("/v1/sessions/{session_id}/board")