mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(goal): validate goal text and preserve the current goal on message replay (#6685)
* fix(goal): validate goal text and preserve current goal on message replay * fix(goal): reject malformed continuation counters before admission * fix(goal): reject external reducer overwrite markers --------- Co-authored-by: Totoro-qaq <279883115+Totoro-qaq@users.noreply.github.com>
This commit is contained in:
Totoro
GitHub
parent
a7dd4849eb
commit
5e67d41c47
@ -851,6 +851,13 @@ This release closes that milestone with **439 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **goal:** External state and run-input writes reject missing or invalid goal
|
||||
objectives, malformed counters and reducer control markers before admission,
|
||||
while retaining valid sparse goal state. Historical
|
||||
checkpoint replay, including regenerate, keeps the current goal and its used
|
||||
continuation budget instead of restoring an older goal; it drops the old answer's
|
||||
completion record. No frontend changes or database migration are required.
|
||||
|
||||
- **channels:** IM conversations now keep their DeerFlow thread across Gateway
|
||||
replicas. The `ChannelManager` kept its chat-to-thread bindings
|
||||
(`channel_name:chat_id[:topic_id]` → `thread_id`) in a per-process JSON file,
|
||||
|
||||
@ -728,6 +728,10 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **目标：** 外部 state 和运行输入在写入前拒绝缺失或无效的目标文本、错误计数及 reducer 控制标记，保持合法的稀疏
|
||||
goal 状态兼容。重放历史 checkpoint（包括重新生成回答）沿用当前 goal 和已用续跑
|
||||
预算，不会恢复旧目标，并移除旧回答的达成记录；不需要前端改动或数据库迁移。
|
||||
|
||||
- **运行时：** JSONL 事件存储在末行写坏后不再丢失事件。写入中途被打断会让文件缺少结尾换行，下一次追加会接在残缺记录后面，
|
||||
两者合成一行无法解析；被截断的多字节字符还会让整个文件解码失败，所有完好记录都读不出来，序号恢复也随之失效。现在追加前
|
||||
若文件末尾不是换行会先补一个分隔符，读取时逐个物理行单独解码、只跳过损坏的那一行，批量追加失败时仍会截回原始大小。
|
||||
|
||||
@ -1919,6 +1919,8 @@ Supported commands:
|
||||
|
||||
After each Gateway-backed run, DeerFlow evaluates the visible conversation, including the assistant's tool calls and shortened tool results, against the active goal with a non-thinking evaluator model. A successful tool result alone does not satisfy a goal, and when the assistant had to guess missing or ambiguous information the evaluator reports `needs_user_input`. The evaluator must return a typed blocker (`missing_evidence`, `needs_user_input`, `run_failed`, `external_wait`, or `goal_not_met_yet`) plus visible evidence. DeerFlow only injects a hidden continuation when the latest assistant turn is durably checkpointed, the blocker is `goal_not_met_yet`, the thread did not change during evaluation, and the no-progress breaker has not fired. The safety cap defaults to 8 hidden continuations, and repeated identical non-progress evaluations stop after 2 attempts. `/goal clear` and any user-authored new input win over queued continuations. When the evaluator considers the goal satisfied, DeerFlow clears it after the run successfully finalizes and publishes the updated thread state. The same checkpoint records the completion as `goal_outcome`, which stays until the next goal change. If required artifact delivery or its receipt persistence fails, the run reports an error and the goal remains active for a later retry; this does not start another hidden continuation.
|
||||
|
||||
Regenerating an earlier response keeps the goal currently set for the chat and its used continuation budget. It does not restore a goal that was cleared or achieved. The earlier answer's completion record is removed when its historical checkpoint is replayed.
|
||||
|
||||
The Web UI shows the active goal above the composer, says why auto-continue stopped, and shows a "Goal met" notice after a goal is met. Editing the last message is off while a goal is set. The same command is available from the TUI and supported IM channels. In the Web UI and supported IM channels, setting `/goal <completion condition>` also starts a run with the condition as the task; status and clear commands only manage goal state. Setting or clearing a goal is rejected while that thread has a run in flight, including a run owned by another Gateway worker, so the goal checkpoint cannot branch away from an active run's checkpoint lineage.
|
||||
|
||||
When your role lacks `runs:create`, the Web UI rejects a new task or `/goal <completion condition>` before preparing the thread or saving the goal, and keeps your draft for retry. Goal status, goal clearing, and `/compact` remain governed by their own endpoint permissions.
|
||||
|
||||
@ -765,6 +765,8 @@ Web UI 输入框支持浏览器侧语音听写。浏览器提供 Web Speech API
|
||||
|
||||
每次 Gateway 驱动的 run 结束后，DeerFlow 会用一个 non-thinking 的评估模型，把可见的对话内容（包括助手的工具调用和截短的工具结果）拿去和激活的 goal 比对。工具调用成功本身不代表 goal 已完成；如果助手不得不猜测缺失或含糊的信息，评估模型会报告 `needs_user_input`。评估模型必须返回一个带类型的 blocker（`missing_evidence`、`needs_user_input`、`run_failed`、`external_wait` 或 `goal_not_met_yet`），并附上可见证据。只有在最近一轮 assistant 回复已被持久化 checkpoint、blocker 是 `goal_not_met_yet`、评估期间 thread 没有变化、且无进展熔断器没有触发时，DeerFlow 才会注入一次 hidden continuation。安全上限默认是 8 次 hidden continuation；连续两次相同的无进展评估后就会停止。`/goal clear` 以及任何用户手动输入的新内容，优先级都高于排队中的 continuation。当 goal 被满足时，DeerFlow 会自动清除它，并发布更新后的 thread 状态。同一个 checkpoint 会把这次达成记为 `goal_outcome`，保留到下一次修改 goal。
|
||||
|
||||
重新生成此前的回答时，会沿用当前对话的 goal 和已用续跑预算，不会恢复已清除或已达成的旧 goal。重放历史 checkpoint 时会移除旧回答的达成记录，避免把它当作新回答已经达成的证明。
|
||||
|
||||
Web UI 会在输入框上方展示当前激活的 goal，说明自动续跑为什么停下，并在 goal 达成后显示「目标已达成」提示；设置了 goal 时不能编辑最后一条消息。同样的命令在 TUI 和受支持的 IM 渠道里也可用。在 Web UI 和受支持的 IM 渠道里，设置 `/goal <完成条件>` 还会以该条件作为任务启动一次 run；状态查询和清除命令则只管理 goal 状态本身。
|
||||
|
||||
### 手动上下文压缩
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
### Gateway API (`app/gateway/`)
|
||||
|
||||
Reject external run/state writes with `sandbox`, `thread_data`, `viewed_images`, or `goal_outcome`.
|
||||
Reject external `sandbox`, `thread_data`, `viewed_images`, `goal_outcome`. Non-null goals need valid text, non-negative int counters/limits and no Overwrite markers; keep sparse values.
|
||||
|
||||
Studio retains sanitized creation metadata.
|
||||
|
||||
|
||||
@ -107,6 +107,7 @@ from deerflow.subagents.status_contract import SUBAGENT_ACCEPTANCE_VERDICT_KEY,
|
||||
from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY, ensure_trace_context, ensure_trace_id
|
||||
from deerflow.utils.assembly_io import run_assembly
|
||||
from deerflow.utils.file_io import await_drained
|
||||
from deerflow.utils.goal_objective import normalize_goal_objective
|
||||
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, UNTRUSTED_INPUT_KEY
|
||||
from deerflow.utils.thread_id import validate_thread_id
|
||||
|
||||
@ -490,6 +491,24 @@ def _normalize_input_messages(
|
||||
SERVER_OWNED_STATE_CHANNELS = frozenset({"sandbox", "thread_data", "viewed_images", "goal_outcome"})
|
||||
|
||||
|
||||
def _normalize_external_goal(value: Any, *, location: str) -> dict[str, Any] | None:
|
||||
"""Require usable goal text without rebuilding caller-supplied goal state."""
|
||||
if value is None:
|
||||
return None
|
||||
if not isinstance(value, Mapping) or not isinstance(value.get("objective"), str):
|
||||
raise HTTPException(status_code=400, detail=f"{location} must be an object with a string objective")
|
||||
if "__overwrite__" in value or value.get("type") == "__overwrite__":
|
||||
raise HTTPException(status_code=400, detail=f"{location} cannot contain an Overwrite marker")
|
||||
try:
|
||||
objective = normalize_goal_objective(value["objective"])
|
||||
except ValueError as exc:
|
||||
raise HTTPException(status_code=400, detail=f"{location}: {exc}") from None
|
||||
for field in ("continuation_count", "max_continuations", "no_progress_count", "max_no_progress_continuations"):
|
||||
if field in value and (type(value[field]) is not int or value[field] < 0):
|
||||
raise HTTPException(status_code=400, detail=f"{location}.{field} must be a non-negative integer")
|
||||
return {**value, "objective": objective}
|
||||
|
||||
|
||||
def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, Any]:
|
||||
"""Validate and sanitize caller-supplied state values before checkpointing.
|
||||
|
||||
@ -514,7 +533,9 @@ def strip_server_owned_state_metadata(values: Mapping[str, Any]) -> dict[str, An
|
||||
|
||||
stripped: dict[str, Any] = {}
|
||||
for channel, value in values.items():
|
||||
if channel == "messages" and value is not None:
|
||||
if channel == "goal":
|
||||
stripped[channel] = _strip_external_metadata_from_message_like(_normalize_external_goal(value, location="values.goal"))
|
||||
elif channel == "messages" and value is not None:
|
||||
stripped[channel] = _normalize_input_messages(value, location="values.messages")
|
||||
elif channel == "delegations" and isinstance(value, list):
|
||||
stripped[channel] = [_strip_external_delegation_verdict(item) for item in value]
|
||||
@ -588,7 +609,7 @@ def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool
|
||||
result = {**result, "delegations": cleaned}
|
||||
# merge_goal ignores None, so only a goal value replaces the goal.
|
||||
if result.get("goal") is not None:
|
||||
result = {**result, "goal_outcome": None}
|
||||
result = {**result, "goal": _normalize_external_goal(result["goal"], location="input.goal"), "goal_outcome": None}
|
||||
return result
|
||||
|
||||
|
||||
|
||||
@ -222,6 +222,26 @@ Only the newest entry carries `messages`, the `goal` while it is active, and
|
||||
entries carry only `title` and `thread_data`. `goal_outcome` is server-owned:
|
||||
state updates and run input that include it are rejected with 400.
|
||||
|
||||
External state updates and run input that supply a non-null `goal` must include
|
||||
a string `objective` that is non-empty after whitespace normalization and at
|
||||
most 4,000 normalized characters. Invalid values return 400 before state writes
|
||||
or run admission. Optional goal timestamps and counters are not required or reset.
|
||||
When present, `continuation_count`, `max_continuations`, `no_progress_count` and
|
||||
`max_no_progress_continuations` must be non-negative integers; booleans, numeric
|
||||
strings, floats and null values are rejected with 400. Valid counters and limits
|
||||
are preserved, including an already-used continuation budget.
|
||||
Goal payloads must be state values, not reducer control wrappers: the reserved
|
||||
`__overwrite__` key and `type: "__overwrite__"` are rejected with 400 at both
|
||||
external boundaries. Ordinary `type`/`value` fields remain compatible.
|
||||
`goal: null` retains its existing semantics: it clears a state update's goal,
|
||||
but does not replace a goal through run input.
|
||||
|
||||
Replaying messages from an earlier root checkpoint, including response regeneration, preserves
|
||||
the thread's current goal and continuation budget. A cleared or achieved goal
|
||||
is not restored from history. A valid goal explicitly supplied in run input
|
||||
still replaces it. Historical replay clears the old `goal_outcome`, because
|
||||
the previous answer's verdict does not certify the replacement answer.
|
||||
|
||||
### Runs
|
||||
|
||||
#### Create Run
|
||||
|
||||
@ -70,7 +70,7 @@ from `on_llm_end` before inspecting the response or touching any run state.
|
||||
**Skill history:** `record_skill_usage` saves lead-run snapshots on terminal
|
||||
answers for paginated history. See `docs/skill-usage-ui.md`.
|
||||
|
||||
**Run delivery receipts:** Journal artifact evidence and terminal status must finalize before an ordinary satisfied goal is cleared. That cleanup uses a durable checkpoint-write reservation; delivery failure retains the ordinary goal without another continuation. Only that clear passes `outcome` to `write_thread_goal`, which removes `goal_outcome` on every other goal write. The scheduled-only exception is described below. Details: `backend/docs/runtime-guidance-details.md`.
|
||||
**Run delivery receipts:** Finalize artifact evidence and terminal status before clearing an ordinary satisfied goal, under a checkpoint-write reservation. Delivery failure retains the goal without continuing. Only that clear writes `goal_outcome`; other goal writes remove it. Historical replay keeps the current goal and budget, and drops the old outcome. Scheduled exception below; details: `backend/docs/runtime-guidance-details.md`.
|
||||
|
||||
**Deferred terminal commit:** With an event store, the worker stages its terminal
|
||||
status locally and commits it only after finalization's receipt and duration
|
||||
|
||||
@ -1341,6 +1341,23 @@ async def run_agent(
|
||||
if scheduled_goal is None:
|
||||
await _clear_stale_scheduled_goal(record=record, run_manager=run_manager, bridge=bridge, checkpointer=checkpointer)
|
||||
|
||||
# Full-mode replay keeps its historical message fork, but the owner's
|
||||
# current goal and budget must win. Read after scheduled-goal preflight
|
||||
# so a cleaned-up occurrence goal cannot be reintroduced by the input.
|
||||
configurable = config.get("configurable", {})
|
||||
selected_checkpoint_id = configurable.get("checkpoint_id")
|
||||
if checkpointer is not None and mode == "full" and isinstance(graph_input, dict) and isinstance(selected_checkpoint_id, str) and selected_checkpoint_id and not configurable.get("checkpoint_ns"):
|
||||
async with _checkpoint_thread_lock(thread_id):
|
||||
current = await accessor.aget(checkpoint_config)
|
||||
if _checkpoint_id(current) != selected_checkpoint_id:
|
||||
current_values = getattr(current, "values", None) or {}
|
||||
graph_input = dict(graph_input)
|
||||
if graph_input.get("goal") is None:
|
||||
# merge_goal ignores plain None, including a goal that
|
||||
# the user cleared or completed after the selected turn.
|
||||
graph_input["goal"] = Overwrite(copy.deepcopy(current_values.get("goal")))
|
||||
graph_input["goal_outcome"] = None
|
||||
|
||||
# Capture the effective (resolved) model name from the agent's metadata.
|
||||
# _resolve_model_name in agent.py may return the default model if the
|
||||
# requested name is not in the allowlist — this update ensures the
|
||||
@ -2689,6 +2706,10 @@ async def _linearize_delta_checkpoint_resume(
|
||||
``None`` when the channel has no constructible default). The abandoned
|
||||
turn stays in checkpoint history as the rewritten head's ancestry.
|
||||
|
||||
The owner's current goal and continuation budget survive replay; the old
|
||||
achievement record cannot certify the new answer. Full mode applies that
|
||||
policy through the run input after scheduled-goal preflight instead.
|
||||
|
||||
Returns the materialized messages when the resume was linearized, or
|
||||
``None`` when there was nothing to do (full mode, no checkpoint selector,
|
||||
a non-root namespace, or a selector that already names the head). Failures
|
||||
@ -2731,6 +2752,8 @@ async def _linearize_delta_checkpoint_resume(
|
||||
selected_values = dict(values)
|
||||
head_values = getattr(head, "values", None) or {}
|
||||
head_values = dict(head_values) if isinstance(head_values, dict) else {}
|
||||
selected_values["goal"] = head_values.get("goal")
|
||||
selected_values["goal_outcome"] = None
|
||||
replacement_values = _complete_state_replacement_values(
|
||||
mutation_graph=mutation_graph,
|
||||
selected_values=selected_values,
|
||||
|
||||
252
backend/tests/test_goal_checkpoint_resume.py
Normal file
252
backend/tests/test_goal_checkpoint_resume.py
Normal file
@ -0,0 +1,252 @@
|
||||
"""Checkpoint replay replaces messages, not the owner's current goal."""
|
||||
|
||||
import asyncio
|
||||
import copy
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import pytest
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.graph import StateGraph
|
||||
from langgraph.types import Command, Overwrite, interrupt
|
||||
|
||||
from deerflow.agents.thread_state import get_thread_state_schema
|
||||
from deerflow.runtime.checkpoint_state import CheckpointStateAccessor, build_state_mutation_graph
|
||||
from deerflow.runtime.goal import build_goal_outcome, build_goal_state, write_thread_goal
|
||||
from deerflow.runtime.runs import worker
|
||||
from deerflow.runtime.runs.manager import RunManager
|
||||
from deerflow.runtime.runs.schemas import RunStatus
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
async def _replay(monkeypatch, *, mode, current_kind, explicit_goal=False, select_head=False, rollback=False, late_goal_update=None):
|
||||
saver = InMemorySaver()
|
||||
thread_id = "goal-checkpoint-resume"
|
||||
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||||
schema = get_thread_state_schema(mode)
|
||||
accessor = CheckpointStateAccessor.bind(build_state_mutation_graph("seed", mode, schema), saver, mode=mode)
|
||||
previous_goal = build_goal_state("Finish the previous objective", now="2026-10-01T00:00:00Z")
|
||||
selected_outcome = None
|
||||
if current_kind == "previous-achievement":
|
||||
selected_outcome = build_goal_outcome(previous_goal, {"satisfied": True, "blocker": "none", "reason": "The earlier answer met its goal"}, reply_message_id="earlier-answer")
|
||||
await accessor.aupdate(
|
||||
config,
|
||||
{
|
||||
"messages": [HumanMessage(id="earlier", content="Earlier request"), AIMessage(id="earlier-answer", content="Earlier answer")],
|
||||
"goal": previous_goal if selected_outcome is None else Overwrite(None),
|
||||
"goal_outcome": selected_outcome,
|
||||
},
|
||||
as_node="seed",
|
||||
)
|
||||
selected = await accessor.aget(config)
|
||||
current_goal = copy.deepcopy(previous_goal)
|
||||
current_goal.update(continuation_count=3, no_progress_count=1, max_continuations=5)
|
||||
if current_kind == "changed":
|
||||
current_goal.update(objective="Finish the revised objective", created_at="2026-10-02T00:00:00Z")
|
||||
if current_kind == "recovered-scheduled":
|
||||
current_goal["created_at"] = "2026-10-02T00:00:00.123456+00:00"
|
||||
outcome = None
|
||||
if current_kind in ("cleared", "achieved"):
|
||||
current_goal = None
|
||||
if current_kind == "achieved":
|
||||
outcome = build_goal_outcome(previous_goal, {"satisfied": True, "blocker": "none", "reason": "Previous answer met the goal"}, reply_message_id="superseded")
|
||||
await accessor.aupdate(
|
||||
config,
|
||||
{
|
||||
"messages": [HumanMessage(id="redo", content="Regenerate this answer"), AIMessage(id="superseded", content="Answer to replace")],
|
||||
"goal": Overwrite(current_goal),
|
||||
"goal_outcome": outcome,
|
||||
},
|
||||
as_node="seed",
|
||||
)
|
||||
head = await accessor.aget(config)
|
||||
prepared_config = {"configurable": {**config["configurable"], "checkpoint_id": (head if select_head else selected).config["configurable"]["checkpoint_id"]}}
|
||||
graph_input = {"messages": [HumanMessage(id="redo", content="Regenerate this answer")]}
|
||||
if explicit_goal:
|
||||
current_goal = build_goal_state("The explicit new request goal", now="2026-10-03T00:00:00Z")
|
||||
graph_input["goal"] = current_goal
|
||||
captured_states = []
|
||||
factory_ready = asyncio.Event()
|
||||
manager = RunManager()
|
||||
record = await manager.create(thread_id)
|
||||
if current_kind == "recovered-scheduled":
|
||||
monkeypatch.setattr(manager, "scheduled_goal_source", AsyncMock(return_value={"run_id": "terminal-scheduled-run"}))
|
||||
|
||||
async def answer(state):
|
||||
captured_states.append(copy.deepcopy(state))
|
||||
if rollback:
|
||||
record.abort_action = "rollback"
|
||||
record.abort_event.set()
|
||||
return {"messages": [AIMessage(id="replacement", content="Replacement answer")]}
|
||||
|
||||
def factory(*, config):
|
||||
del config
|
||||
builder = StateGraph(schema)
|
||||
builder.add_node("answer", answer)
|
||||
builder.set_entry_point("answer")
|
||||
builder.set_finish_point("answer")
|
||||
factory_ready.set()
|
||||
return builder.compile(checkpointer=saver)
|
||||
|
||||
# These tests exercise state selection and the real worker/graph, without
|
||||
# asking a provider to judge a deterministic checkpoint regression.
|
||||
monkeypatch.setattr(worker, "_prepare_goal_continuation_input", AsyncMock(return_value=None))
|
||||
bridge = SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock())
|
||||
|
||||
async def execute():
|
||||
await worker.run_agent(
|
||||
bridge,
|
||||
manager,
|
||||
record,
|
||||
ctx=worker.RunContext(checkpointer=saver, checkpoint_channel_mode=mode),
|
||||
agent_factory=factory,
|
||||
graph_input=graph_input,
|
||||
config=prepared_config,
|
||||
stream_modes=["values"],
|
||||
)
|
||||
|
||||
if late_goal_update:
|
||||
# A user clears the goal after the replay selector was prepared but
|
||||
# before the worker acquires the existing checkpoint-write lock.
|
||||
async with worker._checkpoint_thread_lock(thread_id):
|
||||
task = asyncio.create_task(execute())
|
||||
try:
|
||||
await asyncio.wait_for(factory_ready.wait(), timeout=5)
|
||||
if late_goal_update == "change":
|
||||
current_goal = {**current_goal, "objective": "The latest user revision", "created_at": "2026-10-04T00:00:00Z"}
|
||||
else:
|
||||
current_goal = None
|
||||
await write_thread_goal(saver, thread_id, current_goal)
|
||||
except BaseException:
|
||||
task.cancel()
|
||||
await asyncio.gather(task, return_exceptions=True)
|
||||
raise
|
||||
await asyncio.wait_for(task, timeout=5)
|
||||
else:
|
||||
await execute()
|
||||
final = await accessor.aget(config)
|
||||
return record, captured_states, final, current_goal, head, outcome
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize("current_kind", ["cleared", "achieved", "active", "changed", "previous-achievement"])
|
||||
async def test_checkpoint_replay_preserves_current_goal_and_budget(monkeypatch, mode, current_kind):
|
||||
record, states, final, current_goal, _head, _outcome = await _replay(monkeypatch, mode=mode, current_kind=current_kind)
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert len(states) == 1
|
||||
assert states[0].get("goal") == current_goal
|
||||
assert final.values.get("goal") == current_goal
|
||||
assert states[0].get("goal_outcome") is None
|
||||
assert final.values.get("goal_outcome") is None
|
||||
assert [message.id for message in final.values["messages"]] == ["earlier", "earlier-answer", "redo", "replacement"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_checkpoint_replay_honors_an_explicit_new_goal(monkeypatch, mode):
|
||||
record, states, final, explicit_goal, _head, _outcome = await _replay(monkeypatch, mode=mode, current_kind="active", explicit_goal=True)
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert states[0]["goal"] == explicit_goal
|
||||
assert final.values["goal"] == explicit_goal
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_checkpoint_replay_reads_goal_at_execution_not_preparation(monkeypatch, mode):
|
||||
record, states, final, _goal, _head, _outcome = await _replay(monkeypatch, mode=mode, current_kind="active", late_goal_update="clear")
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert states[0].get("goal") is None
|
||||
assert final.values.get("goal") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_checkpoint_replay_uses_a_goal_edited_after_preparation(monkeypatch, mode):
|
||||
record, states, final, current_goal, _head, _outcome = await _replay(monkeypatch, mode=mode, current_kind="active", late_goal_update="change")
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert states[0]["goal"] == current_goal
|
||||
assert final.values["goal"] == current_goal
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_current_head_resume_retains_existing_goal_outcome(monkeypatch, mode):
|
||||
record, states, final, _goal, _head, outcome = await _replay(monkeypatch, mode=mode, current_kind="achieved", select_head=True)
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert states[0]["goal_outcome"] == outcome
|
||||
assert final.values["goal_outcome"] == outcome
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_cancelled_checkpoint_replay_restores_pre_run_goal_and_outcome(monkeypatch, mode):
|
||||
record, _states, final, _goal, head, _outcome = await _replay(monkeypatch, mode=mode, current_kind="achieved", rollback=True)
|
||||
|
||||
assert record.status == RunStatus.error
|
||||
assert final.values.get("goal") == head.values.get("goal")
|
||||
assert final.values.get("goal_outcome") == head.values.get("goal_outcome")
|
||||
assert [message.id for message in final.values["messages"]] == [message.id for message in head.values["messages"]]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
async def test_checkpoint_replay_does_not_restore_a_cleaned_scheduled_goal(monkeypatch, mode):
|
||||
record, states, final, _goal, _head, _outcome = await _replay(monkeypatch, mode=mode, current_kind="recovered-scheduled")
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert states[0].get("goal") is None
|
||||
assert final.values.get("goal") is None
|
||||
|
||||
|
||||
async def test_historical_full_checkpoint_command_keeps_interrupt_resume_protocol(monkeypatch):
|
||||
saver = InMemorySaver()
|
||||
thread_id = "goal-interrupt-command"
|
||||
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
||||
original_goal = build_goal_state("Finish after approval", now="2026-10-01T00:00:00Z")
|
||||
resumed_answers = []
|
||||
|
||||
async def answer(_state):
|
||||
approved = interrupt("Please approve this step")
|
||||
resumed_answers.append(approved)
|
||||
return {"messages": [AIMessage(id="approved-answer", content=f"Approval: {approved}")]}
|
||||
|
||||
builder = StateGraph(get_thread_state_schema("full"))
|
||||
builder.add_node("answer", answer)
|
||||
builder.set_entry_point("answer")
|
||||
builder.set_finish_point("answer")
|
||||
graph = builder.compile(checkpointer=saver)
|
||||
accessor = CheckpointStateAccessor.bind(graph, saver, mode="full")
|
||||
await graph.ainvoke({"messages": [HumanMessage(id="approval-request", content="Do the approved step")], "goal": original_goal}, config)
|
||||
paused = await accessor.aget(config)
|
||||
assert paused.next == ("answer",)
|
||||
assert paused.tasks[0].interrupts
|
||||
|
||||
# A legitimate goal write advances the head while the graph is paused.
|
||||
# Resume still targets the addressable checkpoint with its interrupt.
|
||||
await write_thread_goal(saver, thread_id, {**original_goal, "continuation_count": 2})
|
||||
head = await accessor.aget(config)
|
||||
assert head.config["configurable"]["checkpoint_id"] != paused.config["configurable"]["checkpoint_id"]
|
||||
stream = Mock(wraps=graph.astream)
|
||||
monkeypatch.setattr(graph, "astream", stream)
|
||||
monkeypatch.setattr(worker, "_prepare_goal_continuation_input", AsyncMock(return_value=None))
|
||||
manager = RunManager()
|
||||
record = await manager.create(thread_id)
|
||||
command = Command(resume="approved")
|
||||
await worker.run_agent(
|
||||
SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock()),
|
||||
manager,
|
||||
record,
|
||||
ctx=worker.RunContext(checkpointer=saver, checkpoint_channel_mode="full"),
|
||||
agent_factory=lambda config: graph,
|
||||
graph_input=command,
|
||||
config=paused.config,
|
||||
stream_modes=["values"],
|
||||
)
|
||||
|
||||
assert record.status == RunStatus.success
|
||||
assert stream.call_args.args[0] is command
|
||||
assert resumed_answers == ["approved"]
|
||||
final = await accessor.aget(config)
|
||||
assert final.next == ()
|
||||
assert final.values["messages"][-1].content == "Approval: approved"
|
||||
275
backend/tests/test_goal_input_validation.py
Normal file
275
backend/tests/test_goal_input_validation.py
Normal file
@ -0,0 +1,275 @@
|
||||
"""Goal writes reject unusable objectives and counters before admission."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from contextlib import asynccontextmanager
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
from _router_auth_helpers import make_authed_test_app
|
||||
from fastapi import HTTPException
|
||||
from fastapi.testclient import TestClient
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.graph import StateGraph
|
||||
from langgraph.types import Overwrite
|
||||
|
||||
from app.gateway import services
|
||||
from app.gateway.auth.models import User
|
||||
from app.gateway.routers import runs, thread_runs, threads
|
||||
from deerflow.agents.thread_state import get_thread_state_schema
|
||||
from deerflow.utils.goal_objective import MAX_GOAL_OBJECTIVE_CHARS
|
||||
|
||||
_JSON_OVERWRITE_GOAL = {
|
||||
"objective": "outer objective",
|
||||
"type": "__overwrite__",
|
||||
"value": {"status": "active", "continuation_count": "many"},
|
||||
}
|
||||
_KEYED_OVERWRITE_GOAL = {"objective": "outer objective", "__overwrite__": {"status": "active", "continuation_count": "many"}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("boundary", [services.normalize_input, services.strip_server_owned_state_metadata])
|
||||
@pytest.mark.parametrize(
|
||||
"goal",
|
||||
[
|
||||
{"status": "active"},
|
||||
{"objective": None},
|
||||
{"objective": 42},
|
||||
{"objective": ""},
|
||||
{"objective": " \n\t "},
|
||||
{"objective": "x" * (MAX_GOAL_OBJECTIVE_CHARS + 1)},
|
||||
"finish",
|
||||
[],
|
||||
],
|
||||
)
|
||||
def test_external_goal_requires_a_usable_objective(boundary, goal):
|
||||
with pytest.raises(HTTPException) as rejected:
|
||||
boundary({"goal": goal})
|
||||
assert rejected.value.status_code == 400
|
||||
assert "goal" in rejected.value.detail.lower()
|
||||
assert "x" * 100 not in rejected.value.detail
|
||||
|
||||
|
||||
@pytest.mark.parametrize("boundary", [services.normalize_input, services.strip_server_owned_state_metadata])
|
||||
@pytest.mark.parametrize("field", ["continuation_count", "max_continuations", "no_progress_count", "max_no_progress_continuations"])
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
[
|
||||
pytest.param("many", id="text"),
|
||||
pytest.param("2", id="numeric-string"),
|
||||
pytest.param(-1, id="negative"),
|
||||
pytest.param(1.5, id="float"),
|
||||
pytest.param(True, id="boolean"),
|
||||
pytest.param(None, id="null"),
|
||||
],
|
||||
)
|
||||
def test_external_goal_rejects_invalid_present_counters(boundary, field, value):
|
||||
source = {"goal": {"objective": "finish", field: value}}
|
||||
with pytest.raises(HTTPException) as rejected:
|
||||
boundary(source)
|
||||
assert rejected.value.status_code == 400
|
||||
assert f"goal.{field}" in rejected.value.detail
|
||||
assert "many" not in rejected.value.detail
|
||||
assert source == {"goal": {"objective": "finish", field: value}}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("boundary", [services.normalize_input, services.strip_server_owned_state_metadata])
|
||||
@pytest.mark.parametrize(
|
||||
"goal",
|
||||
[
|
||||
pytest.param(_JSON_OVERWRITE_GOAL, id="json-overwrite"),
|
||||
pytest.param(_KEYED_OVERWRITE_GOAL, id="reserved-key"),
|
||||
pytest.param({"objective": "finish", "type": "__overwrite__"}, id="marker-without-value"),
|
||||
pytest.param({"objective": "finish", "type": "__overwrite__", "value": None}, id="overwrite-clear"),
|
||||
],
|
||||
)
|
||||
def test_external_goal_rejects_reducer_control_markers(boundary, goal):
|
||||
with pytest.raises(HTTPException) as rejected:
|
||||
boundary({"goal": goal})
|
||||
assert rejected.value.status_code == 400
|
||||
assert "goal" in rejected.value.detail
|
||||
assert "outer objective" not in rejected.value.detail
|
||||
assert "many" not in rejected.value.detail
|
||||
|
||||
|
||||
@pytest.mark.parametrize("boundary", [services.normalize_input, services.strip_server_owned_state_metadata])
|
||||
def test_external_goal_keeps_ordinary_type_and_value_fields(boundary):
|
||||
goal = {"objective": "finish", "continuation_count": 7, "max_continuations": 8, "type": "business", "value": {"continuation_count": "many"}}
|
||||
assert boundary({"goal": goal})["goal"] == goal
|
||||
|
||||
|
||||
def test_trusted_internal_goal_input_preserves_typed_overwrite():
|
||||
overwrite = Overwrite({"objective": "finish", "continuation_count": 7, "max_continuations": 8})
|
||||
assert services.normalize_input({"goal": overwrite}, trusted_internal=True)["goal"] is overwrite
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize("initial_goal", [None, {"objective": "previous", "status": "active", "continuation_count": 7, "max_continuations": 8}], ids=["cleared", "active"])
|
||||
async def test_run_input_rejects_overwrite_before_mutating_existing_goal(mode, initial_goal):
|
||||
checkpointer = InMemorySaver()
|
||||
builder = StateGraph(get_thread_state_schema(mode))
|
||||
builder.add_node("finish", lambda _state: {})
|
||||
builder.set_entry_point("finish")
|
||||
builder.set_finish_point("finish")
|
||||
graph = builder.compile(checkpointer=checkpointer)
|
||||
config = {"configurable": {"thread_id": "goal-overwrite-input", "checkpoint_ns": ""}}
|
||||
# An initialized goal channel is required to exercise LangGraph's sentinel
|
||||
# unpacking; the first write into an unset channel keeps the outer mapping.
|
||||
await graph.ainvoke({"goal": initial_goal}, config)
|
||||
before = await checkpointer.aget_tuple(config)
|
||||
assert before is not None
|
||||
|
||||
with pytest.raises(HTTPException) as rejected:
|
||||
await graph.ainvoke(services.normalize_input({"goal": _JSON_OVERWRITE_GOAL}), config)
|
||||
assert rejected.value.status_code == 400
|
||||
after = await checkpointer.aget_tuple(config)
|
||||
assert after is not None
|
||||
assert after.config == before.config
|
||||
assert after.checkpoint == before.checkpoint
|
||||
assert (await graph.aget_state(config)).values["goal"] == initial_goal
|
||||
|
||||
|
||||
@pytest.mark.parametrize("boundary", [services.normalize_input, services.strip_server_owned_state_metadata])
|
||||
@pytest.mark.parametrize(
|
||||
"counters",
|
||||
[
|
||||
{"continuation_count": 0, "max_continuations": 0, "no_progress_count": 0, "max_no_progress_continuations": 0},
|
||||
{"continuation_count": 7, "max_continuations": 8, "no_progress_count": 1, "max_no_progress_continuations": 2},
|
||||
{"continuation_count": 9, "max_continuations": 8, "no_progress_count": 3, "max_no_progress_continuations": 2},
|
||||
],
|
||||
ids=["zero", "remaining-budget", "used-budget"],
|
||||
)
|
||||
def test_external_goal_preserves_valid_counters_without_resetting_budget(boundary, counters):
|
||||
goal = {"objective": "finish", **counters}
|
||||
assert boundary({"goal": goal})["goal"] == goal
|
||||
|
||||
|
||||
def test_trusted_internal_goal_input_keeps_existing_restore_protocol():
|
||||
source = {"goal": {"objective": "finish", "continuation_count": "many"}}
|
||||
assert services.normalize_input(source, trusted_internal=True) == source
|
||||
|
||||
|
||||
@pytest.mark.parametrize("boundary", [services.normalize_input, services.strip_server_owned_state_metadata])
|
||||
def test_valid_sparse_goals_are_normalized_without_building_fresh_state(boundary):
|
||||
sparse = {"objective": " finish\n the report "}
|
||||
source = {"goal": sparse}
|
||||
admitted = boundary(source)
|
||||
|
||||
assert admitted["goal"] == {"objective": "finish the report"}
|
||||
assert source == {"goal": {"objective": " finish\n the report "}}
|
||||
assert set(admitted["goal"]) == {"objective"}
|
||||
assert boundary({"goal": None}) == {"goal": None}
|
||||
assert boundary({"title": "unrelated update"}) == {"title": "unrelated update"}
|
||||
|
||||
|
||||
def test_goal_replacement_preserves_supplied_counters_and_clears_previous_outcome():
|
||||
goal = {"objective": "finish", "status": "active", "continuation_count": 7, "max_continuations": 8}
|
||||
assert services.normalize_input({"goal": goal}) == {"goal": goal, "goal_outcome": None}
|
||||
assert services.normalize_input({"goal": None}) == {"goal": None}
|
||||
assert services.normalize_input({"goal": goal}, trusted_internal=True) == {"goal": goal}
|
||||
|
||||
|
||||
class _StateRunManager:
|
||||
def __init__(self):
|
||||
self.reservations = 0
|
||||
|
||||
@asynccontextmanager
|
||||
async def reserve_thread_operation(self, *_args, **_kwargs):
|
||||
self.reservations += 1
|
||||
yield
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize(
|
||||
"invalid_goal",
|
||||
[
|
||||
pytest.param({"status": "active"}, id="missing-objective"),
|
||||
pytest.param({"objective": "finish", "continuation_count": "many"}, id="invalid-count"),
|
||||
pytest.param({"objective": "finish", "max_continuations": True}, id="boolean-limit"),
|
||||
pytest.param({"objective": "finish", "no_progress_count": -1}, id="negative-count"),
|
||||
pytest.param({"objective": "finish", "max_no_progress_continuations": None}, id="null-limit"),
|
||||
pytest.param(_JSON_OVERWRITE_GOAL, id="json-overwrite"),
|
||||
pytest.param(_KEYED_OVERWRITE_GOAL, id="reserved-key"),
|
||||
],
|
||||
)
|
||||
def test_state_route_rejects_invalid_goal_without_changing_checkpoint(monkeypatch, mode, invalid_goal):
|
||||
user = User(email="goal-input@example.com", password_hash="x", id=UUID("00000000-0000-0000-0000-000000000001"))
|
||||
app = make_authed_test_app(user_factory=lambda: user, bind_current_user=True)
|
||||
app.include_router(threads.router)
|
||||
app.state.checkpointer = checkpointer = InMemorySaver()
|
||||
app.state.checkpoint_channel_mode = mode
|
||||
app.state.run_manager = manager = _StateRunManager()
|
||||
config = {"configurable": {"thread_id": "goal-input-http", "checkpoint_ns": ""}}
|
||||
|
||||
# Keep the real state mutation graph and saver; assistant-schema resolution
|
||||
# is unrelated to goal admission, so use the base schema here.
|
||||
async def build_accessor(request, *, thread_id, as_node, checkpoint_id=None):
|
||||
return services.build_checkpoint_state_mutation_accessor(request, thread_id=thread_id, as_node=as_node, checkpoint_id=checkpoint_id)
|
||||
|
||||
builder = AsyncMock(side_effect=build_accessor)
|
||||
monkeypatch.setattr(threads, "build_thread_checkpoint_state_mutation_accessor", builder)
|
||||
|
||||
with TestClient(app) as client:
|
||||
valid = client.post("/api/threads/goal-input-http/state", json={"values": {"goal": {"objective": "finish", "status": "active"}}})
|
||||
assert valid.status_code == 200, valid.text
|
||||
assert valid.json()["values"]["goal"] == {"objective": "finish", "status": "active"}
|
||||
before = asyncio.run(checkpointer.aget_tuple(config))
|
||||
assert before is not None
|
||||
builder.reset_mock()
|
||||
reservations = manager.reservations
|
||||
|
||||
rejected = client.post("/api/threads/goal-input-http/state", json={"values": {"goal": invalid_goal, "title": "must not write"}})
|
||||
assert rejected.status_code == 400, rejected.text
|
||||
builder.assert_not_awaited()
|
||||
assert manager.reservations == reservations
|
||||
after = asyncio.run(checkpointer.aget_tuple(config))
|
||||
assert after is not None
|
||||
assert after.config == before.config
|
||||
assert after.checkpoint == before.checkpoint
|
||||
|
||||
cleared = client.post("/api/threads/goal-input-http/state", json={"values": {"goal": None}})
|
||||
assert cleared.status_code == 200, cleared.text
|
||||
assert cleared.json()["values"].get("goal") is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
[
|
||||
"/api/threads/goal-input-http/runs",
|
||||
"/api/threads/goal-input-http/runs/stream",
|
||||
"/api/threads/goal-input-http/runs/wait",
|
||||
"/api/runs/stream",
|
||||
"/api/runs/wait",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"invalid_goal",
|
||||
[
|
||||
pytest.param({"status": "active"}, id="missing-objective"),
|
||||
pytest.param({"objective": "finish", "continuation_count": "many"}, id="invalid-counter"),
|
||||
pytest.param(_JSON_OVERWRITE_GOAL, id="json-overwrite"),
|
||||
pytest.param(_KEYED_OVERWRITE_GOAL, id="reserved-key"),
|
||||
],
|
||||
)
|
||||
def test_run_routes_reject_invalid_goal_before_admission(monkeypatch, path, invalid_goal):
|
||||
app = make_authed_test_app()
|
||||
app.include_router(runs.router)
|
||||
app.include_router(thread_runs.router)
|
||||
app.state.stream_bridge = SimpleNamespace()
|
||||
app.state.run_manager = SimpleNamespace(create_or_reject=AsyncMock())
|
||||
monkeypatch.setattr(services, "get_run_context", lambda _request: SimpleNamespace(thread_store=app.state.thread_store))
|
||||
factory = Mock(side_effect=AssertionError("invalid goal reached graph assembly"))
|
||||
monkeypatch.setattr(services, "resolve_agent_factory", factory)
|
||||
worker = AsyncMock()
|
||||
monkeypatch.setattr(services, "run_agent", worker)
|
||||
|
||||
with TestClient(app, raise_server_exceptions=False) as client:
|
||||
rejected = client.post(path, json={"input": {"goal": invalid_goal}})
|
||||
|
||||
assert rejected.status_code == 400, rejected.text
|
||||
factory.assert_not_called()
|
||||
app.state.run_manager.create_or_reject.assert_not_awaited()
|
||||
worker.assert_not_awaited()
|
||||
Loading…
x
Reference in New Issue
Block a user