From 94f6bb7ce2ec9ed387017cbafd7dc55f5a962c45 Mon Sep 17 00:00:00 2001 From: LineWalker Date: Sun, 9 Aug 2026 14:53:34 +0800 Subject: [PATCH 1/6] =?UTF-8?q?fix(linsight):=20=E5=AD=90=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E8=BD=AE=E6=AC=A1=E9=A2=84=E7=AE=97=E6=8C=89=20task?= =?UTF-8?q?=20=E8=B0=83=E7=94=A8=E5=88=86=E6=A1=B6=EF=BC=8Cwrite=5Ftodos?= =?UTF-8?q?=20=E7=A9=BA=E8=BD=AC=E9=80=80=E8=BF=98=E9=A2=84=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 180 POC 会话 aa352cb4 复盘：主图并行委派 2 个调研子代理，日志显示 `graph=sub turn 29/30`——那是两个子代理**加起来**的数。 根因：deepagents 在建 task 工具时把子代理一次性编译 （subagents.py:584），每次 task 调用都重入同一个 runnable，而中间件的 `_turn_count` 是实例属性且从不重置，于是 N 次委派共吃一份 30 轮额度， 第二个子代理一上来就落在软着陆区。注释里只声称「主图与子代理各自独立」， 从未声称多次 task 调用共享——是隐性缺陷，不是设计意图。 改为按 LangGraph 节点 namespace 前缀分桶（有界 LRU，上限 128）： - 子代理 → 每次 task 调用一个桶，同一次调用跨轮次恒定、并发调用互不相同； - 主图 → 强制单桶。主图 ns 形如 `model:` 每轮都变且不含分隔符， 一旦分桶预算每轮重置，是灾难性回归； - 拿不到 runtime / ns 扁平 → 回落单桶 = 修复前行为，永不劣化。 同时：一次模型调用若只产出 write_todos（纯状态维护）不再扣预算。实测子代理 29 次调用里 10 次是这种空转。两段式记账——软着陆档位仍在调用前决定（它要改写 request），退款放在循环唯一的成功出口，所以瞬时重试/截断补写不会重复退款， 降级轮不退款。退款每桶封顶 10，否则 write_todos 死循环会让计数器永不前进、 软着陆阶梯永不触发，最后死在 GraphRecursionError。 新增 test_subagent_ns_contract.py：纯 langgraph 不调模型，钉住 namespace 的 三条性质。这是升级 langgraph/langchain 时唯一能提前炸出来的守门测试。 Co-Authored-By: Claude Opus 5 (1M context) --- src/backend/bisheng/initdb_config.yaml | 14 +- .../domain/services/resilience_middleware.py | 213 ++++++++++++-- .../linsight/test_subagent_ns_contract.py | 145 ++++++++++ src/backend/test/linsight/test_turn_budget.py | 273 +++++++++++++++++- 4 files changed, 609 insertions(+), 36 deletions(-) create mode 100644 src/backend/test/linsight/test_subagent_ns_contract.py diff --git a/src/backend/bisheng/initdb_config.yaml b/src/backend/bisheng/initdb_config.yaml index 600cd6d380..772f7dded1 100644 --- a/src/backend/bisheng/initdb_config.yaml +++ b/src/backend/bisheng/initdb_config.yaml @@ -104,14 +104,18 @@ linsight: tool_buffer: 100000 # 单个任务最大执行步骤数（LangGraph recursion_limit），防止死循环。 # ⚠️ 这里的“步骤”是 LangGraph super-step，不是模型轮次：一轮模型调用约消耗 4 个 super-step - # （model → 工具循环熔断器 after_model → TodoList after_model → tools）， - # 所以 500 ≈ 115 轮。它是跑飞时的保险丝，真正的业务上限请调 max_model_turns。 - # 若本值低于 max_model_turns 所需，后端会自动抬高并打 warning 日志。 - max_steps: 500 + # （model → 工具循环熔断器 after_model → TodoList after_model → tools）。 + # 另外「只产出 write_todos」的空转轮会退还预算（每个预算桶最多退 10 轮）， + # 所以实际模型调用可能超过 max_model_turns，保险丝要留出这部分余量： + # (115 + 10) × 4 + 20 = 520，取 600 留头。 + # 它是跑飞时的保险丝，真正的业务上限请调 max_model_turns。 + # 若本值低于所需，后端会自动抬高并打 warning 日志。 + max_steps: 600 # 主图轮次预算：一次任务运行最多允许多少次模型调用，到顶前会提示模型收尾（软着陆）。 # ask_user 打断恢复后重新计数（中间件随 agent 重建）。 max_model_turns: 115 - # 子代理（researcher）自己的轮次预算，独立计数（子图跑自己的 Pregel 循环） + # 子代理（researcher）轮次预算。⚠️ 按「每次 task 调用」独立计数，不是整次任务共享： + # 主图并行委派 2 个子代理时，每个各得完整一份，互不抢额度。 max_model_turns_subagent: 30 # 距离预算耗尽还剩多少轮时开始提示模型收尾；最后 2 轮只保留写文件/导出工具， # 归零时不再提供任何工具，模型只能输出文本，图正常结束 diff --git a/src/backend/bisheng/linsight/domain/services/resilience_middleware.py b/src/backend/bisheng/linsight/domain/services/resilience_middleware.py index 79250439cb..4ad8909d73 100644 --- a/src/backend/bisheng/linsight/domain/services/resilience_middleware.py +++ b/src/backend/bisheng/linsight/domain/services/resilience_middleware.py @@ -30,6 +30,7 @@ import asyncio import time +from collections import OrderedDict from collections.abc import Awaitable, Callable from langchain.agents.middleware._retry import calculate_delay @@ -123,6 +124,32 @@ # the whole point of landing softly is to still produce the file. _POST_BUDGET_ALLOWED_TOOLS = _DELIVERABLE_TOOLS | {"write_todos"} +# LangGraph's checkpoint-namespace level separator, mirroring +# ``langgraph._internal._constants.NS_SEP``. Re-declared rather than imported: the +# value is part of the checkpoint wire format and far more stable than the private +# module path. Pinned by ``test/linsight/test_subagent_ns_contract.py``. +_NS_SEP = "|" + +# Upper bound on live turn-budget buckets (see ``_budget_key``). Observed concurrent +# delegations are single-digit; 128 is far above any real plan, and eviction is LRU +# so an active bucket can never be pushed out by stale ones. +_BUDGET_BUCKETS_MAX = 128 + +# Tool calls that are PURE STATE MAINTENANCE: they move no work forward, they only +# republish the plan. deepagents injects ``TodoListMiddleware`` into every subagent +# unconditionally (``deepagents/graph.py:643-651``) — it is not part of the business +# tool subset ``agent_factory._subagent_tools`` builds. Measured on v2.6.0-fix2: 10 of +# a researcher's 29 model calls produced ``write_todos`` and nothing else, i.e. a third +# of a 30-turn budget bought zero research. Those turns are refunded instead. +_STATE_ONLY_TOOLS = frozenset({"write_todos"}) + +# Refunds are CAPPED. Uncapped, a model looping on ``write_todos`` would never advance +# the counter, the soft-landing ladder would never fire, and the run would die on +# GraphRecursionError instead — the exact failure ``_resolve_recursion_limit`` exists +# to prevent. The L3 tool-loop breaker cannot cover this either: it trips on tool +# FAILURES, and ``write_todos`` succeeds every time. +_MAX_STATE_ONLY_REFUNDS = 10 + _BUDGET_SPENT_TOOL_REPLY = ( "⚠️ 本次任务的模型调用次数预算已用尽，{tool_name} 未被执行。" "请立即用已经掌握的材料完成交付：先用 write_file 把最终成果写入 output/ 目录下的交付文件，" @@ -191,6 +218,30 @@ def _with_truncation_nudge(request: ModelRequest) -> ModelRequest: return request.override(messages=[*request.messages, HumanMessage(content=_TRUNCATION_NUDGE)]) +def _is_state_only_turn(response: object) -> bool: + """True iff this model call produced ONLY pure state-maintenance tool calls. + + Deliberately strict — a turn is refunded only when it demonstrably moved nothing + forward. Everything else still costs a turn: + + - no ``AIMessage`` at all → unknown shape, stay conservative; + - no tool calls (a text-only close-out) → that IS the run's real last turn, and + refunding it would keep the ladder from ever reaching stage 3; + - any invalid/malformed tool call → the model attempted real work and failed; + - ``write_todos`` alongside any other tool → real work happened this turn. + """ + ai = _response_ai_message(response) + if ai is None: + return False + if getattr(ai, "invalid_tool_calls", None): + return False + calls = getattr(ai, "tool_calls", None) or [] + if not calls: + return False + names = {tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None) for tc in calls} + return names <= _STATE_ONLY_TOOLS + + def _summarize_tool_calls(ai: AIMessage) -> list[str]: """Compact ``name(argkey1,argkey2)`` per tool call — argument KEYS only, never VALUES, so a large ``write_file`` ``content`` is never dumped into the log. A @@ -268,15 +319,28 @@ def __init__( self.max_degrade = max(0, max_degrade) self.truncation_retry_limit = max(0, truncation_retry_limit) self.is_subagent = is_subagent + # KNOWN GAP (not fixed here to keep the backport surface small): this counter + # has the same cross-``task``-call sharing problem the turn budget had — every + # delegation shares one ``max_degrade`` allowance. Bucket it on ``_budget_key`` + # when touching this next. self._degrade_count = 0 - # Turn budget. Per-instance, and each graph builds its own instance, so the - # main graph and the researcher subagent hold separate budgets. It resets - # whenever the agent is rebuilt — i.e. an ask_user resume grants a fresh - # allowance, matching LangGraph's own ``stop = step + recursion_limit + 1`` - # recomputation on resume. + # Turn budget, bucketed per graph RUN rather than per middleware instance: + # + # - main graph → exactly one bucket. One compiled Pregel loop, one allowance; + # ``budget_sink`` semantics unchanged. + # - subagent → ONE BUCKET PER ``task`` TOOL CALL. deepagents compiles the + # researcher ONCE (``subagents.py:584``) and every ``task`` call re-enters + # that same runnable, so a single counter was silently shared: two parallel + # researchers burned one 30-turn allowance between them and the second one + # started already inside the soft-landing zone (measured, v2.6.0-fix2). + # + # Either way the budget resets when the agent is rebuilt — an ask_user resume + # grants a fresh allowance, matching LangGraph's own + # ``stop = step + recursion_limit + 1`` recomputation on resume. self.turn_limit = max(1, turn_limit) self.soft_landing_turns = max(0, soft_landing_turns) - self._turn_count = 0 + self._turns: OrderedDict[str, int] = OrderedDict() + self._refunds: OrderedDict[str, int] = OrderedDict() # Optional shared dict the task executor reads after the run to tell the # user their result was wrapped up early. self._budget_sink = budget_sink @@ -296,17 +360,83 @@ def _delay(self, attempt: int) -> float: jitter=self.jitter, ) - def _apply_turn_budget(self, request: ModelRequest) -> ModelRequest: + def _budget_key(self, request: object) -> str: + """Which turn-budget bucket this model/tool call belongs to. + + Main graph → always the single default bucket. The main graph is ONE compiled + Pregel loop with ONE allowance, and its node namespace (``model:``) + carries a fresh task id every turn, so bucketing it would hand it a brand-new + budget on every single call. + + Subagent → the parent namespace of the current node. LangGraph gives each + ``task`` tool call its own PUSH task (one ``Send`` per tool call, whose task id + includes the Send index), and every node inside the resulting subgraph run is + namespaced ``|:``. Dropping the last + segment therefore yields a key that is CONSTANT across all turns of one + ``task`` call and DISTINCT between concurrent ones — pinned by + ``test/linsight/test_subagent_ns_contract.py``. + + Anything unexpected (no runtime, as in unit tests or non-graph callers; or a + flat namespace, meaning the subagent graph ran un-nested) falls back to the + shared bucket, which is exactly the pre-fix behaviour — never worse. + """ + if not self.is_subagent: + return "" + runtime = getattr(request, "runtime", None) + info = getattr(runtime, "execution_info", None) + ns = getattr(info, "checkpoint_ns", None) + if not isinstance(ns, str) or _NS_SEP not in ns: + return "" + return ns.rsplit(_NS_SEP, 1)[0] + + def _turns_used(self, key: str = "") -> int: + return self._turns.get(key, 0) + + def _bump_turn(self, key: str) -> int: + """Count one turn against ``key`` and return the new total (LRU-bounded).""" + used = self._turns.pop(key, 0) + 1 + self._turns[key] = used # re-insert → most-recently-used tail + while len(self._turns) > _BUDGET_BUCKETS_MAX: + evicted, _ = self._turns.popitem(last=False) + self._refunds.pop(evicted, None) + return used + + def _refund_turn(self, key: str) -> None: + """Give a pure state-maintenance turn its budget back (bounded per bucket).""" + used = self._turns.get(key, 0) + if used <= 0: + return + if self._refunds.get(key, 0) >= _MAX_STATE_ONLY_REFUNDS: + return + self._turns[key] = used - 1 + self._refunds[key] = self._refunds.get(key, 0) + 1 + + @property + def _turn_count(self) -> int: + """Turns used in the DEFAULT bucket — i.e. the whole budget for the main graph. + + Read-only alias kept so main-graph call sites and existing tests read + unchanged; subagent buckets must be read via ``_turns_used(key)``. + """ + return self._turns_used("") + + def _apply_turn_budget(self, request: ModelRequest) -> tuple[ModelRequest, str]: """Count this turn and apply the soft-landing stage it falls into. Called ONCE per ``wrap_model_call`` — i.e. once per model node execution. The retry loops below re-enter ``handler`` without re-entering this, so a transient retry or a truncation nudge never burns turn budget. + + Returns the (possibly nudged) request together with the budget key it was + counted against, so the caller can refund a pure state-maintenance turn once + the response makes that knowable. """ - self._turn_count += 1 - remaining = self.turn_limit - self._turn_count + key = self._budget_key(request) + graph = "sub" if self.is_subagent else "main" + count = self._bump_turn(key) + remaining = self.turn_limit - count if remaining > self.soft_landing_turns: - return request + return request, key self._mark_soft_landing() if remaining <= 0: @@ -314,29 +444,35 @@ def _apply_turn_budget(self, request: ModelRequest) -> ModelRequest: # routes the graph straight to END. This is what turns "budget # exhausted" into a normal completion instead of a recursion abort. logger.warning( - "[linsight-turn-budget] graph={} turn {}/{} — budget exhausted, forcing a text-only close-out", - "sub" if self.is_subagent else "main", - self._turn_count, + "[linsight-turn-budget] graph={} key={} turn {}/{} — budget exhausted, forcing a text-only close-out", + graph, + key or "-", + count, self.turn_limit, ) - return _with_wrap_up_nudge(request, _LAST_CHANCE_NUDGE, 0).override(tools=[]) + return _with_wrap_up_nudge(request, _LAST_CHANCE_NUDGE, 0).override(tools=[]), key if remaining <= _WRITE_ONLY_TURNS_LEFT: logger.warning( - "[linsight-turn-budget] graph={} turn {}/{} — {} left, narrowing to deliverable tools", - "sub" if self.is_subagent else "main", - self._turn_count, + "[linsight-turn-budget] graph={} key={} turn {}/{} — {} left, narrowing to deliverable tools", + graph, + key or "-", + count, self.turn_limit, remaining, ) - return _only_deliverable_tools(_with_wrap_up_nudge(request, _LAST_CHANCE_NUDGE, remaining)) + return ( + _only_deliverable_tools(_with_wrap_up_nudge(request, _LAST_CHANCE_NUDGE, remaining)), + key, + ) logger.info( - "[linsight-turn-budget] graph={} turn {}/{} — {} left, nudging the model to wrap up", - "sub" if self.is_subagent else "main", - self._turn_count, + "[linsight-turn-budget] graph={} key={} turn {}/{} — {} left, nudging the model to wrap up", + graph, + key or "-", + count, self.turn_limit, remaining, ) - return _with_wrap_up_nudge(request, _WRAP_UP_NUDGE, remaining) + return _with_wrap_up_nudge(request, _WRAP_UP_NUDGE, remaining), key def _budget_blocked_reply(self, request) -> ToolMessage | None: """Refuse an exploratory tool call once the turn budget is spent. @@ -346,16 +482,18 @@ def _budget_blocked_reply(self, request) -> ToolMessage | None: its wrap_tool_call is the outermost one — short-circuiting here also skips the inner guards, which is what we want for a call that never ran. """ - if self._turn_count < self.turn_limit: + key = self._budget_key(request) + if self._turns_used(key) < self.turn_limit: return None tool_call = request.tool_call or {} name = tool_call.get("name") if name in _POST_BUDGET_ALLOWED_TOOLS: return None logger.warning( - "[linsight-turn-budget] graph={} budget spent ({}/{}) — refusing tool call '{}'", + "[linsight-turn-budget] graph={} key={} budget spent ({}/{}) — refusing tool call '{}'", "sub" if self.is_subagent else "main", - self._turn_count, + key or "-", + self._turns_used(key), self.turn_limit, name, ) @@ -383,7 +521,13 @@ def wrap_tool_call(self, request, handler): return handler(request) def _mark_soft_landing(self) -> None: - """Flag the run as wrapped-up-early for the task executor's user-facing note.""" + """Flag the run as wrapped-up-early for the task executor's user-facing note. + + Deliberately sticky: a turn that triggered a wrap-up nudge and was then + refunded (``_refund_turn``) does NOT clear this. The nudge really was sent and + really did shape that turn, so telling the user their result was closed out + early is still true. + """ if self._budget_sink is not None: self._budget_sink["soft_landing"] = True @@ -415,7 +559,7 @@ async def awrap_model_call( # SEPARATE budgets so a truncation retry never eats the exception-retry # budget and vice-versa. ``current`` carries the (possibly nudged) request. # The turn budget is applied ONCE, outside the retry loop. - current = self._apply_turn_budget(request) + current, budget_key = self._apply_turn_budget(request) exc_attempts = 0 trunc_attempts = 0 while True: @@ -450,6 +594,15 @@ async def awrap_model_call( ) current = _with_truncation_nudge(current) continue + # Two-phase turn accounting: the soft-landing STAGE had to be picked before + # the call (it shapes the request), but whether this turn did any real work + # is only knowable from the response. Refund here — the loop's SINGLE + # success exit — so a transient retry or a truncation nudge (both + # ``continue`` above) can never double-refund, and a degraded call (which + # returns from the except branch) is never refunded: it really did burn + # model calls. + if _is_state_only_turn(response): + self._refund_turn(budget_key) return response def wrap_model_call( @@ -457,7 +610,7 @@ def wrap_model_call( request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse | AIMessage: - current = self._apply_turn_budget(request) + current, budget_key = self._apply_turn_budget(request) exc_attempts = 0 trunc_attempts = 0 while True: @@ -479,6 +632,10 @@ def wrap_model_call( trunc_attempts += 1 current = _with_truncation_nudge(current) continue + # Refund a pure state-maintenance turn — see the async twin above for why + # this sits at the loop's single success exit. + if _is_state_only_turn(response): + self._refund_turn(budget_key) return response diff --git a/src/backend/test/linsight/test_subagent_ns_contract.py b/src/backend/test/linsight/test_subagent_ns_contract.py new file mode 100644 index 0000000000..6185e4e60d --- /dev/null +++ b/src/backend/test/linsight/test_subagent_ns_contract.py @@ -0,0 +1,145 @@ +"""Framework contract: how LangGraph namespaces a re-entered subgraph. + +``LinsightModelResilienceMiddleware`` gives the researcher subagent a per-``task``-call +turn budget by bucketing its counter on the LangGraph node namespace. That only works +because of three properties of the framework, none of which we control: + +1. every ``task`` tool call becomes its own PUSH task, so two parallel delegations get + distinct namespaces even though deepagents compiled the subagent exactly ONCE + (``deepagents/middleware/subagents.py:584`` builds ``compiled_subagents`` at + ``_build_task_tool`` time; ``:640-653`` hands the same runnable to every call); +2. inside one such call the namespace prefix is CONSTANT across turns and across + model/tools nodes, so the bucket survives the whole delegation; +3. the main graph's namespace has NO separator, which is what lets the middleware + force it onto a single bucket — bucketing the main graph would reset its budget + every turn. + +This test pins all three WITHOUT calling a model, so a langgraph/langchain upgrade +that changes namespace construction fails here instead of silently handing every +subagent an unlimited budget (property 1/2) or the main graph a broken one (property 3). +""" + +from __future__ import annotations + +import operator +from typing import Annotated, TypedDict + +from langgraph.graph import END, START, StateGraph +from langgraph.runtime import ExecutionInfo, get_runtime +from langgraph.types import Send + +# Mirrors ``langgraph._internal._constants.NS_SEP``. Deliberately re-declared instead +# of imported: the value is part of the checkpoint wire format (far more stable than +# the private module path), and the production code makes the same choice. +NS_SEP = "|" + +_SUB_TURNS = 3 + + +class _SubState(TypedDict): + turns: int + seen: Annotated[list[str], operator.add] + + +class _ParentState(TypedDict): + seen: Annotated[list[str], operator.add] + sub_runs: Annotated[list[list[str]], operator.add] + + +def _current_ns() -> str: + return get_runtime().execution_info.checkpoint_ns + + +def _sub_model(state: _SubState) -> dict: + return {"turns": state["turns"] + 1, "seen": [_current_ns()]} + + +def _sub_tools(state: _SubState) -> dict: + return {"seen": [_current_ns()]} + + +def _sub_route(state: _SubState) -> str: + return "sub_tools" if state["turns"] < _SUB_TURNS else END + + +def _build_subgraph(): + graph = StateGraph(_SubState) + graph.add_node("model_request", _sub_model) + graph.add_node("sub_tools", _sub_tools) + graph.add_edge(START, "model_request") + graph.add_conditional_edges("model_request", _sub_route, ["sub_tools", END]) + graph.add_edge("sub_tools", "model_request") + return graph.compile() + + +# Compiled ONCE at import, exactly like deepagents compiles the researcher once and +# re-enters it on every ``task`` call. Compiling per test would defeat the point. +_SUBGRAPH = _build_subgraph() + + +def _parent_agent(state: _ParentState) -> dict: + return {"seen": [_current_ns()]} + + +def _parent_fan_out(state: _ParentState) -> list[Send]: + # Two delegations from ONE model turn — the shape that exposed the shared budget + # in production (both ``task`` calls arrived in a single tool_calls array). + return [Send("tools", {"turns": 0, "seen": []}), Send("tools", {"turns": 0, "seen": []})] + + +def _parent_tools(state: _SubState) -> dict: + result = _SUBGRAPH.invoke({"turns": 0, "seen": []}) + return {"sub_runs": [result["seen"]]} + + +def _build_parent(): + graph = StateGraph(_ParentState) + graph.add_node("agent", _parent_agent) + graph.add_node("tools", _parent_tools) + graph.add_edge(START, "agent") + graph.add_conditional_edges("agent", _parent_fan_out, ["tools"]) + graph.add_edge("tools", END) + return graph.compile() + + +def _prefix(ns: str) -> str: + """The bucket key the middleware derives — see ``_budget_key``.""" + return ns.rsplit(NS_SEP, 1)[0] + + +def test_execution_info_still_exposes_checkpoint_ns(): + """The attribute the whole scheme reads. Guards dependency upgrades.""" + assert "checkpoint_ns" in ExecutionInfo.__dataclass_fields__ + + +def test_subagent_namespace_prefix_is_stable_within_one_call(): + result = _build_parent().invoke({"seen": [], "sub_runs": []}) + + for run in result["sub_runs"]: + # model_request x3 + sub_tools x2 — every node of the delegation. + assert len(run) == _SUB_TURNS * 2 - 1 + prefixes = {_prefix(ns) for ns in run} + assert len(prefixes) == 1, f"prefix drifted across turns of one task call: {prefixes}" + assert NS_SEP in run[0], f"nested subgraph namespace lost its separator: {run[0]!r}" + + +def test_parallel_task_calls_get_distinct_namespace_prefixes(): + result = _build_parent().invoke({"seen": [], "sub_runs": []}) + + assert len(result["sub_runs"]) == 2 + first, second = ({_prefix(ns) for ns in run}.pop() for run in result["sub_runs"]) + assert first != second, "two task calls shared a bucket key — budgets would be shared again" + + +def test_main_graph_namespace_has_no_separator(): + """Why ``_budget_key`` must return the constant bucket for the main graph. + + The main graph's node namespace carries a fresh task id every turn and no + separator, so bucketing on it would hand the main graph a brand-new budget on + every single model call. + """ + result = _build_parent().invoke({"seen": [], "sub_runs": []}) + + assert result["seen"], "parent node never recorded its namespace" + for ns in result["seen"]: + assert NS_SEP not in ns, f"main-graph namespace unexpectedly nested: {ns!r}" diff --git a/src/backend/test/linsight/test_turn_budget.py b/src/backend/test/linsight/test_turn_budget.py index 243b23521a..ea788365bf 100644 --- a/src/backend/test/linsight/test_turn_budget.py +++ b/src/backend/test/linsight/test_turn_budget.py @@ -17,10 +17,14 @@ from __future__ import annotations +from types import SimpleNamespace + from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from bisheng.linsight.domain.services.resilience_middleware import ( + _BUDGET_BUCKETS_MAX, _DELIVERABLE_TOOLS, + _MAX_STATE_ONLY_REFUNDS, LinsightModelResilienceMiddleware, build_resilience_middleware, ) @@ -36,7 +40,7 @@ def __init__(self, name: str) -> None: class FakeRequest: """ModelRequest stand-in supporting the immutable ``override`` contract.""" - def __init__(self, messages=None, tools=None) -> None: + def __init__(self, messages=None, tools=None, runtime=None) -> None: self.messages = messages if messages is not None else [HumanMessage(content="做一份 PPT")] self.tools = ( tools @@ -49,14 +53,33 @@ def __init__(self, messages=None, tools=None) -> None: FakeTool("read_file"), ] ) + # Defaults to None so every pre-existing case keeps exercising the + # no-runtime path, which must behave exactly as it did before bucketing. + self.runtime = runtime def override(self, **kwargs): - new = FakeRequest(messages=list(self.messages), tools=list(self.tools)) + new = FakeRequest(messages=list(self.messages), tools=list(self.tools), runtime=self.runtime) for key, value in kwargs.items(): setattr(new, key, value) return new +# Real namespaces captured from ``test_subagent_ns_contract.py``: two parallel ``task`` +# calls, each re-entering the SAME compiled researcher subgraph. +NS_A = "tools:a6dc6726-df24-8a53-049a-fbf4c35a1e9c|model_request:0d1c6618-5d9a-1111-2222-333344445555" +NS_A_TOOLS = "tools:a6dc6726-df24-8a53-049a-fbf4c35a1e9c|sub_tools:4c3b921e-1111-2222-3333-444455556666" +NS_B = "tools:5923fee9-ce67-f1a7-a07f-265bbf188878|model_request:f71b1032-5e6b-1111-2222-333344445555" + + +def _runtime(ns: str): + return SimpleNamespace(execution_info=SimpleNamespace(checkpoint_ns=ns)) + + +def ns_request(ns: str, **kwargs): + """A request that looks like it came from inside a namespaced subgraph node.""" + return FakeRequest(runtime=_runtime(ns), **kwargs) + + def make_mw(*, turn_limit=115, soft_landing_turns=8, is_subagent=False, sink=None): return LinsightModelResilienceMiddleware( max_retries=2, @@ -245,8 +268,9 @@ class Conf: class FakeToolRequest: - def __init__(self, name: str, call_id: str = "call_1") -> None: + def __init__(self, name: str, call_id: str = "call_1", ns: str | None = None) -> None: self.tool_call = {"name": name, "args": {}, "id": call_id} + self.runtime = _runtime(ns) if ns is not None else None def tool_handler(): @@ -334,3 +358,246 @@ class LegacyConf: mw = build_resilience_middleware(LegacyConf(), is_subagent=False) assert mw.turn_limit == 115 assert mw.soft_landing_turns == 8 + + +# -------------------------------------------------------------------------- +# Per-``task``-call budget buckets +# +# deepagents compiles the researcher subagent ONCE and re-enters that same +# runnable on every ``task`` call, so one middleware instance serves every +# delegation. Before bucketing, two parallel researchers split ONE 30-turn +# allowance: measured on v2.6.0-fix2, `graph=sub turn 29/30` was the two of them +# added together and the second one started already inside the soft-landing zone. +# -------------------------------------------------------------------------- + + +def test_key_is_the_namespace_prefix_for_a_subagent(): + mw = make_mw(is_subagent=True) + key = mw._budget_key(ns_request(NS_A)) + assert key == "tools:a6dc6726-df24-8a53-049a-fbf4c35a1e9c" + # Different node inside the SAME task call → same bucket. + assert mw._budget_key(ns_request(NS_A_TOOLS)) == key + # A different task call → different bucket. + assert mw._budget_key(ns_request(NS_B)) != key + + +async def test_two_task_calls_get_independent_budgets(): + mw = make_mw(turn_limit=3, soft_landing_turns=8, is_subagent=True) + handler = capturing_handler() + for _ in range(3): + await mw.awrap_model_call(ns_request(NS_A), handler) + assert handler.state["requests"][-1].tools == [] # A is exhausted + + # B starts fresh: 3 left, still above the soft-landing window used here. + request_b = ns_request(NS_B) + await mw.awrap_model_call(request_b, handler) + assert mw._turns_used(mw._budget_key(request_b)) == 1 + assert handler.state["requests"][-1].tools != [] + + +async def test_parallel_task_calls_do_not_share_the_ladder(): + """The regression: interleaved delegations must each get their own stage.""" + mw = make_mw(turn_limit=3, soft_landing_turns=8, is_subagent=True) + handler = capturing_handler() + for _ in range(3): + await mw.awrap_model_call(ns_request(NS_A), handler) + await mw.awrap_model_call(ns_request(NS_B), handler) + + a_turns = mw._turns_used("tools:a6dc6726-df24-8a53-049a-fbf4c35a1e9c") + b_turns = mw._turns_used("tools:5923fee9-ce67-f1a7-a07f-265bbf188878") + assert a_turns == b_turns == 3 # not 6 shared between them + + +async def test_main_graph_ignores_the_namespace(): + """Gate-keeper: the main graph's namespace changes every turn and carries no + separator, so bucketing it would hand it a brand-new budget on every call.""" + mw = make_mw(turn_limit=10, is_subagent=False) + handler = capturing_handler() + await mw.awrap_model_call(ns_request(NS_A), handler) + await mw.awrap_model_call(ns_request(NS_B), handler) + assert mw._turn_count == 2 + assert list(mw._turns) == [""] + + +async def test_missing_runtime_falls_back_to_the_shared_bucket(): + mw = make_mw(turn_limit=10, is_subagent=True) + handler = capturing_handler() + for _ in range(2): + await mw.awrap_model_call(FakeRequest(), handler) + assert mw._turn_count == 2 + + +async def test_flat_namespace_falls_back_to_the_shared_bucket(): + """A subagent graph that ran un-nested has no separator — degrade, never crash.""" + mw = make_mw(turn_limit=10, is_subagent=True) + handler = capturing_handler() + await mw.awrap_model_call(ns_request("model:abc"), handler) + assert mw._turn_count == 1 + + +async def test_tool_refusal_uses_the_calling_task_bucket(): + mw = make_mw(turn_limit=1, is_subagent=True) + await mw.awrap_model_call(ns_request(NS_A), capturing_handler()) # spend A only + + refused_handler = tool_handler() + refused = await mw.awrap_tool_call(FakeToolRequest("read_file", ns=NS_A_TOOLS), refused_handler) + assert refused_handler.state["calls"] == 0 + assert "预算已用尽" in refused.content + + allowed_handler = tool_handler() + allowed = await mw.awrap_tool_call(FakeToolRequest("read_file", ns=NS_B), allowed_handler) + assert allowed_handler.state["calls"] == 1 + assert allowed.content == "executed" + + +async def test_bucket_table_is_bounded(): + mw = make_mw(turn_limit=100, is_subagent=True) + handler = capturing_handler() + for i in range(_BUDGET_BUCKETS_MAX + 40): + await mw.awrap_model_call(ns_request(f"tools:{i}|model_request:x"), handler) + assert len(mw._turns) <= _BUDGET_BUCKETS_MAX + assert len(mw._refunds) <= _BUDGET_BUCKETS_MAX + + +# -------------------------------------------------------------------------- +# State-only turns are refunded +# +# ``TodoListMiddleware`` is injected into every subagent by deepagents, and a +# model call that only re-publishes the todo list buys nothing. Measured: 10 of a +# researcher's 29 calls were exactly that. +# -------------------------------------------------------------------------- + + +def todo_handler(tool_names_seq=("write_todos",), *, invalid=False, fail_times=0, truncate_first=False): + """Handler returning an AIMessage whose tool calls are under our control.""" + state = {"calls": 0, "requests": []} + + async def handler(request): + state["calls"] += 1 + state["requests"].append(request) + if state["calls"] <= fail_times: + import openai + + exc = openai.APITimeoutError.__new__(openai.APITimeoutError) + exc.message = "timeout" + exc.code = None + exc.body = None + raise exc + truncated = truncate_first and state["calls"] == fail_times + 1 + return AIMessage( + content="", + tool_calls=[{"name": name, "args": {}, "id": f"c{i}"} for i, name in enumerate(tool_names_seq)], + invalid_tool_calls=([{"name": "write_file", "args": "{bad", "id": "bad", "error": "x"}] if invalid else []), + response_metadata={"finish_reason": "length"} if truncated else {}, + ) + + handler.state = state + return handler + + +async def test_write_todos_only_turn_is_refunded(): + mw = make_mw(turn_limit=10) + for _ in range(3): + await mw.awrap_model_call(FakeRequest(), todo_handler()) + assert mw._turn_count == 0 + + +async def test_write_todos_plus_another_tool_burns_a_turn(): + mw = make_mw(turn_limit=10) + await mw.awrap_model_call(FakeRequest(), todo_handler(("write_todos", "write_file"))) + assert mw._turn_count == 1 + + +async def test_text_only_close_out_burns_a_turn(): + """Refunding the closing turn would keep the ladder from ever reaching stage 3.""" + mw = make_mw(turn_limit=10) + await mw.awrap_model_call(FakeRequest(), todo_handler(())) + assert mw._turn_count == 1 + + +async def test_invalid_tool_calls_burn_a_turn(): + mw = make_mw(turn_limit=10) + await mw.awrap_model_call(FakeRequest(), todo_handler(("write_todos",), invalid=True)) + assert mw._turn_count == 1 + + +async def test_response_without_an_ai_message_burns_a_turn(): + mw = make_mw(turn_limit=10) + + async def handler(_request): + return SimpleNamespace(result=[]) + + await mw.awrap_model_call(FakeRequest(), handler) + assert mw._turn_count == 1 + + +async def test_refund_happens_once_across_transient_retries(): + """The refund sits at the loop's single success exit, so two retries cannot + turn one state-only turn into a -2 credit.""" + mw = make_mw(turn_limit=10) + handler = todo_handler(fail_times=2) + await mw.awrap_model_call(FakeRequest(), handler) + assert handler.state["calls"] == 3 + assert mw._turn_count == 0 + + +async def test_refund_happens_once_across_truncation_retries(): + mw = make_mw(turn_limit=10) + handler = todo_handler(truncate_first=True) + await mw.awrap_model_call(FakeRequest(), handler) + assert handler.state["calls"] == 2 # truncation nudge retried once + assert mw._turn_count == 0 + + +async def test_degraded_turn_is_not_refunded(): + """A degraded call really did burn model calls — it never reaches the refund.""" + mw = make_mw(turn_limit=10, is_subagent=True) + + async def handler(_request): + import openai + + exc = openai.BadRequestError.__new__(openai.BadRequestError) + exc.message = "content filter" + exc.code = "content_filter" + exc.body = None + raise exc + + await mw.awrap_model_call(FakeRequest(), handler) + assert mw._turn_count == 1 + + +async def test_refunds_are_capped(): + mw = make_mw(turn_limit=100) + for _ in range(_MAX_STATE_ONLY_REFUNDS + 5): + await mw.awrap_model_call(FakeRequest(), todo_handler()) + assert mw._turn_count == 5 + + +async def test_soft_landing_stage_is_chosen_before_the_refund(): + """Two-phase accounting: the stage is picked from the pre-call count (it has to + shape the request), and only afterwards is the turn given back.""" + mw = make_mw(turn_limit=3, soft_landing_turns=8) + handler = todo_handler() + await mw.awrap_model_call(FakeRequest(), handler) + assert tool_names(handler.state["requests"][0]) <= _DELIVERABLE_TOOLS + assert mw._turn_count == 0 + + +async def test_refunds_never_make_the_hard_stop_unreachable(): + """The cap is what keeps a write_todos loop from bypassing the ladder entirely + and dying on GraphRecursionError instead.""" + mw = make_mw(turn_limit=3, soft_landing_turns=8) + handler = todo_handler() + for _ in range(3 + _MAX_STATE_ONLY_REFUNDS + 1): + await mw.awrap_model_call(FakeRequest(), handler) + assert handler.state["requests"][-1].tools == [] + + +async def test_refunds_are_per_bucket(): + mw = make_mw(turn_limit=10, is_subagent=True) + handler = todo_handler() + for _ in range(2): + await mw.awrap_model_call(ns_request(NS_A), handler) + await mw.awrap_model_call(ns_request(NS_B), handler) + assert mw._turns_used("tools:a6dc6726-df24-8a53-049a-fbf4c35a1e9c") == 0 + assert mw._turns_used("tools:5923fee9-ce67-f1a7-a07f-265bbf188878") == 0 From e1b412b3c59494f29311bf08b6f5e3ab4c6ba022 Mon Sep 17 00:00:00 2001 From: LineWalker Date: Sun, 9 Aug 2026 14:53:54 +0800 Subject: [PATCH 2/6] =?UTF-8?q?fix(linsight):=20=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=E6=94=B6=E6=95=9B=EF=BC=8C=E5=B7=B2=E5=AE=8C?= =?UTF-8?q?=E6=88=90=E7=9A=84=E4=BC=9A=E8=AF=9D=E4=B8=8D=E5=86=8D=E6=98=BE?= =?UTF-8?q?=E7=A4=BA=E3=80=8C4/7=E3=80=8D=E5=81=87=E6=AF=94=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 180 POC 会话 aa352cb4 成功交付后仍显示「任务已完成 4/7」，库里还躺着 1 条 IN_PROGRESS + 2 条 NOT_STARTED，而会话状态是 COMPLETED。 linsight_execute_task 实际是 append-only 的「首次规划快照」： - 模型从计划里剪掉的 todo 不产生任何事件。mapper 的注释写着「消失的 todo 标记为 TERMINATED」，实现里却只有一行 `self.ctx.todos = new_projection` ——注释与实现不符； - 三条正常完成路径只 finalize svid 伪任务，从不收口真实 todo 行； - 位置对齐复用旧行 id 时不回写 task_data，标题永远停在第一版草稿。 三处修复： 1. `_set_tasks_failed` 更名 `_terminate_unfinished_tasks`（它写的是 TERMINATED，旧名字既没描述动作也没描述状态），三条完成路径经 `_converge_task_rows_on_completion` 接上。顺序是硬约束：必须在 `_complete_session_pseudo_task` 之后，否则 sweep 会把会话自己那行也标掉。 2. `_diff_todos` 为消失项产出 TaskEnd(status="terminated")，已完成的跳过 （剪掉计划不等于撤销已交付的工作）。**守门点**：`_handle_task_end` 里 `_final_result` 必须收窄为「非 TERMINATED 才赋值」——否则一条被剪掉的 todo 最后到达会经 `_handle_task_completion` 把整个会话判失败。 3. `_save_task_info` 对文案被改写的行刷新 task_data。 前端同步：后端改了前端也不会变——terminated 在 taskStatus 里既非 done 也非 running，渲染成和 not_started 一样的灰圈且仍进分母。TaskPanel 在 `completed && !terminated` 时过滤掉 terminated 行。用户终止/失败两条路径 的观感逐字节不变（有守门测试）。 另：recursion 地板跟随 write_todos 退款上调为 (turns+10)*4+20，并把 initdb_config 模板的 max_steps 提到 600 保持自洽——新增测试直接读模板断言 它不会自己触发「max_steps 低于所需」的告警。 Co-Authored-By: Claude Opus 5 (1M context) --- .../domain/services/stream_event_mapper.py | 50 +++- .../bisheng/linsight/domain/task_exec.py | 88 +++++- .../test_recursion_partial_classification.py | 34 ++- .../test_task_progress_convergence.py | 277 ++++++++++++++++++ .../Linsight/Execution/TaskPanel.test.tsx | 61 ++++ .../Linsight/Execution/TaskPanel.tsx | 19 +- 6 files changed, 500 insertions(+), 29 deletions(-) create mode 100644 src/backend/test/linsight/test_task_progress_convergence.py create mode 100644 src/frontend/client/src/components/Linsight/Execution/TaskPanel.test.tsx diff --git a/src/backend/bisheng/linsight/domain/services/stream_event_mapper.py b/src/backend/bisheng/linsight/domain/services/stream_event_mapper.py index 8b767d147c..21fb0d4314 100644 --- a/src/backend/bisheng/linsight/domain/services/stream_event_mapper.py +++ b/src/backend/bisheng/linsight/domain/services/stream_event_mapper.py @@ -80,6 +80,16 @@ def _stable_task_id(svid: str, content: str) -> str: return hashlib.md5(f"{svid}:{content}".encode()).hexdigest()[:8] +# Todo statuses that count as delivered. A todo dropped from the plan AFTER it was +# completed stays completed — pruning the plan does not un-deliver work. +_TODO_DONE_STATUSES = frozenset({"completed", "done"}) + +# Status carried by the TaskEnd emitted for a todo the model pruned. Matches +# ``ExecuteTaskStatusEnum.TERMINATED`` (imported as a literal to keep this pure +# mapper free of the persistence layer); ``task_exec._handle_task_end`` maps it back. +_TODO_DROPPED_STATUS = "terminated" + + def _truncate(text: str | None) -> tuple[str | None, bool]: if text is None: return None, False @@ -311,6 +321,7 @@ def _diff_todos(self, new_todos: list[dict[str, Any]]) -> list[BaseEvent]: used_old_idx: set[int] = set() new_projection: list[_TodoProjection] = [] newly_generated: list[_TodoProjection] = [] + renamed: list[_TodoProjection] = [] status_events: list[BaseEvent] = [] for pos, item in enumerate(new_todos): @@ -334,6 +345,12 @@ def _diff_todos(self, new_todos: list[dict[str, Any]]) -> list[BaseEvent]: if matched is not None: proj = _TodoProjection(task_id=matched.task_id, content=content, status=status) new_projection.append(proj) + if matched.content != content: + # Level-2 alignment reused an existing row id while the model + # REWROTE the wording. Re-announce it so the DB row's task_data + # (and therefore the panel after a refresh) follows the current + # plan instead of keeping the very first draft's title. + renamed.append(proj) status_events.extend(self._status_transition(matched.status, status, proj)) else: # Level 3: brand new @@ -346,15 +363,33 @@ def _diff_todos(self, new_todos: list[dict[str, Any]]) -> list[BaseEvent]: newly_generated.append(proj) status_events.extend(self._status_transition(None, status, proj)) - # Tasks that vanished from the new list are marked TERMINATED (not - # deleted) — projection drop is enough at the mapper layer. The todo - # projection still drives TaskPanel signals (GenerateSubTask / TaskStart / - # TaskEnd); it no longer needs an in_progress cursor because steps are no - # longer attributed to a todo (B2 段流重构 2026-06). + # Todos that vanished from the new snapshot: the model pruned them from its + # plan. Emit a terminal event so the DB row converges instead of sitting at + # NOT_STARTED/IN_PROGRESS forever — measured on 180, a COMPLETED session left + # 1 IN_PROGRESS + 2 NOT_STARTED behind and the panel was stuck at "4/7". + # Rows that already reached a done state are left alone: they were delivered, + # dropping them from the plan afterwards does not un-deliver them. + terminated_events: list[BaseEvent] = [ + TaskEnd( + task_id=o.task_id, + name=o.content, + status=_TODO_DROPPED_STATUS, + answer="", + # Non-empty data on purpose: _handle_task_end only refreshes + # ``task_data`` when the event carries some, so this also repairs the + # row's title on the way out. + data={"id": o.task_id, "task_id": o.task_id, "name": o.content, "status": _TODO_DROPPED_STATUS}, + ) + for i, o in enumerate(old) + if i not in used_old_idx and o.status not in _TODO_DONE_STATUSES + ] + # The todo projection drives TaskPanel signals (GenerateSubTask / TaskStart / + # TaskEnd); it needs no in_progress cursor because steps are no longer + # attributed to a todo (B2 段流重构 2026-06). self.ctx.todos = new_projection events: list[BaseEvent] = [] - if newly_generated: + if newly_generated or renamed: events.append( GenerateSubTask( task_id=self.ctx.svid, @@ -365,11 +400,12 @@ def _diff_todos(self, new_todos: list[dict[str, Any]]) -> list[BaseEvent]: "name": p.content, "status": p.status, } - for p in newly_generated + for p in (*newly_generated, *renamed) ], ) ) events.extend(status_events) + events.extend(terminated_events) return events def _status_transition(self, old_status: str | None, new_status: str, proj: _TodoProjection) -> list[BaseEvent]: diff --git a/src/backend/bisheng/linsight/domain/task_exec.py b/src/backend/bisheng/linsight/domain/task_exec.py index c2205dda99..4946e343e2 100644 --- a/src/backend/bisheng/linsight/domain/task_exec.py +++ b/src/backend/bisheng/linsight/domain/task_exec.py @@ -33,6 +33,7 @@ ) from bisheng.linsight.domain.services.agent_factory import _resolve_model, create_linsight_agent from bisheng.linsight.domain.services.binary_content_guard import CODE_INTERPRETER_TOOL +from bisheng.linsight.domain.services.resilience_middleware import _MAX_STATE_ONLY_REFUNDS from bisheng.linsight.domain.services.state_message_manager import ( LinsightStateMessageManager, MessageData, @@ -140,7 +141,12 @@ def _resolve_recursion_limit(linsight_conf) -> int: """ configured = int(getattr(linsight_conf, "max_steps", 500) or 0) turn_limit = int(getattr(linsight_conf, "max_model_turns", 115) or 0) - floor = turn_limit * _STEPS_PER_MODEL_TURN + _RECURSION_LIMIT_MARGIN + # Pure state-maintenance turns (a model call that only re-published the todo list) + # are refunded, up to ``_MAX_STATE_ONLY_REFUNDS`` per budget bucket, so a run may + # legitimately make more model calls than ``max_model_turns``. The fuse must still + # blow LATER than the budget, or the soft-landing ladder is bypassed and the run + # aborts into the partial-salvage path instead of finishing normally. + floor = (turn_limit + _MAX_STATE_ONLY_REFUNDS) * _STEPS_PER_MODEL_TURN + _RECURSION_LIMIT_MARGIN if configured >= floor: return configured logger.warning( @@ -980,6 +986,22 @@ async def _complete_session_pseudo_task(self, session_model: LinsightSessionVers except Exception as e: logger.warning(f"Failed to finalize session pseudo task: {e}") + async def _converge_task_rows_on_completion(self) -> None: + """Close out task rows a NORMALLY finished run left hanging. + + ``linsight_execute_task`` used to be append-only: rows were inserted from the + first ``write_todos`` snapshot and only ever updated on a status flip, so a run + that finished after the model reshaped its plan left rows stuck at + NOT_STARTED/IN_PROGRESS forever. The panel then reported a fake ratio on a + COMPLETED session — measured on 180: "任务已完成 4/7" with 1 IN_PROGRESS and 2 + NOT_STARTED still in the table. + + ⚠️ ORDERING: must run AFTER ``_complete_session_pseudo_task``. The sweep walks + every row including the svid pseudo task; finalizing that one first is what + makes the sweep skip it instead of marking the session's own row TERMINATED. + """ + await self._terminate_unfinished_tasks() + async def _save_task_info(self, session_model: LinsightSessionVersion, task_info: list[dict]): """Save Task Information. @@ -1021,6 +1043,24 @@ async def _save_task_info(self, session_model: LinsightSessionVersion, task_info if new_tasks: await LinsightExecuteTaskDao.batch_create_tasks(new_tasks) + # Positional alignment reuses an existing row id when the model REWRITES a + # todo's wording, so without this the row would keep the title from the + # very first plan for the rest of the run. Status/result are untouched. + for task in tasks: + prev = existing_by_id.get(task.id) + if prev is None: + continue + if (prev.task_data or {}).get("name") == (task.task_data or {}).get("name"): + continue + # DAO directly (like the batch insert above): the state-manager helper + # requires a ``status`` and returns a plain dict, both of which would + # be wrong here — the status must not move, and ``merged`` below has + # to stay a list of models. Redis is refreshed by the + # ``set_execution_tasks(merged)`` call a few lines down. + refreshed = await LinsightExecuteTaskDao.update_by_id(task.id, task_data=task.task_data) + if refreshed is not None: + existing_by_id[task.id] = refreshed + merged = [existing_by_id.get(t.id, t) for t in tasks] await self._state_manager.set_execution_tasks(merged) @@ -1413,9 +1453,18 @@ async def _handle_task_start(self, agent, event: TaskStart, session_model: Linsi async def _handle_task_end(self, agent, event: TaskEnd, session_model: LinsightSessionVersion): """Handle task end events""" - status = ( - ExecuteTaskStatusEnum.SUCCESS if event.status == TaskStatus.SUCCESS.value else ExecuteTaskStatusEnum.FAILED - ) + # A "terminated" TaskEnd is a todo the model DROPPED from its plan (see + # StreamEventMapper._diff_todos), not a task that failed. Mapping it to FAILED + # would paint an error row in the panel — and, far worse, feed + # ``self._final_result`` below, which ``_handle_task_completion`` routes into + # ``_handle_task_failure`` for anything != success: a pruned todo arriving last + # would fail the WHOLE session. + if event.status == ExecuteTaskStatusEnum.TERMINATED.value: + status = ExecuteTaskStatusEnum.TERMINATED + elif event.status == TaskStatus.SUCCESS.value: + status = ExecuteTaskStatusEnum.SUCCESS + else: + status = ExecuteTaskStatusEnum.FAILED # F035 fix: TaskEnd.data is frequently empty for deepagents tasks; passing # it as task_data would OVERWRITE the task_data stored at write_todos time @@ -1430,8 +1479,9 @@ async def _handle_task_end(self, agent, event: TaskEnd, session_model: LinsightS await self._state_manager.push_message(MessageData(event_type=MessageEventType.TASK_END, data=task_data)) - # Save Final Result - self._final_result = event + # Save Final Result — a dropped todo is never the run's outcome. + if status is not ExecuteTaskStatusEnum.TERMINATED: + self._final_result = event async def _handle_need_user_input(self, agent, event: NeedUserInput, session_model: LinsightSessionVersion): """Handle events that require user input (park-and-release, F035 §4.6). @@ -1547,8 +1597,8 @@ async def _handle_user_termination(self, session_model: LinsightSessionVersion): await self._state_manager.set_session_version_info(session_model) - # Set all tasks to failed - await self._set_tasks_failed() + # Converge every task row the run never finished + await self._terminate_unfinished_tasks() # Push termination message await self._state_manager.push_message( @@ -1680,6 +1730,7 @@ async def _handle_direct_answer_completion(self, session_model: LinsightSessionV # F035 problem 2: finalize the session pseudo task carrying any # planning/direct-answer steps so it isn't left stuck in_progress. await self._complete_session_pseudo_task(session_model) + await self._converge_task_rows_on_completion() # F035 Track J: land the answer in the unified conversation stream. await linsight_execute_utils.persist_task_turn_message(session_model) await self._state_manager.push_message( @@ -1774,6 +1825,7 @@ async def _handle_task_partial(self, session_model: LinsightSessionVersion): self._flag_phantom_deliverables(session_model, answer, final_files) await self._state_manager.set_session_version_info(session_model) await self._complete_session_pseudo_task(session_model) + await self._converge_task_rows_on_completion() await linsight_execute_utils.persist_task_turn_message(session_model) await self._state_manager.push_message( MessageData(event_type=MessageEventType.FINAL_RESULT, data=session_model.model_dump()) @@ -1823,6 +1875,7 @@ async def _handle_task_success(self, session_model: LinsightSessionVersion): # F035 problem 2: finalize the session pseudo task carrying any # planning/wrap-up steps so it isn't left stuck in_progress. await self._complete_session_pseudo_task(session_model) + await self._converge_task_rows_on_completion() # F035 Track J: land the answer in the unified conversation stream. await linsight_execute_utils.persist_task_turn_message(session_model) await self._state_manager.push_message( @@ -1836,8 +1889,17 @@ async def _handle_task_success(self, session_model: LinsightSessionVersion): raise TaskExecutionError(f"An error occurred while processing the task successfully: {e}") # Modify All Task Failure Processing Logic - async def _set_tasks_failed(self): - """Set all tasks to failed""" + async def _terminate_unfinished_tasks(self): + """Converge every non-terminal task row to TERMINATED. + + Named for what it DOES — the old name (``_set_tasks_failed``) described + neither the action nor the status it writes. Three callers, three meanings, + one action: user stop, task failure, and normal completion, where the model + simply stopped updating todos it never finished. + + Rows already SUCCESS / FAILED / TERMINATED are left alone, so this is + idempotent and can never downgrade a delivered result. + """ try: # Get All Execute Tasks execution_tasks = await self._state_manager.get_execution_tasks() @@ -1853,7 +1915,7 @@ async def _set_tasks_failed(self): task_id=task.id, status=ExecuteTaskStatusEnum.TERMINATED ) except Exception as e: - logger.warning(f"Error setting task failed: {e}") + logger.warning("Error converging unfinished task rows: {}", e) async def _handle_task_failure( self, session_model: LinsightSessionVersion, error_msg: str, *, exc: Exception | None = None @@ -1882,8 +1944,8 @@ async def _handle_task_failure( # whose detail panel shows the failure. await linsight_execute_utils.persist_task_turn_message(session_model) - # Set all tasks to failed - await self._set_tasks_failed() + # Converge every task row the run never finished + await self._terminate_unfinished_tasks() # error_message event: keep ``error`` for backward compatibility (old # clients display it raw); new fields drive the classified friendly card. diff --git a/src/backend/test/linsight/test_recursion_partial_classification.py b/src/backend/test/linsight/test_recursion_partial_classification.py index c123b77ee2..0254bf49b8 100644 --- a/src/backend/test/linsight/test_recursion_partial_classification.py +++ b/src/backend/test/linsight/test_recursion_partial_classification.py @@ -21,11 +21,14 @@ from __future__ import annotations +from pathlib import Path from unittest.mock import AsyncMock +import yaml from langchain_core.messages import AIMessage, HumanMessage, ToolMessage from langgraph.errors import GraphRecursionError +from bisheng.linsight.domain.services.resilience_middleware import _MAX_STATE_ONLY_REFUNDS from bisheng.linsight.domain.services.tool_loop_middleware import LinsightToolLoopError from bisheng.linsight.domain.task_exec import ( _PARTIAL_NO_SALVAGE_STEP_LIMIT, @@ -193,16 +196,29 @@ def __init__(self, max_steps: int, max_model_turns: int = 115) -> None: self.max_model_turns = max_model_turns +def _floor_for(turns: int) -> int: + return (turns + _MAX_STATE_ONLY_REFUNDS) * _STEPS_PER_MODEL_TURN + _RECURSION_LIMIT_MARGIN + + def test_legacy_db_value_is_raised_above_the_turn_budget(): """Existing installs keep ``max_steps: 200`` in the DB config, which would trip at ~50 turns and make the 115-turn budget unreachable.""" resolved = _resolve_recursion_limit(_Conf(max_steps=200)) - assert resolved == 115 * _STEPS_PER_MODEL_TURN + _RECURSION_LIMIT_MARGIN + assert resolved == _floor_for(115) assert resolved > 200 -def test_new_default_is_already_above_the_floor(): - assert _resolve_recursion_limit(_Conf(max_steps=500)) == 500 +def test_shipped_default_is_already_above_the_floor(): + """The value in ``initdb_config.yaml`` must not itself trip the auto-raise warning. + + Guards the coupling that broke once already: raising the floor (for the refund cap) + without bumping the shipped default would make EVERY install log the + "max_steps is below ..." warning on every task. + """ + shipped = yaml.safe_load( + (Path(__file__).resolve().parents[2] / "bisheng" / "initdb_config.yaml").read_text(encoding="utf-8") + )["linsight"]["max_steps"] + assert _resolve_recursion_limit(_Conf(max_steps=shipped)) == shipped def test_operator_raised_ceiling_is_respected(): @@ -211,4 +227,14 @@ def test_operator_raised_ceiling_is_respected(): def test_floor_tracks_a_raised_turn_budget(): resolved = _resolve_recursion_limit(_Conf(max_steps=500, max_model_turns=300)) - assert resolved == 300 * _STEPS_PER_MODEL_TURN + _RECURSION_LIMIT_MARGIN + assert resolved == _floor_for(300) + + +def test_recursion_floor_covers_the_refund_cap(): + """Refunded state-only turns let a run exceed ``max_model_turns`` model calls, so + the fuse has to sit above budget + cap — otherwise the ladder is bypassed and the + run aborts into partial salvage instead of landing softly.""" + floor = _resolve_recursion_limit(_Conf(max_steps=200)) + assert floor >= (115 + _MAX_STATE_ONLY_REFUNDS) * _STEPS_PER_MODEL_TURN + # Strictly above the pre-refund floor: that is the whole point of this change. + assert floor > 115 * _STEPS_PER_MODEL_TURN + _RECURSION_LIMIT_MARGIN diff --git a/src/backend/test/linsight/test_task_progress_convergence.py b/src/backend/test/linsight/test_task_progress_convergence.py new file mode 100644 index 0000000000..6d77debe2b --- /dev/null +++ b/src/backend/test/linsight/test_task_progress_convergence.py @@ -0,0 +1,277 @@ +"""The task table must converge, so the panel never reports a fake ratio. + +Background — session ``aa352cb4…`` (180 POC, v2.6.0-fix2, 2026-08-08): a run that +finished successfully showed "任务已完成 4/7". ``linsight_execute_task`` was +effectively append-only — rows were inserted from the FIRST ``write_todos`` +snapshot, then only ever touched on a status flip: + +- todos the model pruned from its plan produced no event at all (the mapper's own + comment claimed they were "marked TERMINATED", but the code only dropped them + from the in-memory projection), so their rows sat at ``not_started`` forever; +- none of the three normal completion paths swept leftovers, so the session ended + COMPLETED with 1 IN_PROGRESS + 2 NOT_STARTED still in the table; +- a rewritten todo kept the title from the very first draft. + +``asyncio_mode = auto`` — async tests need no decorator. +""" + +from __future__ import annotations + +import hashlib +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from bisheng.linsight.domain.models.linsight_execute_task import ( + ExecuteTaskStatusEnum, + ExecuteTaskTypeEnum, + LinsightExecuteTask, +) +from bisheng.linsight.domain.services.stream_event_mapper import StreamEventMapper +from bisheng.linsight.domain.task_exec import LinsightWorkflowTask +from bisheng_langchain.linsight.event import GenerateSubTask, TaskEnd, TaskStart + +SVID = "1f3c9a20-7b4e-4d11-9c3a-0a1b2c3d4e5f" + + +def _task_id(content: str) -> str: + return hashlib.md5(f"{SVID}:{content}".encode()).hexdigest()[:8] + + +@pytest.fixture +def mapper() -> StreamEventMapper: + return StreamEventMapper(svid=SVID) + + +def _todos(*pairs) -> list[dict]: + return [{"content": c, "status": s} for c, s in pairs] + + +def _feed(mapper: StreamEventMapper, todos: list[dict]): + return mapper._diff_todos(todos) + + +def _of(events, kind): + return [e for e in events if isinstance(e, kind)] + + +# -------------------------------------------------------------------------- +# Mapper: pruned todos produce a terminal event +# -------------------------------------------------------------------------- + + +def test_vanished_todo_emits_a_terminated_task_end(mapper): + _feed(mapper, _todos(("调研", "completed"), ("撰写", "in_progress"), ("配图", "pending"))) + events = _feed(mapper, _todos(("调研", "completed"), ("撰写", "completed"))) + + dropped = [e for e in _of(events, TaskEnd) if e.status == "terminated"] + assert len(dropped) == 1 + assert dropped[0].task_id == _task_id("配图") + # Non-empty data so _handle_task_end also repairs the row's title. + assert dropped[0].data["name"] == "配图" + + +def test_vanished_completed_todo_is_left_alone(mapper): + """Pruning the plan after delivery does not un-deliver the work.""" + _feed(mapper, _todos(("调研", "completed"), ("撰写", "completed"))) + events = _feed(mapper, _todos(("调研", "completed"))) + + assert [e for e in _of(events, TaskEnd) if e.status == "terminated"] == [] + + +def test_vanished_todo_leaves_the_projection(mapper): + _feed(mapper, _todos(("a", "pending"), ("b", "pending"), ("c", "pending"))) + _feed(mapper, _todos(("a", "pending"))) + assert [p.content for p in mapper.ctx.todos] == ["a"] + + +def test_reordering_terminates_nothing(mapper): + """Level-1 exact-content matching must survive a shuffle.""" + _feed(mapper, _todos(("a", "pending"), ("b", "pending"))) + events = _feed(mapper, _todos(("b", "pending"), ("a", "pending"))) + assert [e for e in _of(events, TaskEnd) if e.status == "terminated"] == [] + + +def test_first_snapshot_terminates_nothing(mapper): + events = _feed(mapper, _todos(("a", "pending"), ("b", "pending"))) + assert _of(events, TaskEnd) == [] + assert len(_of(events, GenerateSubTask)) == 1 + + +def test_rewritten_todo_is_republished_for_task_data_sync(mapper): + """Positional alignment keeps the row id, so the new wording has to be pushed + or the panel keeps showing the first draft's title after a refresh.""" + _feed( + mapper, + _todos( + ("写初稿", "in_progress"), + ), + ) + events = _feed( + mapper, + _todos( + ("写初稿并配图", "in_progress"), + ), + ) + + generated = _of(events, GenerateSubTask) + assert len(generated) == 1 + entry = generated[0].subtask[0] + assert entry["id"] == _task_id("写初稿") # id reused + assert entry["name"] == "写初稿并配图" # title refreshed + # Same status → no spurious start/end. + assert _of(events, TaskStart) == [] + + +# -------------------------------------------------------------------------- +# task_exec: a dropped todo must never become the run's outcome +# -------------------------------------------------------------------------- + + +def _exec_task() -> LinsightWorkflowTask: + task = LinsightWorkflowTask() + task.session_version_id = "svid" + sm = MagicMock() + sm.update_execution_task_status = AsyncMock(return_value={"id": "t1"}) + sm.push_message = AsyncMock() + task._state_manager = sm + return task + + +def _end(task_id: str, status: str, data: dict | None = None) -> TaskEnd: + return TaskEnd(task_id=task_id, name="n", status=status, answer="", data=data or {}) + + +async def test_terminated_task_end_writes_terminated_status(): + task = _exec_task() + await task._handle_task_end(None, _end("t1", "terminated", {"name": "配图"}), None) + + kwargs = task._state_manager.update_execution_task_status.await_args.kwargs + assert kwargs["status"] is ExecuteTaskStatusEnum.TERMINATED + assert kwargs["task_data"] == {"name": "配图"} + + +async def test_terminated_task_end_never_becomes_the_final_result(): + """Gate-keeper: ``_handle_task_completion`` routes any non-success + ``_final_result`` into ``_handle_task_failure``. A todo the model pruned + arriving last would therefore fail the WHOLE session.""" + task = _exec_task() + success = _end("t1", "success") + await task._handle_task_end(None, success, None) + await task._handle_task_end(None, _end("t2", "terminated"), None) + + assert task._final_result is success + + +async def test_success_task_end_still_sets_the_final_result(): + task = _exec_task() + success = _end("t1", "success") + await task._handle_task_end(None, success, None) + assert task._final_result is success + kwargs = task._state_manager.update_execution_task_status.await_args.kwargs + assert kwargs["status"] is ExecuteTaskStatusEnum.SUCCESS + + +async def test_failed_task_end_still_maps_to_failed(): + task = _exec_task() + await task._handle_task_end(None, _end("t1", "failed"), None) + kwargs = task._state_manager.update_execution_task_status.await_args.kwargs + assert kwargs["status"] is ExecuteTaskStatusEnum.FAILED + + +# -------------------------------------------------------------------------- +# task_exec: completion sweeps whatever the run left hanging +# -------------------------------------------------------------------------- + + +def _row(tid: str, status: ExecuteTaskStatusEnum, *, pseudo: bool = False) -> LinsightExecuteTask: + return LinsightExecuteTask( + id=tid, + session_version_id="svid", + parent_task_id=None, + task_type=ExecuteTaskTypeEnum.SINGLE, + status=status, + task_data={"name": tid, **({"is_session_global": True} if pseudo else {})}, + history=[], + ) + + +def _sweepable_task() -> LinsightWorkflowTask: + task = LinsightWorkflowTask() + task.session_version_id = "svid" + sm = MagicMock() + sm.get_execution_tasks = AsyncMock( + return_value=[ + _row("done", ExecuteTaskStatusEnum.SUCCESS), + _row("running", ExecuteTaskStatusEnum.IN_PROGRESS), + _row("never", ExecuteTaskStatusEnum.NOT_STARTED), + ] + ) + sm.update_execution_task_status = AsyncMock(return_value={}) + task._state_manager = sm + return task + + +def _swept(task) -> dict[str, ExecuteTaskStatusEnum]: + return { + call.kwargs["task_id"]: call.kwargs["status"] + for call in task._state_manager.update_execution_task_status.await_args_list + } + + +async def test_completion_sweep_converges_unfinished_rows(): + task = _sweepable_task() + await task._converge_task_rows_on_completion() + + swept = _swept(task) + assert swept == {"running": ExecuteTaskStatusEnum.TERMINATED, "never": ExecuteTaskStatusEnum.TERMINATED} + assert "done" not in swept # a delivered row is never downgraded + + +async def test_sweep_is_idempotent(): + task = LinsightWorkflowTask() + task.session_version_id = "svid" + sm = MagicMock() + sm.get_execution_tasks = AsyncMock( + return_value=[ + _row("a", ExecuteTaskStatusEnum.SUCCESS), + _row("b", ExecuteTaskStatusEnum.TERMINATED), + _row("c", ExecuteTaskStatusEnum.FAILED), + ] + ) + sm.update_execution_task_status = AsyncMock(return_value={}) + task._state_manager = sm + + await task._converge_task_rows_on_completion() + assert sm.update_execution_task_status.await_count == 0 + + +async def test_sweep_runs_after_the_pseudo_task_is_finalized(): + """ORDERING gate-keeper: the sweep walks every row including the svid pseudo + task, so finalizing that one first is what keeps the session's own row from + being marked TERMINATED on a successful run.""" + task = LinsightWorkflowTask() + task.session_version_id = "svid" + order: list[str] = [] + + sm = MagicMock() + + async def _update(task_id, status, **kwargs): + order.append(f"{task_id}:{status.value}") + return {} + + sm.update_execution_task_status = AsyncMock(side_effect=_update) + sm.get_execution_tasks = AsyncMock( + return_value=[ + _row("svid", ExecuteTaskStatusEnum.SUCCESS, pseudo=True), + _row("t1", ExecuteTaskStatusEnum.NOT_STARTED), + ] + ) + task._state_manager = sm + + session_model = MagicMock() + session_model.id = "svid" + await task._complete_session_pseudo_task(session_model) + await task._converge_task_rows_on_completion() + + assert order == ["svid:success", "t1:terminated"] diff --git a/src/frontend/client/src/components/Linsight/Execution/TaskPanel.test.tsx b/src/frontend/client/src/components/Linsight/Execution/TaskPanel.test.tsx new file mode 100644 index 0000000000..3ae38178e2 --- /dev/null +++ b/src/frontend/client/src/components/Linsight/Execution/TaskPanel.test.tsx @@ -0,0 +1,61 @@ +/** + * A finished run must not report a fake ratio. + * + * Session `aa352cb4…` (180 POC, 2026-08-08) rendered "任务已完成 4/7" on a run that + * had succeeded: the backend never converged `linsight_execute_task`, so three rows + * the model had pruned from its plan stayed in the list. The backend now sweeps them + * to `terminated`, but `terminated` is neither done nor running, so it used to render + * as a grey ring indistinguishable from `not_started` and still counted toward the + * denominator. This pins the split: hide them on a normally-completed run, keep every + * row when the user stopped the run. + */ +import { render, screen } from '@testing-library/react'; +import { TaskPanel } from './TaskPanel'; + +jest.mock('~/hooks', () => ({ + useLocalize: () => (key: string) => key, +})); + +jest.mock('bisheng-icons', () => ({ + Outlined: new Proxy( + {}, + { + get: () => () => null, + }, + ), +})); + +const task = (id: string, status: string) => ({ id, name: id, status }) as never; + +const FOUR_DONE = [ + task('a', 'success'), + task('b', 'success'), + task('c', 'success'), + task('d', 'success'), +]; +const THREE_PRUNED = [task('e', 'terminated'), task('f', 'terminated'), task('g', 'terminated')]; + +describe('TaskPanel progress ratio', () => { + it('hides pruned rows on a normally completed run', () => { + render(); + expect(screen.getByText('4/4')).toBeInTheDocument(); + expect(screen.queryByText('4/7')).not.toBeInTheDocument(); + }); + + it('keeps every row when the user stopped the run', () => { + // Gate-keeper: the stop path renders exactly as it did before this change. + render(); + expect(screen.getByText('4/7')).toBeInTheDocument(); + }); + + it('keeps not-started rows while the run is still going', () => { + const running = [...FOUR_DONE, task('e', 'in_progress'), task('f', 'not_started')]; + render(); + expect(screen.getByText('4/6')).toBeInTheDocument(); + }); + + it('renders nothing when every row was pruned', () => { + const { container } = render(); + expect(container).toBeEmptyDOMElement(); + }); +}); diff --git a/src/frontend/client/src/components/Linsight/Execution/TaskPanel.tsx b/src/frontend/client/src/components/Linsight/Execution/TaskPanel.tsx index a56819c46a..0181f9445c 100644 --- a/src/frontend/client/src/components/Linsight/Execution/TaskPanel.tsx +++ b/src/frontend/client/src/components/Linsight/Execution/TaskPanel.tsx @@ -30,15 +30,24 @@ export function TaskPanel({ if (!tasks.length) return null; - const doneCount = tasks.filter((t) => isTaskDone(t.status)).length; - const allDone = completed || doneCount === tasks.length; + // On a NORMALLY finished run, `terminated` rows are todos the model pruned from + // its plan (or leftovers the backend swept at completion). They were never + // delivered, so counting them inflated the denominator — a finished run showed + // "4/7" with three grey rings that would never fill. On a STOPPED or FAILED run + // every unreached row is `terminated` too; there the list must stay intact, so + // those paths render byte-identically to before. + const visible = completed && !terminated ? tasks.filter((t) => t.status !== 'terminated') : tasks; + if (!visible.length) return null; + + const doneCount = visible.filter((t) => isTaskDone(t.status)).length; + const allDone = completed || doneCount === visible.length; // Collapsed header surfaces the currently-running task name inline (Figma // 12221-40080): `≣ 任务 N/M ⌃`. While expanded the list // below already shows every task, so the header omits the inline name. // A terminated run is no longer "running", so it shows neither the spinner // nor the shimmering task name. - const runningTask = terminated ? undefined : tasks.find((t) => isTaskRunning(t.status)); + const runningTask = terminated ? undefined : visible.find((t) => isTaskRunning(t.status)); const runningName = runningTask?.name || runningTask?.task_data?.name || ''; const showRunningInline = !open && !allDone && !!runningName; @@ -81,7 +90,7 @@ export function TaskPanel({ className={cn('shrink-0 text-[14px] tabular-nums', showRunningInline ? 'ml-2' : 'ml-1')} style={{ color: MUTED }} > - {doneCount}/{tasks.length} + {doneCount}/{visible.length}
    - {tasks.map((task) => { + {visible.map((task) => { const done = isTaskDone(task.status); const running = isTaskRunning(task.status); return ( From 452e3bbac4fb6968f0604175d66ead08aaf5202a Mon Sep 17 00:00:00 2001 From: LineWalker Date: Sun, 9 Aug 2026 14:54:52 +0800 Subject: [PATCH 3/6] =?UTF-8?q?fix(linsight):=20=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E5=B7=A5=E5=85=B7=E4=B8=8E=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=E6=89=A7=E8=A1=8C=E5=99=A8=E7=9A=84=E4=B8=A4=E5=A5=97=E8=B7=AF?= =?UTF-8?q?=E5=BE=84=E5=91=BD=E5=90=8D=E7=A9=BA=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 文件工具回显带前导斜杠的 /output/x，而代码执行器的 cwd 就是同一个工作区根， 盘上是 output/x。没人说过这件事，180 POC 会话为此付了两次代价： 正向 open("/skills/…") 直接 FileNotFoundError，花 3 次模型往返才摸清； 反向把执行器里看到的宿主路径喂给 read_file，前导斜杠被无脑剥掉后拼成 workspace//root/.cache/… 这种废 key，收尾连续 4 次读图失败。 三件事： 1. 提示词：新增 path_namespace_rules() 供两个执行器共用，并把中文版说明注入 主 system prompt。门控从「skills_present AND has_code_interpreter」双真 改为只看 has_code_interpreter——原来那句唯一的解释藏在技能分支里，没选技能 的会话（正是这次）根本看不到。保留 has_code_interpreter 门控是因为本模块 处处遵守 prompt↔tool lockstep，未绑工具时不得提及工具名（有测试断言）。 e2b 用 include_skills=False：沙箱 copy-in 快照发生在 skills 落地之前， 那里确实没有 skills/，连举例也不能写。 2. normalize_workspace_path 增加可选 file_dir，识别执行器宿主路径前缀。 收益不止「报错更友好」：read 会先查本地缓存 Path(file_dir)/rel，而执行器 就写在那儿，所以折算之后那 4 次读图会直接**读成功**。 只做可证明无歧义的剥离：要求完整 file_dir 前缀且带 / 边界。曾实现过 「同父目录的兄弟任务目录也剥」，被自己的对抗用例证伪（前缀相似的兄弟被 吞掉）后删除——改写路径必须无歧义，猜测只能进错误文案，不能进选文件。 3. advisory 区分读写：/skills、/uploads 是**读**区，open 抛 FileNotFoundError 而已，什么都没丢，沿用写侧「文件被 DISCARDED」的措辞会让模型去追一个根本 不存在的数据丢失问题。并且必须挂到失败返回路径上——绝对路径读必然 exitcode≠0 提前 return，只扩正则会完全失效。 Co-Authored-By: Claude Opus 5 (1M context) --- .../linsight/domain/services/agent_factory.py | 35 ++- .../domain/services/workspace_backend.py | 112 ++++++++-- .../tools/code_interpreter/base_executor.py | 83 ++++++- .../tools/code_interpreter/e2b_executor.py | 7 +- .../tools/code_interpreter/local_executor.py | 18 +- .../test_code_interpreter_output_path.py | 102 +++++++++ .../linsight/test_workspace_path_namespace.py | 211 ++++++++++++++++++ 7 files changed, 531 insertions(+), 37 deletions(-) create mode 100644 src/backend/test/linsight/test_workspace_path_namespace.py diff --git a/src/backend/bisheng/linsight/domain/services/agent_factory.py b/src/backend/bisheng/linsight/domain/services/agent_factory.py index 29a06829d5..2469151745 100644 --- a/src/backend/bisheng/linsight/domain/services/agent_factory.py +++ b/src/backend/bisheng/linsight/domain/services/agent_factory.py @@ -157,12 +157,12 @@ - ask_user(reason, questions)：第 0 步澄清；整个会话最多调用一次。 - write_todos(todos)：维护有编号的待办清单；只翻转 status，不改写已有文案。 -__KB_TOOL_LINE__- write_file / read_file / edit_file / ls：工作区文件工具；交付物写 output/，中间产物写 scratch/。 +__KB_TOOL_LINE__- write_file / read_file / edit_file / ls：工作区文件工具；交付物写 output/，中间产物写 scratch/（工具回显的 /output/x 与 output/x 是同一个文件）。 - export_docx(source_path, dest_path)：把 output/ 下的 markdown 转 Word(.docx)，必须在对应 .md 写好之后。 - export_pdf(source_path, dest_path)：把 output/ 下的 markdown 转 PDF，必须在对应 .md 写好之后。 - task(description, subagent_type="general-purpose")：把独立、可隔离、较重的调研子任务委派给子代理。description 必须自包含——子代理看不到你的对话历史与上下文，只能读到 description，因此完成该子任务所需的全部背景、目标、约束与必要标识都要写进去；不得委派最终交付物撰写，也不得委派“问用户/澄清”。 -# 风格 +__PATH_NAMESPACE_LINE__# 风格 - 简洁，工具调用之间不加多余解释性文字。 - 不要凭空编造事实；交付内容应基于检索到的资料/依据。 @@ -320,10 +320,7 @@ def _build_linsight_system_prompt( "此时**禁止**调用 export_docx / export_pdf 从 markdown 派生同一份交付物——" "那条路径丢掉技能规定的全部排版细节，等于没用技能；" "只有代码执行连续失败、确实产不出成品时，才允许退回 export_docx / export_pdf 兜底，" - "并在收尾里说明退化。3a 的 markdown 规范源仍然要写（界面预览依赖它）。" - "注意：代码执行器直接写在工作区本地目录，它生成的文件**不会**出现在 ls / glob 的结果里" - "（那两个工具读的是对象存储视图）。只要执行返回 exitcode 0 且日志显示写成功，" - "就视为交付物已产出，继续下一步；不要反复 ls / glob 找它，也不要因为“找不到”而重新生成。\n" + "并在收尾里说明退化。3a 的 markdown 规范源仍然要写（界面预览依赖它）。\n" ) else: # No executor bound: the skill's route cannot run, so banning the @@ -337,12 +334,38 @@ def _build_linsight_system_prompt( else: skill_exec_line = "" skill_deliverable_line = "" + + # The two path namespaces only need explaining when a code interpreter is bound — + # that is the ONLY tool with a cwd. Gated on ``has_code_interpreter`` ALONE: + # this used to ride along inside ``skill_deliverable_line``, so a run that bound + # the executor but selected no skill (exactly the 180 session) never saw it and + # burned three model round-trips rediscovering the mapping. Keeping the + # ``has_code_interpreter`` gate preserves the prompt↔tool lockstep this module + # enforces everywhere else (a bound-tool name must never appear otherwise). + path_namespace_line = "" + if has_code_interpreter: + path_namespace_line = ( + "# 工作区路径（同一份文件的两种写法）\n\n" + "- 文件工具（ls / read_file / write_file / edit_file / export_*）用**带前导斜杠**的工作区路径：" + "/output/x.md、/scratch/x.png、/uploads/x.xlsx、/skills//SKILL.md；不带斜杠的相对写法等价。\n" + "- bisheng_code_interpreter 的**当前工作目录就是工作区根**，代码里一律用**去掉前导斜杠**的相对路径：" + 'open("skills//assets/t.html")、plt.savefig("output/chart.png")。' + '写成 open("/skills/…") / open("/output/…") 会落到容器根目录——读必然 FileNotFoundError，' + "写会被丢弃、进不了交付。\n" + "- 反向同理：**不要**把代码里看到的宿主机绝对路径（形如 /root/.cache/…/<8位任务号>/output/a.png）" + "传给 read_file / edit_file，去掉前缀只传 output/a.png。\n" + "- 执行器直接写本地工作目录，它生成的文件**不会**出现在 ls / glob 的结果里" + "（那两个工具读的是对象存储视图）。只要执行返回 exitcode 0 且日志显示写成功，" + "就视为交付物已产出，继续下一步；不要反复 ls / glob 找它，也不要因为“找不到”而重新生成。\n\n" + ) + return ( _LINSIGHT_SYSTEM_PROMPT_TEMPLATE_ZH.replace("__KB_EXEC_LINE__", exec_line) .replace("__KB_TOOL_LINE__", tool_line) .replace("__KB_DELEGATE_LINE__", delegate_line) .replace("__SKILL_EXEC_LINE__", skill_exec_line) .replace("__SKILL_DELIVERABLE_LINE__", skill_deliverable_line) + .replace("__PATH_NAMESPACE_LINE__", path_namespace_line) ) diff --git a/src/backend/bisheng/linsight/domain/services/workspace_backend.py b/src/backend/bisheng/linsight/domain/services/workspace_backend.py index 22740e618d..6d7f4aee79 100644 --- a/src/backend/bisheng/linsight/domain/services/workspace_backend.py +++ b/src/backend/bisheng/linsight/domain/services/workspace_backend.py @@ -160,14 +160,57 @@ def to_dict(self) -> dict: } -def normalize_workspace_path(path: str) -> str: +# First segments that only ever appear in a container/host path, never as a +# workspace zone. Used ONLY to word the error message — never to pick a file. +_HOST_ROOT_HINTS = frozenset({"root", "home", "tmp", "var", "usr", "opt", "mnt", "media", "Users", "app"}) + + +def strip_executor_host_prefix(path: str, file_dir: str | None) -> tuple[str, bool]: + """Drop the code interpreter's host-directory prefix; return ``(path, stripped)``. + + The interpreter's cwd IS this session's workspace cache dir, so the model routinely + reports host paths like ``/root/.cache/bisheng/linsight//output/qa/s08.png`` + and then hands one back to ``read_file``. Without this the leading slash was simply + dropped, producing the nonsense key ``workspace//root/.cache/...`` and a bare + "File not found" that never hinted at the real problem. + + ONE rule, deliberately: the full ``file_dir`` followed by ``/`` (or an exact + match). Never a mere ancestor segment, so a workspace that genuinely contains + ``root/report.md`` keeps it, and a near-miss sibling like ``/tmp/ws/x/...`` + is left alone. + + A "strip any sibling task dir under the same parent" rule was considered — it + would also catch the model quoting a path from an EARLIER turn's log — and + rejected: it cannot distinguish a real sibling task dir from a directory that + merely shares the parent, so it would silently reinterpret paths it has no + business touching. Rewriting a path must be provably unambiguous; opening the + wrong file is worse than an error message. Guessing belongs in the error text + (``WorkspaceBackend._not_found_error``), never in file selection. + """ + if not file_dir or not path.startswith("/"): + return path, False + root = os.path.normpath(file_dir) + if path == root: + return "", True + if path.startswith(root + "/"): + return path[len(root) + 1 :], True + return path, False + + +def normalize_workspace_path(path: str, file_dir: str | None = None) -> str: """Normalize a workspace path to a relative key; reject ``..`` traversal. Accepts both absolute (``/output/a.md``) and relative (``output/a.md``) forms and returns a clean relative key (``output/a.md``). Raises ``ValueError`` on any ``..`` segment so a model/tool cannot escape the session workspace. + + ``file_dir`` is optional so this stays a plain function other modules can call + (and test) without a backend; when given, a code-interpreter host path is folded + back to its workspace key first. Traversal is validated AFTER stripping, so + ``/../../etc/passwd`` still raises. """ - p = (path or "").strip().lstrip("/") + stripped, _ = strip_executor_host_prefix((path or "").strip(), file_dir) + p = stripped.lstrip("/") parts: list[str] = [] for seg in p.split("/"): if seg in ("", "."): @@ -357,6 +400,17 @@ def __init__(self, svid: str, minio, file_dir: str) -> None: os.makedirs(self.file_dir, exist_ok=True) # -- key / cache helpers ------------------------------------------------ + def _ws_rel(self, path: str) -> str: + """``normalize_workspace_path`` bound to this session's executor cache dir. + + Every tool entry point goes through here so a host path pasted back from the + code interpreter resolves instead of turning into a bogus key. + """ + rel = normalize_workspace_path(path, file_dir=self.file_dir) + if rel != (path or "").strip().lstrip("/"): + logger.info("workspace path: folded executor host path {} -> {}", path, rel) + return rel + def _object_key(self, rel_path: str) -> str: """Map a workspace-relative path to its MinIO object key.""" return f"{WORKSPACE_PREFIX}/{self.svid}/{rel_path}" @@ -364,6 +418,32 @@ def _object_key(self, rel_path: str) -> str: def _cache_path(self, rel_path: str) -> Path: return Path(self.file_dir) / rel_path + def _not_found_error(self, file_path: str) -> str: + """ "File not found" plus, when warranted, WHY the path could not resolve. + + A bare "File 'X' not found" was actively unhelpful for the most common + failure: handing back a code-interpreter host path. Naming the two + namespaces is what lets the model fix it in one step instead of retrying + the same path. + """ + _, stripped = strip_executor_host_prefix((file_path or "").strip(), self.file_dir) + if stripped: + rel = normalize_workspace_path(file_path, file_dir=self.file_dir) + return ( + f"File '{file_path}' not found. (Interpreted as workspace path '{rel}' — the code " + f"interpreter's working directory IS the workspace root.) Nothing exists there; " + f"call ls to see what the workspace holds." + ) + head = (file_path or "").strip().lstrip("/").split("/", 1)[0] + if file_path.startswith("/") and head in _HOST_ROOT_HINTS: + return ( + f"File '{file_path}' not found. NOTE: this looks like a path on the code " + f"interpreter's HOST filesystem. The file tools take WORKSPACE paths only — the " + f"interpreter's working directory IS the workspace root, so its 'output/a.png' is " + f"'/output/a.png' here. Retry with the workspace path, or call ls to list it." + ) + return f"File '{file_path}' not found" + def _bucket(self) -> str: return self.minio.bucket @@ -425,7 +505,7 @@ def _to_bytes(content) -> bytes: # -- write -------------------------------------------------------------- def write(self, file_path: str, content) -> WriteResult: - rel = normalize_workspace_path(file_path) + rel = self._ws_rel(file_path) data = self._to_bytes(content) # cache first (fast local), then write-through to MinIO (truth). self._cache_write(rel, data) @@ -434,10 +514,10 @@ def write(self, file_path: str, content) -> WriteResult: # -- read --------------------------------------------------------------- def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult: - rel = normalize_workspace_path(file_path) + rel = self._ws_rel(file_path) data = self._materialize(rel) if data is None: - return ReadResult(error=f"File '{file_path}' not found") + return ReadResult(error=self._not_found_error(file_path)) text = _decode_workspace_text(data, rel) if text is None: # Binary: offset/limit are meaningless here (slicing bytes by "lines" @@ -451,7 +531,7 @@ def read(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult # -- ls (authoritative from MinIO) -------------------------------------- def ls(self, path: str = "") -> LsResult: - rel_prefix = normalize_workspace_path(path) if path else "" + rel_prefix = self._ws_rel(path) if path else "" object_prefix = f"{WORKSPACE_PREFIX}/{self.svid}/" if rel_prefix: object_prefix += rel_prefix @@ -486,10 +566,10 @@ def edit( new_string: str, replace_all: bool = False, ) -> EditResult: - rel = normalize_workspace_path(file_path) + rel = self._ws_rel(file_path) data = self._materialize(rel) if data is None: - return EditResult(error=f"File '{file_path}' not found") + return EditResult(error=self._not_found_error(file_path)) text = _decode_workspace_text(data, rel) if text is None: # Refuse BEFORE any write. A replace-decode here would turn every @@ -523,7 +603,7 @@ def edit( def glob(self, pattern: str, path: str | None = None) -> GlobResult: import fnmatch - base = normalize_workspace_path(path) if path else "" + base = self._ws_rel(path) if path else "" ls_res = self.ls(base) if ls_res.error is not None: return GlobResult(error=ls_res.error) @@ -540,7 +620,7 @@ def glob(self, pattern: str, path: str | None = None) -> GlobResult: def grep(self, pattern: str, path: str | None = None, glob: str | None = None) -> GrepResult: from deepagents.backends.protocol import GrepMatch - base = normalize_workspace_path(path) if path else "" + base = self._ws_rel(path) if path else "" ls_res = self.ls(base) if ls_res.error is not None: return GrepResult(error=ls_res.error) @@ -586,7 +666,7 @@ def upload_files(self, files: list[tuple[str, bytes]]) -> list: responses: list = [] for raw_path, content in files: try: - rel = normalize_workspace_path(raw_path) + rel = self._ws_rel(raw_path) data = self._to_bytes(content) self._cache_write(rel, data) self._minio_put_sync(rel, data) @@ -602,7 +682,7 @@ def download_files(self, paths: list[str]) -> list: responses: list = [] for raw_path in paths: try: - rel = normalize_workspace_path(raw_path) + rel = self._ws_rel(raw_path) except ValueError: responses.append(FileDownloadResponse(path=raw_path, error="invalid_path")) continue @@ -615,7 +695,7 @@ def download_files(self, paths: list[str]) -> list: # -- async surface (write-through truth uses real async MinIO) ---------- async def awrite(self, file_path: str, content) -> WriteResult: - rel = normalize_workspace_path(file_path) + rel = self._ws_rel(file_path) data = self._to_bytes(content) await asyncio.to_thread(self._cache_write, rel, data) await self.minio.put_object( @@ -626,7 +706,7 @@ async def awrite(self, file_path: str, content) -> WriteResult: return WriteResult(path="/" + rel) async def aread(self, file_path: str, offset: int = 0, limit: int = 2000) -> ReadResult: - rel = normalize_workspace_path(file_path) + rel = self._ws_rel(file_path) data = await asyncio.to_thread(self._cache_read, rel) if data is None: try: @@ -646,7 +726,7 @@ async def aread(self, file_path: str, offset: int = 0, limit: int = 2000) -> Rea if data is not None: await asyncio.to_thread(self._cache_write, rel, data) if data is None: - return ReadResult(error=f"File '{file_path}' not found") + return ReadResult(error=self._not_found_error(file_path)) text = _decode_workspace_text(data, rel) if text is None: # Same contract as the sync path: binary is decided whole, never sliced. @@ -689,7 +769,7 @@ def ensure_local(self, file_path: str) -> str | None: cross-turn seed, has to be materialized before the tool list is built or it is invisible to Python code even though ``ls`` shows it. """ - rel = normalize_workspace_path(file_path) + rel = self._ws_rel(file_path) data = self._materialize(rel) if data is None: return None diff --git a/src/backend/bisheng_langchain/gpts/tools/code_interpreter/base_executor.py b/src/backend/bisheng_langchain/gpts/tools/code_interpreter/base_executor.py index 2403302608..3b7bcbc602 100644 --- a/src/backend/bisheng_langchain/gpts/tools/code_interpreter/base_executor.py +++ b/src/backend/bisheng_langchain/gpts/tools/code_interpreter/base_executor.py @@ -17,12 +17,31 @@ # executor can append a corrective notice and the model self-corrects next step. _ABSOLUTE_DELIVERABLE_RE = re.compile(r"""['"]/(?:output|scratch)(?:/|['"])""") +# The read-side twin. ``/skills`` and ``/uploads`` are zones the model READS: the +# workspace tools hand it ``/skills//SKILL.md`` and ``/uploads/``, and +# copying those into code prefixes the container root — ``open()`` then raises +# FileNotFoundError. Nothing is lost, so the write-side wording ("DISCARDED") would +# be actively misleading here; hence a separate pattern and a separate notice. +_ABSOLUTE_PROVISIONED_RE = re.compile(r"""['"]/(?:skills|uploads)(?:/|['"])""") + ABSOLUTE_PATH_NOTICE = ( "\n\n[SYSTEM NOTICE] Your code wrote file(s) to an ABSOLUTE path " "(/output/... or /scratch/...). Files written outside the current working " "directory are DISCARDED and were NOT delivered to the user. Re-run and write " "to the RELATIVE path with no leading slash, e.g. `output/report.pdf` for " - "deliverables or `scratch/temp.png` for intermediate files." + "deliverables or `scratch/temp.png` for intermediate files. The working " + "directory IS the workspace root, so the file tools' `/output/x` is simply " + "`output/x` here." +) + +ABSOLUTE_PROVISIONED_PATH_NOTICE = ( + "\n\n[SYSTEM NOTICE] Your code opened an ABSOLUTE workspace path " + "(/skills/... or /uploads/...). Those leading-slash paths exist only in the FILE " + "TOOLS' view; on this filesystem they live under the CURRENT WORKING DIRECTORY, " + "so a leading slash sends open() to the container root and raises " + "FileNotFoundError. Nothing was lost or discarded — the file simply was not " + "read. Re-run with the same path minus the leading slash, e.g. " + "`skills//SKILL.md` or `uploads/`." ) # Delivery zones of the executor working dir. ``output/`` is the ONLY zone the @@ -31,6 +50,39 @@ OUTPUT_DIR_NAME = "output" SCRATCH_DIR_NAME = "scratch" + +def path_namespace_rules(include_skills: bool = True) -> str: + """The mapping between the two path namespaces the model is shown. + + The file tools (ls / read_file / write_file / edit_file) render every workspace + path with a LEADING SLASH, while the interpreter's cwd IS that same workspace + root — so `/output/a.md` and `output/a.md` are one file, and the model has to + translate in both directions. Nothing said so until now, and a real run burned + three model round-trips discovering it, then failed four `read_file` calls at the + end by handing back a host path it had seen in the interpreter. + + ``include_skills`` is False for E2B: the sandbox copy-in snapshots the working dir + BEFORE skills are materialised, so `skills/` genuinely is not there and promising + it would just point the model at nothing. + """ + zones = "`/output/x` is `output/x`, `/scratch/x` is `scratch/x`, `/uploads/x` is `uploads/x`" + # Every example has to stay inside the zone set this executor actually has, or + # the guidance points at something that is not there. + read_example = "`open('/skills/...')`" if include_skills else "`open('/uploads/...')`" + if include_skills: + zones += ", `/skills//SKILL.md` is `skills//SKILL.md`" + return ( + "WORKING DIRECTORY: your cwd IS the task workspace root. The file tools show " + f"those same files with a LEADING SLASH: {zones}. " + f"In code ALWAYS drop the leading slash — {read_example} or " + "`open('/output/...')` resolves at the CONTAINER ROOT, so reads raise " + "FileNotFoundError and writes are discarded and never delivered. Conversely, " + "never hand a host path you saw here (e.g. `/root/.cache/.../output/a.png`) " + "back to read_file — pass the workspace path `output/a.png`. Create the zone " + "dirs before writing into them: `os.makedirs('output', exist_ok=True)`. " + ) + + # A file created at the working-directory ROOT sits in no zone at all, so it was # never delivered — the model just wrote ``report.xlsx`` instead of # ``output/report.xlsx``. The tool description asks for ``output/`` but that is a @@ -59,17 +111,28 @@ def run(self, code: str) -> Any: @staticmethod def absolute_path_advisory(code: str) -> str: - """Corrective notice to append when ``code`` writes to an absolute - ``/output``/``/scratch`` path (which escapes the harvested working dir and - makes the deliverable silently vanish); empty string otherwise. + """Corrective notice for absolute workspace paths in ``code``; "" if clean. + + Two independent failure modes, two notices, because the consequences differ: - String-literal match only (leading-slash ``/output`` / ``/scratch``), which - is specific enough that false positives are negligible, and the notice is - non-blocking (appended to the tool result, never rejects the run). + - WRITE side (``/output`` / ``/scratch``): the file lands outside the + harvested working dir and silently vanishes from the result panel. + - READ side (``/skills`` / ``/uploads``): ``open()`` raises FileNotFoundError + and nothing is lost — telling the model its file was "DISCARDED" there + would send it chasing a data-loss problem that never happened. + + Both hit at once → both notices, write side first. String-literal match only, + which is specific enough that false positives are negligible, and the notice + is non-blocking (appended to the tool result, never rejects the run). """ - if code and _ABSOLUTE_DELIVERABLE_RE.search(code): - return ABSOLUTE_PATH_NOTICE - return "" + if not code: + return "" + notices = "" + if _ABSOLUTE_DELIVERABLE_RE.search(code): + notices += ABSOLUTE_PATH_NOTICE + if _ABSOLUTE_PROVISIONED_RE.search(code): + notices += ABSOLUTE_PROVISIONED_PATH_NOTICE + return notices @staticmethod def relocation_advisory(moved: list[tuple[str, str]]) -> str: diff --git a/src/backend/bisheng_langchain/gpts/tools/code_interpreter/e2b_executor.py b/src/backend/bisheng_langchain/gpts/tools/code_interpreter/e2b_executor.py index 674fb18e35..b97e7ced3f 100644 --- a/src/backend/bisheng_langchain/gpts/tools/code_interpreter/e2b_executor.py +++ b/src/backend/bisheng_langchain/gpts/tools/code_interpreter/e2b_executor.py @@ -8,7 +8,7 @@ from e2b_code_interpreter import Result, Sandbox from loguru import logger -from bisheng_langchain.gpts.tools.code_interpreter.base_executor import BaseExecutor +from bisheng_langchain.gpts.tools.code_interpreter.base_executor import BaseExecutor, path_namespace_rules # F035 TC-4: copy-in/copy-out thresholds. The sandbox cannot reach MinIO, so the # worker mediates all file transfer (design §9.3.9). @@ -77,7 +77,10 @@ def description(self) -> str: "Write final deliverables to the RELATIVE directory `output/` (e.g. `output/report.pdf`) " "and intermediate files to `scratch/`. NEVER use an absolute path with a leading slash " "such as `/output/...` or `/scratch/...` — files written outside the current working " - "directory are DISCARDED and will NOT be delivered to the user." + "directory are DISCARDED and will NOT be delivered to the user. " + # include_skills=False: the sandbox copy-in snapshots the working dir + # before skills are materialised, so `skills/` is genuinely absent here. + + path_namespace_rules(include_skills=False) ) def init_sandbox(self): diff --git a/src/backend/bisheng_langchain/gpts/tools/code_interpreter/local_executor.py b/src/backend/bisheng_langchain/gpts/tools/code_interpreter/local_executor.py index 5ffce89395..1067b337c7 100644 --- a/src/backend/bisheng_langchain/gpts/tools/code_interpreter/local_executor.py +++ b/src/backend/bisheng_langchain/gpts/tools/code_interpreter/local_executor.py @@ -18,6 +18,7 @@ from bisheng_langchain.gpts.tools.code_interpreter.base_executor import ( OUTPUT_DIR_NAME, BaseExecutor, + path_namespace_rules, ) CODE_BLOCK_PATTERN = r"```(\w*)\n(.*?)\n```" @@ -39,7 +40,8 @@ LOG_TRUNCATED_NOTICE = "[... earlier output truncated ...]\n" PARTIAL_OUTPUT_HEADER = "\nOutput captured before the kill:\n" -LOCAL_DESCRIPTION = """Evaluates python code in native environment. \ +LOCAL_DESCRIPTION = ( + """Evaluates python code in native environment. \ You must send the whole script every time and print your outputs. \ Script should be pure python code that can be evaluated. \ It should be in python format NOT markdown. \ @@ -49,15 +51,19 @@ are subfolders of the current working directory. NEVER use an absolute path with a \ leading slash such as `/output/...` or `/scratch/...` — anything written outside the \ current working directory is DISCARDED and will NOT be delivered to the user. \ +""" + + path_namespace_rules(include_skills=True) + + """\ Do not use things like plot.show() as it will not work; save figures to `output/` \ instead. print() any output and results so you can capture the output. \ AVAILABLE LIBRARIES: this runs in the backend Python environment; these are ALREADY \ installed — pandas, numpy, matplotlib (charts), openpyxl / XlsxWriter (Excel), \ -python-docx (Word), Pillow (images), reportlab (generate PDF), and PyMuPDF a.k.a. \ +python-docx (Word), python-pptx (PowerPoint), Pillow (images), reportlab (generate PDF), and PyMuPDF a.k.a. \ `fitz` (read/parse PDF). To READ text or tables from a PDF, use `import fitz` \ (PyMuPDF); do NOT use pdfminer / pdfplumber / PyPDF2 — they are NOT installed. If an \ import fails, switch to an already-installed library instead of assuming a package \ exists; do NOT run `pip install` (this is a shared, offline environment).""" +) class LocalExecutor(BaseExecutor): @@ -391,7 +397,13 @@ def run(self, code: str) -> Any: # accumulated prefix handed the model {"exitcode": 1, "log": ""} on every # failure and forced it to debug blind. logger.warning("code interpreter block {}/{} exited {}", i + 1, len(code_blocks), exit_code) - return {"exitcode": exit_code, "log": self._tail(logs_all)} + # The advisory has to be attached HERE too, not only on the success + # path below: reading an absolute `/skills/...` raises + # FileNotFoundError, which is exactly a non-zero exit — so the one + # failure the read-side notice exists to explain would otherwise + # never see it. Appended AFTER ``_tail`` (which keeps the tail) so + # the truncation cannot eat it. + return {"exitcode": exit_code, "log": self._tail(logs_all) + self.absolute_path_advisory(original_code)} all_file_list += file_list # Deterministic safety net: if the script wrote a deliverable to an absolute diff --git a/src/backend/test/linsight/test_code_interpreter_output_path.py b/src/backend/test/linsight/test_code_interpreter_output_path.py index b82f18d0e2..eba64c7674 100644 --- a/src/backend/test/linsight/test_code_interpreter_output_path.py +++ b/src/backend/test/linsight/test_code_interpreter_output_path.py @@ -23,6 +23,7 @@ from bisheng_langchain.gpts.tools.code_interpreter.base_executor import ( ABSOLUTE_PATH_NOTICE, + ABSOLUTE_PROVISIONED_PATH_NOTICE, BaseExecutor, ) from bisheng_langchain.gpts.tools.code_interpreter.local_executor import ( @@ -201,5 +202,106 @@ def test_description_guides_to_installed_pdf_and_data_libs(): # names an installed PDF generator + core data lib so the model won't guess assert "reportlab" in d assert "pandas" in d + # Office writers: without python-pptx here the model concludes PPT is + # impossible and reaches for node/pptxgenjs, which is not installed either. + assert "python-docx" in d + assert "python-pptx" in d # shared, offline env — must not encourage pip install assert "pip install" in d + + +# --------------------------------------------------------------------------- +# Read-side advisory: /skills and /uploads are READ zones +# +# The write-side notice says files were "DISCARDED", which is exactly wrong for a +# read: `open('/skills/x/SKILL.md')` raises FileNotFoundError and nothing is lost. +# Telling the model its file vanished sent it hunting a data-loss problem that +# never happened (180 POC, 2026-08-08). +# --------------------------------------------------------------------------- +@pytest.mark.parametrize( + "code", + [ + "open('/skills/html-ppt-templates/SKILL.md')", + 'Path("/uploads/report.xlsx").read_bytes()', + "open('/skills')", + "open('/uploads')", + ], +) +def test_advisory_flags_absolute_provisioned_paths(code): + assert BaseExecutor.absolute_path_advisory(code) == ABSOLUTE_PROVISIONED_PATH_NOTICE + + +@pytest.mark.parametrize( + "code", + [ + "open('skills/x/SKILL.md')", # relative — correct + "open('./skills/x')", + "open('/skillset/x')", # different word + "url = 'https://host/skills/x'", # mid-string, not a path root + "open('/data/skills/x')", # path root is /data + ], +) +def test_read_advisory_silent_for_relative_or_unrelated(code): + assert BaseExecutor.absolute_path_advisory(code) == "" + + +def test_write_and_read_advisories_compose(): + code = "open('/skills/x/SKILL.md'); open('/output/a.pdf','wb')" + advisory = BaseExecutor.absolute_path_advisory(code) + assert ABSOLUTE_PATH_NOTICE in advisory + assert ABSOLUTE_PROVISIONED_PATH_NOTICE in advisory + # write side first, matching the pre-existing single-notice ordering + assert advisory.index(ABSOLUTE_PATH_NOTICE) < advisory.index(ABSOLUTE_PROVISIONED_PATH_NOTICE) + + +def test_run_appends_the_read_advisory_on_the_failure_path(monkeypatch): + """The advisory used to hang off the SUCCESS return only. An absolute + `/skills/...` read raises FileNotFoundError → non-zero exit → early return, so + the one failure this notice exists to explain never saw it.""" + exe = LocalExecutor(minio={}) + exe.local_sync_path = None + monkeypatch.setattr(exe, "insert_set_font_code", lambda code: code) + monkeypatch.setattr( + exe, + "run_with_dir", + lambda code, dir_path, lang: (1, "FileNotFoundError: '/skills/x/SKILL.md'\n", []), + ) + result = exe.run("open('/skills/x/SKILL.md')") + assert result["exitcode"] == 1 + assert "FileNotFoundError" in result["log"] + assert ABSOLUTE_PROVISIONED_PATH_NOTICE in result["log"] + + +def test_failure_advisory_survives_log_truncation(monkeypatch): + """Appended AFTER ``_tail`` — the tail-keeping truncation must not eat it.""" + exe = LocalExecutor(minio={}) + exe.local_sync_path = None + monkeypatch.setattr(exe, "insert_set_font_code", lambda code: code) + monkeypatch.setattr( + exe, + "run_with_dir", + lambda code, dir_path, lang: (1, "x" * (MAX_FAILURE_LOG_CHARS * 2), []), + ) + result = exe.run("open('/skills/x/SKILL.md')") + assert LOG_TRUNCATED_NOTICE in result["log"] + assert ABSOLUTE_PROVISIONED_PATH_NOTICE in result["log"] + + +# --------------------------------------------------------------------------- +# Tool descriptions carry the namespace mapping +# --------------------------------------------------------------------------- +def test_local_description_explains_both_namespaces(): + d = LocalExecutor(minio={}).description + assert "working directory" in d.lower() + assert "leading slash" in d.lower() + assert "skills/" in d + + +def test_e2b_description_omits_skills(): + """E2B copy-in snapshots the working dir BEFORE skills are materialised, so + promising `skills/` there would point the model at nothing.""" + from bisheng_langchain.gpts.tools.code_interpreter.base_executor import path_namespace_rules + + rules = path_namespace_rules(include_skills=False) + assert "skills/" not in rules + assert "`/output/x` is `output/x`" in rules diff --git a/src/backend/test/linsight/test_workspace_path_namespace.py b/src/backend/test/linsight/test_workspace_path_namespace.py new file mode 100644 index 0000000000..9ea1479456 --- /dev/null +++ b/src/backend/test/linsight/test_workspace_path_namespace.py @@ -0,0 +1,211 @@ +"""The file tools and the code interpreter speak two path namespaces. + +The workspace tools render everything with a LEADING SLASH (`/output/a.md`), while +the interpreter's cwd IS that same workspace root, so on disk it is `output/a.md`. +Session ``aa352cb4…`` (180 POC, 2026-08-08) paid for that gap twice: + +- forward: `open("/skills/html-ppt-templates/...")` → FileNotFoundError, and three + model round-trips to work out why; +- backward: the model handed `read_file` a host path it had seen in the interpreter + (`/root/.cache/bisheng/linsight/aa352cb4/output/.../qa/s08.png`), the leading slash + was simply dropped, and it became the nonsense key + `workspace//root/.cache/...` → four silent "File not found"s. + +The backward half is fixed by folding the host prefix back to a workspace key. That +also makes those reads SUCCEED, because the interpreter writes into ``file_dir`` and +``read`` checks that cache before MinIO. + +Stripping must be provably unambiguous: a heuristic here would silently open the +WRONG file, which is worse than an error. Hence the adversarial cases below. + +``asyncio_mode = auto`` — async tests need no decorator. +""" + +from __future__ import annotations + +import tempfile + +import pytest + +# Reuse the in-memory MinIO stand-in rather than duplicating it. ``test/linsight`` +# has no ``__init__.py``, so pytest puts the directory itself on sys.path and this +# resolves as a top-level module. +from test_workspace_backend import FakeMinioStorage + +from bisheng.linsight.domain.services.workspace_backend import ( + WorkspaceBackend, + normalize_workspace_path, + strip_executor_host_prefix, +) + +SVID = "aa352cb420f649d2be391cb09d9ad69d" +FILE_DIR = "/root/.cache/bisheng/linsight/aa352cb4" + + +# --------------------------------------------------------------------------- +# Pure function +# --------------------------------------------------------------------------- + + +def test_the_180_host_path_folds_back_to_a_workspace_key(): + """The exact path that produced four 'File not found' errors in production.""" + assert ( + normalize_workspace_path(f"{FILE_DIR}/output/siyuan-signal-deck/qa/s08.png", file_dir=FILE_DIR) + == "output/siyuan-signal-deck/qa/s08.png" + ) + + +@pytest.mark.parametrize( + "path", + ["/output/a.md", "output/a.md", "/uploads/report.xlsx", "skills/x/SKILL.md"], +) +def test_ordinary_workspace_paths_are_unchanged(path): + assert normalize_workspace_path(path, file_dir=FILE_DIR) == path.lstrip("/") + + +def test_a_real_root_directory_in_the_workspace_is_not_stripped(): + """Adversarial: the workspace may legitimately contain ``root/``. Only the FULL + file_dir prefix counts — never a mere ancestor segment.""" + assert normalize_workspace_path("/root/report.md", file_dir=FILE_DIR) == "root/report.md" + assert normalize_workspace_path("/root/.cache/x", file_dir=FILE_DIR) == "root/.cache/x" + assert normalize_workspace_path("/root/.cache/bisheng/x", file_dir=FILE_DIR) == "root/.cache/bisheng/x" + + +def test_a_prefix_similar_sibling_is_not_stripped(): + """Adversarial: ``/tmp/ws/aa352cb4x/...`` shares a string prefix with + ``/tmp/ws/aa352cb4`` but is a different directory — the ``/`` boundary is what + keeps them apart.""" + assert normalize_workspace_path("/tmp/ws/aa352cb4x/output/a", file_dir="/tmp/ws/aa352cb4") == ( + "tmp/ws/aa352cb4x/output/a" + ) + + +def test_another_sessions_task_dir_is_not_stripped(): + """A sibling task dir is NOT folded in. Catching it would also catch any + unrelated directory that merely shares the parent, and a rewrite rule that + cannot tell those apart would silently open the wrong file. The measured + production failure used the CURRENT session's dir, which rule 1 covers.""" + assert normalize_workspace_path("/root/.cache/bisheng/linsight/bb99ff00/output/a.md", file_dir=FILE_DIR) == ( + "root/.cache/bisheng/linsight/bb99ff00/output/a.md" + ) + + +def test_a_different_subtree_under_the_same_grandparent_is_not_stripped(): + assert normalize_workspace_path("/root/.cache/bisheng/other/output/a.md", file_dir=FILE_DIR) == ( + "root/.cache/bisheng/other/output/a.md" + ) + + +def test_the_file_dir_itself_is_the_workspace_root(): + assert normalize_workspace_path(FILE_DIR, file_dir=FILE_DIR) == "" + + +def test_traversal_is_still_rejected_after_stripping(): + with pytest.raises(ValueError): + normalize_workspace_path(f"{FILE_DIR}/../../etc/passwd", file_dir=FILE_DIR) + with pytest.raises(ValueError): + normalize_workspace_path("../etc/passwd", file_dir=FILE_DIR) + + +def test_without_a_file_dir_the_old_behaviour_is_preserved(): + """``test_skill_provisioning`` calls this as a plain contract function.""" + assert normalize_workspace_path(f"{FILE_DIR}/output/a.md") == "root/.cache/bisheng/linsight/aa352cb4/output/a.md" + + +def test_strip_reports_whether_it_did_anything(): + assert strip_executor_host_prefix(f"{FILE_DIR}/output/a", FILE_DIR) == ("output/a", True) + assert strip_executor_host_prefix("/output/a", FILE_DIR) == ("/output/a", False) + assert strip_executor_host_prefix("/output/a", None) == ("/output/a", False) + + +# --------------------------------------------------------------------------- +# Backend integration +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def minio(): + return FakeMinioStorage() + + +@pytest.fixture() +def workdir(): + with tempfile.TemporaryDirectory() as d: + yield d + + +def _backend(minio, workdir): + return WorkspaceBackend(svid=SVID, minio=minio, file_dir=workdir) + + +def _interpreter_writes(workdir: str, rel: str, data: bytes) -> str: + """Simulate the code interpreter: writes straight into its cwd, never MinIO.""" + from pathlib import Path + + p = Path(workdir) / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_bytes(data) + return str(p) + + +def test_a_host_path_from_the_interpreter_now_reads_successfully(minio, workdir): + """The real payoff: not just a better error, an actual successful read. The + interpreter's output never reaches MinIO, but ``read`` checks the local cache + first — which is the very directory the interpreter wrote to.""" + host_path = _interpreter_writes(workdir, "output/report.md", b"hello from the interpreter") + backend = _backend(minio, workdir) + + result = backend.read(host_path) + + assert result.error is None + assert "hello from the interpreter" in result.file_data["content"] + assert (minio.bucket, f"workspace/{SVID}/output/report.md") not in minio.store + + +async def test_the_async_read_folds_the_same_way(minio, workdir): + host_path = _interpreter_writes(workdir, "output/report.md", b"async hello") + backend = _backend(minio, workdir) + + result = await backend.aread(host_path) + + assert result.error is None + assert "async hello" in result.file_data["content"] + + +def test_ls_accepts_either_namespace(minio, workdir): + _interpreter_writes(workdir, "output/a.txt", b"a") + backend = _backend(minio, workdir) + backend.write("/output/a.txt", "a") + + by_workspace = {e["path"] for e in backend.ls("/output").entries} + by_host = {e["path"] for e in backend.ls(f"{workdir}/output").entries} + assert by_workspace == by_host + + +def test_a_real_root_directory_still_round_trips(minio, workdir): + """Adversarial, end to end: a workspace file literally named ``root/...`` must + still write and read back as itself.""" + backend = _backend(minio, workdir) + backend.write("/root/report.md", "mine") + + assert (minio.bucket, f"workspace/{SVID}/root/report.md") in minio.store + assert "mine" in backend.read("/root/report.md").file_data["content"] + + +def test_missing_host_path_error_names_both_namespaces(minio, workdir): + backend = _backend(minio, workdir) + + err = backend.read(f"{workdir}/output/missing.png").error + assert "workspace path 'output/missing.png'" in err + + err_host = backend.read("/home/user/output/x.png").error + assert "HOST filesystem" in err_host + + err_plain = backend.read("/output/missing.png").error + assert err_plain == "File '/output/missing.png' not found" + + +def test_edit_shares_the_same_error_wording(minio, workdir): + backend = _backend(minio, workdir) + err = backend.edit(f"{workdir}/output/missing.md", "a", "b").error + assert "workspace path 'output/missing.md'" in err From 6433a2bff096fd9475335b56af03ee14f04ed617 Mon Sep 17 00:00:00 2001 From: LineWalker Date: Sun, 9 Aug 2026 17:13:38 +0800 Subject: [PATCH 4/6] =?UTF-8?q?fix(linsight):=20=E5=8E=82=E5=95=86?= =?UTF-8?q?=E4=B8=8A=E6=B8=B8=E5=8F=96=E6=95=B0=E4=B8=AD=E6=96=AD=E7=9A=84?= =?UTF-8?q?=20400=20=E5=88=A4=E4=B8=BA=E5=8F=AF=E9=87=8D=E8=AF=95=EF=BC=8C?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E4=B8=80=E6=AC=A1=E6=80=A7=E5=88=A4=E6=AD=BB?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 180 会话 649ba617（2026-08-09）：模型把一张 deck 截图交给 dashscope，回来的是 400 {"code":"DOWNLOAD_FAILED", "message":"failed to download image, err: Get \"http://…/restful/data-uri/null/…\": read tcp 10.0.2.60:46976-> 10.86.10.104:6000: read: connection reset by peer"} 两端都是厂商自己内网的私有 IP——请求本身没问题，是**他们回取暂存对象时连接被重置**。 但 classify_behavior 把「plain 400」兜进 DEGRADABLE，而主图的 DEGRADABLE 是 re-raise，于是零重试、21 分钟的任务直接判失败。 新增 _TRANSIENT_TRANSPORT_SIGNATURES：当错误里带 Go/POSIX 的传输层失败措辞 （connection reset / refused、broken pipe、i/o timeout、unexpected EOF、 no route to host）时归 RETRYABLE。这些是传输层通用措辞不是厂商错误码，与本模块 既有的扁平签名集（_QUOTA_SIGNATURES / _RATE_LIMIT_SIGNATURES）同一范式， 不引入 per-vendor 分支。 刻意只认传输层失败：404 / 坏 URL / 不支持的格式这类取数失败每次重试都会失败， 必须继续走 DEGRADABLE，不能白烧三次退避。检查点放在 FAIL_FAST 分支之后，所以 欠费/鉴权错误即使正文里恰好带「connection reset」也不会被翻成重试（有守门测试）。 label 层同步：重试耗尽后给「服务繁忙」而不是通用未知卡片。 Co-Authored-By: Claude Opus 5 (1M context) --- .../common/services/llm_error_classifier.py | 43 +++++++ .../linsight/test_llm_error_classifier.py | 106 ++++++++++++++++++ 2 files changed, 149 insertions(+) diff --git a/src/backend/bisheng/common/services/llm_error_classifier.py b/src/backend/bisheng/common/services/llm_error_classifier.py index 8a9533875d..eb5602c441 100644 --- a/src/backend/bisheng/common/services/llm_error_classifier.py +++ b/src/backend/bisheng/common/services/llm_error_classifier.py @@ -116,6 +116,33 @@ class ErrorType(str, enum.Enum): "限流", ) +# Transient TRANSPORT failures a provider reports from ITS OWN upstream fetch, +# wrapped in an otherwise non-transient status. When an endpoint fetches something +# on our behalf — staging a data-URI image into object storage, a sandbox pulling a +# file, a RAG fetcher — its I/O error surfaces as a 4xx even though the REQUEST was +# fine; only their fetch of it failed. Retrying is exactly right, and a plain 400 +# otherwise DEGRADEs, which on the main graph means the whole run dies. +# +# Measured on 180 (2026-08-09, session 649ba617): a deck screenshot handed to the +# model came back as +# 400 {"code":"DOWNLOAD_FAILED", ... "read tcp 10.0.2.60:46976->10.86.10.104:6000: +# read: connection reset by peer"} +# — both endpoints inside the vendor's own network — and killed a 21-minute run +# with zero retries. +# +# These are Go/POSIX transport phrases, not vendor error codes, so this stays +# vendor-agnostic like the other flat signature sets. DELIBERATELY transport-only: +# a fetch that failed with 404 / bad URL / unsupported format is a genuine client +# error and MUST keep degrading rather than burn three backoffs on a sure failure. +_TRANSIENT_TRANSPORT_SIGNATURES: tuple[str, ...] = ( + "connection reset", + "connection refused", + "broken pipe", + "i/o timeout", + "unexpected eof", + "no route to host", +) + # Content-moderation / safety-guardrail codes across mainstream vendors: # Aliyun DashScope (data_inspection_failed / inappropriate_content), # Baidu Qianfan (336003/336005), Zhipu (1301), MiniMax (2013), @@ -231,6 +258,16 @@ def _is_rate_limit(exc: BaseException) -> bool: return any(sig in text for sig in _RATE_LIMIT_SIGNATURES) +def _is_transient_transport(exc: BaseException) -> bool: + """A transport-level failure the PROVIDER hit on its own upstream fetch. + + Checked only after the FAIL_FAST branches, so a quota/auth error carrying such + wording in a nested message can never be turned into a retry. + """ + text = _exc_text(exc) + return any(sig in text for sig in _TRANSIENT_TRANSPORT_SIGNATURES) + + def _is_task_aborted(exc: BaseException) -> bool: """The task was aborted by a framework/loop guard, not a provider error. @@ -300,6 +337,8 @@ def classify_behavior(exc: BaseException) -> Behavior: return Behavior.RETRYABLE if isinstance(exc, (TimeoutError, ConnectionError)): return Behavior.RETRYABLE + if _is_transient_transport(exc): # provider's own upstream fetch blew up mid-flight + return Behavior.RETRYABLE # 3) DEGRADABLE — default bucket: content filter, plain 400, anything else. return Behavior.DEGRADABLE @@ -351,6 +390,10 @@ def label_error(exc: BaseException) -> ErrorType: status = _status_code(exc) if isinstance(exc, openai.InternalServerError) or (status is not None and 500 <= status < 600): return ErrorType.SERVICE_UNAVAILABLE + # Retries exhausted on the provider's own upstream fetch: "service busy" is the + # honest story for the user, not the generic unknown card. + if _is_transient_transport(exc): + return ErrorType.SERVICE_UNAVAILABLE return ErrorType.UNKNOWN diff --git a/src/backend/test/linsight/test_llm_error_classifier.py b/src/backend/test/linsight/test_llm_error_classifier.py index d8c50a5b78..d986d070be 100644 --- a/src/backend/test/linsight/test_llm_error_classifier.py +++ b/src/backend/test/linsight/test_llm_error_classifier.py @@ -273,3 +273,109 @@ def test_classify_for_event_accepts_bare_string(): assert result.error_type == ErrorType.CONTENT_FILTER.value assert result.error_code == 11090 assert result.detail.startswith("Task execution failed") + + +# --------------------------------------------------------------------------- # +# Transient upstream-fetch failures wrapped in a 400 +# +# Session 649ba617 (180 POC, 2026-08-09): a deck screenshot handed to the model +# came back as a 400 whose body said the PROVIDER's own fetch of the staged image +# was reset mid-flight — both endpoints inside the vendor's network. Plain 400s +# DEGRADE, and DEGRADABLE on the main graph re-raises, so one transport blip +# killed a 21-minute run with zero retries. +# --------------------------------------------------------------------------- # + +_DOWNLOAD_RESET_BODY = { + "message": ( + '{"error":{"code":"DOWNLOAD_FAILED","message":"failed to download image, err: ' + 'Get \\"http://ds-restful-api-bj-prod.oss-cn-beijing.aliyuncs.com/restful/data-uri/null/' + 'd7f119b9/f7f86073?Expires=1786436767\\": read tcp 10.0.2.60:46976->10.86.10.104:6000: ' + 'read: connection reset by peer"}}' + ), + "type": "UploadFailed", + "code": "UploadFailed", +} + + +def test_provider_upstream_reset_is_retryable(): + """The exact production payload.""" + exc = make_exc( + openai.BadRequestError, + message=_DOWNLOAD_RESET_BODY["message"], + code="UploadFailed", + body=_DOWNLOAD_RESET_BODY, + status_code=400, + ) + assert classify_behavior(exc) is Behavior.RETRYABLE + + +@pytest.mark.parametrize( + "message", + [ + "failed to download image, err: read: connection reset by peer", + "upstream fetch: write: broken pipe", + "dial tcp 10.0.0.1:6000: i/o timeout", + "unexpected EOF while reading the staged object", + "dial tcp 10.0.0.1:6000: connect: connection refused", + "no route to host", + ], +) +def test_transport_phrases_are_retryable(message): + exc = make_exc(openai.BadRequestError, message=message, status_code=400) + assert classify_behavior(exc) is Behavior.RETRYABLE + + +@pytest.mark.parametrize( + "message", + [ + "failed to download image, err: 404 Not Found", + "failed to download image: unsupported image format", + "invalid_request_error: image url is not accessible", + "malformed request", + ], +) +def test_permanent_fetch_failures_still_degrade(message): + """Transport-only on purpose: a 404 / bad URL / bad format will fail every + retry, so it must keep degrading instead of burning three backoffs.""" + exc = make_exc(openai.BadRequestError, message=message, status_code=400) + assert classify_behavior(exc) is Behavior.DEGRADABLE + + +def test_fail_fast_still_wins_over_a_transport_phrase(): + """Ordering guard: the FAIL_FAST branches run first, so an arrears/auth error + that happens to mention a reset connection is never turned into a retry.""" + arrears = make_exc( + openai.BadRequestError, + message="欠费 stopped the request; connection reset by peer", + status_code=400, + ) + assert classify_behavior(arrears) is Behavior.FAIL_FAST + + auth = make_exc( + openai.AuthenticationError, + message="invalid api key (connection reset by peer)", + status_code=401, + ) + assert classify_behavior(auth) is Behavior.FAIL_FAST + + +def test_content_filter_is_not_hijacked_by_a_transport_phrase(): + """Content moderation must keep degrading — retrying it is pointless.""" + exc = make_exc( + openai.BadRequestError, + message="Output data may contain inappropriate content", + code="data_inspection_failed", + status_code=400, + ) + assert classify_behavior(exc) is Behavior.DEGRADABLE + + +def test_upstream_reset_labels_as_service_unavailable(): + """If retries do run out, the user gets "service busy", not the unknown card.""" + exc = make_exc( + openai.BadRequestError, + message=_DOWNLOAD_RESET_BODY["message"], + body=_DOWNLOAD_RESET_BODY, + status_code=400, + ) + assert label_error(exc) is ErrorType.SERVICE_UNAVAILABLE From 10624289a6d1ac311bb33d30cf4c9fae7408647b Mon Sep 17 00:00:00 2001 From: dolphin Date: Mon, 10 Aug 2026 15:25:21 +0800 Subject: [PATCH 5/6] style(chat): play chat media in the knowledge-space player, centred on a scrim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Daily mode played attachments through bare