diff --git a/backend/AGENTS.md b/backend/AGENTS.md index f460fc239..73caca7c8 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -427,7 +427,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores ` | **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`); `POST /reload` - admin-only process-local prompt-cache invalidation after trusted external filesystem changes | | **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data | | **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); `GET /list` - list; `DELETE /{filename}` - delete | -| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - create a new main-thread branch from a completed assistant turn checkpoint and, when an addressable pre-user replay checkpoint exists, materialize it into the branch namespace so the inherited response remains regeneratable. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - manually summarize older active context into `summary_text` and retain the recent message window, blocked while a run is in flight; unexpected failures are logged server-side and return a generic 500 detail | +| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - create a new main-thread branch from a completed assistant turn checkpoint and, when an addressable pre-user replay checkpoint exists, materialize it into the branch namespace so the inherited response remains regeneratable. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380). Seeded rows are grouped into one synthetic run per inherited turn (`branch-seed-{thread_id}-{n}`, a new turn opening at every persisted human message, including an allowlisted hidden `ask_clarification` reply) because `run_id` is a turn identity to the feed's consumers, not a provenance tag: regenerating an inherited answer supersedes that row's whole `run_id` in `GET /messages/page`, so one shared id for the entire seed deleted the complete inherited history on a branch's first regenerate (#4458); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - manually summarize older active context into `summary_text` and retain the recent message window, blocked while a run is in flight; unexpected failures are logged server-side and return a generic 500 detail | | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - serve artifacts; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index b5ba42527..0e323a18b 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -888,7 +888,7 @@ async def branch_thread(thread_id: str, body: ThreadBranchRequest, request: Requ seed_events = build_branch_history_seed_events( _checkpoint_messages(snapshot), thread_id=new_thread_id, - run_id=f"branch-seed-{new_thread_id}", + run_id_prefix=f"branch-seed-{new_thread_id}", parent_thread_id=thread_id, ) if seed_events: diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index bf5706a33..2112fcffc 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -91,7 +91,7 @@ def build_branch_history_seed_events( messages: Sequence[Any], *, thread_id: str, - run_id: str, + run_id_prefix: str, parent_thread_id: str, ) -> list[dict]: """Serialize a branch checkpoint's messages into run-event message rows. @@ -104,6 +104,18 @@ def build_branch_history_seed_events( checkpoint snapshot the branch was created from keeps the feed consistent with what the branch actually contains. + Rows are grouped into one synthetic run per inherited turn + (``{run_id_prefix}-{n}``), a new turn starting at every persisted human + message — the same boundary a real run has, since a run begins with a + human input (including the allowlisted hidden ``ask_clarification`` + reply, which resumes as its own run). ``run_id`` is a *turn* identity to + the feed's consumers, not merely a provenance tag: regenerating the last + inherited answer resolves that row's ``run_id`` as the superseded source + (``_find_target_run_id``) and ``GET /messages/page`` then drops **every** + row carrying it. One shared id for the whole seed therefore deleted the + complete inherited history on the branch's first regenerate (#4458); one + id per turn confines the drop to the turn actually regenerated. + Mirrors RunJournal's message-event contract so seeded rows are indistinguishable from journaled ones except by the ``branch_seed`` marker: same event types, ``category="message"``, ``content= @@ -125,6 +137,8 @@ def build_branch_history_seed_events( events: list[dict] = [] created_at = datetime.now(UTC).isoformat() seed_metadata = {"branch_seed": True, "branch_parent_thread_id": parent_thread_id} + # Messages ahead of the first human turn (none in practice) stay in turn 0. + turn_index = 0 for raw_message in messages: message = _coerce_seed_message(raw_message) if not isinstance(message, BaseMessage): @@ -132,6 +146,7 @@ def build_branch_history_seed_events( if isinstance(message, HumanMessage): if not _should_persist_human_input_message(message): continue + turn_index += 1 event_type = "llm.human.input" content = restore_original_human_message(message).model_dump() metadata: dict[str, Any] = {"caller": "lead_agent", **seed_metadata} @@ -149,7 +164,7 @@ def build_branch_history_seed_events( events.append( { "thread_id": thread_id, - "run_id": run_id, + "run_id": f"{run_id_prefix}-{turn_index}", "event_type": event_type, "category": "message", "content": content, diff --git a/backend/tests/test_branch_history_seed.py b/backend/tests/test_branch_history_seed.py index 4d391910d..c5ca0de88 100644 --- a/backend/tests/test_branch_history_seed.py +++ b/backend/tests/test_branch_history_seed.py @@ -19,7 +19,7 @@ def _seed(messages): return build_branch_history_seed_events( messages, thread_id="branch-thread", - run_id="branch-seed-branch-thread", + run_id_prefix="branch-seed-branch-thread", parent_thread_id="parent-thread", ) @@ -36,7 +36,7 @@ def test_seed_serializes_visible_history_in_order() -> None: assert [event["event_type"] for event in events] == ["llm.human.input", "llm.ai.response", "llm.tool.result"] assert all(event["category"] == "message" for event in events) assert all(event["thread_id"] == "branch-thread" for event in events) - assert all(event["run_id"] == "branch-seed-branch-thread" for event in events) + assert all(event["run_id"] == "branch-seed-branch-thread-1" for event in events) assert [event["content"]["id"] for event in events] == ["h1", "a1", "t1"] # Human/AI rows carry the journal's caller tag; every row carries the # seed provenance marker. @@ -159,6 +159,60 @@ def test_seed_restores_original_user_content() -> None: assert "original_user_content" not in events[0]["content"]["additional_kwargs"] +def test_seed_scopes_one_synthetic_run_per_inherited_turn() -> None: + """``run_id`` is a turn identity to the feed, not a provenance tag: the + regenerate path supersedes a whole ``run_id``, so packing every inherited + turn into one id deleted the complete history on the first regenerate + (#4458). Each human message opens the next synthetic run.""" + events = _seed( + [ + HumanMessage(id="h1", content="turn one"), + AIMessage(id="a1", content="answer one"), + HumanMessage(id="h2", content="turn two"), + AIMessage(id="a2", content="calling a tool", tool_calls=[{"name": "search", "args": {}, "id": "call-1", "type": "tool_call"}]), + ToolMessage(id="t2", content="tool output", tool_call_id="call-1"), + AIMessage(id="a2b", content="answer two"), + ] + ) + + assert [(event["content"]["id"], event["run_id"]) for event in events] == [ + ("h1", "branch-seed-branch-thread-1"), + ("a1", "branch-seed-branch-thread-1"), + ("h2", "branch-seed-branch-thread-2"), + ("a2", "branch-seed-branch-thread-2"), + ("t2", "branch-seed-branch-thread-2"), + ("a2b", "branch-seed-branch-thread-2"), + ] + + +def test_seed_opens_a_new_turn_at_a_hidden_clarification_reply() -> None: + """An answered clarification resumes as its own run, so the reply is a turn + boundary exactly like a visible user message.""" + response = { + "version": 1, + "kind": "human_input_response", + "source": "ask_clarification", + "request_id": "req-1", + "value": "yes", + "response_kind": "text", + } + events = _seed( + [ + HumanMessage(id="h1", content="question"), + AIMessage(id="a1", content="which one?"), + HumanMessage(id="h-response", content="yes", additional_kwargs={"hide_from_ui": True, "human_input_response": response}), + AIMessage(id="a2", content="answer"), + ] + ) + + assert [event["run_id"] for event in events] == [ + "branch-seed-branch-thread-1", + "branch-seed-branch-thread-1", + "branch-seed-branch-thread-2", + "branch-seed-branch-thread-2", + ] + + def test_seed_roundtrips_through_memory_store_feed() -> None: """put_batch preserves order and list_messages returns the seeded feed.""" store = MemoryRunEventStore() diff --git a/backend/tests/test_thread_messages_page.py b/backend/tests/test_thread_messages_page.py index 246bd00c1..07bfb6129 100644 --- a/backend/tests/test_thread_messages_page.py +++ b/backend/tests/test_thread_messages_page.py @@ -9,10 +9,12 @@ from unittest.mock import AsyncMock, MagicMock import pytest from _router_auth_helpers import make_authed_test_app from fastapi.testclient import TestClient +from langchain_core.messages import AIMessage, HumanMessage from app.gateway.routers import thread_runs from deerflow.runtime import RunRecord from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import build_branch_history_seed_events def _make_app(event_store: MemoryRunEventStore, *, superseded: set[str] | None = None, records=None, feedback=None): @@ -185,6 +187,45 @@ def test_thread_page_filters_all_successfully_superseded_runs_before_filling(): assert body["next_before_seq"] is None +def test_thread_page_keeps_earlier_branch_turns_when_the_last_one_is_regenerated(): + """Regenerating a branch's inherited answer must not delete the turns before + it (#4458). + + Supersession is run-scoped, so it can only stay confined to the regenerated + turn while each seeded turn owns its own synthetic run id — this pins the + seed builder's turn scoping against the filter that consumes it. + """ + store = MemoryRunEventStore() + branch_thread = "thread-1" + + async def seed(): + seed_events = build_branch_history_seed_events( + [ + HumanMessage(id="h1", content="turn one"), + AIMessage(id="a1", content="answer one"), + HumanMessage(id="h2", content="turn two"), + AIMessage(id="a2", content="answer two"), + ], + thread_id=branch_thread, + run_id_prefix=f"branch-seed-{branch_thread}", + parent_thread_id="parent-thread", + ) + await store.put_batch(seed_events) + # The regenerate run re-journals the same human id plus a fresh answer. + await _put_message(store, "live-run", "human", "h2") + await _put_message(store, "live-run", "ai", "a2-new") + # Resolve the superseded source the way `_find_target_run_id` does: + # the run id carried by the regenerated assistant row itself. + return next(event["run_id"] for event in seed_events if event["content"]["id"] == "a2") + + superseded_run_id = asyncio.run(seed()) + app = _make_app(store, superseded={superseded_run_id}) + with TestClient(app) as client: + body = client.get("/api/threads/thread-1/messages/page?limit=50").json() + + assert [row["content"]["id"] for row in body["data"]] == ["h1", "a1", "h2", "a2-new"] + + def test_thread_page_logs_rows_missing_sequence_values(caplog): store = AsyncMock() store.list_messages.return_value = [{"run_id": "run-1", "content": {"type": "human"}}] diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index df2c702f9..99a937a26 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -2013,7 +2013,8 @@ def test_branch_seeds_run_events_with_parent_history(monkeypatch, mode) -> None: assert [row["content"]["id"] for row in rows] == ["h1", "t1", "a1"] assert [row["event_type"] for row in rows] == ["llm.human.input", "llm.tool.result", "llm.ai.response"] assert all(row["category"] == "message" for row in rows) - assert all(row["run_id"] == f"branch-seed-{branch_thread_id}" for row in rows) + # One synthetic run per inherited turn (#4458): this source has a single turn. + assert all(row["run_id"] == f"branch-seed-{branch_thread_id}-1" for row in rows) assert all((row.get("metadata") or {}).get("branch_seed") is True for row in rows) seqs = [row["seq"] for row in rows] assert seqs == sorted(seqs)