fix(agent-session): mark stopped pending messages as paused

src/main/ai/agentSession/AgentSessionRuntimeService.ts

@@ -60,7 +60,7 @@ export interface BeginAgentSessionTurnInput {
   agentId: string
   agentType: string
   modelId: UniqueModelId
-  assistantMessageId?: string
+  assistantMessageId: string
   userMessage?: AgentSessionMessageEntity
   /** Container-level OTel trace id (one trace per session); cached on the entry. */
   traceId?: string
@@ -69,6 +69,7 @@ export interface BeginAgentSessionTurnInput {
 export interface AgentSessionRuntimeHandle {
   listeners: StreamListener[]
   turnId: string
+  abortController: AbortController
 }
 
 export interface OpenAgentSessionTurnStreamInput {
@@ -90,10 +91,11 @@ export interface AgentSessionRuntimeSnapshot {
 
 type AgentSessionTurn = {
   turnId: string
-  assistantMessageId?: string
+  assistantMessageId: string
   userMessage: AgentSessionMessageEntity
   modelId: UniqueModelId
   admitted: boolean
+  abortController: AbortController
   terminalStatus?: AgentSessionRuntimeTerminalStatus
   controller?: ReadableStreamDefaultController<UIMessageChunk>
   activeToolIds: Set<string>
@@ -211,6 +213,7 @@ export class AgentSessionRuntimeService extends BaseService {
       userMessage,
       modelId: input.modelId,
       admitted: false,
+      abortController: new AbortController(),
       activeToolIds: new Set()
     }
 
@@ -231,7 +234,8 @@ export class AgentSessionRuntimeService extends BaseService {
           new AgentSessionRuntimeTerminalListener(this, input.sessionId),
           new TraceFlushListener(input.topicId)
         ],
-        turnId
+        turnId,
+        abortController: turn.abortController
       }
     }
 
@@ -256,7 +260,8 @@ export class AgentSessionRuntimeService extends BaseService {
         new AgentSessionRuntimeTerminalListener(this, input.sessionId),
         new TraceFlushListener(input.topicId)
       ],
-      turnId
+      turnId,
+      abortController: turn.abortController
     }
   }
 
@@ -572,6 +577,13 @@ export class AgentSessionRuntimeService extends BaseService {
     return toolApprovalRegistry.dispatch(approvalId, decision)
   }
 
+  abortPendingTurn(sessionId: string, reason: string): boolean {
+    const turn = this.entries.get(sessionId)?.currentTurn
+    if (!turn || turn.terminalStatus || turn.abortController.signal.aborted) return false
+    turn.abortController.abort(reason)
+    return true
+  }
+
   protected onStop(): void {
     this.closeAll()
     toolApprovalRegistry.clear('agent-session-runtime-stop')
@@ -915,6 +927,7 @@ export class AgentSessionRuntimeService extends BaseService {
       userMessage: nextMessage,
       modelId: entry.modelId,
       admitted: false,
+      abortController: new AbortController(),
       activeToolIds: new Set()
     }
 
@@ -939,6 +952,7 @@ export class AgentSessionRuntimeService extends BaseService {
         messages,
         runtime: { kind: 'agent-session', sessionId: entry.sessionId, turnId }
       },
+      abortController: entry.currentTurn.abortController,
       listeners: [
         this.createPersistenceListener(entry, nextMessage),
         new AgentSessionRuntimeTerminalListener(this, entry.sessionId),
@@ -1005,6 +1019,7 @@ export class AgentSessionRuntimeService extends BaseService {
       modelId: entry.modelId,
       // Pre-admitted: the steer was already delivered via the hook, so `admitTurn` must NOT re-send it.
       admitted: true,
+      abortController: new AbortController(),
       activeToolIds: new Set()
     }
 
@@ -1029,6 +1044,7 @@ export class AgentSessionRuntimeService extends BaseService {
         messages,
         runtime: { kind: 'agent-session', sessionId: entry.sessionId, turnId }
       },
+      abortController: entry.currentTurn.abortController,
       listeners: [
         this.createPersistenceListener(entry, steerMessage),
         new AgentSessionRuntimeTerminalListener(this, entry.sessionId),
@@ -1089,12 +1105,18 @@ export class AgentSessionRuntimeService extends BaseService {
     entry: AgentSessionRuntimeEntry,
     userMessage: AgentSessionMessageEntity
   ): StreamListener {
+    const currentTurn = entry.currentTurn
+    if (!currentTurn) {
+      throw new Error(`Cannot create persistence listener without an active turn: ${entry.sessionId}`)
+    }
+    const { assistantMessageId } = currentTurn
     const userText = extractMessageText(userMessage)
     return new PersistenceListener({
       topicId: entry.topicId,
       modelId: entry.modelId,
       backend: new AgentSessionMessageBackend({
         sessionId: entry.sessionId,
+        assistantMessageId,
         modelId: entry.modelId,
         runtimeResumeToken: () => entry.lastResumeToken,
         afterPersist: async (finalMessage) => {

src/main/ai/agentSession/__tests__/AgentSessionRuntimeService.test.ts

@@ -238,6 +238,32 @@ describe('AgentSessionRuntimeService', () => {
     })
   })
 
+  it('aborts the current turn controller before the stream starts', () => {
+    const service = new AgentSessionRuntimeService()
+    const handle = service.beginTurn(baseTurnInput)
+
+    expect(service.abortPendingTurn('session-1', 'user-requested')).toBe(true)
+    expect(handle.abortController.signal.aborted).toBe(true)
+    expect(handle.abortController.signal.reason).toBe('user-requested')
+  })
+
+  it('does not reuse an aborted controller for a later turn', () => {
+    const service = new AgentSessionRuntimeService()
+    const first = service.beginTurn(baseTurnInput)
+
+    expect(service.abortPendingTurn('session-1', 'user-requested')).toBe(true)
+    void terminalListener(first).onPaused({ status: 'paused', isTopicDone: true })
+
+    const second = service.beginTurn({
+      ...baseTurnInput,
+      assistantMessageId: 'assistant-2',
+      userMessage: userMessage('user-2')
+    })
+
+    expect(first.abortController.signal.aborted).toBe(true)
+    expect(second.abortController.signal.aborted).toBe(false)
+  })
+
   it('marks the runtime idle when the terminal listener observes done', () => {
     const service = new AgentSessionRuntimeService()
     const handle = service.beginTurn(baseTurnInput)
@@ -616,6 +642,30 @@ describe('AgentSessionRuntimeService', () => {
     })
   })
 
+  it('persists empty paused terminals to the active assistant placeholder', async () => {
+    const service = new AgentSessionRuntimeService()
+    const handle = service.beginTurn({ ...baseTurnInput, userMessage: userMessage('user-1') })
+    getEntry(service).lastResumeToken = 'resume-1'
+
+    await persistenceListener(handle).onPaused({
+      status: 'paused',
+      isTopicDone: true,
+      finalMessage: undefined
+    })
+
+    expect(mocks.saveMessage).toHaveBeenCalledWith({
+      sessionId: 'session-1',
+      runtimeResumeToken: 'resume-1',
+      message: {
+        id: 'assistant-1',
+        role: 'assistant',
+        status: 'paused',
+        data: { parts: [] },
+        modelId: 'claude-code::claude-sonnet-4-5'
+      }
+    })
+  })
+
   it('routes runtime events from the selected driver into the active turn', async () => {
     const events = createAsyncQueue<any>()
     const connection = {
@@ -1627,6 +1677,7 @@ describe('AgentSessionRuntimeService', () => {
         ],
         runtime: { kind: 'agent-session', sessionId: 'session-1', turnId: expect.any(String) }
       },
+      abortController: expect.any(AbortController),
       listeners: [
         expect.objectContaining({ id: expect.stringContaining('persistence:agents-db:') }),
         expect.objectContaining({ id: 'agent-runtime:session-1' }),

src/main/ai/agentSession/persistence/AgentSessionMessageBackend.ts

@@ -10,13 +10,14 @@
 import { agentSessionMessageService } from '@data/services/AgentSessionMessageService'
 import type { CherryMessagePart, CherryUIMessage } from '@shared/data/types/message'
 import type { UniqueModelId } from '@shared/data/types/model'
-import { v7 as uuidv7 } from 'uuid'
 
 import { finalizeInterruptedParts, type PersistAssistantInput, type PersistenceBackend } from '../../streamManager'
 
 export interface AgentSessionMessageBackendOptions {
   /** Cherry Studio agent-session id. */
   sessionId: string
+  /** Existing assistant placeholder id to finalize. */
+  assistantMessageId: string
   /** Model id used for this assistant message. */
   modelId?: UniqueModelId
   /** Opaque runtime resume token persisted for future recovery; `undefined` when unknown. */
@@ -27,6 +28,7 @@ export interface AgentSessionMessageBackendOptions {
 
 export class AgentSessionMessageBackend implements PersistenceBackend {
   readonly kind = 'agents-db'
+  readonly canPersistEmptyTerminal = true
   readonly afterPersist?: (finalMessage: CherryUIMessage) => Promise<void>
 
   constructor(private readonly opts: AgentSessionMessageBackendOptions) {
@@ -41,7 +43,7 @@ export class AgentSessionMessageBackend implements PersistenceBackend {
       sessionId: this.opts.sessionId,
       ...(runtimeResumeToken ? { runtimeResumeToken } : {}),
       message: {
-        id: finalMessage?.id ?? uuidv7(),
+        id: finalMessage?.id ?? this.opts.assistantMessageId,
         role: 'assistant',
         status,
         data: { parts },

src/main/ai/streamManager/AiStreamManager.ts

@@ -19,7 +19,7 @@ import type { UniqueModelId } from '@shared/data/types/model'
 import type { SerializedError } from '@shared/types/error'
 import { type UIMessageChunk } from 'ai'
 
-import { isAgentSessionTopic } from '../agentSession/topic'
+import { extractAgentSessionId, isAgentSessionTopic } from '../agentSession/topic'
 import { applyTurnOutputAttributes } from '../observability'
 import type { AiStreamRequest, CallOverrides } from '../types'
 import { buildCompactReplay } from './buildCompactReplay'
@@ -79,6 +79,7 @@ export interface SendModelSpec {
   modelId: UniqueModelId
   request: AiStreamRequest
   rootSpan?: Span
+  abortController?: AbortController
 }
 
 export interface SendInput {
@@ -105,6 +106,7 @@ export interface StartRuntimeTurnInput {
   request: AiStreamRequest
   listeners: StreamListener[]
   rootSpan?: Span
+  abortController?: AbortController
 }
 
 // ── Inspection snapshots ────────────────────────────────────────────
@@ -356,11 +358,18 @@ export class AiStreamManager extends BaseService {
     const isMultiModel = input.models.length > 1
     const executions = new Map<UniqueModelId, StreamExecution>()
 
-    for (const { modelId, request, rootSpan } of input.models) {
+    for (const { modelId, request, rootSpan, abortController } of input.models) {
       if (executions.has(modelId)) {
         throw new Error(`send() got duplicate modelId ${modelId} for topic ${input.topicId}`)
       }
-      const exec = this.createAndLaunchExecution(input.topicId, modelId, request, input.siblingsGroupId, rootSpan)
+      const exec = this.createAndLaunchExecution(
+        input.topicId,
+        modelId,
+        request,
+        input.siblingsGroupId,
+        rootSpan,
+        abortController
+      )
       executions.set(modelId, exec)
     }
 
@@ -380,6 +389,10 @@ export class AiStreamManager extends BaseService {
     // Chat broadcasts to SharedCache so `useChatWithHistory.resumeActiveStream` can attach; prompt is silent.
     stream.lifecycle.onCreated(stream)
 
+    if ([...executions.values()].every((exec) => exec.abortController.signal.aborted)) {
+      stream.status = 'aborted'
+    }
+
     return {
       mode: 'started',
       executionIds: input.models.map((m) => m.modelId)
@@ -437,7 +450,14 @@ export class AiStreamManager extends BaseService {
 
     return this.send({
       topicId: input.topicId,
-      models: [{ modelId: input.modelId, request: input.request, rootSpan: input.rootSpan }],
+      models: [
+        {
+          modelId: input.modelId,
+          request: input.request,
+          rootSpan: input.rootSpan,
+          abortController: input.abortController
+        }
+      ],
       listeners: [...carriedListeners, ...input.listeners]
     })
   }
@@ -541,7 +561,12 @@ export class AiStreamManager extends BaseService {
   /** Abort all executions in a topic. */
   abort(topicId: string, reason: string): void {
     const stream = this.activeStreams.get(topicId)
-    if (!stream || !isLiveStatus(stream.status)) return
+    if (!stream || !isLiveStatus(stream.status)) {
+      if (isAgentSessionTopic(topicId)) {
+        application.get('AgentSessionRuntimeService').abortPendingTurn(extractAgentSessionId(topicId), reason)
+      }
+      return
+    }
     logger.info('Aborting stream', { topicId, reason })
     for (const exec of stream.executions.values()) {
       if (exec.status === 'streaming') {
@@ -961,14 +986,15 @@ export class AiStreamManager extends BaseService {
     modelId: UniqueModelId,
     request: AiStreamRequest,
     siblingsGroupId?: number,
-    rootSpan?: Span
+    rootSpan?: Span,
+    abortController?: AbortController
   ): StreamExecution {
     // `loopPromise` is overwritten right after launch; initialise to a resolved sentinel
     // so the `exec` object reference is stable inside the arrow function below.
     const exec: StreamExecution = {
       modelId,
       anchorMessageId: request.messageId,
-      abortController: new AbortController(),
+      abortController: abortController ?? new AbortController(),
       status: 'streaming',
       buffer: [],
       droppedChunks: 0,