diff --git a/Cargo.lock b/Cargo.lock index 341354c..5639b13 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1128,7 +1128,7 @@ checksum = "52560adf09603e58c9a7ee1fe1dcb95a16927b17c127f0ac02d6e768a0e25bc1" [[package]] name = "cursor-byok-desktop" -version = "0.1.0-beta.2" +version = "0.1.0-beta.3" dependencies = [ "axum", "cursor-server", diff --git a/apps/desktop/package-lock.json b/apps/desktop/package-lock.json index 042fb05..5fafc70 100644 --- a/apps/desktop/package-lock.json +++ b/apps/desktop/package-lock.json @@ -1,12 +1,12 @@ { "name": "cursor-byok-desktop", - "version": "0.1.0-beta.2", + "version": "0.1.0-beta.3", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "cursor-byok-desktop", - "version": "0.1.0-beta.2", + "version": "0.1.0-beta.3", "license": "MIT", "dependencies": { "@floating-ui/dom": "^1.8.0", diff --git a/apps/desktop/package.json b/apps/desktop/package.json index 59e438d..c2d88e9 100644 --- a/apps/desktop/package.json +++ b/apps/desktop/package.json @@ -1,6 +1,6 @@ { "name": "cursor-byok-desktop", - "version": "0.1.0-beta.2", + "version": "0.1.0-beta.3", "description": "Cursor BYOK desktop management application", "type": "module", "scripts": { diff --git a/apps/desktop/src-tauri/Cargo.toml b/apps/desktop/src-tauri/Cargo.toml index b7c770f..c145b08 100644 --- a/apps/desktop/src-tauri/Cargo.toml +++ b/apps/desktop/src-tauri/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cursor-byok-desktop" -version = "0.1.0-beta.2" +version = "0.1.0-beta.3" edition = "2021" publish = false diff --git a/apps/desktop/src-tauri/tauri.conf.json b/apps/desktop/src-tauri/tauri.conf.json index 7d36928..30f012e 100644 --- a/apps/desktop/src-tauri/tauri.conf.json +++ b/apps/desktop/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "Cursor BYOK", - "version": "0.1.0-beta.2", + "version": "0.1.0-beta.3", "identifier": "dev.cursorbyok.desktop", "build": { "beforeDevCommand": "npm run dev", diff --git a/apps/desktop/src/App.tsx b/apps/desktop/src/App.tsx index 4777323..3f39fec 100644 --- a/apps/desktop/src/App.tsx +++ b/apps/desktop/src/App.tsx @@ -10,8 +10,8 @@ import { CursorSettingsPage } from "./pages/CursorSettingsPage"; import { HomePage } from "./pages/HomePage"; import { ProvidersPage } from "./pages/ProvidersPage"; import { SettingsPage } from "./pages/SettingsPage"; -import { checkForUpdate, hasNativeAppLifecycle } from "./native/appLifecycle"; import { useAppStore } from "./store/appStore"; +import { updateStore } from "./store/updateStore"; export function App() { return ( @@ -47,11 +47,9 @@ function AppMessages() { }, [error, showMessage]); useEffect(() => { - if (!hasNativeAppLifecycle()) return; - void checkForUpdate().then(async (update) => { - if (!update) return; - showMessage(t("发现新版本 {version}，可在设置中安装", { version: update.version }), { duration: 6_000 }); - await update.close(); + void updateStore.check().then((version) => { + if (!version) return; + showMessage(t("发现新版本 {version}，可在设置中安装", { version }), { duration: 6_000 }); }).catch(() => { // Startup checks are best-effort; manual checks in Settings report errors. }); diff --git a/apps/desktop/src/components/settings/AppLifecycleSettingsCard.module.scss b/apps/desktop/src/components/settings/AppLifecycleSettingsCard.module.scss index 6ca3e7f..29c20a8 100644 --- a/apps/desktop/src/components/settings/AppLifecycleSettingsCard.module.scss +++ b/apps/desktop/src/components/settings/AppLifecycleSettingsCard.module.scss @@ -21,3 +21,11 @@ font-size: type.$font-size-xs; } } + +.updateDot { + width: 7px; + height: 7px; + flex: 0 0 7px; + background: #ef4444; + border-radius: 50%; +} diff --git a/apps/desktop/src/components/settings/AppLifecycleSettingsCard.tsx b/apps/desktop/src/components/settings/AppLifecycleSettingsCard.tsx index 677d8ff..e255a23 100644 --- a/apps/desktop/src/components/settings/AppLifecycleSettingsCard.tsx +++ b/apps/desktop/src/components/settings/AppLifecycleSettingsCard.tsx @@ -1,13 +1,11 @@ -import { useEffect, useRef, useState } from "react"; -import type { Update } from "@tauri-apps/plugin-updater"; +import { useEffect, useState } from "react"; import { - checkForUpdate, currentAppVersion, hasNativeAppLifecycle, - installUpdate, readAutostart, writeAutostart, } from "../../native/appLifecycle"; +import { updateStore, useUpdateStore } from "../../store/updateStore"; import { Button } from "../ui/Button"; import { Switch } from "../ui/Switch"; import { TitledCard } from "../ui/TitledCard"; @@ -17,13 +15,10 @@ import styles from "./AppLifecycleSettingsCard.module.scss"; export function AppLifecycleSettingsCard() { const message = useMessage(); const native = hasNativeAppLifecycle(); - const updateRef = useRef(null); + const { availableVersion, checking, installing } = useUpdateStore(); const [version, setVersion] = useState("…"); const [autostart, setAutostart] = useState(false); const [loadingAutostart, setLoadingAutostart] = useState(native); - const [checking, setChecking] = useState(false); - const [installing, setInstalling] = useState(false); - const [availableVersion, setAvailableVersion] = useState(null); useEffect(() => { let disposed = false; @@ -34,12 +29,7 @@ export function AppLifecycleSettingsCard() { .catch((cause) => message(cause instanceof Error ? cause.message : String(cause))) .finally(() => { if (!disposed) setLoadingAutostart(false); }); } - return () => { - disposed = true; - const update = updateRef.current; - updateRef.current = null; - if (update) void update.close(); - }; + return () => { disposed = true; }; }, [message, native]); const toggleAutostart = async (enabled: boolean) => { @@ -57,29 +47,17 @@ export function AppLifecycleSettingsCard() { const checkUpdate = async () => { try { - setChecking(true); - const previous = updateRef.current; - updateRef.current = null; - if (previous) await previous.close(); - const update = await checkForUpdate(); - updateRef.current = update; - setAvailableVersion(update?.version ?? null); - message(update ? t("发现新版本 {version}", { version: update.version }) : t("当前已是最新版本")); + const nextVersion = await updateStore.check(); + message(nextVersion ? t("发现新版本 {version}", { version: nextVersion }) : t("当前已是最新版本")); } catch (cause) { message(cause instanceof Error ? cause.message : String(cause)); - } finally { - setChecking(false); } }; const updateNow = async () => { - const update = updateRef.current; - if (!update) return; try { - setInstalling(true); - await installUpdate(update); + await updateStore.install(); } catch (cause) { - setInstalling(false); message(cause instanceof Error ? cause.message : String(cause)); } }; @@ -107,6 +85,7 @@ export function AppLifecycleSettingsCard() { {availableVersion ? : :
@@ -207,6 +209,7 @@ export function AppLayout() { ? : } {item.label} + {item.path === "/settings" && availableVersion &&
} diff --git a/apps/desktop/src/store/updateStore.ts b/apps/desktop/src/store/updateStore.ts new file mode 100644 index 0000000..a5dfed9 --- /dev/null +++ b/apps/desktop/src/store/updateStore.ts @@ -0,0 +1,74 @@ +import { useSyncExternalStore } from "react"; +import type { Update } from "@tauri-apps/plugin-updater"; +import { + checkForUpdate, + hasNativeAppLifecycle, + installUpdate, +} from "../native/appLifecycle"; + +export type UpdateSnapshot = { + availableVersion: string | null; + checking: boolean; + installing: boolean; +}; + +let snapshot: UpdateSnapshot = { + availableVersion: null, + checking: false, + installing: false, +}; +let availableUpdate: Update | null = null; +let pendingCheck: Promise | null = null; +const listeners = new Set<() => void>(); + +function update(patch: Partial) { + snapshot = { ...snapshot, ...patch }; + listeners.forEach((listener) => listener()); +} + +async function replaceAvailableUpdate(next: Update | null) { + const previous = availableUpdate; + availableUpdate = next; + update({ availableVersion: next?.version ?? null }); + if (previous && previous !== next) await previous.close(); +} + +export const updateStore = { + subscribe(listener: () => void) { + listeners.add(listener); + return () => listeners.delete(listener); + }, + getSnapshot: () => snapshot, + + async check(): Promise { + if (!hasNativeAppLifecycle()) return null; + if (pendingCheck) return pendingCheck; + update({ checking: true }); + pendingCheck = (async () => { + const next = await checkForUpdate(); + await replaceAvailableUpdate(next); + return next?.version ?? null; + })(); + try { + return await pendingCheck; + } finally { + pendingCheck = null; + update({ checking: false }); + } + }, + + async install(): Promise { + const current = availableUpdate; + if (!current) return; + update({ installing: true }); + try { + await installUpdate(current); + } finally { + update({ installing: false }); + } + }, +}; + +export function useUpdateStore(): UpdateSnapshot { + return useSyncExternalStore(updateStore.subscribe, updateStore.getSnapshot); +} diff --git a/server/src/cursor/actor.rs b/server/src/cursor/actor.rs index 590c669..df2a2cc 100644 --- a/server/src/cursor/actor.rs +++ b/server/src/cursor/actor.rs @@ -8,6 +8,7 @@ use crate::{ blob_sync::BlobSynchronizer, checkpoint::CheckpointBuilder, context_sync::RequestContextSynchronizer, + interaction, proto::agent::v1 as pb, request, session::CursorSession, @@ -272,6 +273,9 @@ impl CursorActor { )) => { let _ = blob_sync.handle_client(message).await; } + Some(pb::agent_client_message::Message::ClientHeartbeat(_)) => { + let _ = handle.emit(&interaction::heartbeat()); + } // TODO: ConversationAction has two different delivery paths that // must not be conflated: // diff --git a/server/src/cursor/interaction/mod.rs b/server/src/cursor/interaction/mod.rs index bac8dfc..b2e6d7d 100644 --- a/server/src/cursor/interaction/mod.rs +++ b/server/src/cursor/interaction/mod.rs @@ -66,6 +66,12 @@ pub fn thinking_completed(elapsed: Duration) -> pb::AgentServerMessage { )) } +pub fn heartbeat() -> pb::AgentServerMessage { + server_interaction(pb::interaction_update::Message::Heartbeat( + pb::HeartbeatUpdate {}, + )) +} + pub fn arguments_delta(call: &ToolCall, delta: &str) -> Result { Ok(server_interaction( pb::interaction_update::Message::PartialToolCall(pb::PartialToolCallUpdate { diff --git a/server/src/cursor/run_sse.rs b/server/src/cursor/run_sse.rs index 16e6991..2abbda8 100644 --- a/server/src/cursor/run_sse.rs +++ b/server/src/cursor/run_sse.rs @@ -3,29 +3,24 @@ use axum::{ http::{header, HeaderValue, Response, StatusCode}, }; use bytes::Bytes; +use std::convert::Infallible; use tokio::sync::mpsc; use tokio_stream::StreamExt; +use tokio_util::sync::CancellationToken; use crate::{ - cursor::{observability::CursorTraceRecorder, CursorSessionRegistry}, + cursor::{connect::END_STREAM_FLAG, observability::CursorTraceRecorder, CursorSessionRegistry}, Result, }; pub async fn stream(registry: &CursorSessionRegistry, request_id: &str) -> Result> { let handle = registry.get_or_create(request_id).await?; - let mut receiver = handle.subscribe(); + let receiver = handle.subscribe(); let trace = handle.trace().cloned(); if let Some(trace) = &trace { trace.response_started(StatusCode::OK.as_u16()).await; } - let body_stream = async_stream::stream! { - let mut trace = TraceStreamSink::new(trace, "byok_server"); - while let Some(chunk) = receiver.recv().await { - trace.chunk(&chunk); - yield Ok::(chunk); - } - trace.finish(None); - }; + let body_stream = local_body_stream(receiver, handle.cancellation(), trace); let mut response = Response::new(Body::from_stream(body_stream)); *response.status_mut() = StatusCode::OK; response.headers_mut().insert( @@ -41,6 +36,63 @@ pub async fn stream(registry: &CursorSessionRegistry, request_id: &str) -> Resul Ok(response) } +fn local_body_stream( + mut receiver: mpsc::UnboundedReceiver, + cancellation: CancellationToken, + trace: Option, +) -> impl tokio_stream::Stream> { + async_stream::stream! { + let mut guard = LocalRunGuard::new(cancellation); + let mut trace = TraceStreamSink::new(trace, "byok_server"); + while let Some(chunk) = receiver.recv().await { + let terminal = is_end_stream_frame(&chunk); + trace.chunk(&chunk); + if terminal { + guard.complete(); + trace.finish(None); + } + yield Ok::(chunk); + if terminal { + return; + } + } + guard.complete(); + trace.finish(None); + } +} + +fn is_end_stream_frame(frame: &Bytes) -> bool { + frame + .first() + .is_some_and(|flags| flags & END_STREAM_FLAG != 0) +} + +struct LocalRunGuard { + cancellation: CancellationToken, + completed: bool, +} + +impl LocalRunGuard { + fn new(cancellation: CancellationToken) -> Self { + Self { + cancellation, + completed: false, + } + } + + fn complete(&mut self) { + self.completed = true; + } +} + +impl Drop for LocalRunGuard { + fn drop(&mut self) { + if !self.completed { + self.cancellation.cancel(); + } + } +} + pub async fn upstream( registry: CursorSessionRegistry, request_id: String, @@ -127,7 +179,11 @@ impl TraceStreamSink { impl Drop for TraceStreamSink { fn drop(&mut self) { - self.finish(None); + if self.sender.is_some() { + self.finish(Some( + "response stream dropped before completion".to_string(), + )); + } } } @@ -143,3 +199,38 @@ impl Drop for UpstreamRunGuard { .finish_upstream(self.request_id.clone(), self.generation); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::cursor::{connect, proto::agent::v1 as pb}; + + #[tokio::test] + async fn local_stream_cancels_when_the_client_disconnects() { + let (sender, receiver) = mpsc::unbounded_channel(); + let cancellation = CancellationToken::new(); + sender + .send(connect::encode_message(&pb::AgentServerMessage::default()).unwrap()) + .unwrap(); + let mut stream = Box::pin(local_body_stream(receiver, cancellation.clone(), None)); + + stream.next().await.unwrap().unwrap(); + + drop(sender); + drop(stream); + assert!(cancellation.is_cancelled()); + } + + #[tokio::test] + async fn terminal_frame_does_not_cancel_a_completed_local_run() { + let (sender, receiver) = mpsc::unbounded_channel(); + let cancellation = CancellationToken::new(); + sender.send(connect::encode_end_stream()).unwrap(); + let mut stream = Box::pin(local_body_stream(receiver, cancellation.clone(), None)); + + let terminal = stream.next().await.unwrap().unwrap(); + assert!(is_end_stream_frame(&terminal)); + drop(stream); + assert!(!cancellation.is_cancelled()); + } +} diff --git a/server/src/provider/openai_responses.rs b/server/src/provider/openai_responses.rs index c656ba6..11c7a9d 100644 --- a/server/src/provider/openai_responses.rs +++ b/server/src/provider/openai_responses.rs @@ -140,11 +140,11 @@ impl Provider for OpenAiResponsesProvider { } if text_open { text_open = false; yield ModelEvent::TextEnd; } } - "response.reasoning_summary_text.delta" => { + "response.reasoning_summary_text.delta" | "response.reasoning_text.delta" => { if !thinking_open { thinking_open = true; yield ModelEvent::ThinkingStart; } if let Some(delta) = value.get("delta").and_then(Value::as_str) { yield ModelEvent::ThinkingDelta(delta.into()); } } - "response.reasoning_summary_text.done" => { + "response.reasoning_summary_text.done" | "response.reasoning_text.done" => { if thinking_open { thinking_open = false; yield ModelEvent::ThinkingEnd; } } "response.output_item.added" => { diff --git a/server/tests/interrupt.rs b/server/tests/interrupt.rs index 7c72440..2420562 100644 --- a/server/tests/interrupt.rs +++ b/server/tests/interrupt.rs @@ -146,6 +146,54 @@ async fn registry_shutdown_cancels_runs_and_closes_run_sse_outputs() { assert_eq!(output.recv().await, None); } +#[tokio::test] +async fn client_heartbeat_returns_a_server_protocol_heartbeat() { + let (_directory, store) = fixtures::temp_store().await; + let assets = PromptAssets::load( + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("prompt/cursor") + .as_path(), + ) + .unwrap(); + let registry = CursorSessionRegistry::new( + store, + Arc::new(fake_provider::FakeProvider::default()), + PromptCompiler::new(assets), + Default::default(), + ); + let handle = registry.get_or_create("heartbeat-run").await.unwrap(); + let mut output = handle.subscribe(); + + handle + .command(CursorCommand::Append { + seqno: 0, + message: Box::new(pb::AgentClientMessage { + message: Some(pb::agent_client_message::Message::ClientHeartbeat( + pb::ClientHeartbeat {}, + )), + }), + }) + .await + .unwrap(); + + let frame = tokio::time::timeout(std::time::Duration::from_secs(1), output.recv()) + .await + .unwrap() + .unwrap(); + let (_, payload) = connect::decode_frames(&frame).unwrap().pop().unwrap(); + let message = pb::AgentServerMessage::decode(payload).unwrap(); + assert!(matches!( + message.message, + Some(pb::agent_server_message::Message::InteractionUpdate( + pb::InteractionUpdate { + message: Some(pb::interaction_update::Message::Heartbeat(_)), + } + )) + )); + + registry.shutdown().await; +} + #[tokio::test] async fn runtime_user_message_action_aborts_active_exec_before_canceled_end_stream() { let (_directory, store) = fixtures::temp_store().await; diff --git a/server/tests/provider_stream.rs b/server/tests/provider_stream.rs index 86dc076..e428f52 100644 --- a/server/tests/provider_stream.rs +++ b/server/tests/provider_stream.rs @@ -329,6 +329,37 @@ async fn openai_responses_raw_stream_does_not_invent_reasoning_effort() { assert_eq!(replayed, ["opaque-1", "opaque-2"]); } +#[tokio::test] +async fn openai_responses_streams_openrouter_reasoning_text_events() { + let (base_url, _requests, server) = fixture_server( + "/v1/responses", + concat!( + "data: {\"type\":\"response.reasoning_text.delta\",\"delta\":\"still working\"}\n\n", + "data: {\"type\":\"response.reasoning_text.done\"}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{}}\n\n", + ), + ) + .await; + let provider = OpenAiResponsesProvider::new( + reqwest::Client::new(), + config(ProviderKind::OpenAiResponses, base_url, None), + ); + + let events = collect(provider.stream(invocation(), CancellationToken::new())).await; + server.abort(); + + assert!(events + .iter() + .any(|event| matches!(event, ModelEvent::ThinkingStart))); + assert!(events.iter().any( + |event| matches!(event, ModelEvent::ThinkingDelta(delta) if delta == "still working") + )); + assert!(events + .iter() + .any(|event| matches!(event, ModelEvent::ThinkingEnd))); + assert_eq!(events.last(), Some(&ModelEvent::Done(FinishReason::Stop))); +} + #[tokio::test] async fn openai_responses_reasoning_item_done_closes_an_open_summary() { let (base_url, _requests, server) = fixture_server(