diff --git a/README.md b/README.md index 35ed26d7d9..5685e1a23e 100644 --- a/README.md +++ b/README.md @@ -1072,6 +1072,8 @@ Skill Markdown and bundled text resources use UTF-8. Skill-creator CLI and revie Users can explicitly activate an enabled skill for a single turn by starting the request with `/skill-name`, for example `/data-analysis analyze uploads/foo.csv`. DeerFlow loads that skill's `SKILL.md` as hidden current-turn context while leaving the base prompt limited to skill metadata. Slash activation respects disabled skills, custom-agent skill whitelists, and existing channel commands such as `/new` and `/help`. +After an answer loads skills, its toolbar includes **Skills used**. Hover over or click the icon to see the skills and their sources; hovering a skill name underlines it. Select a skill to inspect its `SKILL.md` in the resizable side panel (a drawer on mobile). The view uses snapshots captured during successful configured read-tool loads or explicit slash activation, so later edits or removal of a skill do not rewrite its history. Repeated loads appear once per run, in first-load order. Range reads and bounded snapshots are labeled as partial; older conversations without captured evidence do not show this menu. Copy returns the captured Markdown, including YAML frontmatter. Package-relative links and images remain readable references rather than navigating away from the conversation. Skill loads performed through other tools, such as shell commands, are not inferred from answer text. + An enabled skill's `allowed-tools` policy applies only after that skill is explicitly slash-activated or captured in the agent's active skill context after a `read_file` load. Merely enabling, advertising, or listing a skill in a custom agent or subagent `skills` allowlist does not reduce that agent's normal toolset; subagents use the same progressive discovery and activation policy as the lead agent. During a slash-activated run, that explicit skill's policy is authoritative: reading another `SKILL.md` may provide instructions but cannot widen the slash skill's tools. Without slash activation, policies from skills actually loaded into active context retain their union semantics. Once active, the policy filters both model-visible tool schemas and tool execution. Framework discovery tools (`tool_search` and `describe_skill`) remain available so an allowed deferred tool or installed skill can still be discovered, but discovery and promotion never grant permission to execute a business tool omitted from `allowed-tools`. `task` is not framework-exempt; a restrictive skill must list it explicitly to delegate to a subagent. Per-step policy decisions are internal runtime context and are removed from observable or persisted context copies. Registry failures and an active set with no remaining valid skill fail closed to framework-safe tools; individual stale paths are ignored only when another valid active skill remains. This is best-effort behavioral scoping, not a hard security boundary: loading skill instructions through another tool is not captured, and active-skill entries can be evicted from bounded context. When you install `.skill` archives through the Gateway, DeerFlow accepts standard space-separated `allowed-tools`, optional frontmatter metadata, and the Claude-compatible `argument-hint` field instead of rejecting otherwise valid external skills. YAML lists remain supported for `allowed-tools` and preserve exact runtime names. Exact portable spellings such as `WebFetch`, `WebSearch`, `Glob`, `Grep`, and `Read` map to DeerFlow's `web_fetch`, `web_search`, `glob`, `grep`, and `read_file` tools; lowercase or otherwise unknown scalar names remain unchanged so custom and MCP tools keep their exact runtime spelling. Parenthesized entries such as `Bash(tvly *)` are tokenized as one literal entry, including spaces, quoted text, and escaped parentheses, but remain inactive because DeerFlow does not inspect tool arguments; declare `bash` only when the skill may use the full Bash tool. diff --git a/README_zh.md b/README_zh.md index d37f96f910..dd565231eb 100644 --- a/README_zh.md +++ b/README_zh.md @@ -643,6 +643,8 @@ Skills 是 DeerFlow 能做“几乎任何事”的关键。 Skills 采用按需渐进加载,不会一次性把所有内容都塞进上下文。只有任务确实需要时才加载,这样能把上下文窗口控制得更干净,也更适合对 token 比较敏感的模型。 +回答加载过技能时,底部工具栏会显示「使用的技能」。悬停或点击图标可查看本轮技能及其来源,悬停技能名称显示下划线,点击名称即可在可拖拽侧栏(手机端为抽屉)查看当时加载的 `SKILL.md` 快照。自动读取和 `/技能名` 显式调用均会记录,同一技能按首次加载顺序去重;之后修改或删除技能不会改变这份历史。复制得到包含 YAML 元数据的原始 Markdown 快照;包内相对链接和图片显示为引用,不会误跳转离开对话。范围读取或超出快照大小上限时会标明内容不完整。旧对话、失败读取以及通过 shell 等其他工具加载的技能不会凭回答文本推断为已使用。 + 通过 Gateway 安装 `.skill` 压缩包时,DeerFlow 会接受标准的可选 frontmatter 元数据,比如 `version`、`author`、`compatibility`,不会把本来合法的外部 skill 拒之门外。 Tools 也是同样的思路。DeerFlow 自带一组核心工具:网页搜索、网页抓取、网页渲染截图、文件操作、bash 执行;同时也支持通过 MCP Server 和 Python 函数扩展自定义工具。你可以替换任何一项,也可以继续往里加。 diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 6dfb70961a..5d13700031 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -42,6 +42,7 @@ from deerflow.agents.human_input import read_human_input_response from deerflow.agents.middlewares.dynamic_context_middleware import _DYNAMIC_CONTEXT_REMINDER_KEY, _REMINDER_DATE_KEY from deerflow.agents.middlewares.input_sanitization_middleware import frame_untrusted_text from deerflow.agents.middlewares.message_utils import _SUMMARY_MESSAGE_NAME, is_genuine_user_message +from deerflow.agents.middlewares.skill_usage import SKILL_USAGE_KEY, SKILL_USAGES_KEY from deerflow.agents.middlewares.tool_receipt import TOOL_RECEIPT_KEY, TOOL_RECEIPT_LEDGER_KEY from deerflow.agents.middlewares.tool_transform_meta import TOOL_TRANSFORMS_KEY from deerflow.agents.middlewares.view_image_middleware import _IMAGE_CONTEXT_MESSAGE_MARKER_KEY @@ -139,6 +140,8 @@ _SERVER_OWNED_MESSAGE_METADATA_KEYS = ( TOOL_RECEIPT_KEY, TOOL_RECEIPT_LEDGER_KEY, TOOL_TRANSFORMS_KEY, + SKILL_USAGE_KEY, + SKILL_USAGES_KEY, # Attached when a values frame is serialized, for display ordering only. # A replayed message carrying it back would write a thread-scoped seq # into the checkpoint, which a fork then re-seeds and reassigns (#4380). diff --git a/backend/docs/runtime-guidance-details.md b/backend/docs/runtime-guidance-details.md new file mode 100644 index 0000000000..a75ec8ee0a --- /dev/null +++ b/backend/docs/runtime-guidance-details.md @@ -0,0 +1,222 @@ +# Runtime implementation details + +The runtime AGENTS.md indexes these detailed contracts. Preserve their behavior when changing the corresponding code. + +**Message feed seq stamping** (#4666): a checkpoint carries no position of its +own and loses messages to summarization, so a client merging a `values` frame +with the seq-ordered `run_events` feed cannot place a checkpoint-kept message +once the feed's loaded page window no longer reaches back to it. +`RunEventStore.get_message_seqs(thread_id, identities)` resolves the seq the +store already assigned, keyed by +`runtime/events/message_identity.py::message_identity` — the backend half of +the identity rule `frontend/src/core/threads/hooks.ts::messageIdentity` applies +(tool messages by `tool_call_id`; `X` / `X__user` human copies collapse to +one). The two halves must stay in sync: a mismatch is silent, degrading +placement rather than raising. `runs/worker.py::_MessageSeqStamper` attaches +the result as `additional_kwargs.deerflow_seq` on root `values` frames only — +subgraph frames are not part of the thread feed's ordering, and nothing is +written back to the checkpoint. The run-scoped cache makes the compaction frame +the only one that costs a lookup, and the stamper soft-resolves the user id +once at build time — like the worker's write paths beside it — so a launch path +that never inherits the auth contextvar (a null-owner scheduled task) still +stamps instead of the db store's strict `AUTO` default raising per frame. A +resolved seq is cached for the run (earliest-seq-wins makes it final), but a +miss is not: a message this run produces reaches a frame before `RunJournal` +flushes it, so it misses and is persisted moments later. Misses are re-asked +when `RunJournal.feed_generation` — bumped once per successful event-store +write, never while the buffer merely fills — shows the feed gained rows, which +keeps the retry bounded by writes rather than by frames and makes a failed +lookup cost one generation instead of the run. REST +reads (`GET /threads/{id}/state`, `POST /threads/{id}/history`) stamp through +`events/message_seq.py::stamp_messages_with_seq`, the request-scoped +counterpart: everything a checkpoint still holds is already persisted, so one +batched lookup resolves the whole list. The db store prefilters candidate rows +in SQL (a LIKE clause per wanted raw id, wildcards escaped; an id `json.dumps` +would escape falls the set back to the full scan) so a wanted identity absent +from the feed — a message still streaming — does not force a full +fetch-and-decode of every message row's tool outputs on long threads. +`deerflow_seq` is server-owned display metadata: the gateway strips it from +client input, because a welded-in seq goes stale when a fork re-seeds the feed +(#4380). + +**Run delivery receipts** (`runtime/journal.py` + `runs/worker.py`): +`RunJournal` records each non-empty artifact update once per tool `Command` for +the terminal `run.delivery` event. When a command contains multiple messages, a +unique tool name resolved from matching `ToolMessage` entries supplies +attribution; additional command messages do not duplicate artifact paths or +counts. If multiple different tool names resolve for one flat artifact update, +the paths remain counted but unattributed because the command does not carry a +per-path mapping. `RunJournal` callbacks set `run_inline=True`: they do only +in-memory bookkeeping or schedule async writes, and staying on the run's +event-loop thread serializes parallel tool callbacks before terminal delivery +recording and flushing. Each worker creates a separate journal per run before +cancellable/fallible preflight work, so checkpoint compatibility failures and +cancellation while waiting for prior finalization still emit a zero-delivery +receipt. The worker flushes ordinary journal events, idempotently persists the +run-scoped receipt, and only then persists the staged terminal run status. A +receipt failure is retried on a short bounded schedule while the owning worker +still knows the real outcome and holds the lease. Delivery candidates are every +regular file created or modified under `/mnt/user-data/outputs`; internal +process-feedback files are excluded (the scanner's `EXCLUDED_DIR_NAMES` plus +the configured `tool_output.storage_subdir`), so a run that only externalized +oversized tool outputs does not fail delivery. At least one candidate must be +covered by a path attributed by the journal to `present_files`; presenting only +an unrelated pre-existing path does not satisfy delivery. Receipts for such +runs add `produced_paths`, `presented_paths`, `matched_paths`, `verification`, +`stage`, and `satisfied` to the Slice 1 fact fields. Missing a matching +presentation becomes a run error; a successful presentation is also downgraded +to error if its receipt cannot be durably verified. Runs without changed +outputs preserve ordinary chat behavior and the original receipt shape. Orphan +recovery first atomically claims an expired lease, then uses the same singleton +write to backfill a zero-delivery receipt — a stale recovery scan cannot +overwrite a live run's later detailed receipt, an event-store outage does not +undo the terminal takeover, and an existing detailed receipt is preserved when +a worker crashed after writing it. Event stores serialize `put_if_absent` with +ordinary thread writers: memory and JSONL provide the documented +single-process guarantee, while the DB store adds per-thread in-process locks +and PostgreSQL advisory locks for cross-process writers. Moving journal +construction ahead of preflight is receipt-only on early failure paths: a +separate boundary flag preserves the previous completion-data semantics, so +checkpoint incompatibility or cancellation while waiting for an older +finalizing run does not persist an empty completion snapshot. Worker tests pin +one accumulated receipt across multiple goal-continuation `_stream_once` calls; +journal tests drive LangChain's real async callback dispatcher against a single +journal to pin serialized, deduplicated parallel tool callbacks. + +**Extension changed-run discovery** (`runtime/runs/store/` and +`extensions/run_evidence.py`) orders public run-record changes by the +backend-owned `(change_seq, run_id)` key rather than timestamps or per-thread +event sequence numbers. The singleton SQL clock allocates positions in the same +transaction as each public record mutation; memory uses a process-local counter. +The clock row serializes position-bearing SQL mutations globally until their +transactions commit, so every mutator must acquire it before any run-row lock. +An atomic thread operation uses one position for its interrupted rows and new +row, with `run_id` ordering ties. High-frequency progress snapshots and lease +heartbeats do not advance this position: they are not lifecycle-discovery +signals, progress evidence remains available from the event stream, and +excluding them bounds clock contention. A later lifecycle change exposes the +run row's latest `updated_at` and accumulated progress fields to internal readers. +Rows from before the migration retain `change_seq=0` and page deterministically +by run id. Because a later mutation only moves a row forward, concurrent paging +may replay a run but cannot move an unseen run behind the committed cursor. +Deletes produce no tombstone and therefore do not advance the cursor; consumers +that require deletion reconciliation must poll authoritative status for known +runs and treat a missing result as absent. +The extension-facing cursor is versioned, opaque, and bound to the reader's +fixed user scope. `RunEventStore.list_events()` accepts the same explicit +`user_id` override as `list_messages()`; evidence readers must propagate their +bound scope (including global `None`) rather than resolving an ambient request +user. The adapter deep-copies event content and redacted metadata before +exposing them, detaching nested mutable payloads from host storage. + +Gateway `POST /api/threads/{id}/history` uses that lookup to migrate legacy AI +messages. An exhaustive miss preserves the human-boundary fallback; an +incomplete lookup removes unproven synthesized IDs. Its metadata-only +write-on-read cache stores `run_message_ids` for every audited AI ID (including +exhaustive misses) plus required `run_durations`; duration presence alone does +not prove attribution. Historical `body.before` reads write the audit to the +head, and the merge may retain IDs no longer in materialized history, which +readers ignore. Migration must acquire the durable `checkpoint_write` +reservation, then repeat the whole message audit and batch-reload required run +rows before persisting. Post-admission exact hits replace foreground exact or +boundary mappings, and recomputed final durations replace foreground snapshots. +Successful workers keep their durable run row active through the final duration +checkpoint write, so a peer migration cannot enter during terminalization. +The first `RunManager.list_by_thread()` hydration page uses a 100-row floor or +the number of required IDs, whichever is larger; missing exact runs use targeted +`get()` calls. + +**Terminal run cleanup explicitly breaks graph-scoped references while preserving the existing `RunRecord` grace period.** Every `agent.astream()` iterator is closed in `_stream_once`, including abort/exception/early-break paths. A close failure after an abort is warning-only and cannot replace the user-requested `interrupted` outcome; normal-completion close failures still surface, and an in-flight stream exception remains authoritative over a secondary close failure. Journal construction and cancellable preflight work (including MCP task projection and the prior-finalization wait) live inside the worker's guarded body, so cancellation before agent startup still terminalizes the run and closes its stream. `run_agent()` wraps the complete terminal-finalization sequence in an outer teardown guard, so cancellation or failure from any terminal-stage await cannot skip `RunJournal.close()`, removal of the journal, `__pregel_runtime`, and internal runtime-context values from every runnable config, or release of local graph/payload references. That guard schedules bridge cleanup, run-record cleanup, and cyclic GC even when interruption happens before the terminal stream marker or terminal publication itself fails, so neither a cancelled observer nor a delivery-backend outage can strand process-local run state. A non-`Exception` `BaseException` caught while awaiting the completion hook or task-stop notification (including host-task cancellation) is deferred through the ordinary remaining finalization, with the first interruption preserved and every caught host-task `CancelledError` balanced by calling `Task.uncancel()` until the current task’s cumulative cancellation count is clear. Task-stop fan-out runs in one child task and every host wait uses `shield`, so repeated cancellation of the worker cannot cancel that fan-out or skip later observers; the worker keeps awaiting the same child task. A rogue observer that raises its own `CancelledError` remains contained by the extension dispatcher and distinguishable from host cancellation. This guarantee applies only to cancellation caught during those hook stages: clearing the finalizing barrier and publishing END remain direct awaits, so another cancellation in the subsequent critical tail retains forceful-termination semantics instead of creating an unbounded shield. If that tail completes without another interruption, the first deferred interruption is re-raised after END; a barrier-clear failure prevents END publication, while an END failure is raised after the barrier is clear. `RunJournal.flush()` clears its `_pending_progress_task` after awaiting or cancelling it; ordinary `close()` detaches the event store/progress reporter and clears callback bookkeeping only after that flush succeeds, preserving the buffer for retry on a transient store failure. A fenced worker instead calls `close(flush=False)`, which cancels pending journal work and detaches without initiating another event-store write after lease ownership is lost; its final detach runs even if a second cancellation interrupts pending-task shutdown. `RunManager.cleanup(run_id)` retains the process-local `RunRecord`, completed task, and request payload for its default 300-second local join/status window before releasing them, and evicts only when a `RunStore` backs the manager — without one there is no hydration fallback, so the record is retained rather than silently dropping the run's history. Durable history remains in `RunStore`; `StreamBridge` data keeps its separate 60-second late-subscriber window, and both cleanup coroutines run in a fresh empty `contextvars.Context`. A contextless full cyclic-GC pass, coalesced to at most once every 10 seconds and dispatched through the default executor, bounds the lifetime of unreachable LangGraph callback/loop cycles without synchronously walking the heap in the event-loop timer; passes taking at least 100 ms are logged at INFO because CPython GC may still impose interpreter-level pauses. + +**Checkpoint channel benchmark**: `scripts/benchmark/checkpoint/bench_channels.py` +runs paired `full`/`delta` message-only StateGraphs in a fresh child process per +case, using sync `InMemorySaver` or `SqliteSaver` so reducer, serialization, and +saver costs stay separate from Gateway/async scheduling. Optional +`AsyncPostgresSaver` cases are enabled only when `TEST_POSTGRES_URI` is set. +Postgres cases use a unique thread and remove only that benchmark thread through +the saver's public `adelete_thread` API after measurement. It reports deterministic +correctness digests, write windows/percentiles, warm and graph-rebuilt cold reads, +backend-neutral checkpoint/blob/write row and byte fields, aggregate logical +checkpoint/write bytes, SQLite DB/WAL/SHM footprint, reducer replay time, and +peak RSS as versioned JSONL. SQLite embeds channel blobs in its checkpoint +payload, so its separate blob metrics are zero; Postgres reports its +`checkpoint_blobs` table separately. Byte fields describe each saver's serialized +representation and should not be treated as identical encodings across backends. +The controller alternates mode order and rejects +performance data when paired modes materialize different state. Its default 1 GiB +estimated cumulative full-payload cap skips both modes of an oversized pair when +`full` is selected, including every delta cadence in a `--snapshot-frequencies` +sweep; intentional `--modes delta` diagnostics bypass this full-payload cap, so +size those runs explicitly. Use `--allow-large-cases` only +on a provisioned machine. Duplicate CSV matrix values are ignored with a warning; +use `--repetitions` for repeated samples. Summarize paired successful repetitions +with `scripts/benchmark/checkpoint/summarize_channels.py` (all ratios are +`delta/full`). `--profile-dir /tmp/checkpoint-profiles` writes one cProfile +artifact per case for attribution. Profiled rows carry `profiled: true`, and the +summarizer automatically excludes them from baseline summaries with a warning. +Storage-size collection relies on saver-specific diagnostic layouts; if those +layouts change, the timing/correctness row remains successful while storage +fields become `null` and `storage_stats_error` records the diagnostic failure. +Example: + +```bash +cd backend +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ + --backends sqlite --updates 100,500,999,1000,1001 --payload-bytes 128 \ + --repetitions 7 --output /tmp/checkpoint-bench.jsonl +TEST_POSTGRES_URI=postgresql://... \ +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ + --backends sqlite,postgres --updates 100 --payload-bytes 128 \ + --output /tmp/checkpoint-cross-backend.jsonl +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_channels.py \ + /tmp/checkpoint-bench.jsonl +``` + +The production-shaped layer lives in +`scripts/benchmark/checkpoint/bench_production.py`: per-case child processes +run graph-level `ainvoke` turns through the real lead-agent graph (scripted +deterministic model, real `AsyncSqliteSaver`), then measure +`GET /threads/{id}/state` and `POST /threads/{id}/history` through the real +Gateway route stack in the same event loop (httpx ASGITransport), split into +cold/warm accessor-graph-cache samples. It sweeps `snapshot_frequency` +(config: `checkpoint_delta.snapshot_frequency`, process-frozen like the mode), +pairs every delta frequency against the same full row, and fails both +rows of a pair when materialized or wire digests diverge. Each case must have +more than the two discarded warm-up turns, and SQLite DB/WAL/SHM sizes are +captured while the saver is still open so they represent the online storage +footprint. Summarize with +`scripts/benchmark/checkpoint/summarize_production.py` (ratios are +`delta/full`; it also emits `snapshot_write_spike` and `cache_effect_ms`, +the decision inputs for the production snapshot-frequency and accessor-cache +defaults). Harness tests live in `tests/test_bench_checkpoint_production.py` +and `tests/test_summarize_checkpoint_production.py`; timing thresholds are +not CI gates. The matrix test pins that every `(repetition, turns)` group +contains both modes and that their execution order flips between consecutive +groups, including across repetition boundaries. + +Operational limits learned from the first runs (the default matrix is too +large to run blindly): + +- The default `--timeout-seconds 900` is insufficient for delta mode at + `snapshot_frequency=1000` once turns reach 500 (measured: delta-500 takes + ~1100-1200s; delta-2000 takes ~45min). Pass an explicit + `--timeout-seconds` for any large matrix, and treat the turns=2000 corner + as practical only at small snapshot frequencies. +- Full-mode 2000-turn runs produce a ~33GB sqlite DB. Point `TMPDIR` at real + disk, not tmpfs (the benchmark uses `tempfile.TemporaryDirectory`, which + honors `TMPDIR`), or the run dies mid-case. +- The history route clamps `limit` to 100 (`le=100` on + `ThreadHistoryRequest.limit`), so `--history-limits` values above 100 are + measured and reported by their effective (clamped) limit. + +Example: + +```bash +cd backend +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_production.py \ + --turns 10,100,500,1000,2000 --payload-bytes 128 \ + --snapshot-frequencies 10,50,100,500,1000 \ + --repetitions 7 --output /tmp/production-bench.jsonl +PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py \ + /tmp/production-bench.jsonl +``` diff --git a/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py index ebc8d8cb97..fa848ac7e3 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/skill_activation_middleware.py @@ -18,6 +18,7 @@ from langchain.agents.middleware import AgentMiddleware from langchain.agents.middleware.types import ModelRequest, ModelResponse from langchain_core.messages import AIMessage, HumanMessage +from deerflow.agents.middlewares.skill_usage import SKILL_USAGE_KEY, build_skill_usage, record_skill_usage from deerflow.runtime.events.catalog import ( MIDDLEWARE_SKILL_ACTIVATION_TAG, MIDDLEWARE_SKILL_SECRETS_TAG, @@ -358,13 +359,41 @@ Follow this skill before choosing a general workflow. Load supporting resources messages.insert(target_index, activation_msg) return request.override(messages=messages), activation - def _handle_model_request(self, request: ModelRequest, *, hook: str) -> ModelRequest | AIMessage: + def _handle_model_request(self, request: ModelRequest, *, hook: str) -> tuple[ModelRequest | AIMessage, _Activation | None]: prepared, activation = self._prepare_model_request(request, hook=hook) if isinstance(prepared, AIMessage): - return prepared + return prepared, None effective = prepared if prepared is not None else request self._resolve_secret_bindings(effective, activation, hook=hook) - return effective + if activation is not None: + record_skill_usage(getattr(request, "runtime", None), self._usage_snapshot(activation)) + return effective, activation + + @staticmethod + def _usage_snapshot(activation: _Activation) -> dict | None: + return build_skill_usage( + activation.container_file_path, + activation.skill_content, + skills_root=posixpath.dirname(activation.container_file_path), + activation="slash", + name=activation.skill_name, + category=activation.category, + ) + + @staticmethod + def _stamp_usage(response: ModelResponse | AIMessage, activation: _Activation | None) -> ModelResponse | AIMessage: + if activation is None: + return response + # Embedded/checkpoint clients also keep evidence on the first response. + # The journal independently attaches a run aggregate before serialization. + usage = SkillActivationMiddleware._usage_snapshot(activation) + if usage is not None: + messages = [response] if isinstance(response, AIMessage) else getattr(response, "result", []) + for message in messages: + if isinstance(message, AIMessage): + message.additional_kwargs = {**message.additional_kwargs, SKILL_USAGE_KEY: usage} + break + return response def _resolve_secret_bindings(self, request: ModelRequest, activation: _Activation | None, *, hook: str) -> None: """Recompute the per-run secret injection set (binding point A+, #3861/#3914). @@ -576,10 +605,10 @@ Follow this skill before choosing a general workflow. Load supporting resources request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse], ) -> ModelResponse | AIMessage: - prepared = self._handle_model_request(request, hook="wrap_model_call") + prepared, activation = self._handle_model_request(request, hook="wrap_model_call") if isinstance(prepared, AIMessage): return prepared - return handler(prepared) + return self._stamp_usage(handler(prepared), activation) @override async def awrap_model_call( @@ -587,7 +616,7 @@ Follow this skill before choosing a general workflow. Load supporting resources request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]], ) -> ModelResponse | AIMessage: - prepared = await asyncio.to_thread(self._handle_model_request, request, hook="awrap_model_call") + prepared, activation = await asyncio.to_thread(self._handle_model_request, request, hook="awrap_model_call") if isinstance(prepared, AIMessage): return prepared - return await handler(prepared) + return self._stamp_usage(await handler(prepared), activation) diff --git a/backend/packages/harness/deerflow/agents/middlewares/skill_usage.py b/backend/packages/harness/deerflow/agents/middlewares/skill_usage.py new file mode 100644 index 0000000000..e9b93e1066 --- /dev/null +++ b/backend/packages/harness/deerflow/agents/middlewares/skill_usage.py @@ -0,0 +1,68 @@ +"""Bounded, display-only snapshots of skill content actually loaded by the agent. + +Kept on the producing message so normal run history supplies ownership, ordering +and persistence. This metadata must never be used for tool/secret authorization. +""" + +import hashlib +import logging +import posixpath + +import yaml + +from deerflow.agents.middlewares.skill_context import _FRONT_MATTER_RE, build_skill_entry_metadata_from_read +from deerflow.sandbox.read_file_contract import READ_FILE_NO_CONTENT_RESULTS, READ_FILE_TRUNCATION_PREFIX + +SKILL_USAGE_KEY = "skill_usage" +SKILL_USAGES_KEY = "skill_usages" +MAX_SKILL_SNAPSHOT_CHARS = 100_000 +logger = logging.getLogger(__name__) + + +def record_skill_usage(runtime: object, usage: dict | None) -> None: + """Register evidence before the next model callback snapshots canonical history.""" + context = getattr(runtime, "context", None) + if usage is None or not isinstance(context, dict) or context.get("is_subagent"): + return + journal = context.get("__run_journal") + record = getattr(journal, "record_skill_usage", None) + if callable(record): + try: + record(usage) + except Exception: + logger.warning("Failed to record skill usage display snapshot", exc_info=True) + + +def build_skill_usage( + path: str, + content: str, + *, + skills_root: str, + activation: str = "automatic", + name: str | None = None, + category: str | None = None, + partial: bool = False, +) -> dict | None: + entry = build_skill_entry_metadata_from_read(path, content, skills_root=skills_root) + if entry is None or not content.strip() or content.strip() in READ_FILE_NO_CONTENT_RESULTS: + return None + normalized_path = entry["path"] + relative = posixpath.relpath(normalized_path, posixpath.normpath(skills_root)) + if name is None: + match = _FRONT_MATTER_RE.match(content) + try: + metadata = yaml.safe_load(match.group(1)) if match else None + except yaml.YAMLError: + metadata = None + value = metadata.get("name") if isinstance(metadata, dict) else None + name = value if isinstance(value, str) and value.strip() else posixpath.basename(posixpath.dirname(normalized_path)) + return { + "name": name[:256], + "description": entry["description"], + "category": category or {"public": "public", "custom": "custom", "integrations": "integrations", "legacy": "legacy"}.get(relative.split("/")[0], "legacy"), + "path": normalized_path, + "content": content[:MAX_SKILL_SNAPSHOT_CHARS], + "content_hash": hashlib.sha256(content.encode("utf-8")).hexdigest(), + "activation": activation, + "partial": partial or len(content) > MAX_SKILL_SNAPSHOT_CHARS or READ_FILE_TRUNCATION_PREFIX in content, + } diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py index 8efc78b86c..fcfb81d17a 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py @@ -14,10 +14,13 @@ from langgraph.types import Command from deerflow.agents.middlewares.skill_context import ( SKILL_CONTEXT_ENTRY_KEY, + _tool_call_id, _tool_call_path, build_skill_entry_metadata_from_read, ) +from deerflow.agents.middlewares.skill_usage import SKILL_USAGE_KEY, build_skill_usage from deerflow.agents.middlewares.tool_result_meta import ( + TOOL_META_KEY, normalize_tool_result, stamp_exception_meta, ) @@ -98,6 +101,9 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): return message if getattr(message, "status", "success") == "error": return message + tool_meta = message.additional_kwargs.get(TOOL_META_KEY) + if isinstance(tool_meta, dict) and tool_meta.get("status") == "error": + return message content = message.content if isinstance(message.content, str) else None if content is None: return message @@ -109,15 +115,47 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): return message existing = dict(message.additional_kwargs or {}) existing[SKILL_CONTEXT_ENTRY_KEY] = dict(entry) + args = request.tool_call.get("args") or {} + usage = build_skill_usage( + path, + content, + skills_root=self._skills_root, + partial=args.get("start_line") is not None or args.get("end_line") is not None, + ) + if usage is not None: + existing[SKILL_USAGE_KEY] = usage message.additional_kwargs = existing return message def _maybe_stamp(self, result: ToolMessage | Command, request: ToolCallRequest) -> ToolMessage | Command: """Apply producer-bound metadata for tool results that need it.""" - if not isinstance(result, ToolMessage): - return result tool_name = str(request.tool_call.get("name") or "") - return self._stamp_skill_read_metadata(result, request, tool_name=tool_name) + tool_call_id = _tool_call_id(request.tool_call) + + def stamp(message: ToolMessage) -> None: + # Tool-returned kwargs are untrusted. Only this middleware may add + # skill evidence after checking the configured producer and path. + existing = dict(message.additional_kwargs or {}) + existing.pop(SKILL_CONTEXT_ENTRY_KEY, None) + existing.pop(SKILL_USAGE_KEY, None) + message.additional_kwargs = existing + if tool_call_id is None or str(message.tool_call_id) == tool_call_id: + self._stamp_skill_read_metadata(message, request, tool_name=tool_name) + + if isinstance(result, ToolMessage): + stamp(result) + return result + update = getattr(result, "update", None) + if not isinstance(update, dict): + return result + messages = update.get("messages") + if isinstance(messages, ToolMessage): + stamp(messages) + elif isinstance(messages, (list, tuple)): + for message in messages: + if isinstance(message, ToolMessage): + stamp(message) + return result @override def wrap_tool_call( @@ -133,9 +171,9 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: logger.exception("Tool execution failed (sync): name=%s id=%s", request.tool_call.get("name"), request.tool_call.get("id")) return self._build_error_message(request, exc) - return normalize_tool_result( - self._maybe_stamp(result, request), - tool_call_id=str(request.tool_call.get("id") or ""), + return self._maybe_stamp( + normalize_tool_result(result, tool_call_id=str(request.tool_call.get("id") or "")), + request, ) @override @@ -152,9 +190,9 @@ class ToolErrorHandlingMiddleware(AgentMiddleware[AgentState]): except Exception as exc: logger.exception("Tool execution failed (async): name=%s id=%s", request.tool_call.get("name"), request.tool_call.get("id")) return self._build_error_message(request, exc) - return normalize_tool_result( - self._maybe_stamp(result, request), - tool_call_id=str(request.tool_call.get("id") or ""), + return self._maybe_stamp( + normalize_tool_result(result, tool_call_id=str(request.tool_call.get("id") or "")), + request, ) diff --git a/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py index 54825cc81c..85e5303d09 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/tool_output_budget_middleware.py @@ -24,6 +24,7 @@ read-before-write middleware's own policy; both rewrite through the shared from __future__ import annotations import asyncio +import hashlib import logging import os import posixpath @@ -40,12 +41,16 @@ from langchain_core.messages import AIMessage, ToolMessage from langgraph.prebuilt.tool_node import ToolCallRequest from langgraph.types import Command +from deerflow.agents.middlewares.skill_context import _tool_call_path, build_skill_entry_metadata_from_read +from deerflow.agents.middlewares.skill_usage import MAX_SKILL_SNAPSHOT_CHARS, SKILL_USAGE_KEY, record_skill_usage from deerflow.agents.middlewares.tool_call_args import ToolCallOccurrence, pair_tool_call_results, rewrite_messages_tool_call_args from deerflow.agents.middlewares.tool_output_synopsis import render_tool_output_preview from deerflow.agents.middlewares.tool_result_meta import TOOL_META_KEY from deerflow.agents.middlewares.tool_transform_meta import append_tool_transform from deerflow.community.ragflow.sources import budget_source_artifact +from deerflow.config.summarization_config import DEFAULT_SKILL_FILE_READ_TOOL_NAMES from deerflow.config.tool_output_config import ToolOutputConfig +from deerflow.constants import DEFAULT_SKILLS_CONTAINER_PATH from deerflow.sandbox.sandbox_provider import get_sandbox_provider if TYPE_CHECKING: @@ -537,9 +542,11 @@ def _needs_budget(result: ToolMessage | Command, config: ToolOutputConfig) -> bo return _tool_message_over_budget(result, config) update = getattr(result, "update", None) if isinstance(update, dict): - for msg in update.get("messages", []): - if isinstance(msg, ToolMessage) and _tool_message_over_budget(msg, config): - return True + messages = update.get("messages", []) + if isinstance(messages, ToolMessage): + return _tool_message_over_budget(messages, config) + if isinstance(messages, (list, tuple)): + return any(isinstance(msg, ToolMessage) and _tool_message_over_budget(msg, config) for msg in messages) return False @@ -558,7 +565,10 @@ def _patch_result( return result messages = update.get("messages") - if not isinstance(messages, list): + if isinstance(messages, ToolMessage): + patched = _patch_tool_message(messages, config, outputs_path, sandbox) + return result if patched is messages else dc_replace(result, update={**update, "messages": patched}) + if not isinstance(messages, (list, tuple)): return result new_messages: list[Any] = [] @@ -575,7 +585,59 @@ def _patch_result( if not changed: return result - return dc_replace(result, update={**update, "messages": new_messages}) + return dc_replace(result, update={**update, "messages": tuple(new_messages) if isinstance(messages, tuple) else new_messages}) + + +def _record_visible_skill_usage( + result: ToolMessage | Command, + request: ToolCallRequest, + *, + skill_read_tool_names: frozenset[str], + skills_root: str, +) -> ToolMessage | Command: + """Register the snapshot after output budgeting has determined model-visible content.""" + tool_call = request.tool_call + tool_name = str(tool_call.get("name") or "") + tool_call_id = str(tool_call.get("id") or "") + path = _tool_call_path(tool_call) + runtime = getattr(request, "runtime", None) + + def record(message: ToolMessage) -> ToolMessage: + if SKILL_USAGE_KEY not in message.additional_kwargs: + return message + usage = message.additional_kwargs[SKILL_USAGE_KEY] + entry = build_skill_entry_metadata_from_read(path, message.content, skills_root=skills_root) if path is not None and isinstance(message.content, str) else None + if tool_name not in skill_read_tool_names or str(message.tool_call_id) != tool_call_id or not isinstance(usage, dict) or entry is None or usage.get("path") != entry["path"]: + kwargs = dict(message.additional_kwargs) + kwargs.pop(SKILL_USAGE_KEY, None) + return message.model_copy(update={"additional_kwargs": kwargs}) + visible_hash = hashlib.sha256(message.content.encode("utf-8")).hexdigest() + if visible_hash != usage.get("content_hash"): + usage = { + **usage, + "content": message.content[:MAX_SKILL_SNAPSHOT_CHARS], + "content_hash": visible_hash, + "partial": True, + } + message = message.model_copy(update={"additional_kwargs": {**message.additional_kwargs, SKILL_USAGE_KEY: usage}}) + record_skill_usage(runtime, usage) + return message + + if isinstance(result, ToolMessage): + return record(result) + update = getattr(result, "update", None) + if not isinstance(update, dict): + return result + messages = update.get("messages") + if isinstance(messages, ToolMessage): + updated = record(messages) + return result if updated is messages else dc_replace(result, update={**update, "messages": updated}) + if not isinstance(messages, (list, tuple)): + return result + updated = [record(message) if isinstance(message, ToolMessage) else message for message in messages] + if all(new is old for new, old in zip(updated, messages)): + return result + return dc_replace(result, update={**update, "messages": tuple(updated) if isinstance(messages, tuple) else updated}) def _patch_model_messages(messages: list[Any], config: ToolOutputConfig) -> list[Any] | None: @@ -724,19 +786,31 @@ def _normalized_path_arg(args: Mapping[str, Any]) -> str | None: class ToolOutputBudgetMiddleware(AgentMiddleware[AgentState]): """Enforce per-result budget on tool outputs via externalization or truncation.""" - def __init__(self, config: ToolOutputConfig | None = None) -> None: + def __init__( + self, + config: ToolOutputConfig | None = None, + *, + skill_read_tool_names: list[str] | None = None, + skills_root: str = DEFAULT_SKILLS_CONTAINER_PATH, + ) -> None: super().__init__() self._config = config if config is not None else _default_config() + self._skill_read_tool_names = frozenset(skill_read_tool_names if skill_read_tool_names is not None else DEFAULT_SKILL_FILE_READ_TOOL_NAMES) + self._skills_root = skills_root def release_policy_parameters(self) -> dict[str, object]: - return {"config": self._config.model_dump(mode="python")} + return {"config": self._config.model_dump(mode="python"), "skill_read_tool_names": sorted(self._skill_read_tool_names), "skills_root": self._skills_root} @classmethod def from_app_config(cls, app_config: Any) -> ToolOutputBudgetMiddleware: tool_output = getattr(app_config, "tool_output", None) - if isinstance(tool_output, ToolOutputConfig): - return cls(config=tool_output) - return cls() + summarization = getattr(app_config, "summarization", None) + skills = getattr(app_config, "skills", None) + return cls( + config=tool_output if isinstance(tool_output, ToolOutputConfig) else None, + skill_read_tool_names=getattr(summarization, "skill_file_read_tool_names", None), + skills_root=getattr(skills, "container_path", DEFAULT_SKILLS_CONTAINER_PATH), + ) # -- tool call hooks --------------------------------------------------- @@ -747,13 +821,11 @@ class ToolOutputBudgetMiddleware(AgentMiddleware[AgentState]): handler: Callable[[ToolCallRequest], ToolMessage | Command], ) -> ToolMessage | Command: result = handler(request) - if not self._config.enabled: - return result - if not _needs_budget(result, self._config): - return result - outputs_path = _resolve_outputs_path(request) - sandbox = _resolve_sandbox(request) - return _patch_result(result, self._config, outputs_path, sandbox) + if self._config.enabled and _needs_budget(result, self._config): + outputs_path = _resolve_outputs_path(request) + sandbox = _resolve_sandbox(request) + result = _patch_result(result, self._config, outputs_path, sandbox) + return _record_visible_skill_usage(result, request, skill_read_tool_names=self._skill_read_tool_names, skills_root=self._skills_root) @override async def awrap_tool_call( @@ -762,17 +834,14 @@ class ToolOutputBudgetMiddleware(AgentMiddleware[AgentState]): handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]], ) -> ToolMessage | Command: result = await handler(request) - if not self._config.enabled: - return result - if not _needs_budget(result, self._config): - return result - outputs_path = _resolve_outputs_path(request) - # _resolve_sandbox only touches runtime.state and the provider's - # in-memory sandbox registry, so it is safe to call on the event - # loop. The actual sandbox I/O (mkdir/write/test) happens inside - # _patch_result, which is offloaded to a worker thread below. - sandbox = _resolve_sandbox(request) - return await asyncio.to_thread(_patch_result, result, self._config, outputs_path, sandbox) + if self._config.enabled and _needs_budget(result, self._config): + outputs_path = _resolve_outputs_path(request) + # _resolve_sandbox only touches runtime.state and the provider's + # in-memory sandbox registry, so it is safe to call on the event + # loop. The actual sandbox I/O happens in the worker thread. + sandbox = _resolve_sandbox(request) + result = await asyncio.to_thread(_patch_result, result, self._config, outputs_path, sandbox) + return _record_visible_skill_usage(result, request, skill_read_tool_names=self._skill_read_tool_names, skills_root=self._skills_root) # -- model call hooks (historical context budgeting) ------------------- diff --git a/backend/packages/harness/deerflow/runtime/AGENTS.md b/backend/packages/harness/deerflow/runtime/AGENTS.md index 507c24f420..b83b6dc40c 100644 --- a/backend/packages/harness/deerflow/runtime/AGENTS.md +++ b/backend/packages/harness/deerflow/runtime/AGENTS.md @@ -24,42 +24,7 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c **Run rollback flow** (`runtime/runs/worker.py`): `_capture_rollback_point` materializes the complete pre-run state via the accessor and captures raw `pending_writes` via `aget_tuple` into an immutable `RollbackPoint` before the run starts — capture failure disables rollback (fail-closed), never restores partial state. In `full` mode, cancel-with-rollback forks from the pre-run checkpoint via the mutation graph and inherits non-message channels from that parent. In `delta` mode, forking is unsafe once the cancelled path has attached sibling writes to the pre-run checkpoint, so rollback replaces every captured channel on the current head, using `Overwrite` for reducers and schema defaults for current-head-only channels. Resume and rollback state rewrites copy only the selected/captured checkpoint's server-authored agent binding; missing or malformed bindings remain unbound. Both modes reattach only the captured pre-run pending writes to the restored checkpoint. Edit replay runs (`metadata.replay_kind="edit"`) also restore the pre-run checkpoint on failed, timed-out, or interrupted completion and publish the restored `values` snapshot to the stream before `end`, so clients do not remain on a transient edited branch when the replay did not produce a successful replacement. -**Message feed seq stamping** (#4666): a checkpoint carries no position of its -own and loses messages to summarization, so a client merging a `values` frame -with the seq-ordered `run_events` feed cannot place a checkpoint-kept message -once the feed's loaded page window no longer reaches back to it. -`RunEventStore.get_message_seqs(thread_id, identities)` resolves the seq the -store already assigned, keyed by -`runtime/events/message_identity.py::message_identity` — the backend half of -the identity rule `frontend/src/core/threads/hooks.ts::messageIdentity` applies -(tool messages by `tool_call_id`; `X` / `X__user` human copies collapse to -one). The two halves must stay in sync: a mismatch is silent, degrading -placement rather than raising. `runs/worker.py::_MessageSeqStamper` attaches -the result as `additional_kwargs.deerflow_seq` on root `values` frames only — -subgraph frames are not part of the thread feed's ordering, and nothing is -written back to the checkpoint. The run-scoped cache makes the compaction frame -the only one that costs a lookup, and the stamper soft-resolves the user id -once at build time — like the worker's write paths beside it — so a launch path -that never inherits the auth contextvar (a null-owner scheduled task) still -stamps instead of the db store's strict `AUTO` default raising per frame. A -resolved seq is cached for the run (earliest-seq-wins makes it final), but a -miss is not: a message this run produces reaches a frame before `RunJournal` -flushes it, so it misses and is persisted moments later. Misses are re-asked -when `RunJournal.feed_generation` — bumped once per successful event-store -write, never while the buffer merely fills — shows the feed gained rows, which -keeps the retry bounded by writes rather than by frames and makes a failed -lookup cost one generation instead of the run. REST -reads (`GET /threads/{id}/state`, `POST /threads/{id}/history`) stamp through -`events/message_seq.py::stamp_messages_with_seq`, the request-scoped -counterpart: everything a checkpoint still holds is already persisted, so one -batched lookup resolves the whole list. The db store prefilters candidate rows -in SQL (a LIKE clause per wanted raw id, wildcards escaped; an id `json.dumps` -would escape falls the set back to the full scan) so a wanted identity absent -from the feed — a message still streaming — does not force a full -fetch-and-decode of every message row's tool outputs on long threads. -`deerflow_seq` is server-owned display metadata: the gateway strips it from -client input, because a welded-in seq goes stale when a fork re-seeds the feed -(#4380). +**Message sequence placement:** Keep backend and frontend message identity rules aligned. Details: `backend/docs/runtime-guidance-details.md`. **LLM response callback coalescing** (`runtime/journal.py`): a provider may fire `on_llm_end` twice for one LangChain run id, first without usage (or with all token @@ -80,49 +45,10 @@ authoritative run token summary, but it cannot mutate the append-only message ev caller attribution, fallback state, or tool-call bookkeeping. Closed journals return from `on_llm_end` before inspecting the response or touching any run state. -**Run delivery receipts** (`runtime/journal.py` + `runs/worker.py`): -`RunJournal` records each non-empty artifact update once per tool `Command` for -the terminal `run.delivery` event. When a command contains multiple messages, a -unique tool name resolved from matching `ToolMessage` entries supplies -attribution; additional command messages do not duplicate artifact paths or -counts. If multiple different tool names resolve for one flat artifact update, -the paths remain counted but unattributed because the command does not carry a -per-path mapping. `RunJournal` callbacks set `run_inline=True`: they do only -in-memory bookkeeping or schedule async writes, and staying on the run's -event-loop thread serializes parallel tool callbacks before terminal delivery -recording and flushing. Each worker creates a separate journal per run before -cancellable/fallible preflight work, so checkpoint compatibility failures and -cancellation while waiting for prior finalization still emit a zero-delivery -receipt. The worker flushes ordinary journal events, idempotently persists the -run-scoped receipt, and only then persists the staged terminal run status. A -receipt failure is retried on a short bounded schedule while the owning worker -still knows the real outcome and holds the lease. Delivery candidates are every -regular file created or modified under `/mnt/user-data/outputs`; internal -process-feedback files are excluded (the scanner's `EXCLUDED_DIR_NAMES` plus -the configured `tool_output.storage_subdir`), so a run that only externalized -oversized tool outputs does not fail delivery. At least one candidate must be -covered by a path attributed by the journal to `present_files`; presenting only -an unrelated pre-existing path does not satisfy delivery. Receipts for such -runs add `produced_paths`, `presented_paths`, `matched_paths`, `verification`, -`stage`, and `satisfied` to the Slice 1 fact fields. Missing a matching -presentation becomes a run error; a successful presentation is also downgraded -to error if its receipt cannot be durably verified. Runs without changed -outputs preserve ordinary chat behavior and the original receipt shape. Orphan -recovery first atomically claims an expired lease, then uses the same singleton -write to backfill a zero-delivery receipt — a stale recovery scan cannot -overwrite a live run's later detailed receipt, an event-store outage does not -undo the terminal takeover, and an existing detailed receipt is preserved when -a worker crashed after writing it. Event stores serialize `put_if_absent` with -ordinary thread writers: memory and JSONL provide the documented -single-process guarantee, while the DB store adds per-thread in-process locks -and PostgreSQL advisory locks for cross-process writers. Moving journal -construction ahead of preflight is receipt-only on early failure paths: a -separate boundary flag preserves the previous completion-data semantics, so -checkpoint incompatibility or cancellation while waiting for an older -finalizing run does not persist an empty completion snapshot. Worker tests pin -one accumulated receipt across multiple goal-continuation `_stream_once` calls; -journal tests drive LangChain's real async callback dispatcher against a single -journal to pin serialized, deduplicated parallel tool callbacks. +**Skill history:** `record_skill_usage` saves lead-run snapshots on terminal +answers for paginated history. See `docs/skill-usage-ui.md`. + +**Run delivery receipts:** Journal artifact evidence and terminal status must finalize in order. Details: `backend/docs/runtime-guidance-details.md`. **Deferred-tool promotion event deduplication** (`runtime/journal.py`): one `RunJournal` owns the lead graph's run-scoped atomic promotion claim. Parallel @@ -189,50 +115,9 @@ checkpoint-write admission boundary must repeat the complete audit after admission; a pre-admission exact hit can be superseded by a later event just as a pre-admission miss can become an exact hit. -**Extension changed-run discovery** (`runtime/runs/store/` and -`extensions/run_evidence.py`) orders public run-record changes by the -backend-owned `(change_seq, run_id)` key rather than timestamps or per-thread -event sequence numbers. The singleton SQL clock allocates positions in the same -transaction as each public record mutation; memory uses a process-local counter. -The clock row serializes position-bearing SQL mutations globally until their -transactions commit, so every mutator must acquire it before any run-row lock. -An atomic thread operation uses one position for its interrupted rows and new -row, with `run_id` ordering ties. High-frequency progress snapshots and lease -heartbeats do not advance this position: they are not lifecycle-discovery -signals, progress evidence remains available from the event stream, and -excluding them bounds clock contention. A later lifecycle change exposes the -run row's latest `updated_at` and accumulated progress fields to internal readers. -Rows from before the migration retain `change_seq=0` and page deterministically -by run id. Because a later mutation only moves a row forward, concurrent paging -may replay a run but cannot move an unseen run behind the committed cursor. -Deletes produce no tombstone and therefore do not advance the cursor; consumers -that require deletion reconciliation must poll authoritative status for known -runs and treat a missing result as absent. -The extension-facing cursor is versioned, opaque, and bound to the reader's -fixed user scope. `RunEventStore.list_events()` accepts the same explicit -`user_id` override as `list_messages()`; evidence readers must propagate their -bound scope (including global `None`) rather than resolving an ambient request -user. The adapter deep-copies event content and redacted metadata before -exposing them, detaching nested mutable payloads from host storage. +**Changed-run discovery:** Use the durable `(change_seq, run_id)` cursor and repeat history audits after admission. Details: `backend/docs/runtime-guidance-details.md`. -Gateway `POST /api/threads/{id}/history` uses that lookup to migrate legacy AI -messages. An exhaustive miss preserves the human-boundary fallback; an -incomplete lookup removes unproven synthesized IDs. Its metadata-only -write-on-read cache stores `run_message_ids` for every audited AI ID (including -exhaustive misses) plus required `run_durations`; duration presence alone does -not prove attribution. Historical `body.before` reads write the audit to the -head, and the merge may retain IDs no longer in materialized history, which -readers ignore. Migration must acquire the durable `checkpoint_write` -reservation, then repeat the whole message audit and batch-reload required run -rows before persisting. Post-admission exact hits replace foreground exact or -boundary mappings, and recomputed final durations replace foreground snapshots. -Successful workers keep their durable run row active through the final duration -checkpoint write, so a peer migration cannot enter during terminalization. -The first `RunManager.list_by_thread()` hydration page uses a 100-row floor or -the number of required IDs, whichever is larger; missing exact runs use targeted -`get()` calls. - -**Terminal run cleanup explicitly breaks graph-scoped references while preserving the existing `RunRecord` grace period.** Every `agent.astream()` iterator is closed in `_stream_once`, including abort/exception/early-break paths. A close failure after an abort is warning-only and cannot replace the user-requested `interrupted` outcome; normal-completion close failures still surface, and an in-flight stream exception remains authoritative over a secondary close failure. Journal construction and cancellable preflight work (including MCP task projection and the prior-finalization wait) live inside the worker's guarded body, so cancellation before agent startup still terminalizes the run and closes its stream. `run_agent()` wraps the complete terminal-finalization sequence in an outer teardown guard, so cancellation or failure from any terminal-stage await cannot skip `RunJournal.close()`, removal of the journal, `__pregel_runtime`, and internal runtime-context values from every runnable config, or release of local graph/payload references. That guard schedules bridge cleanup, run-record cleanup, and cyclic GC even when interruption happens before the terminal stream marker or terminal publication itself fails, so neither a cancelled observer nor a delivery-backend outage can strand process-local run state. A non-`Exception` `BaseException` caught while awaiting the completion hook or task-stop notification (including host-task cancellation) is deferred through the ordinary remaining finalization, with the first interruption preserved and every caught host-task `CancelledError` balanced by calling `Task.uncancel()` until the current task’s cumulative cancellation count is clear. Task-stop fan-out runs in one child task and every host wait uses `shield`, so repeated cancellation of the worker cannot cancel that fan-out or skip later observers; the worker keeps awaiting the same child task. A rogue observer that raises its own `CancelledError` remains contained by the extension dispatcher and distinguishable from host cancellation. This guarantee applies only to cancellation caught during those hook stages: clearing the finalizing barrier and publishing END remain direct awaits, so another cancellation in the subsequent critical tail retains forceful-termination semantics instead of creating an unbounded shield. If that tail completes without another interruption, the first deferred interruption is re-raised after END; a barrier-clear failure prevents END publication, while an END failure is raised after the barrier is clear. `RunJournal.flush()` clears its `_pending_progress_task` after awaiting or cancelling it; ordinary `close()` detaches the event store/progress reporter and clears callback bookkeeping only after that flush succeeds, preserving the buffer for retry on a transient store failure. A fenced worker instead calls `close(flush=False)`, which cancels pending journal work and detaches without initiating another event-store write after lease ownership is lost; its final detach runs even if a second cancellation interrupts pending-task shutdown. `RunManager.cleanup(run_id)` retains the process-local `RunRecord`, completed task, and request payload for its default 300-second local join/status window before releasing them, and evicts only when a `RunStore` backs the manager — without one there is no hydration fallback, so the record is retained rather than silently dropping the run's history. Durable history remains in `RunStore`; `StreamBridge` data keeps its separate 60-second late-subscriber window, and both cleanup coroutines run in a fresh empty `contextvars.Context`. A contextless full cyclic-GC pass, coalesced to at most once every 10 seconds and dispatched through the default executor, bounds the lifetime of unreachable LangGraph callback/loop cycles without synchronously walking the heap in the event-loop timer; passes taking at least 100 ms are logged at INFO because CPython GC may still impose interpreter-level pauses. +**Terminal run cleanup:** Close streams, journals, and graph references even on cancellation. Details: `backend/docs/runtime-guidance-details.md`. **`RunManager._runs` holds only records this worker admitted.** A cross-worker idempotent reuse returns the `store_only` row from `_record_from_store()` unregistered: the peer never finalizes or `cleanup()`s it, so a registered copy stays `pending`/`running`, 409s later same-thread admissions, hides the owner's orphan from reconciliation, and sends a peer `cancel()` down the local-owner path. Pinned by `test_peer_idempotent_reuse_*` and `test_peer_cancel_of_reused_run_*` (`tests/test_multi_worker_run_ownership.py`) plus `tests/test_gateway_services.py::test_start_run_peer_idempotent_reuse_*`. @@ -247,98 +132,8 @@ the number of required IDs, whichever is larger; missing exact runs use targeted - `runtime/checkpoint_cache/` + `runtime/checkpointer/cached_saver.py` — delta-mode checkpoint history cache; checkpoint state reads MUST go through `CheckpointStateAccessor`, and the checkpointer may be a `CachedHistorySaver` wrapper — never rely on concrete saver types - Tests: `tests/test_checkpoint_mode.py` (freeze/detect/gate), `tests/test_checkpoint_state.py` (accessor/mutation graph), `tests/test_delta_channel_checkpointers.py` (saver parity), `tests/test_threads_checkpoint_mode.py`, `tests/test_gateway_checkpoint_mode.py` (dual-mode e2e parity), `tests/test_context_compaction.py` (mutation-graph write, no scheduling), `tests/test_run_worker_rollback.py`, `tests/test_cached_history_saver.py` + `tests/test_cached_history_saver_integration.py` (history cache) -**Checkpoint channel benchmark**: `scripts/benchmark/checkpoint/bench_channels.py` -runs paired `full`/`delta` message-only StateGraphs in a fresh child process per -case, using sync `InMemorySaver` or `SqliteSaver` so reducer, serialization, and -saver costs stay separate from Gateway/async scheduling. Optional -`AsyncPostgresSaver` cases are enabled only when `TEST_POSTGRES_URI` is set. -Postgres cases use a unique thread and remove only that benchmark thread through -the saver's public `adelete_thread` API after measurement. It reports deterministic -correctness digests, write windows/percentiles, warm and graph-rebuilt cold reads, -backend-neutral checkpoint/blob/write row and byte fields, aggregate logical -checkpoint/write bytes, SQLite DB/WAL/SHM footprint, reducer replay time, and -peak RSS as versioned JSONL. SQLite embeds channel blobs in its checkpoint -payload, so its separate blob metrics are zero; Postgres reports its -`checkpoint_blobs` table separately. Byte fields describe each saver's serialized -representation and should not be treated as identical encodings across backends. -The controller alternates mode order and rejects -performance data when paired modes materialize different state. Its default 1 GiB -estimated cumulative full-payload cap skips both modes of an oversized pair when -`full` is selected, including every delta cadence in a `--snapshot-frequencies` -sweep; intentional `--modes delta` diagnostics bypass this full-payload cap, so -size those runs explicitly. Use `--allow-large-cases` only -on a provisioned machine. Duplicate CSV matrix values are ignored with a warning; -use `--repetitions` for repeated samples. Summarize paired successful repetitions -with `scripts/benchmark/checkpoint/summarize_channels.py` (all ratios are -`delta/full`). `--profile-dir /tmp/checkpoint-profiles` writes one cProfile -artifact per case for attribution. Profiled rows carry `profiled: true`, and the -summarizer automatically excludes them from baseline summaries with a warning. -Storage-size collection relies on saver-specific diagnostic layouts; if those -layouts change, the timing/correctness row remains successful while storage -fields become `null` and `storage_stats_error` records the diagnostic failure. -Example: +**Checkpoint benchmarks:** Paired full/delta cases and production-shaped runs are documented in `backend/docs/runtime-guidance-details.md`. -```bash -cd backend -PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ - --backends sqlite --updates 100,500,999,1000,1001 --payload-bytes 128 \ - --repetitions 7 --output /tmp/checkpoint-bench.jsonl -TEST_POSTGRES_URI=postgresql://... \ -PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_channels.py \ - --backends sqlite,postgres --updates 100 --payload-bytes 128 \ - --output /tmp/checkpoint-cross-backend.jsonl -PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_channels.py \ - /tmp/checkpoint-bench.jsonl -``` - -The production-shaped layer lives in -`scripts/benchmark/checkpoint/bench_production.py`: per-case child processes -run graph-level `ainvoke` turns through the real lead-agent graph (scripted -deterministic model, real `AsyncSqliteSaver`), then measure -`GET /threads/{id}/state` and `POST /threads/{id}/history` through the real -Gateway route stack in the same event loop (httpx ASGITransport), split into -cold/warm accessor-graph-cache samples. It sweeps `snapshot_frequency` -(config: `checkpoint_delta.snapshot_frequency`, process-frozen like the mode), -pairs every delta frequency against the same full row, and fails both -rows of a pair when materialized or wire digests diverge. Each case must have -more than the two discarded warm-up turns, and SQLite DB/WAL/SHM sizes are -captured while the saver is still open so they represent the online storage -footprint. Summarize with -`scripts/benchmark/checkpoint/summarize_production.py` (ratios are -`delta/full`; it also emits `snapshot_write_spike` and `cache_effect_ms`, -the decision inputs for the production snapshot-frequency and accessor-cache -defaults). Harness tests live in `tests/test_bench_checkpoint_production.py` -and `tests/test_summarize_checkpoint_production.py`; timing thresholds are -not CI gates. The matrix test pins that every `(repetition, turns)` group -contains both modes and that their execution order flips between consecutive -groups, including across repetition boundaries. - -Operational limits learned from the first runs (the default matrix is too -large to run blindly): - -- The default `--timeout-seconds 900` is insufficient for delta mode at - `snapshot_frequency=1000` once turns reach 500 (measured: delta-500 takes - ~1100-1200s; delta-2000 takes ~45min). Pass an explicit - `--timeout-seconds` for any large matrix, and treat the turns=2000 corner - as practical only at small snapshot frequencies. -- Full-mode 2000-turn runs produce a ~33GB sqlite DB. Point `TMPDIR` at real - disk, not tmpfs (the benchmark uses `tempfile.TemporaryDirectory`, which - honors `TMPDIR`), or the run dies mid-case. -- The history route clamps `limit` to 100 (`le=100` on - `ThreadHistoryRequest.limit`), so `--history-limits` values above 100 are - measured and reported by their effective (clamped) limit. - -Example: - -```bash -cd backend -PYTHONPATH=. uv run python scripts/benchmark/checkpoint/bench_production.py \ - --turns 10,100,500,1000,2000 --payload-bytes 128 \ - --snapshot-frequencies 10,50,100,500,1000 \ - --repetitions 7 --output /tmp/production-bench.jsonl -PYTHONPATH=. uv run python scripts/benchmark/checkpoint/summarize_production.py \ - /tmp/production-bench.jsonl -``` # Referenced conversation capability `RunContext.conversation_reader` is a host-provided per-run callback. The worker @@ -361,4 +156,4 @@ Drain tasks are named `jsonl-mutation:{thread_id}` for asyncio task dumps. Multi the admitted group keeps its records on success or completes rollback on failure. This is a store-local guarantee, not a change to RunJournal cancellation policy or JSONL's single-process deployment constraint. Regression coverage is in -`tests/test_jsonl_event_store_cancellation.py`. \ No newline at end of file +`tests/test_jsonl_event_store_cancellation.py`. diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index 95577a4759..d88f03a966 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -34,6 +34,7 @@ from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMes from langgraph.types import Command from deerflow.agents.human_input import read_human_input_response +from deerflow.agents.middlewares.skill_usage import MAX_SKILL_SNAPSHOT_CHARS, SKILL_USAGES_KEY from deerflow.runtime.events.catalog import ( LLM_AI_RESPONSE_EVENT, LLM_ERROR_EVENT, @@ -54,6 +55,7 @@ logger = logging.getLogger(__name__) _LEGACY_SUMMARY_MESSAGE_NAME = "summary" _PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES = frozenset({"ask_clarification", "sandbox_network"}) +_MAX_RUN_SKILL_SNAPSHOTS = 64 @dataclass @@ -288,6 +290,10 @@ class RunJournal(BaseCallbackHandler): self._memory_context_recorded = False self._tool_promotion_claim_lock = threading.Lock() self._claimed_tool_promotions: set[str] = set() + # Only lead producers receive this journal in runtime.context. Keep + # display snapshots separate from tool/secret authorization state. + self._skill_usage_lock = threading.Lock() + self._skill_usages: dict[str, dict[str, Any]] = {} # Convenience fields self._last_ai_msg: str | None = None @@ -509,11 +515,18 @@ class RunJournal(BaseCallbackHandler): call_index = self._llm_call_index self._seen_llm_starts.add(rid) + content = message.model_dump() + if is_canonical_callback and caller == "lead_agent" and isinstance(message, AIMessage) and not message.tool_calls: + with self._skill_usage_lock: + skill_usages = deepcopy(list(self._skill_usages.values())) + if skill_usages: + content["additional_kwargs"] = {**content.get("additional_kwargs", {}), SKILL_USAGES_KEY: skill_usages} + response_events.append( self._make_event( event_type=LLM_AI_RESPONSE_EVENT.event_type, category=LLM_AI_RESPONSE_EVENT.category, - content=message.model_dump(), + content=content, metadata={ "caller": caller, "usage": usage_dict, @@ -620,6 +633,30 @@ class RunJournal(BaseCallbackHandler): # -- Internal methods -- + def record_skill_usage(self, usage: Mapping[str, Any]) -> None: + """Capture first-load display evidence from a successful lead producer. + + Tool wrappers run after ``on_tool_end`` serialized the raw response; + slash producers register before invoking their model handler. The next + terminal lead response carries these snapshots in the history feed so + pagination and checkpoint compaction do not lose the menu's evidence. + No event-store or other loop-bound work happens on this producer path. + """ + with self._skill_usage_lock: + if self._closed: + return + path = usage.get("path") + content = usage.get("content") + if not isinstance(path, str) or not path or not isinstance(content, str): + return + if path in self._skill_usages or len(self._skill_usages) >= _MAX_RUN_SKILL_SNAPSHOTS: + return + snapshot = deepcopy(dict(usage)) + if len(content) > MAX_SKILL_SNAPSHOT_CHARS: + snapshot["content"] = content[:MAX_SKILL_SNAPSHOT_CHARS] + snapshot["partial"] = True + self._skill_usages[path] = snapshot + @staticmethod def _message_identity(message: BaseMessage) -> str | None: tool_call_id = getattr(message, "tool_call_id", None) @@ -1156,6 +1193,8 @@ class RunJournal(BaseCallbackHandler): def _detach_runtime_dependencies(self) -> None: """Drop every external or potentially cyclic run-scoped reference.""" self._closed = True + with self._skill_usage_lock: + self._skill_usages.clear() self._store = None self._progress_reporter = None self._buffer.clear() diff --git a/backend/packages/harness/deerflow/sandbox/read_file_contract.py b/backend/packages/harness/deerflow/sandbox/read_file_contract.py new file mode 100644 index 0000000000..4c26ac1c01 --- /dev/null +++ b/backend/packages/harness/deerflow/sandbox/read_file_contract.py @@ -0,0 +1,17 @@ +"""Text markers shared by read_file and its display-only consumers.""" + +READ_FILE_EMPTY = "(empty)" +READ_FILE_START_LINE_EXCEEDS = "(start_line exceeds file length)" +READ_FILE_INVALID_START_LINE = "(start_line must be >= 1)" +READ_FILE_INVALID_END_LINE = "(end_line must be >= 1)" +READ_FILE_EMPTY_RANGE = "(start_line > end_line — no lines in range)" +READ_FILE_NO_CONTENT_RESULTS = frozenset( + { + READ_FILE_EMPTY, + READ_FILE_START_LINE_EXCEEDS, + READ_FILE_INVALID_START_LINE, + READ_FILE_INVALID_END_LINE, + READ_FILE_EMPTY_RANGE, + } +) +READ_FILE_TRUNCATION_PREFIX = "... [truncated:" diff --git a/backend/packages/harness/deerflow/sandbox/tools.py b/backend/packages/harness/deerflow/sandbox/tools.py index c94d6ff6e8..9f831da0ba 100644 --- a/backend/packages/harness/deerflow/sandbox/tools.py +++ b/backend/packages/harness/deerflow/sandbox/tools.py @@ -40,6 +40,14 @@ from deerflow.sandbox.lease import ( ) from deerflow.sandbox.overwrite import unwrap_sandbox from deerflow.sandbox.path_patterns import build_output_mask_pattern, normalize_mask_tail, replace_output_path_matches +from deerflow.sandbox.read_file_contract import ( + READ_FILE_EMPTY, + READ_FILE_EMPTY_RANGE, + READ_FILE_INVALID_END_LINE, + READ_FILE_INVALID_START_LINE, + READ_FILE_START_LINE_EXCEEDS, + READ_FILE_TRUNCATION_PREFIX, +) from deerflow.sandbox.sandbox import Sandbox from deerflow.sandbox.sandbox_provider import SandboxProvider, get_sandbox_provider from deerflow.sandbox.search import GrepMatch @@ -1909,9 +1917,9 @@ def _read_file_truncation_marker(*, line_offset: int, shown_lines: int, total_li span = f"of {total_lines}" if line_offset == 0 else f"of {first}-{line_offset + total_lines}" if inside_line is None: shown = f"first {shown_lines}" if line_offset == 0 else f"lines {first}-{line_offset + shown_lines}" - return f"... [truncated: showing {shown} {span} lines ({kept} of {total} chars). Continue with start_line={line_offset + shown_lines + 1}, or use start_line/end_line to read a specific range] ..." + return f"{READ_FILE_TRUNCATION_PREFIX} showing {shown} {span} lines ({kept} of {total} chars). Continue with start_line={line_offset + shown_lines + 1}, or use start_line/end_line to read a specific range] ..." line = line_offset + inside_line - head = f"\n... [truncated: showing first {kept} of {total} chars, cut inside line {line} {span} lines" + head = f"\n{READ_FILE_TRUNCATION_PREFIX} showing first {kept} of {total} chars, cut inside line {line} {span} lines" if continuation == "next": return f"{head}. Continue with start_line={line}, or use start_line/end_line to read a specific range] ..." if continuation == "whole_line": @@ -1945,7 +1953,7 @@ def _read_file_marker_reserve(*, line_offset: int, total_lines: int, total: int) # Emitted when the budget cannot even hold a marker: the model must still # learn the output was cut and how to read it in pieces. -_READ_FILE_TINY_BUDGET_MARKER = "... [truncated: {total} chars exceed the {max_chars}-char read limit; use start_line/end_line to read a smaller range] ..." +_READ_FILE_TINY_BUDGET_MARKER = READ_FILE_TRUNCATION_PREFIX + " {total} chars exceed the {max_chars}-char read limit; use start_line/end_line to read a smaller range] ..." def _truncate_read_file_output(output: str, max_chars: int, *, line_offset: int = 0, joined_lines: bool = False, ends_at_eof: bool = True) -> str: @@ -2574,12 +2582,12 @@ def read_file_tool( skill_name = _extract_skill_name_from_skills_path(path) or "unknown" return f"Error: Skill '{skill_name}' is disabled. Access to its files is blocked. Enable the skill in settings before using it." if start_line is not None and start_line < 1: - return "(start_line must be >= 1)" + return READ_FILE_INVALID_START_LINE effective_start = start_line or 1 if end_line is not None and end_line < 1: - return "(end_line must be >= 1)" + return READ_FILE_INVALID_END_LINE if end_line is not None and effective_start > end_line: - return "(start_line > end_line — no lines in range)" + return READ_FILE_EMPTY_RANGE requested_path = path use_line_range = start_line is not None or end_line is not None @@ -2593,9 +2601,9 @@ def read_file_tool( # tell them apart so a continuation named by a truncation # marker is not reported as beyond the file. if _ranged_read_hits_a_line(runtime, path, start_line): - return "(empty)" - return "(start_line exceeds file length)" - return "(empty)" + return READ_FILE_EMPTY + return READ_FILE_START_LINE_EXCEEDS + return READ_FILE_EMPTY try: from deerflow.config.app_config import get_app_config diff --git a/backend/tests/test_run_journal_skill_usage.py b/backend/tests/test_run_journal_skill_usage.py new file mode 100644 index 0000000000..774e535f0a --- /dev/null +++ b/backend/tests/test_run_journal_skill_usage.py @@ -0,0 +1,111 @@ +"""Skill snapshots survive callback ordering, pagination, and checkpoint removal.""" + +import hashlib +from uuid import uuid4 + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from langchain_core.outputs import ChatGeneration, LLMResult + +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import RunJournal + + +def snapshot(name="report", *, content="# Original instructions", activation="automatic"): + return { + "name": name, + "description": "Create reports", + "category": "custom", + "path": f"/mnt/skills/custom/{name}/SKILL.md", + "content": content, + "content_hash": hashlib.sha256(content.encode()).hexdigest(), + "activation": activation, + "partial": False, + } + + +def respond(journal, message, *, run_id=None, tags=None): + journal.on_llm_end(LLMResult(generations=[[ChatGeneration(message=message)]]), run_id=run_id or uuid4(), tags=tags) + + +@pytest.mark.anyio +async def test_terminal_history_contains_loads_recorded_after_tool_callback_and_before_slash_response(): + store = MemoryRunEventStore() + journal = RunJournal("run", "thread", store) + slash = snapshot("slash", activation="slash") + journal.record_skill_usage(slash) + respond(journal, AIMessage(content="", tool_calls=[{"id": "read", "name": "read_file", "args": {}}])) + # Tool callbacks happen inside the handler, before the wrapper knows the + # read succeeded. The raw tool event therefore cannot contain this metadata. + journal.on_tool_end(ToolMessage(content="# Original instructions", tool_call_id="read"), run_id=uuid4()) + automatic = snapshot() + journal.record_skill_usage(automatic) + # Repeat loads retain the first snapshot and its original position. + journal.record_skill_usage(snapshot(content="# Edited later")) + automatic["content"] = "mutated by producer" + terminal = AIMessage(id="final", content="Done") + respond(journal, terminal) + await journal.close() + + # This uses only the durable feed, without any retained checkpoint, and + # requests a page too small to include the original load events. + page = await store.list_messages("thread", limit=1) + assert len(page) == 1 + assert page[0]["content"]["id"] == "final" + assert page[0]["content"]["additional_kwargs"]["skill_usages"] == [slash, snapshot()] + assert "skill_usages" not in terminal.additional_kwargs + all_rows = await store.list_messages("thread") + assert "skill_usages" not in all_rows[0]["content"]["additional_kwargs"] + assert "skill_usage" not in all_rows[1]["content"]["additional_kwargs"] + + +@pytest.mark.anyio +async def test_only_canonical_lead_terminal_responses_receive_display_snapshots(): + store = MemoryRunEventStore() + journal = RunJournal("run", "thread", store) + journal.record_skill_usage(snapshot()) + respond(journal, AIMessage(id="subagent", content="Child answer"), tags=["subagent:research"]) + respond(journal, AIMessage(id="summary", content="Summary"), tags=["middleware:summarization"]) + respond(journal, AIMessage(id="tool-call", content="Reading", tool_calls=[{"id": "read", "name": "read_file", "args": {}}])) + callback_id = uuid4() + respond(journal, AIMessage(id="final", content="Done"), run_id=callback_id) + journal.record_skill_usage(snapshot("later")) + respond(journal, AIMessage(id="replay", content="Replay", usage_metadata={"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}), run_id=callback_id) + respond(journal, AIMessage(id="next", content="Next answer")) + await journal.close() + + rows = await store.list_events("thread", "run", event_types=["llm.ai.response"]) + by_id = {row["content"]["id"]: row["content"]["additional_kwargs"] for row in rows} + assert by_id["subagent"] == {} + assert by_id["summary"] == {} + assert by_id["tool-call"] == {} + assert by_id["final"]["skill_usages"] == [snapshot()] + assert by_id["next"]["skill_usages"] == [snapshot(), snapshot("later")] + assert "replay" not in by_id + + +@pytest.mark.anyio +async def test_skill_aggregation_is_bounded_and_close_drops_snapshots(): + store = MemoryRunEventStore() + journal = RunJournal("run", "thread", store) + for index in range(70): + journal.record_skill_usage(snapshot(str(index), content="x" * 100_001)) + respond(journal, AIMessage(content="Done")) + await journal.close() + rows = await store.list_messages("thread", limit=1) + usages = rows[0]["content"]["additional_kwargs"]["skill_usages"] + assert len(usages) == 64 + assert all(len(usage["content"]) == 100_000 and usage["partial"] for usage in usages) + assert journal._skill_usages == {} + journal.record_skill_usage(snapshot("after-close")) + assert journal._skill_usages == {} + + +@pytest.mark.anyio +async def test_empty_journal_keeps_history_unchanged(): + store = MemoryRunEventStore() + journal = RunJournal("run", "thread", store) + respond(journal, AIMessage(content="No skills")) + await journal.close() + rows = await store.list_messages("thread") + assert "skill_usages" not in rows[0]["content"]["additional_kwargs"] diff --git a/backend/tests/test_skill_usage.py b/backend/tests/test_skill_usage.py new file mode 100644 index 0000000000..adab7d54ef --- /dev/null +++ b/backend/tests/test_skill_usage.py @@ -0,0 +1,230 @@ +"""Display snapshots record successful loads without depending on live skill files.""" + +import asyncio +import hashlib +from types import SimpleNamespace + +import pytest +from langchain_core.messages import AIMessage, ToolMessage +from langgraph.types import Command + +from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware +from deerflow.agents.middlewares.tool_output_budget_middleware import ToolOutputBudgetMiddleware +from deerflow.config.app_config import AppConfig +from deerflow.config.sandbox_config import SandboxConfig +from deerflow.config.tool_output_config import ToolOutputConfig +from deerflow.sandbox.read_file_contract import READ_FILE_NO_CONTENT_RESULTS + + +def read_result(content, *, path="/mnt/skills/custom/report/SKILL.md", status="success", args=None, asynchronous=False): + request = SimpleNamespace(tool_call={"name": "read_file", "id": "read-1", "args": {"path": path, **(args or {})}}) + message = ToolMessage(content=content, tool_call_id="read-1", status=status) + middleware = ToolErrorHandlingMiddleware() + if asynchronous: + + async def handler(_request): + return message + + return asyncio.run(middleware.awrap_tool_call(request, handler)) + return middleware.wrap_tool_call(request, lambda _: message) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_successful_read_captures_loaded_snapshot(asynchronous): + content = "---\nname: quarterly-report\ndescription: Summarize results.\n---\n# Report\nUse source data." + message = read_result(content, asynchronous=asynchronous) + assert message.additional_kwargs["skill_usage"] == { + "name": "quarterly-report", + "description": "Summarize results.", + "category": "custom", + "path": "/mnt/skills/custom/report/SKILL.md", + "content": content, + "content_hash": hashlib.sha256(content.encode()).hexdigest(), + "activation": "automatic", + "partial": False, + } + + +@pytest.mark.parametrize( + "content,status,path", + [ + ("Error: File not found", "success", "/mnt/skills/custom/report/SKILL.md"), + ("denied", "error", "/mnt/skills/custom/report/SKILL.md"), + ("(start_line exceeds file length)", "success", "/mnt/skills/custom/report/SKILL.md"), + ("(empty)", "success", "/mnt/skills/custom/report/SKILL.md"), + ("body", "success", "/mnt/user-data/SKILL.md"), + ("body", "success", "/mnt/skills/../../private/SKILL.md"), + ("body", "success", "/mnt/skills/custom/report/scripts/run.py"), + ], +) +def test_failed_or_unrelated_reads_are_not_usage(content, status, path): + assert "skill_usage" not in read_result(content, status=status, path=path).additional_kwargs + + +@pytest.mark.parametrize("content", sorted(READ_FILE_NO_CONTENT_RESULTS)) +def test_read_file_no_content_markers_are_not_skill_usage(content): + assert "skill_usage" not in read_result(content).additional_kwargs + + +def test_range_and_size_limited_snapshots_are_truthfully_marked_partial(): + assert read_result("A section", args={"start_line": 4}).additional_kwargs["skill_usage"]["partial"] + content = "A" * 110_000 + snapshot = read_result(content).additional_kwargs["skill_usage"] + assert snapshot["partial"] + assert len(snapshot["content"]) <= 100_000 + assert snapshot["content_hash"] == hashlib.sha256(content.encode()).hexdigest() + + +def test_tiny_read_budget_marker_is_partial(): + content = "... [truncated: 123 chars exceed the 80-char read limit; use start_line/end_line to read a smaller range] ..." + assert read_result(content).additional_kwargs["skill_usage"]["partial"] + + +def test_external_messages_cannot_forge_skill_usage(): + from app.gateway.services import _strip_external_message_metadata, _strip_external_metadata_from_message_like + + message = AIMessage(content="hello", additional_kwargs={"skill_usage": {"name": "forged"}, "skill_usages": [{"name": "forged"}]}) + assert "skill_usage" not in _strip_external_message_metadata(message).additional_kwargs + assert "skill_usages" not in _strip_external_message_metadata(message).additional_kwargs + raw = {"type": "ai", "content": "hello", "additional_kwargs": {"skill_usage": {"name": "forged"}}} + assert "skill_usage" not in _strip_external_metadata_from_message_like(raw)["additional_kwargs"] + + +def test_successful_read_registers_snapshot_before_next_model_callback(): + recorded = [] + request = SimpleNamespace( + tool_call={"name": "read_file", "id": "read-1", "args": {"path": "/mnt/skills/custom/report/SKILL.md"}}, + runtime=SimpleNamespace(context={"__run_journal": SimpleNamespace(record_skill_usage=recorded.append)}), + ) + read = ToolErrorHandlingMiddleware() + result = ToolOutputBudgetMiddleware().wrap_tool_call(request, lambda inner_request: read.wrap_tool_call(inner_request, lambda _: ToolMessage(content="# Instructions", tool_call_id="read-1"))) + assert recorded == [result.additional_kwargs["skill_usage"]] + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("externalize", [False, True]) +@pytest.mark.parametrize("tool_name", ["read_file", "custom_read"]) +def test_budgeted_skill_read_records_only_visible_snapshot(asynchronous, externalize, tool_name, tmp_path): + recorded = [] + request = SimpleNamespace( + tool_call={"name": tool_name, "id": "read-1", "args": {"path": "/mnt/skills/custom/report/SKILL.md"}}, + runtime=SimpleNamespace( + context={"__run_journal": SimpleNamespace(record_skill_usage=recorded.append)}, + state={"thread_data": {"outputs_path": str(tmp_path)}} if externalize else {}, + ), + ) + raw = "# Instructions\n" + "Use the source data.\n" * 100 + message = ToolMessage(content=raw, tool_call_id="read-1", name=tool_name) + app_config = AppConfig(sandbox=SandboxConfig(use="test")) + app_config.summarization.skill_file_read_tool_names = [tool_name] + read = ToolErrorHandlingMiddleware(app_config=app_config) + app_config.tool_output = ToolOutputConfig( + exempt_tools=[], + externalize_min_chars=100 if externalize else 0, + fallback_max_chars=100, + fallback_head_chars=40, + fallback_tail_chars=20, + ) + budget = ToolOutputBudgetMiddleware.from_app_config(app_config) + if asynchronous: + + async def inner(_request): + return message + + async def wrapped(inner_request): + return await read.awrap_tool_call(inner_request, inner) + + result = asyncio.run(budget.awrap_tool_call(request, wrapped)) + else: + result = budget.wrap_tool_call(request, lambda inner_request: read.wrap_tool_call(inner_request, lambda _: message)) + assert result.content != raw + usage = result.additional_kwargs["skill_usage"] + assert usage["content"] == result.content + assert usage["content_hash"] == hashlib.sha256(result.content.encode()).hexdigest() + assert usage["partial"] is True + assert recorded == [usage] + + +@pytest.mark.parametrize("tool_name,path", [("bash", "/mnt/skills/custom/report/SKILL.md"), ("read_file", "/mnt/user-data/report/SKILL.md")]) +@pytest.mark.parametrize("wrapped_in_command", [False, True]) +def test_tool_supplied_skill_usage_cannot_claim_a_skill_read(tool_name, path, wrapped_in_command): + recorded = [] + request = SimpleNamespace( + tool_call={"name": tool_name, "id": "read-1", "args": {"path": path}}, + runtime=SimpleNamespace(context={"__run_journal": SimpleNamespace(record_skill_usage=recorded.append)}), + ) + forged = read_result("# Instructions").additional_kwargs["skill_usage"] + message = ToolMessage( + content="# unrelated output", + tool_call_id="read-1", + name=tool_name, + additional_kwargs={"skill_usage": forged, "skill_context_entry": {"path": forged["path"], "description": "forged"}}, + ) + read = ToolErrorHandlingMiddleware() + response = Command(update={"messages": [message]}) if wrapped_in_command else message + result = ToolOutputBudgetMiddleware().wrap_tool_call(request, lambda inner_request: read.wrap_tool_call(inner_request, lambda _: response)) + output = result.update["messages"][0] if wrapped_in_command else result + assert "skill_usage" not in output.additional_kwargs + assert "skill_context_entry" not in output.additional_kwargs + assert recorded == [] + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("messages_shape", ["list", "tuple", "single"]) +def test_skill_read_inside_command_records_post_budget_snapshot(asynchronous, messages_shape): + recorded = [] + request = SimpleNamespace( + tool_call={"name": "read_file", "id": "read-1", "args": {"path": "/mnt/skills/custom/report/SKILL.md"}}, + runtime=SimpleNamespace(context={"__run_journal": SimpleNamespace(record_skill_usage=recorded.append)}, state={}), + ) + raw = "# Instructions\n" + "Use source data.\n" * 100 + unrelated = ToolMessage(content="# Not this call", tool_call_id="other", name="read_file") + loaded = ToolMessage(content=raw, tool_call_id="read-1", name="read_file") + messages = [unrelated, loaded] if messages_shape != "single" else [loaded] + response = Command(update={"messages": messages[0] if messages_shape == "single" else tuple(messages) if messages_shape == "tuple" else messages}) + read = ToolErrorHandlingMiddleware() + budget = ToolOutputBudgetMiddleware(ToolOutputConfig(exempt_tools=[], externalize_min_chars=0, fallback_max_chars=100)) + if asynchronous: + + async def inner(_request): + return response + + async def wrapped(inner_request): + return await read.awrap_tool_call(inner_request, inner) + + result = asyncio.run(budget.awrap_tool_call(request, wrapped)) + else: + result = budget.wrap_tool_call(request, lambda inner_request: read.wrap_tool_call(inner_request, lambda _: response)) + updated = result.update["messages"] + updated_messages = [updated] if isinstance(updated, ToolMessage) else updated + if messages_shape != "single": + assert "skill_usage" not in updated_messages[0].additional_kwargs + loaded_result = updated_messages[-1] + assert loaded_result.content != raw + usage = loaded_result.additional_kwargs["skill_usage"] + assert usage["content"] == loaded_result.content + assert usage["partial"] is True + assert recorded == [usage] + + +@pytest.mark.parametrize("asynchronous", [False, True]) +@pytest.mark.parametrize("content,kwargs", [('{"error":"permission denied"}', {}), ("denied", {"deerflow_tool_meta": {"status": "error"}})]) +def test_structured_read_failures_never_register_usage(asynchronous, content, kwargs): + recorded = [] + request = SimpleNamespace( + tool_call={"name": "read", "id": "read-1", "args": {"path": "/mnt/skills/custom/report/SKILL.md"}}, + runtime=SimpleNamespace(context={"__run_journal": SimpleNamespace(record_skill_usage=recorded.append)}), + ) + message = ToolMessage(content=content, tool_call_id="read-1", additional_kwargs=kwargs) + middleware = ToolErrorHandlingMiddleware() + if asynchronous: + + async def handler(_): + return message + + result = asyncio.run(middleware.awrap_tool_call(request, handler)) + else: + result = middleware.wrap_tool_call(request, lambda _: message) + assert result.additional_kwargs["deerflow_tool_meta"]["status"] == "error" + assert "skill_usage" not in result.additional_kwargs + assert recorded == [] diff --git a/backend/tests/test_slash_skills.py b/backend/tests/test_slash_skills.py index ea41869e4e..1ce45f0a87 100644 --- a/backend/tests/test_slash_skills.py +++ b/backend/tests/test_slash_skills.py @@ -852,3 +852,38 @@ def test_skill_activation_middleware_reports_invalid_utf8_skill_file_safely(monk assert isinstance(result, AIMessage) assert "could not be loaded safely" in result.content + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_slash_usage_snapshot_is_persisted_on_first_model_response_only(monkeypatch, tmp_path, asynchronous): + from langchain.agents.middleware.types import ModelResponse + + content = "---\nname: data-analysis\ndescription: Analyze data.\n---\n# Original instructions" + skill = _make_skill(tmp_path, "data-analysis", content=content) + monkeypatch.setattr(middleware_module, "get_or_new_skill_storage", lambda **kwargs: _make_storage(tmp_path, [skill])) + middleware = SkillActivationMiddleware(slash_source_owner_token=_SLASH_SOURCE_OWNER_TOKEN) + recorded = [] + runtime = SimpleNamespace(context={"__run_journal": SimpleNamespace(record_skill_usage=recorded.append)}) + request = _make_model_request([HumanMessage(content="/data-analysis analyze", id="user-1")], runtime=runtime) + + def call(): + response = ModelResponse(result=[AIMessage(content="Analyzing")]) + if asynchronous: + + async def handler(_request): + return response + + return asyncio.run(middleware.awrap_model_call(request, handler)) + return middleware.wrap_model_call(request, lambda _: response) + + first = call() + snapshot = first.result[0].additional_kwargs["skill_usage"] + assert recorded == [snapshot] + assert snapshot["activation"] == "slash" + assert snapshot["name"] == "data-analysis" + assert snapshot["category"] == "custom" + assert snapshot["content"] == content + assert snapshot["path"] == "/mnt/skills/custom/data-analysis/SKILL.md" + skill.skill_file.write_text("Changed later", encoding="utf-8") + assert snapshot["content"] == content + assert "skill_usage" not in call().result[0].additional_kwargs diff --git a/docs/skill-usage-ui.md b/docs/skill-usage-ui.md new file mode 100644 index 0000000000..f537cedff8 --- /dev/null +++ b/docs/skill-usage-ui.md @@ -0,0 +1,49 @@ +# Skill usage in answer details + +The answer toolbar shows skills that the lead agent actually loaded during that +answer's run. A skill catalog entry, a mention in answer text, a failed read, or +a client-supplied claim does not count as usage. Shell reads and subagent loads +are outside this evidence path. Messages without captured usage omit the entry. + +## Capture and history + +`skill_usage.py` makes display-only snapshots after successful configured skill +reads and explicit slash activation. The read boundary discards tool-supplied +skill metadata and stamps only successful messages from the configured read +tool and requested `SKILL.md` path, including matching ToolMessages inside a +`Command`. The output-budget middleware checks that producer and path again, +then updates and registers the snapshot after any externalization or truncation, +so the bounded display snapshot and its hash describe the final model-visible +output. Slash activation stamps the first AI response that received it. +Each snapshot carries the canonical path, category, name, description, loaded +content, SHA-256, activation mode, and a partial flag for range or truncated +reads. Content is bounded to 100,000 characters. The snapshot never authorizes +a tool or secret, and Gateway strips it from external messages. + +Both producers register with `runtime.context["__run_journal"]`. A tool-end +callback can serialize output before middleware adds metadata, so the journal +keeps the first snapshot per path in load order (up to 64 skills) and copies the +cumulative list to `additional_kwargs.skill_usages` on canonical lead AI answers +without tool calls. This survives pagination and checkpoint compaction of +earlier load messages. Native subagents do not receive the lead journal. Provider +messages are unchanged; replay does not replace the original snapshot list; +closing the journal releases it. The existing message run ID owns run scope, +and serialization/history must preserve the display kwargs. + +## Presentation + +`core/skills/usage.ts` groups server-owned snapshots by run, deduplicating by +path in first-load order. The terminal `skill_usages` aggregate takes precedence +over singular snapshots. The menu anchors on the run's last assistant bubble; +when live run IDs are absent, visible human and clarification-result boundaries +separate runs, including continuations whose human replies are hidden. + +Answer details use the generic `workspace/message-details` menu, provider, and +panel. Add later detail types with a descriptor (`id`, `title`, `content`, +optional `actions`) instead of a separate drawer. `ChatBox` renders the detail +inside its resizable desktop group or mobile sheet, closing other panels on +selection. Heavy content loads on interaction. Hover opens the Skills menu; +selecting a name focuses the captured `SKILL.md` heading, and explicit close +restores focus to its trigger. A partial snapshot shows a notice. Copy returns +the original Markdown, including frontmatter. Package-relative links and images +stay inert because the snapshot has no trustworthy live package destination. diff --git a/frontend/docs/conversation-ui-ownership.md b/frontend/docs/conversation-ui-ownership.md new file mode 100644 index 0000000000..10727b9a67 --- /dev/null +++ b/frontend/docs/conversation-ui-ownership.md @@ -0,0 +1,30 @@ +# Conversation UI ownership + +This expands the ownership index in `frontend/src/AGENTS.md`. + +- `src/components/workspace/model-picker-content.tsx` owns the compact model + list, favorite grouping, and the anchored non-modal picker shared by the main + composer and Side Chat. Each row keeps model selection and its inline + favorite star as sibling buttons. The picker deliberately follows the + pre-favorites two-line row density and does not add a search field. Favorites + are stored by + `core/models/favorites-store.ts` under a user-scoped browser key and only + reorder derived display arrays: never sort `useModels().models`, promote a + favorite to the default model, prune a temporarily unavailable favorite, or + merge the main and Side Chat selection callbacks. Keep favorite buttons out + of model-selection buttons; the two call sites continue to own their triggers + and their distinct mode/reasoning-effort transitions. +- `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. +- `src/components/workspace/messages/message-list.tsx` owns human-input card answered/latest/pending gating; entry pages only translate a submitted card response into `sendMessage` calls. +- `src/components/workspace/browser-view/browser-view-panel.tsx` forwards each physical pointer click as one `click` input; do not also emit `down`/`up` for the same gesture because the remote Playwright click would run twice. +- `src/components/workspace/browser-view/use-browser-stream.ts` requests binary JPEG + frames with `frame_format=binary`; status, URL, tabs, and navigation rejection + messages remain JSON. `LatestBrowserFrameBuffer` keeps only the newest pending + frame, publishes through `useSyncExternalStore` at most once per animation + frame, and owns object-URL revocation. Keep the Gateway's legacy JSON/base64 + frame path for older clients. diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 904f2fa381..a4b98fee51 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -1,10 +1,11 @@ ### Data Flow -Artifact state, upload metadata, and workspace changes carry raw filesystem -paths. `urlOfArtifact` and `resolveArtifactURL` encode those paths without -decoding literal percent sequences. Only Markdown destinations are URL inputs: -`resolveMarkdownArtifactURL` decodes once at that boundary, and relative image -resolution matches the decoded name against raw artifact paths before encoding. +Answer details use `workspace/message-details` descriptors; skill run scoping +is documented in `docs/skill-usage-ui.md`. + +Artifact URLs encode raw filesystem paths, preserving literal percent sequences. +Only Markdown destinations decode once; relative images match decoded names +against raw artifact paths before encoding. 1. Optional composer helpers such as `core/input-polish` can rewrite the local draft before submission, and `core/voice-input` can transcribe browser microphone input into that same local draft; confirmed user input then flows to thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming 2. Stream events update thread state (messages, artifacts, todos, goal). The main thread stream uses the LangGraph SDK's `throttle: true` mode so updates received in the same macrotask coalesce before React is notified; do not replace it with a numeric delay without validating the SDK's trailing-debounce behavior on a continuous stream. @@ -162,46 +163,18 @@ Array previews coalesce consecutive generated markers only at the end into one o ### Interaction Ownership -- `src/components/workspace/model-picker-content.tsx` owns the compact model - list, favorite grouping, and the anchored non-modal picker shared by the main - composer and Side Chat. Each row keeps model selection and its inline - favorite star as sibling buttons. The picker deliberately follows the - pre-favorites two-line row density and does not add a search field. Favorites - are stored by - `core/models/favorites-store.ts` under a user-scoped browser key and only - reorder derived display arrays: never sort `useModels().models`, promote a - favorite to the default model, prune a temporarily unavailable favorite, or - merge the main and Side Chat selection callbacks. Keep favorite buttons out - of model-selection buttons; the two call sites continue to own their triggers - and their distinct mode/reasoning-effort transitions. -- `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. -- `src/components/workspace/messages/message-list.tsx` owns human-input card answered/latest/pending gating; entry pages only translate a submitted card response into `sendMessage` calls. -- `src/components/workspace/browser-view/browser-view-panel.tsx` forwards each physical pointer click as one `click` input; do not also emit `down`/`up` for the same gesture because the remote Playwright click would run twice. -- `src/components/workspace/browser-view/use-browser-stream.ts` requests binary JPEG - frames with `frame_format=binary`; status, URL, tabs, and navigation rejection - messages remain JSON. `LatestBrowserFrameBuffer` keeps only the newest pending - frame, publishes through `useSyncExternalStore` at most once per animation - frame, and owns object-URL revocation. Keep the Gateway's legacy JSON/base64 - frame path for older clients. +Model picker, chat entry pages, branch tree, and browser-control ownership are +detailed in `frontend/docs/conversation-ui-ownership.md`. + - `src/core/threads/hooks.ts` owns pre-submit upload state and thread submission. - `src/components/workspace/chats/chat-box.tsx` owns the desktop right-panel layout, and **all three** right panels (artifacts, sidecar, browser) share one `ResizablePanelGroup` — do not fork a non-resizable branch per panel kind, which is how the artifacts divider silently lost its drag handle (#4465). Open/close is `collapse()` / `resize()` on the side panel's imperative handle, not conditional rendering, so the width can animate. Three constraints hold that together: the size transition is applied from the group as `[&>[data-panel]]:transition-[flex-grow]` because the sized flex item is the library's own `[data-panel]` element rather than the child `className` lands on; it is applied only while an open/close is in flight, so a drag is not interpolated frame by frame; and during the animation the panel content is held at its final width in `cqw` and clipped, because a reflowing message list re-runs its scroll-to-bottom (pinned by `tests/e2e/sidecar-chat.spec.ts`'s no-animated-scroll test) and a re-wrapping composer changes which responsive labels it shows. Because the panel is `collapsible`, the library can also collapse it to `0%` on its own when a drag crosses `minSize`, without going through the state that owns it. `onResize` records the last positive size while the pointer moves, but the owning `sidecar` / `browserView` / `artifactsOpen` state must only mirror a final `0%` layout from `onLayoutChanged`, after pointer release; closing on the first `0%` resize frame breaks a continuous drag that reaches the edge and then reverses before release. -Clarification ToolMessages delimit completed runs for streaming message grouping, -including continuations submitted with hidden human replies. Do not classify all -messages after the last visible human as unresolved once a clarification result -has arrived. The processing renderer keeps tool-calling messages intact for -association and usage accounting, but renders text accompanying -`ask_clarification` outside the execution panel (including mixed tool calls). - -`findCurrentTurnStartIndex` owns the boundary rule for both full and incremental -message grouping. Incremental prefix/tail splitting applies only at human -boundaries; clarification results also belong to the preceding processing group, -so derive the full grouping and stabilize references at clarification boundaries. +`ask_clarification` ToolMessages delimit runs despite hidden human replies; skill +usage splits after the clarification group. Keep the tool call in its processing +group and render accompanying text outside the execution panel, including mixed +tool calls. `findCurrentTurnStartIndex` owns full and incremental grouping: +split incremental prefix/tail only at human boundaries, then stabilize +clarification references from full grouping. ### Knowledge source citations diff --git a/frontend/src/components/workspace/chats/chat-box.tsx b/frontend/src/components/workspace/chats/chat-box.tsx index 344c89557f..f739b496b0 100644 --- a/frontend/src/components/workspace/chats/chat-box.tsx +++ b/frontend/src/components/workspace/chats/chat-box.tsx @@ -22,12 +22,17 @@ import { SheetHeader, SheetTitle, } from "@/components/ui/sheet"; +import { useI18n } from "@/core/i18n/hooks"; import { env } from "@/env"; import { useIsMobile } from "@/hooks/use-mobile"; import { cn } from "@/lib/utils"; import { useArtifacts } from "../artifacts/context"; import { useMaybeBrowserView } from "../browser-view/context"; +import { + MessageDetailsProvider, + useMaybeMessageDetails, +} from "../message-details/context"; import { useThread } from "../messages/context"; import { useMaybeSidecar } from "../sidecar/context"; @@ -68,16 +73,50 @@ const SidecarPanel = dynamic( { loading: RightPanelLoading }, ); +const MessageDetailsPanel = dynamic( + () => + import("../message-details/message-details-panel").then( + (module) => module.MessageDetailsPanel, + ), + { loading: RightPanelLoading }, +); + const RIGHT_PANEL_ANIMATION_MS = 280; const RIGHT_PANEL_DEFAULT_SIZE = "40%"; -type RightPanelKind = "sidecar" | "artifacts" | "browser"; +type RightPanelKind = "sidecar" | "artifacts" | "browser" | "details"; -const ChatBox: React.FC<{ +interface ChatBoxProps { children: React.ReactNode; threadId: string; browserEnabled?: boolean; -}> = ({ children, threadId, browserEnabled = true }) => { +} + +function ChatBox(props: ChatBoxProps) { + const { setOpen: setArtifactsOpen } = useArtifacts(); + const sidecar = useMaybeSidecar(); + const browserView = useMaybeBrowserView(); + const closeOtherPanels = useCallback(() => { + setArtifactsOpen(false); + sidecar?.close(); + browserView?.close(); + }, [setArtifactsOpen, sidecar, browserView]); + + return ( + + + + ); +} + +const ChatBoxContent: React.FC = ({ + children, + threadId, + browserEnabled = true, +}) => { + const { t } = useI18n(); + const messageDetails = useMaybeMessageDetails(); + const detailsOpen = messageDetails?.open ?? false; const { thread } = useThread(); const isMobile = useIsMobile(); const pathname = usePathname(); @@ -158,7 +197,9 @@ const ChatBox: React.FC<{ ? "browser" : artifactPanelOpen ? "artifacts" - : null; + : detailsOpen + ? "details" + : null; const rightPanelOpen = activeRightPanel !== null; const [renderedRightPanel, setRenderedRightPanel] = useState(activeRightPanel); @@ -212,11 +253,20 @@ const ChatBox: React.FC<{ sidecar?.close(); } else if (activeRightPanel === "browser") { browserView?.close(); + } else if (activeRightPanel === "details") { + messageDetails?.close(); } else if (activeRightPanel === "artifacts") { setArtifactsOpen(false); } }, - [activeRightPanel, browserView, resizableIdBase, setArtifactsOpen, sidecar], + [ + activeRightPanel, + browserView, + messageDetails, + resizableIdBase, + setArtifactsOpen, + sidecar, + ], ); useEffect(() => { @@ -285,7 +335,22 @@ const ChatBox: React.FC<{ } }, [browserEnabled, browserView]); + useEffect(() => { + if (detailsOpen && (sidecarOpen || browserViewOpen || artifactPanelOpen)) { + messageDetails?.close({ restoreFocus: false }); + } + }, [ + detailsOpen, + sidecarOpen, + browserViewOpen, + artifactPanelOpen, + messageDetails, + ]); + const rightPanelContent = useMemo(() => { + if (renderedRightPanel === "details") { + return ; + } if (renderedRightPanel === "browser") { return ; } @@ -357,6 +422,9 @@ const ChatBox: React.FC<{ if (open) { return; } + if (detailsOpen) { + messageDetails?.close(); + } if (sidecarOpen) { sidecar?.close(); } @@ -378,7 +446,9 @@ const ChatBox: React.FC<{ ? "Sidecar" : renderedRightPanel === "browser" ? "Browser" - : "Artifacts"} + : renderedRightPanel === "details" + ? (messageDetails?.selectedDetail?.title ?? t.common.more) + : "Artifacts"} Browse the side panel for this conversation. diff --git a/frontend/src/components/workspace/copy-button.tsx b/frontend/src/components/workspace/copy-button.tsx index ae7aea3bc0..40dc4e6ecb 100644 --- a/frontend/src/components/workspace/copy-button.tsx +++ b/frontend/src/components/workspace/copy-button.tsx @@ -43,9 +43,9 @@ export function CopyButton({ {...props} > {copied ? ( - + ) : ( - + )} diff --git a/frontend/src/components/workspace/message-details/context.tsx b/frontend/src/components/workspace/message-details/context.tsx new file mode 100644 index 0000000000..2e0b3889b3 --- /dev/null +++ b/frontend/src/components/workspace/message-details/context.tsx @@ -0,0 +1,80 @@ +"use client"; + +import { + createContext, + useCallback, + useContext, + useEffect, + useRef, + useMemo, + useState, + type ReactNode, +} from "react"; + +export interface MessageDetail { + id: string; + title: string; + content: ReactNode; + actions?: ReactNode; +} + +interface MessageDetailsContextValue { + open: boolean; + selectedDetail: MessageDetail | null; + select: (detail: MessageDetail, origin?: HTMLElement) => void; + close: (options?: { restoreFocus?: boolean }) => void; +} + +const MessageDetailsContext = createContext( + null, +); + +export function MessageDetailsProvider({ + children, + onSelect, +}: { + children: ReactNode; + onSelect?: () => void; +}) { + const [open, setOpen] = useState(false); + const [selectedDetail, setSelectedDetail] = useState( + null, + ); + const originRef = useRef(undefined); + const restoreFocusRef = useRef(false); + useEffect(() => { + if (!open && restoreFocusRef.current) { + restoreFocusRef.current = false; + if (originRef.current?.isConnected) { + originRef.current.focus({ preventScroll: true }); + } + } + }, [open]); + const select = useCallback( + (detail: MessageDetail, origin?: HTMLElement) => { + originRef.current = origin; + restoreFocusRef.current = false; + onSelect?.(); + setSelectedDetail(detail); + setOpen(true); + }, + [onSelect], + ); + const close = useCallback((options?: { restoreFocus?: boolean }) => { + restoreFocusRef.current = options?.restoreFocus !== false; + setOpen(false); + }, []); + const value = useMemo( + () => ({ open, selectedDetail, select, close }), + [open, selectedDetail, select, close], + ); + return ( + + {children} + + ); +} + +export function useMaybeMessageDetails() { + return useContext(MessageDetailsContext); +} diff --git a/frontend/src/components/workspace/message-details/message-details-menu.tsx b/frontend/src/components/workspace/message-details/message-details-menu.tsx new file mode 100644 index 0000000000..a7a5e98c48 --- /dev/null +++ b/frontend/src/components/workspace/message-details/message-details-menu.tsx @@ -0,0 +1,147 @@ +"use client"; + +import { useEffect, useRef, useState, type ReactNode } from "react"; + +import { Button } from "@/components/ui/button"; +import { + DropdownMenu, + DropdownMenuContent, + DropdownMenuItem, + DropdownMenuLabel, + DropdownMenuSeparator, + DropdownMenuTrigger, +} from "@/components/ui/dropdown-menu"; + +export interface MessageDetailsMenuEntry { + id: string; + title: string; + subtitle?: string; + onSelect: (origin?: HTMLElement) => void; +} + +export function MessageDetailsMenu({ + label, + title, + icon, + entries, +}: { + label: string; + title: string; + icon: ReactNode; + entries: MessageDetailsMenuEntry[]; +}) { + const [open, setOpen] = useState(false); + const triggerRef = useRef(null); + const selectedEntryRef = useRef(false); + const interaction = useRef<"hover" | "press">("press"); + const openTimer = useRef | null>(null); + const closeTimer = useRef | null>(null); + const clearTimers = () => { + if (openTimer.current) clearTimeout(openTimer.current); + if (closeTimer.current) clearTimeout(closeTimer.current); + openTimer.current = null; + closeTimer.current = null; + }; + useEffect( + () => () => { + if (openTimer.current) clearTimeout(openTimer.current); + if (closeTimer.current) clearTimeout(closeTimer.current); + }, + [], + ); + const enter = (pointerType: string) => { + if (pointerType !== "mouse") return; + clearTimers(); + if (!open) { + openTimer.current = setTimeout(() => { + interaction.current = "hover"; + setOpen(true); + openTimer.current = null; + }, 120); + } + }; + const leave = (pointerType: string) => { + if (pointerType !== "mouse") return; + clearTimers(); + if (interaction.current === "hover") { + closeTimer.current = setTimeout(() => { + setOpen(false); + closeTimer.current = null; + }, 180); + } + }; + const press = () => { + clearTimers(); + interaction.current = "press"; + }; + + if (entries.length === 0) return null; + return ( + { + clearTimers(); + if (nextOpen) selectedEntryRef.current = false; + setOpen(nextOpen); + }} + > + + + + enter(event.pointerType)} + onPointerLeave={(event) => leave(event.pointerType)} + // Radix forwards this FocusScope hook at runtime but omits it from + // DropdownMenuContent's public type; keep hover from moving composer focus. + {...{ + onOpenAutoFocus: (event: Event) => { + if (interaction.current === "hover") event.preventDefault(); + }, + }} + onCloseAutoFocus={(event) => { + if (interaction.current === "hover" || selectedEntryRef.current) + event.preventDefault(); + }} + > + + {title} + + + {entries.map((entry) => ( + { + selectedEntryRef.current = true; + entry.onSelect(triggerRef.current ?? undefined); + }} + > + + {entry.title} + + {entry.subtitle && ( + + {entry.subtitle} + + )} + + ))} + + + ); +} diff --git a/frontend/src/components/workspace/message-details/message-details-panel.tsx b/frontend/src/components/workspace/message-details/message-details-panel.tsx new file mode 100644 index 0000000000..db08c23f2e --- /dev/null +++ b/frontend/src/components/workspace/message-details/message-details-panel.tsx @@ -0,0 +1,63 @@ +"use client"; + +import { FileTextIcon, XIcon } from "lucide-react"; +import { useEffect, useRef } from "react"; + +import { Button } from "@/components/ui/button"; +import { useI18n } from "@/core/i18n/hooks"; + +import { useMaybeMessageDetails } from "./context"; + +export function MessageDetailsPanel() { + const { t } = useI18n(); + const details = useMaybeMessageDetails(); + const selected = details?.selectedDetail; + const headingRef = useRef(null); + useEffect(() => { + if (details?.open && selected) { + headingRef.current?.focus({ preventScroll: true }); + } + }, [details?.open, selected]); + if (!details?.open || !selected) return null; + + return ( +
{ + if (event.key === "Escape" && !event.defaultPrevented) { + event.preventDefault(); + event.stopPropagation(); + details.close(); + } + }} + > +
+ +

+ {selected.title} +

+ {selected.actions} + +
+
+ {selected.content} +
+
+ ); +} diff --git a/frontend/src/components/workspace/messages/message-list-item.tsx b/frontend/src/components/workspace/messages/message-list-item.tsx index 933949cc99..4a0b2717a0 100644 --- a/frontend/src/components/workspace/messages/message-list-item.tsx +++ b/frontend/src/components/workspace/messages/message-list-item.tsx @@ -264,7 +264,7 @@ export function MessageListItem({ disabled={isEditPending || isSubmittingEdit} onClick={startEditing} > - + )} diff --git a/frontend/src/components/workspace/messages/message-list.tsx b/frontend/src/components/workspace/messages/message-list.tsx index 1c7fc3a38b..fad1a45fdc 100644 --- a/frontend/src/components/workspace/messages/message-list.tsx +++ b/frontend/src/components/workspace/messages/message-list.tsx @@ -73,6 +73,10 @@ import { buildMessageSidecarContext, type SidecarContext, } from "@/core/sidecar"; +import { + getSkillUsageByGroupIndex, + type SkillUsage, +} from "@/core/skills/usage"; import type { Subtask } from "@/core/tasks"; import { useUpdateSubtask } from "@/core/tasks/context"; import { collectRenderedSubtasks } from "@/core/tasks/subtask-render"; @@ -83,6 +87,7 @@ import { ArtifactFileList } from "../artifacts/artifact-file-list"; import { useMaybeBrowserView } from "../browser-view"; import { CopyButton } from "../copy-button"; import { useMaybeSidecar } from "../sidecar/context"; +import { SkillUsageMenu } from "../skill-usage/skill-usage-menu"; import { Tooltip } from "../tooltip"; import { ConversationOutline } from "./conversation-outline"; @@ -545,6 +550,10 @@ export function MessageList({ () => getWorkspaceChangeAnchorGroupIndices(groupedMessages), [groupedMessages], ); + const skillUsageByGroupIndex = useMemo( + () => getSkillUsageByGroupIndex(groupedMessages), + [groupedMessages], + ); useEffect(() => { setClientDurationsByGroupId((current) => { let next: Map | undefined; @@ -891,6 +900,7 @@ export function MessageList({ isStreaming: boolean, enableBranchForTurn: boolean, enableRegenerateForTurn: boolean, + skills?: SkillUsage[], ) => { const clipboardData = getAssistantTurnCopyData(messages, { isStreaming }); const actionTarget = [...messages] @@ -905,8 +915,9 @@ export function MessageList({ } return ( -
+
{clipboardData && } + {!isStreaming && skills && } {enableBranchForTurn && !isStreaming && actionTarget?.id && @@ -973,7 +984,7 @@ export function MessageList({ > , ); diff --git a/frontend/src/components/workspace/skill-usage/skill-usage-copy-action.tsx b/frontend/src/components/workspace/skill-usage/skill-usage-copy-action.tsx new file mode 100644 index 0000000000..896a075408 --- /dev/null +++ b/frontend/src/components/workspace/skill-usage/skill-usage-copy-action.tsx @@ -0,0 +1,28 @@ +"use client"; + +import { CopyIcon } from "lucide-react"; +import { toast } from "sonner"; + +import { Button } from "@/components/ui/button"; +import { writeTextToClipboard } from "@/core/clipboard"; +import { useI18n } from "@/core/i18n/hooks"; + +export function SkillUsageCopyAction({ content }: { content: string }) { + const { t } = useI18n(); + return ( + + ); +} diff --git a/frontend/src/components/workspace/skill-usage/skill-usage-menu.tsx b/frontend/src/components/workspace/skill-usage/skill-usage-menu.tsx new file mode 100644 index 0000000000..0edb13d620 --- /dev/null +++ b/frontend/src/components/workspace/skill-usage/skill-usage-menu.tsx @@ -0,0 +1,46 @@ +"use client"; + +import { BoxIcon } from "lucide-react"; +import dynamic from "next/dynamic"; + +import { useI18n } from "@/core/i18n/hooks"; +import { type SkillUsage } from "@/core/skills/usage"; + +import { useMaybeMessageDetails } from "../message-details/context"; +import { MessageDetailsMenu } from "../message-details/message-details-menu"; + +import { SkillUsageCopyAction } from "./skill-usage-copy-action"; +import { skillSourceLabel } from "./source-label"; + +const SkillUsagePanel = dynamic(() => + import("./skill-usage-panel").then((module) => module.SkillUsagePanel), +); + +export function SkillUsageMenu({ skills }: { skills: SkillUsage[] }) { + const { t } = useI18n(); + const details = useMaybeMessageDetails(); + if (!details || skills.length === 0) return null; + + return ( + } + entries={skills.map((skill) => ({ + id: `${skill.path}:${skill.content_hash}`, + title: skill.name, + subtitle: skillSourceLabel(skill.category, t), + onSelect: (origin) => + details.select( + { + id: `skill:${skill.path}:${skill.content_hash}`, + title: "SKILL.md", + content: , + actions: , + }, + origin, + ), + }))} + /> + ); +} diff --git a/frontend/src/components/workspace/skill-usage/skill-usage-panel.tsx b/frontend/src/components/workspace/skill-usage/skill-usage-panel.tsx new file mode 100644 index 0000000000..63b140d425 --- /dev/null +++ b/frontend/src/components/workspace/skill-usage/skill-usage-panel.tsx @@ -0,0 +1,68 @@ +"use client"; + +import { BoxIcon } from "lucide-react"; + +import { useI18n } from "@/core/i18n/hooks"; +import { type SkillUsage } from "@/core/skills/usage"; + +import { MarkdownContent } from "../messages/markdown-content"; + +import { snapshotMarkdownComponents } from "./snapshot-markdown-components"; +import { skillSourceLabel } from "./source-label"; + +export function SkillUsagePanel({ skill }: { skill: SkillUsage }) { + const { t } = useI18n(); + // The snapshot remains intact for copying; metadata is presented separately. + const body = skill.content.replace( + /^\uFEFF?---\r?\n[\s\S]*?\r?\n(?:---|\.\.\.)[ \t]*(?:\r?\n|$)/, + "", + ); + + return ( +
+
+

+ {skill.path} +

+
+ + + +

+ {skill.name} +

+ + {skillSourceLabel(skill.category, t)} + +
+
+
{t.skillUsage.name}
+
{skill.name}
+
{t.skillUsage.description}
+
+ {skill.description || "—"} +
+
+
+ {skill.partial && ( +

+ {t.skillUsage.partial} +

+ )} +
+ +
+
+ ); +} diff --git a/frontend/src/components/workspace/skill-usage/snapshot-markdown-components.tsx b/frontend/src/components/workspace/skill-usage/snapshot-markdown-components.tsx new file mode 100644 index 0000000000..0b3c12e5f1 --- /dev/null +++ b/frontend/src/components/workspace/skill-usage/snapshot-markdown-components.tsx @@ -0,0 +1,57 @@ +import { type StreamdownComponentOverrides } from "@/core/streamdown/components"; + +import { + createMarkdownLinkComponent, + isSafeHref, +} from "../messages/markdown-link"; + +const MarkdownLink = createMarkdownLinkComponent(); + +// Only SKILL.md is captured. Package-relative links and images have no resource +// backing in this snapshot, so never resolve them against the current chat URL. +export const snapshotMarkdownComponents: StreamdownComponentOverrides = { + a: ({ href, children, className, title }) => { + if (href && /^(https?:\/\/|mailto:|tel:)/i.test(href) && isSafeHref(href)) { + return ( + + {children} + + ); + } + return ( + + {children} + + ); + }, + img: ({ src, alt, title }) => { + if ( + typeof src === "string" && + /^https?:\/\//i.test(src) && + isSafeHref(src) + ) { + // Preserve the shared renderer's remote image behavior without fetching + // unresolved package paths or requiring a new skill-resource endpoint. + return ( + {alt + ); + } + return ( + + {alt?.trim() ? alt : typeof src === "string" ? src : ""} + + ); + }, +}; diff --git a/frontend/src/components/workspace/skill-usage/source-label.ts b/frontend/src/components/workspace/skill-usage/source-label.ts new file mode 100644 index 0000000000..e4cb3548fd --- /dev/null +++ b/frontend/src/components/workspace/skill-usage/source-label.ts @@ -0,0 +1,16 @@ +import { type Translations } from "@/core/i18n/locales/types"; + +export function skillSourceLabel(category: string, t: Translations) { + switch (category) { + case "public": + return t.skillUsage.builtIn; + case "custom": + return t.skillUsage.custom; + case "integrations": + return t.skillUsage.integration; + case "legacy": + return t.skillUsage.legacy; + default: + return category; + } +} diff --git a/frontend/src/core/i18n/locales/en-US.ts b/frontend/src/core/i18n/locales/en-US.ts index f30976cf3f..929e2d1f8a 100644 --- a/frontend/src/core/i18n/locales/en-US.ts +++ b/frontend/src/core/i18n/locales/en-US.ts @@ -184,6 +184,20 @@ export const enUS: Translations = { showBrowser: "Open browser panel", }, + skillUsage: { + used: "Skills used", + title: "Skills", + name: "Name", + description: "Description", + builtIn: "Built-in", + custom: "Custom", + integration: "Integration", + legacy: "Legacy", + copy: "Copy skill snapshot", + partial: + "This is a partial snapshot of the skill instructions loaded during this run.", + }, + runDuration: { reasoning: "Reasoning", working: "Working…", diff --git a/frontend/src/core/i18n/locales/types.ts b/frontend/src/core/i18n/locales/types.ts index dfbbd5d4f9..405feb5553 100644 --- a/frontend/src/core/i18n/locales/types.ts +++ b/frontend/src/core/i18n/locales/types.ts @@ -160,6 +160,19 @@ export interface Translations { showBrowser: string; }; + skillUsage: { + used: string; + title: string; + name: string; + description: string; + builtIn: string; + custom: string; + integration: string; + legacy: string; + copy: string; + partial: string; + }; + runDuration: { reasoning: string; working: string; diff --git a/frontend/src/core/i18n/locales/zh-CN.ts b/frontend/src/core/i18n/locales/zh-CN.ts index 2729fcd5b6..ec4ff9a10a 100644 --- a/frontend/src/core/i18n/locales/zh-CN.ts +++ b/frontend/src/core/i18n/locales/zh-CN.ts @@ -178,6 +178,19 @@ export const zhCN: Translations = { showBrowser: "打开浏览器面板", }, + skillUsage: { + used: "使用的技能", + title: "技能", + name: "名称", + description: "描述", + builtIn: "内置", + custom: "自定义", + integration: "集成", + legacy: "旧版", + copy: "复制技能快照", + partial: "此快照仅包含本次运行加载的部分技能内容。", + }, + runDuration: { reasoning: "思考过程", working: "执行中…", diff --git a/frontend/src/core/skills/usage.ts b/frontend/src/core/skills/usage.ts new file mode 100644 index 0000000000..bf964e985f --- /dev/null +++ b/frontend/src/core/skills/usage.ts @@ -0,0 +1,122 @@ +import type { Message } from "@langchain/langgraph-sdk"; + +import { getMessageRunId } from "@/core/messages/run-duration"; +import type { MessageGroup } from "@/core/messages/utils"; + +export const SKILL_USAGES_KEY = "skill_usages"; + +/** Display evidence captured at load time, never a catalog/authorization source. */ +export interface SkillUsage { + name: string; + description: string; + category: string; + path: string; + content: string; + content_hash: string; + activation: "automatic" | "slash"; + partial: boolean; +} + +export function readSkillUsage(message: Message): SkillUsage | undefined { + if (message.type !== "ai" && message.type !== "tool") return undefined; + if (message.type === "tool" && message.status === "error") return undefined; + return parseSkillUsage(message.additional_kwargs?.skill_usage); +} + +function parseSkillUsage(raw: unknown): SkillUsage | undefined { + if (!raw || typeof raw !== "object") return undefined; + const value = raw as Record; + if ( + typeof value.name !== "string" || + !value.name || + typeof value.description !== "string" || + typeof value.category !== "string" || + typeof value.path !== "string" || + !value.path.endsWith("/SKILL.md") || + typeof value.content !== "string" || + typeof value.content_hash !== "string" || + !/^[a-f0-9]{64}$/.test(value.content_hash) || + (value.activation !== "automatic" && value.activation !== "slash") || + typeof value.partial !== "boolean" + ) + return undefined; + return value as unknown as SkillUsage; +} + +/** One menu on the last assistant bubble per run, in first-load order. + * Live messages may not yet carry run ids; use visible human and clarification + * boundaries until history provides those ids. Never infer usage from answer + * text or the catalog. + */ +export function getSkillUsageByGroupIndex( + groups: MessageGroup[], +): Map { + const byRun = new Map< + string, + { skills: Map; anchor?: number } + >(); + let start = 0; + while (start < groups.length) { + let end = start + 1; + while ( + end < groups.length && + groups[end]?.type !== "human" && + groups[end - 1]?.type !== "assistant:clarification" + ) + end++; + const runIds = new Set(); + for (let i = start; i < end; i++) { + for (const message of groups[i]!.messages) { + if (message.type === "human") continue; + const runId = getMessageRunId(message); + if (runId) runIds.add(runId); + } + } + let currentKey = + runIds.size === 1 ? `run:${[...runIds][0]}` : `turn:${start}`; + for (let i = start; i < end; i++) { + const group = groups[i]!; + for (const message of group.messages) { + if (message.type !== "ai" && message.type !== "tool") continue; + const runId = getMessageRunId(message); + if (runId) currentKey = `run:${runId}`; + let run = byRun.get(currentKey); + if (!run) { + run = { skills: new Map() }; + byRun.set(currentKey, run); + } + const aggregate: unknown = + message.additional_kwargs?.[SKILL_USAGES_KEY]; + const usages = + message.type === "ai" && Array.isArray(aggregate) + ? aggregate.map(parseSkillUsage) + : [readSkillUsage(message)]; + if (message.type === "ai" && Array.isArray(aggregate)) { + // The aggregate preserves first-load order/content even when earlier + // reads are absent from the currently loaded history page. + const canonical = new Map(); + for (const usage of usages) { + if (usage && !canonical.has(usage.path)) + canonical.set(usage.path, usage); + } + for (const [path, usage] of run.skills) { + if (!canonical.has(path)) canonical.set(path, usage); + } + run.skills = canonical; + } + for (const usage of usages) { + if (usage && !run.skills.has(usage.path)) + run.skills.set(usage.path, usage); + } + if (group.type === "assistant" && message.type === "ai") run.anchor = i; + } + } + start = end; + } + const result = new Map(); + for (const run of byRun.values()) { + if (run.anchor !== undefined && run.skills.size > 0) + result.set(run.anchor, [...run.skills.values()]); + } + return result; +} diff --git a/frontend/src/core/threads/message-order.ts b/frontend/src/core/threads/message-order.ts index 4b6066f801..78a4aef336 100644 --- a/frontend/src/core/threads/message-order.ts +++ b/frontend/src/core/threads/message-order.ts @@ -29,6 +29,7 @@ import type { Message } from "@langchain/langgraph-sdk"; import { getMessageRunId } from "../messages/run-duration"; import { isHiddenFromUIMessage } from "../messages/utils"; +import { SKILL_USAGES_KEY } from "../skills/usage"; // Thread-global feed position, attached by the backend to history rows and to // `values` frame messages it has already persisted. Mirrors MESSAGE_SEQ_KEY in @@ -220,6 +221,7 @@ export function mergeMessages( const anySeqByIdentity = new Map(); const savedTurnDurations = new Map(); const savedRunIds = new Map(); + const savedSkillUsages = new Map(); const collectTrustedSeq = (message: Message) => { const identity = messageIdentity(message); if (!identity) { @@ -251,6 +253,15 @@ export function mergeMessages( if (identity && runId) { savedRunIds.set(identity, runId); } + if ( + identity && + Array.isArray(message.additional_kwargs?.[SKILL_USAGES_KEY]) + ) { + savedSkillUsages.set( + identity, + message.additional_kwargs[SKILL_USAGES_KEY], + ); + } if (identity && message.additional_kwargs?.turn_duration !== undefined) { savedTurnDurations.set( identity, @@ -413,10 +424,14 @@ export function mergeMessages( const shouldRestoreTurnDuration = savedTurnDurations.has(identity) && message.additional_kwargs?.turn_duration === undefined; + const shouldRestoreSkillUsages = + savedSkillUsages.has(identity) && + message.additional_kwargs?.[SKILL_USAGES_KEY] === undefined; if ( !shouldRestoreSeq && !shouldRestoreRunId && - !shouldRestoreTurnDuration + !shouldRestoreTurnDuration && + !shouldRestoreSkillUsages ) { return message; } @@ -429,6 +444,9 @@ export function mergeMessages( ...(shouldRestoreTurnDuration ? { turn_duration: savedTurnDurations.get(identity) } : {}), + ...(shouldRestoreSkillUsages + ? { [SKILL_USAGES_KEY]: savedSkillUsages.get(identity) } + : {}), }, } as Message; }); diff --git a/frontend/tests/e2e/skill-usage.spec.ts b/frontend/tests/e2e/skill-usage.spec.ts new file mode 100644 index 0000000000..455fb5459b --- /dev/null +++ b/frontend/tests/e2e/skill-usage.spec.ts @@ -0,0 +1,337 @@ +import { mkdir } from "node:fs/promises"; +import { join } from "node:path"; + +import { expect, test, type Page } from "@playwright/test"; + +import { mockLangGraphAPI } from "./utils/mock-api"; + +const THREAD_ID = "00000000-0000-0000-0000-000000009922"; +const RUN_ID = "skill-usage-run"; +const skills = [ + { + name: "using-superpowers", + description: "开始任务前识别适用的技能,并按技能工作流推进。", + category: "custom", + path: "/mnt/skills/custom/using-superpowers/SKILL.md", + content: + "---\nname: using-superpowers\ndescription: 开始任务前识别适用的技能,并按技能工作流推进。\n---\n# Using Superpowers\n\n在执行任务前,先检查是否有适用的技能。\n\n## 工作流程\n\n1. 理解用户请求,识别任务类型。\n2. 读取相关技能的说明与参考资料。\n3. 按工作流完成任务,持续验证结果。\n\n## 完成前验证\n\n以实际命令输出和检查结果为依据。记录遇到的问题与处理方式,再汇报最终结果。\n\n> 此处展示的是本轮加载时保存的技能内容。", + content_hash: "a".repeat(64), + activation: "slash", + partial: false, + }, + { + name: "verification-before-completion", + description: "交付前运行检查,并用实际结果验证完成情况。", + category: "public", + path: "/mnt/skills/public/verification-before-completion/SKILL.md", + content: + "# Verification Before Completion\n\n## 验证步骤\n\n- 运行相关测试与静态检查。\n- 检查实际界面与交互。\n- 将验证结果与交付内容一并提供。", + content_hash: "b".repeat(64), + activation: "automatic", + partial: false, + }, +]; + +function messages(includeSkills = true) { + return [ + { + type: "human", + id: "user", + content: "帮我检查这次改动,完成后给我看看结果。", + run_id: RUN_ID, + }, + { + type: "ai", + id: "start", + content: "我会先读取适用的技能,再检查改动和运行验证。", + run_id: RUN_ID, + additional_kwargs: includeSkills ? { skill_usage: skills[0] } : {}, + }, + { + type: "ai", + id: "read", + content: "", + run_id: RUN_ID, + tool_calls: [ + { + id: "read-call", + name: "read_file", + args: { path: skills[1]!.path, description: "读取完成前验证技能" }, + }, + ], + }, + { + type: "tool", + id: "loaded", + content: skills[1]!.content, + name: "read_file", + tool_call_id: "read-call", + run_id: RUN_ID, + additional_kwargs: includeSkills ? { skill_usage: skills[1] } : {}, + }, + { + type: "ai", + id: "final", + content: + "检查完成,本次改动已在本地验证。\n\n- **功能检查**:入口、列表和详情面板均可正常使用。\n- **历史记录**:刷新后仍能查看本轮使用的技能。\n- **界面适配**:桌面使用侧栏,手机使用抽屉。\n\n代码和验证结果已保留在本地。", + run_id: RUN_ID, + }, + ]; +} + +async function setup(page: Page, includeSkills = true) { + await page.addInitScript(() => localStorage.setItem("theme", "dark")); + await page + .context() + .addCookies([ + { name: "locale", value: "zh-CN", url: "http://localhost:3000" }, + ]); + mockLangGraphAPI(page, { + threads: [ + { + thread_id: THREAD_ID, + title: "技能使用记录", + messages: messages(includeSkills), + }, + ], + }); + await page.goto(`/workspace/chats/${THREAD_ID}`); + await expect(page.getByText("检查完成,本次改动已在本地验证。")).toBeVisible({ + timeout: 60_000, + }); +} + +async function screenshot(page: Page, name: string) { + const directory = process.env.SKILL_USAGE_SCREENSHOT_DIR; + if (!directory) return; + await mkdir(directory, { recursive: true }); + await page.screenshot({ + path: join(directory, name), + fullPage: true, + animations: "disabled", + }); + if (name.startsWith("skills-menu")) { + const menu = await page.getByRole("menu").boundingBox(); + const trigger = await page + .getByRole("button", { name: "使用的技能", exact: true }) + .boundingBox(); + if (menu && trigger) { + const x = Math.max(0, Math.min(menu.x, trigger.x) - 12); + const y = Math.max(0, menu.y - 12); + await page.screenshot({ + path: join(directory, name.replace(".png", "-closeup.png")), + animations: "disabled", + clip: { + x, + y, + width: + Math.max(menu.x + menu.width, trigger.x + trigger.width) - x + 12, + height: trigger.y + trigger.height - y + 12, + }, + }); + } + } +} + +test("shows one skill menu per run, switches snapshots and restores after reload", async ({ + page, +}) => { + test.setTimeout(120_000); + await page.setViewportSize({ width: 1440, height: 1000 }); + await setup(page); + const trigger = page.getByRole("button", { name: "使用的技能", exact: true }); + await expect(trigger).toHaveCount(1); + const toolbarSizes = await trigger + .locator("..") + .locator("button") + .evaluateAll((buttons) => + buttons.map((button) => ({ + button: Math.round(button.getBoundingClientRect().width), + icon: Math.round( + button.querySelector("svg")!.getBoundingClientRect().width, + ), + color: getComputedStyle(button).color, + })), + ); + expect(toolbarSizes).toHaveLength(4); + expect( + toolbarSizes.every(({ button, icon }) => button === 32 && icon === 16), + ).toBe(true); + expect(new Set(toolbarSizes.map(({ color }) => color)).size).toBe(1); + await trigger.hover(); + await expect( + page.getByRole("menuitem", { name: /using-superpowers.*自定义/ }), + ).toBeVisible(); + await expect( + page.getByRole("menuitem", { + name: /verification-before-completion.*内置/, + }), + ).toBeVisible(); + await screenshot(page, "skills-menu.png"); + const skillName = page + .getByRole("menuitem", { name: /using-superpowers/ }) + .getByText("using-superpowers"); + await skillName.hover(); + await expect(skillName).toHaveCSS("text-decoration-line", "underline"); + await screenshot(page, "skills-menu-name-hover.png"); + await page.getByRole("menuitem", { name: /using-superpowers/ }).click(); + const panel = page.getByRole("region", { name: "SKILL.md", exact: true }); + await expect( + panel.getByRole("heading", { name: "Using Superpowers", exact: true }), + ).toBeVisible(); + await expect(panel.getByText(skills[0]!.path)).toBeVisible(); + await expect( + panel.getByRole("button", { name: "复制技能快照" }), + ).toBeVisible(); + await screenshot(page, "skills-detail-desktop.png"); + const handle = page.locator('[data-slot="resizable-handle"]').first(); + await handle.hover(); + const initialWidth = (await panel.boundingBox())!.width; + await page.mouse.down(); + const bounds = (await handle.boundingBox())!; + await page.mouse.move(bounds.x - 120, bounds.y + bounds.height / 2, { + steps: 12, + }); + await page.mouse.up(); + await expect + .poll(async () => (await panel.boundingBox())!.width) + .toBeGreaterThan(initialWidth + 50); + await trigger.click(); + await page + .getByRole("menuitem", { name: /verification-before-completion/ }) + .click(); + await expect( + panel.getByRole("heading", { name: "Verification Before Completion" }), + ).toBeVisible(); + await panel.getByRole("button", { name: "关闭", exact: true }).click(); + await expect(panel).toBeHidden(); + await page.reload(); + await expect(trigger).toHaveCount(1); + await trigger.click(); + await page.getByRole("menuitem", { name: /using-superpowers/ }).click(); + await expect( + panel.getByRole("heading", { name: "Using Superpowers", exact: true }), + ).toBeVisible(); +}); + +test("mobile opens a readable detail sheet and supports closing", async ({ + page, +}) => { + await page.setViewportSize({ width: 390, height: 844 }); + await setup(page); + await page.getByRole("button", { name: "使用的技能", exact: true }).click(); + await page.getByRole("menuitem", { name: /using-superpowers/ }).click(); + const panel = page.getByRole("region", { name: "SKILL.md", exact: true }); + await expect( + panel.getByRole("heading", { name: "Using Superpowers", exact: true }), + ).toBeVisible(); + expect((await panel.boundingBox())!.width).toBeLessThan(391); + await screenshot(page, "skills-detail-mobile.png"); + await page.keyboard.press("Escape"); + await expect(panel).toBeHidden(); +}); + +test("older history without load evidence does not invent skill usage", async ({ + page, +}) => { + await setup(page, false); + await expect( + page.getByRole("button", { name: "使用的技能", exact: true }), + ).toHaveCount(0); +}); + +test("details share the browser panel and can reopen after drag collapse", async ({ + page, +}) => { + await page.setViewportSize({ width: 1440, height: 1000 }); + await setup(page); + const trigger = page.getByRole("button", { name: "使用的技能", exact: true }); + const openDetail = async () => { + await trigger.click(); + await page.getByRole("menuitem", { name: /using-superpowers/ }).click(); + await expect( + page.getByRole("region", { name: "SKILL.md", exact: true }), + ).toBeVisible(); + }; + await openDetail(); + await page.getByTestId("browser-trigger").click(); + await expect( + page.getByRole("region", { name: "SKILL.md", exact: true }), + ).toBeHidden(); + await openDetail(); + await expect(page.getByTestId("browser-trigger")).not.toHaveAttribute( + "aria-label", + "关闭", + ); + const handle = page.locator('[data-slot="resizable-handle"]').first(); + await handle.hover(); + const bounds = (await handle.boundingBox())!; + await page.mouse.down(); + await page.mouse.move(1438, bounds.y + bounds.height / 2, { steps: 15 }); + await page.mouse.up(); + await expect( + page.getByRole("region", { name: "SKILL.md", exact: true }), + ).toBeHidden(); + await openDetail(); +}); + +test("terminal-only history keeps original raw copy and scrolls long snapshots safely", async ({ + page, +}) => { + await page.setViewportSize({ width: 1440, height: 900 }); + await page.context().grantPermissions(["clipboard-read", "clipboard-write"]); + const content = `---\nname: long-report\ndescription: Long report workflow\n---\n# Long report\n\n[Local template](templates/apa.md)\n\n![Local diagram](images/flow.png)\n\n${"Paragraph with instructions.\n\n".repeat(160)}## End of snapshot`; + const original = { ...skills[0]!, content }; + await page + .context() + .addCookies([ + { name: "locale", value: "zh-CN", url: "http://localhost:3000" }, + ]); + mockLangGraphAPI(page, { + threads: [ + { + thread_id: THREAD_ID, + title: "Archived skill history", + messages: [ + { + type: "ai", + id: "final-only", + content: "历史已恢复。", + run_id: RUN_ID, + additional_kwargs: { skill_usages: [original] }, + }, + ], + }, + ], + }); + await page.goto(`/workspace/chats/${THREAD_ID}`); + await page.getByRole("button", { name: "使用的技能", exact: true }).click(); + await page.getByRole("menuitem", { name: /using-superpowers/ }).click(); + const panel = page.getByRole("region", { name: "SKILL.md", exact: true }); + await expect( + panel.getByRole("heading", { name: "Long report", exact: true }), + ).toBeVisible(); + await panel.getByRole("button", { name: "复制技能快照" }).click(); + await expect + .poll(() => page.evaluate(() => navigator.clipboard.readText())) + .toBe(content); + await expect(panel.getByRole("link", { name: "Local template" })).toHaveCount( + 0, + ); + await expect( + panel.getByText("Local template", { exact: true }), + ).toBeVisible(); + await expect(panel.locator('img[src="images/flow.png"]')).toHaveCount(0); + await panel.hover(); + await page.mouse.wheel(0, 15000); + await expect( + panel.getByRole("heading", { name: "End of snapshot" }), + ).toBeInViewport(); + await expect( + panel.getByRole("button", { name: "复制技能快照" }), + ).toBeInViewport(); + await panel.getByRole("button", { name: "关闭", exact: true }).click(); + await expect( + page.getByRole("button", { name: "使用的技能", exact: true }), + ).toBeFocused(); +}); diff --git a/frontend/tests/unit/components/workspace/message-details/message-details-menu.dom.test.tsx b/frontend/tests/unit/components/workspace/message-details/message-details-menu.dom.test.tsx new file mode 100644 index 0000000000..6f696b4a9c --- /dev/null +++ b/frontend/tests/unit/components/workspace/message-details/message-details-menu.dom.test.tsx @@ -0,0 +1,99 @@ +import { afterEach, describe, expect, it, rs } from "@rstest/core"; +import { + act, + cleanup, + fireEvent, + render, + screen, +} from "@testing-library/react"; + +import { MessageDetailsMenu } from "@/components/workspace/message-details/message-details-menu"; + +function renderMenu(onSelect = rs.fn()) { + render( + <> + +