diff --git a/frontend-web/src/styles/globals.css b/frontend-web/src/styles/globals.css
index c045a0bb..cc03e47f 100644
--- a/frontend-web/src/styles/globals.css
+++ b/frontend-web/src/styles/globals.css
@@ -870,3 +870,16 @@
body {
font-size: calc(1rem * var(--ui-font-scale, 1));
}
+
+/* Strategy 顶栏:浅色 / 深色背景图随 html.dark 切换 */
+.strategy-app-header {
+ background-color: transparent;
+ background-image: url("../assets/img/red-header.png");
+ background-size: 100% 100%;
+ background-repeat: no-repeat;
+ background-position: center;
+}
+
+html.dark .strategy-app-header {
+ background-image: url("../assets/img/dark-blue-header.svg");
+}
diff --git a/offline-backend-20260512/backend/app/gateway/authz.py b/offline-backend-20260512/backend/app/gateway/authz.py
index 89fd943c..0d8d2899 100644
--- a/offline-backend-20260512/backend/app/gateway/authz.py
+++ b/offline-backend-20260512/backend/app/gateway/authz.py
@@ -31,21 +31,43 @@ from __future__ import annotations
import functools
import inspect
-
-from app.gateway.auth.disabled_mode import get_assumed_user_for_disabled_auth, is_auth_disabled
+import logging
from collections.abc import Callable
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar
from fastapi import HTTPException, Request
+from app.gateway.auth.disabled_mode import get_assumed_user_for_disabled_auth, is_auth_disabled
+
if TYPE_CHECKING:
from app.gateway.auth.models import User
+logger = logging.getLogger(__name__)
+
P = ParamSpec("P")
T = TypeVar("T")
+async def _thread_checkpoint_exists(request: Request, thread_id: str) -> bool:
+ """Return True if a LangGraph checkpoint exists for ``thread_id``.
+
+ A checkpoint with no matching ``threads_meta`` row is the signature of a
+ DB-switch orphan (sqlite↔mysql/postgres juggling): the conversation state
+ survives in the checkpointer while the business row is gone. Normal thread
+ deletion removes the checkpoint too, so this can never resurrect a
+ deliberately-deleted thread.
+ """
+ try:
+ from app.gateway.deps import get_checkpointer
+
+ checkpointer = get_checkpointer(request)
+ tup = await checkpointer.aget_tuple({"configurable": {"thread_id": thread_id}})
+ return tup is not None
+ except Exception:
+ return False
+
+
# Permission constants
class Permissions:
"""Permission constants for resource:action format."""
@@ -205,6 +227,7 @@ def require_permission(
action: str,
owner_check: bool = False,
require_existing: bool = False,
+ claim_orphan: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]:
"""Decorator that checks permission for resource:action.
@@ -221,6 +244,15 @@ def require_permission(
**destructive / mutating** routes (DELETE, PATCH,
state-update) so a deleted thread can't be re-targeted
by another user via the missing-row code path.
+ claim_orphan: Only meaningful with ``owner_check=True`` and
+ ``require_existing=True``. When the ``threads_meta``
+ row is missing **and** a LangGraph checkpoint exists
+ for the thread (a DB-switch orphan), adopt it for the
+ current user instead of 404ing. Use on **run-create**
+ routes so a conversation whose business row was lost
+ (sqlite↔mysql swap) stays answerable. Safe: normal
+ deletion removes the checkpoint, so this never revives
+ a deliberately-deleted thread.
Usage:
# Read-style: legacy untracked threads are allowed
@@ -289,16 +321,44 @@ def require_permission(
from app.gateway.deps import get_thread_store
thread_store = get_thread_store(request)
+ user_id = str(auth.user.id)
allowed = await thread_store.check_access(
thread_id,
- str(auth.user.id),
+ user_id,
require_existing=require_existing,
)
if not allowed:
- raise HTTPException(
- status_code=404,
- detail=f"Thread {thread_id} not found",
- )
+ # Tell apart three denial causes so the client gets an
+ # accurate message and so DB-switch orphans can be adopted
+ # rather than hard-404'd:
+ # 1. row exists but owned by another user -> not accessible
+ # 2. no row, no checkpoint -> truly missing
+ # 3. no row + checkpoint + claim_orphan -> orphan, claim
+ existing = await thread_store.get(thread_id, user_id=None)
+ if existing is not None:
+ # Distinct detail string lets the frontend show "无权访问"
+ # instead of "不存在" for a cross-user / stale-session hit.
+ raise HTTPException(
+ status_code=404,
+ detail=f"Thread {thread_id} not accessible",
+ )
+ claimed = False
+ if claim_orphan and await _thread_checkpoint_exists(request, thread_id):
+ try:
+ await thread_store.create(thread_id, user_id=user_id)
+ logger.info(
+ "Claimed orphan thread %s for user %s (checkpoint present, no meta row)",
+ thread_id,
+ user_id,
+ )
+ claimed = True
+ except Exception:
+ logger.warning("Failed to claim orphan thread %s", thread_id, exc_info=True)
+ if not claimed:
+ raise HTTPException(
+ status_code=404,
+ detail=f"Thread {thread_id} not found",
+ )
return await func(*args, **kwargs)
diff --git a/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py b/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py
index 9526e80d..d1eea26a 100644
--- a/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py
+++ b/offline-backend-20260512/backend/app/gateway/roundtable_run_policy.py
@@ -38,11 +38,14 @@
``docs/roundtable-dag-orchestration.md`` §4 / §12.1 / §12.8 / §12.10。
与 ``seat`` 的差异:``seat`` 禁 ``web_search``(旧「圆桌一律禁搜索」),
``seat_parallel`` **放开搜索**(本期诉求);``seat`` 已放开 ``bash``,本角色仍禁。
+ - ``summary`` —— Step3「总结报告」:写 Markdown 总结报告 + 支持问答/增量改。
+ 工具面与 ``report`` 完全一致(继承 ``seat`` 的 bash 放开 + 额外禁 ``str_replace`` /
+ 读前序席位),单列一个角色名便于以后独立调优。
- ``report`` —— Step3 报告绘制:在 ``seat`` 基础上**额外禁 ``str_replace``**:
* 继承 ``seat`` 的 bash 放开(2026-06)—— Step3 也可跑 bash 做辅助查询,同
``seat`` 警示:**结果必须 write_file 落盘**,否则前端虚拟预览会空白
(预览只抓 ``write_file.args.content``,抓不到脚本产物 → 预览停在半截
- ``write_file``,正是「没写完就去生成 py 文件」的现象)。
+ ``write_file``,正是「没写完就去生成 py 文件」的现象);
* 禁 ``str_replace`` → 强制返工用整文件 ``write_file`` 重写
(``str_replace`` 没有 ``content`` 字段,会让前端虚拟预览读到 undefined
而清空沙箱)。
@@ -112,6 +115,10 @@ _SEAT_PARALLEL_EXCLUDED: list[str] = [
# report:在 seat 基础上额外禁 str_replace(bash 已在 seat 禁)——强制整文件 write_file 重写,
# 让前端虚拟预览能拿到完整 args.content。report 是 Step3 单独绘制,不读前序席位 → 禁 read_peer_delivery。
_REPORT_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "str_replace", "read_peer_delivery"]
+# summary:Step3「总结报告」智能体(写 Markdown 报告 + 支持问答/增量改)。工具面与 report
+# 完全一致——可写文件(write_file 整文件重写),禁 str_replace(整文件重写才能让右侧 md 预览
+# 实时刷新)/bash/联网/派活/呈现文件/看图/读前序席位。单列出来便于以后独立调优。
+_SUMMARY_EXCLUDED: list[str] = [*_SEAT_EXCLUDED, "str_replace", "read_peer_delivery"]
# leader 内禀的 skill_stop:``agent_orchestration`` 调用后停图。调用方可在此基础上
# 再叠加自己的 skill_stop(见 ``merge_skill_stop_names``)。
@@ -167,7 +174,16 @@ def roundtable_run_policy(role: str) -> RoundtableRunPolicy:
thinking_enabled=True,
reasoning_effort="medium",
)
- raise ValueError(f"unknown roundtable role: {role!r} (expected leader/seat/seat_parallel/report)")
+ if role == "summary":
+ return RoundtableRunPolicy(
+ excluded_tools=list(_SUMMARY_EXCLUDED),
+ skill_stop_names=[],
+ thinking_enabled=True,
+ reasoning_effort="medium",
+ )
+ raise ValueError(
+ f"unknown roundtable role: {role!r} (expected leader/seat/seat_parallel/report/summary)"
+ )
def merge_skill_stop_names(policy_names: list[str], caller_names: list[str] | None) -> list[str]:
diff --git a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py
index 07e4dba7..43f9eb21 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed.py
@@ -63,18 +63,24 @@ COORDINATOR_AGENT_ID = "roundtable-coordinator"
# Step 3「结果绘制」用的方案可视化总结智能体:读各席位交付 + 共识,画自包含图表 HTML。
# 与上面三个一样是内置单例,种子化一次常驻磁盘+DB,绝不每次调用重建。
REPORT_AGENT_ID = "roundtable-report"
+# Step 3「总结报告」用的方案总结报告智能体:读各席位交付 + 共识,撰写一份 Markdown
+# 总结报告(写入 outputs/方案总结报告.md),并按业务链总控编排提示组织。与 report 并存
+# 的另一内置单例(report 画 HTML 看板,summary 写 Markdown 报告 + 支持问答/增量改)。
+SUMMARY_AGENT_ID = "roundtable-summary"
# 验证版 SOUL.md + config.yaml 的源码资源目录(随 git 发布)。每个子目录名即 agent id,
# 内含该 agent 的 config.yaml 与 SOUL.md,是 seeder 落盘时字节级复制的来源。
_ASSETS_DIR = Path(__file__).parent / "_roundtable_seed_assets"
# 这几个功能型/内置 agent 任何部署都必须存在;顺序无关紧要(逐个独立补建)。
-# 前三个支撑 Step 1 澄清 / 推荐 / Step 2 总控;roundtable-report 支撑 Step 3 结果绘制。
+# 前三个支撑 Step 1 澄清 / 推荐 / Step 2 总控;roundtable-report 支撑 Step 3 结果绘制
+# (HTML 看板);roundtable-summary 支撑 Step 3 总结报告(Markdown + 问答/增量改)。
_FUNCTIONAL_AGENT_IDS: tuple[str, ...] = (
INTENT_AGENT_ID,
RECOMMENDER_AGENT_ID,
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
+ SUMMARY_AGENT_ID,
)
# seeder 会复制资源目录里的这些文件(存在即复制,缺失则跳过)。
diff --git a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md
new file mode 100644
index 00000000..734d36f7
--- /dev/null
+++ b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/SOUL.md
@@ -0,0 +1,73 @@
+# 圆桌方案总结报告智能体
+
+## 核心定位
+你是多智能体圆桌研讨 **Step 3** 的「方案总结报告智能体」。各席位(情报收集 / 环境评估 / 方案设计 / 风险审查 / 执行规划 / 总结陈述等)已经完成研讨并产出交付,总控也给出了最终共识。
+
+你的**职责**:把这些文字成果归纳、提炼,**撰写成一份条理清晰、以文字叙述为主的 Markdown 总结报告**。
+
+这份报告是给**用户阅读**的——目的是让用户能清晰、轻松地看懂「这群智能体讨论出了什么方案、为什么是它」。所以**可读性第一**:用通顺的中文把研讨脉络和方案讲清楚。
+
+你**不**调研、**不**联网搜索、**不**派活、**不**绘制 HTML 页面 / 不画图表——只做归纳与撰写。
+
+## 输入说明
+用户消息里会给你这些原料(可能有缺省):
+1. **任务意图摘要**:目标、约束、关键假设(来自 Step 1);
+2. **各席位最终交付**:每个席位的名称 + 它产出的完整 markdown 正文;
+3. **总控最终共识结论**;
+4. **(可能有)业务链总控编排提示**:用户为这条业务链写给总控(协调)智能体的一段指引。它代表用户对这条业务链产出的期望——**这是最高优先级的口径要求**,请把它作为本次总结的**重点方向、结构与风格的依据**,严格按它强调的关注点来组织你的报告。
+
+请通读全部原料:提炼每个席位的核心观点 / 交付,串成一份完整的方案总结。
+
+## 产出要求(最重要)
+
+你每一轮都要做**两件事**,缺一不可:
+
+### A. 写一个 Markdown 文件
+- 用 `write_file` **一次性**把完整报告写入 `/mnt/user-data/outputs/方案总结报告.md`;
+- **就一个 `.md` 文件**——不要拆成多个文件、不要分多次拼接;
+- **纯 Markdown 文本**(标题、列表、表格、引用、加粗等),**不**输出 HTML、**不**画图表(SVG / Chart 等一律不要)、**不**引任何外链;
+- **严禁**写 Python / shell / 任何脚本去"生成"文件,也**严禁**用 bash——你必须自己把 Markdown 文本直接写进 `write_file`;
+- `write_file` **支持很长的内容,不会截断**——请放心一次性把整篇报告写进去。
+
+### B. 在对话回复里给出报告正文
+- 写完文件后,**把这份报告的完整正文也输出到对话回复里**(直接给用户读),不要只回一句「已写入文件」。
+- 用户既能在对话里直接读到报告,也能在右侧预览 / 下载这个 `.md` 文件——两边内容应当一致。
+
+## 报告结构(叙事型,文字为主线)
+按下面的章节顺序组织,缺料的章节可省略或合并。**若有「业务链总控编排提示」,以它的要求优先**:
+
+| # | 章节 | 内容 |
+|---|------|------|
+| ① | 标题 + 一句话结论 | 用一句话点明最终方案 / 结论 |
+| ② | 任务背景与目标 | 交代要解决什么、目标与关键约束 |
+| ③ | 各席位观点与交付 | **每个席位一段提炼文字**(这位专家关注什么、给出了什么关键结论 / 建议) |
+| ④ | 综合分析 | 把跨席位的信息串起来分析(风险全景、资源权衡、阶段安排等),可用 Markdown 表格 / 列表结构化呈现 |
+| ⑤ | 总控最终共识方案 | **重点章节**,详写最终拍板的方案是什么、核心要点、为什么这么定 |
+| ⑥ | 执行路线 | 落地步骤 / 阶段说明 |
+
+善用 Markdown 表格、有序 / 无序列表,把要点、阶段、里程碑、风险项、优劣势等结构化呈现,提升可读性。
+
+## 多轮交互(用户会在输入框里继续和你对话)
+用户读完报告后,可能在输入框里继续给你发消息。你要**自己判断**意图,分两种情况处理:
+
+1. **提出修改意见**(如「把执行路线写得更细」「补一节预算分析」「风险部分太长,精简」):
+ - 在**原报告基础上做增量修改**——只改用户提到的地方,其余内容保持不变;
+ - 用 `write_file` **整文件重写**到同一路径 `/mnt/user-data/outputs/方案总结报告.md`(`str_replace` 已被禁用,整文件重写才能让右侧预览实时刷新);
+ - 然后在对话回复里给出修改后的完整报告正文。
+
+2. **就报告 / 研讨内容提问**(如「为什么没采用方案 B?」「风险最高的是哪一项?」):
+ - **只在对话里用文字回答**,依据已有的研讨原料和报告内容作答;
+ - **不要**改写报告文件、**不要**重新 `write_file`。
+
+判断不准时,倾向于「先简短回答,再问用户是否需要据此更新报告」。
+
+## 禁止事项
+- ❌ 生成 HTML / 绘制可视化页面 / 画图表(只写 Markdown 文本)
+- ❌ 把报告拆成多文件、分多次拼接
+- ❌ 联网搜索、派活给其他智能体
+- ❌ 写 Python / shell / 任何脚本去生成文件,或用 bash 跑脚本产出文件
+- ❌ 写完文件后只回一句「已写入文件」而不给出报告正文(用户要在对话里读到报告)
+- ❌ 用户只是提问时还去改写报告文件
+- ❌ 在报告任何位置添加「AI 自动生成」「AI generated」之类的标签、徽章或水印
+- ❌ 编造与原料明显矛盾的内容
+- ❌ 忽视「业务链总控编排提示」的口径要求(它优先级最高)
diff --git a/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/config.yaml b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/config.yaml
new file mode 100644
index 00000000..d9a32803
--- /dev/null
+++ b/offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-summary/config.yaml
@@ -0,0 +1,3 @@
+description: 圆桌方案总结报告智能体,读取各席位的最终交付与总控共识,撰写一份以文字叙述为主的 Markdown 总结报告(写入 outputs/方案总结报告.md),同时把报告正文输出到对话供用户阅读,并支持用户在输入框里就报告提问或提出修改意见。它会遵循业务链写给总控的编排提示来组织报告。本身不调研、不联网、不派活、不绘制 HTML / 图表。
+id: roundtable-summary
+name: 方案总结报告
diff --git a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
index cf3a1590..a26f5742 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/multi_agent.py
@@ -66,6 +66,7 @@ from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable
from app.gateway.routers._roundtable_seed import (
COORDINATOR_AGENT_ID,
REPORT_AGENT_ID,
+ SUMMARY_AGENT_ID,
ensure_roundtable_functional_agents,
)
from deerflow.config import get_app_config
@@ -915,6 +916,26 @@ async def init_report_thread(request: Request) -> dict[str, str]:
return {"agent_name": REPORT_AGENT_ID, "thread_id": thread_id}
+@router.post("/summary/init")
+async def init_summary_thread(request: Request) -> dict[str, str]:
+ """为 Step 3「总结报告」创建一个供方案总结报告智能体(``roundtable-summary``)
+ 运行的线程,返回 ``{"agent_name", "thread_id"}``。
+
+ 与 ``/report/init`` 同构,只是换成另一个内置单例 ``roundtable-summary``:
+ - ``roundtable-summary`` 是**内置单例**(``_roundtable_seed`` 种子化,启动即同步进
+ DB),这里**只建一个轻量 thread**,绝不重建 agent;
+ - 线程打 ``_ROUNDTABLE_THREAD_METADATA`` 标记成系统会话,从「最近对话」里隐藏;
+ - 入口先 ``ensure_roundtable_functional_agents()`` 自愈(磁盘目录被手删也能补建);
+ - 建好后前端用现有 ``POST /run/stream``(``agent_type="special"``、
+ ``agent_name="roundtable-summary"``、``d_agent_thread_id={SUMMARY_AGENT_ID: thread_id}``)
+ 跑这一轮(后续问答/增量改**复用同一线程**,不重新 init),智能体把 Markdown 报告
+ 写进该线程的 ``outputs/方案总结报告.md``,再由沙箱预览。
+ """
+ ensure_roundtable_functional_agents()
+ thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+ return {"agent_name": SUMMARY_AGENT_ID, "thread_id": thread_id}
+
+
# ---------------------------------------------------------------------------
# 上游 LangGraph SSE 透传(intent / recommend / leader / special 共用)
# ---------------------------------------------------------------------------
@@ -1572,6 +1593,8 @@ async def _special_run(
# 后台进程内网关用同一份 policy,保证「后台挂起」与「前端手动调用」传参一致。
if agent_name == REPORT_AGENT_ID:
role = "report"
+ elif agent_name == SUMMARY_AGENT_ID:
+ role = "summary"
elif no_file:
role = "seat_parallel"
else:
diff --git a/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py b/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py
index 38129fa5..1f108dcc 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/thread_runs.py
@@ -21,7 +21,7 @@ from pydantic import BaseModel, Field
from app.gateway.authz import require_permission
from app.gateway.deps import get_checkpointer, get_current_user, get_feedback_repo, get_run_event_store, get_run_manager, get_run_store, get_stream_bridge
-from app.gateway.services import sse_consumer, start_run
+from app.gateway.services import immediate_end_sse, sse_consumer, start_run
from deerflow.runtime import RunRecord, serialize_channel_values
logger = logging.getLogger(__name__)
@@ -96,7 +96,7 @@ def _record_to_response(record: RunRecord) -> RunResponse:
@router.post("/{thread_id}/runs", response_model=RunResponse)
-@require_permission("runs", "create", owner_check=True, require_existing=True)
+@require_permission("runs", "create", owner_check=True, require_existing=True, claim_orphan=True)
async def create_run(thread_id: str, body: RunCreateRequest, request: Request) -> RunResponse:
"""Create a background run (returns immediately)."""
record = await start_run(body, thread_id, request)
@@ -104,7 +104,7 @@ async def create_run(thread_id: str, body: RunCreateRequest, request: Request) -
@router.post("/{thread_id}/runs/stream")
-@require_permission("runs", "create", owner_check=True, require_existing=True)
+@require_permission("runs", "create", owner_check=True, require_existing=True, claim_orphan=True)
async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) -> StreamingResponse:
"""Create a run and stream events via SSE.
@@ -132,7 +132,7 @@ async def stream_run(thread_id: str, body: RunCreateRequest, request: Request) -
@router.post("/{thread_id}/runs/wait", response_model=dict)
-@require_permission("runs", "create", owner_check=True, require_existing=True)
+@require_permission("runs", "create", owner_check=True, require_existing=True, claim_orphan=True)
async def wait_run(thread_id: str, body: RunCreateRequest, request: Request) -> dict:
"""Create a run and block until it completes, returning the final state."""
record = await start_run(body, thread_id, request)
@@ -173,7 +173,11 @@ async def get_run(thread_id: str, run_id: str, request: Request) -> RunResponse:
run_mgr = get_run_manager(request)
record = run_mgr.get(run_id)
if record is None or record.thread_id != thread_id:
- raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
+ # In-memory miss: fall back to the persistent store so a run still
+ # resolves after a Gateway restart instead of 404ing.
+ record = await run_mgr.get_persisted(run_id)
+ if record is None or record.thread_id != thread_id:
+ raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
return _record_to_response(record)
@@ -222,11 +226,27 @@ async def join_run(thread_id: str, run_id: str, request: Request) -> StreamingRe
bridge = get_stream_bridge(request)
run_mgr = get_run_manager(request)
record = run_mgr.get(run_id)
- if record is None or record.thread_id != thread_id:
+ if record is not None and record.thread_id == thread_id:
+ return StreamingResponse(
+ sse_consumer(bridge, record, request, run_mgr),
+ media_type="text/event-stream",
+ headers={
+ "Cache-Control": "no-cache",
+ "Connection": "keep-alive",
+ "X-Accel-Buffering": "no",
+ },
+ )
+
+ # Not in memory: a run that still exists in the store but has no live task
+ # means the Gateway restarted (or the finished run was cleaned up). There
+ # is nothing to reconnect to, so stream a clean terminal end instead of a
+ # 404 — the SDK then clears its reconnect key and renders history.
+ persisted = await run_mgr.get_persisted(run_id)
+ if persisted is None or persisted.thread_id != thread_id:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
return StreamingResponse(
- sse_consumer(bridge, record, request, run_mgr),
+ immediate_end_sse(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
@@ -255,7 +275,22 @@ async def stream_existing_run(
run_mgr = get_run_manager(request)
record = run_mgr.get(run_id)
if record is None or record.thread_id != thread_id:
- raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
+ # In-memory miss. If the run survives in the store it's a dead record
+ # from before a restart (or already finished) — nothing to join or
+ # cancel, so emit a clean terminal end so the SDK stops reconnecting
+ # instead of 404ing. A truly unknown run still 404s.
+ persisted = await run_mgr.get_persisted(run_id)
+ if persisted is None or persisted.thread_id != thread_id:
+ raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
+ return StreamingResponse(
+ immediate_end_sse(),
+ media_type="text/event-stream",
+ headers={
+ "Cache-Control": "no-cache",
+ "Connection": "keep-alive",
+ "X-Accel-Buffering": "no",
+ },
+ )
# Cancel if an action was requested (stop-button / interrupt flow)
if action is not None:
diff --git a/offline-backend-20260512/backend/app/gateway/services.py b/offline-backend-20260512/backend/app/gateway/services.py
index c5c2baec..359dba8b 100644
--- a/offline-backend-20260512/backend/app/gateway/services.py
+++ b/offline-backend-20260512/backend/app/gateway/services.py
@@ -42,6 +42,20 @@ logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
+async def immediate_end_sse():
+ """Yield a single terminal ``end`` SSE frame and stop.
+
+ Used when a run exists in the persistent store but not in the in-memory
+ registry — typically after a Gateway restart, or once a finished run's
+ record has been cleaned up. There is no live stream to join, so instead of
+ returning 404 (which the SDK's ``joinStream`` surfaces as a hard error and
+ never clears its reconnect key for) we stream a clean end. The SDK treats
+ this as success, clears ``lg:stream:{threadId}``, and the conversation
+ falls back to history rendering — killing the repeating switch-back 404.
+ """
+ yield format_sse("end", None)
+
+
def format_sse(event: str, data: Any, *, event_id: str | None = None) -> str:
"""Format a single SSE frame.
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py
index 9bd39c06..64deed00 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py
@@ -112,6 +112,52 @@ class RunManager:
"""Return a run record by ID, or ``None``."""
return self._runs.get(run_id)
+ def _record_from_store(self, data: dict) -> RunRecord:
+ """Rebuild a :class:`RunRecord` from a persisted store row.
+
+ The reconstructed record has no live ``task``/``abort_event`` wired to a
+ running coroutine — it represents *history*, not an in-flight run.
+ """
+ try:
+ status = RunStatus(data.get("status") or RunStatus.success.value)
+ except ValueError:
+ status = RunStatus.success
+ return RunRecord(
+ run_id=data["run_id"],
+ thread_id=data["thread_id"],
+ assistant_id=data.get("assistant_id"),
+ status=status,
+ # Disconnect mode is irrelevant for a store-only record (no task to
+ # cancel); ``continue_`` keeps sse_consumer's finally block a no-op.
+ on_disconnect=DisconnectMode.continue_,
+ multitask_strategy=data.get("multitask_strategy", "reject"),
+ metadata=data.get("metadata") or {},
+ kwargs=data.get("kwargs") or {},
+ created_at=data.get("created_at") or "",
+ updated_at=data.get("updated_at") or "",
+ error=data.get("error"),
+ )
+
+ async def get_persisted(self, run_id: str) -> RunRecord | None:
+ """Load a run from the backing store (process-restart fallback).
+
+ The in-memory registry is wiped on every Gateway restart, but the
+ ``RunStore`` row survives. When :meth:`get` misses, callers fall back
+ here so a stale ``joinStream``/``GET run`` can report the run's final
+ status instead of 404ing. Returns ``None`` when there is no store, no
+ matching row, or the lookup fails (e.g. no user context).
+ """
+ if self._store is None:
+ return None
+ try:
+ data = await self._store.get(run_id)
+ except Exception:
+ logger.warning("Failed to load run %s from store", run_id, exc_info=True)
+ return None
+ if not data:
+ return None
+ return self._record_from_store(data)
+
async def list_by_thread(self, thread_id: str) -> list[RunRecord]:
"""Return all runs for a given thread, newest first."""
async with self._lock: