From 960cdc784edfa0d112136dbe7728a9d5f3cc6f25 Mon Sep 17 00:00:00 2001 From: Raing5Days Date: Thu, 5 Nov 2026 00:33:49 +0800 Subject: [PATCH 1/2] fix: keep host dialog queue polls and workspace state quiet when a session is not loaded Fixes #3729. Switching sessions/workspaces (especially local <-> remote) left passive observers polling manage_dialog_queue for a session that was never restored on the execution host, producing an ERROR log roughly every 10s ("Session is not loaded on the execution host"). Three changes: - DialogScheduler::execute_queue_request now lets read-only actions (List, Get) answer from the scheduler-owned receipts when the session is not loaded on this host, instead of failing every poll. Mutations (Submit, Cancel, Promote) still refuse user work for an unloaded session so nothing can silently disappear. - WorkspaceManager.setLoading/setError no longer emit state-quiet events; repeating an unchanged value made listeners observe workspace:loading followed by workspace:error in the same tick, which looked like a load that failed instantly. - The transport adapter classifies a structured host_unavailable answer from get_external_source_* polls as expected (debug-level) instead of logging each poll as ERROR; ApiClient reuses the same classifier. Verification: cargo test -p openbitfun-core (host_queue_ filters, 20 passed incl. new degradation test); web-ui vitest workspaceManager + tauri-adapter + HostPendingQueuePanel (49 passed); web-ui type-check; pnpm run fmt:rs. AI-assisted. --- .../coordination/host_message_queue.rs | 13 +++- .../coordination/host_message_queue_tests.rs | 71 +++++++++++++++++++ .../api/adapters/tauri-adapter.test.ts | 25 +++++++ .../api/adapters/tauri-adapter.ts | 47 ++++++++++++ .../api/service-api/ApiClient.ts | 14 ++-- .../business/workspaceManager.test.ts | 61 ++++++++++++++++ .../services/business/workspaceManager.ts | 6 ++ 7 files changed, 232 insertions(+), 5 deletions(-) diff --git a/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs b/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs index f8343f6f22..4dcd5f723e 100644 --- a/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs +++ b/src/crates/assembly/core/src/agentic/coordination/host_message_queue.rs @@ -364,7 +364,18 @@ impl DialogScheduler { let session = &request.session_id; openbitfun_core_types::validate_session_id(session).map_err(error)?; let _admission = self.host_queue_locks.lock(session).await; - if self.session_manager.get_session(session).is_none() { + // Passive observers (queue panels, timeline projection) poll a session + // that may not be restored on this execution host yet, e.g. right after + // a local <-> remote workspace switch. Reads may degrade: the scheduler + // keeps its own receipts, so report that state without failing every + // poll. Mutations keep failing — accepting user work against a session + // this runtime cannot vouch for would silently lose it. + let session_unloaded = self.session_manager.get_session(session).is_none(); + if !matches!( + request.action, + DialogQueueAction::List | DialogQueueAction::Get { .. } + ) && session_unloaded + { return Err(PortError::new( PortErrorKind::NotFound, "Session is not loaded on the execution host", diff --git a/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs b/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs index 328601ad9d..8d8530ffcc 100644 --- a/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs +++ b/src/crates/assembly/core/src/agentic/coordination/host_message_queue_tests.rs @@ -39,6 +39,77 @@ async fn fixture() -> ( (scheduler, sessions, root, snapshot.queue_epoch) } +/// A session switched away from (local <-> remote) is no longer loaded on the +/// execution host. Passive queue reads must keep answering from the +/// scheduler's own receipts instead of failing every poll. +#[tokio::test] +async fn host_queue_reads_degrade_for_a_session_not_loaded_on_the_host() { + let (scheduler, sessions, _root, epoch) = fixture().await; + scheduler + .manage_host_queue(request( + Some(&epoch), + Action::Submit { + message: message("queued-a"), + }, + )) + .await + .unwrap(); + sessions.evict_loaded_session_for_test("host-queue-session"); + + let list = scheduler + .manage_host_queue(request(None, Action::List)) + .await + .unwrap(); + assert_eq!(list.queue_epoch, epoch, "receipts stay readable"); + assert_eq!(list.items.len(), 1); + assert_eq!(list.items[0].turn_id, "queued-a"); + assert_eq!(list.used, 1); + + let receipt = scheduler + .manage_host_queue(request( + Some(&epoch), + Action::Get { + turn_id: "queued-a".into(), + }, + )) + .await + .unwrap(); + assert_eq!( + receipt.receipt.as_ref().map(|item| item.turn_id.as_str()), + Some("queued-a") + ); + + // A mutation still refuses to accept user work for an unloaded session. + let submit = scheduler + .manage_host_queue(request( + Some(&epoch), + Action::Submit { + message: message("queued-b"), + }, + )) + .await + .unwrap_err(); + assert_eq!( + submit.message, + "Session is not loaded on the execution host" + ); + let promote = scheduler + .manage_host_queue(request( + Some(&epoch), + Action::Promote { + turn_id: "queued-a".into(), + operation_id: "promote-a".into(), + expected_active_turn_id: None, + }, + )) + .await + .unwrap_err(); + assert_eq!( + promote.message, + "Session is not loaded on the execution host" + ); +} + #[tokio::test] async fn host_queue_subagent_conversation_admission_precedes_receipts_and_queue_mutation() { use openbitfun_core_types::{SessionContinuationPolicy, SessionKind}; diff --git a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts b/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts index 4043fbf3eb..03823791f2 100644 --- a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts +++ b/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.test.ts @@ -52,6 +52,31 @@ describe('Tauri adapter expected errors', () => { )).toBe(false); }); + it('classifies a structured host_unavailable external-source answer as expected', () => { + expect(isExpectedTauriRequestError( + 'get_external_source_control_snapshot', + { request: { workspaceId: 'ws-1', forceRefresh: false } }, + new Error(JSON.stringify({ + code: 'host_unavailable', + detail: 'The remote workspace is not running the external compatibility service', + retryable: true, + })), + )).toBe(true); + }); + + it('keeps unexpected external-source failures visible', () => { + expect(isExpectedTauriRequestError( + 'get_external_source_control_snapshot', + { request: { workspaceId: 'ws-1', forceRefresh: false } }, + new Error('Internal runtime failure'), + )).toBe(false); + expect(isExpectedTauriRequestError( + 'update_external_integration_policy_command', + { request: {} }, + new Error(JSON.stringify({ code: 'host_unavailable', detail: 'unavailable host', retryable: true })), + )).toBe(false); + }); + it('records adapter init and invoke timings for each request', async () => { invokeMock.mockResolvedValueOnce({ ok: true }); const adapter = new TauriTransportAdapter(); diff --git a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts b/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts index b0b594bac2..c72279f261 100644 --- a/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts +++ b/src/web-ui/src/infrastructure/api/adapters/tauri-adapter.ts @@ -125,7 +125,54 @@ function registerSharedTauriEventListener( return registration; } +/** Structured external-source operation errors travel as the serialized + * `ExternalSourceOperationError` contract ({code, detail, retryable}). */ +interface ExternalSourceOperationErrorShape { + code?: unknown; + detail?: unknown; + retryable?: unknown; +} + +function parseExternalSourceOperationError(error: unknown): ExternalSourceOperationErrorShape | null { + const candidates = [error, (error as { details?: unknown } | null)?.details]; + for (const candidate of candidates) { + if (!candidate || typeof candidate !== 'object') continue; + const record = candidate as Record; + if (typeof record.code === 'string' && typeof record.detail === 'string') { + return candidate as ExternalSourceOperationErrorShape; + } + // A serialized payload inside `message` (Tauri rejects as `Error`) or + // `error`/`details` wrappers from an older host also counts. + for (const key of ['message', 'error'] as const) { + const value = record[key]; + if (typeof value !== 'string') continue; + try { + const nested: unknown = JSON.parse(value); + if (nested && typeof nested === 'object' + && typeof (nested as Record).code === 'string' + && typeof (nested as Record).detail === 'string') { + return nested as ExternalSourceOperationErrorShape; + } + } catch { + // Plain prose stays classified by the caller below. + } + } + } + return null; +} + +/** + * A host that by design cannot serve a request is not a product failure. + * Remote workspaces without the external compatibility service answer + * `get_external_source_*` with structured `host_unavailable`; callers already + * degrade quietly, so the transport must not log the poll as an ERROR. + */ export function isExpectedTauriRequestError(action: string, params: unknown, error: unknown): boolean { + if (action.startsWith('get_external_source_')) { + const parsed = parseExternalSourceOperationError(error); + if (parsed?.code === 'host_unavailable') return true; + } + if (action !== 'get_config') { return false; } diff --git a/src/web-ui/src/infrastructure/api/service-api/ApiClient.ts b/src/web-ui/src/infrastructure/api/service-api/ApiClient.ts index e5d512b557..3e83a6fa12 100644 --- a/src/web-ui/src/infrastructure/api/service-api/ApiClient.ts +++ b/src/web-ui/src/infrastructure/api/service-api/ApiClient.ts @@ -3,6 +3,7 @@ import { getTransportAdapter, isPeerLocalOnlyCommand, + isExpectedTauriRequestError, ITransportAdapter, type TransportRequestTiming, } from '../adapters'; @@ -543,11 +544,16 @@ export class ApiClient implements IApiClient { throw this.createApiError('REQUEST_TIMEOUT', 'Request timeout', error); } const errorMessage = transportErrorMessage(error); - - - const isExpectedError = errorMessage.includes('not found') || + + + const isExpectedError = errorMessage.includes('not found') || errorMessage.includes('Config path') || - errorMessage.includes('Configuration error'); + errorMessage.includes('Configuration error') || + // A host that by design cannot serve a poll (a + // remote workspace without the external + // compatibility service) is quiet control flow, + // not a product failure. + isExpectedTauriRequestError(config.command, config.args, error); const optionalConfigNotFound = isOptionalConfigNotFoundCommand(config, error); diff --git a/src/web-ui/src/infrastructure/services/business/workspaceManager.test.ts b/src/web-ui/src/infrastructure/services/business/workspaceManager.test.ts index 07712b4cd9..605d625fb2 100644 --- a/src/web-ui/src/infrastructure/services/business/workspaceManager.test.ts +++ b/src/web-ui/src/infrastructure/services/business/workspaceManager.test.ts @@ -14,6 +14,7 @@ const globalStateMocks = vi.hoisted(() => ({ getCurrentWorkspace: vi.fn(), getPrimaryAssistantWorkspace: vi.fn(), updateWorkspaceInfo: vi.fn(), + scanWorkspaceInfo: vi.fn(), })); const listenMock = vi.hoisted(() => vi.fn()); @@ -70,6 +71,7 @@ function configureGlobalState(): void { globalStateMocks.getCurrentWorkspace.mockResolvedValue(null); globalStateMocks.getPrimaryAssistantWorkspace.mockResolvedValue(null); globalStateMocks.updateWorkspaceInfo.mockReset(); + globalStateMocks.scanWorkspaceInfo.mockReset(); } async function getFreshWorkspaceManager() { @@ -1063,3 +1065,62 @@ describe('WorkspaceManager host catalog hints', () => { expect(manager.getState().openedWorkspaces.size).toBe(1); }); }); + +describe('WorkspaceManager state-quiet loading and error events', () => { + const workspace = { + id: 'workspace-quiet', + name: 'Quiet repo', + rootPath: '/repo/quiet', + workspaceKind: 'normal', + identity: null, + }; + + beforeEach(() => { + vi.clearAllMocks(); + configureGlobalState(); + globalStateMocks.initializeWorkspaceStartupState.mockResolvedValue({ + cleanupRemovedCount: 0, + recentWorkspaces: [workspace], + openedWorkspaces: [workspace], + currentWorkspace: workspace, + legacyRemoteWorkspace: null, + }); + globalStateMocks.getCurrentWorkspace.mockResolvedValue(workspace); + globalStateMocks.getRecentWorkspaces.mockResolvedValue([workspace]); + globalStateMocks.getOpenedWorkspaces.mockResolvedValue([workspace]); + }); + + function collectEvents(manager: { + addEventListener: (listener: (event: { type: string }) => void) => () => void; + }): { types: string[]; unsubscribe: () => void } { + const types: string[] = []; + const unsubscribe = manager.addEventListener(event => { + types.push(event.type); + }); + return { types, unsubscribe }; + } + + it('does not emit an instant-failure loading/error pair when a load starts cleanly', async () => { + const manager = await getFreshWorkspaceManager(); + await manager.initialize(); + + const { types } = collectEvents(manager); + globalStateMocks.scanWorkspaceInfo.mockResolvedValue(workspace); + await manager.scanWorkspaceInfo(); + + expect(types).not.toContain('workspace:error'); + expect(types.filter(type => type === 'workspace:loading')).toEqual(['workspace:loading']); + }); + + it('still reports real failures through workspace:error', async () => { + const manager = await getFreshWorkspaceManager(); + await manager.initialize(); + + const { types } = collectEvents(manager); + globalStateMocks.scanWorkspaceInfo.mockRejectedValue(new Error('scan failed')); + await expect(manager.scanWorkspaceInfo()).rejects.toThrow('scan failed'); + + expect(types).toContain('workspace:error'); + expect(manager.getState().error).toBe('scan failed'); + }); +}); diff --git a/src/web-ui/src/infrastructure/services/business/workspaceManager.ts b/src/web-ui/src/infrastructure/services/business/workspaceManager.ts index 92e9e795cf..834e0414fb 100644 --- a/src/web-ui/src/infrastructure/services/business/workspaceManager.ts +++ b/src/web-ui/src/infrastructure/services/business/workspaceManager.ts @@ -347,10 +347,16 @@ class WorkspaceManager { } private setLoading(loading: boolean): void { + // State-quiet events: repeating an unchanged value would have listeners + // observe `workspace:loading` and `workspace:error` in the same tick — + // e.g. loading:true then error:null right after — and treat the pair as a + // load that failed instantly. + if (this.state.loading === loading) return; this.updateState({ loading }, { type: 'workspace:loading', loading }); } private setError(error: string | null): void { + if (this.state.error === error) return; this.updateState({ error }, { type: 'workspace:error', error }); } From 7a7eddce31690e4033cbd7cad97c191d852168ec Mon Sep 17 00:00:00 2001 From: Raing5Days Date: Thu, 5 Nov 2026 08:44:28 +0800 Subject: [PATCH 2/2] fix: keep the transport classifier importable in ApiClient tests The ApiClient command-level log path now consumes isExpectedTauriRequestError, but ApiClient.test.ts replaced the '../adapters' module with a stub that did not export it, so every rejected invoke in that suite crashed with a TypeError ("preserves structured transport error facts across command wrapping", "keeps message-only transport errors parseable by domain adapters"). Keep the real classifier via vi.importActual while the transport adapter itself stays mocked. CI confirmed this was the only real failure: the macos-15 Rust/CLI jobs were cancelled mid-run (no logs, BlobNotFound) and the Rust / CLI Validation aggregator only failed because of that abandonment. Verification: web-ui vitest ApiClient + tauri-adapter + workspaceManager suites (63 passed). AI-assisted. --- .../api/service-api/ApiClient.test.ts | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/web-ui/src/infrastructure/api/service-api/ApiClient.test.ts b/src/web-ui/src/infrastructure/api/service-api/ApiClient.test.ts index c9e79bf020..a7e637610f 100644 --- a/src/web-ui/src/infrastructure/api/service-api/ApiClient.test.ts +++ b/src/web-ui/src/infrastructure/api/service-api/ApiClient.test.ts @@ -27,10 +27,16 @@ const loggerMocks = vi.hoisted(() => ({ error: vi.fn(), })); -vi.mock('../adapters', () => ({ - getTransportAdapter: () => adapterMocks, - isPeerLocalOnlyCommand: (command: string) => command === 'account_status', -})); +vi.mock('../adapters', async () => { + const actual = await vi.importActual('../adapters'); + return { + getTransportAdapter: () => adapterMocks, + isPeerLocalOnlyCommand: (command: string) => command === 'account_status', + // The command-level log path consumes the transport classifier, so keep + // the real one instead of a stub that would break every rejection. + isExpectedTauriRequestError: actual.isExpectedTauriRequestError, + }; +}); vi.mock('@/shared/utils/logger', () => ({ createLogger: () => loggerMocks,