feat(hub): add ServerChan task notifications (#515)

* fix(hub): 修复发送后状态显示延迟

* feat(hub): 接入Server酱任务通知

* fix(hub): 仅在会话结束时发送完成通知

* fix(hub): avoid reviving inactive queued sessions

* fix(hub): address notification review feedback

* fix(hub): expire queued thinking on hub clock

* fix(hub): 修复任务通知 review 反馈

cli/src/agent/runnerLifecycle.ts

@@ -1,4 +1,5 @@
 import type { ApiSessionClient } from '@/api/apiSession'
+import type { SessionEndReason } from '@hapi/protocol'
 import { logger } from '@/ui/logger'
 import { restoreTerminalState } from '@/ui/terminalState'
 
@@ -13,6 +14,7 @@ type RunnerLifecycleOptions = {
 export type RunnerLifecycle = {
     setExitCode: (code: number) => void
     setArchiveReason: (reason: string) => void
+    setSessionEndReason: (reason: SessionEndReason) => void
     markCrash: (error: unknown) => void
     cleanup: () => Promise<void>
     cleanupAndExit: (codeOverride?: number) => Promise<void>
@@ -22,6 +24,7 @@ export type RunnerLifecycle = {
 export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLifecycle {
     let exitCode = 0
     let archiveReason = 'User terminated'
+    let sessionEndReason: SessionEndReason = 'terminated'
     let cleanupStarted = false
     let cleanupPromise: Promise<void> | null = null
 
@@ -36,7 +39,7 @@ export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLi
             archiveReason
         }))
 
-        options.session.sendSessionDeath()
+        options.session.sendSessionDeath(sessionEndReason)
         await options.session.flush()
         await options.session.close()
     }
@@ -90,10 +93,15 @@ export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLi
         archiveReason = reason
     }
 
+    const setSessionEndReason = (reason: SessionEndReason) => {
+        sessionEndReason = reason
+    }
+
     const markCrash = (error: unknown) => {
         logger.debug(`${logPrefix} Unhandled error:`, error)
         exitCode = 1
         archiveReason = 'Session crashed'
+        sessionEndReason = 'error'
     }
 
     const registerProcessHandlers = () => {
@@ -119,6 +127,7 @@ export function createRunnerLifecycle(options: RunnerLifecycleOptions): RunnerLi
     return {
         setExitCode,
         setArchiveReason,
+        setSessionEndReason,
         markCrash,
         cleanup,
         cleanupAndExit,

cli/src/agent/runners/runAgentSession.test.ts

@@ -0,0 +1,122 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const harness = vi.hoisted(() => ({
+    sendSessionDeath: vi.fn(),
+    userMessageHandler: null as null | ((message: { content: { text: string; attachments: unknown[] } }, localId: string) => void),
+    promptError: null as Error | null,
+    cancelPrompt: vi.fn(async () => {}),
+    cancelAll: vi.fn(async () => {}),
+    stopServer: vi.fn(),
+    disconnect: vi.fn(async () => {})
+}))
+
+vi.mock('@/agent/sessionFactory', () => ({
+    bootstrapSession: vi.fn(async () => ({
+        session: {
+            updateAgentState: vi.fn(),
+            onUserMessage: vi.fn((handler) => {
+                harness.userMessageHandler = handler
+            }),
+            keepAlive: vi.fn(),
+            sendSessionEvent: vi.fn(),
+            sendAgentMessage: vi.fn(),
+            sendSessionDeath: harness.sendSessionDeath,
+            flush: vi.fn(async () => {}),
+            close: vi.fn(),
+            rpcHandlerManager: {
+                registerHandler: vi.fn()
+            }
+        },
+        sessionInfo: {
+            permissionMode: 'default'
+        }
+    }))
+}))
+
+vi.mock('@/agent/AgentRegistry', () => ({
+    AgentRegistry: {
+        create: vi.fn(() => ({
+            initialize: vi.fn(async () => {}),
+            newSession: vi.fn(async () => 'agent-session-1'),
+            prompt: vi.fn(async () => {
+                if (harness.promptError) {
+                    throw harness.promptError
+                }
+            }),
+            cancelPrompt: harness.cancelPrompt,
+            respondToPermission: vi.fn(async () => {}),
+            onPermissionRequest: vi.fn(),
+            disconnect: harness.disconnect
+        }))
+    }
+}))
+
+vi.mock('@/agent/permissionAdapter', () => ({
+    PermissionAdapter: vi.fn(function PermissionAdapter() {
+        return {
+            cancelAll: harness.cancelAll
+        }
+    })
+}))
+
+vi.mock('@/claude/utils/startHappyServer', () => ({
+    startHappyServer: vi.fn(async () => ({
+        url: 'http://127.0.0.1:1234',
+        stop: harness.stopServer
+    }))
+}))
+
+vi.mock('@/utils/spawnHappyCLI', () => ({
+    getHappyCliCommand: vi.fn(() => ({ command: 'hapi', args: [], env: [] }))
+}))
+
+vi.mock('@/claude/registerKillSessionHandler', () => ({
+    registerKillSessionHandler: vi.fn()
+}))
+
+vi.mock('@/utils/invokedCwd', () => ({
+    getInvokedCwd: vi.fn(() => '/tmp/project')
+}))
+
+vi.mock('@/ui/logger', () => ({
+    logger: {
+        debug: vi.fn(),
+        warn: vi.fn()
+    }
+}))
+
+vi.mock('@/utils/attachmentFormatter', () => ({
+    formatMessageWithAttachments: vi.fn((text: string) => text)
+}))
+
+import { runAgentSession } from './runAgentSession'
+
+describe('runAgentSession', () => {
+    beforeEach(() => {
+        harness.sendSessionDeath.mockClear()
+        harness.userMessageHandler = null
+        harness.promptError = null
+        harness.cancelPrompt.mockClear()
+        harness.cancelAll.mockClear()
+        harness.stopServer.mockClear()
+        harness.disconnect.mockClear()
+    })
+
+    it('reports unhandled ACP runner failures as error, not completed', async () => {
+        harness.cancelAll.mockImplementationOnce(async () => {
+            throw new Error('cancel failed')
+        })
+
+        const running = runAgentSession({ agentType: 'acp' })
+        for (let i = 0; i < 5; i++) {
+            await Promise.resolve()
+        }
+        expect(harness.userMessageHandler).not.toBeNull()
+        harness.userMessageHandler?.({ content: { text: 'hello', attachments: [] } }, 'local-1')
+
+        await expect(running).rejects.toThrow('cancel failed')
+
+        expect(harness.sendSessionDeath).toHaveBeenCalledWith('error')
+        expect(harness.sendSessionDeath).not.toHaveBeenCalledWith('completed')
+    })
+})

cli/src/agent/runners/runAgentSession.ts

@@ -14,6 +14,7 @@ import { formatMessageWithAttachments } from '@/utils/attachmentFormatter';
 import { getInvokedCwd } from '@/utils/invokedCwd';
 import { PermissionModeSchema } from '@hapi/protocol/schemas';
 import { isPermissionModeAllowedForFlavor } from '@hapi/protocol';
+import type { SessionEndReason } from '@hapi/protocol';
 
 function emitReadyIfIdle(props: {
     queueSize: () => number;
@@ -146,6 +147,7 @@ export async function runAgentSession(opts: {
 
     registerKillSessionHandler(session.rpcHandlerManager, handleKillSession);
 
+    let sessionEndReason: SessionEndReason = 'completed';
     try {
         while (!shouldExit) {
             waitAbortController = new AbortController();
@@ -191,10 +193,16 @@ export async function runAgentSession(opts: {
                 });
             }
         }
+        if (shouldExit) {
+            sessionEndReason = 'terminated';
+        }
+    } catch (error) {
+        sessionEndReason = 'error';
+        throw error;
     } finally {
         clearInterval(keepAliveInterval);
         await permissionAdapter.cancelAll('Session ended');
-        session.sendSessionDeath();
+        session.sendSessionDeath(sessionEndReason);
         await session.flush();
         session.close();
         await backend.disconnect();

cli/src/api/apiSession.ts

@@ -10,6 +10,7 @@ import { AsyncLock } from '@/utils/lock'
 import type { RawJSONLines } from '@/claude/types'
 import { configuration } from '@/configuration'
 import { AGENT_MESSAGE_PAYLOAD_TYPE } from "@hapi/protocol"
+import type { SessionEndReason } from '@hapi/protocol'
 import type { ClientToServerEvents, ServerToClientEvents, Update } from '@hapi/protocol'
 import {
     TerminalClosePayloadSchema,
@@ -499,9 +500,9 @@ export class ApiSessionClient extends EventEmitter {
         this.socket.emit('messages-consumed', { sid: this.sessionId, localIds })
     }
 
-    sendSessionDeath(): void {
+    sendSessionDeath(reason?: SessionEndReason): void {
         void cleanupUploadDir(this.sessionId)
-        this.socket.emit('session-end', { sid: this.sessionId, time: Date.now() })
+        this.socket.emit('session-end', { sid: this.sessionId, time: Date.now(), reason })
     }
 
     updateMetadata(handler: (metadata: Metadata) => Metadata): void {

cli/src/claude/runClaude.ts

@@ -390,6 +390,7 @@ export async function runClaude(options: StartOptions = {}): Promise<void> {
     if (localFailure?.exitReason === 'exit') {
         lifecycle.setExitCode(1);
         lifecycle.setArchiveReason(`Local launch failed: ${formatFailureReason(localFailure.message)}`);
+        lifecycle.setSessionEndReason('error');
     }
 
     if (loopFailed) {
@@ -397,5 +398,9 @@ export async function runClaude(options: StartOptions = {}): Promise<void> {
         throw loopError;
     }
 
+    if (!localFailure) {
+        lifecycle.setSessionEndReason('completed');
+    }
+
     await lifecycle.cleanupAndExit();
 }

cli/src/codex/runCodex.ts

@@ -214,6 +214,8 @@ export async function runCodex(opts: {
         };
     });
 
+    let crashed = false;
+
     try {
         await loop({
             path: workingDirectory,
@@ -236,6 +238,7 @@ export async function runCodex(opts: {
             }
         });
     } catch (error) {
+        crashed = true;
         lifecycle.markCrash(error);
         logger.debug('[codex] Loop error:', error);
     } finally {
@@ -243,6 +246,9 @@ export async function runCodex(opts: {
         if (localFailure?.exitReason === 'exit') {
             lifecycle.setExitCode(1);
             lifecycle.setArchiveReason(`Local launch failed: ${formatFailureReason(localFailure.message)}`);
+            lifecycle.setSessionEndReason('error');
+        } else if (!crashed) {
+            lifecycle.setSessionEndReason('completed');
         }
         await lifecycle.cleanupAndExit();
     }

cli/src/cursor/runCursor.ts

@@ -108,6 +108,8 @@ export async function runCursor(opts: {
         return { applied: { permissionMode: currentPermissionMode } };
     });
 
+    let crashed = false;
+
     try {
         await loop({
             path: workingDirectory,
@@ -127,6 +129,7 @@ export async function runCursor(opts: {
             }
         });
     } catch (error) {
+        crashed = true;
         lifecycle.markCrash(error);
         logger.debug('[cursor] Loop error:', error);
     } finally {
@@ -134,6 +137,9 @@ export async function runCursor(opts: {
         if (localFailure?.exitReason === 'exit') {
             lifecycle.setExitCode(1);
             lifecycle.setArchiveReason(`Local launch failed: ${formatFailureReason(localFailure.message)}`);
+            lifecycle.setSessionEndReason('error');
+        } else if (!crashed) {
+            lifecycle.setSessionEndReason('completed');
         }
         await lifecycle.cleanupAndExit();
     }

cli/src/gemini/runGemini.test.ts

@@ -9,6 +9,7 @@ const mockGeminiSession = vi.hoisted(() => ({
 const harness = vi.hoisted(() => ({
     bootstrapArgs: [] as Array<Record<string, unknown>>,
     geminiLoopArgs: [] as Array<Record<string, unknown>>,
+    geminiLoopError: null as Error | null,
     session: {
         onUserMessage: vi.fn(),
         rpcHandlerManager: {
@@ -30,6 +31,9 @@ vi.mock('@/agent/sessionFactory', () => ({
 vi.mock('./loop', () => ({
     geminiLoop: vi.fn(async (options: Record<string, unknown>) => {
         harness.geminiLoopArgs.push(options);
+        if (harness.geminiLoopError) {
+            throw harness.geminiLoopError;
+        }
         const onSessionReady = options.onSessionReady as ((session: unknown) => void) | undefined;
         if (onSessionReady) {
             onSessionReady(mockGeminiSession);
@@ -41,15 +45,18 @@ vi.mock('@/claude/registerKillSessionHandler', () => ({
     registerKillSessionHandler: vi.fn()
 }));
 
+const lifecycleMock = vi.hoisted(() => ({
+    registerProcessHandlers: vi.fn(),
+    cleanupAndExit: vi.fn(async () => {}),
+    markCrash: vi.fn(),
+    setExitCode: vi.fn(),
+    setArchiveReason: vi.fn(),
+    setSessionEndReason: vi.fn()
+}));
+
 vi.mock('@/agent/runnerLifecycle', () => ({
     createModeChangeHandler: vi.fn(() => vi.fn()),
-    createRunnerLifecycle: vi.fn(() => ({
-        registerProcessHandlers: vi.fn(),
-        cleanupAndExit: vi.fn(async () => {}),
-        markCrash: vi.fn(),
-        setExitCode: vi.fn(),
-        setArchiveReason: vi.fn()
-    })),
+    createRunnerLifecycle: vi.fn(() => lifecycleMock),
     setControlledByUser: vi.fn()
 }));
 
@@ -88,10 +95,17 @@ describe('runGemini', () => {
     beforeEach(() => {
         harness.bootstrapArgs.length = 0;
         harness.geminiLoopArgs.length = 0;
+        harness.geminiLoopError = null;
         mockGeminiSession.setModel.mockReset();
         mockGeminiSession.setPermissionMode.mockReset();
         harness.session.onUserMessage.mockReset();
         harness.session.rpcHandlerManager.registerHandler.mockReset();
+        lifecycleMock.registerProcessHandlers.mockClear();
+        lifecycleMock.cleanupAndExit.mockClear();
+        lifecycleMock.markCrash.mockClear();
+        lifecycleMock.setExitCode.mockClear();
+        lifecycleMock.setArchiveReason.mockClear();
+        lifecycleMock.setSessionEndReason.mockClear();
         resolveGeminiRuntimeConfigMock.mockReset();
     });
 
@@ -252,4 +266,18 @@ describe('runGemini', () => {
 
         expect(harness.geminiLoopArgs[0]?.resumeSessionId).toBeUndefined();
     });
+
+    it('preserves crash session end reason instead of overwriting it as completed', async () => {
+        resolveGeminiRuntimeConfigMock.mockReturnValue({
+            model: 'gemini-2.5-pro',
+            modelSource: 'default'
+        });
+        harness.geminiLoopError = new Error('loop failed');
+
+        await runGemini({});
+
+        expect(lifecycleMock.markCrash).toHaveBeenCalledWith(harness.geminiLoopError);
+        expect(lifecycleMock.setSessionEndReason).not.toHaveBeenCalledWith('completed');
+        expect(lifecycleMock.cleanupAndExit).toHaveBeenCalled();
+    });
 });

cli/src/gemini/runGemini.ts

@@ -160,6 +160,8 @@ export async function runGemini(opts: {
         return { applied };
     });
 
+    let crashed = false;
+
     try {
         await geminiLoop({
             path: workingDirectory,
@@ -179,6 +181,7 @@ export async function runGemini(opts: {
             }
         });
     } catch (error) {
+        crashed = true;
         lifecycle.markCrash(error);
         logger.debug('[gemini] Loop error:', error);
     } finally {
@@ -186,6 +189,9 @@ export async function runGemini(opts: {
         if (localFailure?.exitReason === 'exit') {
             lifecycle.setExitCode(1);
             lifecycle.setArchiveReason(`Local launch failed: ${localFailure.message.slice(0, 200)}`);
+            lifecycle.setSessionEndReason('error');
+        } else if (!crashed) {
+            lifecycle.setSessionEndReason('completed');
         }
         await lifecycle.cleanupAndExit();
     }

cli/src/opencode/runOpencode.ts

@@ -114,6 +114,8 @@ export async function runOpencode(opts: {
         return { applied: { permissionMode: currentPermissionMode } };
     });
 
+    let crashed = false;
+
     try {
         await opencodeLoop({
             path: workingDirectory,
@@ -133,6 +135,7 @@ export async function runOpencode(opts: {
             }
         });
     } catch (error) {
+        crashed = true;
         lifecycle.markCrash(error);
         logger.debug('[opencode] Loop error:', error);
     } finally {
@@ -140,6 +143,9 @@ export async function runOpencode(opts: {
         if (localFailure?.exitReason === 'exit') {
             lifecycle.setExitCode(1);
             lifecycle.setArchiveReason(`Local launch failed: ${localFailure.message.slice(0, 200)}`);
+            lifecycle.setSessionEndReason('error');
+        } else if (!crashed) {
+            lifecycle.setSessionEndReason('completed');
         }
         await lifecycle.cleanupAndExit();
     }