From 734f9e0e92d91a809603c3e6b204c620755602da Mon Sep 17 00:00:00 2001 From: Daoyuan Li <94409450+DaoyuanLi2816@users.noreply.github.com> Date: Wed, 7 Oct 2026 02:20:11 -0700 Subject: [PATCH] fix(runtime): omit leading reasoning from run summaries (#6419) --- README.md | 2 + backend/app/gateway/routers/thread_runs.py | 6 +- .../harness/deerflow/runtime/AGENTS.md | 8 +- .../harness/deerflow/runtime/journal.py | 3 +- .../packages/harness/deerflow/utils/AGENTS.md | 3 + .../harness/deerflow/utils/llm_text.py | 27 ++++ .../tests/test_run_journal_visible_summary.py | 126 ++++++++++++++++++ .../tests/test_thread_regenerate_prepare.py | 18 ++- 8 files changed, 182 insertions(+), 11 deletions(-) create mode 100644 backend/tests/test_run_journal_visible_summary.py diff --git a/README.md b/README.md index cc9d79ed3b..01f443b81d 100644 --- a/README.md +++ b/README.md @@ -2485,6 +2485,8 @@ Current MVP capabilities: - When `channel_connections.enabled: true`, send scheduled task updates to the task owner's connected IM identities (outbox + delivery worker) on apps that support proactive push, which today is WeCom; Settings shows for each app whether updates are sent there, and other apps get none. Each occurrence sends at most one message: a finished, failed or goal-missed run, the automatic pause after three missed goals, a pause by the agent (its stop condition was met) or the task finishing (all `max_runs` done, `end_at` reached); when several apply, the pause or finish wins and still says how the last run went. A one-time task sends only its run's outcome. The message is queued in the same database transaction that records the outcome, so runs finalized after a crash or a lost lease notify exactly once. It reads on its own, in your web UI language (else `channel_connections.notification_locale`): the task title, what happened, a one-line result when the agent replied, and "Open DeerFlow → Scheduled tasks for details.", with no IDs and no links. Plain manual "run now" trials and interrupted runs stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, interrupted by a restart). Channel/transport outages park deliveries without exhausting retries, for up to about a day; platform rejections retry for roughly 15 minutes before settling as `failed`. An identity you disconnect while a delivery is waiting is never pushed to: the row is dropped as `failed`. - Browse execution history in pages of 50; older pages pause automatic refresh, with an explicit return to the latest runs. Counts appear only after a successful read; loading and failed reads are not reported as zero runs. +Derived run summaries omit leading inline `` reasoning before limiting the text, so a long thought does not replace the actual answer in previews or scheduled notices. Original response history and literal tags in the answer are unchanged; previously stored run summaries are not rewritten. + **Filter execution history through the API** To inspect failures without downloading every successful occurrence, authenticated clients with `threads:read` can request `GET /api/scheduled-tasks/{task_id}/runs?status=failed&limit=50&offset=0` for an owned task. The optional `status` accepts `queued`, `launching`, `running`, `success`, `failed`, `skipped`, `interrupted`, or `unmet`; these are occurrence statuses, so task statuses such as `completed` are invalid (422). diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 149c6e7e6f..86d8f833e0 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -69,6 +69,7 @@ from deerflow.runtime import CancelOutcome, ConflictError, RunRecord, RunStatus, from deerflow.runtime.runs.store.base import format_run_cursor_created_at, normalize_run_created_at_iso from deerflow.runtime.secret_context import redact_config_secrets, redact_metadata_secrets from deerflow.runtime.user_context import get_effective_user_id +from deerflow.utils.llm_text import strip_leading_think_blocks from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY, get_original_user_content_text, message_to_text from deerflow.utils.thread_id import ThreadId from deerflow.workspace_changes import get_workspace_changes_response @@ -592,10 +593,11 @@ def _run_last_ai_matches_message(record: RunRecord, message: Any) -> bool: last_ai_message = (record.last_ai_message or "").strip() if not last_ai_message: return False - target_text = _message_text(message).strip() + target_text = _message_text(message) if not target_text: return False - return last_ai_message == target_text[: len(last_ai_message)] + # Match both historical raw summaries and newer visible-answer summaries. + return any(last_ai_message == text[: len(last_ai_message)] for text in (target_text.strip(), strip_leading_think_blocks(target_text))) async def _find_target_run_id( diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 49ab468ce1..040c3c3468 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -28,10 +28,10 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c **Thread message cursors:** `list_messages` applies both exclusive bounds before `limit`, paging forward whenever `after_seq` is supplied. -**Human-input capture** (`runtime/journal.py`): track capture separately from -the optional display summary. Image-only input has no text but must still stop -the batch scan and later model calls from appending another human-input event. -`tests/test_run_journal.py` covers callback and full/delta graph paths. +**Journal capture**: Image-only input stops batch scans and later human-event +capture despite an empty summary. AI summaries strip leading think sections +before the 2000-character cap, preserving events, literal answer tags, and +prior useful summaries. Tests: `test_run_journal*.py` (callbacks, full/delta). **Per-call LLM telemetry** (`runtime/journal.py`): `RunJournal` adds observation-only keys to `llm.ai.response` metadata and to `llm.error` metadata (previously empty): `langchain_run_id`, diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 488189ac79..ae5076d7ea 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -47,6 +47,7 @@ from deerflow.runtime.events.catalog import ( RUN_ERROR_EVENT, RUN_START_EVENT, ) +from deerflow.utils.llm_text import strip_leading_think_blocks from deerflow.utils.messages import message_to_text, restore_original_human_message if TYPE_CHECKING: @@ -411,7 +412,7 @@ class RunJournal(BaseCallbackHandler): is_ai_message = isinstance(message, AIMessage) or getattr(message, "type", None) == "ai" if not is_ai_message or (caller is not None and caller != "lead_agent"): return None - text = self._message_text(message).strip() + text = strip_leading_think_blocks(self._message_text(message)) return text[:2000] if text else None def _record_message_summary(self, message: BaseMessage, *, caller: str | None = None) -> None: diff --git a/backend/packages/harness/deerflow/utils/AGENTS.md b/backend/packages/harness/deerflow/utils/AGENTS.md index 10f9fdbbe9..98df8ca010 100644 --- a/backend/packages/harness/deerflow/utils/AGENTS.md +++ b/backend/packages/harness/deerflow/utils/AGENTS.md @@ -21,6 +21,9 @@ suffix. Keep its case-insensitive tag handling, optional whitespace before the closing `>`, and the `truncate_unclosed` behavior. Regression coverage lives in `tests/test_utils_llm_text.py`. +Display summaries use `strip_leading_think_blocks` before limiting text; +`test_run_journal_visible_summary.py` preserves literal tags in the answer. + ### Agent / Tool Assembly Off-Load `start_run` also resolves the factory through `run_assembly` before admission: diff --git a/backend/packages/harness/deerflow/utils/llm_text.py b/backend/packages/harness/deerflow/utils/llm_text.py index 9dd1ec24db..e7ee061b64 100644 --- a/backend/packages/harness/deerflow/utils/llm_text.py +++ b/backend/packages/harness/deerflow/utils/llm_text.py @@ -57,6 +57,33 @@ def strip_think_blocks(text: str, *, truncate_unclosed: bool = True) -> str: return "".join(parts).strip() +def strip_leading_think_blocks(text: str) -> str: + """Remove leading reasoning for display summaries, leaving tags in the answer intact. + + Unlike structured-response parsing, a displayed answer may explain the + tag in prose or code. Only a leading XML-style reasoning section is + removed; an unfinished section has no answer to summarize. + """ + start = 0 + while True: + whitespace_start = start + while start < len(text) and text[start].isspace(): + start += 1 + newline = text.rfind("\n", whitespace_start, start) + indent = text[max(newline + 1, whitespace_start) : start] + if (whitespace_start == 0 or newline >= 0) and ("\t" in indent or indent.startswith(" ")): + break # Root-level indented Markdown code is a literal example. + opening = _THINK_OPEN_PREFIX_RE.match(text, start) + if opening is None or (opening.end() < len(text) and not (text[opening.end()].isspace() or text[opening.end()] == ">")): + break + end = text.find(">", opening.end()) + closing = _find_think_close(text, end + 1) if end >= 0 else None + if closing is None: + return "" + start = closing[1] + return text[start:].strip() + + def strip_markdown_code_fence(text: str) -> str: """Remove a single wrapping markdown code fence when present.""" stripped = text.strip() diff --git a/backend/tests/test_run_journal_visible_summary.py b/backend/tests/test_run_journal_visible_summary.py new file mode 100644 index 0000000000..a6bbfb39fd --- /dev/null +++ b/backend/tests/test_run_journal_visible_summary.py @@ -0,0 +1,126 @@ +"""Derived run summaries exclude leading reasoning without rewriting messages.""" + +from uuid import uuid4 + +import httpx +import pytest +from langchain_core.messages import AIMessage, HumanMessage +from langchain_core.outputs import ChatGeneration, LLMResult +from langchain_openai import ChatOpenAI + +from app.scheduler.notification_text import render_notification_text +from deerflow.persistence.scheduled_task_runs.sql import run_summary +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import RunJournal + +ANSWER = "The release checklist is complete." + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("content", "expected"), + [ + (f"Compare candidates.{ANSWER}", ANSWER), + (f"\n\nCompare candidates.\n\n{ANSWER}", ANSWER), + (f'Compare candidates.\n{ANSWER}', ANSWER), + (f"First.\nSecond.\n{ANSWER}", ANSWER), + (f"First. Second.{ANSWER}", ANSWER), + ("" + "r" * 2400 + f"\n{ANSWER}", ANSWER), + ("Only reasoning.", None), + ("Unfinished reasoning.", None), + ("literal example\n```", "```xml\nliteral example\n```"), + ("`literal example`", "`literal example`"), + ("Use literal example in this document.", "Use literal example in this document."), + (" indented code", "indented code"), + ("ordinary XML", "ordinary XML"), + ("Reasoning.\n`literal example`", "`literal example`"), + ([{"type": "text", "text": "Reasoning."}, {"type": "text", "text": ANSWER}], ANSWER), + (f"Reasoning.{'a' * 2500}", "a" * 2000), + ], + ids=[ + "short", + "multiline", + "attributes", + "two-blocks", + "same-line-blocks", + "long-reasoning", + "only-reasoning", + "unclosed", + "partial-tag", + "plain", + "fenced-literal", + "inline-literal", + "prose-literal", + "indented-literal", + "not-tag", + "literal-after-reasoning", + "content-blocks", + "long-answer", + ], +) +async def test_visible_summary_preserves_original_response(content, expected): + store = MemoryRunEventStore() + journal = RunJournal("run-visible", "thread-visible", store) + message = AIMessage(content=content, usage_metadata={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}) + original = message.model_dump() + journal.on_llm_end(LLMResult(generations=[[ChatGeneration(message=message)]]), run_id=uuid4(), tags=["lead_agent"]) + await journal.flush() + + completion = journal.get_completion_data() + assert completion["last_ai_message"] == expected + assert completion["message_count"] == 1 + assert completion["total_tokens"] == 15 + events = await store.list_events("thread-visible", "run-visible") + response = next(event for event in events if event["event_type"] == "llm.ai.response") + assert response["content"]["content"] == content + assert message.model_dump() == original + + +@pytest.mark.anyio +async def test_thought_only_response_keeps_previous_useful_summary(): + journal = RunJournal("run-visible", "thread-visible", MemoryRunEventStore()) + for content in (ANSWER, "Unfinished reasoning."): + journal.on_llm_end(LLMResult(generations=[[ChatGeneration(message=AIMessage(content=content))]]), run_id=uuid4(), tags=["lead_agent"]) + await journal.flush() + assert journal.get_completion_data()["last_ai_message"] == ANSWER + assert journal.get_completion_data()["message_count"] == 2 + + +@pytest.mark.anyio +async def test_openai_compatible_reply_reaches_notice_without_inline_reasoning(): + content = "" + "r" * 2400 + f"\n{ANSWER}" + requests = [] + + def respond(request): + assert request.method == "POST" and request.url.path == "/v1/chat/completions" + requests.append(request) + return httpx.Response( + 200, + json={ + "id": "offline-response", + "object": "chat.completion", + "created": 0, + "model": "reasoning-model", + "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": content}}], + "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}, + }, + ) + + store = MemoryRunEventStore() + journal = RunJournal("run-visible", "thread-visible", store) + async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client: + model = ChatOpenAI(model="reasoning-model", api_key="offline-test-key", base_url="https://model.invalid/v1", http_async_client=client) + message = await model.ainvoke([HumanMessage(content="Check the release checklist")], config={"callbacks": [journal], "tags": ["lead_agent"]}) + await journal.flush() + summary = journal.get_completion_data()["last_ai_message"] + notice = render_notification_text({"event": "run_completed", "run_id": "run-visible", "payload": {"run_status": "success", "task_title": "Check release", "result_summary": run_summary(summary)}}) + + assert len(requests) == 1 + assert message.content == content + assert run_summary(summary) == ANSWER + assert f"Result: {ANSWER}" in notice + assert "" not in notice and "rrrr" not in notice + events = await store.list_events("thread-visible", "run-visible") + assert any(event["content"].get("content") == content for event in events if event["event_type"] == "llm.ai.response") diff --git a/backend/tests/test_thread_regenerate_prepare.py b/backend/tests/test_thread_regenerate_prepare.py index 8121ed6348..a92fe652d9 100644 --- a/backend/tests/test_thread_regenerate_prepare.py +++ b/backend/tests/test_thread_regenerate_prepare.py @@ -12,7 +12,7 @@ from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, Tool from langgraph.checkpoint.base import empty_checkpoint, uuid6 from langgraph.checkpoint.memory import InMemorySaver -from deerflow.runtime import RunStatus +from deerflow.runtime import DisconnectMode, RunRecord, RunStatus from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY @@ -1541,18 +1541,28 @@ def test_prepare_regenerate_payload_rejects_non_latest_assistant(): assert exc.value.detail == "Only the latest assistant message can be regenerated" -def test_prepare_regenerate_payload_falls_back_to_matching_run_when_events_are_missing(): +@pytest.mark.parametrize( + ("content", "summary"), + [ + ("answer", "answer"), + ("Reasoning.answer", "Reasoning.answer"), + ("Reasoning.answer", "answer"), + ("" + "r" * 2400 + "" + "a" * 2500, "a" * 2000), + ], + ids=["plain", "legacy-summary", "visible-summary", "long-reasoning"], +) +def test_prepare_regenerate_payload_falls_back_to_matching_run_when_events_are_missing(content, summary): from app.gateway.routers.thread_runs import _prepare_regenerate_payload human = HumanMessage(id="human-1", content="question") - ai = AIMessage(id="ai-1", content="answer") + ai = AIMessage(id="ai-1", content=content) base = _checkpoint("ckpt-base", []) after_human = _checkpoint("ckpt-human", [human]) latest = _checkpoint("ckpt-ai", [human, ai]) checkpointer = FakeCheckpointer([latest, after_human, base]) run_manager = FakeRunManager( [ - SimpleNamespace(run_id="run-latest", status=RunStatus.success, last_ai_message="answer"), + RunRecord(run_id="run-latest", thread_id="thread-1", assistant_id="lead_agent", status=RunStatus.success, on_disconnect=DisconnectMode.continue_, last_ai_message=summary), SimpleNamespace(run_id="run-older", status=RunStatus.error, last_ai_message="answer"), ] )