chore(persistence): 增加 Workflow Studio 表重建开关与迁移兼容修正

提交内容：为数据库生命周期增加按白名单清单重建 Workflow Studio 表的受限流程，限定仅 MySQL 且检查当前连接数据库名称一致，避免误删非目标 schema。

启动流程中新增 DEER_FLOW_REBUILD_WORKFLOW_SCHEMA 环境变量入口与持久层重建 API，并在 lifespan 里按需执行；导出到 persistence 包，便于后续调用与测试复用。

迁移与模型修复：将 workflow 相关大文本字段切到 MySQL LONGTEXT，不再给 TEXT/LONGTEXT 设置 server_default；新增 draft_canvas_schema_json 的迁移回填与非空约束变更，规避 MySQL 5.7 对默认值限制。

测试覆盖：补充 workflow 定义 ddl 长文本断言与 schema 重建保护性测试（仅允许清单表、数据库名不匹配直接拒绝），并把新测试文件加入版本追踪。

.gitignore

@@ -126,10 +126,11 @@ offline-backend-20260512/backend/tests/*
 # counter, event seq) — required in CI per docs/WORKFLOW_STUDIO_BACKEND_REMEDIATION_ZH.md P1-06.
 !offline-backend-20260512/backend/tests/test_workflow_schemas.py
 !offline-backend-20260512/backend/tests/test_workflow_definitions.py
-!offline-backend-20260512/backend/tests/test_workflow_engine.py
+!offline-backend-20260512/backend/tests/test_workflow_engine.py
-!offline-backend-20260512/backend/tests/test_workflow_runs.py
+!offline-backend-20260512/backend/tests/test_workflow_runs.py
-!offline-backend-20260512/backend/tests/test_workflow_concurrency.py
+!offline-backend-20260512/backend/tests/test_workflow_concurrency.py
-# Writing-mode first-report pipeline enforcement regressions.
+!offline-backend-20260512/backend/tests/test_workflow_schema_rebuild.py
+# Writing-mode first-report pipeline enforcement regressions.
 !offline-backend-20260512/backend/tests/test_writing_mode_report_middleware.py
 !offline-backend-20260512/backend/tests/test_writing_rewrite_prompt.py
 # LLM call timeout retry classification regression coverage.

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

@@ -996,11 +996,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     logger.info(f"Starting API Gateway on {config.host}:{config.port}")
 
     # Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store)
-    async with langgraph_runtime(app):
+    async with langgraph_runtime(app):
-        logger.info("LangGraph runtime initialised")
+        logger.info("LangGraph runtime initialised")
-
+
-        # Run Alembic migrations before anything else touches the schema.
+        # One-time repair switch for a failed, unused Workflow Studio schema.
-        await _run_db_migrations()
+        # The helper is intentionally MySQL-only and drops an exact allow-list
+        # in the configured current database; it never expands ``workflow_%``.
+        # Keep this opt-in so established workflow data is never cleared during
+        # an ordinary restart.
+        if os.getenv("DEER_FLOW_REBUILD_WORKFLOW_SCHEMA", "").lower() in {"1", "true", "yes"}:
+            from deerflow.persistence.engine import rebuild_workflow_schema_for_mysql
+
+            logger.warning(
+                "DEER_FLOW_REBUILD_WORKFLOW_SCHEMA is enabled; rebuilding the unused workflow schema"
+            )
+            await rebuild_workflow_schema_for_mysql()
+
+        # Run Alembic migrations before anything else touches the schema.
+        await _run_db_migrations()
 
         # Local Wiki mirror/index scans are opt-in and run outside request
         # handling. Per-library database leases keep multi-worker startup safe.

offline-backend-20260512/backend/packages/harness/deerflow/client.py

@@ -264,12 +264,35 @@ class DeerFlowClient:
         return [{"name": tc["name"], "args": tc["args"], "id": tc.get("id")} for tc in tool_calls]
 
     @staticmethod
-    def _ai_text_event(msg_id: str | None, text: str, usage: dict | None) -> "StreamEvent":
+    def _ai_text_event(
-        """Build a ``messages-tuple`` AI text event, attaching usage when present."""
+        msg_id: str | None,
-        data: dict[str, Any] = {"type": "ai", "content": text, "id": msg_id}
+        text: str,
-        if usage:
+        usage: dict | None,
-            data["usage_metadata"] = usage
+        reasoning: str = "",
-        return StreamEvent(type="messages-tuple", data=data)
+    ) -> "StreamEvent":
+        """Build a ``messages-tuple`` AI text event, preserving reasoning deltas."""
+        data: dict[str, Any] = {"type": "ai", "content": text, "id": msg_id}
+        if usage:
+            data["usage_metadata"] = usage
+        # Thinking-enabled OpenAI-compatible providers commonly put their
+        # streaming tokens in ``additional_kwargs.reasoning_content`` while
+        # ``content`` remains empty. Keep it on the wire for consumers that
+        # render a custom stream (such as the multi-agent roundtable).
+        if reasoning:
+            data["additional_kwargs"] = {"reasoning_content": reasoning}
+        return StreamEvent(type="messages-tuple", data=data)
+
+    @staticmethod
+    def _reasoning_text(msg: Any) -> str:
+        """Return a provider reasoning delta when it is available as text."""
+        additional_kwargs = getattr(msg, "additional_kwargs", None) or {}
+        if not isinstance(additional_kwargs, dict):
+            return ""
+        reasoning = (
+            additional_kwargs.get("reasoning_content")
+            or additional_kwargs.get("reasoning")
+        )
+        return reasoning if isinstance(reasoning, str) else ""
 
     @staticmethod
     def _ai_tool_calls_event(msg_id: str | None, tool_calls) -> "StreamEvent":
@@ -618,14 +641,15 @@ class DeerFlowClient:
 
                 msg_id = getattr(msg_chunk, "id", None)
 
-                if isinstance(msg_chunk, AIMessage):
+                if isinstance(msg_chunk, AIMessage):
-                    text = self._extract_text(msg_chunk.content)
+                    text = self._extract_text(msg_chunk.content)
-                    counted_usage = _account_usage(msg_id, msg_chunk.usage_metadata)
+                    reasoning = self._reasoning_text(msg_chunk)
-
+                    counted_usage = _account_usage(msg_id, msg_chunk.usage_metadata)
-                    if text:
+
-                        if msg_id:
+                    if text or reasoning:
-                            streamed_ids.add(msg_id)
+                        if msg_id:
-                        yield self._ai_text_event(msg_id, text, counted_usage)
+                            streamed_ids.add(msg_id)
+                        yield self._ai_text_event(msg_id, text, counted_usage, reasoning)
 
                     if msg_chunk.tool_calls:
                         if msg_id:
@@ -661,9 +685,10 @@ class DeerFlowClient:
                     if msg.tool_calls:
                         yield self._ai_tool_calls_event(msg_id, msg.tool_calls)
 
-                    text = self._extract_text(msg.content)
+                    text = self._extract_text(msg.content)
-                    if text:
+                    reasoning = self._reasoning_text(msg)
-                        yield self._ai_text_event(msg_id, text, counted_usage)
+                    if text or reasoning:
+                        yield self._ai_text_event(msg_id, text, counted_usage, reasoning)
 
                 elif isinstance(msg, ToolMessage):
                     yield self._tool_message_event(msg)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/__init__.py

@@ -8,6 +8,18 @@ Usage:
     from deerflow.persistence import init_engine, close_engine, get_session_factory
 """
 
-from deerflow.persistence.engine import close_engine, get_engine, get_session_factory, init_engine
+from deerflow.persistence.engine import (
-
+    close_engine,
-__all__ = ["close_engine", "get_engine", "get_session_factory", "init_engine"]
+    get_engine,
+    get_session_factory,
+    init_engine,
+    rebuild_workflow_schema_for_mysql,
+)
+
+__all__ = [
+    "close_engine",
+    "get_engine",
+    "get_session_factory",
+    "init_engine",
+    "rebuild_workflow_schema_for_mysql",
+]

offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py

@@ -38,7 +38,21 @@ _session_factory: async_sessionmaker[AsyncSession] | None = None
 # "got Future attached to a different loop". We capture the engine's home loop here so
 # best-effort writes coming from worker threads / other loops can be routed back to it
 # (see ``fire_and_forget_db_write``).
-_engine_loop: asyncio.AbstractEventLoop | None = None
+_engine_loop: asyncio.AbstractEventLoop | None = None
+
+# These are the complete workflow-studio persistence tables owned by DeerFlow.
+# Keep this explicit allow-list instead of treating ``workflow_%`` as a SQL
+# wildcard: another application might share a schema and have its own table
+# with that prefix.
+_WORKFLOW_SCHEMA_TABLE_NAMES = (
+    "workflow_definitions",
+    "workflow_versions",
+    "workflow_runs",
+    "workflow_node_runs",
+    "workflow_run_events",
+    "workflow_run_artifacts",
+    "workflow_data_sources",
+)
 
 
 def _add_runtime_vendor_path() -> None:
@@ -527,9 +541,73 @@ def get_engine() -> AsyncEngine | None:
     return _engine
 
 
-def get_engine_loop() -> asyncio.AbstractEventLoop | None:
+def get_engine_loop() -> asyncio.AbstractEventLoop | None:
-    """Return the event loop that owns the engine's connection pool (or None)."""
+    """Return the event loop that owns the engine's connection pool (or None)."""
-    return _engine_loop
+    return _engine_loop
+
+
+async def rebuild_workflow_schema_for_mysql() -> list[str]:
+    """Drop and recreate DeerFlow's workflow tables in the current MySQL schema.
+
+    This is a deliberately narrow recovery operation for an unused Workflow
+    Studio installation. It never uses a prefix wildcard, never changes the
+    selected database, and refuses to run unless the engine is connected to
+    MySQL and ``SELECT DATABASE()`` confirms the URL's exact target schema.
+
+    Returns the allow-listed workflow tables that existed before the rebuild.
+    """
+    engine = get_engine()
+    if engine is None:
+        raise RuntimeError("Cannot rebuild workflow schema: persistence engine is not initialized")
+    if engine.dialect.name != "mysql":
+        raise RuntimeError(
+            "DEER_FLOW_REBUILD_WORKFLOW_SCHEMA is supported only for a MySQL persistence backend"
+        )
+
+    expected_database = engine.url.database
+    if not expected_database:
+        raise RuntimeError("Cannot rebuild workflow schema: MySQL URL has no database name")
+
+    # Make the operation safe when this helper is invoked independently from
+    # normal app startup as well.
+    import deerflow.persistence.models  # noqa: F401 — register ORM models
+    from deerflow.persistence.base import Base
+
+    missing_models = [name for name in _WORKFLOW_SCHEMA_TABLE_NAMES if name not in Base.metadata.tables]
+    if missing_models:
+        raise RuntimeError(
+            "Cannot rebuild workflow schema: ORM metadata is missing " + ", ".join(missing_models)
+        )
+    workflow_tables = [Base.metadata.tables[name] for name in _WORKFLOW_SCHEMA_TABLE_NAMES]
+
+    def _rebuild(sync_conn) -> list[str]:
+        from sqlalchemy import inspect as sa_inspect
+
+        current_database = sync_conn.exec_driver_sql("SELECT DATABASE()").scalar_one_or_none()
+        if current_database != expected_database:
+            raise RuntimeError(
+                "Refusing workflow schema rebuild: connected database "
+                f"{current_database!r} does not match configured database {expected_database!r}"
+            )
+
+        existing = set(sa_inspect(sync_conn).get_table_names())
+        present = [name for name in _WORKFLOW_SCHEMA_TABLE_NAMES if name in existing]
+        # SQLAlchemy orders parent/child tables around the foreign key from
+        # workflow_versions to workflow_definitions. Only tables in the exact
+        # allow-list above are ever passed to drop_all/create_all.
+        Base.metadata.drop_all(sync_conn, tables=workflow_tables, checkfirst=True)
+        Base.metadata.create_all(sync_conn, tables=workflow_tables, checkfirst=True)
+        return present
+
+    async with engine.begin() as conn:
+        rebuilt = await conn.run_sync(_rebuild)
+
+    logger.warning(
+        "Rebuilt workflow schema in current MySQL database %s; tables reset: %s",
+        expected_database,
+        ", ".join(rebuilt) if rebuilt else "none (created fresh)",
+    )
+    return rebuilt
 
 
 def fire_and_forget_db_write(coro) -> None:

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260828_01_workflow_studio_tables.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import sqlalchemy as sa
 from alembic import op
+from sqlalchemy.dialects.mysql import LONGTEXT
 
 revision = "20260828_01"
 down_revision = "20260825_01"
@@ -19,16 +20,21 @@ depends_on = None
 def upgrade() -> None:
     bind = op.get_bind()
     inspector = sa.inspect(bind)
+    # Match PortableLongText used by the ORM. MySQL TEXT only holds 64 KiB,
+    # which is too small for serialized workflow graphs.
+    long_text = LONGTEXT() if bind.dialect.name == "mysql" else sa.Text()
     if not inspector.has_table("workflow_definitions"):
         op.create_table(
             "workflow_definitions",
             sa.Column("id", sa.String(length=64), primary_key=True),
             sa.Column("name", sa.String(length=255), nullable=False),
-            sa.Column("description", sa.Text(), nullable=False, server_default=""),
+            # Older MySQL versions reject defaults on TEXT/LONGTEXT columns.
+            # The repository always writes both fields explicitly.
+            sa.Column("description", sa.Text(), nullable=False),
             sa.Column("owner_id", sa.String(length=64), nullable=False),
             sa.Column("status", sa.String(length=32), nullable=False, server_default="draft"),
             sa.Column("draft_revision", sa.Integer(), nullable=False, server_default="0"),
-            sa.Column("draft_graph_json", sa.Text(), nullable=False),
+            sa.Column("draft_graph_json", long_text, nullable=False),
             sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
             sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
         )
@@ -41,13 +47,13 @@ def upgrade() -> None:
             sa.Column("id", sa.String(length=64), primary_key=True),
             sa.Column("workflow_id", sa.String(length=64), nullable=False),
             sa.Column("version_number", sa.Integer(), nullable=False),
-            sa.Column("graph_json", sa.Text(), nullable=False),
+            sa.Column("graph_json", long_text, nullable=False),
             sa.Column("graph_hash", sa.String(length=64), nullable=False),
             sa.Column("input_schema", sa.Text(), nullable=True),
             sa.Column("output_schema", sa.Text(), nullable=True),
             sa.Column("published_by", sa.String(length=64), nullable=True),
             sa.Column("published_at", sa.DateTime(timezone=True), nullable=False),
-            sa.Column("change_note", sa.Text(), nullable=False, server_default=""),
+            sa.Column("change_note", sa.Text(), nullable=False),
             sa.ForeignKeyConstraint(["workflow_id"], ["workflow_definitions.id"], ondelete="CASCADE"),
             sa.UniqueConstraint("workflow_id", "version_number", name="uq_workflow_versions_wf_num"),
         )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260829_02_workflow_canvas_schema.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import sqlalchemy as sa
 from alembic import op
+from sqlalchemy.dialects.mysql import LONGTEXT
 
 revision = "20260829_02"
 down_revision = "20260829_01"
@@ -28,10 +29,30 @@ def upgrade() -> None:
         return
     existing = {column["name"] for column in inspector.get_columns("workflow_definitions")}
     if "draft_canvas_schema_json" not in existing:
+        # Follow roundtable/deep-research migrations: large JSON is LONGTEXT on
+        # MySQL, and we must not use a TEXT/LONGTEXT server default (unsupported
+        # by MySQL 5.7). Add it nullable, backfill legacy rows, then enforce the
+        # ORM's NOT NULL contract.
+        canvas_type = LONGTEXT() if bind.dialect.name == "mysql" else sa.Text()
         op.add_column(
             "workflow_definitions",
-            sa.Column("draft_canvas_schema_json", sa.Text(), nullable=False, server_default="'{}'"),
+            sa.Column("draft_canvas_schema_json", canvas_type, nullable=True),
         )
+        op.execute(
+            sa.text(
+                "UPDATE workflow_definitions "
+                "SET draft_canvas_schema_json = '{}' "
+                "WHERE draft_canvas_schema_json IS NULL"
+            )
+        )
+        # ``batch_alter_table`` uses a native ALTER TABLE on MySQL while also
+        # handling SQLite's lack of ALTER COLUMN for legacy SQLite deployments.
+        with op.batch_alter_table("workflow_definitions") as batch_op:
+            batch_op.alter_column(
+                "draft_canvas_schema_json",
+                existing_type=canvas_type,
+                nullable=False,
+            )
     if "published_canvas_schema_hash" not in existing:
         op.add_column(
             "workflow_definitions",

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

@@ -33,7 +33,10 @@ class WorkflowDefinitionRow(Base):
     # Coze canvas document (opaque JSON text). Never interpreted by the runtime;
     # the executable WorkflowGraph lives in draft_graph_json above.
     draft_canvas_schema_json: Mapped[str] = mapped_column(
-        PortableLongText(), default="{}", server_default=sa_text("'{}'"), nullable=False
+        # Keep this as an ORM-side default only. MySQL 5.7 rejects DEFAULT on
+        # TEXT/LONGTEXT columns, while workflow creation always supplies this
+        # value explicitly as well.
+        PortableLongText(), default="{}", nullable=False
     )
     # sha256 of the canvas document the current published version was compiled
     # from (audit trail only; publish always freezes the execution graph).

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

@@ -7,6 +7,8 @@ import json
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
+from sqlalchemy.dialects import mysql
+from sqlalchemy.schema import CreateTable
 
 from app.gateway.routers import workflow_resources, workflows, workflows_coze_compat
 from deerflow.persistence.workflows import (
@@ -14,10 +16,20 @@ from deerflow.persistence.workflows import (
     WorkflowDraftConflictError,
     make_workflow_store,
 )
+from deerflow.persistence.workflows.model import WorkflowDefinitionRow
 from deerflow.workflows.samples import load_sample_graph
 from deerflow.workflows.validator import validate_workflow_graph
 
 
+def test_workflow_definition_mysql_uses_longtext_without_a_server_default() -> None:
+    """MySQL 5.7 rejects DEFAULT values on TEXT/LONGTEXT columns."""
+    ddl = str(CreateTable(WorkflowDefinitionRow.__table__).compile(dialect=mysql.dialect())).upper()
+
+    assert "DRAFT_GRAPH_JSON LONGTEXT NOT NULL" in ddl
+    assert "DRAFT_CANVAS_SCHEMA_JSON LONGTEXT NOT NULL" in ddl
+    assert "DRAFT_CANVAS_SCHEMA_JSON LONGTEXT NOT NULL DEFAULT" not in ddl
+
+
 @pytest.mark.asyncio
 async def test_memory_store_create_draft_publish() -> None:
     store = make_workflow_store(None)

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

@@ -0,0 +1,90 @@
+"""Safety tests for the one-time MySQL Workflow Studio schema rebuild."""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from types import SimpleNamespace
+
+import pytest
+import sqlalchemy as sa
+
+import deerflow.persistence.engine as persistence_engine
+from deerflow.persistence.base import Base
+
+
+class _ScalarResult:
+    def __init__(self, value: str | None) -> None:
+        self._value = value
+
+    def scalar_one_or_none(self) -> str | None:
+        return self._value
+
+
+class _SyncConnection:
+    def __init__(self, database: str | None) -> None:
+        self._database = database
+
+    def exec_driver_sql(self, statement: str) -> _ScalarResult:
+        assert statement == "SELECT DATABASE()"
+        return _ScalarResult(self._database)
+
+
+class _AsyncConnection:
+    def __init__(self, sync_connection: _SyncConnection) -> None:
+        self._sync_connection = sync_connection
+
+    async def run_sync(self, callback):
+        return callback(self._sync_connection)
+
+
+class _AsyncEngine:
+    def __init__(self, database: str, connected_database: str | None) -> None:
+        self.dialect = SimpleNamespace(name="mysql")
+        self.url = SimpleNamespace(database=database)
+        self._connection = _AsyncConnection(_SyncConnection(connected_database))
+
+    @asynccontextmanager
+    async def begin(self):
+        yield self._connection
+
+
+@pytest.mark.asyncio
+async def test_rebuild_workflow_schema_only_targets_known_tables(monkeypatch: pytest.MonkeyPatch) -> None:
+    engine = _AsyncEngine("deerflow", "deerflow")
+    monkeypatch.setattr(persistence_engine, "get_engine", lambda: engine)
+    monkeypatch.setattr(
+        sa,
+        "inspect",
+        lambda _connection: SimpleNamespace(
+            get_table_names=lambda: {"workflow_definitions", "unrelated_workflow_cache"}
+        ),
+    )
+
+    calls: dict[str, list[str]] = {}
+    monkeypatch.setattr(
+        Base.metadata,
+        "drop_all",
+        lambda _connection, *, tables, checkfirst: calls.setdefault("drop", [table.name for table in tables]),
+    )
+    monkeypatch.setattr(
+        Base.metadata,
+        "create_all",
+        lambda _connection, *, tables, checkfirst: calls.setdefault("create", [table.name for table in tables]),
+    )
+
+    rebuilt = await persistence_engine.rebuild_workflow_schema_for_mysql()
+
+    expected = list(persistence_engine._WORKFLOW_SCHEMA_TABLE_NAMES)
+    assert rebuilt == ["workflow_definitions"]
+    assert calls == {"drop": expected, "create": expected}
+    assert "unrelated_workflow_cache" not in calls["drop"]
+
+
+@pytest.mark.asyncio
+async def test_rebuild_workflow_schema_refuses_unexpected_database(monkeypatch: pytest.MonkeyPatch) -> None:
+    engine = _AsyncEngine("deerflow", "another_database")
+    monkeypatch.setattr(persistence_engine, "get_engine", lambda: engine)
+    monkeypatch.setattr(sa, "inspect", lambda _connection: pytest.fail("must not inspect tables"))
+
+    with pytest.raises(RuntimeError, match="does not match configured database"):
+        await persistence_engine.rebuild_workflow_schema_for_mysql()