diff --git a/.gitignore b/.gitignore index b5fa1a07..371449b7 100644 --- a/.gitignore +++ b/.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_runs.py -!offline-backend-20260512/backend/tests/test_workflow_concurrency.py -# Writing-mode first-report pipeline enforcement regressions. +!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_concurrency.py +!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. diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py index 25cbbeea..067e5520 100644 --- a/offline-backend-20260512/backend/app/gateway/app.py +++ b/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): - logger.info("LangGraph runtime initialised") - - # Run Alembic migrations before anything else touches the schema. - await _run_db_migrations() + async with langgraph_runtime(app): + logger.info("LangGraph runtime initialised") + + # One-time repair switch for a failed, unused Workflow Studio schema. + # 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. diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/client.py b/offline-backend-20260512/backend/packages/harness/deerflow/client.py index 1e76d27a..94b60a3e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/client.py +++ b/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": - """Build a ``messages-tuple`` AI text event, attaching usage when present.""" - data: dict[str, Any] = {"type": "ai", "content": text, "id": msg_id} - if usage: - data["usage_metadata"] = usage - return StreamEvent(type="messages-tuple", data=data) + def _ai_text_event( + msg_id: str | None, + text: str, + usage: dict | None, + reasoning: str = "", + ) -> "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): - text = self._extract_text(msg_chunk.content) - counted_usage = _account_usage(msg_id, msg_chunk.usage_metadata) - - if text: - if msg_id: - streamed_ids.add(msg_id) - yield self._ai_text_event(msg_id, text, counted_usage) + if isinstance(msg_chunk, AIMessage): + text = self._extract_text(msg_chunk.content) + reasoning = self._reasoning_text(msg_chunk) + counted_usage = _account_usage(msg_id, msg_chunk.usage_metadata) + + if text or reasoning: + if msg_id: + 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) - if text: - yield self._ai_text_event(msg_id, text, counted_usage) + text = self._extract_text(msg.content) + reasoning = self._reasoning_text(msg) + 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) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/__init__.py index 09b577c7..0e406979 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/__init__.py +++ b/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 - -__all__ = ["close_engine", "get_engine", "get_session_factory", "init_engine"] +from deerflow.persistence.engine import ( + close_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", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py index e7addcc0..c50d3602 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py +++ b/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: - """Return the event loop that owns the engine's connection pool (or None).""" - return _engine_loop +def get_engine_loop() -> asyncio.AbstractEventLoop | None: + """Return the event loop that owns the engine's connection pool (or None).""" + 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: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260828_01_workflow_studio_tables.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260828_01_workflow_studio_tables.py index b3ee9e22..6b784ae9 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260828_01_workflow_studio_tables.py +++ b/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"), ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260829_02_workflow_canvas_schema.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260829_02_workflow_canvas_schema.py index a18a8ad9..0f7a1058 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260829_02_workflow_canvas_schema.py +++ b/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", diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflows/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflows/model.py index 3eb307d9..e001b59b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/workflows/model.py +++ b/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). diff --git a/offline-backend-20260512/backend/tests/test_workflow_definitions.py b/offline-backend-20260512/backend/tests/test_workflow_definitions.py index 52aa5cc9..5bb44ee3 100644 --- a/offline-backend-20260512/backend/tests/test_workflow_definitions.py +++ b/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) diff --git a/offline-backend-20260512/backend/tests/test_workflow_schema_rebuild.py b/offline-backend-20260512/backend/tests/test_workflow_schema_rebuild.py new file mode 100644 index 00000000..8019dbd0 --- /dev/null +++ b/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()