代码提交

offline-backend-20260512/backend/CLAUDE.md

@@ -107,6 +107,19 @@ CI runs these regression tests for every pull request via [.github/workflows/bac
 
 ## Architecture
 
+### Position roundtable recovery persistence
+
+`/api/position-roundtable` uses LangGraph threads as the live execution source
+and stores an additional SQL recovery copy in
+`position_roundtable_sessions`/`position_roundtable_nodes`. The recovery copy
+contains complete intent/node/summary/action-plan conversations, unsent
+composer drafts, workflow snapshots, and bounded full text for generated
+Markdown/JSON artifacts. This lets history and downstream closing agents keep
+working when container-local checkpoint or sandbox volumes are replaced.
+Revision `20260723_05` adds the conversation columns. Session deletion
+explicitly removes child nodes so SQLite deployments do not depend on FK
+cascade pragmas.
+
 ### Harness / App Split
 
 The backend is split into two layers with a strict dependency direction:

offline-backend-20260512/backend/Dockerfile

@@ -64,7 +64,7 @@ CMD ["sh", "-c", "cd backend && PYTHONPATH=. uv run uvicorn app.gateway.app:app
 
 # ── Stage 3: Runtime ──────────────────────────────────────────────────────────
 # Clean image without build-essential — reduces size (~200 MB) and attack surface.
-FROM python:3.12-slim-bookworm
+FROM python:3.12-slim-bookworm AS runtime
 
 # Copy Node.js runtime from builder (provides npx for MCP servers)
 COPY --from=builder /usr/bin/node /usr/bin/node

offline-backend-20260512/backend/README.md

@@ -449,3 +449,11 @@ See the [LICENSE](../LICENSE) file in the project root.
 ## Contributing
 
 See [CONTRIBUTING.md](CONTRIBUTING.md) for contribution guidelines.
+## Position roundtable durable history
+
+The position-roundtable workflow stores database-owned recovery snapshots for
+Step-1 intent chat, every position-node chat, the summary/action-plan chats,
+unsent composer drafts, and generated text deliverables. LangGraph checkpoints
+remain the execution source, but SQL history continues to render after a
+checkpoint or sandbox-output volume is replaced. Apply Alembic revision
+`20260723_05` (or simply upgrade to `head`) before deploying this version.

offline-backend-20260512/backend/app/gateway/routers/position_roundtable.py

@@ -2,8 +2,10 @@
 
 Unlike the existing multi-agent roundtable, this feature has no coordinator and
 no background roundtable job. The API only persists the task/intent/chain
-snapshot and indexes one direct-agent node per business-chain seat. Actual chat
-messages and artifacts stay in the established LangGraph thread stores.
+snapshot and indexes one direct-agent node per business-chain seat. LangGraph
+remains the live execution source, while SQL also stores recovery copies of
+all conversations, drafts and delivered artifact bodies so checkpoint-volume
+replacement cannot turn an existing collaboration history into a blank page.
 """
 
 from __future__ import annotations
@@ -25,6 +27,11 @@ NODE_STATUSES = {"locked", "ready", "running", "done", "stale", "rejected", "err
 _HANDOFF_TEXT_BUDGET = 12_000
 _MAX_INLINE_ARTIFACTS_PER_NODE = 4
 _MAX_INLINE_ARTIFACT_CHARS = 1_800
+# The handoff prompt remains intentionally small, but the recovery copy stored
+# in SQL must retain a complete normal Markdown/JSON deliverable.  The ceiling
+# protects the database from an accidental multi-gigabyte log while covering
+# substantially larger reports than this workspace normally produces.
+_MAX_PERSISTED_ARTIFACT_CHARS = 4_000_000
 _BUILTIN_MATERIAL_BUDGET = 24_000
 SUMMARY_AGENT_ID = "roundtable-summary"
 ACTION_PLAN_AGENT_ID = "position-action-planner"
@@ -58,6 +65,7 @@ class NodeResponse(BaseModel):
     status: str
     latest_answer: str | None = None
     artifact_manifest: list[dict[str, Any]] = Field(default_factory=list)
+    conversation_snapshot: dict[str, Any] | None = None
     revision: int = 0
     upstream_revision: dict[str, Any] | None = None
     rejection_count: int = 0
@@ -75,10 +83,13 @@ class SessionResponse(BaseModel):
     chain_snapshot: dict[str, Any] | None = None
     intent_thread_id: str | None = None
     intent_snapshot: dict[str, Any] | None = None
+    intent_conversation: dict[str, Any] | None = None
     summary_thread_id: str | None = None
     summary_snapshot: dict[str, Any] | None = None
+    summary_conversation: dict[str, Any] | None = None
     action_plan_thread_id: str | None = None
     action_plan_snapshot: dict[str, Any] | None = None
+    action_plan_conversation: dict[str, Any] | None = None
     status: str
     created_at: str | None = None
     updated_at: str | None = None
@@ -100,6 +111,14 @@ class TaskSnapshotUpdateRequest(BaseModel):
 class IntentUpdateRequest(BaseModel):
     intent_thread_id: str | None = Field(default=None, max_length=128)
     intent_snapshot: dict[str, Any] | None = None
+    conversation_snapshot: dict[str, Any] | None = None
+
+
+class ConversationStateUpdateRequest(BaseModel):
+    """Durable UI recovery state written while a normal chat is in progress."""
+
+    thread_id: str | None = Field(default=None, max_length=128)
+    conversation_snapshot: dict[str, Any] = Field(default_factory=dict)
 
 
 class ActivateRequest(BaseModel):
@@ -146,6 +165,14 @@ class CompleteBuiltinRunRequest(BaseModel):
     model: str | None = Field(default=None, max_length=256)
 
 
+class DemoSeedRequest(BaseModel):
+    """Canonical frontend fixture imported into the normal SQL session model."""
+
+    demo_key: str = Field(min_length=1, max_length=96)
+    session: dict[str, Any] = Field(default_factory=dict)
+    nodes: list[dict[str, Any]] = Field(default_factory=list, max_length=32)
+
+
 def _current_user_id(request: Request) -> str:
     user = getattr(request.state, "user", None)
     return str(user.id) if user is not None else get_effective_user_id()
@@ -515,13 +542,20 @@ def _artifact_refs(artifacts: Any) -> list[dict[str, Any]]:
     for artifact in artifacts[:20]:
         if not isinstance(artifact, dict):
             continue
-        refs.append(
-            {
-                "name": artifact.get("name") or artifact.get("filename") or "未命名产物",
-                "path": artifact.get("path") or artifact.get("artifact_url") or "",
-                "mime_type": artifact.get("mimeType") or artifact.get("mime_type") or None,
-            }
-        )
+        ref = {
+            "name": artifact.get("name") or artifact.get("filename") or "未命名产物",
+            "path": artifact.get("path") or artifact.get("artifact_url") or "",
+            "mime_type": artifact.get("mimeType") or artifact.get("mime_type") or None,
+        }
+        # A database recovery payload is already constrained when it is first
+        # persisted. Preserve it here so downstream agents can still consume
+        # an upstream report after the owning sandbox/output disk is gone.
+        content = artifact.get("content")
+        if isinstance(content, str):
+            ref["content"] = content
+            if artifact.get("content_truncated"):
+                ref["content_truncated"] = True
+        refs.append(ref)
     return refs
 
 
@@ -567,7 +601,7 @@ async def _artifact_refs_with_content(
     never be allowed to select another thread's user-data directory.
     """
     refs = _artifact_refs(artifacts)
-    if not thread_id or text_budget <= 0:
+    if text_budget <= 0:
         return refs, 0
 
     consumed = 0
@@ -587,6 +621,18 @@ async def _artifact_refs_with_content(
         if available <= 0:
             break
         excerpt_limit = min(_MAX_INLINE_ARTIFACT_CHARS, available)
+        persisted_content = ref.get("content")
+        if isinstance(persisted_content, str):
+            content = _truncate_text(persisted_content, excerpt_limit)
+            ref["content"] = content
+            ref["content_truncated"] = bool(
+                ref.get("content_truncated") or len(persisted_content) > len(content)
+            )
+            consumed += len(content)
+            inline_count += 1
+            continue
+        if not thread_id:
+            continue
         try:
             actual_path = await aresolve_thread_virtual_path(thread_id, normalized_path)
             if not await asyncio.to_thread(actual_path.is_file):
@@ -718,7 +764,7 @@ def _artifact_signature(artifact: dict[str, Any]) -> tuple[str, str, str]:
 
 
 def _artifact_manifest_refs(artifacts: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Persist only the display metadata received from the thread artifact API."""
+    """Persist display metadata plus any database recovery payload."""
     refs: list[dict[str, Any]] = []
     for artifact in artifacts:
         if not isinstance(artifact, dict):
@@ -726,18 +772,59 @@ def _artifact_manifest_refs(artifacts: list[dict[str, Any]]) -> list[dict[str, A
         path = str(artifact.get("path") or artifact.get("artifact_url") or "").strip()
         if not path.replace("\\", "/").lstrip("/").startswith("mnt/user-data/outputs/"):
             continue
-        refs.append(
-            {
-                "name": str(artifact.get("name") or artifact.get("filename") or path.rsplit("/", 1)[-1]),
-                "path": path,
-                "mime_type": artifact.get("mime_type") or artifact.get("mimeType") or None,
-                "modified_at": artifact.get("modified_at") or artifact.get("modifiedAt") or None,
-                "size_bytes": artifact.get("size_bytes") or artifact.get("sizeBytes") or None,
-            }
-        )
+        ref = {
+            "name": str(artifact.get("name") or artifact.get("filename") or path.rsplit("/", 1)[-1]),
+            "path": path,
+            "mime_type": artifact.get("mime_type") or artifact.get("mimeType") or None,
+            "modified_at": artifact.get("modified_at") or artifact.get("modifiedAt") or None,
+            "size_bytes": artifact.get("size_bytes") or artifact.get("sizeBytes") or None,
+        }
+        content = artifact.get("content")
+        if isinstance(content, str):
+            ref["content"] = content
+            if artifact.get("content_truncated"):
+                ref["content_truncated"] = True
+        refs.append(ref)
     return refs
 
 
+async def _persist_artifact_contents(
+    thread_id: str | None,
+    artifacts: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Copy text deliverables into SQL so history survives ephemeral disks."""
+    refs = _artifact_manifest_refs(artifacts)
+    if not thread_id:
+        return refs
+
+    enriched: list[dict[str, Any]] = []
+    for artifact in refs:
+        path = str(artifact.get("path") or "")
+        suffix = Path(path).suffix.lower()
+        if suffix not in _TEXT_ARTIFACT_SUFFIXES:
+            enriched.append(artifact)
+            continue
+        try:
+            resolved = await aresolve_thread_virtual_path(thread_id, path)
+            content = await asyncio.to_thread(resolved.read_text, encoding="utf-8", errors="replace")
+        except Exception:
+            # Artifact enrichment is recovery hardening, not a reason to fail
+            # an otherwise successful delivery. Keep the manifest metadata if
+            # the sandbox disappeared or its resolver/provider is unavailable.
+            enriched.append(artifact)
+            continue
+        if len(content) > _MAX_PERSISTED_ARTIFACT_CHARS:
+            artifact = {
+                **artifact,
+                "content": content[:_MAX_PERSISTED_ARTIFACT_CHARS],
+                "content_truncated": True,
+            }
+        else:
+            artifact = {**artifact, "content": content}
+        enriched.append(artifact)
+    return enriched
+
+
 async def _write_builtin_markdown_fallback(
     *, thread_id: str, kind: str, content: str
 ) -> dict[str, Any]:
@@ -784,7 +871,8 @@ async def _effective_delivery_material(
 
     Stale/rejected rows are deliberately excluded.  The persisted node answer
     provides a quick synopsis; the Markdown files provide the actual
-    deliverable body, read through the owning thread's virtual-path resolver.
+    deliverable body, preferably from its database recovery copy and otherwise
+    through the owning thread's virtual-path resolver.
     """
     remaining = _BUILTIN_MATERIAL_BUDGET
     lines: list[str] = ["【岗位有效交付】"]
@@ -920,6 +1008,30 @@ def _build_builtin_snapshot(
     }
 
 
+@router.post("/demo/seed", response_model=SessionDetailResponse)
+async def seed_demo_session(
+    request: Request, body: DemoSeedRequest
+) -> SessionDetailResponse:
+    """Idempotently materialise the frontend demo JSON for the current user.
+
+    The endpoint intentionally accepts a complete recovery snapshot instead of
+    inventing a second demo-only schema.  Imported rows therefore exercise the
+    same history, position switching, artifact and conversation code paths as
+    real collaboration sessions.
+    """
+
+    row, nodes = await _get_store(request).ensure_demo_session(
+        _current_user_id(request),
+        demo_key=body.demo_key,
+        session_payload=body.session,
+        nodes=body.nodes,
+    )
+    return SessionDetailResponse(
+        **row,
+        nodes=[_project_node(node) for node in nodes],
+    )
+
+
 @router.get("/sessions", response_model=list[SessionResponse])
 async def list_sessions(request: Request, external_task_id: str | None = None) -> list[SessionResponse]:
     rows = await _get_store(request).list_sessions(
@@ -964,6 +1076,39 @@ async def update_intent_snapshot(
         user_id,
         intent_thread_id=body.intent_thread_id,
         intent_snapshot=body.intent_snapshot,
+        intent_conversation=body.conversation_snapshot,
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    return _project_session(row)
+
+
+@router.put("/sessions/{session_id}/summary/state", response_model=SessionResponse)
+async def update_summary_conversation(
+    request: Request, session_id: str, body: ConversationStateUpdateRequest
+) -> SessionResponse:
+    store, user_id, _ = await _require_writable_session(request, session_id)
+    row = await store.update_session(
+        session_id,
+        user_id,
+        summary_thread_id=body.thread_id,
+        summary_conversation=body.conversation_snapshot,
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    return _project_session(row)
+
+
+@router.put("/sessions/{session_id}/action-plan/state", response_model=SessionResponse)
+async def update_action_plan_conversation(
+    request: Request, session_id: str, body: ConversationStateUpdateRequest
+) -> SessionResponse:
+    store, user_id, _ = await _require_writable_session(request, session_id)
+    row = await store.update_session(
+        session_id,
+        user_id,
+        action_plan_thread_id=body.thread_id,
+        action_plan_conversation=body.conversation_snapshot,
     )
     if row is None:
         raise HTTPException(status_code=404, detail="Position roundtable session not found")
@@ -1016,6 +1161,16 @@ async def archive_session(request: Request, session_id: str) -> SessionResponse:
     return _project_session(row)
 
 
+@router.delete("/sessions/{session_id}")
+async def delete_session(request: Request, session_id: str) -> dict[str, bool]:
+    deleted = await _get_store(request).delete_session(
+        session_id, _current_user_id(request)
+    )
+    if not deleted:
+        raise HTTPException(status_code=404, detail="Position roundtable session not found")
+    return {"deleted": True}
+
+
 @router.get("/sessions/{session_id}/nodes/{node_key}", response_model=NodeResponse)
 async def get_node(request: Request, session_id: str, node_key: str) -> NodeResponse:
     row = await _get_store(request).get_node(session_id, node_key, _current_user_id(request))
@@ -1048,6 +1203,38 @@ async def bind_node_thread(
     return _project_node(row)
 
 
+@router.put(
+    "/sessions/{session_id}/nodes/{node_key}/state",
+    response_model=NodeResponse,
+)
+async def update_node_conversation(
+    request: Request,
+    session_id: str,
+    node_key: str,
+    body: ConversationStateUpdateRequest,
+) -> NodeResponse:
+    store, user_id, _ = await _require_writable_session(request, session_id)
+    current = await store.get_node(session_id, node_key, user_id)
+    if current is None:
+        raise HTTPException(status_code=404, detail="Position roundtable node not found")
+    if (
+        body.thread_id
+        and current.get("thread_id")
+        and current["thread_id"] != body.thread_id
+    ):
+        raise HTTPException(status_code=409, detail="Position node is bound to another thread")
+    row = await store.update_node(
+        session_id,
+        node_key,
+        user_id,
+        thread_id=body.thread_id if body.thread_id else current.get("thread_id"),
+        conversation_snapshot=body.conversation_snapshot,
+    )
+    if row is None:
+        raise HTTPException(status_code=404, detail="Position roundtable node not found")
+    return _project_node(row)
+
+
 @router.post("/sessions/{session_id}/nodes/{node_key}/context")
 async def build_node_context(request: Request, session_id: str, node_key: str) -> dict[str, Any]:
     """Return the bounded hidden task/intent/upstream context for a node turn."""
@@ -1069,6 +1256,11 @@ async def complete_node_turn(
     # Artifact listing is best-effort on the client.  Never erase a previously
     # delivered report merely because one listing request transiently failed.
     effective_artifacts = body.artifact_manifest or list(node.get("artifact_manifest") or [])
+    if body.artifact_manifest:
+        effective_artifacts = await _persist_artifact_contents(
+            node.get("thread_id"),
+            effective_artifacts,
+        )
     has_markdown = any(_is_markdown_artifact(item) for item in effective_artifacts)
     markdown_changed = _markdown_delivery_changed(node.get("artifact_manifest"), effective_artifacts)
     requires_fresh_delivery = node.get("status") in {"rejected", "stale"}
@@ -1240,7 +1432,7 @@ async def _complete_builtin_run(
     if body.source_revisions != current_revisions:
         raise HTTPException(status_code=409, detail="岗位产物版本已变化，请基于最新交付重新生成")
 
-    artifacts = _artifact_manifest_refs(body.artifact_manifest)
+    artifacts = await _persist_artifact_contents(body.thread_id, body.artifact_manifest)
     markdown = [artifact for artifact in artifacts if _is_markdown_artifact(artifact)]
     if not markdown:
         # The report text is already visible in the chat stream. Do not lose
@@ -1252,6 +1444,7 @@ async def _complete_builtin_run(
             kind=kind,
             content=body.content,
         )
+        fallback_markdown["content"] = body.content
         artifacts.append(fallback_markdown)
         markdown = [fallback_markdown]
     if kind == "action_plan" and not any(_is_json_artifact(artifact) for artifact in artifacts):

offline-backend-20260512/backend/app/gateway/routers/system_settings.py

@@ -536,8 +536,8 @@ async def update_publish_squares(body: PublishSquaresUpdateRequest, request: Req
 class AppearanceResponse(BaseModel):
     """Effective system-wide appearance defaults."""
 
-    default_theme: Literal["light", "dark", "lightblue", "lightgreen"] = Field(
-        ..., description="System default theme — light, dark, lightblue or lightgreen"
+    default_theme: Literal["light", "dark", "lightblue", "lightgreen", "iceblue"] = Field(
+        ..., description="System default theme — light, dark, lightblue, lightgreen or iceblue"
     )
     default_skill_view: Literal["card", "list"] = Field(..., description="System default skill list display mode — card or list")
     default_agent_view: Literal["card", "list"] = Field(..., description="System default agent list display mode — card or list")
@@ -546,7 +546,7 @@ class AppearanceResponse(BaseModel):
 
 
 class AppearanceUpdateRequest(BaseModel):
-    default_theme: Literal["light", "dark", "lightblue", "lightgreen"] | None = Field(default=None)
+    default_theme: Literal["light", "dark", "lightblue", "lightgreen", "iceblue"] | None = Field(default=None)
     default_skill_view: Literal["card", "list"] | None = Field(default=None)
     default_agent_view: Literal["card", "list"] | None = Field(default=None)
     default_layout_mode: Literal["full", "compact"] | None = Field(default=None)

offline-backend-20260512/backend/packages/harness/deerflow/config/system_settings.py

@@ -102,9 +102,9 @@ class AppearanceSettings(BaseModel):
     toggle, and any future first-time loads.
     """
 
-    default_theme: Literal["light", "dark", "lightblue", "lightgreen"] = Field(
+    default_theme: Literal["light", "dark", "lightblue", "lightgreen", "iceblue"] = Field(
         default="light",
-        description="Theme applied for users who haven't picked their own (light / dark / lightblue / lightgreen)",
+        description="Theme applied for users who haven't picked their own (light / dark / lightblue / lightgreen / iceblue)",
     )
     default_skill_view: Literal["card", "list"] = Field(
         default="list",

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260723_05_position_roundtable_conversations.py

@@ -0,0 +1,69 @@
+"""Persist complete position-roundtable conversation recovery snapshots.
+
+Revision ID: 20260723_05
+Revises: 20260721_04
+Create Date: 2026-07-23
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+revision = "20260723_05"
+down_revision = "20260721_04"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    return table_name in inspect(op.get_bind()).get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    return column_name in {
+        column["name"] for column in inspect(op.get_bind()).get_columns(table_name)
+    }
+
+
+def upgrade() -> None:
+    long_text = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
+    session_table = "position_roundtable_sessions"
+    if _has_table(session_table):
+        for column in (
+            "intent_conversation",
+            "summary_conversation",
+            "action_plan_conversation",
+        ):
+            if not _has_column(session_table, column):
+                op.add_column(
+                    session_table,
+                    sa.Column(column, long_text, nullable=True),
+                )
+
+    node_table = "position_roundtable_nodes"
+    if _has_table(node_table) and not _has_column(node_table, "conversation_snapshot"):
+        op.add_column(
+            node_table,
+            sa.Column("conversation_snapshot", long_text, nullable=True),
+        )
+
+
+def downgrade() -> None:
+    node_table = "position_roundtable_nodes"
+    if _has_table(node_table) and _has_column(node_table, "conversation_snapshot"):
+        op.drop_column(node_table, "conversation_snapshot")
+
+    session_table = "position_roundtable_sessions"
+    if _has_table(session_table):
+        for column in (
+            "action_plan_conversation",
+            "summary_conversation",
+            "intent_conversation",
+        ):
+            if _has_column(session_table, column):
+                op.drop_column(session_table, column)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/position_roundtable/model.py

@@ -2,9 +2,10 @@
 
 This deliberately does **not** reuse ``roundtable_drafts`` or
 ``roundtable_jobs``. A session captures the selected business chain and the
-confirmed task intent; every chain seat then has one durable node record. The
-node points at an existing LangGraph thread, which continues to own the actual
-message history and generated artifacts.
+confirmed task intent; every chain seat then has one durable node record.
+LangGraph threads remain the execution source, while these tables also retain
+database-owned recovery copies of conversations and text deliverables so the
+workspace survives replacement of container-local checkpoint/output volumes.
 """
 
 from __future__ import annotations
@@ -31,13 +32,20 @@ class PositionRoundtableSessionRow(Base):
     chain_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     intent_thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
     intent_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # Database-owned recovery copy for the complete Step-1 conversation.  The
+    # normal LangGraph checkpoint remains the execution source, while this
+    # snapshot lets the position workspace recover after a checkpoint volume
+    # or container-local store has been replaced.
+    intent_conversation: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     # The two built-in closing agents own normal LangGraph threads too.  Keep
     # their ids on the position session so a refresh can continue the same
     # report/planning dialogue instead of silently starting a fresh context.
     summary_thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
     summary_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    summary_conversation: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     action_plan_thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
     action_plan_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    action_plan_conversation: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     # intent_pending -> active -> completed / archived. The router owns legal
     # transitions; the DB keeps it as a concise, portable string.
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="intent_pending", index=True)
@@ -71,12 +79,16 @@ class PositionRoundtableNodeRow(Base):
     seat_index: Mapped[int] = mapped_column(Integer, nullable=False)
     agent_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
     position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
-    # The real messages remain in the standard LangGraph/checkpoint store.
+    # The LangGraph/checkpoint store remains the execution source for messages.
     thread_id: Mapped[str | None] = mapped_column(String(128), nullable=True, unique=True)
     # locked / ready / running / done / stale / rejected / error
     status: Mapped[str] = mapped_column(String(32), nullable=False, default="locked", index=True)
     latest_answer: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     artifact_manifest: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # A complete, JSON-serialisable copy of the node chat plus its composer
+    # draft.  It is intentionally colocated with the node so deleting a
+    # position Session removes every recovery snapshot through the FK cascade.
+    conversation_snapshot: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     upstream_revision: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
     rejection_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/position_roundtable/sql.py

@@ -5,9 +5,10 @@ from __future__ import annotations
 import json
 from datetime import UTC, datetime
 from typing import Any
-from uuid import uuid4
+from uuid import NAMESPACE_URL, uuid4, uuid5
 
-from sqlalchemy import select
+from sqlalchemy import delete, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.position_roundtable.model import (
@@ -55,10 +56,13 @@ class PositionRoundtableRepository:
             "chain_snapshot": _loads(row.chain_snapshot),
             "intent_thread_id": row.intent_thread_id,
             "intent_snapshot": _loads(row.intent_snapshot),
+            "intent_conversation": _loads(row.intent_conversation),
             "summary_thread_id": row.summary_thread_id,
             "summary_snapshot": _loads(row.summary_snapshot),
+            "summary_conversation": _loads(row.summary_conversation),
             "action_plan_thread_id": row.action_plan_thread_id,
             "action_plan_snapshot": _loads(row.action_plan_snapshot),
+            "action_plan_conversation": _loads(row.action_plan_conversation),
             "status": row.status,
             "created_at": _time(row.created_at),
             "updated_at": _time(row.updated_at),
@@ -78,6 +82,7 @@ class PositionRoundtableRepository:
             "status": row.status,
             "latest_answer": row.latest_answer,
             "artifact_manifest": _loads(row.artifact_manifest) or [],
+            "conversation_snapshot": _loads(row.conversation_snapshot),
             "revision": row.revision,
             "upstream_revision": _loads(row.upstream_revision),
             "rejection_count": row.rejection_count or 0,
@@ -110,6 +115,134 @@ class PositionRoundtableRepository:
             await session.refresh(row)
             return self._session_dict(row)
 
+    async def ensure_demo_session(
+        self,
+        user_id: str,
+        *,
+        demo_key: str,
+        session_payload: dict[str, Any],
+        nodes: list[dict[str, Any]],
+    ) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+        """Create one deterministic, user-scoped demonstration session.
+
+        The browser owns the canonical JSON fixture so the same payload can be
+        rendered locally when the gateway is unavailable.  When the gateway is
+        reachable this method imports that payload exactly once.  Stable ids
+        make repeated initialisation and React StrictMode double effects safe,
+        while the user id in every UUID seed avoids cross-account collisions
+        on the globally unique LangGraph thread columns.
+        """
+
+        def stable_id(label: str) -> str:
+            return uuid5(
+                NAMESPACE_URL,
+                f"deerflow:position-roundtable-demo:{user_id}:{demo_key}:{label}",
+            ).hex
+
+        session_id = stable_id("session")
+        existing = await self.get_session(session_id, user_id)
+        if existing is not None:
+            return existing, (await self.list_nodes(session_id, user_id)) or []
+
+        now = datetime.now(UTC)
+        intent_thread_id = (
+            stable_id("intent-thread") if session_payload.get("intent_thread_id") else None
+        )
+        summary_thread_id = (
+            stable_id("summary-thread") if session_payload.get("summary_thread_id") else None
+        )
+        action_plan_thread_id = (
+            stable_id("action-plan-thread")
+            if session_payload.get("action_plan_thread_id")
+            else None
+        )
+
+        def with_thread_id(value: Any, thread_id: str | None) -> Any:
+            if not isinstance(value, dict) or not thread_id:
+                return value
+            return {**value, "thread_id": thread_id}
+
+        row = PositionRoundtableSessionRow(
+            id=session_id,
+            user_id=user_id,
+            external_task_id=session_payload.get("external_task_id") or None,
+            task_snapshot=_dumps(session_payload.get("task_snapshot") or {}),
+            chain_id=session_payload.get("chain_id") or None,
+            chain_snapshot=_dumps(session_payload.get("chain_snapshot")),
+            intent_thread_id=intent_thread_id,
+            intent_snapshot=_dumps(session_payload.get("intent_snapshot")),
+            intent_conversation=_dumps(session_payload.get("intent_conversation")),
+            summary_thread_id=summary_thread_id,
+            summary_snapshot=_dumps(
+                with_thread_id(session_payload.get("summary_snapshot"), summary_thread_id)
+            ),
+            summary_conversation=_dumps(session_payload.get("summary_conversation")),
+            action_plan_thread_id=action_plan_thread_id,
+            action_plan_snapshot=_dumps(
+                with_thread_id(
+                    session_payload.get("action_plan_snapshot"), action_plan_thread_id
+                )
+            ),
+            action_plan_conversation=_dumps(
+                session_payload.get("action_plan_conversation")
+            ),
+            status=str(session_payload.get("status") or "completed"),
+            created_at=now,
+            updated_at=now,
+        )
+
+        node_rows: list[PositionRoundtableNodeRow] = []
+        for index, node in enumerate(nodes):
+            node_key = str(node.get("node_key") or f"demo-node-{index + 1}")
+            node_rows.append(
+                PositionRoundtableNodeRow(
+                    id=stable_id(f"node:{node_key}"),
+                    session_id=session_id,
+                    node_key=node_key,
+                    stage_index=int(node.get("stage_index") or 0),
+                    seat_index=int(node.get("seat_index") or 0),
+                    agent_id=str(node.get("agent_id") or f"demo-agent-{index + 1}"),
+                    position_id=node.get("position_id") or None,
+                    thread_id=(
+                        stable_id(f"node-thread:{node_key}")
+                        if node.get("thread_id")
+                        else None
+                    ),
+                    status=str(node.get("status") or "done"),
+                    latest_answer=node.get("latest_answer") or None,
+                    artifact_manifest=_dumps(node.get("artifact_manifest") or []),
+                    conversation_snapshot=_dumps(node.get("conversation_snapshot")),
+                    revision=int(node.get("revision") or 0),
+                    upstream_revision=_dumps(node.get("upstream_revision")),
+                    rejection_count=int(node.get("rejection_count") or 0),
+                    last_rejection=_dumps(node.get("last_rejection")),
+                    invalidated_by=node.get("invalidated_by") or None,
+                    created_at=now,
+                    updated_at=now,
+                )
+            )
+
+        try:
+            async with self._sf() as session:
+                session.add(row)
+                session.add_all(node_rows)
+                await session.commit()
+                await session.refresh(row)
+                for node_row in node_rows:
+                    await session.refresh(node_row)
+        except IntegrityError:
+            # Another tab may have seeded the same user/demo key between the
+            # initial lookup and commit.  The deterministic row is the desired
+            # result, so recover it instead of returning a spurious 500.
+            existing = await self.get_session(session_id, user_id)
+            if existing is None:
+                raise
+            return existing, (await self.list_nodes(session_id, user_id)) or []
+
+        return self._session_dict(row), [
+            self._node_dict(node_row) for node_row in node_rows
+        ]
+
     async def list_sessions(
         self, user_id: str, *, external_task_id: str | None = None, limit: int = 50
     ) -> list[dict[str, Any]]:
@@ -136,10 +269,13 @@ class PositionRoundtableRepository:
         task_snapshot: dict[str, Any] | object = _UNSET,
         intent_thread_id: str | None | object = _UNSET,
         intent_snapshot: dict[str, Any] | None | object = _UNSET,
+        intent_conversation: dict[str, Any] | None | object = _UNSET,
         summary_thread_id: str | None | object = _UNSET,
         summary_snapshot: dict[str, Any] | None | object = _UNSET,
+        summary_conversation: dict[str, Any] | None | object = _UNSET,
         action_plan_thread_id: str | None | object = _UNSET,
         action_plan_snapshot: dict[str, Any] | None | object = _UNSET,
+        action_plan_conversation: dict[str, Any] | None | object = _UNSET,
         status: str | object = _UNSET,
     ) -> dict[str, Any] | None:
         async with self._sf() as session:
@@ -152,14 +288,20 @@ class PositionRoundtableRepository:
                 row.intent_thread_id = intent_thread_id or None
             if intent_snapshot is not _UNSET:
                 row.intent_snapshot = _dumps(intent_snapshot)
+            if intent_conversation is not _UNSET:
+                row.intent_conversation = _dumps(intent_conversation)
             if summary_thread_id is not _UNSET:
                 row.summary_thread_id = summary_thread_id or None
             if summary_snapshot is not _UNSET:
                 row.summary_snapshot = _dumps(summary_snapshot)
+            if summary_conversation is not _UNSET:
+                row.summary_conversation = _dumps(summary_conversation)
             if action_plan_thread_id is not _UNSET:
                 row.action_plan_thread_id = action_plan_thread_id or None
             if action_plan_snapshot is not _UNSET:
                 row.action_plan_snapshot = _dumps(action_plan_snapshot)
+            if action_plan_conversation is not _UNSET:
+                row.action_plan_conversation = _dumps(action_plan_conversation)
             if status is not _UNSET:
                 row.status = str(status)
             row.updated_at = datetime.now(UTC)
@@ -265,6 +407,7 @@ class PositionRoundtableRepository:
         status: str | object = _UNSET,
         latest_answer: str | None | object = _UNSET,
         artifact_manifest: list[dict[str, Any]] | object = _UNSET,
+        conversation_snapshot: dict[str, Any] | None | object = _UNSET,
         upstream_revision: dict[str, Any] | object = _UNSET,
         last_rejection: dict[str, Any] | None | object = _UNSET,
         invalidated_by: str | None | object = _UNSET,
@@ -293,6 +436,8 @@ class PositionRoundtableRepository:
                 row.latest_answer = latest_answer
             if artifact_manifest is not _UNSET:
                 row.artifact_manifest = _dumps(artifact_manifest)
+            if conversation_snapshot is not _UNSET:
+                row.conversation_snapshot = _dumps(conversation_snapshot)
             if upstream_revision is not _UNSET:
                 row.upstream_revision = _dumps(upstream_revision)
             if last_rejection is not _UNSET:
@@ -392,5 +537,38 @@ class PositionRoundtableRepository:
     async def archive_session(self, session_id: str, user_id: str) -> dict[str, Any] | None:
         return await self.update_session(session_id, user_id, status="archived")
 
+    async def delete_session(self, session_id: str, user_id: str) -> bool:
+        """Delete one user-owned history record and all of its node snapshots."""
+        async with self._sf() as session:
+            parent = (
+                await session.execute(
+                    select(PositionRoundtableSessionRow)
+                    .where(
+                        PositionRoundtableSessionRow.id == session_id,
+                        PositionRoundtableSessionRow.user_id == user_id,
+                    )
+                    .limit(1)
+                )
+            ).scalar_one_or_none()
+            if parent is None:
+                return False
+
+            # Do not rely on database-specific FK cascade settings. SQLite
+            # deployments may not have PRAGMA foreign_keys enabled, which
+            # would otherwise leave orphaned conversation/artifact snapshots.
+            await session.execute(
+                delete(PositionRoundtableNodeRow).where(
+                    PositionRoundtableNodeRow.session_id == session_id
+                )
+            )
+            await session.execute(
+                delete(PositionRoundtableSessionRow).where(
+                    PositionRoundtableSessionRow.id == session_id,
+                    PositionRoundtableSessionRow.user_id == user_id,
+                )
+            )
+            await session.commit()
+            return True
+
     async def list_results(self, session_id: str, user_id: str) -> list[dict[str, Any]] | None:
         return await self.list_nodes(session_id, user_id)

offline-backend-20260512/backend/tests/test_position_roundtable_sessions.py

@@ -89,6 +89,200 @@ async def test_session_activation_freezes_chain_and_creates_nodes_once(store):
     assert [node["node_key"] for node in repeated_nodes] == [node["node_key"] for node in created_nodes]
 
 
+@pytest.mark.asyncio
+async def test_conversation_snapshots_round_trip_and_history_delete_removes_nodes(store):
+    """数据库恢复副本必须覆盖三个对话阶段、岗位节点和未发送草稿。"""
+    created = await store.create_session(
+        "user-a",
+        external_task_id="task-persisted",
+        task_snapshot={"title": "持久化测试"},
+    )
+    intent_conversation = {
+        "messages": [{"id": "intent-user-1", "type": "human", "content": "请分析任务"}],
+        "timeline": [{"kind": "message", "message_id": "intent-user-1"}],
+        "draft": "尚未发送的意图补充",
+    }
+    summary_conversation = {
+        "messages": [{"id": "summary-ai-1", "type": "ai", "content": "方案总结"}],
+        "draft": "总结修改意见",
+    }
+    action_conversation = {
+        "messages": [{"id": "action-ai-1", "type": "ai", "content": "行动规划"}],
+        "draft": "",
+    }
+    updated = await store.update_session(
+        created["id"],
+        "user-a",
+        intent_thread_id="intent-thread",
+        intent_conversation=intent_conversation,
+        summary_conversation=summary_conversation,
+        action_plan_conversation=action_conversation,
+    )
+    assert updated is not None
+    assert updated["intent_conversation"] == intent_conversation
+    assert updated["summary_conversation"] == summary_conversation
+    assert updated["action_plan_conversation"] == action_conversation
+
+    _, nodes = await store.activate_session(
+        created["id"],
+        "user-a",
+        chain_id="chain-a",
+        chain_snapshot={"id": "chain-a"},
+        nodes=[
+            {
+                "node_key": "stage-1-seat-1-agent-a",
+                "stage_index": 0,
+                "seat_index": 0,
+                "agent_id": "agent-a",
+                "position_id": "planning",
+                "status": "ready",
+            }
+        ],
+    )
+    node_conversation = {
+        "messages": [{"id": "node-ai-1", "type": "ai", "content": "岗位答复"}],
+        "draft": "岗位草稿",
+    }
+    saved_node = await store.update_node(
+        created["id"],
+        nodes[0]["node_key"],
+        "user-a",
+        conversation_snapshot=node_conversation,
+    )
+    assert saved_node is not None
+    assert saved_node["conversation_snapshot"] == node_conversation
+
+    # Ownership checks apply to delete as well.
+    assert not await store.delete_session(created["id"], "user-b")
+    assert await store.get_session(created["id"], "user-a") is not None
+    assert await store.delete_session(created["id"], "user-a")
+    assert await store.get_session(created["id"], "user-a") is None
+    assert await store.list_nodes(created["id"], "user-a") is None
+
+
+@pytest.mark.asyncio
+async def test_demo_seed_is_complete_idempotent_and_user_scoped(store):
+    payload = {
+        "external_task_id": "position-demo-full-v1",
+        "task_snapshot": {"title": "暑期活动"},
+        "chain_id": "position-demo-chain",
+        "chain_snapshot": {
+            "id": "position-demo-chain",
+            "seats": [
+                {
+                    "agent_id": "demo-agent",
+                    "name": "演示分析智能体",
+                    "position_id": "planning",
+                }
+            ],
+        },
+        "intent_thread_id": "frontend-placeholder",
+        "intent_snapshot": {
+            "objective": "形成完整宣传方案",
+            "constraints": ["预算不少于 15 万元"],
+        },
+        "intent_conversation": {
+            "messages": [{"id": 1, "sender": "agent", "content": "意图已确认"}],
+            "inputDraft": "",
+        },
+        "summary_thread_id": "frontend-placeholder",
+        "summary_snapshot": {"status": "done", "content": "方案总结"},
+        "summary_conversation": {
+            "messages": [{"id": "summary-ai", "type": "ai", "content": "方案总结"}]
+        },
+        "action_plan_thread_id": "frontend-placeholder",
+        "action_plan_snapshot": {"status": "done", "content": "行动规划"},
+        "action_plan_conversation": {
+            "messages": [{"id": "action-ai", "type": "ai", "content": "行动规划"}]
+        },
+        "status": "completed",
+    }
+    nodes = [
+        {
+            "node_key": "stage-1-seat-1-demo-agent",
+            "stage_index": 0,
+            "seat_index": 0,
+            "agent_id": "demo-agent",
+            "position_id": "planning",
+            "thread_id": "frontend-placeholder",
+            "status": "done",
+            "latest_answer": "报告已完成",
+            "artifact_manifest": [
+                {
+                    "name": "演示报告.md",
+                    "path": "mnt/user-data/outputs/演示报告.md",
+                    "mime_type": "text/markdown",
+                    "content": "# 演示报告\n\n完整正文",
+                }
+            ],
+            "conversation_snapshot": {
+                "messages": [{"id": "node-ai", "type": "ai", "content": "报告已完成"}],
+                "inputDraft": "",
+            },
+            "revision": 1,
+        }
+    ]
+
+    first, first_nodes = await store.ensure_demo_session(
+        "user-a",
+        demo_key="full-v1",
+        session_payload=payload,
+        nodes=nodes,
+    )
+    repeated, repeated_nodes = await store.ensure_demo_session(
+        "user-a",
+        demo_key="full-v1",
+        session_payload={**payload, "task_snapshot": {"title": "不应覆盖"}},
+        nodes=[],
+    )
+    other_user, other_nodes = await store.ensure_demo_session(
+        "user-b",
+        demo_key="full-v1",
+        session_payload=payload,
+        nodes=nodes,
+    )
+
+    assert first["id"] == repeated["id"]
+    assert first["task_snapshot"]["title"] == "暑期活动"
+    assert repeated["task_snapshot"]["title"] == "暑期活动"
+    assert first["intent_conversation"] == payload["intent_conversation"]
+    assert first["summary_snapshot"]["thread_id"] == first["summary_thread_id"]
+    assert first["action_plan_snapshot"]["thread_id"] == first["action_plan_thread_id"]
+    assert first_nodes[0]["artifact_manifest"][0]["content"] == "# 演示报告\n\n完整正文"
+    assert repeated_nodes[0]["id"] == first_nodes[0]["id"]
+    assert other_user["id"] != first["id"]
+    assert other_user["intent_thread_id"] != first["intent_thread_id"]
+    assert other_nodes[0]["thread_id"] != first_nodes[0]["thread_id"]
+
+
+@pytest.mark.asyncio
+async def test_embedded_artifact_content_survives_missing_sandbox(monkeypatch):
+    """后续岗位优先读取 SQL 正文，不能因沙箱卷丢失而只拿到文件名。"""
+    async def unexpected_resolve(*_args, **_kwargs):
+        raise AssertionError("embedded database content should avoid sandbox resolution")
+
+    monkeypatch.setattr(
+        position_roundtable_router,
+        "aresolve_thread_virtual_path",
+        unexpected_resolve,
+    )
+    artifacts, consumed = await position_roundtable_router._artifact_refs_with_content(
+        "removed-thread",
+        [
+            {
+                "name": "report.md",
+                "path": "mnt/user-data/outputs/report.md",
+                "mime_type": "text/markdown",
+                "content": "# 数据库恢复报告\n\n完整正文",
+            }
+        ],
+        10_000,
+    )
+
+    assert artifacts[0]["content"] == "# 数据库恢复报告\n\n完整正文"
+    assert consumed == len("# 数据库恢复报告\n\n完整正文")
+
+
 def test_task_external_context_reaches_agent_runtime_context():
     """岗位会商的左侧任务快照不能在网关白名单中丢失。"""
     task_context = {