diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts index 15e67f9b..b093fb20 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts @@ -63,6 +63,12 @@ import { ProtocolValidationError, assertValidProtocolPayload, } from "./protocol-validation" +import { + ReferenceSearchSession, + type ReferenceSearchSnapshot, +} from "./reference-search-session" + +export type { ReferenceSearchSnapshot } from "./reference-search-session" export type JsonRpcId = number | string @@ -581,6 +587,7 @@ class AcpClient { private pendingQuestions = new Map() private sessionDiscovery = new Map>() private lastEventTime = 0 + private referenceSearchSession: ReferenceSearchSession | null = null constructor(private readonly options: CreateDevoClientOptions) {} project = { @@ -887,7 +894,25 @@ class AcpClient { } find = { - files: async (_params: { query: string }) => ({ data: [] }), + // @ mention file search uses connection-local search/* RPC + notifications. + files: async (params: { query: string }) => { + const session = this.ensureReferenceSearchSession() + await session.startOrUpdate(params.query) + return { data: session.filePaths() } + }, + } + + referenceSearch = { + startOrUpdate: async (params: { query: string }) => ({ + data: await this.ensureReferenceSearchSession().startOrUpdate(params.query), + }), + cancel: async () => { + await this.ensureReferenceSearchSession().cancel() + return { data: null } + }, + subscribe: (listener: (snapshot: ReferenceSearchSnapshot) => void) => + this.ensureReferenceSearchSession().subscribe(listener), + getState: () => this.ensureReferenceSearchSession().getState(), } worktree = { @@ -1160,6 +1185,25 @@ class AcpClient { }) } + /** Devo extension RPCs that are not yet present in the generated ACP schema bundle. */ + private async requestExtension(method: string, params: unknown): Promise { + await this.open() + if (!this.transport) throw new Error("Devo ACP transport is not connected") + const extensionMethod = method.startsWith("_devo/") ? method : `_devo/${method}` + return this.transport.request(extensionMethod, params, this.options.directory) + } + + private ensureReferenceSearchSession(): ReferenceSearchSession { + if (!this.referenceSearchSession) { + const cwd = this.options.directory ?? defaultCwd() + this.referenceSearchSession = new ReferenceSearchSession( + (method, params) => this.requestExtension(method, params), + cwd, + ) + } + return this.referenceSearchSession + } + private handleTransportEvent(event: DevoAcpTransportEvent): void { if (event.type === "closed") { if (this.transport) { @@ -1179,6 +1223,14 @@ class AcpClient { this.handleSessionUpdate(notification) return } + if ( + event.type === "notification" && + event.method && + event.params && + this.referenceSearchSession?.handleNotification(event.method, event.params) + ) { + return + } if ( event.type === "notification" && (event.method === "workspace/changes/updated" || diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/reference-search-session.test.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/reference-search-session.test.ts new file mode 100644 index 00000000..db0d9452 --- /dev/null +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/reference-search-session.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, it, vi } from "vitest" + +import { ReferenceSearchSession } from "./reference-search-session" + +describe("ReferenceSearchSession", () => { + it("ignores stale search notifications for a different query", async () => { + const request = vi.fn(async (method: string, params: unknown) => { + if (method === "search/start") { + return { + snapshot: { + search_id: "search-1", + query: "src", + results: [], + total_file_match_count: 0, + scanned_file_count: 0, + file_search_complete: false, + }, + } + } + throw new Error(`unexpected request ${method}: ${JSON.stringify(params)}`) + }) + const session = new ReferenceSearchSession(request, "/workspace") + const snapshots: string[] = [] + session.subscribe((snapshot) => { + snapshots.push(snapshot.query) + }) + + await session.startOrUpdate("src") + const handled = session.handleNotification("search/updated", { + search_id: "search-1", + query: "old", + results: [], + total_file_match_count: 0, + scanned_file_count: 0, + file_search_complete: true, + }) + + expect(handled).toBe(true) + expect(snapshots).toEqual(["src"]) + }) + + it("prefers workspace-relative display_name over absolute file_path", async () => { + const request = vi.fn(async () => ({ + snapshot: { + search_id: "search-1", + query: "lib", + results: [ + { + kind: "file", + display_name: "src/lib.rs", + insert_text: "src/lib.rs", + file_path: "C:\\workspace\\src\\lib.rs", + }, + ], + total_file_match_count: 1, + scanned_file_count: 1, + file_search_complete: true, + }, + })) + const session = new ReferenceSearchSession(request, "C:\\workspace") + await session.startOrUpdate("lib") + + expect(session.filePaths()).toEqual(["src/lib.rs"]) + }) + + it("maps completed file results to paths", async () => { + const request = vi.fn(async () => ({ + snapshot: { + search_id: "search-1", + query: "lib", + results: [ + { + kind: "file", + display_name: "src/lib.rs", + insert_text: "src/lib.rs", + file_path: "src/lib.rs", + }, + ], + total_file_match_count: 1, + scanned_file_count: 1, + file_search_complete: false, + }, + })) + const session = new ReferenceSearchSession(request, "/workspace") + await session.startOrUpdate("lib") + session.handleNotification("search/completed", { + search_id: "search-1", + query: "lib", + results: [ + { + kind: "file", + display_name: "src/lib.rs", + insert_text: "src/lib.rs", + file_path: "src/lib.rs", + }, + ], + total_file_match_count: 1, + scanned_file_count: 1, + file_search_complete: true, + }) + + expect(session.filePaths()).toEqual(["src/lib.rs"]) + }) +}) diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/reference-search-session.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/reference-search-session.ts new file mode 100644 index 00000000..49680c8d --- /dev/null +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/reference-search-session.ts @@ -0,0 +1,181 @@ +/** + * Connection-local composer `@` reference search session. + * + * `@` mention file search uses `search/start`, `search/update`, and `search/cancel` + * RPCs plus `search/updated` / `search/completed` notifications — not a separate + * `find.files` endpoint. See L2-DES-APP-003 Search Protocol Rules. + * + * The server owns fuzzy ranking; clients render snapshots and ignore stale + * `query` / `search_id` pairs when the popup query has already moved on. + */ + +export type ReferenceSearchResultKind = "skill" | "mcp" | "file" + +export interface ReferenceSearchResult { + kind: ReferenceSearchResultKind + display_name: string + insert_text: string + description?: string + mention_path?: string + file_path?: string + is_disabled?: boolean + disabled_reason?: string +} + +export interface ReferenceSearchSnapshot { + search_id: string + query: string + results: ReferenceSearchResult[] + total_file_match_count: number + scanned_file_count: number + file_search_complete: boolean +} + +export interface ReferenceSearchFailedPayload { + search_id: string + query: string + message: string +} + +type SearchRequest = (method: string, params: unknown) => Promise + +type SnapshotListener = (snapshot: ReferenceSearchSnapshot) => void + +type ReferenceSearchState = { + snapshot: ReferenceSearchSnapshot | null + error: string | null +} + +function devoExtensionInnerMethod(method: string): string { + return method.startsWith("_devo/") ? method.slice("_devo/".length) : method +} + +function parseSnapshot(payload: unknown): ReferenceSearchSnapshot | null { + if (!payload || typeof payload !== "object") return null + if ("snapshot" in payload) { + const snapshot = (payload as { snapshot?: unknown }).snapshot + return parseSnapshot(snapshot) + } + const candidate = payload as Partial + if (typeof candidate.search_id !== "string" || typeof candidate.query !== "string") { + return null + } + return { + search_id: candidate.search_id, + query: candidate.query, + results: Array.isArray(candidate.results) ? (candidate.results as ReferenceSearchResult[]) : [], + total_file_match_count: candidate.total_file_match_count ?? 0, + scanned_file_count: candidate.scanned_file_count ?? 0, + file_search_complete: candidate.file_search_complete ?? false, + } +} + +function filePathsFromSnapshot(snapshot: ReferenceSearchSnapshot): string[] { + return snapshot.results + .filter((result) => result.kind === "file") + .map((result) => result.display_name) + .filter((path) => path.trim().length > 0) +} + +export class ReferenceSearchSession { + private searchId: string | null = null + private activeQuery = "" + private snapshot: ReferenceSearchSnapshot | null = null + private error: string | null = null + private readonly listeners = new Set() + + constructor( + private readonly request: SearchRequest, + private readonly cwd: string, + ) {} + + subscribe(listener: SnapshotListener): () => void { + this.listeners.add(listener) + if (this.snapshot) listener(this.snapshot) + return () => { + this.listeners.delete(listener) + } + } + + getState(): ReferenceSearchState { + return { snapshot: this.snapshot, error: this.error } + } + + filePaths(): string[] { + return this.snapshot ? filePathsFromSnapshot(this.snapshot) : [] + } + + async startOrUpdate(query: string): Promise { + this.activeQuery = query + this.error = null + const snapshot = this.searchId + ? await this.update(query) + : await this.start(query) + this.applySnapshot(snapshot) + return snapshot + } + + async cancel(): Promise { + const searchId = this.searchId + this.searchId = null + this.activeQuery = "" + this.snapshot = null + this.error = null + if (!searchId) return + try { + await this.request("search/cancel", { search_id: searchId }) + } catch { + // Best-effort cleanup when the popup closes. + } + } + + handleNotification(method: string, payload: unknown): boolean { + const innerMethod = devoExtensionInnerMethod(method) + if (innerMethod === "search/failed") { + const failed = payload as Partial + if (!failed.search_id || this.searchId !== failed.search_id) return true + if (failed.query && failed.query !== this.activeQuery) return true + this.error = failed.message ?? "reference search failed" + if (this.snapshot) { + this.applySnapshot({ + ...this.snapshot, + file_search_complete: true, + }) + } + return true + } + if (innerMethod !== "search/updated" && innerMethod !== "search/completed") { + return false + } + const snapshot = parseSnapshot(payload) + if (!snapshot) return true + if (this.searchId && snapshot.search_id !== this.searchId) return true + if (snapshot.query !== this.activeQuery) return true + this.applySnapshot(snapshot) + return true + } + + private async start(query: string): Promise { + const result = (await this.request("search/start", { + cwd: this.cwd, + query, + })) as { snapshot: ReferenceSearchSnapshot } + this.searchId = result.snapshot.search_id + return result.snapshot + } + + private async update(query: string): Promise { + const result = (await this.request("search/update", { + search_id: this.searchId, + query, + })) as { snapshot: ReferenceSearchSnapshot } + return result.snapshot + } + + private applySnapshot(snapshot: ReferenceSearchSnapshot): void { + this.snapshot = snapshot + for (const listener of this.listeners) { + listener(snapshot) + } + } +} diff --git a/apps/desktop/src/renderer/components/chat/mention-popover.tsx b/apps/desktop/src/renderer/components/chat/mention-popover.tsx index faf270e4..9d0df2c3 100644 --- a/apps/desktop/src/renderer/components/chat/mention-popover.tsx +++ b/apps/desktop/src/renderer/components/chat/mention-popover.tsx @@ -92,7 +92,7 @@ export const MentionPopover = memo( ) // --- Data: file search (enabled whenever popover is open, even with empty query) --- - const { files } = useFileSearch(directory, query, open) + const { files, isLoading, error } = useFileSearch(directory, query, open) const fileOptions = useMemo( () => files.slice(0, 20).map((f) => ({ type: "file" as const, path: f, display: f })), [files], @@ -174,6 +174,8 @@ export const MentionPopover = memo( const agentItems = allOptions.filter((o) => o.type === "agent") const fileItems = allOptions.filter((o) => o.type === "file") const hasResults = allOptions.length > 0 + const showLoading = isLoading && !hasResults + const showError = !!error && !hasResults && !isLoading let globalIndex = 0 @@ -196,7 +198,15 @@ export const MentionPopover = memo(
{!hasResults && (
- {query ? "No results found" : "No files or agents available"} + {showLoading + ? query + ? `Searching for "${query}"…` + : "Searching files and agents…" + : showError + ? error + : query + ? "No results found" + : "No files or agents available"}
)} diff --git a/apps/desktop/src/renderer/hooks/use-file-search.ts b/apps/desktop/src/renderer/hooks/use-file-search.ts index 11dfc3a9..3b3a69ad 100644 --- a/apps/desktop/src/renderer/hooks/use-file-search.ts +++ b/apps/desktop/src/renderer/hooks/use-file-search.ts @@ -1,16 +1,28 @@ /** - * Hook for searching files in the project via the Devo server. - * Provides debounced file search with caching. - * When query is empty, fetches an initial set of files. + * Hook for server-backed `@` reference file search in the composer. + * + * Uses connection-local `search/start` + `search/update` RPCs and listens for + * `search/updated` / `search/completed` notifications. Replaces the legacy + * `find.files` stub; debounce + cancel behavior matches the TUI composer. */ -import { useQuery } from "@tanstack/react-query" import { useEffect, useRef, useState } from "react" +import type { ReferenceSearchSnapshot } from "@devo-ai/sdk/v2/client" import { getProjectClient } from "../services/connection-manager" const FILE_SEARCH_DEBOUNCE_MS = 150 +function filePathsFromSnapshot(snapshot: ReferenceSearchSnapshot): string[] { + return snapshot.results + .filter((result) => result.kind === "file") + .map((result) => result.display_name) + .filter((path) => path.trim().length > 0) +} + export function useFileSearch(directory: string | null, query: string, enabled = true) { const [debouncedQuery, setDebouncedQuery] = useState(query) + const [files, setFiles] = useState([]) + const [isLoading, setIsLoading] = useState(false) + const [error, setError] = useState(null) const timerRef = useRef | null>(null) useEffect(() => { @@ -23,21 +35,50 @@ export function useFileSearch(directory: string | null, query: string, enabled = } }, [query]) - const { data, isLoading } = useQuery({ - queryKey: ["file-search", directory, debouncedQuery], - queryFn: async () => { - const client = getProjectClient(directory!) - if (!client) return [] - // Empty query returns initial/recent files from the server - const result = await client.find.files({ query: debouncedQuery }) - return (result.data ?? []) as string[] - }, - enabled: !!directory && enabled, - staleTime: 10_000, - }) + useEffect(() => { + if (!directory || !enabled) { + setFiles([]) + setIsLoading(false) + setError(null) + return + } + + const client = getProjectClient(directory) + if (!client) { + setFiles([]) + setIsLoading(false) + setError(null) + return + } + + let cancelled = false + const applySnapshot = (snapshot: ReferenceSearchSnapshot) => { + if (cancelled) return + setFiles(filePathsFromSnapshot(snapshot).slice(0, 20)) + setIsLoading(!snapshot.file_search_complete) + setError(client.referenceSearch.getState().error) + } + + const unsubscribe = client.referenceSearch.subscribe(applySnapshot) + setIsLoading(true) + setError(null) + void client.referenceSearch.startOrUpdate({ query: debouncedQuery }).catch((searchError) => { + if (cancelled) return + setFiles([]) + setIsLoading(false) + setError(searchError instanceof Error ? searchError.message : "file search failed") + }) + + return () => { + cancelled = true + unsubscribe() + void client.referenceSearch.cancel() + } + }, [directory, debouncedQuery, enabled]) return { - files: data ?? [], + files, isLoading, + error, } } diff --git a/apps/desktop/src/renderer/services/devo.ts b/apps/desktop/src/renderer/services/devo.ts index 5fcd8ca6..805ca9f2 100644 --- a/apps/desktop/src/renderer/services/devo.ts +++ b/apps/desktop/src/renderer/services/devo.ts @@ -1,5 +1,6 @@ import type { DevoClient, + ReferenceSearchSnapshot, WorkspaceChangeScope, WorkspaceChangesReadResult, WorkspaceDiffDetail, @@ -374,12 +375,19 @@ export async function listCommands( } /** - * Search for files in the project. - * Returns file paths as strings (from the Devo /find/file endpoint). + * Search for files in the project via server-backed reference search. + * Returns file paths from the active `search/*` session snapshot. */ export async function findFiles(client: DevoClient, query: string): Promise { - const result = await client.find.files({ query }) - return (result.data ?? []) as string[] + const result = await client.referenceSearch.startOrUpdate({ query }) + return filePathsFromReferenceSnapshot(result.data) +} + +function filePathsFromReferenceSnapshot(snapshot: ReferenceSearchSnapshot): string[] { + return snapshot.results + .filter((result) => result.kind === "file") + .map((result) => result.display_name) + .filter((path) => path.trim().length > 0) } /** diff --git a/apps/web/content/docs/using-devo/fuzzy-file-search.mdx b/apps/web/content/docs/using-devo/fuzzy-file-search.mdx index 4639f620..37d07328 100644 --- a/apps/web/content/docs/using-devo/fuzzy-file-search.mdx +++ b/apps/web/content/docs/using-devo/fuzzy-file-search.mdx @@ -90,3 +90,20 @@ you continue typing. If you press `Enter` while the popup has no selected result, Devo closes the popup instead of submitting the prompt. This prevents accidental submissions while reference search is still active. + +## How Search Stays In Sync + +Reference search is **server-backed** and **connection-local**: + +- The client opens a search session with `search/start` and updates it with + `search/update` as you type after `@`. +- The server returns an initial snapshot immediately, then streams file matches + through `search/updated` and finishes with `search/completed`. +- The popup stops showing a loading state when the latest snapshot sets + `file_search_complete` to true. +- These notifications belong to the requesting client connection. They are not + session transcript events and do not require a separate `events/subscribe` + filter to match a session id. + +If file search fails, clients should leave loading state and show a short +degraded message rather than waiting indefinitely. diff --git a/apps/web/content/docs/using-devo/fuzzy-file-search.zh.mdx b/apps/web/content/docs/using-devo/fuzzy-file-search.zh.mdx index ac40c11a..1bc3f9b6 100644 --- a/apps/web/content/docs/using-devo/fuzzy-file-search.zh.mdx +++ b/apps/web/content/docs/using-devo/fuzzy-file-search.zh.mdx @@ -78,3 +78,14 @@ Find where provider-hosted web search is configured. 按 `Esc` 后，Devo 会对当前 token dismiss popup，直到 token 改变才重新打开。这样 popup 不会在你继续输入时干扰你。 如果 popup 没有选中结果时按 `Enter`，Devo 会关闭 popup，而不是提交 prompt。这样可以避免 reference search 活跃时误提交。 + +## 搜索如何保持同步 + +`@` 引用搜索由 **服务端驱动**，且是 **连接级（connection-local）** 的： + +- 客户端通过 `search/start` 打开搜索会话，并在 `@` 后继续输入时用 `search/update` 更新查询。 +- 服务端会立即返回初始快照，随后通过 `search/updated` 流式推送文件匹配，并以 `search/completed` 结束。 +- 当最新快照的 `file_search_complete` 为 `true` 时，popup 会结束 loading 状态。 +- 这些通知只发给发起搜索的客户端连接，不属于会话 transcript，也不依赖 `events/subscribe` 的 session 过滤。 + +若文件搜索失败，客户端应退出 loading 并显示简短降级提示，而不是一直等待。 diff --git a/crates/core/src/history/compaction.rs b/crates/core/src/history/compaction.rs index 59b585ac..2651482a 100644 --- a/crates/core/src/history/compaction.rs +++ b/crates/core/src/history/compaction.rs @@ -1,19 +1,26 @@ //! Compaction — summarise conversation history via a separate LLM call //! when the token budget is exceeded. //! -//! Two compaction modes: +//! Two compaction modes (`CompactionKind`) choose different preserve strategies: //! -//! * **Auto** — triggered automatically when the context window is reached. -//! The operation is skipped if the history is already within budget. -//! * **Proactive** — explicitly requested by the user (e.g. via a `/compact` -//! command). The compaction always runs, regardless of the current budget. +//! * **Auto** — token-budget threshold in the query loop (`query.rs`). +//! Preserves a tail window of roughly [`COMPACT_USER_MESSAGE_MAX_TOKENS`] +//! estimated tokens via [`split_by_user_message_budget`], regardless of user +//! message boundaries. Example: `[user1, asst1, user2, asst2, user3]` may +//! become `[summary, asst2, user3]` when `asst2` and `user3` fit the tail +//! budget but `user2` does not. +//! * **Proactive** — `/compact` or provider `context_too_long` retry. +//! Preserves from the latest user message onward via +//! [`preserve_suffix_from_latest_user_message`]. Example: the same history +//! becomes `[summary, user3]` only. //! //! The compaction flow: //! //! 1. Filter out `Reason` items (reasoning text is not useful for summaries). -//! 2. Separate items into a "to‑compact" (old) and "to‑preserve" (recent) set -//! based on a user‑message token budget. -//! 3. Call the summarizer LLM with the `prompt.md` template. +//! 2. Separate items into a "to‑compact" prefix and "to‑preserve" suffix. +//! Auto uses a tail token budget; Proactive uses the latest-user suffix. +//! 3. Call the summarizer LLM with the `prompt.md` template appended as the +//! last developer message after the to-compact history. //! 4. Wrap the returned summary with the `summary_prefix.md` template. //! 5. Build a new history: `[summary_msg, …preserved_items]`. //! 6. If the summarizer LLM call fails with a context‑length error, move the @@ -33,11 +40,14 @@ use crate::response_item::ResponseItem; use devo_protocol::RequestContent; use devo_protocol::RequestMessage; +use devo_protocol::RequestRole; use super::TokenInfo; use super::normalize; const SUMMARIZATION_PROMPT: &str = include_str!("../../prompts/compact/prompt.md"); +/// Tail preserve budget for [`CompactionKind::Auto`]: walk backward from the +/// end of history and keep items until this estimated-token budget is full. const COMPACT_USER_MESSAGE_MAX_TOKENS: usize = 20_000; // --------------------------------------------------------------------------- @@ -77,8 +87,8 @@ pub enum CompactionError { /// this module does not depend directly on a specific provider SDK. #[async_trait] pub trait HistorySummarizer: Send + Sync { - /// Send `messages` (system prompt followed by the to‑compact history) - /// to the model and return the generated summary text. + /// Send `messages` (to-compact history followed by a developer compaction + /// prompt) to the model and return the generated summary text. async fn summarize(&self, messages: Vec) -> Result; } @@ -108,14 +118,24 @@ impl Default for CompactionConfig { // CompactionKind — how compaction was triggered // --------------------------------------------------------------------------- -/// Whether compaction was triggered automatically or proactively by the user. +/// Whether compaction was triggered automatically or proactively. +/// +/// Call-site mapping: +/// - [`CompactionKind::Auto`]: `query.rs` token-budget threshold before a turn. +/// - [`CompactionKind::Proactive`]: `/compact` (`server/.../compaction.rs`) and +/// provider `context_too_long` retry in `query.rs`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum CompactionKind { - /// Automatic compaction triggered when the context window is reached. - /// Skips if the history is already within budget. + /// Automatic compaction when context pressure is high. + /// + /// Skips when [`should_compact`] says the session is already within budget. + /// Preserve strategy: [`split_by_user_message_budget`] over the tail + /// [`COMPACT_USER_MESSAGE_MAX_TOKENS`] window (items, not user turns). Auto, - /// Proactive compaction explicitly requested by the user. - /// Always runs regardless of the current token budget. + /// Forced compaction that always runs. + /// + /// Preserve strategy: [`preserve_suffix_from_latest_user_message`] — from + /// the last `Role::User` item through the end of history. Proactive, } @@ -165,7 +185,9 @@ pub async fn compact_history( let mut filtered = normalize::filter_reason(items); normalize::pair_tool_call_items(&mut filtered); - // 2. Determine what to summarize vs what to preserve in the final history. + // 2. Pick preserve strategy from compaction kind. + // Auto: tail token window (may include assistant/tool items before the + // latest user). Proactive: suffix from the latest user message only. let (mut to_compact, mut preserved) = if config.kind == CompactionKind::Proactive { ( filtered.clone(), @@ -194,17 +216,7 @@ pub async fn compact_history( const MAX_TRANSIENT_RETRIES: u32 = 5; loop { - let mut messages = Vec::with_capacity(to_compact.len() + 1); - messages.push(RequestMessage { - role: "system".to_string(), - content: vec![RequestContent::Text { - text: SUMMARIZATION_PROMPT.trim().to_string(), - }], - }); - for item in &to_compact { - messages.push(RequestMessage::from(item)); - } - merge_consecutive_assistant_messages(&mut messages); + let messages = summarizer_request_messages(&to_compact); let summary = match summarizer.summarize(messages).await { Ok(s) => s, @@ -251,11 +263,19 @@ pub async fn compact_history( /// Splits items into a "to compact" prefix and a "to preserve" suffix. /// -/// Walks backward from the end, accumulating item token estimates until the -/// budget is exhausted. Everything before the split point is marked for -/// compaction; everything from the split point onward is preserved. -/// At least one item is always preserved (the last item) when items exist, -/// regardless of the budget. +/// Used by [`CompactionKind::Auto`]. Walks backward from the end, accumulating +/// per-item token estimates until `budget_tokens` would be exceeded. +/// +/// # Example +/// +/// History `[user1, asst1, user2, asst2(large), user3]` with a small budget +/// that fits only `asst2` and `user3`: +/// - `to_compact` = `[user1, asst1, user2]` +/// - `preserve` = `[asst2, user3]` +/// - result after compaction = `[summary, asst2, user3]` +/// +/// Items are [`ResponseItem`] records (messages, tool calls, tool outputs), not +/// whole turns. At least the last item is preserved when history is non-empty. fn split_by_user_message_budget( items: &[ResponseItem], budget_tokens: usize, @@ -292,6 +312,31 @@ fn split_by_user_message_budget( } } +fn summarizer_request_messages(to_compact: &[ResponseItem]) -> Vec { + let mut messages: Vec = to_compact.iter().map(RequestMessage::from).collect(); + merge_consecutive_assistant_messages(&mut messages); + messages.push(RequestMessage { + role: RequestRole::Developer.as_str().to_string(), + content: vec![RequestContent::Text { + text: SUMMARIZATION_PROMPT.trim().to_string(), + }], + }); + messages +} + +/// Preserves history from the latest user message through the end. +/// +/// Used by [`CompactionKind::Proactive`] (`/compact` and `context_too_long` +/// retry). Assistant and tool items after that user are kept; everything +/// before the user is summarized. +/// +/// # Example +/// +/// History `[user1, asst1, user2, asst2, user3]` always yields: +/// - `preserve` = `[user3]` +/// - result after compaction = `[summary, user3]` +/// +/// Returns an empty vector when no user message exists. fn preserve_suffix_from_latest_user_message(items: &[ResponseItem]) -> Vec { let Some(latest_user_index) = items.iter().rposition( |item| matches!(item, ResponseItem::Message(msg) if msg.role == devo_protocol::Role::User), @@ -415,47 +460,66 @@ mod tests { }, ]; - let mut messages = vec![RequestMessage { - role: "system".to_string(), - content: vec![RequestContent::Text { - text: "dummy prompt".into(), - }], - }]; - for item in &items { - messages.push(RequestMessage::from(item)); - } + let mut messages: Vec = items.iter().map(RequestMessage::from).collect(); merge_consecutive_assistant_messages(&mut messages); - // system, user, assistant(merged), user, user - assert_eq!(messages.len(), 5); - assert_eq!(messages[0].role, "system"); - - assert_eq!(messages[1].role, "user"); - assert_eq!(messages[1].content.len(), 1); + // user, assistant(merged), user, user + assert_eq!(messages.len(), 4); + assert_eq!(messages[0].role, "user"); + assert_eq!(messages[0].content.len(), 1); - assert_eq!(messages[2].role, "assistant"); - assert_eq!(messages[2].content.len(), 3); + assert_eq!(messages[1].role, "assistant"); + assert_eq!(messages[1].content.len(), 3); assert!(matches!( - &messages[2].content[0], + &messages[1].content[0], RequestContent::Text { .. } )); assert!( - matches!(&messages[2].content[1], RequestContent::ToolUse { id, .. } if id == "call-1") + matches!(&messages[1].content[1], RequestContent::ToolUse { id, .. } if id == "call-1") ); assert!( - matches!(&messages[2].content[2], RequestContent::ToolUse { id, .. } if id == "call-2") + matches!(&messages[1].content[2], RequestContent::ToolUse { id, .. } if id == "call-2") ); + assert_eq!(messages[2].role, "user"); + assert_eq!(messages[2].content.len(), 1); + assert!( + matches!(&messages[2].content[0], RequestContent::ToolResult { tool_use_id, .. } if tool_use_id == "call-1") + ); assert_eq!(messages[3].role, "user"); assert_eq!(messages[3].content.len(), 1); assert!( - matches!(&messages[3].content[0], RequestContent::ToolResult { tool_use_id, .. } if tool_use_id == "call-1") + matches!(&messages[3].content[0], RequestContent::ToolResult { tool_use_id, .. } if tool_use_id == "call-2") ); - assert_eq!(messages[4].role, "user"); - assert_eq!(messages[4].content.len(), 1); + } + + #[test] + fn compaction_summarizer_messages_put_prompt_last() { + let items = vec![ + ResponseItem::Message(Message::user("hello")), + ResponseItem::Message(Message::assistant_text("ok")), + ResponseItem::ToolCall { + id: "call-1".into(), + name: "read".into(), + input: serde_json::json!({"filePath": "/tmp/a.txt"}), + }, + ResponseItem::ToolCallOutput { + tool_use_id: "call-1".into(), + content: "content".into(), + is_error: false, + }, + ]; + + let messages = summarizer_request_messages(&items); + let last = messages + .last() + .expect("summarizer messages should not be empty"); + + assert_eq!(last.role, RequestRole::Developer.as_str()); assert!( - matches!(&messages[4].content[0], RequestContent::ToolResult { tool_use_id, .. } if tool_use_id == "call-2") + matches!(&last.content[0], RequestContent::Text { text } if text.contains("CONTEXT CHECKPOINT COMPACTION")) ); + assert_ne!(messages[0].role, RequestRole::Developer.as_str()); } #[test] @@ -493,6 +557,86 @@ mod tests { assert_eq!(preserve.len() + compact.len(), items.len()); } + #[tokio::test] + async fn auto_compaction_preserves_tail_by_token_budget_not_latest_user_only() { + struct StubSummarizer; + + #[async_trait] + impl HistorySummarizer for StubSummarizer { + async fn summarize( + &self, + _messages: Vec, + ) -> Result { + Ok("summary".to_string()) + } + } + + let large_tail = "x".repeat(40_000); + let items = vec![ + ResponseItem::Message(Message::user("old user")), + ResponseItem::Message(Message::assistant_text("old assistant")), + ResponseItem::Message(Message::user(large_tail.clone())), + ResponseItem::Message(Message::assistant_text(large_tail)), + ResponseItem::Message(Message::user("latest user")), + ]; + let token_info = TokenInfo { + input_tokens: 200, + cached_input_tokens: 0, + output_tokens: 0, + }; + let config = CompactionConfig { + budget: TokenBudget { + auto_compact_token_limit: Some(100), + ..TokenBudget::new(200_000, 8192) + }, + kind: CompactionKind::Auto, + }; + + let action = compact_history(&items, &token_info, &StubSummarizer, &config) + .await + .expect("auto compaction should succeed"); + + let proactive_config = CompactionConfig { + budget: config.budget.clone(), + kind: CompactionKind::Proactive, + }; + let proactive_action = + compact_history(&items, &token_info, &StubSummarizer, &proactive_config) + .await + .expect("proactive compaction should succeed"); + + match (action, proactive_action) { + ( + CompactAction::Replaced(auto_compacted), + CompactAction::Replaced(proactive_compacted), + ) => { + assert_eq!( + proactive_compacted[1..], + [ResponseItem::Message(Message::user("latest user"))] + ); + assert!( + auto_compacted.len() > proactive_compacted.len(), + "auto compaction should preserve more tail items than proactive latest-user suffix" + ); + assert_eq!( + auto_compacted.last(), + Some(&ResponseItem::Message(Message::user("latest user"))) + ); + assert!( + auto_compacted.iter().any(|item| { + matches!( + item, + ResponseItem::Message(msg) + if msg.role == devo_protocol::Role::Assistant + ) + }), + "auto compaction should preserve assistant tail items beyond the latest user message" + ); + } + _ => panic!("expected both compaction modes to replace history"), + } + } + #[tokio::test] async fn proactive_compaction_summarizes_all_history_and_preserves_latest_user_suffix() { struct StubSummarizer; @@ -504,6 +648,13 @@ mod tests { messages: Vec, ) -> Result { assert_eq!(messages.len(), 4); + let last = messages + .last() + .expect("summarizer messages should not be empty"); + assert_eq!(last.role, RequestRole::Developer.as_str()); + assert!( + matches!(&last.content[0], RequestContent::Text { text } if text.contains("CONTEXT CHECKPOINT COMPACTION")) + ); Ok("summary".to_string()) } } diff --git a/crates/core/src/query.rs b/crates/core/src/query.rs index 08bdab8b..1189f978 100644 --- a/crates/core/src/query.rs +++ b/crates/core/src/query.rs @@ -415,14 +415,17 @@ fn provider_retry_decision( /// Compact session messages using LLM-backed summarization. /// -/// Converts session messages to ResponseItems, runs compact_history() -/// with the history module's LLM summarizer, and converts the compacted -/// items back to Messages. +/// `kind` selects the preserve strategy inside [`compact_history`]: +/// - [`CompactionKind::Auto`]: preventive compaction when the session token +/// budget is high; keeps a tail token window. +/// - [`CompactionKind::Proactive`]: forced compaction after provider +/// `context_too_long`; keeps from the latest user message onward. async fn summarize_and_compact( session: &mut SessionState, provider: &Arc, request_model: &str, max_tokens: usize, + kind: CompactionKind, ) { let items: Vec = session .prompt_source_messages() @@ -439,7 +442,7 @@ async fn summarize_and_compact( let config = CompactionConfig { budget: session.config.token_budget.clone(), - kind: CompactionKind::Proactive, + kind, }; let summarizer = @@ -904,11 +907,14 @@ pub async fn query( budget_steer_injected = true; } info!("token budget threshold exceeded, running LLM compaction"); + // Auto: preserve tail items up to COMPACT_USER_MESSAGE_MAX_TOKENS. + // Example: [user1, asst1, user2, asst2, user3] -> [summary, asst2, user3]. summarize_and_compact( session, &provider, &turn_config.request_model, turn_config.model.max_tokens.unwrap_or(4096) as usize, + CompactionKind::Auto, ) .await; } @@ -1036,11 +1042,14 @@ pub async fn query( match provider_retry_decision(&e, &mut retry_count, &mut context_compacted) { ProviderRetryDecision::CompactAndRetry => { warn!("context_too_long - compacting and retrying"); + // Proactive: must compact even if token estimates disagree + // with the provider; preserve from latest user only. summarize_and_compact( session, &provider, &turn_config.request_model, turn_config.model.max_tokens.unwrap_or(4096) as usize, + CompactionKind::Proactive, ) .await; session.turn_count -= 1; @@ -1214,11 +1223,14 @@ pub async fn query( match provider_retry_decision(&e, &mut retry_count, &mut context_compacted) { ProviderRetryDecision::CompactAndRetry => { warn!("context_too_long - compacting and retrying"); + // Proactive: must compact even if token estimates disagree + // with the provider; preserve from latest user only. summarize_and_compact( session, &provider, &turn_config.request_model, turn_config.model.max_tokens.unwrap_or(4096) as usize, + CompactionKind::Proactive, ) .await; session.turn_count -= 1; diff --git a/crates/server/src/runtime/connection.rs b/crates/server/src/runtime/connection.rs index 11fb2795..7182bdfe 100644 --- a/crates/server/src/runtime/connection.rs +++ b/crates/server/src/runtime/connection.rs @@ -640,12 +640,52 @@ impl ServerRuntime { self.record_subagent_output_event(event).await; } + /// Deliver a connection-local notification to one client connection. + /// + /// Connection-local search notifications (`search/updated`, `search/completed`, + /// `search/failed`) are not session transcript events. They carry no + /// `session_id` and must reach the requesting connection even when the only + /// active subscriptions are session-scoped (`session_id=Some(...)`). + pub(super) async fn emit_connection_local_to_connection( + &self, + connection_id: u64, + method: &str, + event: ServerEvent, + ) { + debug_assert!(is_connection_local_notification(method)); + let notification = { + let mut connections = self.connections.lock().await; + let Some(connection) = connections.get_mut(&connection_id) else { + return; + }; + if !connection.should_deliver_connection_local(method) { + return; + } + let event_seq = connection.next_seq(); + let event = event.with_seq(event_seq); + let (method, value) = acp_notification_from_server_event(method, &event); + Some(( + connection.outbound_tx.clone(), + OutboundFrame::notification(connection_id, method, event_seq, value), + )) + }; + if let Some((outbound_tx, frame)) = notification { + let _ = enqueue_outbound_notification(&outbound_tx, frame, "connection_notifications") + .await; + } + } + pub(super) async fn emit_to_connection( &self, connection_id: u64, method: &str, event: ServerEvent, ) { + if is_connection_local_notification(method) { + self.emit_connection_local_to_connection(connection_id, method, event) + .await; + return; + } let session_id = event.session_id(); let child_parent_by_session = self.child_parent_by_session().await; let notification = { @@ -1096,7 +1136,23 @@ pub(crate) struct ConnectionRuntime { pending_client_requests: HashMap>>, } +/// Returns whether `method` is a connection-local composer notification. +/// +/// These events are scoped to the requesting client connection, not to a +/// durable session subscription. +pub(super) fn is_connection_local_notification(method: &str) -> bool { + matches!( + method, + "search/updated" | "search/completed" | "search/failed" + ) +} + impl ConnectionRuntime { + /// Connection-local notifications bypass session subscription filters. + pub(super) fn should_deliver_connection_local(&self, method: &str) -> bool { + !self.opt_out_notification_methods.contains(method) + } + pub(super) fn should_deliver( &self, method: &str, @@ -1612,4 +1668,73 @@ mod tests { Ok(()) } + + /// Trace: L2-DES-CLIENT-002 + /// Verifies: connection-local search notifications deliver despite session-only subscriptions. + #[tokio::test] + async fn connection_local_search_notifications_ignore_session_subscriptions() -> Result<()> { + use devo_protocol::ReferenceSearchId; + use devo_protocol::ReferenceSearchSnapshot; + + let data_root = TempDir::new()?; + let runtime = build_runtime(data_root.path()); + let session_id = SessionId::new(); + let (owner_outbound, mut owner_receiver) = super::outbound::test_outbound_channel(4); + let owner_connection_id = runtime + .register_connection(ClientTransportKind::Stdio, owner_outbound) + .await; + let (other_outbound, mut other_receiver) = super::outbound::test_outbound_channel(4); + let _other_connection_id = runtime + .register_connection(ClientTransportKind::Stdio, other_outbound) + .await; + + runtime + .subscribe_connection_to_session(owner_connection_id, session_id, None) + .await; + + let snapshot = ReferenceSearchSnapshot { + search_id: ReferenceSearchId::new(), + query: "src".to_string(), + results: Vec::new(), + total_file_match_count: 0, + scanned_file_count: 0, + file_search_complete: true, + }; + runtime + .emit_connection_local_to_connection( + owner_connection_id, + "search/completed", + ServerEvent::ReferenceSearchCompleted(snapshot), + ) + .await; + + let owner_message = tokio::time::timeout(Duration::from_secs(1), owner_receiver.recv()) + .await? + .expect("owner receives connection-local search notification"); + assert_eq!(owner_message["method"], "_devo/search/completed"); + assert!( + tokio::time::timeout(Duration::from_millis(50), other_receiver.recv()) + .await + .is_err(), + "unrelated connection must not receive connection-local search notification" + ); + + Ok(()) + } + + /// Trace: L2-DES-CLIENT-002 + /// Verifies: session-scoped notifications still require matching subscriptions. + #[test] + fn session_scoped_notifications_require_matching_subscription() { + let subscribed_session = SessionId::new(); + let subscription = SubscriptionFilter { + session_id: Some(subscribed_session), + event_types: HashSet::new(), + include_child_agents: false, + }; + let child_parent_by_session = HashMap::new(); + + assert!(!subscription.session_matches(None, &child_parent_by_session)); + assert!(subscription.session_matches(Some(subscribed_session), &child_parent_by_session)); + } } diff --git a/crates/server/src/runtime/handlers/compaction.rs b/crates/server/src/runtime/handlers/compaction.rs index 8c2527b9..687e7441 100644 --- a/crates/server/src/runtime/handlers/compaction.rs +++ b/crates/server/src/runtime/handlers/compaction.rs @@ -147,6 +147,8 @@ impl ServerRuntime { let config = CompactionConfig { budget: core_session.config.token_budget.clone(), + // Proactive: user-requested /compact; preserve latest user suffix. + // Example: [user1, asst1, user2, asst2, user3] -> [summary, user3]. kind: CompactionKind::Proactive, }; diff --git a/crates/server/src/runtime/reference_search.rs b/crates/server/src/runtime/reference_search.rs index 57e09b41..7cbc4d34 100644 --- a/crates/server/src/runtime/reference_search.rs +++ b/crates/server/src/runtime/reference_search.rs @@ -1,7 +1,9 @@ -//! Server-owned reference search sessions for TUI composer `@` lookups. +//! Server-owned reference search sessions for composer `@` lookups. //! //! The runtime aggregates skill metadata, configured MCP servers, and live file //! search snapshots into protocol rows so UI clients only render and select. +//! Incremental file results are pushed to the requesting client through +//! connection-local `search/updated` and `search/completed` notifications. use std::num::NonZero; use std::path::PathBuf; @@ -207,7 +209,7 @@ impl ServerRuntime { file_matches: Vec::new(), total_file_match_count: 0, scanned_file_count: 0, - file_search_complete: true, + file_search_complete: false, file_session, }; let snapshot = state.snapshot(&search_id); @@ -219,9 +221,7 @@ impl ServerRuntime { self.spawn_reference_search_update_task(connection_id, update_rx); - if !snapshot.query.trim().is_empty() - && let Some(state) = self.reference_searches.lock().await.get_mut(&search_id) - { + if let Some(state) = self.reference_searches.lock().await.get_mut(&search_id) { state.file_search_complete = false; state.file_session.update_query(&snapshot.query); return Ok(state.snapshot(&search_id)); @@ -247,12 +247,8 @@ impl ServerRuntime { state.file_matches.clear(); state.total_file_match_count = 0; state.scanned_file_count = 0; - if state.query.trim().is_empty() { - state.file_search_complete = true; - } else { - state.file_search_complete = false; - state.file_session.update_query(&state.query); - } + state.file_search_complete = false; + state.file_session.update_query(&state.query); Ok(state.snapshot(¶ms.search_id)) } @@ -309,7 +305,8 @@ impl ServerRuntime { ServerEvent::ReferenceSearchUpdated(snapshot), ) }; - self.emit_to_connection(connection_id, method, event).await; + self.emit_connection_local_to_connection(connection_id, method, event) + .await; } fn mcp_sources(runtime_context: &SessionRuntimeContext) -> Vec { diff --git a/crates/tui/src/worker.rs b/crates/tui/src/worker.rs index a9c94e75..e3019fc2 100644 --- a/crates/tui/src/worker.rs +++ b/crates/tui/src/worker.rs @@ -38,6 +38,7 @@ use devo_protocol::ProviderVendorListParams; use devo_protocol::ProviderVendorUpsertParams; use devo_protocol::ReferenceSearchCancelParams; use devo_protocol::ReferenceSearchId; +use devo_protocol::ReferenceSearchSnapshot; use devo_protocol::ReferenceSearchStartParams; use devo_protocol::ReferenceSearchUpdateParams; use devo_protocol::SessionHistoryMetadata; @@ -2793,6 +2794,19 @@ async fn run_worker_inner( message = %payload.message, "reference search failed" ); + // End the composer loading state instead of waiting forever + // for a completion notification that will never arrive. + let snapshot = ReferenceSearchSnapshot { + search_id: payload.search_id, + query: payload.query, + results: Vec::new(), + total_file_match_count: 0, + scanned_file_count: 0, + file_search_complete: true, + }; + let _ = event_tx.send(WorkerEvent::ReferenceSearchUpdated { + snapshot, + }); } } "session/title/updated" => { diff --git a/specs/L2/app/L2-DES-APP-003-client-server-protocol.md b/specs/L2/app/L2-DES-APP-003-client-server-protocol.md index 4d1109a5..ec682fd3 100644 --- a/specs/L2/app/L2-DES-APP-003-client-server-protocol.md +++ b/specs/L2/app/L2-DES-APP-003-client-server-protocol.md @@ -386,6 +386,7 @@ Rules: - `search/start` must return promptly with a server-owned `search_id` or a structured rejection. - `search/updated`, `search/completed`, and `search/failed` are sent only to the requesting client connection unless a later requirement defines shared picker state. +- Connection-local search notifications must be delivered to the requesting connection even when the only active notification subscriptions are session-scoped (`session_id=Some(...)`). Search snapshots carry no `session_id` and must not depend on `events/subscribe` session matching. - `search/update` should supersede prior query work for the same `search_id`. - Search query text is the literal keyword after the client prefix. It must not require or encode provider selectors such as file, skill, or MCP. - If `providers` is omitted, the server should use the default provider set for prefixed input.