Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-server into dhk-1

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

@@ -1,8 +1,10 @@
-import asyncio
-import logging
-import os
-import sys
-from collections.abc import AsyncGenerator
+import asyncio
+import logging
+import os
+import queue
+import sys
+import threading
+from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 from dotenv import load_dotenv
@@ -245,8 +247,7 @@ async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
 
 async def _run_db_migrations() -> None:
     """Run Alembic migrations at startup so schema changes are applied automatically."""
-    import asyncio
-    from pathlib import Path
+    from pathlib import Path
 
     from alembic import command
     from alembic.config import Config
@@ -259,18 +260,51 @@ async def _run_db_migrations() -> None:
         logger.info("Persistence backend=memory — skipping Alembic migrations")
         return
 
-    db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
-    migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
-
-    def _run_sync() -> None:
-        cfg = Config()
-        cfg.set_main_option("script_location", migrations_dir)
-        cfg.set_main_option("sqlalchemy.url", db_url)
-        command.upgrade(cfg, "head")
-
-    try:
-        await asyncio.get_running_loop().run_in_executor(None, _run_sync)
-        logger.info("Database migrations applied (head)")
+    if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in {"1", "true", "yes"}:
+        logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations")
+        return
+
+    db_url = engine.url.render_as_string(hide_password=False).replace("%", "%%")
+    migrations_dir = str(Path(_persistence_pkg.__file__).parent / "migrations")
+    result_queue: queue.Queue[BaseException | None] = queue.Queue(maxsize=1)
+
+    def _run_sync() -> None:
+        try:
+            cfg = Config()
+            cfg.set_main_option("script_location", migrations_dir)
+            cfg.set_main_option("sqlalchemy.url", db_url)
+            command.upgrade(cfg, "head")
+        except BaseException as exc:
+            result_queue.put(exc)
+        else:
+            result_queue.put(None)
+
+    try:
+        try:
+            timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
+        except ValueError:
+            timeout_s = 60.0
+
+        worker = threading.Thread(target=_run_sync, name="deerflow-alembic-upgrade", daemon=True)
+        worker.start()
+        try:
+            deadline = asyncio.get_running_loop().time() + timeout_s
+            while result_queue.empty():
+                if asyncio.get_running_loop().time() >= deadline:
+                    raise asyncio.TimeoutError
+                await asyncio.sleep(0.2)
+            result = result_queue.get_nowait()
+        except asyncio.TimeoutError:
+            logger.error(
+                "Database migration did not finish within %.0fs; continuing startup on create_all schema. "
+                "Likely cause: MySQL metadata-lock wait or a slow DDL. Set DEER_FLOW_SKIP_ALEMBIC=1 "
+                "to skip startup migrations, or inspect MySQL with SHOW FULL PROCESSLIST.",
+                timeout_s,
+            )
+            return
+        if result is not None:
+            raise result
+        logger.info("Database migrations applied (head)")
     except Exception:
         logger.exception("Database migration failed — schema may be out of date")
 

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

@@ -37,9 +37,17 @@ def should_check_csrf(request: Request) -> bool:
     if request.method not in ("POST", "PUT", "DELETE", "PATCH"):
         return False
 
-    path = request.url.path.rstrip("/")
+    path = request.url.path
+    stripped = path.rstrip("/")
     # Exempt /api/v1/auth/me endpoint
-    if path == "/api/v1/auth/me":
+    if stripped == "/api/v1/auth/me":
+        return False
+    # Anonymous public endpoints (mirror auth_middleware._PUBLIC_PATH_PREFIXES).
+    # These intentionally have no cookie session, so the Double-Submit Cookie
+    # pattern is both impossible (no token to compare) and unnecessary (no
+    # session for an attacker to ride). Without this exemption every public
+    # PUT/POST returns 403 even though auth was bypassed earlier in the chain.
+    if path.startswith("/api/public/"):
         return False
     return True
 

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

@@ -1,4 +1,4 @@
-"""Public (unauthenticated) read-only API for scheduled tasks.
+"""Public (unauthenticated) API for scheduled tasks.
 
 Mounted under the ``/api/public/`` prefix so the auth middleware lets these
 routes through without a session cookie. Intended for embedding scheduled
@@ -12,9 +12,15 @@ Surface area:
 - ``GET  /api/public/scheduled-tasks/{task_id}/runs``         — runs for a task (latest first)
 - ``GET  /api/public/scheduled-tasks/runs/{run_id}``          — single run (result included)
 - ``GET  /api/public/scheduled-tasks/runs/{run_id}/files/{i}``— inline file content / preview
+- ``PUT  /api/public/scheduled-tasks/runs/{run_id}/files/{i}``— overwrite an editable result file
+- ``PUT  /api/public/scheduled-tasks/runs/{run_id}/body``     — overwrite the inline Markdown body
 - ``GET  /api/public/scheduled-tasks/runs/{run_id}/download`` — zip / md download of result
+- ``DELETE /api/public/scheduled-tasks/runs/{run_id}``        — remove a single run (no auth)
 
-These are read-only. No write/mutation routes belong here.
+The PUT endpoints intentionally skip ownership checks so the anonymous public
+viewer at ``#/public/scheduled-tasks/<task_id>`` can fix typos in result MD
+files / body content. Tighten this if the deployment requires read-only
+sharing.
 """
 
 from __future__ import annotations
@@ -30,6 +36,7 @@ from urllib.parse import quote
 
 from fastapi import APIRouter, HTTPException, Query
 from fastapi.responses import FileResponse, Response, StreamingResponse
+from pydantic import BaseModel, Field
 
 from deerflow.runtime.scheduler import get_scheduled_task_service
 
@@ -39,6 +46,17 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/public/scheduled-tasks", tags=["public-scheduled-tasks"])
 
 
+_MAX_PUBLIC_EDIT_BYTES = 5 * 1024 * 1024  # 5 MB — mirrors the authenticated edit cap
+
+
+class PublicRunFileUpdateRequest(BaseModel):
+    content: str = Field(..., description="New text content to write into the run file")
+
+
+class PublicRunBodyUpdateRequest(BaseModel):
+    content: str = Field(..., description="New Markdown body for the run's result")
+
+
 # ---------------------------------------------------------------------------
 # Sanitisation helpers
 # ---------------------------------------------------------------------------
@@ -93,6 +111,15 @@ def _redact_run(run: dict[str, Any]) -> dict[str, Any]:
                 # to fetch content via the public file endpoint.
                 public_file = {k: v for k, v in item.items() if k != "path"}
                 public_file["index"] = idx
+                # Mark whether the underlying archive is still on disk. The
+                # public viewer filters out tabs whose source is gone so users
+                # never see the raw "Result file is no longer available" 404.
+                path_value = item.get("path")
+                public_file["available"] = (
+                    isinstance(path_value, str)
+                    and bool(path_value)
+                    and Path(path_value).is_file()
+                )
                 cleaned_files.append(public_file)
             result = {**result, "files": cleaned_files}
             cleaned["result"] = result
@@ -302,3 +329,57 @@ async def public_download_run(run_id: str):
         media_type="application/zip",
         headers={"Content-Disposition": _content_disposition(f"{base_name}.zip")},
     )
+
+
+@router.put("/runs/{run_id}/files/{file_index}")
+async def public_update_run_file(
+    run_id: str,
+    file_index: int,
+    body: PublicRunFileUpdateRequest,
+) -> dict[str, Any]:
+    """Overwrite an MD/text result file on disk and refresh its size in the run row.
+
+    Anonymous — no ownership check. See module docstring for the rationale.
+    """
+    if len(body.content.encode("utf-8")) > _MAX_PUBLIC_EDIT_BYTES:
+        raise HTTPException(status_code=413, detail="Edited content exceeds the 5 MB limit")
+    service = get_scheduled_task_service()
+    try:
+        return await service.admin_update_run_file(run_id, file_index, body.content)
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc))
+    except PermissionError as exc:
+        # Surfaces archive-path escape attempts. Treat as 400 since the caller is
+        # already trusted to be reaching a real run row.
+        raise HTTPException(status_code=400, detail=str(exc))
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc))
+
+
+@router.delete("/runs/{run_id}", status_code=204)
+async def public_delete_run(run_id: str) -> Response:
+    """Remove a single run regardless of owner.
+
+    Anonymous — no ownership check, mirroring the existing public PUT endpoints.
+    Tighten this if the deployment requires read-only public sharing.
+    """
+    service = get_scheduled_task_service()
+    deleted = await service.admin_delete_run(run_id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
+    return Response(status_code=204)
+
+
+@router.put("/runs/{run_id}/body")
+async def public_update_run_body(run_id: str, body: PublicRunBodyUpdateRequest) -> dict[str, Any]:
+    """Overwrite the inline Markdown body (``result.content``) of a run.
+
+    Anonymous — no ownership check. Used by the public viewer's "正文" tab.
+    """
+    if len(body.content.encode("utf-8")) > _MAX_PUBLIC_EDIT_BYTES:
+        raise HTTPException(status_code=413, detail="Edited content exceeds the 5 MB limit")
+    service = get_scheduled_task_service()
+    try:
+        return await service.admin_update_run_body(run_id, body.content)
+    except FileNotFoundError as exc:
+        raise HTTPException(status_code=404, detail=str(exc))

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

@@ -31,6 +31,7 @@ router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
 
 class ScheduledTaskCreateRequest(BaseModel):
     name: str = Field(default="Scheduled task")
+    description: str | None = None
     prompt: str
     schedule: str
     timezone: str | None = None
@@ -40,6 +41,7 @@ class ScheduledTaskCreateRequest(BaseModel):
 
 class ScheduledTaskUpdateRequest(BaseModel):
     name: str | None = None
+    description: str | None = None
     prompt: str | None = None
     schedule: str | None = None
     timezone: str | None = None
@@ -547,6 +549,7 @@ async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Reque
         task = await service.create_task(
             user_id=user_id,
             name=body.name,
+            description=body.description,
             prompt=body.prompt,
             schedule=body.schedule,
             timezone=body.timezone,

offline-backend-20260512/backend/config.yaml

@@ -153,12 +153,11 @@ memory:
     user_char_limit: 1375
     deduplicate_on_load: true
 
-  hindsight: # provider=hindsight 时生效,需另行部署 Hindsight 服务
-    mode: local_external # 'cloud' | 'local_embedded' | 'local_external'
+  hindsight:                       # provider=hindsight 时生效,需另行部署 Hindsight 服务
+    mode: local_external           # 'cloud' | 'local_embedded' | 'local_external'
     api_url: http://47.88.25.99:18888
-    api_key:
-      "" # 部署 Hindsight 后填,或改成 $HINDSIGHT_API_KEY 并设同名环境变量
-      # 注意:写成 $VAR 但环境变量未设置会导致配置加载失败
+    api_key: ""                    # 部署 Hindsight 后填,或改成 $HINDSIGHT_API_KEY 并设同名环境变量
+                                   # 注意:写成 $VAR 但环境变量未设置会导致配置加载失败
     bank_id_template: "deerflow-user-{user_id}"
     work_bank_id_template: "deerflow-work-{user_id}-{agent_id}"
     memory_mode: context # 'context' | 'tools' | 'hybrid'

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

@@ -11,13 +11,16 @@ None and fall back to in-memory implementations.
 from __future__ import annotations
 
 import asyncio
-import json
-import logging
-import sys
-from importlib.util import find_spec
-from pathlib import Path
-
-from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
+import json
+import logging
+import os
+import sys
+from importlib.util import find_spec
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.engine import make_url
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
 
 
 def _json_serializer(obj: object) -> str:
@@ -77,6 +80,10 @@ async def _run_alembic_upgrade(url: str) -> None:
       operable on the ``create_all`` schema; the operator just won't get
       in-place fixes until they investigate.
     """
+    if os.getenv("DEER_FLOW_SKIP_ALEMBIC", "").lower() in ("1", "true", "yes"):
+        logger.warning("DEER_FLOW_SKIP_ALEMBIC is set; skipping Alembic migrations (schema stays at create_all)")
+        return
+
     ini_path = Path(__file__).parent / "migrations" / "alembic.ini"
     if not ini_path.exists():
         logger.warning("Alembic config not found at %s; skipping auto-migrate", ini_path)
@@ -102,14 +109,31 @@ async def _run_alembic_upgrade(url: str) -> None:
         # at the directory of alembic.ini itself — already correct for us.
         command.upgrade(cfg, "head")
 
+    # Bound the upgrade so a stuck DDL (e.g. MySQL metadata-lock wait) cannot
+    # hang startup forever. Tunable via DEER_FLOW_ALEMBIC_TIMEOUT (seconds).
+    # NOTE: asyncio.wait_for cancels the *await* on timeout; the worker thread
+    # itself keeps running (Python can't kill threads). That's acceptable —
+    # the orphan thread will exit when its MySQL connection eventually errors
+    # or when the process restarts. The point here is to unblock startup.
     try:
-        await asyncio.to_thread(_do_upgrade)
+        timeout_s = float(os.getenv("DEER_FLOW_ALEMBIC_TIMEOUT", "60"))
+    except ValueError:
+        timeout_s = 60.0
+
+    try:
+        await asyncio.wait_for(asyncio.to_thread(_do_upgrade), timeout=timeout_s)
         logger.info("Alembic migrations applied (or already up-to-date)")
+    except asyncio.TimeoutError:
+        logger.error(
+            "Alembic upgrade did not finish within %.0fs; continuing startup on create_all schema. "
+            "Likely cause: MySQL metadata-lock wait — check SHOW FULL PROCESSLIST for a blocking transaction.",
+            timeout_s,
+        )
     except Exception:
         logger.exception("Auto-applying Alembic migrations failed; continuing with create_all schema")
 
 
-async def _auto_create_postgres_db(url: str) -> None:
+async def _auto_create_postgres_db(url: str) -> None:
     """Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
 
     The target database name is extracted from *url*.  The connection is
@@ -132,11 +156,50 @@ async def _auto_create_postgres_db(url: str) -> None:
         async with maint_engine.connect() as conn:
             await conn.execute(text(f'CREATE DATABASE "{db_name}"'))
         logger.info("Auto-created PostgreSQL database: %s", db_name)
-    finally:
-        await maint_engine.dispose()
-
-
-async def init_engine(
+    finally:
+        await maint_engine.dispose()
+
+
+async def _auto_create_mysql_db(url: str) -> None:
+    """Create the target MySQL database when it is missing.
+
+    ``create_all`` can create tables, but MySQL requires the database/schema to
+    exist before the application URL can connect. We connect to the built-in
+    ``mysql`` maintenance database on the same server and issue a conservative
+    ``CREATE DATABASE IF NOT EXISTS``.
+    """
+
+    parsed = make_url(url)
+    db_name = parsed.database
+    if not db_name:
+        raise ValueError("Cannot auto-create MySQL database: no database name in URL")
+
+    maint_url = parsed.set(database="mysql")
+    maint_engine = create_async_engine(
+        maint_url.render_as_string(hide_password=False),
+        isolation_level="AUTOCOMMIT",
+        pool_pre_ping=True,
+    )
+    quoted = "`" + db_name.replace("`", "``") + "`"
+    try:
+        async with maint_engine.connect() as conn:
+            await conn.execute(
+                text(
+                    f"CREATE DATABASE IF NOT EXISTS {quoted} "
+                    "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
+                )
+            )
+        logger.info("Auto-created MySQL database if missing: %s", db_name)
+    finally:
+        await maint_engine.dispose()
+
+
+def _is_mysql_unknown_database_error(exc: Exception) -> bool:
+    message = str(exc).lower()
+    return "unknown database" in message or "1049" in message
+
+
+async def init_engine(
     backend: str,
     *,
     url: str = "",
@@ -239,25 +302,41 @@ async def init_engine(
         # This is expected during initial scaffolding or minimal installs.
         logger.debug("deerflow.persistence.models not found; skipping auto-create tables")
 
-    try:
-        async with _engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-    except Exception as exc:
-        if backend == "postgres" and "does not exist" in str(exc):
+    try:
+        async with _engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+    except Exception as exc:
+        if backend == "postgres" and "does not exist" in str(exc):
             # Database not yet created — attempt to auto-create it, then retry.
             await _auto_create_postgres_db(url)
             # Rebuild engine against the now-existing database
             await _engine.dispose()
             _engine = create_async_engine(url, echo=echo, pool_size=pool_size, pool_pre_ping=True, json_serializer=_json_serializer)
-            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
-            async with _engine.begin() as conn:
-                await conn.run_sync(Base.metadata.create_all)
-        else:
-            raise
+            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+            async with _engine.begin() as conn:
+                await conn.run_sync(Base.metadata.create_all)
+        elif backend == "mysql" and _is_mysql_unknown_database_error(exc):
+            # Database not yet created. Create it, rebuild the app engine
+            # against the target DB, then let SQLAlchemy create all tables.
+            await _auto_create_mysql_db(url)
+            await _engine.dispose()
+            _engine = create_async_engine(
+                url,
+                echo=echo,
+                pool_size=pool_size,
+                pool_pre_ping=True,
+                pool_recycle=1800,
+                json_serializer=_json_serializer,
+            )
+            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+            async with _engine.begin() as conn:
+                await conn.run_sync(Base.metadata.create_all)
+        else:
+            raise
 
     # Apply in-place schema fixes (column widenings, etc.) that ``create_all``
     # cannot do on its own. Failures don't abort startup — see the helper.
-    await _run_alembic_upgrade(url)
+    # await _run_alembic_upgrade(url)
 
     logger.info("Persistence engine initialized: backend=%s", backend)
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/env.py

@@ -9,9 +9,11 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import os
 from logging.config import fileConfig
 
 from alembic import context
+from sqlalchemy import text
 from sqlalchemy.ext.asyncio import create_async_engine
 
 from deerflow.persistence.base import Base
@@ -44,13 +46,27 @@ def run_migrations_offline() -> None:
         context.run_migrations()
 
 
-def do_run_migrations(connection):
-    render_as_batch = connection.dialect.name == "sqlite"
-    context.configure(
-        connection=connection,
-        target_metadata=target_metadata,
-        render_as_batch=render_as_batch,  # Required for SQLite ALTER TABLE support
-    )
+def do_run_migrations(connection):
+    render_as_batch = connection.dialect.name == "sqlite"
+
+    # On MySQL, shrink the session lock waits so a blocked DDL fails fast
+    # instead of hanging on the server default (lock_wait_timeout = 31536000s
+    # = 1 year; innodb_lock_wait_timeout = 50s). Without this, an idle
+    # transaction on another connection can freeze startup indefinitely.
+    # Tunable via DEER_FLOW_ALEMBIC_LOCK_WAIT (seconds, default 15).
+    if connection.dialect.name == "mysql":
+        try:
+            lock_wait = int(os.getenv("DEER_FLOW_ALEMBIC_LOCK_WAIT", "15"))
+        except ValueError:
+            lock_wait = 15
+        connection.execute(text(f"SET SESSION lock_wait_timeout = {lock_wait}"))
+        connection.execute(text(f"SET SESSION innodb_lock_wait_timeout = {lock_wait}"))
+
+    context.configure(
+        connection=connection,
+        target_metadata=target_metadata,
+        render_as_batch=render_as_batch,  # Required for SQLite ALTER TABLE support
+    )
     with context.begin_transaction():
         context.run_migrations()
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260521_02_merge_agent_pins_and_memory_heads.py

@@ -0,0 +1,24 @@
+"""Merge agent pin/featured and user memory config migration heads.
+
+Revision ID: 20260521_02
+Revises: 20260520_03, 20260521_01
+Create Date: 2026-05-21
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+
+revision: str = "20260521_02"
+down_revision: str | tuple[str, str] = ("20260520_03", "20260521_01")
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260522_01_scheduled_task_description.py

@@ -0,0 +1,36 @@
+"""Add description column to scheduled_tasks.
+
+Revision ID: 20260522_01
+Revises: 20260521_02
+Create Date: 2026-05-22
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260522_01"
+down_revision = "20260521_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return any(column["name"] == column_name for column in inspector.get_columns(table_name))
+
+
+def upgrade() -> None:
+    if not _has_column("scheduled_tasks", "description"):
+        op.add_column(
+            "scheduled_tasks",
+            sa.Column("description", sa.Text(), nullable=True),
+        )
+
+
+def downgrade() -> None:
+    if _has_column("scheduled_tasks", "description"):
+        op.drop_column("scheduled_tasks", "description")

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

@@ -31,6 +31,7 @@ class ScheduledTaskRow(Base):
     source_agent_name: Mapped[str | None] = mapped_column(String(128), index=True)
     execution_agent_name: Mapped[str | None] = mapped_column(String(128), index=True)
     name: Mapped[str] = mapped_column(String(127))
+    description: Mapped[str | None] = mapped_column(Text)
     prompt: Mapped[str] = mapped_column(Text)
     schedule_text: Mapped[str] = mapped_column(String(256))
     cron_expr: Mapped[str] = mapped_column(String(128))

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

@@ -76,6 +76,7 @@ class ScheduledTaskRepository(ScheduledTaskStore):
             source_agent_name=data.get("source_agent_name"),
             execution_agent_name=data.get("execution_agent_name"),
             name=data["name"],
+            description=data.get("description"),
             prompt=data["prompt"],
             schedule_text=data["schedule_text"],
             cron_expr=data["cron_expr"],

offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py

@@ -18,6 +18,7 @@ from zoneinfo import ZoneInfo
 from croniter import croniter
 from langgraph.checkpoint.base import empty_checkpoint
 
+from deerflow.config.agents_config import AGENT_ID_PATTERN
 from deerflow.config.paths import get_paths
 from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
 from deerflow.runtime import serialize_channel_values
@@ -305,12 +306,51 @@ class ScheduledTaskService:
         """
         run = await self.store.get_run(run_id, user_id)
         if run is None:
-            # Either the run doesn't exist, or the caller is a subscriber, not owner.
             candidate = await self.store.get_run_any(run_id)
             if candidate is None:
                 raise FileNotFoundError(f"Run {run_id} not found")
             raise PermissionError("Only the run owner may edit result files")
+        return await self._apply_run_file_edit(run, run_id, file_index, new_content)
 
+    async def admin_update_run_file(
+        self,
+        run_id: str,
+        file_index: int,
+        new_content: str,
+    ) -> dict[str, Any]:
+        """Owner-less variant of :meth:`update_run_file` used by the public viewer.
+
+        The public scheduled-task page is unauthenticated, so it cannot supply a
+        ``user_id``. Mutations from that page route here instead. The disk-write
+        and result-row update logic is identical to the owner path.
+        """
+        run = await self.store.get_run_any(run_id)
+        if run is None:
+            raise FileNotFoundError(f"Run {run_id} not found")
+        return await self._apply_run_file_edit(run, run_id, file_index, new_content)
+
+    async def admin_update_run_body(self, run_id: str, new_content: str) -> dict[str, Any]:
+        """Overwrite the inline Markdown body (``result.content``) of a run.
+
+        Used by the public scheduled-task viewer so anonymous editors can fix
+        typos in the "正文" tab. Unlike :meth:`admin_update_run_file`, no disk
+        artifact backs the body — it lives entirely in the run row.
+        """
+        run = await self.store.get_run_any(run_id)
+        if run is None:
+            raise FileNotFoundError(f"Run {run_id} not found")
+        result = run.get("result") if isinstance(run.get("result"), dict) else {}
+        new_result = {**result, "content": new_content}
+        await self.store.update_run(run_id, {"result": new_result})
+        return {"content": new_content}
+
+    async def _apply_run_file_edit(
+        self,
+        run: dict[str, Any],
+        run_id: str,
+        file_index: int,
+        new_content: str,
+    ) -> dict[str, Any]:
         result = run.get("result") or {}
         files = result.get("files") if isinstance(result, dict) else None
         if not isinstance(files, list) or file_index < 0 or file_index >= len(files):
@@ -406,6 +446,7 @@ class ScheduledTaskService:
         prompt: str,
         schedule: str,
         timezone: str | None = None,
+        description: str | None = None,
         source_thread_id: str | None = None,
         source_agent_name: str | None = None,
         execution_agent_name: str | None = None,
@@ -420,6 +461,7 @@ class ScheduledTaskService:
         context = dict(execution_context or {})
         if execution_agent_name:
             context["agent_name"] = execution_agent_name
+        normalized_description = description.strip() if isinstance(description, str) else None
         task = await self.store.create_task(
             {
                 "task_id": str(uuid.uuid4()),
@@ -429,6 +471,7 @@ class ScheduledTaskService:
                 "source_agent_name": source_agent_name,
                 "execution_agent_name": execution_agent_name,
                 "name": final_name,
+                "description": normalized_description or None,
                 "prompt": prompt.strip(),
                 "schedule_text": schedule.strip(),
                 "cron_expr": cron_expr,
@@ -451,6 +494,15 @@ class ScheduledTaskService:
         for key in ("name", "prompt", "enabled", "execution_agent_name"):
             if key in data and data[key] is not None:
                 update_data[key] = data[key]
+        # Description is nullable — accept ``None`` (clear) and empty strings
+        # (also clear) explicitly, so the field can be removed via PATCH.
+        if "description" in data:
+            raw = data["description"]
+            if isinstance(raw, str):
+                trimmed = raw.strip()
+                update_data["description"] = trimmed or None
+            else:
+                update_data["description"] = None
         # Names are matched against the per-user uniqueness constraint, so keep
         # the stored value trimmed to match create_task's normalization.
         if isinstance(update_data.get("name"), str):
@@ -819,8 +871,17 @@ class ScheduledTaskService:
         try:
             request = SimpleNamespace(app=self.app)
             execution_context = dict(task.get("execution_context") or {})
-            if task.get("execution_agent_name"):
-                execution_context["agent_name"] = task["execution_agent_name"]
+            stored_agent_ref = task.get("execution_agent_name")
+            if stored_agent_ref:
+                resolved_agent_id = await self._resolve_agent_id(stored_agent_ref, task["user_id"])
+                if resolved_agent_id:
+                    execution_context["agent_name"] = resolved_agent_id
+                else:
+                    logger.warning(
+                        "Scheduled task %s references unknown agent %r; falling back to default lead agent",
+                        task.get("task_id"),
+                        stored_agent_ref,
+                    )
             execution_context.update(
                 {
                     "is_scheduled_run": True,
@@ -845,6 +906,35 @@ class ScheduledTaskService:
         finally:
             reset_current_user(token)
 
+    async def _resolve_agent_id(self, stored_ref: str, user_id: str) -> str | None:
+        """Resolve a scheduled task's ``execution_agent_name`` to a stable agent id.
+
+        Older tasks (created before the frontend was fixed) stored the agent's
+        Chinese **display name** here; the runtime validator only accepts ids
+        matching ``^[A-Za-z0-9_-]+$``, so those rows would 422 forever. We
+        gracefully recover by looking the stored value up first as an id, then
+        as a display name. Returns ``None`` when nothing visible matches — the
+        caller then falls back to the default lead agent rather than crashing.
+        """
+        store = getattr(self.app.state, "agent_store", None)
+        if store is None:
+            return stored_ref if AGENT_ID_PATTERN.fullmatch(stored_ref) else None
+        if AGENT_ID_PATTERN.fullmatch(stored_ref):
+            try:
+                row = await store.get_visible(stored_ref, user_id)
+            except Exception:
+                row = None
+            if row is not None:
+                return row["id"]
+        try:
+            agents = await store.list_visible(user_id)
+        except Exception:
+            return None
+        for agent in agents:
+            if agent.get("name") == stored_ref:
+                return agent.get("id")
+        return None
+
     async def _extract_latest_ai_text(self, thread_id: str) -> str | None:
         checkpointer = self.app.state.checkpointer
         config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}

offline-backend-20260512/config.yaml

@@ -148,7 +148,7 @@ memory:
 
   hindsight:                       # provider=hindsight 时生效,需另行部署 Hindsight 服务
     mode: local_external           # 'cloud' | 'local_embedded' | 'local_external'
-    api_url: http://localhost:8765
+    api_url: http://47.88.25.99:18888
     api_key: ""                    # 部署 Hindsight 后填,或改成 $HINDSIGHT_API_KEY 并设同名环境变量
                                    # 注意:写成 $VAR 但环境变量未设置会导致配置加载失败
     bank_id_template: "deerflow-user-{user_id}"