diff --git a/README.md b/README.md index 32e5a140c..125c29909 100644 --- a/README.md +++ b/README.md @@ -1042,7 +1042,7 @@ Interrupted first-turn runs still persist a fallback conversation title, so stop Streaming Markdown responses animate only newly arrived words; text that is already visible is not faded out and replayed when the next chunk extends the same block. -In the Web UI, completed assistant turns can be branched into a new main conversation. The new thread starts from that turn's checkpoint and keeps the preceding replay checkpoint, so the branched response can be regenerated immediately. The latest response can also be regenerated after an interruption, even when its streamed partial text never reached a checkpoint. Regenerating the latest response preserves the thread's current title, including a title you renamed manually after the original response. Legacy or imported histories without checkpoint parent links use a bounded chronological fallback; if no earlier replay checkpoint exists, branching still succeeds with the legacy single-checkpoint shape, while regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are left unchanged rather than attempting an unsafe checkpoint copy. Because workspace files are not checkpointed, the branch only receives a best-effort copy of the current workspace when you branch from the latest turn; branching from an older turn keeps just the restored message history so the branch never inherits files that were created in a later part of the conversation. +In the Web UI, completed assistant turns can be branched into a new main conversation. Automatically inherited branch titles use the next free language-neutral numeric suffix (`Title (2)`, then `Title (3)` for another sibling or a branch of the numbered conversation) so generated sibling titles remain distinct without persisting a locale-specific label; matching explicit or renamed sibling titles also reserve their displayed suffix even though they carry no generated sequence metadata. An explicit API title is preserved. Renaming a branch clears its generated sequence, so its next automatic branch starts from the renamed title at `(2)`. Recent chats also groups loaded branches directly beneath a loaded parent with subtle tree connectors. Missing parents, malformed or cyclic lineage, and branches in a different pinned state stay visible at the top level instead of being hidden or moved across the pinned boundary. The new thread starts from that turn's checkpoint and keeps the preceding replay checkpoint, so the branched response can be regenerated immediately. The latest response can also be regenerated after an interruption, even when its streamed partial text never reached a checkpoint. Regenerating the latest response preserves the thread's current title, including a title you renamed manually after the original response. Legacy or imported histories without checkpoint parent links use a bounded chronological fallback; if no earlier replay checkpoint exists, branching still succeeds with the legacy single-checkpoint shape, while regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are left unchanged rather than attempting an unsafe checkpoint copy. Because workspace files are not checkpointed, the branch only receives a best-effort copy of the current workspace when you branch from the latest turn; branching from an older turn keeps just the restored message history so the branch never inherits files that were created in a later part of the conversation. The Web UI reports completed task time once per run. This is total wall-clock time—including model reasoning, tool calls, and waiting—not a per-step or model-only thinking duration. Reasoning content remains available through its own separate disclosure. diff --git a/README_zh.md b/README_zh.md index 1e9f465f3..ae7f89e94 100644 --- a/README_zh.md +++ b/README_zh.md @@ -576,7 +576,7 @@ Tools 也是同样的思路。DeerFlow 自带一组核心工具：网页搜索 Gateway 生成后续建议时，现在会先把普通字符串输出和 block/list 风格的富文本内容统一归一化，再去解析 JSON 数组响应，因此不同 provider 的内容包装方式不会再悄悄把建议吞掉。 -Web UI 支持从已完成的 assistant 回复分叉出一个新的主对话。新 thread 会保留该轮回复的 checkpoint 以及用户消息之前的重放 checkpoint，因此分叉后可以立即重新生成该回复。对于缺少 checkpoint 父链接的旧历史或导入历史，Gateway 会进行有界的时间顺序查找；如果不存在更早的重放 checkpoint，分叉仍会按旧版单-checkpoint 形态成功创建，但无法重新生成继承的回复。已有的单-checkpoint 分叉会保持不变，不会通过不安全的 checkpoint 复制尝试修复。只有从最新回合分叉时才会尽力复制当前 thread 的工作区文件；从历史回合分叉不会带入后续时间线创建的文件。 +Web UI 支持从已完成的 assistant 回复分叉出一个新的主对话。自动继承的分叉标题会使用下一个空闲的数字后缀（`标题 (2)`、`标题 (3)`……）；显式指定或手动重命名得到的同名后缀也会占号，即使它没有生成序号 metadata，后续自动分叉也不会与它重名。API 调用方显式提供的标题保持不变；重命名会清除旧的生成序号，因此从新标题继续自动分叉时会重新从 `(2)` 开始。最近对话列表还会把已加载的分叉直接排列在已加载的父对话下方，并显示低干扰的树形连接线。父对话尚未加载、谱系数据错误或成环、父子置顶状态不一致时，分叉会安全地保留在顶层，不会被隐藏或跨越置顶边界移动。新 thread 会保留该轮回复的 checkpoint 以及用户消息之前的重放 checkpoint，因此分叉后可以立即重新生成该回复。对于缺少 checkpoint 父链接的旧历史或导入历史，Gateway 会进行有界的时间顺序查找；如果不存在更早的重放 checkpoint，分叉仍会按旧版单-checkpoint 形态成功创建，但无法重新生成继承的回复。已有的单-checkpoint 分叉会保持不变，不会通过不安全的 checkpoint 复制尝试修复。只有从最新回合分叉时才会尽力复制当前 thread 的工作区文件；从历史回合分叉不会带入后续时间线创建的文件。 ```text # sandbox 容器内的路径 diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 5aa0f108d..7b823a704 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -163,6 +163,10 @@ modules that only need lightweight types, config, or registries should import the concrete submodule instead of adding eager package-root imports that pull in the tool graph or subagent executor during state/schema imports. +`ThreadMetaStore.search()` keeps JSON filter semantics identical across memory, +SQLite, and PostgreSQL: missing differs from null, bool differs from int, and +float filters accept integer or real JSON numbers through `json_value_matches`. + ## Development Workflow ### Test-Driven Development (TDD) — MANDATORY diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index bdb0f5428..9ccaf293c 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -53,7 +53,7 @@ reads/searches. | **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/config/credentials` - atomically switch the caller's per-user Lark app after validating the new `app_id`/`app_secret` through the official CLI's live tenant-token probe, revoke/remove the previous OAuth tokens, and restore the prior credential tree if the switch fails; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants. Config and auth flows carry a server-issued, per-user generation persisted under the credential lock; a rejected direct switch leaves the current generation unchanged, stale completions return 409, and browser re-registration uses the same token-clearing/revocation transaction as direct credential switches. | | **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. Thread-scoped runtime channels (`sandbox`, `thread_data`) are not copied onto the branch: the parent's `sandbox_id` binds path mappings and the release lifecycle to the parent's workspace, so the branch lazily acquires its own sandbox instead. 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 | +| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - branch a completed assistant turn with a replay checkpoint; inherited titles take next-free displayed sibling suffixes, including explicit/renamed ones, while explicit titles stay unchanged. Durable `branch` admission rejects races. 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. Thread-scoped runtime channels (`sandbox`, `thread_data`) are not copied onto the branch: the parent's `sandbox_id` binds path mappings and the release lifecycle to the parent's workspace, so the branch lazily acquires its own sandbox instead. 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}` - stream regular text and binary artifacts with `FileResponse`, including byte-`Range` 206/416 behavior used by bounded text previews and media seeking; 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. `PUT /{path}` atomically replaces an existing UTF-8 text file under `/mnt/user-data/outputs` when its expected SHA-256 still matches; active runs conflict, and non-mounted sandbox providers receive the same update explicitly. Atomic replacement applies the existing POSIX permission handling when descriptor-based APIs are available and otherwise keeps the platform-native temporary-file permissions (Windows). | | **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)`) | @@ -153,7 +153,7 @@ JSONL event stores when `GATEWAY_WORKERS > 1`. - Store-only hydrated runs are readable history. In multi-worker mode with heartbeat enabled, cancel on a store-only run records `runs.cancel_action` / `cancel_requested_at` while the owner's lease is live; the first action wins even if a retry later lands on the owner. `RunStore.request_cancel()` and owner completion through `finalize_if_not_cancelled()` are competing active-row CAS operations, so an accepted cancel cannot be overwritten by a later success. `RunStore.renew_lease()` renews and observes the request atomically in the SQL implementation. The owner then executes the normal process-local interrupt/rollback and terminal stream path without transferring the lease. An expired owner is still taken over and marked `error`. `wait=true` and cancel-then-stream use the shared bridge to observe owner finalization; a non-standard process-local bridge returns accepted 202 instead of subscribing to an unreachable stream. In single-worker mode (heartbeat off), store-only runs still return 409. - A local worker's `RunRecord.lease_expires_at` is the last durably confirmed ownership deadline. `_renew_leases()` bounds each renewal attempt by that deadline: transient store exceptions remain retryable while it is valid, but an exception or blocked call that reaches expiry sets the process-local `ownership_lost` fence, raises `abort_event`, and cancels the run task. Successful renewals collect durable cancellation actions; after all local renewals have been attempted, heartbeat only signals the corresponding process-local tasks, leaving status writes and rollback cleanup to the worker finalization path. Fenced workers do not perform subsequent journal/delivery-receipt, progress/completion/status, checkpoint/thread-metadata, or `on_run_completed` writes; the peer recovery path owns the terminal receipt. `RunStore.update_run_completion()` also refuses to replace a different terminal status, closing the peer-takeover/late-finalization race. `grace_seconds` delays peer reclamation for clock skew but is not extra execution time for an owner that can no longer confirm its lease. Already-committed remote tool side effects remain outside this local cancellation boundary. - Startup/orphan reconciliation must claim stale active rows with `RunStore.claim_for_takeover()`, not a plain `update_status()`. The final claim re-checks `status` and lease expiry atomically, so a heartbeat renewal between the candidate scan and the recovery write keeps the run active. -- Run admission and independent writes are first-class thread operations. `runs.operation_kind` distinguishes user-visible `run` rows from internal `checkpoint_write` and `artifact_write` reservations, while every active kind shares the existing durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds. +- Run admission and independent writes are first-class thread operations. `runs.operation_kind` distinguishes `run` from `checkpoint_write`, `artifact_write`, `branch`, and `delete`; every active kind shares the durable active-thread uniqueness constraint. New operation kinds must go through `RunStore.create_thread_operation_atomic()` and `RunManager.reserve_thread_operation()` rather than adding another lock or metadata marker. Live and lease-less reservations are non-interruptible; an expired leased reservation can be reclaimed immediately by interrupt/rollback admission without waiting for orphan reconciliation. Lease-less rows stay fail-closed because the store cannot distinguish a stale row from a live writer in another heartbeat-disabled worker; a rare failed delete therefore requires startup reconciliation, and heartbeat-disabled multi-worker deployment remains unsupported. Reservation bodies are attached to their caller task so loss detected by lease renewal cancels the writer before it can continue after takeover; the context manager translates that lease-loss cancellation to `ConflictError` after cleanup so Gateway mutation routes return a retryable 409 instead of dropping the HTTP request. The cleanup scope begins immediately after durable admission, including the await that attaches the caller task, so cancellation cannot strand a locally renewed pending reservation. A failed renewal is revalidated under the manager lock before cancellation; if the reservation completed and unregistered while the store update was in flight, its request task must not be cancelled after the write. Reservations are excluded from run history/reporting and from run-only helpers such as `list_by_thread()` and `has_inflight()`, release uses the captured owner rather than ambient user context, and local cleanup still runs when the best-effort store delete fails. `RunStore.create_run_atomic()` remains a deprecated compatibility shim for external stores that only admit normal runs; new stores must implement `create_thread_operation_atomic()` to support internal operation kinds. - Gateway checkpoint mutations outside run execution must use `services.reserve_checkpoint_write()`, which composes the process-local thread lock with the durable `checkpoint_write` reservation. Manual compaction, `POST /threads/{id}/state`, and both goal mutation routes (`PUT` / `DELETE /threads/{id}/goal`, including creation of a missing goal checkpoint) use this boundary, so an existing run blocks the write and the reservation blocks new reject/interrupt/rollback runs across workers. - `POST /wait` (both thread-scoped and `/api/runs/wait`) drains the stream bridge via `wait_for_run_completion()` instead of bare `await record.task`, so it honours the run's `on_disconnect` setting and cancels the background run on real client disconnect rather than returning a stale checkpoint (issue #3265). - Memory and Redis `StreamBridge` implementations retain only `stream_bridge.queue_maxsize` data events. A syntactically valid `Last-Event-ID` older than the retained watermark, or a live subscriber that falls behind it, yields `StreamGap` before any partial replay. `sse_consumer` maps that control item to an id-less SSE `gap` payload (`stream_replay_gap`) and intentionally leaves the run active; internal `/wait` consumers resume from its latest retained ID because they only need terminal completion. Redis checks bounds plus the non-blocking read in one transaction, using blocking `XREAD` only as a wake-up before repeating the atomic snapshot. For a no-cursor subscriber that established a wait on an empty stream, the first wake response remains provisional until that next snapshot verifies its tail is still retained; this closes the pre-first-delivery trimming window without changing malformed-cursor live tailing. The correctness tradeoff is one three-command snapshot pipeline per poll plus the blocking wake round trip while idle. Malformed cursor behavior remains backend-specific. Memory treats a syntactically numeric cursor below its watermark conservatively as a gap even when the evicted timestamp can no longer be verified; unknown ids at or above the watermark retain the legacy replay-from-earliest policy. diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index ff1e45e74..8aeb1d95c 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -102,6 +102,7 @@ def _checkpoint_mode_http_error(exc: Exception, thread_id: str) -> HTTPException _SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"}) _SIDECAR_METADATA_KEY = "deerflow_sidecar" _BRANCH_METADATA_KEY = "deerflow_branch" +_BRANCH_TITLE_SEQUENCE_METADATA_KEY = "branch_title_sequence" # Thread-scoped runtime channels a branch must NOT inherit from its parent: # ``sandbox.sandbox_id`` binds path mappings and the release lifecycle to the # *parent* thread, so copying it would make the branch read/write the parent's @@ -112,6 +113,9 @@ _BRANCH_METADATA_KEY = "deerflow_branch" _BRANCH_EXCLUDED_CHANNELS = frozenset({"sandbox", "thread_data"}) _BRANCH_HISTORY_SCAN_LIMIT = 200 _BRANCH_HISTORY_RAW_SCAN_LIMIT = _BRANCH_HISTORY_SCAN_LIMIT * 2 +_BRANCH_TITLE_MAX_LENGTH = 256 +_BRANCH_TITLE_SEQUENCE_MAX = 9_007_199_254_740_991 +_BRANCH_SIBLING_PAGE_SIZE = 100 def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]: @@ -329,16 +333,66 @@ async def _copy_branch_user_data(source_thread_id: str, target_thread_id: str) - return "failed" -def _default_branch_display_name(source_title: Any, *, source_is_branch: bool = False) -> str | None: +def _next_branch_title_sequence(source_sequence: Any, *, source_is_branch: bool) -> int: + if source_is_branch and isinstance(source_sequence, int) and not isinstance(source_sequence, bool) and 2 <= source_sequence < _BRANCH_TITLE_SEQUENCE_MAX: + return source_sequence + 1 + return 2 + + +def _format_branch_display_name(base: str, sequence: int) -> str | None: + suffix = f" ({sequence})" + truncated_base = base[: _BRANCH_TITLE_MAX_LENGTH - len(suffix)].rstrip() + return f"{truncated_base}{suffix}" if truncated_base else None + + +def _default_branch_title( + source_title: Any, + *, + source_is_branch: bool = False, + source_sequence: Any = None, + sibling_records: list[dict[str, Any]] | None = None, +) -> tuple[str | None, int | None]: if not isinstance(source_title, str): - return None + return None, None display_name = source_title.strip() if source_is_branch: while display_name.lower().startswith("branch:"): display_name = display_name[len("branch:") :].strip() - return display_name or None + if not display_name: + return None, None + + sequence = _next_branch_title_sequence(source_sequence, source_is_branch=source_is_branch) + base = display_name + if sequence > 2: + source_suffix = f" ({sequence - 1})" + if display_name.endswith(source_suffix): + base = display_name[: -len(source_suffix)].rstrip() + + occupied_titles = {sibling.get("display_name") for sibling in sibling_records or [] if isinstance(sibling.get("display_name"), str)} + display_name = _format_branch_display_name(base, sequence) + while display_name in occupied_titles: + if sequence >= _BRANCH_TITLE_SEQUENCE_MAX: + return None, None + sequence += 1 + display_name = _format_branch_display_name(base, sequence) + return display_name, sequence if display_name is not None else None + + +async def _branch_sibling_records(thread_store: Any, parent_thread_id: str) -> list[dict[str, Any]]: + siblings: list[dict[str, Any]] = [] + offset = 0 + while True: + page = await thread_store.search( + metadata={"branch_parent_thread_id": parent_thread_id}, + limit=_BRANCH_SIBLING_PAGE_SIZE, + offset=offset, + ) + siblings.extend(page) + if len(page) < _BRANCH_SIBLING_PAGE_SIZE: + return siblings + offset += len(page) # --------------------------------------------------------------------------- @@ -807,6 +861,27 @@ async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadRe @require_permission("threads", "write", owner_check=True, require_existing=True) async def branch_thread(thread_id: ThreadId, body: ThreadBranchRequest, request: Request) -> ThreadBranchResponse: """Create a new main-thread branch from a completed assistant turn.""" + try: + async with goal_thread_lock(thread_id): + async with get_run_manager(request).reserve_thread_operation( + thread_id, + kind=ThreadOperationKind.branch, + user_id=get_effective_user_id(), + ): + return await _branch_thread_with_reservation(thread_id, body, request) + except ConflictError: + raise HTTPException( + status_code=409, + detail="Thread has work in flight. Branch it after the work finishes.", + ) from None + + +async def _branch_thread_with_reservation( + thread_id: ThreadId, + body: ThreadBranchRequest, + request: Request, +) -> ThreadBranchResponse: + """Create a branch while holding the source thread's exclusive reservation.""" from app.gateway.deps import get_thread_store thread_store = get_thread_store(request) @@ -857,10 +932,18 @@ async def branch_thread(thread_id: ThreadId, body: ThreadBranchRequest, request: "branch_created_at": now, } - display_name = body.title or _default_branch_display_name( - source_record.get("display_name"), - source_is_branch=source_metadata.get(_BRANCH_METADATA_KEY) is True, - ) + if body.title: + display_name = body.title + else: + sibling_records = await _branch_sibling_records(thread_store, thread_id) + display_name, title_sequence = _default_branch_title( + source_record.get("display_name"), + source_is_branch=source_metadata.get(_BRANCH_METADATA_KEY) is True, + source_sequence=source_metadata.get(_BRANCH_TITLE_SEQUENCE_METADATA_KEY), + sibling_records=sibling_records, + ) + if title_sequence is not None: + branch_metadata[_BRANCH_TITLE_SEQUENCE_METADATA_KEY] = title_sequence thread_owner_user_id = get_trusted_internal_owner_user_id(request) thread_owner_kwargs = {"user_id": thread_owner_user_id} if thread_owner_user_id else {} @@ -889,6 +972,8 @@ async def branch_thread(thread_id: ThreadId, body: ThreadBranchRequest, request: values[key] = Overwrite(list(value) if key == "messages" and isinstance(value, list) else value) else: values[key] = value + if display_name is not None: + values["title"] = display_name return values # Stamp both synthetic checkpoints with the branch-creation time because @@ -1323,7 +1408,11 @@ async def update_thread_state(thread_id: ThreadId, body: ThreadStateUpdateReques new_title = body.values["title"] if new_title: try: - await thread_store.update_display_name(thread_id, new_title) + await thread_store.update_display_name( + thread_id, + new_title, + remove_metadata_keys=(_BRANCH_TITLE_SEQUENCE_METADATA_KEY,), + ) except Exception: logger.debug("Failed to sync title to thread_meta for %s (non-fatal)", sanitize_log_param(thread_id)) diff --git a/backend/packages/harness/deerflow/persistence/json_compat.py b/backend/packages/harness/deerflow/persistence/json_compat.py index 442b29e22..c59ef7921 100644 --- a/backend/packages/harness/deerflow/persistence/json_compat.py +++ b/backend/packages/harness/deerflow/persistence/json_compat.py @@ -3,6 +3,7 @@ from __future__ import annotations import re +from collections.abc import Mapping from dataclasses import dataclass from typing import Any @@ -57,6 +58,36 @@ def validate_metadata_filter_value(value: object) -> bool: return True +def json_value_matches(values: Mapping[str, Any], key: str, expected: object) -> bool: + """Match one JSON object field with the same type semantics as :class:`JsonMatch`. + + Missing keys differ from explicit JSON null, booleans differ from integers, + integer filters accept only JSON integers, and float filters accept either + JSON integer or real values. Callers validate filter keys and values at + their API boundary; unsupported expected values never match. + """ + if key not in values: + return False + + actual = values[key] + if expected is None: + return actual is None + if isinstance(expected, bool): + return isinstance(actual, bool) and actual is expected + if isinstance(expected, int): + return isinstance(actual, int) and not isinstance(actual, bool) and actual == expected + if isinstance(expected, float): + if not isinstance(actual, (int, float)) or isinstance(actual, bool): + return False + try: + return float(actual) == expected + except OverflowError: + return False + if isinstance(expected, str): + return isinstance(actual, str) and actual == expected + return False + + class JsonMatch(ColumnElement): """Dialect-portable ``column[key] == value`` for JSON columns. @@ -75,6 +106,7 @@ class JsonMatch(ColumnElement): _traverse_internals = [ ("column", InternalTraversal.dp_clauseelement), ("key", InternalTraversal.dp_string), + ("value_type", InternalTraversal.dp_string), ("value", InternalTraversal.dp_plain_obj), ] @@ -87,6 +119,10 @@ class JsonMatch(ColumnElement): raise TypeError(f"JsonMatch value must be None, bool, int, float, or str; got: {type(value).__name__!r}") self.column = column self.key = key + # Python considers True == 1 == 1.0 and gives them the same hash. + # Include the JSON filter type in SQLAlchemy's cache key so a compiled + # boolean predicate can never be reused for a numeric query (or vice versa). + self.value_type = type(value).__name__ self.value = value super().__init__() diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/base.py b/backend/packages/harness/deerflow/persistence/thread_meta/base.py index f9fecb30d..869ca2b31 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/base.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/base.py @@ -65,7 +65,14 @@ class ThreadMetaStore(abc.ABC): pass @abc.abstractmethod - async def update_display_name(self, thread_id: str, display_name: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + async def update_display_name( + self, + thread_id: str, + display_name: str, + *, + remove_metadata_keys: tuple[str, ...] = (), + user_id: str | None | _AutoSentinel = AUTO, + ) -> None: pass @abc.abstractmethod diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/memory.py b/backend/packages/harness/deerflow/persistence/thread_meta/memory.py index 310b13032..5673a23ae 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/memory.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/memory.py @@ -11,6 +11,7 @@ from typing import Any from langgraph.store.base import BaseStore +from deerflow.persistence.json_compat import json_value_matches from deerflow.persistence.thread_meta.base import THREAD_PINNED_METADATA_KEY, ThreadMetaStore from deerflow.runtime.user_context import AUTO, _AutoSentinel, resolve_user_id from deerflow.utils.time import coerce_iso, now_iso @@ -84,8 +85,6 @@ class MemoryThreadMetaStore(ThreadMetaStore): """ resolved_user_id = resolve_user_id(user_id, method_name="MemoryThreadMetaStore.search") filter_dict: dict[str, Any] = {} - if metadata: - filter_dict.update(metadata) if status: filter_dict["status"] = status if resolved_user_id is not None: @@ -108,6 +107,8 @@ class MemoryThreadMetaStore(ThreadMetaStore): search_offset += len(page) records = [self._item_to_dict(item) for item in items] + if metadata: + records = [record for record in records if isinstance(record.get("metadata"), dict) and all(json_value_matches(record["metadata"], key, value) for key, value in metadata.items())] records.sort(key=self._sort_key, reverse=True) return records[offset : offset + limit] @@ -120,11 +121,22 @@ class MemoryThreadMetaStore(ThreadMetaStore): return True return record_user_id == user_id - async def update_display_name(self, thread_id: str, display_name: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None: + async def update_display_name( + self, + thread_id: str, + display_name: str, + *, + remove_metadata_keys: tuple[str, ...] = (), + user_id: str | None | _AutoSentinel = AUTO, + ) -> None: record = await self._get_owned_record(thread_id, user_id, "MemoryThreadMetaStore.update_display_name") if record is None: return record["display_name"] = display_name + metadata = dict(record.get("metadata") or {}) + for key in remove_metadata_keys: + metadata.pop(key, None) + record["metadata"] = metadata record["updated_at"] = now_iso() await self._store.aput(THREADS_NS, thread_id, record) diff --git a/backend/packages/harness/deerflow/persistence/thread_meta/sql.py b/backend/packages/harness/deerflow/persistence/thread_meta/sql.py index e282e0ef5..7f9f310e1 100644 --- a/backend/packages/harness/deerflow/persistence/thread_meta/sql.py +++ b/backend/packages/harness/deerflow/persistence/thread_meta/sql.py @@ -170,14 +170,26 @@ class ThreadMetaRepository(ThreadMetaStore): thread_id: str, display_name: str, *, + remove_metadata_keys: tuple[str, ...] = (), user_id: str | None | _AutoSentinel = AUTO, ) -> None: - """Update the display_name (title) for a thread.""" + """Update the display name and remove caller-selected stale metadata atomically.""" resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.update_display_name") async with self._sf() as session: - if not await self._check_ownership(session, thread_id, resolved_user_id): + if session.get_bind().dialect.name == "sqlite": + await session.execute(text("BEGIN IMMEDIATE")) + row = await session.get(ThreadMetaRow, thread_id) + else: + result = await session.execute(select(ThreadMetaRow).where(ThreadMetaRow.thread_id == thread_id).with_for_update()) + row = result.scalar_one_or_none() + if row is None or (resolved_user_id is not None and row.user_id != resolved_user_id): return - await session.execute(update(ThreadMetaRow).where(ThreadMetaRow.thread_id == thread_id).values(display_name=display_name, updated_at=datetime.now(UTC))) + row.display_name = display_name + metadata = dict(row.metadata_json or {}) + for key in remove_metadata_keys: + metadata.pop(key, None) + row.metadata_json = metadata + row.updated_at = datetime.now(UTC) await session.commit() async def update_status( diff --git a/backend/packages/harness/deerflow/runtime/runs/schemas.py b/backend/packages/harness/deerflow/runtime/runs/schemas.py index 10ff8706c..15ace3ae9 100644 --- a/backend/packages/harness/deerflow/runtime/runs/schemas.py +++ b/backend/packages/harness/deerflow/runtime/runs/schemas.py @@ -9,6 +9,7 @@ class ThreadOperationKind(StrEnum): run = "run" checkpoint_write = "checkpoint_write" artifact_write = "artifact_write" + branch = "branch" delete = "delete" diff --git a/backend/tests/test_memory_thread_meta_isolation.py b/backend/tests/test_memory_thread_meta_isolation.py index 25c9298f0..25a776010 100644 --- a/backend/tests/test_memory_thread_meta_isolation.py +++ b/backend/tests/test_memory_thread_meta_isolation.py @@ -53,6 +53,61 @@ async def test_search_isolation(store): assert [r["thread_id"] for r in results] == ["t-beta"] +@pytest.mark.anyio +@pytest.mark.no_auto_user +async def test_search_matches_nested_thread_metadata(store): + with _as_user(USER_A): + await store.create("root", metadata={}) + await store.create("child", metadata={"branch_parent_thread_id": "root"}) + await store.create("other", metadata={"branch_parent_thread_id": "elsewhere"}) + + results = await store.search(metadata={"branch_parent_thread_id": "root"}) + + assert [record["thread_id"] for record in results] == ["child"] + + +@pytest.mark.anyio +@pytest.mark.no_auto_user +async def test_search_metadata_preserves_json_type_and_presence_contract(store): + with _as_user(USER_A): + await store.create("missing", metadata={}) + await store.create("null", metadata={"value": None}) + await store.create("bool", metadata={"value": True}) + await store.create("int", metadata={"value": 1}) + await store.create("float", metadata={"value": 1.0}) + + null_hits = await store.search(metadata={"value": None}) + bool_hits = await store.search(metadata={"value": True}) + int_hits = await store.search(metadata={"value": 1}) + float_hits = await store.search(metadata={"value": 1.0}) + + assert [record["thread_id"] for record in null_hits] == ["null"] + assert [record["thread_id"] for record in bool_hits] == ["bool"] + assert [record["thread_id"] for record in int_hits] == ["int"] + assert {record["thread_id"] for record in float_hits} == {"float", "int"} + + +@pytest.mark.anyio +@pytest.mark.no_auto_user +async def test_update_display_name_can_remove_stale_metadata(store): + with _as_user(USER_A): + await store.create( + "branch", + display_name="Original (2)", + metadata={"branch_title_sequence": 2, "keep": True}, + ) + await store.update_display_name( + "branch", + "Report Q4", + remove_metadata_keys=("branch_title_sequence",), + ) + result = await store.get("branch") + + assert result is not None + assert result["display_name"] == "Report Q4" + assert result["metadata"] == {"keep": True} + + @pytest.mark.anyio @pytest.mark.no_auto_user async def test_get_isolation(store): diff --git a/backend/tests/test_thread_meta_repo.py b/backend/tests/test_thread_meta_repo.py index 14fe6ddd9..e18b9ccec 100644 --- a/backend/tests/test_thread_meta_repo.py +++ b/backend/tests/test_thread_meta_repo.py @@ -46,6 +46,21 @@ class TestThreadMetaRepository: record = await repo.create("t1", metadata={"key": "value"}) assert record["metadata"] == {"key": "value"} + @pytest.mark.anyio + async def test_update_display_name_can_remove_stale_metadata_atomically(self, repo): + await repo.create("t1", display_name="Original (2)", metadata={"branch_title_sequence": 2, "keep": True}) + + await repo.update_display_name( + "t1", + "Report Q4", + remove_metadata_keys=("branch_title_sequence",), + ) + + record = await repo.get("t1") + assert record is not None + assert record["display_name"] == "Report Q4" + assert record["metadata"] == {"keep": True} + @pytest.mark.anyio async def test_get_nonexistent(self, repo): assert await repo.get("nonexistent") is None @@ -416,6 +431,20 @@ class TestThreadMetaRepository: hits = {r["thread_id"] for r in await repo.search(metadata={"score": 3.14})} assert hits == {"t1", "t3"} + @pytest.mark.anyio + async def test_search_metadata_float_matches_integer_but_not_boolean(self, repo): + await repo.create("int", metadata={"score": 1}) + await repo.create("float", metadata={"score": 1.0}) + await repo.create("bool", metadata={"score": True}) + + bool_hits = {record["thread_id"] for record in await repo.search(metadata={"score": True})} + int_hits = {record["thread_id"] for record in await repo.search(metadata={"score": 1})} + float_hits = {record["thread_id"] for record in await repo.search(metadata={"score": 1.0})} + + assert bool_hits == {"bool"} + assert int_hits == {"int"} + assert float_hits == {"float", "int"} + @pytest.mark.anyio async def test_search_metadata_mixed_types_same_key(self, repo): """Each type query only matches its own type, even when the key is shared.""" @@ -608,3 +637,16 @@ class TestJsonMatchCompilation: with pytest.raises(ValueError, match="Key escaped validation"): str(elem.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True})) + + +class TestJsonValueMatches: + def test_distinguishes_missing_null_bool_int_and_float(self): + from deerflow.persistence.json_compat import json_value_matches + + assert json_value_matches({}, "value", None) is False + assert json_value_matches({"value": None}, "value", None) is True + assert json_value_matches({"value": 1}, "value", True) is False + assert json_value_matches({"value": True}, "value", 1) is False + assert json_value_matches({"value": 1.0}, "value", 1) is False + assert json_value_matches({"value": 1}, "value", 1.0) is True + assert json_value_matches({"value": 1.0}, "value", 1.0) is True diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index 567e1d746..619badf20 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -513,6 +513,42 @@ def test_delete_thread_route_rejects_active_thread_operation_without_deleting_me assert asyncio.run(store.aget(THREADS_NS, "thread-active-delete")) is not None +def test_branch_thread_route_rejects_concurrent_source_operation_without_creating_child(): + class RejectingRunManager(_ThreadTestRunManager): + @asynccontextmanager + async def reserve_thread_operation(self, _thread_id: str, **_kwargs): + raise ConflictError("Thread already has active work") + yield # pragma: no cover - required by asynccontextmanager + + app, store, _checkpointer = _build_thread_app() + app.state.run_manager = RejectingRunManager() + asyncio.run( + store.aput( + THREADS_NS, + "thread-active-branch", + { + "thread_id": "thread-active-branch", + "user_id": None, + "status": "idle", + "created_at": "", + "updated_at": "", + "metadata": {}, + }, + ) + ) + + with TestClient(app) as client: + response = client.post( + "/api/threads/thread-active-branch/branches", + json={"message_id": "ai-1", "message_ids": ["ai-1"]}, + ) + + assert response.status_code == 409 + assert response.json()["detail"] == "Thread has work in flight. Branch it after the work finishes." + children = asyncio.run(app.state.thread_store.search(metadata={"branch_parent_thread_id": "thread-active-branch"}, user_id=None)) + assert children == [] + + def test_legacy_thread_metadata_mutation_is_rejected(): app, store, _checkpointer = _build_thread_app() legacy_thread_id = "legacy.thread" @@ -2382,7 +2418,54 @@ def test_branch_thread_from_older_assistant_turn_creates_truncated_thread() -> N assert response.status_code == 200, response.text body = response.json() new_thread_id = body["thread_id"] + sibling_response = client.post( + f"/api/threads/{source_thread_id}/branches", + json={"message_id": "ai-2", "message_ids": ["ai-2"]}, + ) + assert sibling_response.status_code == 200, sibling_response.text + sibling_thread_id = sibling_response.json()["thread_id"] + explicit_collision_response = client.post( + f"/api/threads/{source_thread_id}/branches", + json={"message_id": "ai-2", "message_ids": ["ai-2"], "title": "Original chat (4)"}, + ) + assert explicit_collision_response.status_code == 200, explicit_collision_response.text + explicit_collision_thread_id = explicit_collision_response.json()["thread_id"] + after_explicit_response = client.post( + f"/api/threads/{source_thread_id}/branches", + json={"message_id": "ai-2", "message_ids": ["ai-2"]}, + ) + assert after_explicit_response.status_code == 200, after_explicit_response.text + after_explicit_thread_id = after_explicit_response.json()["thread_id"] + nested_response = client.post( + f"/api/threads/{new_thread_id}/branches", + json={"message_id": "ai-2", "message_ids": ["ai-2"]}, + ) + assert nested_response.status_code == 200, nested_response.text + nested_thread_id = nested_response.json()["thread_id"] + rename_response = client.post( + f"/api/threads/{new_thread_id}/state", + json={"values": {"title": "Report Q4"}}, + ) + assert rename_response.status_code == 200, rename_response.text + renamed_branch_response = client.post( + f"/api/threads/{new_thread_id}/branches", + json={"message_id": "ai-2", "message_ids": ["ai-2"]}, + ) + assert renamed_branch_response.status_code == 200, renamed_branch_response.text + renamed_branch_thread_id = renamed_branch_response.json()["thread_id"] + explicit_response = client.post( + f"/api/threads/{source_thread_id}/branches", + json={"message_id": "ai-2", "message_ids": ["ai-2"], "title": "Deliberate branch title"}, + ) + assert explicit_response.status_code == 200, explicit_response.text + explicit_thread_id = explicit_response.json()["thread_id"] state_response = client.get(f"/api/threads/{new_thread_id}/state") + sibling_state_response = client.get(f"/api/threads/{sibling_thread_id}/state") + explicit_collision_state_response = client.get(f"/api/threads/{explicit_collision_thread_id}/state") + after_explicit_state_response = client.get(f"/api/threads/{after_explicit_thread_id}/state") + nested_state_response = client.get(f"/api/threads/{nested_thread_id}/state") + renamed_branch_state_response = client.get(f"/api/threads/{renamed_branch_thread_id}/state") + explicit_state_response = client.get(f"/api/threads/{explicit_thread_id}/state") search_response = client.post("/api/threads/search", json={"limit": 10}) assert body["parent_thread_id"] == source_thread_id @@ -2391,12 +2474,46 @@ def test_branch_thread_from_older_assistant_turn_creates_truncated_thread() -> N assert body["workspace_clone_mode"] == "skipped_historical_turn" assert state_response.status_code == 200, state_response.text - messages = state_response.json()["values"]["messages"] + state_values = state_response.json()["values"] + messages = state_values["messages"] assert [message["id"] for message in messages] == ["human-1", "ai-1", "human-2", "ai-2"] assert "Third answer" not in [message.get("content") for message in messages] + assert state_values["title"] == "Report Q4" + assert sibling_state_response.status_code == 200, sibling_state_response.text + assert sibling_state_response.json()["values"]["title"] == "Original chat (3)" + assert explicit_collision_state_response.status_code == 200, explicit_collision_state_response.text + assert explicit_collision_state_response.json()["values"]["title"] == "Original chat (4)" + assert after_explicit_state_response.status_code == 200, after_explicit_state_response.text + assert after_explicit_state_response.json()["values"]["title"] == "Original chat (5)" + assert nested_state_response.status_code == 200, nested_state_response.text + assert nested_state_response.json()["values"]["title"] == "Original chat (3)" + assert renamed_branch_state_response.status_code == 200, renamed_branch_state_response.text + assert renamed_branch_state_response.json()["values"]["title"] == "Report Q4 (2)" + assert explicit_state_response.status_code == 200, explicit_state_response.text + assert explicit_state_response.json()["values"]["title"] == "Deliberate branch title" assert search_response.status_code == 200, search_response.text branch_entry = next(item for item in search_response.json() if item["thread_id"] == new_thread_id) - assert branch_entry["values"]["title"] == "Original chat" + assert branch_entry["values"]["title"] == "Report Q4" + assert "branch_title_sequence" not in branch_entry["metadata"] + sibling_entry = next(item for item in search_response.json() if item["thread_id"] == sibling_thread_id) + assert sibling_entry["values"]["title"] == "Original chat (3)" + assert sibling_entry["metadata"]["branch_title_sequence"] == 3 + explicit_collision_entry = next(item for item in search_response.json() if item["thread_id"] == explicit_collision_thread_id) + assert explicit_collision_entry["values"]["title"] == "Original chat (4)" + assert "branch_title_sequence" not in explicit_collision_entry["metadata"] + after_explicit_entry = next(item for item in search_response.json() if item["thread_id"] == after_explicit_thread_id) + assert after_explicit_entry["values"]["title"] == "Original chat (5)" + assert after_explicit_entry["metadata"]["branch_title_sequence"] == 5 + nested_entry = next(item for item in search_response.json() if item["thread_id"] == nested_thread_id) + assert nested_entry["values"]["title"] == "Original chat (3)" + renamed_branch_entry = next(item for item in search_response.json() if item["thread_id"] == renamed_branch_thread_id) + assert renamed_branch_entry["values"]["title"] == "Report Q4 (2)" + assert renamed_branch_entry["metadata"]["branch_title_sequence"] == 2 + explicit_entry = next(item for item in search_response.json() if item["thread_id"] == explicit_thread_id) + assert explicit_entry["values"]["title"] == "Deliberate branch title" + assert "branch_title_sequence" not in explicit_entry["metadata"] + branch_reservations = [reservation for reservation in app.state.run_manager.reservations if reservation[1]["kind"] == ThreadOperationKind.branch] + assert len(branch_reservations) == 7 def test_branch_thread_uses_materialized_history_and_overwrites_fresh_seed(monkeypatch) -> None: @@ -3199,10 +3316,51 @@ def test_update_thread_state_rejects_unknown_state_fields(monkeypatch) -> None: assert "not_a_state_field" in response.json()["detail"] -def test_branch_display_name_strips_legacy_branch_prefix_only_for_branch_sources() -> None: - assert threads._default_branch_display_name("Original chat") == "Original chat" - assert threads._default_branch_display_name("Branch: Original chat") == "Branch: Original chat" - assert threads._default_branch_display_name("Branch: Branch: Original chat", source_is_branch=True) == "Original chat" +def test_branch_title_adds_next_free_language_neutral_numeric_suffix() -> None: + assert threads._default_branch_title("Original chat") == ("Original chat (2)", 2) + assert threads._default_branch_title("Roadmap (2026)") == ("Roadmap (2026) (2)", 2) + assert threads._default_branch_title("Original chat", sibling_records=[{"display_name": "Original chat (2)", "metadata": {"branch_title_sequence": 2}}]) == ( + "Original chat (3)", + 3, + ) + assert threads._default_branch_title("Original chat (2)", source_is_branch=True, source_sequence=2) == ("Original chat (3)", 3) + assert threads._default_branch_title("Roadmap (2026)", source_is_branch=True) == ("Roadmap (2026) (2)", 2) + assert threads._default_branch_title("Roadmap (2026)", source_is_branch=True, source_sequence=2) == ("Roadmap (2026) (3)", 3) + assert threads._default_branch_title("Branch: Branch: Original chat", source_is_branch=True) == ("Original chat (2)", 2) + assert threads._default_branch_title("Report Q4", source_is_branch=True, sibling_records=[{"display_name": "Original chat (3)", "metadata": {"branch_title_sequence": 3}}]) == ( + "Report Q4 (2)", + 2, + ) + assert threads._default_branch_title( + "Original chat", + sibling_records=[ + {"display_name": "Original chat (3)", "metadata": {"branch_title_sequence": 3}}, + {"display_name": "Explicit title", "metadata": {}}, + {"display_name": "Original chat (2)", "metadata": {"branch_title_sequence": "2"}}, + ], + ) == ("Original chat (4)", 4) + assert threads._default_branch_title( + "Original chat", + sibling_records=[ + {"display_name": "Original chat (2)", "metadata": {}}, + {"display_name": "Unrelated (3)", "metadata": {"branch_title_sequence": 3}}, + ], + ) == ("Original chat (3)", 3) + assert threads._default_branch_title(" ") == (None, None) + capped, sequence = threads._default_branch_title("x" * 256) + assert capped is not None + assert sequence == 2 + assert len(capped) == 256 + assert capped.endswith(" (2)") + + +def test_next_branch_title_sequence_accepts_only_bounded_numeric_branch_metadata() -> None: + assert threads._next_branch_title_sequence(2, source_is_branch=True) == 3 + assert threads._next_branch_title_sequence(8, source_is_branch=True) == 9 + assert threads._next_branch_title_sequence(8, source_is_branch=False) == 2 + assert threads._next_branch_title_sequence(True, source_is_branch=True) == 2 + assert threads._next_branch_title_sequence("8", source_is_branch=True) == 2 + assert threads._next_branch_title_sequence(threads._BRANCH_TITLE_SEQUENCE_MAX, source_is_branch=True) == 2 def test_branch_thread_rejects_sidecar_threads() -> None: diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 49a6af1fd..cb6424905 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -107,6 +107,7 @@ Edit-and-rerun is deliberately latest-turn-only. `core/messages/utils.ts::getLat - `src/app/workspace/chats/[thread_id]/page.tsx` owns composer busy-state wiring. - `src/app/workspace/chats/[thread_id]/page.tsx` owns branch-from-turn submission and navigation; sidecar `MessageList` instances do not receive the branch action. +- `core/threads/thread-branch-tree.ts` projects only loaded, same-pin branch lineage into Recent chats. Missing, malformed, cross-pin, self, or cyclic parents stay top-level; unpinned groups follow their freshest descendant while pinned root order stays stable. `recent-chat-list.tsx` caps visual indentation without changing the recursive order. - `src/app/workspace/chats/[thread_id]/page.tsx` and `src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx` own edit-and-rerun submission wiring because the page must preserve normal/custom-agent run context; `MessageList` only detects the latest editable user turn and renders the inline editor. - `src/app/workspace/chats/[thread_id]/page.tsx` gates the Workspace Browser trigger and browser right panel on `/api/features -> browser_control.enabled`; `src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx` applies the same capability gate and additionally requires the Custom Agent's tool groups to be unrestricted or include `browser`. Default/failed feature discovery hides the browser control so optional backend installs do not show a dead Live socket. - `src/app/workspace/chats/[thread_id]/page.tsx` and `src/app/workspace/agents/[agent_name]/chats/[thread_id]/page.tsx` own active-goal display state for their composer overlays. diff --git a/frontend/src/components/workspace/recent-chat-list.tsx b/frontend/src/components/workspace/recent-chat-list.tsx index 6b93ee073..956128615 100644 --- a/frontend/src/components/workspace/recent-chat-list.tsx +++ b/frontend/src/components/workspace/recent-chat-list.tsx @@ -55,6 +55,7 @@ import { usePinThread, useRenameThread, } from "@/core/threads/hooks"; +import { flattenThreadBranches } from "@/core/threads/thread-branch-tree"; import { buildThreadListModel } from "@/core/threads/thread-list-model"; import type { AgentThread, AgentThreadState } from "@/core/threads/types"; import { @@ -103,6 +104,15 @@ export function RecentChatList() { ? [...threadListModel.displayedThreads, activeThread] : threadListModel.displayedThreads; }, [threadIdFromPath, threadListModel]); + const branchList = useMemo(() => { + const entries = flattenThreadBranches(displayedThreads); + return { + entriesById: new Map( + entries.map((entry) => [entry.thread.thread_id, entry]), + ), + threads: entries.map((entry) => entry.thread), + }; + }, [displayedThreads]); const sentinelRef = useRef(null); useEffect(() => { @@ -291,12 +301,22 @@ export function RecentChatList() { { const isActive = pathOfThread(thread) === pathname; const channelSource = channelSourceOfThread(thread); const pinned = isThreadPinned(thread); + const branchEntry = branchList.entriesById.get( + thread.thread_id, + ); + const parentTitle = branchEntry?.parentThread + ? titleOfThread(branchEntry.parentThread) + : null; + const title = titleOfThread(thread); + const branchLabel = parentTitle + ? t.chats.branchLabel(title, parentTitle) + : undefined; return ( 0 + ? branchEntry.depth + : undefined + } + data-branch-parent-id={ + branchEntry?.parentThread?.thread_id + } href={pathOfThread(thread)} + title={branchLabel} > + {branchEntry && branchEntry.depth > 0 && ( + + )} {pinned && ( )} - - {titleOfThread(thread)} - + {title} {channelSource && ( `${title}, branch of ${parentTitle}`, loadMoreToSearch: "Load more to search older conversations", loadingMore: "Loading more...", loadOlderChats: "Load older chats", diff --git a/frontend/src/core/i18n/locales/types.ts b/frontend/src/core/i18n/locales/types.ts index bfe668fe6..af0df8e20 100644 --- a/frontend/src/core/i18n/locales/types.ts +++ b/frontend/src/core/i18n/locales/types.ts @@ -509,6 +509,7 @@ export interface Translations { // Chats chats: { searchChats: string; + branchLabel: (title: string, parentTitle: string) => string; loadMoreToSearch: string; loadingMore: string; loadOlderChats: string; diff --git a/frontend/src/core/i18n/locales/zh-CN.ts b/frontend/src/core/i18n/locales/zh-CN.ts index d8511b4da..153001688 100644 --- a/frontend/src/core/i18n/locales/zh-CN.ts +++ b/frontend/src/core/i18n/locales/zh-CN.ts @@ -595,6 +595,7 @@ export const zhCN: Translations = { // Chats chats: { searchChats: "搜索对话", + branchLabel: (title, parentTitle) => `${title}，分叉自 ${parentTitle}`, loadMoreToSearch: "加载更多以搜索更早的对话", loadingMore: "正在加载...", loadOlderChats: "加载更早的对话", diff --git a/frontend/src/core/threads/thread-branch-tree.ts b/frontend/src/core/threads/thread-branch-tree.ts new file mode 100644 index 000000000..5dcb3aefb --- /dev/null +++ b/frontend/src/core/threads/thread-branch-tree.ts @@ -0,0 +1,160 @@ +import type { AgentThread } from "./types"; +import { isThreadPinned } from "./utils"; + +const THREAD_BRANCH_METADATA_KEY = "deerflow_branch"; +const THREAD_BRANCH_PARENT_METADATA_KEY = "branch_parent_thread_id"; + +export type ThreadBranchEntry = { + thread: AgentThread; + parentThread?: AgentThread; + depth: number; + isLastSibling: boolean; +}; + +function recencyOfThread(thread: AgentThread) { + const timestamp = Date.parse(thread.updated_at ?? thread.created_at ?? ""); + return Number.isFinite(timestamp) ? timestamp : 0; +} + +function branchParentId(thread: AgentThread) { + if (thread.metadata?.[THREAD_BRANCH_METADATA_KEY] !== true) { + return null; + } + const parentId = thread.metadata?.[THREAD_BRANCH_PARENT_METADATA_KEY]; + if (typeof parentId !== "string") { + return null; + } + return parentId.trim() || null; +} + +/** + * Project the loaded flat thread page into a safe visual lineage. + * + * Only loaded parents in the same pinned partition can own a child. Missing, + * malformed, cross-pin, self-parented, and cyclic links remain top-level so + * partial pagination or bad metadata can never hide a conversation. + */ +export function flattenThreadBranches( + threads: readonly AgentThread[], +): ThreadBranchEntry[] { + if (threads.length < 2) { + return threads.map((thread) => ({ + thread, + depth: 0, + isLastSibling: true, + })); + } + + const byId = new Map(threads.map((thread) => [thread.thread_id, thread])); + const sourceIndex = new Map( + threads.map((thread, index) => [thread.thread_id, index]), + ); + const candidateParentByChild = new Map(); + + for (const thread of threads) { + const parentId = branchParentId(thread); + const parent = parentId ? byId.get(parentId) : undefined; + if ( + !parent || + parent.thread_id === thread.thread_id || + isThreadPinned(parent) !== isThreadPinned(thread) + ) { + continue; + } + candidateParentByChild.set(thread.thread_id, parent.thread_id); + } + + const hasCyclicAncestry = (threadId: string) => { + const visited = new Set([threadId]); + let parentId = candidateParentByChild.get(threadId); + while (parentId) { + if (visited.has(parentId)) { + return true; + } + visited.add(parentId); + parentId = candidateParentByChild.get(parentId); + } + return false; + }; + + const parentByChild = new Map(); + for (const [childId, parentId] of candidateParentByChild) { + if (!hasCyclicAncestry(childId)) { + parentByChild.set(childId, parentId); + } + } + + const childrenByParent = new Map(); + for (const thread of threads) { + const parentId = parentByChild.get(thread.thread_id); + if (!parentId) continue; + const children = childrenByParent.get(parentId) ?? []; + children.push(thread); + childrenByParent.set(parentId, children); + } + + for (const [parentId, children] of childrenByParent) { + const parent = byId.get(parentId); + if (!parent || isThreadPinned(parent)) continue; + children.sort( + (left, right) => + recencyOfThread(right) - recencyOfThread(left) || + (sourceIndex.get(left.thread_id) ?? 0) - + (sourceIndex.get(right.thread_id) ?? 0), + ); + } + + const groupRecencyCache = new Map(); + const groupRecency = (thread: AgentThread): number => { + const cached = groupRecencyCache.get(thread.thread_id); + if (cached !== undefined) return cached; + const recency = (childrenByParent.get(thread.thread_id) ?? []).reduce( + (latest, child) => Math.max(latest, groupRecency(child)), + recencyOfThread(thread), + ); + groupRecencyCache.set(thread.thread_id, recency); + return recency; + }; + + const roots = threads.filter( + (thread) => !parentByChild.has(thread.thread_id), + ); + roots.sort((left, right) => { + const pinnedDifference = + Number(isThreadPinned(right)) - Number(isThreadPinned(left)); + if (pinnedDifference) return pinnedDifference; + if (isThreadPinned(left)) { + return ( + (sourceIndex.get(left.thread_id) ?? 0) - + (sourceIndex.get(right.thread_id) ?? 0) + ); + } + return ( + groupRecency(right) - groupRecency(left) || + (sourceIndex.get(left.thread_id) ?? 0) - + (sourceIndex.get(right.thread_id) ?? 0) + ); + }); + + const entries: ThreadBranchEntry[] = []; + const emitted = new Set(); + const emit = ( + thread: AgentThread, + depth: number, + isLastSibling: boolean, + parentThread?: AgentThread, + ) => { + if (emitted.has(thread.thread_id)) return; + emitted.add(thread.thread_id); + entries.push({ thread, parentThread, depth, isLastSibling }); + const children = childrenByParent.get(thread.thread_id) ?? []; + children.forEach((child, index) => + emit(child, depth + 1, index === children.length - 1, thread), + ); + }; + + roots.forEach((root) => emit(root, 0, true)); + // Defense in depth: no malformed lineage should make a loaded thread vanish. + threads.forEach((thread) => emit(thread, 0, true)); + return entries; +} diff --git a/frontend/tests/e2e/branch-thread.spec.ts b/frontend/tests/e2e/branch-thread.spec.ts index b879a7e77..0df19a542 100644 --- a/frontend/tests/e2e/branch-thread.spec.ts +++ b/frontend/tests/e2e/branch-thread.spec.ts @@ -4,6 +4,7 @@ import { MOCK_THREAD_ID, MOCK_THREAD_ID_2, mockLangGraphAPI, + THREAD_PINNED_METADATA_KEY, } from "./utils/mock-api"; test.describe("Branch from turn", () => { @@ -103,7 +104,72 @@ test.describe("Branch from turn", () => { const branchThreadLink = page.locator( `a[href="/workspace/chats/${MOCK_THREAD_ID_2}"]`, ); - await expect(branchThreadLink).toContainText("Original chat"); + await expect(branchThreadLink).toContainText("Original chat (2)"); await expect(branchThreadLink).not.toContainText("Branch:"); + await expect(branchThreadLink).toHaveAttribute( + "data-branch-parent-id", + MOCK_THREAD_ID, + ); + await expect(branchThreadLink).toHaveAttribute("data-branch-depth", "1"); + await expect(branchThreadLink).toHaveAttribute( + "aria-label", + "Original chat (2), branch of Original chat", + ); + await expect(branchThreadLink.getByTestId("thread-branch-stem")).toHaveText( + "└─", + ); + + const recentChatHrefs = await page + .locator( + 'a[data-sidebar="menu-button"][href^="/workspace/chats/"]:not([href="/workspace/chats/new"])', + ) + .evaluateAll((links) => links.map((link) => link.getAttribute("href"))); + expect(recentChatHrefs).toEqual([ + `/workspace/chats/${MOCK_THREAD_ID}`, + `/workspace/chats/${MOCK_THREAD_ID_2}`, + ]); + }); + + test("keeps a pinned branch top-level when its parent is unpinned", async ({ + page, + }) => { + mockLangGraphAPI(page, { + threads: [ + { + thread_id: MOCK_THREAD_ID, + title: "Unpinned parent", + updated_at: "2026-08-24T00:00:00Z", + }, + { + thread_id: MOCK_THREAD_ID_2, + title: "Pinned branch (2)", + updated_at: "2026-08-24T00:01:00Z", + metadata: { + deerflow_branch: true, + branch_parent_thread_id: MOCK_THREAD_ID, + [THREAD_PINNED_METADATA_KEY]: true, + }, + }, + ], + }); + + await page.goto("/workspace/chats/new"); + + const branchLink = page.locator( + `a[href="/workspace/chats/${MOCK_THREAD_ID_2}"]`, + ); + await expect(branchLink).toBeVisible(); + await expect(branchLink).not.toHaveAttribute("data-branch-depth"); + await expect(branchLink.getByTestId("thread-branch-stem")).toHaveCount(0); + + const recentChatHrefs = await page + .locator( + 'a[data-sidebar="menu-button"][href^="/workspace/chats/"]:not([href="/workspace/chats/new"])', + ) + .evaluateAll((links) => links.map((link) => link.getAttribute("href"))); + expect(recentChatHrefs).toEqual([ + `/workspace/chats/${MOCK_THREAD_ID_2}`, + `/workspace/chats/${MOCK_THREAD_ID}`, + ]); }); }); diff --git a/frontend/tests/e2e/utils/mock-api.ts b/frontend/tests/e2e/utils/mock-api.ts index 31a7eb9a7..f75f3bb41 100644 --- a/frontend/tests/e2e/utils/mock-api.ts +++ b/frontend/tests/e2e/utils/mock-api.ts @@ -886,7 +886,25 @@ export function mockLangGraphAPI(page: Page, options?: MockAPIOptions) { if (sourceThread?.metadata?.deerflow_branch === true) { sourceTitle = sourceTitle?.replace(/^(Branch:\s*)+/i, "").trim(); } - const title = body.title ?? sourceTitle; + const sourceSequence = + sourceThread?.metadata?.deerflow_branch === true && + Number.isSafeInteger(sourceThread.metadata.branch_title_sequence) && + Number(sourceThread.metadata.branch_title_sequence) >= 2 && + Number(sourceThread.metadata.branch_title_sequence) < + Number.MAX_SAFE_INTEGER + ? Number(sourceThread.metadata.branch_title_sequence) + : undefined; + const sequence = sourceSequence === undefined ? 2 : sourceSequence + 1; + const sourceSuffix = sourceSequence ? ` (${sourceSequence})` : undefined; + const baseTitle = + sourceSuffix && sourceTitle?.endsWith(sourceSuffix) + ? sourceTitle.slice(0, -sourceSuffix.length).trimEnd() + : sourceTitle; + const title = + body.title ?? + (baseTitle + ? `${baseTitle.slice(0, 256 - ` (${sequence})`.length).trimEnd()} (${sequence})` + : undefined); upsertThread({ thread_id: MOCK_THREAD_ID_2, @@ -894,6 +912,7 @@ export function mockLangGraphAPI(page: Page, options?: MockAPIOptions) { updated_at: new Date().toISOString(), metadata: { deerflow_branch: true, + ...(!body.title && title ? { branch_title_sequence: sequence } : {}), branch_parent_thread_id: sourceThreadId, branch_parent_message_id: body.message_id, branch_parent_checkpoint_id: "mock-checkpoint", diff --git a/frontend/tests/unit/core/threads/thread-branch-tree.test.ts b/frontend/tests/unit/core/threads/thread-branch-tree.test.ts new file mode 100644 index 000000000..51c75c999 --- /dev/null +++ b/frontend/tests/unit/core/threads/thread-branch-tree.test.ts @@ -0,0 +1,196 @@ +import { describe, expect, it } from "@rstest/core"; + +import { flattenThreadBranches } from "@/core/threads/thread-branch-tree"; +import type { AgentThread } from "@/core/threads/types"; + +function thread( + id: string, + updatedAt: string, + metadata: Record = {}, +): AgentThread { + return { + thread_id: id, + updated_at: updatedAt, + created_at: updatedAt, + metadata, + status: "idle", + values: { title: id }, + } as AgentThread; +} + +function branch( + id: string, + parentId: unknown, + updatedAt: string, + metadata: Record = {}, +) { + return thread(id, updatedAt, { + deerflow_branch: true, + branch_parent_thread_id: parentId, + ...metadata, + }); +} + +function summarize(entries: ReturnType) { + return entries.map((entry) => ({ + depth: entry.depth, + id: entry.thread.thread_id, + isLastSibling: entry.isLastSibling, + parentId: entry.parentThread?.thread_id, + })); +} + +describe("flattenThreadBranches", () => { + it("nests loaded siblings and lifts the group by its freshest descendant", () => { + const parent = thread("parent", "2026-01-01T00:00:00Z"); + const child = branch("child", "parent", "2026-01-04T00:00:00Z"); + const sibling = branch("sibling", "parent", "2026-01-03T00:00:00Z"); + const other = thread("other", "2026-01-02T00:00:00Z"); + + expect( + summarize(flattenThreadBranches([child, sibling, other, parent])), + ).toEqual([ + { + depth: 0, + id: "parent", + isLastSibling: true, + parentId: undefined, + }, + { + depth: 1, + id: "child", + isLastSibling: false, + parentId: "parent", + }, + { + depth: 1, + id: "sibling", + isLastSibling: true, + parentId: "parent", + }, + { + depth: 0, + id: "other", + isLastSibling: true, + parentId: undefined, + }, + ]); + }); + + it("keeps recursive lineage and immediate-parent identity", () => { + const root = thread("root", "2026-01-04T00:00:00Z"); + const child = branch("child", "root", "2026-01-03T00:00:00Z"); + const grandchild = branch("grandchild", "child", "2026-01-02T00:00:00Z"); + + expect(summarize(flattenThreadBranches([root, child, grandchild]))).toEqual( + [ + { + depth: 0, + id: "root", + isLastSibling: true, + parentId: undefined, + }, + { + depth: 1, + id: "child", + isLastSibling: true, + parentId: "root", + }, + { + depth: 2, + id: "grandchild", + isLastSibling: true, + parentId: "child", + }, + ], + ); + }); + + it("does not move a pinned branch under an unpinned parent", () => { + const pinnedChild = branch( + "pinned-child", + "parent", + "2026-01-02T00:00:00Z", + { deerflow_pinned: true }, + ); + const parent = thread("parent", "2026-01-01T00:00:00Z"); + + expect(summarize(flattenThreadBranches([pinnedChild, parent]))).toEqual([ + { + depth: 0, + id: "pinned-child", + isLastSibling: true, + parentId: undefined, + }, + { + depth: 0, + id: "parent", + isLastSibling: true, + parentId: undefined, + }, + ]); + }); + + it("preserves pinned root order while nesting same-state children", () => { + const first = thread("first", "2026-01-01T00:00:00Z", { + deerflow_pinned: true, + }); + const second = thread("second", "2026-01-04T00:00:00Z", { + deerflow_pinned: true, + }); + const child = branch("child", "first", "2026-01-05T00:00:00Z", { + deerflow_pinned: true, + }); + + expect(summarize(flattenThreadBranches([first, second, child]))).toEqual([ + { + depth: 0, + id: "first", + isLastSibling: true, + parentId: undefined, + }, + { + depth: 1, + id: "child", + isLastSibling: true, + parentId: "first", + }, + { + depth: 0, + id: "second", + isLastSibling: true, + parentId: undefined, + }, + ]); + }); + + it("keeps missing, malformed, forged, self-parented, and cyclic branches visible at top level", () => { + const missing = branch("missing", "not-loaded", "2026-01-06T00:00:00Z"); + const malformed = branch("malformed", 42, "2026-01-05T00:00:00Z"); + const forged = thread("forged", "2026-01-04T00:00:00Z", { + branch_parent_thread_id: "missing", + }); + const self = branch("self", "self", "2026-01-03T00:00:00Z"); + const cycleA = branch("cycle-a", "cycle-b", "2026-01-02T00:00:00Z"); + const cycleB = branch("cycle-b", "cycle-a", "2026-01-01T00:00:00Z"); + + const entries = flattenThreadBranches([ + missing, + malformed, + forged, + self, + cycleA, + cycleB, + ]); + + expect(entries.map((entry) => entry.thread.thread_id)).toEqual([ + "missing", + "malformed", + "forged", + "self", + "cycle-a", + "cycle-b", + ]); + expect(entries.every((entry) => entry.depth === 0)).toBe(true); + }); +});