"""圆桌真网关（Phase 3b）—— 进程内驱动真实 leader / seat / report agent run。 实现 ``RoundtableGateway`` 协议（定义在 harness 的 roundtable_orchestrator.types）， 但**进程内**跑真模型：复用 ``services.start_run``（scheduled task 同款）驱动一轮 agent， 跑完从 checkpointer 读最终 messages，用 ``multi_agent`` 的纯解析函数解析派活 / 澄清。 不走 HTTP、不需要 Request（用 ``SimpleNamespace(app=app, state=...)`` 当虚拟 request， ``set_current_user`` 由执行器设好）。 本网关位于 app 层（依赖 app.state + multi_agent helpers），实现 harness 定义的协议—— 不违反 harness→app 边界（harness 永不 import app；app import harness 合法）。 ⚠️ 需在能起完整后端栈 + 真模型的环境验证（见 scripts/probe_roundtable_job.py）。 """ from __future__ import annotations import asyncio import json import logging import os import uuid from datetime import UTC, datetime from types import SimpleNamespace from typing import Any from langgraph.checkpoint.base import empty_checkpoint from app.gateway.roundtable_run_policy import roundtable_run_policy from app.gateway.routers._roundtable_seed import ( COORDINATOR_AGENT_ID, REPORT_AGENT_ID, SUMMARY_AGENT_ID, ensure_roundtable_functional_agents, ) from deerflow.agents.roundtable_orchestrator.types import ( LeaderTurn, ReportResult, SeatRef, SeatTurn, SummaryResult, ) from deerflow.config.paths import get_paths from deerflow.runtime.serialization import serialize_channel_values logger = logging.getLogger(__name__) _ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True} _REPORT_FILE = "方案总览.html" _SUMMARY_FILE = "方案总结报告.md" # 各角色的工具裁剪 / thinking / reasoning 全部取自单一数据源 roundtable_run_policy # （与 multi_agent 的 _leader_run / _special_run 同源），不再在本文件硬编码——历史上 # 这里漏禁了 web_search，导致后台席位 / 报告能联网搜索而前端不能，现由 policy 统一。 # ── 后台作业「按用户限流」（2026-06-16）───────────────────────────────────────── # 痛点：进 2+ 个后台挂起后，整个网关变慢——建立智能体连接、等待模型响应都慢。 # 根因：① checkpointer 用单连接 aiosqlite（AsyncSqliteSaver，内部一把 asyncio.Lock 串 # 行全进程的 checkpoint 读写）；② 后台作业无并发上限——每个作业进程内驱动真实 leader/ # seat/report run，DAG/recommend 的 stage 还用 asyncio.gather 把席位真并行铺开。多个后台 # 作业同时把几十个 agent run 压到同一把 SQLite 锁 + 同一个离线模型端点上，前台实时会商 # 的 checkpoint 读写 / 模型调用只能排队 → 「干啥都慢」。 # # 解法：给**每个用户**一个并发额度（默认 2），后台每个 agent 轮次（leader/seat/report/ # summary/dashboard）真正开跑前先取一个额度，跑完释放。同一用户的所有后台作业共享这 2 个 # 额度——随便挂多少个作业都能立刻建好，但同一瞬间真正在打模型/写 checkpoint 的后台轮次 # 至多 2 个，不会把共享资源打满。**前台实时路径（multi_agent._leader_run/_special_run、 # 普通聊天）不经过本文件的 `_run_agent_turn`，永不取这个额度，故前台不受限。** _BACKGROUND_SLOTS_PER_USER = max(1, int(os.getenv("ROUNDTABLE_BACKGROUND_SLOTS_PER_USER", "2") or "2")) # 按 user_id 缓存 Semaphore。网关全程跑在同一个事件循环上，普通 dict 取/建即可（无跨线程）。 _user_background_sems: dict[str, asyncio.Semaphore] = {} def _user_background_semaphore(user_id: str) -> asyncio.Semaphore: """取该用户的后台并发额度信号量（懒建，每用户一把）。""" sem = _user_background_sems.get(user_id) if sem is None: sem = asyncio.Semaphore(_BACKGROUND_SLOTS_PER_USER) _user_background_sems[user_id] = sem return sem # ── 进程内底层操作 ─────────────────────────────────────────────────────────── async def _create_thread(app) -> str: """同进程建一个带圆桌 metadata 的 thread + 空 checkpoint（镜像 _create_thread_direct）。""" checkpointer = app.state.checkpointer thread_store = app.state.thread_store thread_id = str(uuid.uuid4()) await thread_store.create(thread_id, assistant_id=None, metadata=dict(_ROUNDTABLE_THREAD_METADATA)) config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} await checkpointer.aput( config, empty_checkpoint(), {"step": -1, "source": "input", "writes": None, "parents": {}, "created_at": datetime.now(UTC).isoformat()}, {}, ) return thread_id async def _read_thread_messages(app, thread_id: str) -> list[dict[str, Any]]: """读 thread 最新 checkpoint 的 messages（已 serialize 成 dict 列表）。""" checkpointer = app.state.checkpointer config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} ct = await checkpointer.aget_tuple(config) if ct is None: return [] checkpoint = getattr(ct, "checkpoint", {}) or {} values = serialize_channel_values(checkpoint.get("channel_values", {}) or {}) return values.get("messages") or [] def _latest_ai_message(messages: list[dict[str, Any]]) -> dict[str, Any] | None: for m in reversed(messages): if not isinstance(m, dict): continue if (m.get("type") or m.get("role")) in {"ai", "assistant"}: return m return None def _message_text(message: dict[str, Any] | None) -> str: if not message: return "" content = message.get("content") if isinstance(content, str): return content if isinstance(content, list): parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"] return "\n".join(parts) return "" def _latest_ai_text(messages: list[dict[str, Any]]) -> str: """取**最后一条有正文**的 AI 消息文本。 席位常以「调一堆工具」的 AIMessage 收尾（正文为空、只有 tool_calls，甚至 dangling）， 直接取最后一条 AI 消息会拿到空串 → 报告就拿不到席位产物。这里向前找到真正有文字的那条。 """ for m in reversed(messages): if not isinstance(m, dict): continue if (m.get("type") or m.get("role")) not in {"ai", "assistant"}: continue text = _message_text(m) if text.strip(): return text return "" def _normalize_tool_calls(tool_calls: Any) -> list[dict[str, Any]]: """把 checkpoint 里的 tool_calls 归一化成 {name, args, id}（丢掉无名项）。""" out: list[dict[str, Any]] = [] for tc in tool_calls or []: if not isinstance(tc, dict): continue name = tc.get("name") or "" if not name: continue out.append({"name": name, "args": tc.get("args") or {}, "id": tc.get("id") or ""}) return out def _tool_calls_since_last_human(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: """收集「最后一条 human 消息之后」各 AI 消息的 tool_calls（即**本轮**调用），保序。 席位 thread 跨轮累积历史，直接扫全量会把往轮的工具调用也算进来；以最后一条 human （本轮 run 的输入）为界，只取其后的 AI 工具调用，对应前端「这一条气泡的步骤卡」。 """ last_human = -1 for i, m in enumerate(messages): if isinstance(m, dict) and (m.get("type") or m.get("role")) in {"human", "user"}: last_human = i out: list[dict[str, Any]] = [] for m in messages[last_human + 1 :]: if not isinstance(m, dict): continue if (m.get("type") or m.get("role")) not in {"ai", "assistant"}: continue out.extend(_normalize_tool_calls(m.get("tool_calls"))) return out async def _run_agent_turn( app, *, user_id: str, thread_id: str, agent_name: str, message: str, model: str | None, excluded_tools: list[str], skill_stop_names: list[str], thinking_enabled: bool, reasoning_effort: str, subagent_enabled: bool = False, ) -> list[dict[str, Any]]: """进程内驱动 agent_name 在 thread_id 上跑一轮（发 message），阻塞到完成，返回最终 messages。 镜像 multi_agent ``_run_payload`` 的 body 形状：agent 通过 ``context.agent_name`` 选， assistant_id 固定 lead_agent；关闭 memory recall / subagent / plan mode（临时研讨任务）。 并发受**该用户的后台额度**约束（``_user_background_semaphore``）：真正开跑（start_run + await run task）前先取额度，跑完释放。同一用户的所有后台作业共享这几个额度，避免 2+ 后台挂起把单连接 SQLite checkpointer / 离线模型打满、拖垮前台实时会商。 """ from app.gateway.routers.thread_runs import RunCreateRequest from app.gateway.services import start_run body = RunCreateRequest( assistant_id="lead_agent", input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]}, config={"recursion_limit": 1000}, context={ "agent_name": agent_name, "model_name": model or "", "mode": "pro", "reasoning_effort": reasoning_effort, "thinking_enabled": thinking_enabled, "is_plan_mode": False, # ultra 席位才置真 → 解锁 task 子代理工具（席位 policy 未禁 task）。leader 恒 False。 "subagent_enabled": subagent_enabled, "thread_id": thread_id, "memory_recall_disabled": True, # 与 multi_agent._run_payload 对齐:关 builtin 记忆注入,避免单例总控/席位 # 被上轮会话 memory.json 带偏(臆造席位 / 直接干活)。详见该处注释。 "memory_injection_enabled": False, }, excluded_tools=excluded_tools, skill_stop_names=skill_stop_names, on_disconnect="continue", multitask_strategy="reject", ) # 虚拟 request：带上空 ``state``（auth 中间件在真请求上会写 state.user / state.auth； # 进程内没有这层，下游凡是读 request.state.X 的代码必须能拿到 None 而不是 AttributeError）。 request = SimpleNamespace(app=app, state=SimpleNamespace(user=None, auth=None)) # 取该用户的后台额度后再开跑：start_run + await task 是最重的部分（建 agent、调模型、 # 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。 async with _user_background_semaphore(user_id): record = await start_run(body, thread_id, request) # type: ignore[arg-type] if getattr(record, "task", None) is not None: await record.task # 阻塞到这轮 agent 跑完，checkpoint 已落盘 return await _read_thread_messages(app, thread_id) def _read_report_html(app, thread_id: str, user_id: str) -> str | None: """从 report thread 的 outputs 目录读出报告 HTML。 优先 ``方案总览.html``；找不到就扫目录里**任意 .html**（取体积最大的那个）， 容忍报告 agent 用了别的文件名。 """ try: outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id) preferred = outputs_dir / _REPORT_FILE if preferred.exists(): return preferred.read_text(encoding="utf-8") if outputs_dir.exists(): htmls = sorted( outputs_dir.glob("*.html"), key=lambda p: p.stat().st_size, reverse=True, ) if htmls: logger.info("report html: 方案总览.html not found, using %s", htmls[0].name) return htmls[0].read_text(encoding="utf-8") logger.warning("report html: no .html found in %s", outputs_dir) except Exception: # noqa: BLE001 logger.warning("read report html failed for thread %s", thread_id, exc_info=True) return None def _read_summary_md(app, thread_id: str, user_id: str) -> str | None: """从 summary thread 的 outputs 目录读出总结报告 Markdown。 优先 ``方案总结报告.md``；找不到就扫目录里**任意 .md**（取体积最大的那个）， 容忍总结 agent 用了别的文件名。镜像 ``_read_report_html``。 """ try: outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id) preferred = outputs_dir / _SUMMARY_FILE if preferred.exists(): return preferred.read_text(encoding="utf-8") if outputs_dir.exists(): mds = sorted( outputs_dir.glob("*.md"), key=lambda p: p.stat().st_size, reverse=True, ) if mds: logger.info("summary md: 方案总结报告.md not found, using %s", mds[0].name) return mds[0].read_text(encoding="utf-8") logger.warning("summary md: no .md found in %s", outputs_dir) except Exception: # noqa: BLE001 logger.warning("read summary md failed for thread %s", thread_id, exc_info=True) return None # Step3 产物自愈重试次数（首轮之外的纠偏轮数）。 # 实测痛点（2026-06-12，h:\xiugai\后台日志.txt）：deepseek 报告 agent 说「let me write # it all at once」后发出超长 write_file，单次输出被 max_tokens 截断 → 工具调用从未执行、 # run 以 success 收尾但 outputs 下没有 .html → step3.html 回退成开场白文字。 # 自愈：检测产物缺失/截断后，同线程追加纠偏消息（要求 write_file 分段 + append 续写）重试。 # # 取值依据（2026-06-15 修复「后台 HTML 看板只生成一半、概率渲染失败」）：模型单轮输出受 # ``config.yaml`` 的 ``max_tokens``（deepseek-chat = 4096）硬截。一份完整看板 HTML 几十 KB # ≈ 上万 token，单轮写不完；旧值 2（最多 3 轮 ≈ 12k token）对长报告不够 → 落库半截 HTML → # 渲染失败。调到 5（最多 6 轮 append ≈ 24k token，足够绝大多数看板）显著降低截断概率。 # 循环命中完整产物（```` / 非空 md / 合法 json）即 break，短报告不会白跑额外轮次。 _ARTIFACT_RETRY_LIMIT = 5 _REPORT_RETRY_MISSING = ( f"检测到 /mnt/user-data/outputs/{_REPORT_FILE} 还没有生成——你上一轮的 write_file 没有执行成功" "（最常见原因：试图一次写入整份长 HTML，输出超长被截断，工具调用没有真正落盘）。\n" "请立即重新写入，并**严格分段**：\n" f"1. 先用 write_file 写入 /mnt/user-data/outputs/{_REPORT_FILE} 的开头部分（ 起，约 200-300 行）；\n" "2. 再用 write_file 的 append=true 参数分多次续写后续片段，每段同样控制在 300 行以内；\n" "3. 直到文档以 收尾。\n" "不要在对话里粘贴 HTML 源码，写完用一两句话总结即可。" ) _REPORT_RETRY_TRUNCATED = ( f"检测到 /mnt/user-data/outputs/{_REPORT_FILE} 已写入但**不完整**（没有 收尾，疑似上一轮输出被截断）。\n" "请用 write_file 的 append=true 参数从中断处**继续补全剩余内容**（不要重写已有部分），" "每段控制在 300 行以内，分多次续写直到文档以 收尾。不要在对话里粘贴源码。" ) _SUMMARY_RETRY_MISSING = ( f"检测到 /mnt/user-data/outputs/{_SUMMARY_FILE} 还没有生成——你上一轮的 write_file 没有执行成功" "（可能是单次输出过长被截断）。请立即用 write_file 重新写入完整报告；若内容很长，" "先写前半部分、再用 write_file 的 append=true 参数分段续写。写完后把报告正文也输出到对话回复里。" ) _DASHBOARD_FENCE = "```report-json" _DASHBOARD_RETRY = ( "你刚才没有输出有效的 ```report-json 代码块（或其中的 JSON 无法解析）。" "请重新在回复**最末尾**输出一个完整且**严格合法**的 ```report-json 代码块" "（标准 JSON：无注释、无尾逗号、字符串用双引号），代码块之外只允许一两句说明；" "不要调用任何工具、不要写文件、不要输出 HTML。" ) def _extract_report_json(text: str) -> str | None: """从模型回复里提取最后一个 ```report-json 代码块的 JSON 字符串（须可解析）。""" start = text.rfind(_DASHBOARD_FENCE) if start < 0: return None after = text[start + len(_DASHBOARD_FENCE) :] close = after.find("```") body = (after[:close] if close >= 0 else after).strip() if not body: return None try: json.loads(body) except Exception: # noqa: BLE001 return None return body def _html_complete(html: str | None) -> bool: """报告 HTML 是否完整可用（非空 + 有 收尾，截断检测）。""" return bool(html and html.strip()) and "" in html.lower() def _build_seat_intro(agents: list[SeatRef]) -> str: """镜像 init 注入 coordinator 的「可调度席位清单」。""" lines = [f"- agent_name: {a.agent_id}（{a.name}）" for a in agents] return ( "【圆桌系统初始化】本次研讨可调度的席位清单如下，请严格按 agent_name 通过 " "agent_orchestration 技能派活，不要派给不在清单内的 agent：\n" + "\n".join(lines) ) # ── 网关实现 ───────────────────────────────────────────────────────────────── class InProcessRoundtableGateway: """进程内真网关。executor 先调 ``init_threads`` 建好线程，再跑编排引擎。""" def __init__( self, app, *, user_id: str, agents: list[SeatRef], model: str | None = None, seat_thinking_enabled: bool | None = None, seat_reasoning_effort: str | None = None, seat_subagent_enabled: bool = False, ) -> None: self._app = app self._user_id = user_id self._agents = list(agents) self._model = model # 席位执行模式（沿用 Step2 选择）：覆盖 seat policy 的 thinking/reasoning（None → policy 默认）， # subagent 仅 ultra 档为真。只作用 special 席位；leader 派活不受影响。 self._seat_thinking_enabled = seat_thinking_enabled self._seat_reasoning_effort = seat_reasoning_effort self._seat_subagent_enabled = seat_subagent_enabled self._seat_intro = "" self._first_leader = True async def init_threads(self) -> tuple[dict[str, str], str]: """建 coordinator + 各席位 thread，返回 (thread_ids, coordinator_name)。 席位清单通过「首轮 leader 消息前缀」注入（coordinator thread 跨轮累积历史， 首轮带上即长期可见），无需单独写 checkpoint。 """ ensure_roundtable_functional_agents() thread_ids: dict[str, str] = {} for a in self._agents: thread_ids[a.agent_id] = await _create_thread(self._app) thread_ids[COORDINATOR_AGENT_ID] = await _create_thread(self._app) self._seat_intro = _build_seat_intro(self._agents) return thread_ids, COORDINATOR_AGENT_ID async def run_leader(self, *, thread_ids, coordinator_name, message, model) -> LeaderTurn: msg = message if self._first_leader and self._seat_intro: msg = self._seat_intro + "\n\n" + message self._first_leader = False policy = roundtable_run_policy("leader") messages = await _run_agent_turn( self._app, user_id=self._user_id, thread_id=thread_ids[coordinator_name], agent_name=coordinator_name, message=msg, model=model or self._model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, ) ai = _latest_ai_message(messages) # tool_calls 取最后一条 AI 消息（派活就在这条）；正文取最后一条有文字的 AI 消息。 final_text = _latest_ai_text(messages) tool_calls = (ai.get("tool_calls") if ai else None) or [] # 本轮工具调用（派活 / 澄清），带进 LeaderTurn 供草稿 run 重建步骤卡。 turn_tool_calls = _normalize_tool_calls(tool_calls) # 澄清优先 clar = next((tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"), None) if clar is not None: cargs = clar.get("args") or {} question = str(cargs.get("question") or "").strip() or "总控需要补充信息，请补充后继续。" return LeaderTurn(final_text=final_text, clarification=question, tool_calls=turn_tool_calls) # 派活解析（复用 multi_agent 纯函数） from app.gateway.routers.multi_agent import _extract_dispatch dispatched: list[tuple[str, str]] = [] for tc in tool_calls: try: name, task = _extract_dispatch(tc) if name and task: dispatched.append((name, task)) except Exception: # noqa: BLE001 logger.warning("dispatch parse failed for tool_call", exc_info=True) return LeaderTurn(final_text=final_text, dispatched=dispatched, tool_calls=turn_tool_calls) async def run_seat(self, *, thread_ids, agent_id, task, model) -> SeatTurn: policy = roundtable_run_policy("seat") # 席位模式覆盖 policy 的 thinking/reasoning（None → policy 默认），与前端 _special_run 一致。 messages = await _run_agent_turn( self._app, user_id=self._user_id, thread_id=thread_ids[agent_id], agent_name=agent_id, message=task, model=model or self._model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=( policy.thinking_enabled if self._seat_thinking_enabled is None else self._seat_thinking_enabled ), reasoning_effort=( policy.reasoning_effort if self._seat_reasoning_effort is None else self._seat_reasoning_effort ), subagent_enabled=self._seat_subagent_enabled, ) return SeatTurn(final_text=_latest_ai_text(messages), tool_calls=_tool_calls_since_last_human(messages)) async def create_report_thread(self) -> str: ensure_roundtable_functional_agents() return await _create_thread(self._app) async def run_report(self, *, report_thread_id, prompt, model) -> ReportResult: """跑报告 agent，**带产物自愈**：首轮后 HTML 缺失/截断（无 ）→ 同线程追加 纠偏消息（分段 + append 续写）重试，至多 ``_ARTIFACT_RETRY_LIMIT`` 轮。""" policy = roundtable_run_policy("report") summary = "" html: str | None = None message = prompt for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): messages = await _run_agent_turn( self._app, user_id=self._user_id, thread_id=report_thread_id, agent_name=REPORT_AGENT_ID, message=message, model=model or self._model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, ) summary = _latest_ai_text(messages) or summary html = _read_report_html(self._app, report_thread_id, self._user_id) if _html_complete(html): break if attempt >= _ARTIFACT_RETRY_LIMIT: break message = _REPORT_RETRY_TRUNCATED if (html and html.strip()) else _REPORT_RETRY_MISSING logger.warning( "report html %s after attempt %d for thread %s; retrying with corrective prompt", "truncated" if (html and html.strip()) else "missing", attempt + 1, report_thread_id, ) # 截断的 HTML 也比纯文字强（浏览器容忍未闭合标签），仅在完全没有文件时才回退正文。 final_html = html if (html and html.strip()) else summary return ReportResult(html=final_html, summary=summary, model=model or self._model) async def create_summary_thread(self) -> str: ensure_roundtable_functional_agents() return await _create_thread(self._app) async def run_summary(self, *, summary_thread_id, prompt, model) -> SummaryResult: """进程内驱动 roundtable-summary 写「方案总结报告.md」+ 对话正文（与前端 useStep3Summary 同一智能体 / 同一 run policy / 同款产物），返回与 DraftStep3SummarySnapshot 对齐的结果。线程是真实持久线程——加载草稿后可续问。 带产物自愈：md 文件缺失时同线程追加纠偏消息重试（同 ``run_report``）。""" policy = roundtable_run_policy("summary") closing = "" md: str | None = None message = prompt for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): messages = await _run_agent_turn( self._app, user_id=self._user_id, thread_id=summary_thread_id, agent_name=SUMMARY_AGENT_ID, message=message, model=model or self._model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, ) closing = _latest_ai_text(messages) or closing md = _read_summary_md(self._app, summary_thread_id, self._user_id) if md and md.strip(): break if attempt >= _ARTIFACT_RETRY_LIMIT: break message = _SUMMARY_RETRY_MISSING logger.warning( "summary md missing after attempt %d for thread %s; retrying with corrective prompt", attempt + 1, summary_thread_id, ) final_md = md if (md and md.strip()) else closing return SummaryResult(md=final_md, closing=closing, model=model or self._model, thread_id=summary_thread_id) async def run_dashboard_data(self, *, prompt, model) -> str | None: """大屏多页（模式二）的结构化数据：进程内驱动 roundtable-report 输出 ```report-json（prompt 已声明禁工具/禁文件），返回可解析的 JSON 字符串。 缺失 / 坏 JSON → 同线程纠偏重试（同 run_report/run_summary 的自愈语义）。""" ensure_roundtable_functional_agents() thread_id = await _create_thread(self._app) policy = roundtable_run_policy("report") message = prompt for attempt in range(1 + _ARTIFACT_RETRY_LIMIT): messages = await _run_agent_turn( self._app, user_id=self._user_id, thread_id=thread_id, agent_name=REPORT_AGENT_ID, message=message, model=model or self._model, excluded_tools=policy.excluded_tools, skill_stop_names=policy.skill_stop_names, thinking_enabled=policy.thinking_enabled, reasoning_effort=policy.reasoning_effort, ) body = _extract_report_json(_latest_ai_text(messages)) if body: return body if attempt >= _ARTIFACT_RETRY_LIMIT: break message = _DASHBOARD_RETRY logger.warning( "dashboard report-json missing/invalid after attempt %d for thread %s; retrying", attempt + 1, thread_id, ) return None