diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/EnUsMessages.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/EnUsMessages.ets index 9f3fe26d4a..dc078b5a75 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/EnUsMessages.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/EnUsMessages.ets @@ -21,6 +21,9 @@ export const EN_US_MESSAGES: [string, string][] = [ ['common.running', 'Running'], ['common.waiting', 'Waiting'], + ['notification.taskCompletedTitle', 'Task completed'], + ['notification.taskCompletedBody', 'Your BitFun task is ready. Tap to open BitFun and view the result.'], + ['chatHome.prompt', 'What do you want to do today?'], ['chatHome.placeholder', 'Ask BitFun'], ['chatHome.localHint', 'This chat stays on the phone. Switch to Remote to read and write projects on your computer.'], diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/ZhCnMessages.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/ZhCnMessages.ets index 95a1c113af..5d80e7a6a0 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/ZhCnMessages.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/i18n/ZhCnMessages.ets @@ -21,6 +21,9 @@ export const ZH_CN_MESSAGES: [string, string][] = [ ['common.running', '运行中'], ['common.waiting', '等待中'], + ['notification.taskCompletedTitle', '任务已完成'], + ['notification.taskCompletedBody', 'BitFun 任务已经完成，点击打开 BitFun 查看结果。'], + ['chatHome.prompt', '今天想做什么？'], ['chatHome.placeholder', '问问 BitFun'], ['chatHome.localHint', '这是手机上的对话。要读写电脑上的项目，请切换到远程。'], diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/model/RemoteModels.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/model/RemoteModels.ets index 2657c48554..8f9ddf2c7f 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/model/RemoteModels.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/model/RemoteModels.ets @@ -445,6 +445,7 @@ export interface PollSessionResponse extends CommandStatusResponse { title?: string; new_messages?: ChatMessageResponse[]; total_msg_count?: number; + message_snapshot?: ChatMessageResponse[]; active_turn?: ActiveTurnSnapshotResponse; model_catalog?: RemoteModelCatalog; } @@ -456,6 +457,7 @@ export interface PollSessionResult { title: string; newMessages: ChatMessage[]; totalMessageCount: number; + messageSnapshot?: ChatMessage[]; activeTurn?: ChatMessage; modelCatalog?: RemoteModelCatalog; } diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/AppSidebar.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/AppSidebar.ets index ac65225360..4474035b5c 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/AppSidebar.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/AppSidebar.ets @@ -18,9 +18,9 @@ import { SidebarConnectionActionPolicy } from '../policy/SidebarConnectionAction /** * How many conversations show before the rest go behind an overflow row. * - * The conversations and the workspaces share one scroll now, so the list can no - * longer take every pixel it wants: six rows leave the workspace section - * visible on a phone without scrolling to find it. + * The conversations and the workspaces share one scroll. Recent history comes + * after the device/workspace navigation, so cap each expansion to keep the + * bottom section browseable without turning the sidebar into an unbounded list. */ const RECENT_BATCH: number = 6; @@ -60,7 +60,7 @@ export struct AppSidebar { @Local archivedSessionsExpanded: boolean = false; @Local visibleRecentCount: number = RECENT_BATCH; /** - * The workspace section, appended below the conversations. + * The device/workspace section, rendered before recent conversations. * * It is a slot rather than a component this file constructs because the * sidebar has no business knowing about remote state; the host wires it. @@ -81,18 +81,17 @@ export struct AppSidebar { } Stack({ alignContent: Alignment.Bottom }) { - // Conversations and workspaces scroll together. They are different - // kinds of thing — a timeline and a set of places — so they stack - // rather than compete for the same pane, and neither needs a label - // saying whether it runs here or on a desktop. + // Places are the primary navigation, so devices and workspaces come + // first. Recent conversations are history and stay at the bottom of + // the same scroll instead of pushing the current working context down. Scroll() { Column() { - if (this.showConversationSection) { - this.ConversationSection() - } if (this.showWorkspaceSection) { this.contentSlot() } + if (this.showConversationSection) { + this.ConversationSection() + } } .width('100%') .alignItems(HorizontalAlign.Start) diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ChatTimeline.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ChatTimeline.ets index 6131a8311e..92ff2ff650 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ChatTimeline.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ChatTimeline.ets @@ -6,7 +6,6 @@ import { ObservableConversationUiMessage, toConversationUiMessage } from '../state/ConversationUiModels'; -import { ConversationMessageRenderPolicy } from '../policy/ConversationMessageRenderPolicy'; import { RemoteI18n } from '../../i18n/RemoteI18n'; import { ChatTimelineItem, @@ -190,7 +189,7 @@ export struct ChatTimeline { .onAppear(() => { this.logRenderedItem(repeatItem.item); }) - } else if (repeatItem.item.message && this.shouldRenderTimelineMessage(repeatItem.item)) { + } else if (repeatItem.item.message) { ListItem() { if (repeatItem.item.message!.role === 'user') { ChatUserMessageRow({ @@ -250,7 +249,7 @@ export struct ChatTimeline { // temporary object through an @Builder freezes the first empty active // snapshot on current ArkUI builds. ForEach(this.activeTimelineItems(), (item: ObservableChatTimelineItem) => { - if (item.message && this.shouldRenderTimelineMessage(item)) { + if (item.message) { ListItem() { ChatAssistantTimelineRow({ row: item, @@ -459,27 +458,6 @@ export struct ChatTimeline { return summary; } - private shouldRenderTimelineMessage(item: ChatTimelineItem): boolean { - if (!item.message) { - return false; - } - try { - const uiMessage = toConversationUiMessage(item.message); - const render = ConversationMessageRenderPolicy.shouldRender(uiMessage); - RemoteLogger.info( - `chat timeline decide type=${item.type} id=${item.id} role=${uiMessage.role} ` + - `render=${render ? 1 : 0} text=${uiMessage.text ? uiMessage.text.length : -1} ` + - `thinking=${uiMessage.thinking ? uiMessage.thinking.length : 0} ` + - `tools=${uiMessage.tools ? uiMessage.tools.length : 0} ` + - `items=${uiMessage.items ? uiMessage.items.length : 0}` - ); - return render; - } catch (err) { - RemoteLogger.error(`chat timeline decide failed type=${item.type} id=${item.id} err=${err}`); - return false; - } - } - private logTimelineProjection(reason: string): void { const parts: string[] = []; this.timelineItems.forEach((item: ChatTimelineItem) => { diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ConversationViewContract.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ConversationViewContract.ets index 5fe3898692..c91522fb90 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ConversationViewContract.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/components/ConversationViewContract.ets @@ -1,5 +1,7 @@ import { ChatTimelineItem } from '../../model/ChatTimelineModels'; +import { ConversationMessageRenderPolicy } from '../policy/ConversationMessageRenderPolicy'; import { ChatSurface } from '../state/ChatSurface'; +import { toConversationUiMessage } from '../state/ConversationUiModels'; import { ComposerPresentation } from './ComposerBar'; /** Layout and host chrome that vary independently from conversation data. */ @@ -28,7 +30,18 @@ export class ConversationViewContract { } static visibleTimelineItems(timelineItems: T[]): T[] { - return timelineItems.filter((item: T) => item.type !== 'empty_state'); + // Repeat.virtualScroll requires every input entry to create one child. A + // hollow assistant is deliberately invisible, so passing it through and + // conditionally omitting its ListItem corrupts the virtual index/height + // map and can move adjacent user bubbles outside the viewport. Keep the + // renderability fence at the view contract instead: everything handed to + // the virtual list is guaranteed to own a visible row. + return timelineItems.filter((item: T) => { + if (item.type === 'empty_state' || !item.message) { + return false; + } + return ConversationMessageRenderPolicy.shouldRender(toConversationUiMessage(item.message)); + }); } static hasRealTimelineItem(timelineItems: ChatTimelineItem[]): boolean { diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntime.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntime.ets index 7b16f944e5..7522cb99f5 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntime.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntime.ets @@ -74,6 +74,7 @@ export class AppRootRuntime extends AppRootRuntimeComposition { onPageShow(): void { RemoteLogger.info(`page show state=${(this.remotePageState.connectionState as ConnectionState)} route=${this.appShellViewModel.currentRoute()}`); + this.taskCompletionNotificationController.onForeground(); this.remoteActivityViewModel.resume(); // Idempotent: covers the pairing that happened after the last cold start. void this.startWatchProvisioning(); @@ -84,7 +85,17 @@ export class AppRootRuntime extends AppRootRuntimeComposition { onPageHide(): void { RemoteLogger.info(`page hide state=${(this.remotePageState.connectionState as ConnectionState)} route=${this.appShellViewModel.currentRoute()}`); this.settingsController.stopPresencePolling(); - this.remoteActivityViewModel.invalidate(); + const activeRemoteTurn = this.remotePageState.activeTurnMessage; + if (this.isRemoteConversationContext(this.remotePageState.activeSession.sessionId)) { + this.trackRemoteTaskCompletion(this.remotePageState.activeSession.sessionId, activeRemoteTurn); + } + const keepRemotePolling = this.taskCompletionNotificationController.isTracking('remote'); + const watchingCompletion = this.taskCompletionNotificationController.onBackground(); + if (keepRemotePolling && watchingCompletion) { + this.remoteActivityViewModel.suspendKeepingActiveTurnPoll(); + } else { + this.remoteActivityViewModel.invalidate(); + } this.remoteConnectionCoordinator.invalidate(); this.remotePageState.setBusy(false); this.generalChatDraftLifecycleController.persistVisible(this.generalChatPageState.chatInput); @@ -93,6 +104,7 @@ export class AppRootRuntime extends AppRootRuntimeComposition { aboutToDisappear(): void { this.settingsController.stopPresencePolling(); this.watchProvisionController.stop(); + this.taskCompletionNotificationController.stop(); this.remoteActivityViewModel.invalidate(); this.remoteConnectionCoordinator.invalidate(); this.remotePageState.setBusy(false); diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntimeComposition.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntimeComposition.ets index 97e4770ffd..1881e5a0d3 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntimeComposition.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/runtime/AppRootRuntimeComposition.ets @@ -95,6 +95,11 @@ import { RemoteWorkspaceViewModel } from '../viewmodel/RemoteWorkspaceViewModel' import { RemoteSessionViewModel } from '../viewmodel/RemoteSessionViewModel'; import { GeneralChatConversationViewModel } from '../viewmodel/GeneralChatConversationViewModel'; import { ModelProviderGeneralChatAdapter } from '../../services/general-chat/ModelProviderGeneralChatAdapter'; +import { HarmonyTaskCompletionNotificationPort } from '../../services/HarmonyTaskCompletionNotificationPort'; +import { + TaskCompletionNotificationController, + TaskCompletionObservation +} from '../../services/TaskCompletionNotificationController'; export enum ConnectionState { Idle = 'idle', @@ -148,6 +153,15 @@ export abstract class AppRootRuntimeComposition { abstract toggleVoiceInput(): Promise; readonly sessionManager: RemoteSessionManager = new RemoteSessionManager(); + readonly taskCompletionNotificationController: TaskCompletionNotificationController = + new TaskCompletionNotificationController( + new HarmonyTaskCompletionNotificationPort((): Context => this.host.context()), + (observation: TaskCompletionObservation): void => { + if (observation.source === 'remote') { + this.remoteChatPollingLifecycleController.stop(); + } + } + ); readonly workspaceRepository: RemoteWorkspaceRepository = new RemoteWorkspaceRepository(this.sessionManager); readonly workspaceCoordinator: RemoteWorkspaceCoordinator = @@ -491,6 +505,11 @@ export abstract class AppRootRuntimeComposition { this.chatTimelineStore.clearPendingActiveTurn(pendingActiveId); this.conversationController.syncRemoteTimeline(); } + this.taskCompletionNotificationController.track({ + source: 'remote', + sessionId: this.remotePageState.activeSession.sessionId || '', + turnId + }); this.remoteChatPollingLifecycleController.nudge(); }, onSendFailed: ( @@ -582,6 +601,25 @@ export abstract class AppRootRuntimeComposition { }, onSnapshot: (snapshot: RemoteChatPollingSnapshot) => { this.conversationController.applyRemoteSnapshot(snapshot); + if (snapshot.activeTurn && snapshot.activeTurn.id.length > 0) { + const status = (snapshot.activeTurn.status || '').toLowerCase(); + if (status === 'failed' || status === 'error' || status === 'cancelled' || status === 'canceled') { + if (this.taskCompletionNotificationController.cancel('remote', snapshot.sessionId)) { + this.remoteChatPollingLifecycleController.stop(); + } + } else { + this.trackRemoteTaskCompletion(snapshot.sessionId, snapshot.activeTurn); + } + } else if (snapshot.sessionState.toLowerCase() === 'idle' && snapshot.completedTurnId.length === 0 && + this.taskCompletionNotificationController.cancel('remote', snapshot.sessionId)) { + this.remoteChatPollingLifecycleController.stop(); + } + if (snapshot.completedTurnId.length > 0 && + this.taskCompletionNotificationController.complete( + 'remote', snapshot.sessionId, snapshot.completedTurnId + )) { + this.remoteChatPollingLifecycleController.stop(); + } }, onError: (error: Object) => { this.remotePageState.setStatusText(ConnectionErrorPolicy.errorText(error)); @@ -660,9 +698,29 @@ export abstract class AppRootRuntimeComposition { currentActiveTurnId: (): string => this.currentActiveTurnId(), latestUserMessageText: (): string => this.conversationController.latestUserMessageText(), syncTimeline: (): void => this.conversationController.syncGeneralTimeline(), - refreshSessions: (): void => this.generalChatCommandController.refreshSessions() + refreshSessions: (): void => this.generalChatCommandController.refreshSessions(), + onTaskStarted: (sessionId: string, turnId: string): void => { + this.taskCompletionNotificationController.track({ source: 'general', sessionId, turnId }); + }, + onTaskCompleted: (sessionId: string, turnId: string): void => { + this.taskCompletionNotificationController.complete('general', sessionId, turnId); + }, + onTaskStopped: (sessionId: string, turnId: string): void => { + this.taskCompletionNotificationController.cancel('general', sessionId, turnId); + } } ); + + protected trackRemoteTaskCompletion(sessionId: string, activeTurn: ChatMessage): void { + const status = (activeTurn.status || '').toLowerCase(); + if (status !== 'active' && status !== 'completed' && status !== 'done' && status !== 'success') { + return; + } + const explicitTurnId = (activeTurn.turnId || '').trim(); + const turnId = explicitTurnId.length > 0 ? explicitTurnId : + (activeTurn.id.indexOf('active-') === 0 ? activeTurn.id.slice('active-'.length) : ''); + this.taskCompletionNotificationController.track({ source: 'remote', sessionId, turnId }); + } readonly remoteConnectionController: RemoteConnectionController = new RemoteConnectionController( this.remotePageState, diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/GeneralChatConversationViewModel.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/GeneralChatConversationViewModel.ets index 3f4a54516c..5fb1304917 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/GeneralChatConversationViewModel.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/GeneralChatConversationViewModel.ets @@ -28,6 +28,9 @@ export interface GeneralChatConversationViewModelHooks { readonly latestUserMessageText: () => string; readonly syncTimeline: () => void; readonly refreshSessions: () => void; + readonly onTaskStarted?: (sessionId: string, turnId: string) => void; + readonly onTaskCompleted?: (sessionId: string, turnId: string) => void; + readonly onTaskStopped?: (sessionId: string, turnId: string) => void; } /** Owns General Chat stream state and publishes all updates through ConversationViewModel. */ @@ -94,7 +97,7 @@ export class GeneralChatConversationViewModel { this.flushDelta(sessionId); this.hooks.refreshSessions(); if (this.stream.text().length > 0) { - this.finishStream(sessionId, result); + await this.finishStream(sessionId, result); } else { const finalMessage = this.messageForActiveTurn(result.assistantMessage); this.resetStream(); @@ -106,6 +109,9 @@ export class GeneralChatConversationViewModel { message: finalMessage }); await this.persistAssistantMessage(sessionId, finalMessage); + if (this.hooks.onTaskCompleted) { + this.hooks.onTaskCompleted(sessionId, finalMessage.turnId || ''); + } this.hooks.syncTimeline(); this.pageState.setBusy(false); this.pageState.setServiceState(GeneralChatServiceState.Ready); @@ -123,6 +129,7 @@ export class GeneralChatConversationViewModel { } const hadStream = hasStream || this.pageState.activeTurnMessage.id.length > 0; const currentSessionId = this.stream.currentSessionId(); + const currentTurnId = this.stream.currentTurnId() || this.hooks.currentActiveTurnId(); const requestInFlight = this.stream.isRequestInFlight(); if (cancelled && currentSessionId.length > 0) { this.flushDelta(currentSessionId); @@ -131,6 +138,9 @@ export class GeneralChatConversationViewModel { this.stream.markRequestCancelled(); this.command.cancelCurrentRequest(); } + if (hadStream && this.hooks.onTaskStopped) { + this.hooks.onTaskStopped(currentSessionId || this.pageState.activeSession.sessionId || '', currentTurnId); + } this.stream.reset(); if (cancelled && this.pageState.activeTurnMessage.id.length > 0) { const active = this.pageState.activeTurnMessage; @@ -175,11 +185,15 @@ export class GeneralChatConversationViewModel { private beginStream(sessionId: string): void { const activeTurn = this.stream.begin(sessionId); + const turnId = activeTurn.turnId || activeTurn.id.replace('active-', ''); this.timeline.dispatch({ type: 'turn_started', sessionId, - turnId: activeTurn.turnId || activeTurn.id.replace('active-', '') + turnId }); + if (this.hooks.onTaskStarted) { + this.hooks.onTaskStarted(sessionId, turnId); + } this.hooks.syncTimeline(); } @@ -207,7 +221,7 @@ export class GeneralChatConversationViewModel { this.hooks.syncTimeline(); } - private finishStream(sessionId: string, result: GeneralChatSendResult): void { + private async finishStream(sessionId: string, result: GeneralChatSendResult): Promise { const currentSessionId = this.stream.currentSessionId(); const visible = this.hooks.isVisible(sessionId); RemoteLogger.info(`general chat finish check session=${this.shortId(sessionId)} current=${this.shortId(currentSessionId)} visible=${visible}`); @@ -222,7 +236,10 @@ export class GeneralChatConversationViewModel { turnId: finalMessage.turnId || this.hooks.currentActiveTurnId(), message: finalMessage }); - this.persistAssistantMessage(sessionId, finalMessage); + await this.persistAssistantMessage(sessionId, finalMessage); + if (this.hooks.onTaskCompleted) { + this.hooks.onTaskCompleted(sessionId, finalMessage.turnId || ''); + } this.hooks.syncTimeline(); this.resetStream(); this.pageState.setBusy(false); @@ -255,6 +272,12 @@ export class GeneralChatConversationViewModel { images: SelectedImageAttachment[], err: Object ): Promise { + if (this.hooks.onTaskStopped) { + this.hooks.onTaskStopped( + sessionId, + this.stream.currentTurnId() || this.hooks.currentActiveTurnId() + ); + } this.stream.setRequestInFlight(false); if (this.stream.takeRequestCancelled()) { const finalMessage = this.stream.takePendingFinalMessage(); diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteActivityViewModel.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteActivityViewModel.ets index 632e3d4681..953405cc63 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteActivityViewModel.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteActivityViewModel.ets @@ -157,6 +157,12 @@ export class RemoteActivityViewModel { this.stopHeartbeat(); } + /** Keeps only the active-turn poll alive during a transient background window. */ + suspendKeepingActiveTurnPoll(): void { + this.gate.invalidate(); + this.stopHeartbeat(); + } + private async reconnect(token: number): Promise { const shouldRestore = this.hooks.isRemoteChat() && this.hooks.activeSession().sessionId.length > 0; const session = this.hooks.activeSession(); diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteTranscriptController.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteTranscriptController.ets index 3bf7b1c8ee..f72c5f4221 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteTranscriptController.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/pages/viewmodel/RemoteTranscriptController.ets @@ -255,7 +255,7 @@ export class RemoteTranscriptController { } runtime.timeline.applySnapshot(snapshot); this.syncRemoteTimeline(); - if (snapshot.newMessages.length > 0) { + if (snapshot.newMessages.length > 0 || snapshot.messageSnapshot !== undefined) { this.cacheRemoteTranscript(snapshot.sessionId); } this.knownPollVersionValue = snapshot.cursor.pollVersion; diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatSessionController.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatSessionController.ets index 027f4e4674..7b2e888848 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatSessionController.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatSessionController.ets @@ -19,6 +19,7 @@ export interface ChatSessionSnapshot { title: string; sessionState: string; newMessages: ChatMessage[]; + messageSnapshot?: ChatMessage[]; activeTurn?: ChatMessage; modelCatalog?: RemoteModelCatalog; shouldSyncAfterTurnEnded: boolean; @@ -26,6 +27,8 @@ export interface ChatSessionSnapshot { // transcript it was handing out tails of is no longer the one on screen. // Nothing below can resume from an offset into a list that shrank. historyRewritten: boolean; + /** Exact completed turn whose terminal assistant is now durable. */ + completedTurnId: string; } export interface ChatSessionControllerCallbacks { @@ -58,6 +61,7 @@ export class ChatSessionController { // Invalidates in-flight requests when the session or lifecycle changes. private generation: number = 0; private hasActiveRunningTurn: boolean = false; + private awaitingPersistedTurnId: string = ''; private turnJustEndedAt: number = 0; private readonly activeIntervalMs: number = 350; private readonly settleIntervalMs: number = 500; @@ -84,6 +88,8 @@ export class ChatSessionController { }; this.activeTurn = activeTurn && activeTurn.id.length > 0 ? activeTurn : undefined; this.hasActiveRunningTurn = this.isRunningTurn(this.activeTurn); + this.awaitingPersistedTurnId = this.isCompletedTurn(this.activeTurn) ? + this.turnId(this.activeTurn) : ''; this.turnJustEndedAt = 0; this.stopped = false; RemoteLogger.info(`poller start session=${ChatSessionController.shortId(sessionId)} active=${this.activeTurn ? this.activeTurn.id : ''}`); @@ -102,6 +108,7 @@ export class ChatSessionController { this.polling = false; this.stopped = true; this.hasActiveRunningTurn = false; + this.awaitingPersistedTurnId = ''; this.turnJustEndedAt = 0; if (clearActiveTurn) { this.activeTurn = undefined; @@ -197,7 +204,8 @@ export class ChatSessionController { private applyPollResult(result: PollSessionResult): void { const hadRunningTurn = this.hasActiveRunningTurn; const incomingMessages = result.newMessages || []; - const hasAssistantMessage = incomingMessages.some((message: ChatMessage) => { + const persistedProjection = result.messageSnapshot || incomingMessages; + const hasAssistantMessage = persistedProjection.some((message: ChatMessage) => { return message.role === 'assistant'; }); const historyRewritten = result.totalMessageCount > 0 && @@ -217,10 +225,18 @@ export class ChatSessionController { if (result.activeTurn && result.activeTurn.id.length > 0) { this.activeTurn = result.activeTurn; + if (this.isCompletedTurn(result.activeTurn)) { + this.awaitingPersistedTurnId = this.turnId(result.activeTurn); + } } else if (result.changed && (hasAssistantMessage || this.shouldClearMissingActiveTurn(result))) { this.activeTurn = undefined; } + const completedTurnId = this.persistedTerminalTurnId(persistedProjection); + if (completedTurnId.length > 0) { + this.awaitingPersistedTurnId = ''; + } + const isRunningNow = this.isRunningTurn(this.activeTurn); const turnEndedNow = hadRunningTurn && !isRunningNow; if (turnEndedNow) { @@ -235,8 +251,10 @@ export class ChatSessionController { result.changed || activeTurnChanged, result, incomingMessages, + result.messageSnapshot, turnEndedNow || (isSettlingEndedTurn && !isRunningNow), - historyRewritten + historyRewritten, + completedTurnId ); } @@ -253,8 +271,10 @@ export class ChatSessionController { changed: boolean, result: PollSessionResult, newMessages: ChatMessage[], + messageSnapshot: ChatMessage[] | undefined, shouldSyncAfterTurnEnded: boolean, - historyRewritten: boolean + historyRewritten: boolean, + completedTurnId: string ): void { this.callbacks.onSnapshot({ sessionId: this.sessionId, @@ -267,10 +287,12 @@ export class ChatSessionController { title: result.title || '', sessionState: result.sessionState || '', newMessages, + messageSnapshot, activeTurn: this.activeTurn, modelCatalog: result.modelCatalog, shouldSyncAfterTurnEnded, - historyRewritten + historyRewritten, + completedTurnId }); } @@ -308,6 +330,45 @@ export class ChatSessionController { return !!turn && turn.id.length > 0 && (turn.status || '').toLowerCase() === 'active'; } + private isCompletedTurn(turn?: ChatMessage): boolean { + if (!turn || turn.id.length === 0) { + return false; + } + const status = (turn.status || '').toLowerCase(); + return status === 'completed' || status === 'done' || status === 'success'; + } + + private persistedTerminalTurnId(messages: ChatMessage[]): string { + const expected = this.awaitingPersistedTurnId; + if (expected.length === 0) { + return ''; + } + const matched = messages.some((message: ChatMessage) => { + if (message.role !== 'assistant' || !this.isTerminalMessage(message)) { + return false; + } + const messageTurnId = (message.turnId || '').trim(); + return messageTurnId === expected || message.id === `${expected}_assistant`; + }); + return matched ? expected : ''; + } + + private isTerminalMessage(message: ChatMessage): boolean { + const status = (message.status || '').toLowerCase(); + return status === 'completed' || status === 'done' || status === 'success'; + } + + private turnId(turn?: ChatMessage): string { + if (!turn) { + return ''; + } + const explicit = (turn.turnId || '').trim(); + if (explicit.length > 0) { + return explicit; + } + return turn.id.indexOf('active-') === 0 ? turn.id.slice('active-'.length) : ''; + } + private static shortId(value: string): string { if (value.length <= 8) { return value; diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatTimelineStore.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatTimelineStore.ets index 9929543163..367736bf31 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatTimelineStore.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/services/ChatTimelineStore.ets @@ -344,7 +344,9 @@ export class ChatTimelineStore { applySnapshot(snapshot: ChatSessionSnapshot): void { this.setCursor(snapshot.cursor); - if (snapshot.newMessages.length > 0) { + if (snapshot.messageSnapshot) { + this.setPersistedMessages(snapshot.messageSnapshot); + } else if (snapshot.newMessages.length > 0) { this.mergePersistedMessages(snapshot.newMessages); } this.setActiveTurn(snapshot.activeTurn); diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/services/HarmonyTaskCompletionNotificationPort.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/services/HarmonyTaskCompletionNotificationPort.ets new file mode 100644 index 0000000000..4853f7eab3 --- /dev/null +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/services/HarmonyTaskCompletionNotificationPort.ets @@ -0,0 +1,120 @@ +import { common, wantAgent } from '@kit.AbilityKit'; +import { backgroundTaskManager } from '@kit.BackgroundTasksKit'; +import { notificationManager } from '@kit.NotificationKit'; +import { RemoteI18n } from '../i18n/RemoteI18n'; +import { RemoteLogger } from './RemoteLogger'; +import { + TaskCompletionNotificationPort, + TaskCompletionObservation +} from './TaskCompletionNotificationController'; + +const BACKGROUND_REASON: string = 'Waiting for an active BitFun task to finish'; + +/** HarmonyOS platform leaf for task-completion notifications. */ +export class HarmonyTaskCompletionNotificationPort implements TaskCompletionNotificationPort { + private readonly contextProvider: () => Context; + private permissionPrepared: boolean = false; + private suspendRequestId: number = 0; + + constructor(contextProvider: () => Context) { + this.contextProvider = contextProvider; + } + + async prepare(): Promise { + if (this.permissionPrepared) { + return; + } + this.permissionPrepared = true; + try { + if (!notificationManager.isNotificationEnabledSync()) { + await notificationManager.requestEnableNotification(this.uiAbilityContext()); + } + } catch (err) { + RemoteLogger.warn(`task completion notification permission unavailable: ${HarmonyTaskCompletionNotificationPort.errorText(err)}`); + } + } + + keepAlive(onExpired: () => void): boolean { + this.releaseKeepAlive(); + try { + const info = backgroundTaskManager.requestSuspendDelay(BACKGROUND_REASON, (): void => { + this.suspendRequestId = 0; + RemoteLogger.info('task completion background observation expired'); + onExpired(); + }); + this.suspendRequestId = info.requestId; + RemoteLogger.info(`task completion background observation started delayMs=${info.actualDelayTime}`); + return true; + } catch (err) { + RemoteLogger.warn(`task completion background observation unavailable: ${HarmonyTaskCompletionNotificationPort.errorText(err)}`); + return false; + } + } + + releaseKeepAlive(): void { + if (this.suspendRequestId === 0) { + return; + } + const requestId = this.suspendRequestId; + this.suspendRequestId = 0; + try { + backgroundTaskManager.cancelSuspendDelay(requestId); + } catch (err) { + RemoteLogger.warn(`task completion background observation release failed: ${HarmonyTaskCompletionNotificationPort.errorText(err)}`); + } + } + + async publish(observation: TaskCompletionObservation): Promise { + try { + if (!notificationManager.isNotificationEnabledSync()) { + RemoteLogger.info('task completion notification skipped: notifications disabled'); + return; + } + const context = this.uiAbilityContext(); + const notificationId = HarmonyTaskCompletionNotificationPort.notificationId(observation); + const launchAgent = await wantAgent.getWantAgent({ + wants: [{ + bundleName: context.abilityInfo.bundleName, + abilityName: context.abilityInfo.name + }], + actionType: wantAgent.OperationType.START_ABILITY, + requestCode: notificationId, + actionFlags: [wantAgent.WantAgentFlags.UPDATE_PRESENT_FLAG] + }); + await notificationManager.publish({ + id: notificationId, + appMessageId: `bitfun-task-completed:${observation.source}:${observation.sessionId}:${observation.turnId}`, + notificationSlotType: notificationManager.SlotType.SERVICE_INFORMATION, + tapDismissed: true, + wantAgent: launchAgent, + content: { + notificationContentType: notificationManager.ContentType.NOTIFICATION_CONTENT_BASIC_TEXT, + normal: { + title: RemoteI18n.t('notification.taskCompletedTitle'), + text: RemoteI18n.t('notification.taskCompletedBody') + } + } + }); + RemoteLogger.info(`task completion notification published source=${observation.source}`); + } catch (err) { + RemoteLogger.warn(`task completion notification publish failed: ${HarmonyTaskCompletionNotificationPort.errorText(err)}`); + } + } + + private uiAbilityContext(): common.UIAbilityContext { + return this.contextProvider() as common.UIAbilityContext; + } + + private static notificationId(observation: TaskCompletionObservation): number { + const value = `${observation.source}:${observation.sessionId}:${observation.turnId}`; + let hash = 17; + for (let index = 0; index < value.length; index += 1) { + hash = ((hash * 31) + value.charCodeAt(index)) & 0x7FFFFFFF; + } + return hash === 0 ? 1 : hash; + } + + private static errorText(err: Object): string { + return err instanceof Error ? err.message : JSON.stringify(err); + } +} diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/services/RemoteSessionManager.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/services/RemoteSessionManager.ets index 9b6a078274..0e318f074d 100644 --- a/src/apps/mobile/harmonyos/entry/src/main/ets/services/RemoteSessionManager.ets +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/services/RemoteSessionManager.ets @@ -338,6 +338,9 @@ export class RemoteSessionManager implements RemoteChatCommandClient, RemoteFile const response = await this.send(command); const rawMessages = response.new_messages || []; const newMessages = rawMessages.map((item: ChatMessageResponse) => RemoteResponseMapper.chatMessage(item)); + const rawSnapshot = response.message_snapshot; + const messageSnapshot = rawSnapshot ? + rawSnapshot.map((item: ChatMessageResponse) => RemoteResponseMapper.chatMessage(item)) : undefined; const result: PollSessionResult = { version: response.version || sinceVersion, changed: response.changed, @@ -345,13 +348,14 @@ export class RemoteSessionManager implements RemoteChatCommandClient, RemoteFile title: response.title || '', newMessages, totalMessageCount: response.total_msg_count || knownMessageCount, + messageSnapshot, activeTurn: response.active_turn ? RemoteResponseMapper.activeTurnToMessage(response.active_turn, response.version || sinceVersion) : undefined, modelCatalog: response.model_catalog }; const rawActive = response.active_turn; - RemoteLogger.info(`poll_session done session=${RemoteSessionManager.shortId(sessionId)} changed=${result.changed ? '1' : '0'} state=${result.sessionState} active=${rawActive ? rawActive.turn_id : ''} activeStatus=${rawActive ? rawActive.status : ''} rawText=${rawActive && rawActive.text ? rawActive.text.length : 0} rawThinking=${rawActive && rawActive.thinking ? rawActive.thinking.length : 0} rawItems=${rawActive && rawActive.items ? rawActive.items.length : 0} rawItemText=${RemoteSessionManager.itemsContentLength(rawActive ? rawActive.items : undefined)} projectedText=${result.activeTurn ? result.activeTurn.text.length : 0} new=${newMessages.length} ${newMessages.length > 0 ? TranscriptIntegrityPolicy.describeMessages(newMessages) + ' ' : ''}ms=${Date.now() - startedAt}`); + RemoteLogger.info(`poll_session done session=${RemoteSessionManager.shortId(sessionId)} changed=${result.changed ? '1' : '0'} state=${result.sessionState} active=${rawActive ? rawActive.turn_id : ''} activeStatus=${rawActive ? rawActive.status : ''} rawText=${rawActive && rawActive.text ? rawActive.text.length : 0} rawThinking=${rawActive && rawActive.thinking ? rawActive.thinking.length : 0} rawItems=${rawActive && rawActive.items ? rawActive.items.length : 0} rawItemText=${RemoteSessionManager.itemsContentLength(rawActive ? rawActive.items : undefined)} projectedText=${result.activeTurn ? result.activeTurn.text.length : 0} new=${newMessages.length} snapshot=${messageSnapshot ? messageSnapshot.length : 0} ${newMessages.length > 0 ? TranscriptIntegrityPolicy.describeMessages(newMessages) + ' ' : ''}ms=${Date.now() - startedAt}`); return result; } diff --git a/src/apps/mobile/harmonyos/entry/src/main/ets/services/TaskCompletionNotificationController.ets b/src/apps/mobile/harmonyos/entry/src/main/ets/services/TaskCompletionNotificationController.ets new file mode 100644 index 0000000000..86a5565b20 --- /dev/null +++ b/src/apps/mobile/harmonyos/entry/src/main/ets/services/TaskCompletionNotificationController.ets @@ -0,0 +1,128 @@ +export type TaskCompletionSource = 'remote' | 'general'; + +export interface TaskCompletionObservation { + source: TaskCompletionSource; + sessionId: string; + turnId: string; +} + +export interface TaskCompletionNotificationPort { + prepare(): Promise; + keepAlive(onExpired: () => void): boolean; + releaseKeepAlive(): void; + publish(observation: TaskCompletionObservation): Promise; +} + +export type TaskCompletionObservationExpired = (observation: TaskCompletionObservation) => void; + +/** + * Owns the foreground/background completion-notification state machine. + * + * Platform notification and background-task APIs stay behind the injected + * port so lifecycle policy remains deterministic and unit-testable. + */ +export class TaskCompletionNotificationController { + private readonly port: TaskCompletionNotificationPort; + private readonly onExpired: TaskCompletionObservationExpired; + private observation?: TaskCompletionObservation; + private backgrounded: boolean = false; + + constructor( + port: TaskCompletionNotificationPort, + onExpired: TaskCompletionObservationExpired = (_observation: TaskCompletionObservation): void => {} + ) { + this.port = port; + this.onExpired = onExpired; + } + + track(observation: TaskCompletionObservation): void { + const sessionId = observation.sessionId.trim(); + const turnId = observation.turnId.trim(); + if (sessionId.length === 0 || turnId.length === 0) { + return; + } + if (this.backgrounded && this.observation && this.observation.sessionId !== sessionId) { + return; + } + this.observation = { + source: observation.source, + sessionId, + turnId + }; + void this.port.prepare(); + } + + onBackground(): boolean { + if (!this.observation) { + return false; + } + this.backgrounded = true; + const observation = this.observation; + const keptAlive = this.port.keepAlive((): void => { + const expired = this.observation; + if (!this.backgrounded || !expired) { + return; + } + this.backgrounded = false; + this.observation = undefined; + this.onExpired(expired); + }); + if (!keptAlive) { + this.backgrounded = false; + this.observation = undefined; + this.onExpired(observation); + return false; + } + return true; + } + + onForeground(): void { + this.backgrounded = false; + this.observation = undefined; + this.port.releaseKeepAlive(); + } + + complete(source: TaskCompletionSource, sessionId: string, turnId: string): boolean { + const observation = this.observation; + if (!observation || observation.source !== source || + observation.sessionId !== sessionId.trim() || observation.turnId !== turnId.trim()) { + return false; + } + const shouldPublish = this.backgrounded; + this.backgrounded = false; + this.observation = undefined; + this.port.releaseKeepAlive(); + if (shouldPublish) { + void this.port.publish(observation); + } + return shouldPublish; + } + + /** Clears a failed or cancelled observation and reports whether it was backgrounded. */ + cancel(source: TaskCompletionSource, sessionId: string, turnId: string = ''): boolean { + const observation = this.observation; + const expectedTurnId = turnId.trim(); + if (!observation || observation.source !== source || observation.sessionId !== sessionId.trim() || + (expectedTurnId.length > 0 && observation.turnId !== expectedTurnId)) { + return false; + } + const wasBackgrounded = this.backgrounded; + this.backgrounded = false; + this.observation = undefined; + this.port.releaseKeepAlive(); + return wasBackgrounded; + } + + stop(): void { + this.onForeground(); + } + + isWatchingInBackground(): boolean { + return this.backgrounded && this.observation !== undefined; + } + + isTracking(source: TaskCompletionSource): boolean { + return this.observation !== undefined && this.observation.source === source; + } + +} diff --git a/src/apps/mobile/harmonyos/entry/src/test/LifecycleUnit.test.ets b/src/apps/mobile/harmonyos/entry/src/test/LifecycleUnit.test.ets index 9b9dc277d1..da13e81741 100644 --- a/src/apps/mobile/harmonyos/entry/src/test/LifecycleUnit.test.ets +++ b/src/apps/mobile/harmonyos/entry/src/test/LifecycleUnit.test.ets @@ -97,6 +97,11 @@ import { AppRouteContract } from '../main/ets/pages/navigation/AppRouteContract'; import { TimeFormat } from '../main/ets/services/TimeFormat'; +import { + TaskCompletionNotificationController, + TaskCompletionNotificationPort, + TaskCompletionObservation +} from '../main/ets/services/TaskCompletionNotificationController'; import { X25519 } from '../main/ets/services/X25519'; import { ChatMessage, @@ -165,7 +170,128 @@ import { FakeGeneralChatConfigStore } from './LocalTestFixtures'; +class FakeTaskCompletionNotificationPort implements TaskCompletionNotificationPort { + prepareCount: number = 0; + keepAliveCount: number = 0; + releaseCount: number = 0; + canKeepAlive: boolean = true; + published: TaskCompletionObservation[] = []; + expiration?: () => void; + + async prepare(): Promise { + this.prepareCount += 1; + } + + keepAlive(onExpired: () => void): boolean { + this.keepAliveCount += 1; + this.expiration = onExpired; + return this.canKeepAlive; + } + + releaseKeepAlive(): void { + this.releaseCount += 1; + } + + async publish(observation: TaskCompletionObservation): Promise { + this.published.push(observation); + } + + expire(): void { + if (this.expiration) { + this.expiration(); + } + } +} + export default function lifecycleUnitTest() { + describe('TaskCompletionNotificationController', () => { + it('publishes once only after the exact background task finishes', 0, () => { + const port = new FakeTaskCompletionNotificationPort(); + const controller = new TaskCompletionNotificationController(port); + + controller.track({ source: 'remote', sessionId: 'session-1', turnId: 'turn-1' }); + expect(controller.onBackground()).assertTrue(); + expect(controller.complete('remote', 'session-1', 'turn-other')).assertFalse(); + expect(controller.complete('remote', 'session-1', 'turn-1')).assertTrue(); + expect(controller.complete('remote', 'session-1', 'turn-1')).assertFalse(); + + expect(port.keepAliveCount).assertEqual(1); + expect(port.releaseCount).assertEqual(1); + expect(port.published.length).assertEqual(1); + expect(port.published[0].turnId).assertEqual('turn-1'); + }); + + it('clears a foreground completion without publishing later', 0, () => { + const port = new FakeTaskCompletionNotificationPort(); + const controller = new TaskCompletionNotificationController(port); + controller.track({ source: 'general', sessionId: 'session-1', turnId: 'turn-1' }); + + expect(controller.complete('general', 'session-1', 'turn-1')).assertFalse(); + + expect(controller.onBackground()).assertFalse(); + expect(port.published.length).assertEqual(0); + }); + + it('clears a failed background task without publishing', 0, () => { + const port = new FakeTaskCompletionNotificationPort(); + const controller = new TaskCompletionNotificationController(port); + controller.track({ source: 'remote', sessionId: 'session-1', turnId: 'turn-1' }); + expect(controller.onBackground()).assertTrue(); + + expect(controller.cancel('remote', 'session-1', 'turn-1')).assertTrue(); + + expect(controller.isWatchingInBackground()).assertFalse(); + expect(port.published.length).assertEqual(0); + }); + + it('does not arm an observation without an exact turn id', 0, () => { + const port = new FakeTaskCompletionNotificationPort(); + const controller = new TaskCompletionNotificationController(port); + + controller.track({ source: 'general', sessionId: 'session-1', turnId: '' }); + + expect(controller.onBackground()).assertFalse(); + expect(port.keepAliveCount).assertEqual(0); + }); + + it('stops observation when HarmonyOS cannot grant background time', 0, () => { + const port = new FakeTaskCompletionNotificationPort(); + port.canKeepAlive = false; + const expired: TaskCompletionObservation[] = []; + const controller = new TaskCompletionNotificationController( + port, + (observation: TaskCompletionObservation): void => { + expired.push(observation); + } + ); + + controller.track({ source: 'remote', sessionId: 'session-1', turnId: 'turn-1' }); + + expect(controller.onBackground()).assertFalse(); + expect(controller.isWatchingInBackground()).assertFalse(); + expect(expired.length).assertEqual(1); + }); + + it('expires a running background observation without publishing', 0, () => { + const port = new FakeTaskCompletionNotificationPort(); + const expired: TaskCompletionObservation[] = []; + const controller = new TaskCompletionNotificationController( + port, + (observation: TaskCompletionObservation): void => { + expired.push(observation); + } + ); + controller.track({ source: 'general', sessionId: 'session-2', turnId: 'turn-2' }); + expect(controller.onBackground()).assertTrue(); + + port.expire(); + + expect(controller.isWatchingInBackground()).assertFalse(); + expect(expired.length).assertEqual(1); + expect(port.published.length).assertEqual(0); + }); + }); + describe('RemoteControlSettingsPagePolicy', () => { it('opens account mode directly on the signed-out login page', 0, () => { expect(RemoteControlSettingsPagePolicy.resolve(true, false, false, false)).assertEqual('login'); diff --git a/src/apps/mobile/harmonyos/entry/src/test/RemoteControllersUnit.test.ets b/src/apps/mobile/harmonyos/entry/src/test/RemoteControllersUnit.test.ets index a53e28830b..948f65d850 100644 --- a/src/apps/mobile/harmonyos/entry/src/test/RemoteControllersUnit.test.ets +++ b/src/apps/mobile/harmonyos/entry/src/test/RemoteControllersUnit.test.ets @@ -2352,6 +2352,19 @@ export default function remoteControllersUnitTest() { expect(ConversationViewContract.shouldShowSuggestions(false, false, emptyItems)).assertFalse(); expect(ConversationViewContract.visibleTimelineItems(emptyItems).length).assertEqual(0); }); + + it('keeps user rows when a legacy remote transcript ends in a hollow assistant', 0, () => { + const timelineItems = ChatTimelineProjector.project([ + chatMessage('remote-user-1', 'user', 'Hello'), + chatMessage('remote-assistant-1', 'assistant', '') + ], [], RemoteUiState.emptyActiveTurn(), false); + + const visibleItems = ConversationViewContract.visibleTimelineItems(timelineItems); + + expect(visibleItems.length).assertEqual(1); + expect(visibleItems[0].type).assertEqual('user_message'); + expect(visibleItems[0].message!.text).assertEqual('Hello'); + }); }); describe('AppRouteContract', () => { @@ -3295,6 +3308,8 @@ export default function remoteControllersUnitTest() { it('keeps completed active turn until final assistant message arrives', 0, async () => { const manager = new FakePollSessionManager(); const snapshots: ChatSessionSnapshot[] = []; + const persistedAssistant = chatMessage('assistant-final-1', 'assistant', 'Final text'); + persistedAssistant.turnId = 'turn-2'; const controller = new ChatSessionController(manager, { onSnapshot: (snapshot: ChatSessionSnapshot) => { snapshots.push(snapshot); @@ -3310,13 +3325,13 @@ export default function remoteControllersUnitTest() { title: 'Session', newMessages: [], totalMessageCount: 1, - activeTurn: chatMessage('active-turn-2', 'assistant', 'Final text', 'completed') + activeTurn: activeChatMessage('turn-2', 'Final text', 'completed') })]; controller.start('session-1', { pollVersion: 1, knownMessageCount: 1, knownModelCatalogVersion: 0 - }, chatMessage('active-turn-2', 'assistant', 'Final text', 'active')); + }, activeChatMessage('turn-2', 'Final text', 'active')); await delay(20); manager.results = [pollResult({ @@ -3334,9 +3349,7 @@ export default function remoteControllersUnitTest() { changed: true, sessionState: 'idle', title: 'Session', - newMessages: [ - chatMessage('assistant-final-1', 'assistant', 'Final text') - ], + newMessages: [persistedAssistant], totalMessageCount: 2 })]; await controller.pollNow(); @@ -3344,8 +3357,44 @@ export default function remoteControllersUnitTest() { expect(snapshots.length).assertEqual(3); expect(snapshots[0].activeTurn ? snapshots[0].activeTurn.status : '').assertEqual('completed'); + expect(snapshots[0].completedTurnId).assertEqual(''); expect(snapshots[1].activeTurn ? snapshots[1].activeTurn.id : '').assertEqual('active-turn-2'); expect(snapshots[2].activeTurn ? snapshots[2].activeTurn.id : '').assertEqual(''); + expect(snapshots[2].completedTurnId).assertEqual('turn-2'); + }); + + it('recognizes a durable final assistant from the full message snapshot', 0, async () => { + const manager = new FakePollSessionManager(); + const snapshots: ChatSessionSnapshot[] = []; + const persistedAssistant = chatMessage('assistant-final-3', 'assistant', 'Final text'); + persistedAssistant.turnId = 'turn-3'; + manager.results = [pollResult({ + version: 3, + changed: true, + sessionState: 'idle', + title: 'Session', + newMessages: [], + messageSnapshot: [persistedAssistant], + totalMessageCount: 2, + activeTurn: activeChatMessage('turn-3', 'Final text', 'completed') + })]; + const controller = new ChatSessionController(manager, { + onSnapshot: (snapshot: ChatSessionSnapshot) => snapshots.push(snapshot), + onError: (_error: Object) => {}, + canPoll: (_sessionId: string) => true + }); + + controller.start('session-3', { + pollVersion: 2, + knownMessageCount: 1, + knownModelCatalogVersion: 0 + }, activeChatMessage('turn-3', 'Final text', 'active')); + await delay(20); + controller.stop(false); + + expect(snapshots.length).assertEqual(1); + expect(snapshots[0].completedTurnId).assertEqual('turn-3'); + expect(snapshots[0].messageSnapshot ? snapshots[0].messageSnapshot.length : 0).assertEqual(1); }); it('resyncs when a completed assistant message is updated during the settle window', 0, async () => { diff --git a/src/crates/services/services-integrations/src/remote_connect.rs b/src/crates/services/services-integrations/src/remote_connect.rs index f322e1d380..7a89ad6842 100644 --- a/src/crates/services/services-integrations/src/remote_connect.rs +++ b/src/crates/services/services-integrations/src/remote_connect.rs @@ -3596,33 +3596,50 @@ pub fn remote_persisted_poll_response( message_snapshot: Option>, model_catalog: Option, ) -> RemoteResponse { - let turn_finished = tracker.is_turn_finished(); - let has_assistant_msg = new_messages - .iter() - .any(|message| message.role == "assistant"); + let finished_turn = tracker + .is_turn_finished() + .then(|| tracker.snapshot_active_turn()) + .flatten(); + let has_persisted_terminal_assistant = finished_turn.as_ref().is_some_and(|turn| { + new_messages + .iter() + .chain(message_snapshot.iter().flat_map(|messages| messages.iter())) + .any(|message| { + message.role == "assistant" + && message.turn_id.as_deref() == Some(turn.turn_id.as_str()) + && matches!(message.status.as_deref(), Some("done") | Some("completed")) + }) + }); + let completed_turn_waiting_for_assistant = finished_turn + .as_ref() + .is_some_and(|turn| turn.status == "completed" && !has_persisted_terminal_assistant); - let active_turn = if turn_finished && has_assistant_msg { - tracker.finalize_completed_turn(); - None - } else if turn_finished { - let status = tracker.turn_status(); - if status == "completed" { - tracker.snapshot_active_turn() - } else { + let active_turn = match finished_turn { + Some(turn) if turn.status == "completed" && has_persisted_terminal_assistant => { + tracker.finalize_completed_turn(); + None + } + Some(turn) if turn.status == "completed" => Some(turn), + Some(_) => { tracker.finalize_completed_turn(); tracker.mark_persistence_clean_if_version(version); None } - } else { - tracker.snapshot_active_turn() + None => tracker.snapshot_active_turn(), }; let (send_messages, send_total, send_snapshot) = if let Some(snapshot) = message_snapshot { - tracker.mark_persistence_clean_if_version(version); + // A history fence may race the final Turn write. Keep polling until + // this exact completed Turn has a durable assistant projection instead + // of clearing the dirty bit after a snapshot that only contains older + // assistant messages. + if !completed_turn_waiting_for_assistant { + tracker.mark_persistence_clean_if_version(version); + } // Keep the additive delta for older clients that do not know the // optional replacement field yet. (Some(new_messages), Some(total_msg_count), Some(snapshot)) - } else if turn_finished && !has_assistant_msg { + } else if completed_turn_waiting_for_assistant { (None, None, None) } else { if !new_messages.is_empty() || active_turn.is_none() { @@ -4066,6 +4083,23 @@ mod tests { history_read_count: Arc, } + fn poll_test_message(id: &str, role: &str, turn_id: &str, status: Option<&str>) -> ChatMessage { + ChatMessage { + id: id.to_string(), + role: role.to_string(), + content: format!("{role} content"), + timestamp: "1".to_string(), + metadata: None, + turn_id: Some(turn_id.to_string()), + status: status.map(str::to_string), + error: None, + tools: None, + thinking: None, + items: None, + images: None, + } + } + #[async_trait::async_trait] impl RemotePollRuntimeHost for FakePollHost { fn ensure_tracker(&self, _session_id: &str) -> Arc { @@ -4257,6 +4291,162 @@ mod tests { assert!(!tracker.is_history_snapshot_required()); } + #[tokio::test] + async fn completed_turn_finalizes_when_its_assistant_only_exists_in_replacement_snapshot() { + let tracker = Arc::new(RemoteSessionStateTracker::new("session-a".to_string())); + tracker.handle_agentic_event(&AgenticEvent::DialogTurnStarted { + session_id: "session-a".to_string(), + turn_id: "turn-current".to_string(), + turn_index: 1, + user_input: "hello".to_string(), + original_user_input: None, + user_message_metadata: None, + }); + tracker.handle_agentic_event(&AgenticEvent::DialogTurnCompleted { + session_id: "session-a".to_string(), + turn_id: "turn-current".to_string(), + total_rounds: 1, + total_tools: 0, + duration_ms: 1, + partial_recovery_reason: None, + success: Some(true), + finish_reason: Some("complete".to_string()), + has_final_response: Some(true), + }); + tracker.handle_agentic_event(&AgenticEvent::SessionHistoryChanged { + session_id: "session-a".to_string(), + settled_turn_id: Some("turn-current".to_string()), + }); + let version = tracker.version(); + let assistant = poll_test_message( + "turn-current-assistant", + "assistant", + "turn-current", + Some("done"), + ); + let host = FakePollHost { + tracker: tracker.clone(), + storage_dir: Some(PathBuf::from("/workspace/project/.bitfun/sessions")), + messages: vec![assistant.clone()], + history_read_count: Arc::new(AtomicUsize::new(0)), + }; + + let response = handle_remote_poll_command( + &host, + &RemoteCommand::PollSession { + session_id: "session-a".to_string(), + since_version: version, + // The controller already counted the streaming assistant, so + // completion only changes its persisted content/status. + known_msg_count: 1, + known_model_catalog_version: None, + }, + ) + .await; + + let RemoteResponse::SessionPoll { + active_turn, + message_snapshot, + .. + } = response + else { + panic!("expected session poll response"); + }; + assert!(active_turn.is_none()); + assert_eq!(message_snapshot, Some(vec![assistant])); + assert!(tracker.snapshot_active_turn().is_none()); + assert!(!tracker.is_persistence_dirty()); + } + + #[tokio::test] + async fn completed_turn_ignores_older_assistant_and_retries_until_its_result_is_persisted() { + let tracker = Arc::new(RemoteSessionStateTracker::new("session-a".to_string())); + tracker.handle_agentic_event(&AgenticEvent::DialogTurnStarted { + session_id: "session-a".to_string(), + turn_id: "turn-current".to_string(), + turn_index: 1, + user_input: "hello".to_string(), + original_user_input: None, + user_message_metadata: None, + }); + tracker.handle_agentic_event(&AgenticEvent::DialogTurnCompleted { + session_id: "session-a".to_string(), + turn_id: "turn-current".to_string(), + total_rounds: 1, + total_tools: 0, + duration_ms: 1, + partial_recovery_reason: None, + success: Some(true), + finish_reason: Some("complete".to_string()), + has_final_response: Some(true), + }); + tracker.handle_agentic_event(&AgenticEvent::SessionHistoryChanged { + session_id: "session-a".to_string(), + settled_turn_id: Some("turn-current".to_string()), + }); + let version = tracker.version(); + let older_assistant = poll_test_message( + "turn-older-assistant", + "assistant", + "turn-older", + Some("done"), + ); + let first_host = FakePollHost { + tracker: tracker.clone(), + storage_dir: Some(PathBuf::from("/workspace/project/.bitfun/sessions")), + messages: vec![older_assistant.clone()], + history_read_count: Arc::new(AtomicUsize::new(0)), + }; + + let first_response = handle_remote_poll_command( + &first_host, + &RemoteCommand::PollSession { + session_id: "session-a".to_string(), + since_version: version, + known_msg_count: 1, + known_model_catalog_version: None, + }, + ) + .await; + let RemoteResponse::SessionPoll { active_turn, .. } = first_response else { + panic!("expected session poll response"); + }; + assert_eq!( + active_turn.as_ref().map(|turn| turn.turn_id.as_str()), + Some("turn-current") + ); + assert!(tracker.is_persistence_dirty()); + + let current_assistant = poll_test_message( + "turn-current-assistant", + "assistant", + "turn-current", + Some("done"), + ); + let retry_host = FakePollHost { + tracker: tracker.clone(), + storage_dir: Some(PathBuf::from("/workspace/project/.bitfun/sessions")), + messages: vec![older_assistant, current_assistant], + history_read_count: Arc::new(AtomicUsize::new(0)), + }; + let retry_response = handle_remote_poll_command( + &retry_host, + &RemoteCommand::PollSession { + session_id: "session-a".to_string(), + since_version: version, + known_msg_count: 1, + known_model_catalog_version: None, + }, + ) + .await; + let RemoteResponse::SessionPoll { active_turn, .. } = retry_response else { + panic!("expected session poll response"); + }; + assert!(active_turn.is_none()); + assert!(tracker.snapshot_active_turn().is_none()); + assert!(!tracker.is_persistence_dirty()); + } + #[test] fn failed_active_turn_snapshot_preserves_the_runtime_error() { let tracker = RemoteSessionStateTracker::new("session-a".to_string());