diff --git a/coworker/audit.py b/coworker/audit.py
index 5199a7b7d..37dfdbc7d 100644
--- a/coworker/audit.py
+++ b/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)
diff --git a/coworker/conversations.py b/coworker/conversations.py
index 300ca9cc9..2147afb7d 100644
--- a/coworker/conversations.py
+++ b/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
         ]
diff --git a/coworker/engine.py b/coworker/engine.py
index e492443c4..a7e8c94ea 100644
--- a/coworker/engine.py
+++ b/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,
diff --git a/coworker/server/app.py b/coworker/server/app.py
index c2e60e159..16bb000f0 100644
--- a/coworker/server/app.py
+++ b/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")
diff --git a/coworker/server/manager.py b/coworker/server/manager.py
index cf9b740fa..34a46a71d 100644
--- a/coworker/server/manager.py
+++ b/coworker/server/manager.py
@@ -745,6 +745,8 @@ def get_engine(
 
             engine.compaction_state = CompactionState.from_dict(record.compaction)
         engine.compaction_settings = self.compaction_settings
+        if record is not None and record.plan and record.plan.get("id"):
+            engine.audit_context["plan_id"] = str(record.plan["id"])
         self._engines[session_id] = engine
         if is_new_session:
             self._emit_session_created(session_id, agent_name)
@@ -1167,10 +1169,223 @@ async def approve(args, tool_call_id=None):
                     "approved": False,
                     "feedback": resp.get("feedback") or "the user rejected the plan",
                 }
-            return {"approved": True, "mode": resp.get("mode") or "interactive"}
+            plan_record = self.save_plan_artifact(session_id, str(args.get("plan", "")))
+            return {
+                "approved": True,
+                "mode": resp.get("mode") or "interactive",
+                "plan_id": plan_record.get("id"),
+            }
 
         return approve
 
+    # -- Plan artifacts (#623) --------------------------------------------------
+    def save_plan_artifact(
+        self,
+        session_id: str,
+        plan_text: str,
+        plan_id: Optional[str] = None,
+        title: Optional[str] = None,
+    ) -> dict[str, Any]:
+        """Persist a plan proposal as a first-class replayable artifact (#623)."""
+        from datetime import datetime, timezone
+
+        plan_id = plan_id or f"plan-{uuid.uuid4().hex[:8]}"
+        scratch_dir = Path(self._provision_scratch(session_id))
+        plans_dir = scratch_dir / "plans"
+        plans_dir.mkdir(parents=True, exist_ok=True)
+
+        plan_filename = f"{plan_id}.md"
+        plan_path = plans_dir / plan_filename
+        plan_path.write_text(plan_text, encoding="utf-8")
+
+        # Also update plan.md at root of scratch as canonical active plan
+        (scratch_dir / "plan.md").write_text(plan_text, encoding="utf-8")
+
+        # Derive a readable title if not supplied
+        if not title:
+            first_line = plan_text.strip().splitlines()[0] if plan_text.strip() else ""
+            title = first_line.lstrip("#").strip()[:80] or f"Plan {plan_id}"
+
+        plan_record = {
+            "id": plan_id,
+            "session_id": session_id,
+            "title": title,
+            "path": f"plans/{plan_filename}",
+            "plan": plan_text,
+            "created_at": datetime.now(timezone.utc).isoformat(),
+        }
+
+        # Persist in session store
+        self.session_store.set_plan(session_id, plan_record)
+
+        # Update in-memory engine if live
+        engine = self._engines.get(session_id)
+        if engine is not None:
+            engine.audit_context["plan_id"] = plan_id
+            if hasattr(engine, "record") and engine.record:
+                engine.record.plan = plan_record
+
+        # Audit persistence
+        self.audit_store.append(
+            {
+                "session_id": session_id,
+                "stage": "plan_persisted",
+                "status": "approved",
+                "tool": "propose_plan",
+                "resource": str(plan_path),
+                "plan_id": plan_id,
+            }
+        )
+
+        return plan_record
+
+    def get_session_plan(self, session_id: str) -> Optional[dict[str, Any]]:
+        """Retrieve the persisted plan artifact for a session (#623)."""
+        record = self.session_store.load(session_id)
+        if record and record.plan:
+            return record.plan
+        # Fallback to reading from scratch directory if it exists
+        scratch = self.scratch_base() / session_id
+        plans_dir = scratch / "plans"
+        if plans_dir.is_dir():
+            md_files = sorted(
+                plans_dir.glob("*.md"),
+                key=lambda f: f.stat().st_mtime,
+                reverse=True,
+            )
+            if md_files:
+                target_file = md_files[0]
+                try:
+                    content = target_file.read_text(encoding="utf-8")
+                    plan_id = target_file.stem
+                    first_line = content.strip().splitlines()[0] if content.strip() else ""
+                    title = first_line.lstrip("#").strip()[:80] or f"Plan {plan_id}"
+                    return {
+                        "id": plan_id,
+                        "session_id": session_id,
+                        "title": title,
+                        "path": f"plans/{target_file.name}",
+                        "plan": content,
+                    }
+                except Exception:
+                    pass
+        plan_file = scratch / "plan.md"
+        if plan_file.exists():
+            try:
+                content = plan_file.read_text(encoding="utf-8")
+                first_line = content.strip().splitlines()[0] if content.strip() else ""
+                title = first_line.lstrip("#").strip()[:80] or f"Plan {session_id[:8]}"
+                return {
+                    "id": f"plan-{session_id[:8]}",
+                    "session_id": session_id,
+                    "title": title,
+                    "path": "plan.md",
+                    "plan": content,
+                }
+            except Exception:
+                pass
+        return None
+
+    def list_plans(self) -> list[dict[str, Any]]:
+        """List all saved plan artifacts across sessions (#623)."""
+        plans: list[dict[str, Any]] = []
+        for s in self.session_store.list():
+            if s.plan:
+                item = dict(s.plan)
+                item.setdefault("session_id", s.session_id)
+                item.setdefault("session_title", s.title)
+                plans.append(item)
+        return plans
+
+    def replay_plan(
+        self,
+        session_id: Optional[str] = None,
+        plan_id: Optional[str] = None,
+        workspace: Optional[str] = None,
+    ) -> dict[str, Any]:
+        """Replay an approved plan artifact in a fresh session (#623)."""
+        origin_plan: Optional[dict[str, Any]] = None
+        if session_id:
+            origin_plan = self.get_session_plan(session_id)
+        if not origin_plan and plan_id:
+            for p in self.list_plans():
+                if p.get("id") == plan_id:
+                    origin_plan = p
+                    session_id = p.get("session_id")
+                    break
+
+        if not origin_plan:
+            raise ValueError(
+                f"No plan artifact found for session={session_id} plan_id={plan_id}"
+            )
+
+        plan_text = origin_plan.get("plan", "")
+        origin_plan_id = origin_plan.get("id") or (f"plan-{session_id[:8]}" if session_id else f"plan-{uuid.uuid4().hex[:8]}")
+        origin_record = self.session_store.load(session_id) if session_id else None
+        target_ws = workspace or (origin_record.workspace if origin_record else None)
+        model = (origin_record.model if origin_record else None) or self.model
+        agent = (origin_record.agent if origin_record else None) or "code"
+
+        new_session_id = f"replay-{uuid.uuid4().hex[:8]}"
+
+        new_scratch = Path(self._provision_scratch(new_session_id))
+        plans_dir = new_scratch / "plans"
+        plans_dir.mkdir(parents=True, exist_ok=True)
+        (new_scratch / "plan.md").write_text(plan_text, encoding="utf-8")
+        (plans_dir / f"{origin_plan_id}.md").write_text(plan_text, encoding="utf-8")
+
+        new_plan_record = {
+            **origin_plan,
+            "origin_session_id": session_id,
+            "origin_plan_id": origin_plan_id,
+            "session_id": new_session_id,
+        }
+
+        # Seed initial session record with prompt and plan artifact
+        initial_messages = [
+            {"role": "user", "content": f"Execute the approved plan:\n\n{plan_text}"}
+        ]
+        new_record = SessionRecord(
+            session_id=new_session_id,
+            workspace=target_ws or str(new_scratch),
+            model=model,
+            mode=Mode.INTERACTIVE.value,
+            title=f"Replay: {origin_plan.get('title', 'Plan')}",
+            agent=agent,
+            messages=initial_messages,
+            plan=new_plan_record,
+        )
+        self.session_store.save(new_record)
+
+        # Build engine to register it and set audit context
+        new_engine = self.get_engine(
+            new_session_id,
+            workspace=target_ws,
+            agent=agent,
+        )
+        if new_engine is not None:
+            new_engine.audit_context["plan_id"] = origin_plan_id
+            new_engine.audit_context["replay_from"] = session_id or ""
+
+        self.audit_store.append(
+            {
+                "session_id": new_session_id,
+                "stage": "plan_replayed",
+                "status": "started",
+                "tool": "replay_plan",
+                "resource": f"session:{session_id or 'none'}",
+                "plan_id": origin_plan_id,
+            }
+        )
+
+        return {
+            "session_id": new_session_id,
+            "plan_id": origin_plan_id,
+            "workspace": target_ws or str(new_scratch),
+            "agent": agent,
+            "plan": new_plan_record,
+        }
+
     def persist_session(self, session_id: str) -> None:
         """Save the cached engine's thread (so a prompt's pending tool call survives a crash)."""
         engine = self._engines.get(session_id)
diff --git a/coworker/sessions.py b/coworker/sessions.py
index 0e18cf27d..98ee28d9d 100644
--- a/coworker/sessions.py
+++ b/coworker/sessions.py
@@ -44,3 +44,5 @@ class SessionRecord:
     # lead_session, space}. Leads gain their entry when the staffing gate creates the
     # team. Drives tool binding (board actor identity) + the sidebar's expandable entry.
     team: dict[str, Any] = field(default_factory=dict)
+    # Plan artifact: durable, replayable approved plan artifact (#623).
+    plan: dict[str, Any] = field(default_factory=dict)
diff --git a/surfaces/gui/src/App.tsx b/surfaces/gui/src/App.tsx
index 67cff675c..819447ba0 100644
--- a/surfaces/gui/src/App.tsx
+++ b/surfaces/gui/src/App.tsx
@@ -2212,6 +2212,7 @@ export function App() {
             teamChatUnread={curSession?.team?.chat_unread || 0}
             onOpenTeamChat={() => setChatTeam(curSession?.team?.team_id || "")}
             onOpenWorker={(w) => void selectSession(w.session_id, w.workspace, w.agent)}
+            onOpenSession={(id, ws, ag) => void selectSession(id, ws || "", ag || "code")}
             openBoardKey={boardRailKey}
           />
           {boardOpen && board && board.space && (
diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts
index a69e0ceca..315dca764 100644
--- a/surfaces/gui/src/api.ts
+++ b/surfaces/gui/src/api.ts
@@ -408,6 +408,47 @@ export async function revealArtifact(
   return res.json();
 }
 
+// -- Replayable plan artifacts (#623) ----------------------------------------
+export interface PlanArtifact {
+  id: string;
+  session_id: string;
+  title: string;
+  path: string;
+  plan: string;
+  created_at?: string;
+  origin_session_id?: string;
+  origin_plan_id?: string;
+}
+
+export async function getSessionPlan(sessionId: string): Promise<PlanArtifact | null> {
+  const res = await fetch(`${httpBase()}/v1/sessions/${encodeURIComponent(sessionId)}/plan`);
+  if (!res.ok) return null;
+  return res.json();
+}
+
+export async function replayPlan(
+  sessionId: string,
+  planId?: string,
+  workspace?: string,
+): Promise<{ session_id: string; plan_id: string; workspace: string; agent: string; plan: PlanArtifact }> {
+  const res = await fetch(`${httpBase()}/v1/sessions/${encodeURIComponent(sessionId)}/plan/replay`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ plan_id: planId, workspace }),
+  });
+  if (!res.ok) {
+    const data = await res.json().catch(() => ({}));
+    throw new Error(data.error || `Failed to replay plan: ${res.statusText}`);
+  }
+  return res.json();
+}
+
+export async function listPlans(): Promise<PlanArtifact[]> {
+  const res = await fetch(`${httpBase()}/v1/plans`);
+  if (!res.ok) return [];
+  return res.json();
+}
+
 // -- session roots (orphan Cowork: scratch + added folders) -------------------
 export interface RootInfo {
   path: string;
diff --git a/surfaces/gui/src/components/Icon.tsx b/surfaces/gui/src/components/Icon.tsx
index da3b5ce89..dc47cfb76 100644
--- a/surfaces/gui/src/components/Icon.tsx
+++ b/surfaces/gui/src/components/Icon.tsx
@@ -42,6 +42,7 @@ export type IconName =
   | "table"
   | "mic"
   | "stop"
+  | "play"
   | "warning"
   | "x";
 
@@ -364,6 +365,12 @@ export function Icon({
           <path d="M9.5 11.5h5" />
         </svg>
       );
+    case "play":
+      return (
+        <svg {...s}>
+          <polygon points="7 4 19 12 7 20 7 4" fill="currentColor" stroke="none" />
+        </svg>
+      );
     case "trash":
       return (
         <svg {...s}>
diff --git a/surfaces/gui/src/components/RightRail.tsx b/surfaces/gui/src/components/RightRail.tsx
index 307404d07..fae229454 100644
--- a/surfaces/gui/src/components/RightRail.tsx
+++ b/surfaces/gui/src/components/RightRail.tsx
@@ -8,6 +8,7 @@ import {
   getJournalCases,
   getRoots,
   readArtifact,
+  replayPlan,
   revealArtifact,
   type ArtifactContent,
   type ArtifactInfo,
@@ -76,8 +77,8 @@ interface Props {
   teamMembers?: SessionInfo[];
   teamChatEnabled?: boolean;
   teamChatUnread?: number;
-  onOpenTeamChat?: () => void;
   onOpenWorker?: (s: SessionInfo) => void;
+  onOpenSession?: (id: string, ws?: string, ag?: string) => void;
   // Bumped when a [.](board:) chip in the transcript is clicked — expands the Board section.
   openBoardKey?: number;
 }
@@ -107,6 +108,7 @@ export function RightRail({
   teamChatUnread = 0,
   onOpenTeamChat,
   onOpenWorker,
+  onOpenSession,
   openBoardKey = 0,
 }: Props) {
   const { t } = useTranslation();
@@ -244,6 +246,7 @@ export function RightRail({
           content={content}
           onReload={reloadSelected}
           onBack={() => setSelected(null)}
+          onOpenSession={onOpenSession}
           onOpenEntry={(path) =>
             setSelected({
               path,
@@ -570,6 +573,7 @@ function ArtifactViewer({
   onReload,
   onBack,
   onOpenEntry,
+  onOpenSession,
 }: {
   sessionId: string;
   artifact: ArtifactInfo;
@@ -578,9 +582,11 @@ function ArtifactViewer({
   onBack: () => void;
   // Folder listings: open a child entry in the viewer (files and subfolders alike).
   onOpenEntry?: (path: string) => void;
+  onOpenSession?: (id: string, ws?: string, ag?: string) => void;
 }) {
   const { t } = useTranslation();
   const [reloadKey, setReloadKey] = useState(0);
+  const [replaying, setReplaying] = useState(false);
   // UX-038: the ambiguous icon cluster collapsed into ONE labeled ⋯ menu; the
   // breadcrumb parent is the back action and ✕ closes. Copy CONTENTS is the
   // primary copy — the path copy (a 2026-07-12 tester fix) lives under it, labeled.
@@ -599,7 +605,27 @@ function ArtifactViewer({
   const isApp = content?.kind === "sheet" || content?.kind === "pdf" || content?.kind === "office";
   // Text-bearing kinds can copy their contents; images/PDFs/sheets have nothing textual to copy.
   const copyableText = typeof content?.content === "string" && !content?.error;
+  const isPlan =
+    artifact.name === "plan.md" ||
+    artifact.path.startsWith("plans/") ||
+    artifact.path.endsWith("/plan.md");
   const crumbRoot = artifact.origin === "files" ? t("rail.crumb_files") : t("rail.artifacts_title");
+
+  const handleReplay = async () => {
+    if (replaying) return;
+    setReplaying(true);
+    try {
+      const res = await replayPlan(sessionId);
+      if (onOpenSession && res.session_id) {
+        onOpenSession(res.session_id, res.workspace, res.agent);
+      }
+    } catch (err) {
+      console.error("Failed to replay plan:", err);
+    } finally {
+      setReplaying(false);
+    }
+  };
+
   const item = (
     testid: string,
     icon: Parameters<typeof Icon>[0]["name"],
@@ -638,6 +664,18 @@ function ArtifactViewer({
           <div className="artifact-path">{artifact.path}</div>
         </div>
         <div className="rail-actions">
+          {isPlan && (
+            <button
+              className="artifact-icon-btn"
+              data-testid="artifact-rerun-plan"
+              onClick={handleReplay}
+              disabled={replaying}
+              aria-label={t("rail.plan_rerun")}
+              title={replaying ? t("rail.plan_replaying") : t("rail.plan_rerun")}
+            >
+              <Icon name="play" size={16} />
+            </button>
+          )}
           {isHtml && (
             <button
               className="artifact-icon-btn"
@@ -663,6 +701,13 @@ function ArtifactViewer({
             </button>
             {menuOpen && (
               <div className="artifact-menu" data-testid="artifact-menu">
+                {isPlan &&
+                  item(
+                    "artifact-menu-rerun-plan",
+                    "play",
+                    replaying ? t("rail.plan_replaying") : t("rail.plan_rerun"),
+                    handleReplay,
+                  )}
                 {copyableText &&
                   item("artifact-copy-contents", "copy", t("rail.copy_contents"), () =>
                     navigator.clipboard?.writeText(content?.content || ""),
diff --git a/surfaces/gui/src/locales/en.json b/surfaces/gui/src/locales/en.json
index 869ea7397..7b3a7b8c8 100644
--- a/surfaces/gui/src/locales/en.json
+++ b/surfaces/gui/src/locales/en.json
@@ -270,6 +270,8 @@
     "copy_contents": "Copy contents",
     "open_in_browser": "Open in browser",
     "reveal_in_finder": "Reveal in Finder",
+    "plan_rerun": "Re-run this plan",
+    "plan_replaying": "Replaying…",
     "folder_empty": "This folder is empty.",
     "table_truncated": "Showing first {{max}} of {{total}} rows.",
     "board_chip_blocked_one": "{{count}} blocked",
diff --git a/surfaces/gui/src/locales/zh.json b/surfaces/gui/src/locales/zh.json
index 6ac0f33c3..1d9ec1be6 100644
--- a/surfaces/gui/src/locales/zh.json
+++ b/surfaces/gui/src/locales/zh.json
@@ -266,6 +266,8 @@
     "copy_contents": "复制内容",
     "open_in_browser": "在浏览器中打开",
     "reveal_in_finder": "在访达中显示",
+    "plan_rerun": "重新运行此计划",
+    "plan_replaying": "正在重放…",
     "folder_empty": "此文件夹为空。",
     "table_truncated": "显示前 {{max}} 行，共 {{total}} 行。",
     "board_chip_blocked_other": "{{count}} 个受阻",
diff --git a/tests/test_plan_replay.py b/tests/test_plan_replay.py
new file mode 100644
index 000000000..e7f4abedb
--- /dev/null
+++ b/tests/test_plan_replay.py
@@ -0,0 +1,264 @@
+"""Tests for replayable plan artifacts (#623)."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+import pytest
+from fastapi.testclient import TestClient
+
+import aisuite as ai
+from coworker.audit import AuditStore
+from coworker.conversations import ConversationStore
+from coworker.engine import TurnEngine
+from coworker.permissions import Mode, PermissionEngine
+from coworker.providers import (
+    AssistantTurn,
+    ModelCapabilities,
+    ProviderClient,
+    ToolCall,
+)
+from coworker.server import SessionManager, create_app
+from coworker.sessions import SessionRecord
+from coworker.tools import ToolRegistry
+from coworker.tools.plan import propose_plan_tool
+
+
+class ScriptedProvider(ProviderClient):
+    def __init__(self, turns=None):
+        self._turns = list(turns or [])
+
+    def complete(self, *, model, messages, tools=None, **settings):
+        if not self._turns:
+            return AssistantTurn(text="ok", finish_reason="stop")
+        return self._turns.pop(0)
+
+    def capabilities(self, model):
+        return ModelCapabilities()
+
+
+def _tool_turn(name, args, call_id="call_1"):
+    return AssistantTurn(
+        tool_calls=[ToolCall(id=call_id, name=name, arguments=args)],
+        finish_reason="tool_calls",
+    )
+
+
+def _text_turn(text):
+    return AssistantTurn(text=text, finish_reason="stop")
+
+
+def test_conversation_store_plan_persistence(tmp_path):
+    store = ConversationStore(tmp_path / "sessions.db")
+    rec = SessionRecord(
+        session_id="sess-1",
+        workspace=str(tmp_path),
+        model="gpt-5.5",
+        mode=Mode.PLAN.value,
+        title="Session 1",
+        plan={"id": "plan-1", "title": "My Plan", "plan": "1. step one\n2. step two"},
+    )
+    store.save(rec)
+
+    loaded = store.load("sess-1")
+    assert loaded is not None
+    assert loaded.plan == {"id": "plan-1", "title": "My Plan", "plan": "1. step one\n2. step two"}
+
+    # Update plan via set_plan
+    store.set_plan("sess-1", {"id": "plan-2", "title": "Updated", "plan": "revised"})
+    updated = store.load("sess-1")
+    assert updated is not None
+    assert updated.plan["id"] == "plan-2"
+    assert updated.plan["title"] == "Updated"
+
+    # Verify list() loads plan
+    all_sess = store.list()
+    assert len(all_sess) == 1
+    assert all_sess[0].plan["id"] == "plan-2"
+    store.close()
+
+
+def test_audit_store_plan_id_tracking_and_filtering(tmp_path):
+    store = AuditStore(tmp_path / "audit.db")
+    store.append({
+        "session_id": "sess-a",
+        "stage": "tool_executed",
+        "tool": "write_file",
+        "plan_id": "plan-alpha",
+    })
+    store.append({
+        "session_id": "sess-b",
+        "stage": "tool_executed",
+        "tool": "read_file",
+        "plan_id": "plan-beta",
+    })
+    store.append({
+        "session_id": "sess-a",
+        "stage": "tool_executed",
+        "tool": "run_shell",
+        "plan_id": "plan-alpha",
+    })
+
+    alpha_events = store.list(plan_id="plan-alpha")
+    assert len(alpha_events) == 2
+    assert all(e["plan_id"] == "plan-alpha" for e in alpha_events)
+
+    beta_events = store.list(plan_id="plan-beta")
+    assert len(beta_events) == 1
+    assert beta_events[0]["plan_id"] == "plan-beta"
+
+    gamma_events = store.list(plan_id="plan-gamma")
+    assert len(gamma_events) == 0
+    store.close()
+
+
+def test_turn_engine_tags_audit_context_on_plan_approval(tmp_path):
+    recorded_audit = []
+
+    def sink(ev):
+        recorded_audit.append(dict(ev))
+
+    async def approve(args, tool_call_id=None):
+        return {"approved": True, "mode": "auto", "plan_id": "plan-xyz"}
+
+    registry = ToolRegistry()
+    registry.register_all(ai.toolkits.files(root=str(tmp_path), allow_write=True))
+    registry.register(propose_plan_tool())
+    permissions = PermissionEngine(workspace_root=tmp_path, mode=Mode.PLAN)
+
+    engine = TurnEngine(
+        provider=ScriptedProvider([
+            _tool_turn("propose_plan", {"plan": "1. create plan.txt"}),
+            _tool_turn("write_file", {"path": "plan.txt", "content": "hello\n"}, "call_2"),
+            _text_turn("done"),
+        ]),
+        registry=registry,
+        permissions=permissions,
+        model="gpt-5.5",
+        plan_approver=approve,
+        audit_sink=sink,
+    )
+
+    async def _run():
+        return [ev async for ev in engine.run("start")]
+
+    asyncio.run(_run())
+
+    assert engine.audit_context.get("plan_id") == "plan-xyz"
+    # Verify subsequent write_file tool audit events carry plan_id
+    write_audits = [e for e in recorded_audit if e.get("tool") == "write_file"]
+    assert len(write_audits) > 0
+    assert all(e.get("plan_id") == "plan-xyz" for e in write_audits)
+
+
+def test_manager_save_and_get_session_plan(tmp_path):
+    manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider())
+    plan_text = "# Reorganize Documentation\n1. Move guides\n2. Update index"
+    saved = manager.save_plan_artifact(
+        session_id="s1",
+        plan_text=plan_text,
+        plan_id="plan-doc-1",
+    )
+
+    assert saved["id"] == "plan-doc-1"
+    assert saved["title"] == "Reorganize Documentation"
+    assert saved["path"] == "plans/plan-doc-1.md"
+
+    # Check scratch files
+    scratch = Path(manager.scratch_base()) / "s1"
+    assert (scratch / "plan.md").exists()
+    assert (scratch / "plans" / "plan-doc-1.md").exists()
+    assert (scratch / "plan.md").read_text(encoding="utf-8") == plan_text
+
+    # Retrieve via get_session_plan
+    plan = manager.get_session_plan("s1")
+    assert plan is not None
+    assert plan["id"] == "plan-doc-1"
+    assert plan["plan"] == plan_text
+
+    # Check list_plans
+    all_plans = manager.list_plans()
+    assert len(all_plans) == 1
+    assert all_plans[0]["id"] == "plan-doc-1"
+
+    # Check audit log
+    audits = manager.audit_store.list(plan_id="plan-doc-1")
+    assert any(a["stage"] == "plan_persisted" and a["status"] == "approved" for a in audits)
+
+
+def test_manager_replay_plan(tmp_path):
+    manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider())
+    plan_text = "# Automated Verification Plan\n- Run tests\n- Lint code"
+    manager.save_plan_artifact(
+        session_id="s-origin",
+        plan_text=plan_text,
+        plan_id="plan-auto-1",
+    )
+
+    replay_res = manager.replay_plan(session_id="s-origin")
+    new_sid = replay_res["session_id"]
+    assert new_sid.startswith("replay-")
+    assert replay_res["plan_id"] == "plan-auto-1"
+
+    # Verify new session record in store
+    new_rec = manager.session_store.load(new_sid)
+    assert new_rec is not None
+    assert new_rec.plan["id"] == "plan-auto-1"
+    assert new_rec.plan["origin_session_id"] == "s-origin"
+    assert any("Execute the approved plan" in m["content"] for m in new_rec.messages)
+
+    # Verify new session engine carries audit_context
+    new_engine = manager.get_engine(new_sid)
+    assert new_engine.audit_context["plan_id"] == "plan-auto-1"
+    assert new_engine.audit_context["replay_from"] == "s-origin"
+
+    # Verify audit event logged for replay
+    replay_audits = [
+        e for e in manager.audit_store.list(session_id=new_sid)
+        if e["stage"] == "plan_replayed"
+    ]
+    assert len(replay_audits) == 1
+    assert replay_audits[0]["plan_id"] == "plan-auto-1"
+    assert replay_audits[0]["status"] == "started"
+
+
+def test_rest_plan_endpoints(tmp_path):
+    manager = SessionManager(workspace=tmp_path, provider=ScriptedProvider())
+    client = TestClient(create_app(manager))
+
+    plan_text = "# Core Refactor Plan\n1. Decouple modules"
+    manager.save_plan_artifact(
+        session_id="s-rest",
+        plan_text=plan_text,
+        plan_id="plan-rest-1",
+    )
+
+    # GET /v1/sessions/{session_id}/plan
+    resp = client.get("/v1/sessions/s-rest/plan")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["id"] == "plan-rest-1"
+    assert data["plan"] == plan_text
+
+    # GET /v1/sessions/{nonexistent}/plan
+    resp = client.get("/v1/sessions/nonexistent/plan")
+    assert resp.status_code == 404
+
+    # GET /v1/plans
+    resp = client.get("/v1/plans")
+    assert resp.status_code == 200
+    assert any(p["id"] == "plan-rest-1" for p in resp.json())
+
+    # POST /v1/sessions/{session_id}/plan/replay
+    resp = client.post("/v1/sessions/s-rest/plan/replay", json={})
+    assert resp.status_code == 200
+    replay_data = resp.json()
+    assert replay_data["session_id"].startswith("replay-")
+    assert replay_data["plan_id"] == "plan-rest-1"
+
+    # POST /v1/plans/replay by plan_id
+    resp = client.post("/v1/plans/replay", json={"plan_id": "plan-rest-1"})
+    assert resp.status_code == 200
+    replay_data2 = resp.json()
+    assert replay_data2["session_id"].startswith("replay-")
+    assert replay_data2["plan_id"] == "plan-rest-1"