diff --git a/cli/src/agent/runnerLifecycle.ts b/cli/src/agent/runnerLifecycle.ts index f21ec30a..0ae8faa9 100644 --- a/cli/src/agent/runnerLifecycle.ts +++ b/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 cleanupAndExit: (codeOverride?: number) => Promise @@ -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 | 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, diff --git a/cli/src/agent/runners/runAgentSession.test.ts b/cli/src/agent/runners/runAgentSession.test.ts new file mode 100644 index 00000000..4b3e302e --- /dev/null +++ b/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') + }) +}) diff --git a/cli/src/agent/runners/runAgentSession.ts b/cli/src/agent/runners/runAgentSession.ts index 19219b7a..654bcd85 100644 --- a/cli/src/agent/runners/runAgentSession.ts +++ b/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(); diff --git a/cli/src/api/apiSession.ts b/cli/src/api/apiSession.ts index 5a7a8db3..2ac631c4 100644 --- a/cli/src/api/apiSession.ts +++ b/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 { diff --git a/cli/src/claude/runClaude.ts b/cli/src/claude/runClaude.ts index c8b31395..c7aae047 100644 --- a/cli/src/claude/runClaude.ts +++ b/cli/src/claude/runClaude.ts @@ -390,6 +390,7 @@ export async function runClaude(options: StartOptions = {}): Promise { 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 { throw loopError; } + if (!localFailure) { + lifecycle.setSessionEndReason('completed'); + } + await lifecycle.cleanupAndExit(); } diff --git a/cli/src/codex/runCodex.ts b/cli/src/codex/runCodex.ts index c04e305c..bb17a247 100644 --- a/cli/src/codex/runCodex.ts +++ b/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(); } diff --git a/cli/src/cursor/runCursor.ts b/cli/src/cursor/runCursor.ts index 3417cb83..66ed2377 100644 --- a/cli/src/cursor/runCursor.ts +++ b/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(); } diff --git a/cli/src/gemini/runGemini.test.ts b/cli/src/gemini/runGemini.test.ts index 52ee2eb8..694013f5 100644 --- a/cli/src/gemini/runGemini.test.ts +++ b/cli/src/gemini/runGemini.test.ts @@ -9,6 +9,7 @@ const mockGeminiSession = vi.hoisted(() => ({ const harness = vi.hoisted(() => ({ bootstrapArgs: [] as Array>, geminiLoopArgs: [] as Array>, + 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) => { 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(); + }); }); diff --git a/cli/src/gemini/runGemini.ts b/cli/src/gemini/runGemini.ts index d0e7b1d9..e93935fb 100644 --- a/cli/src/gemini/runGemini.ts +++ b/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(); } diff --git a/cli/src/opencode/runOpencode.ts b/cli/src/opencode/runOpencode.ts index 3f9e8d93..0fc781a3 100644 --- a/cli/src/opencode/runOpencode.ts +++ b/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(); } diff --git a/hub/src/config/serverSettings.ts b/hub/src/config/serverSettings.ts index 095fc2db..fd894c50 100644 --- a/hub/src/config/serverSettings.ts +++ b/hub/src/config/serverSettings.ts @@ -13,6 +13,8 @@ import { getSettingsFile, readSettings, writeSettings } from './settings' export interface ServerSettings { telegramBotToken: string | null telegramNotification: boolean + serverChanSendKey: string | null + serverChanNotification: boolean listenHost: string listenPort: number publicUrl: string @@ -24,6 +26,8 @@ export interface ServerSettingsResult { sources: { telegramBotToken: 'env' | 'file' | 'default' telegramNotification: 'env' | 'file' | 'default' + serverChanSendKey: 'env' | 'file' | 'default' + serverChanNotification: 'env' | 'file' | 'default' listenHost: 'env' | 'file' | 'default' listenPort: 'env' | 'file' | 'default' publicUrl: 'env' | 'file' | 'default' @@ -87,6 +91,8 @@ export async function loadServerSettings(dataDir: string): Promise file > null + let serverChanSendKey: string | null = null + if (process.env.SERVERCHAN_SENDKEY) { + serverChanSendKey = process.env.SERVERCHAN_SENDKEY + sources.serverChanSendKey = 'env' + if (settings.serverChanSendKey === undefined) { + settings.serverChanSendKey = serverChanSendKey + needsSave = true + } + } else if (settings.serverChanSendKey !== undefined) { + serverChanSendKey = settings.serverChanSendKey + sources.serverChanSendKey = 'file' + } + + // serverChanNotification: env > file > true + let serverChanNotification = true + if (process.env.SERVERCHAN_NOTIFICATION !== undefined) { + serverChanNotification = process.env.SERVERCHAN_NOTIFICATION === 'true' + sources.serverChanNotification = 'env' + if (settings.serverChanNotification === undefined) { + settings.serverChanNotification = serverChanNotification + needsSave = true + } + } else if (settings.serverChanNotification !== undefined) { + serverChanNotification = settings.serverChanNotification + sources.serverChanNotification = 'file' + } + // listenHost: env > file (new or old name) > default let listenHost = '127.0.0.1' if (process.env.HAPI_LISTEN_HOST) { @@ -212,6 +246,8 @@ export async function loadServerSettings(dataDir: string): Promise { it('returns the event type from a role-wrapped envelope', () => { @@ -76,3 +76,120 @@ describe('extractMessageEventType', () => { expect(extractMessageEventType(event)).toBeNull() }) }) + +describe('extractTaskNotification', () => { + it('extracts task notification from system output payload', () => { + const event: SyncEvent = { + type: 'message-received', + sessionId: 'session-1', + message: { + id: 'message-task-system', + seq: 4, + localId: null, + createdAt: 0, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'system', + subtype: 'task_notification', + status: 'completed', + summary: 'Background command stopped' + } + } + } + } + } + + expect(extractTaskNotification(event)).toEqual({ + status: 'completed', + summary: 'Background command stopped' + }) + }) + + it('extracts task notification from sidechain user output payload', () => { + const event: SyncEvent = { + type: 'message-received', + sessionId: 'session-1', + message: { + id: 'message-task-user', + seq: 5, + localId: null, + createdAt: 0, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'user', + message: { + content: ' Done completed ' + } + } + } + } + } + } + + expect(extractTaskNotification(event)).toEqual({ + status: 'completed', + summary: 'Done' + }) + }) + + it('extracts task notification from direct user output content payload', () => { + const event: SyncEvent = { + type: 'message-received', + sessionId: 'session-1', + message: { + id: 'message-task-user-direct', + seq: 6, + localId: null, + createdAt: 0, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'user', + content: ' Direct done completed ' + } + } + } + } + } + + expect(extractTaskNotification(event)).toEqual({ + status: 'completed', + summary: 'Direct done' + }) + }) + + it('returns null when task notification summary is missing', () => { + const event: SyncEvent = { + type: 'message-received', + sessionId: 'session-1', + message: { + id: 'message-task-user-empty', + seq: 6, + localId: null, + createdAt: 0, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'user', + message: { + content: ' killed ' + } + } + } + } + } + } + + expect(extractTaskNotification(event)).toBeNull() + }) +}) diff --git a/hub/src/notifications/eventParsing.ts b/hub/src/notifications/eventParsing.ts index ac615520..4f8b5271 100644 --- a/hub/src/notifications/eventParsing.ts +++ b/hub/src/notifications/eventParsing.ts @@ -38,3 +38,86 @@ export function extractMessageEventType(event: SyncEvent): string | null { const eventType = data?.type return typeof eventType === 'string' ? eventType : null } + +export type TaskNotificationEvent = { + summary: string + status?: string +} + +function extractTaskNotificationFromSystemOutput(message: unknown): TaskNotificationEvent | null { + if (!isObject(message) || message.type !== 'output') { + return null + } + + const data = isObject(message.data) ? message.data : null + if (!data || data.type !== 'system' || data.subtype !== 'task_notification') { + return null + } + + const summary = typeof data.summary === 'string' ? data.summary.trim() : '' + if (!summary) { + return null + } + + const status = typeof data.status === 'string' ? data.status.trim() : undefined + return { summary, status } +} + +function extractTaskNotificationFromUserOutput(message: unknown): TaskNotificationEvent | null { + if (!isObject(message) || message.type !== 'output') { + return null + } + + const data = isObject(message.data) ? message.data : null + if (!data || data.type !== 'user') { + return null + } + + const wrappedMessage = isObject(data.message) ? data.message : null + const content = typeof data.content === 'string' + ? data.content + : wrappedMessage?.content + if (typeof content !== 'string') { + return null + } + + const trimmed = content.trimStart() + if (!trimmed.startsWith('')) { + return null + } + + const summary = trimmed.match(/([\s\S]*?)<\/summary>/)?.[1]?.trim() + if (!summary) { + return null + } + + const status = trimmed.match(/([\s\S]*?)<\/status>/)?.[1]?.trim() || undefined + return { summary, status } +} + +export function extractTaskNotification(event: SyncEvent): TaskNotificationEvent | null { + if (event.type !== 'message-received') { + return null + } + + const message = event.message?.content + if (!isObject(message)) { + return null + } + + const roleWrapped = isObject(message.content) ? message.content : null + const candidates = roleWrapped ? [roleWrapped, message] : [message] + for (const candidate of candidates) { + const fromSystem = extractTaskNotificationFromSystemOutput(candidate) + if (fromSystem) { + return fromSystem + } + + const fromUser = extractTaskNotificationFromUserOutput(candidate) + if (fromUser) { + return fromUser + } + } + + return null +} diff --git a/hub/src/notifications/notificationHub.test.ts b/hub/src/notifications/notificationHub.test.ts index 0cfaa7c4..b744deba 100644 --- a/hub/src/notifications/notificationHub.test.ts +++ b/hub/src/notifications/notificationHub.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'bun:test' import type { Session, SyncEvent, SyncEventListener, SyncEngine } from '../sync/syncEngine' -import type { NotificationChannel } from './notificationTypes' +import type { SessionEndReason } from '@hapi/protocol' +import type { NotificationChannel, TaskNotification } from './notificationTypes' import { NotificationHub } from './notificationHub' const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) @@ -32,6 +33,8 @@ class FakeSyncEngine { class StubChannel implements NotificationChannel { readonly readySessions: Session[] = [] readonly permissionSessions: Session[] = [] + readonly taskNotifications: Array<{ session: Session; notification: TaskNotification }> = [] + readonly sessionCompletions: Session[] = [] async sendReady(session: Session): Promise { this.readySessions.push(session) @@ -40,6 +43,14 @@ class StubChannel implements NotificationChannel { async sendPermissionRequest(session: Session): Promise { this.permissionSessions.push(session) } + + async sendTaskNotification(session: Session, notification: TaskNotification): Promise { + this.taskNotifications.push({ session, notification }) + } + + async sendSessionCompletion(session: Session): Promise { + this.sessionCompletions.push(session) + } } function createSession(overrides: Partial = {}): Session { @@ -156,4 +167,81 @@ describe('NotificationHub', () => { hub.stop() }) + + it('sends task notifications for task_notification system messages', async () => { + const engine = new FakeSyncEngine() + const channel = new StubChannel() + const hub = new NotificationHub(engine as unknown as SyncEngine, [channel], { + permissionDebounceMs: 1, + readyCooldownMs: 20 + }) + + const session = createSession() + engine.setSession(session) + + const taskEvent: SyncEvent = { + type: 'message-received', + sessionId: session.id, + message: { + id: 'message-task', + seq: 2, + localId: null, + createdAt: 0, + content: { + role: 'agent', + content: { + type: 'output', + data: { + type: 'system', + subtype: 'task_notification', + status: 'completed', + summary: 'Commit T4 finished' + } + } + } + } + } + + engine.emit(taskEvent) + await sleep(5) + + expect(channel.taskNotifications).toHaveLength(1) + expect(channel.taskNotifications[0]?.notification).toEqual({ + status: 'completed', + summary: 'Commit T4 finished' + }) + + hub.stop() + }) + + it('sends session completion only for completed session-ended events', async () => { + const engine = new FakeSyncEngine() + const channel = new StubChannel() + const hub = new NotificationHub(engine as unknown as SyncEngine, [channel], { + permissionDebounceMs: 1, + readyCooldownMs: 20 + }) + + const completedSession = createSession({ id: 'session-completed', active: false }) + const terminatedSession = createSession({ id: 'session-terminated', active: false }) + engine.setSession(completedSession) + engine.setSession(terminatedSession) + + engine.emit({ + type: 'session-ended', + sessionId: completedSession.id, + reason: 'completed' satisfies SessionEndReason + }) + engine.emit({ + type: 'session-ended', + sessionId: terminatedSession.id, + reason: 'terminated' satisfies SessionEndReason + }) + await sleep(5) + + expect(channel.sessionCompletions).toHaveLength(1) + expect(channel.sessionCompletions[0]?.id).toBe(completedSession.id) + + hub.stop() + }) }) diff --git a/hub/src/notifications/notificationHub.ts b/hub/src/notifications/notificationHub.ts index b4a3d16e..bfe109ab 100644 --- a/hub/src/notifications/notificationHub.ts +++ b/hub/src/notifications/notificationHub.ts @@ -1,6 +1,7 @@ import type { Session, SyncEngine, SyncEvent } from '../sync/syncEngine' -import type { NotificationChannel, NotificationHubOptions } from './notificationTypes' -import { extractMessageEventType } from './eventParsing' +import type { SessionEndReason } from '@hapi/protocol' +import type { NotificationChannel, NotificationHubOptions, TaskNotification } from './notificationTypes' +import { extractMessageEventType, extractTaskNotification } from './eventParsing' export class NotificationHub { private readonly channels: NotificationChannel[] @@ -54,6 +55,15 @@ export class NotificationHub { return } + if (event.type === 'session-ended' && event.sessionId) { + if (event.reason === 'completed') { + this.sendSessionCompletion(event.sessionId, event.reason).catch((error) => { + console.error('[NotificationHub] Failed to send session completion notification:', error) + }) + } + return + } + if (event.type === 'message-received' && event.sessionId) { const eventType = extractMessageEventType(event) if (eventType === 'ready') { @@ -61,6 +71,13 @@ export class NotificationHub { console.error('[NotificationHub] Failed to send ready notification:', error) }) } + + const taskNotification = extractTaskNotification(event) + if (taskNotification) { + this.sendTaskNotification(event.sessionId, taskNotification).catch((error) => { + console.error('[NotificationHub] Failed to send task notification:', error) + }) + } } } @@ -146,6 +163,24 @@ export class NotificationHub { await this.notifyReady(session) } + private async sendTaskNotification(sessionId: string, notification: TaskNotification): Promise { + const session = this.getNotifiableSession(sessionId) + if (!session) { + return + } + + await this.notifyTask(session, notification) + } + + private async sendSessionCompletion(sessionId: string, reason: SessionEndReason): Promise { + const session = this.syncEngine.getSession(sessionId) + if (!session) { + return + } + + await this.notifySessionCompletion(session, reason) + } + private async notifyReady(session: Session): Promise { for (const channel of this.channels) { try { @@ -165,4 +200,27 @@ export class NotificationHub { } } } + + private async notifyTask(session: Session, notification: TaskNotification): Promise { + for (const channel of this.channels) { + try { + await channel.sendTaskNotification(session, notification) + } catch (error) { + console.error('[NotificationHub] Failed to send task notification:', error) + } + } + } + + private async notifySessionCompletion(session: Session, reason: SessionEndReason): Promise { + for (const channel of this.channels) { + if (typeof channel.sendSessionCompletion !== 'function') { + continue + } + try { + await channel.sendSessionCompletion(session, reason) + } catch (error) { + console.error('[NotificationHub] Failed to send session completion notification:', error) + } + } + } } diff --git a/hub/src/notifications/notificationTypes.ts b/hub/src/notifications/notificationTypes.ts index 3e3ba289..07e2b642 100644 --- a/hub/src/notifications/notificationTypes.ts +++ b/hub/src/notifications/notificationTypes.ts @@ -1,8 +1,16 @@ import type { Session } from '../sync/syncEngine' +import type { SessionEndReason } from '@hapi/protocol' + +export type TaskNotification = { + summary: string + status?: string +} export type NotificationChannel = { sendReady: (session: Session) => Promise sendPermissionRequest: (session: Session) => Promise + sendTaskNotification: (session: Session, notification: TaskNotification) => Promise + sendSessionCompletion?: (session: Session, reason: SessionEndReason) => Promise } export type NotificationHubOptions = { diff --git a/hub/src/push/pushNotificationChannel.test.ts b/hub/src/push/pushNotificationChannel.test.ts new file mode 100644 index 00000000..8ba04c7f --- /dev/null +++ b/hub/src/push/pushNotificationChannel.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, it } from 'bun:test' +import type { Session } from '../sync/syncEngine' +import { PushNotificationChannel } from './pushNotificationChannel' +import type { PushPayload } from './pushService' + +function createSession(overrides: Partial = {}): Session { + return { + id: 'session-task-toast', + namespace: 'default', + name: 'Demo task', + active: true, + metadata: { flavor: 'codex' }, + ...overrides + } as Session +} + +describe('PushNotificationChannel', () => { + it('sends task notifications to visible web clients before falling back to push', async () => { + const pushed: Array<{ namespace: string; payload: PushPayload }> = [] + const toasts: unknown[] = [] + const channel = new PushNotificationChannel( + { + sendToNamespace: async (namespace: string, payload: PushPayload) => { + pushed.push({ namespace, payload }) + } + } as never, + { + sendToast: async (_namespace: string, event: unknown) => { + toasts.push(event) + return 1 + } + } as never, + { + hasVisibleConnection: () => true + } as never, + '' + ) + + await channel.sendTaskNotification(createSession(), { + status: 'completed', + summary: 'Background work finished' + }) + + expect(toasts).toHaveLength(1) + expect(pushed).toHaveLength(0) + }) + + it('does not reuse one replacement tag for all task notifications in a session', async () => { + const pushed: Array<{ namespace: string; payload: PushPayload }> = [] + const channel = new PushNotificationChannel( + { + sendToNamespace: async (namespace: string, payload: PushPayload) => { + pushed.push({ namespace, payload }) + } + } as never, + { + sendToast: async () => 0 + } as never, + { + hasVisibleConnection: () => false + } as never, + '' + ) + + await channel.sendTaskNotification(createSession(), { + status: 'completed', + summary: 'First task' + }) + await channel.sendTaskNotification(createSession(), { + status: 'failed', + summary: 'Second task' + }) + + expect(pushed).toHaveLength(2) + expect(pushed[0].payload.tag).toBeUndefined() + expect(pushed[1].payload.tag).toBeUndefined() + }) +}) diff --git a/hub/src/push/pushNotificationChannel.ts b/hub/src/push/pushNotificationChannel.ts index 76e73d24..de3dbe88 100644 --- a/hub/src/push/pushNotificationChannel.ts +++ b/hub/src/push/pushNotificationChannel.ts @@ -1,5 +1,5 @@ import type { Session } from '../sync/syncEngine' -import type { NotificationChannel } from '../notifications/notificationTypes' +import type { NotificationChannel, TaskNotification } from '../notifications/notificationTypes' import { getAgentName, getSessionName } from '../notifications/sessionInfo' import type { SSEManager } from '../sse/sseManager' import type { VisibilityTracker } from '../visibility/visibilityTracker' @@ -92,6 +92,48 @@ export class PushNotificationChannel implements NotificationChannel { await this.pushService.sendToNamespace(session.namespace, payload) } + async sendTaskNotification(session: Session, notification: TaskNotification): Promise { + if (!session.active) { + return + } + + const agentName = getAgentName(session) + const name = getSessionName(session) + const normalizedStatus = notification.status?.trim().toLowerCase() + const isFailure = normalizedStatus === 'failed' + || normalizedStatus === 'error' + || normalizedStatus === 'killed' + || normalizedStatus === 'aborted' + + const payload: PushPayload = { + title: isFailure ? 'Task failed' : 'Task completed', + body: `${agentName} · ${name} · ${notification.summary}`, + data: { + type: 'task-notification', + sessionId: session.id, + url: this.buildSessionPath(session.id) + } + } + + const url = payload.data?.url ?? this.buildSessionPath(session.id) + if (this.visibilityTracker.hasVisibleConnection(session.namespace)) { + const delivered = await this.sseManager.sendToast(session.namespace, { + type: 'toast', + data: { + title: payload.title, + body: payload.body, + sessionId: session.id, + url + } + }) + if (delivered > 0) { + return + } + } + + await this.pushService.sendToNamespace(session.namespace, payload) + } + private buildSessionPath(sessionId: string): string { return `/sessions/${sessionId}` } diff --git a/hub/src/serverchan/channel.test.ts b/hub/src/serverchan/channel.test.ts new file mode 100644 index 00000000..a7671447 --- /dev/null +++ b/hub/src/serverchan/channel.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it, mock } from 'bun:test' +import type { SessionEndReason } from '@hapi/protocol' +import type { Session } from '../sync/syncEngine' +import { ServerChanChannel } from './channel' + +function createSession(overrides: Partial = {}): Session { + return { + id: 'session-1', + namespace: 'default', + seq: 1, + createdAt: 0, + updatedAt: 0, + active: true, + activeAt: 0, + metadata: { + path: 'F:\\develop\\code\\usdt', + host: 'DESKTOP' + }, + metadataVersion: 0, + agentState: null, + agentStateVersion: 0, + thinking: false, + thinkingAt: 0, + model: null, + modelReasoningEffort: null, + effort: null, + ...overrides + } +} + +describe('ServerChanChannel', () => { + it('does not send completed task notifications', async () => { + const fetchMock = mock(async () => new Response('ok', { status: 200 })) + const originalFetch = globalThis.fetch + globalThis.fetch = fetchMock as unknown as typeof fetch + + try { + const channel = new ServerChanChannel('SCT_TEST', 'https://hapi.example.com') + await channel.sendTaskNotification(createSession(), { + status: 'completed', + summary: 'Subtask finished' + }) + + expect(fetchMock).not.toHaveBeenCalled() + } finally { + globalThis.fetch = originalFetch + } + }) + + it('sends failed task notifications', async () => { + const fetchMock = mock(async () => new Response('ok', { status: 200 })) + const originalFetch = globalThis.fetch + globalThis.fetch = fetchMock as unknown as typeof fetch + + try { + const channel = new ServerChanChannel('SCT_TEST', 'https://hapi.example.com') + await channel.sendTaskNotification(createSession(), { + status: 'failed', + summary: 'Subtask failed' + }) + + expect(fetchMock).toHaveBeenCalledTimes(1) + const call = fetchMock.mock.calls[0] as unknown[] | undefined + const url = call?.[0] + const init = call?.[1] as RequestInit | undefined + expect(String(url)).toContain('https://sctapi.ftqq.com/SCT_TEST.send') + expect((init?.body as URLSearchParams).get('title')).toBe('HAPI Task failed') + } finally { + globalThis.fetch = originalFetch + } + }) + + it('sends session completion notifications', async () => { + const fetchMock = mock(async () => new Response('ok', { status: 200 })) + const originalFetch = globalThis.fetch + globalThis.fetch = fetchMock as unknown as typeof fetch + + try { + const channel = new ServerChanChannel('SCT_TEST', 'https://hapi.example.com') + await channel.sendSessionCompletion(createSession({ + id: 'session-complete', + metadata: { + path: 'F:\\develop\\code\\usdt', + host: 'DESKTOP', + name: 'USDT review' + } + }), 'completed' satisfies SessionEndReason) + + expect(fetchMock).toHaveBeenCalledTimes(1) + const call = fetchMock.mock.calls[0] as unknown[] | undefined + const url = call?.[0] + const init = call?.[1] as RequestInit | undefined + expect(String(url)).toContain('https://sctapi.ftqq.com/SCT_TEST.send') + expect((init?.body as URLSearchParams).get('title')).toBe('HAPI Session completed') + } finally { + globalThis.fetch = originalFetch + } + }) +}) diff --git a/hub/src/serverchan/channel.ts b/hub/src/serverchan/channel.ts new file mode 100644 index 00000000..71871695 --- /dev/null +++ b/hub/src/serverchan/channel.ts @@ -0,0 +1,89 @@ +import type { Session } from '../sync/syncEngine' +import type { SessionEndReason } from '@hapi/protocol' +import type { NotificationChannel, TaskNotification } from '../notifications/notificationTypes' +import { getAgentName, getSessionName } from '../notifications/sessionInfo' + +function buildSessionUrl(baseUrl: string, sessionId: string): string { + try { + return new URL(`/sessions/${sessionId}`, baseUrl).toString() + } catch { + const normalized = baseUrl.replace(/\/+$/, '') + return `${normalized}/sessions/${sessionId}` + } +} + +export class ServerChanChannel implements NotificationChannel { + constructor( + private readonly sendKey: string, + private readonly publicUrl: string + ) {} + + async sendReady(session: Session): Promise { + if (!session.active) { + return + } + + const agentName = getAgentName(session) + const name = getSessionName(session) + const url = buildSessionUrl(this.publicUrl, session.id) + await this.send('HAPI Ready for input', `${agentName} 正在等待输入\n\n会话：${name}\n\n${url}`) + } + + async sendPermissionRequest(session: Session): Promise { + if (!session.active) { + return + } + + const name = getSessionName(session) + const request = session.agentState?.requests + ? Object.values(session.agentState.requests)[0] + : null + const toolName = request?.tool ? ` (${request.tool})` : '' + const url = buildSessionUrl(this.publicUrl, session.id) + await this.send('HAPI Permission Request', `${name}${toolName}\n\n${url}`) + } + + async sendTaskNotification(session: Session, notification: TaskNotification): Promise { + if (!session.active) { + return + } + + const agentName = getAgentName(session) + const name = getSessionName(session) + const status = notification.status?.trim().toLowerCase() + const isFailure = status === 'failed' || status === 'error' || status === 'killed' || status === 'aborted' + if (!isFailure) { + return + } + const url = buildSessionUrl(this.publicUrl, session.id) + await this.send('HAPI Task failed', `${agentName} · ${name}\n\n${notification.summary}\n\n${url}`) + } + + async sendSessionCompletion(session: Session, _reason: SessionEndReason): Promise { + const agentName = getAgentName(session) + const name = getSessionName(session) + const url = buildSessionUrl(this.publicUrl, session.id) + await this.send('HAPI Session completed', `${agentName} · ${name}\n\n会话已结束。\n\n${url}`) + } + + private async send(title: string, desp: string): Promise { + const url = `https://sctapi.ftqq.com/${this.sendKey}.send` + const body = new URLSearchParams({ + title, + desp + }) + + const response = await fetch(url, { + method: 'POST', + headers: { + 'content-type': 'application/x-www-form-urlencoded' + }, + body + }) + + if (!response.ok) { + const text = await response.text().catch(() => '') + throw new Error(`Server酱发送失败: HTTP ${response.status} ${response.statusText}${text ? ` - ${text}` : ''}`) + } + } +} diff --git a/hub/src/socket/handlers/cli/sessionHandlers.ts b/hub/src/socket/handlers/cli/sessionHandlers.ts index 940a7d6b..b3a89292 100644 --- a/hub/src/socket/handlers/cli/sessionHandlers.ts +++ b/hub/src/socket/handlers/cli/sessionHandlers.ts @@ -9,6 +9,7 @@ import { extractTeamStateFromMessageContent, applyTeamStateDelta } from '../../. import { extractBackgroundTaskDelta } from '../../../sync/backgroundTasks' import { shouldRecordSessionActivity } from '../../../sync/sessionActivity' import type { CliSocketWithData } from '../../socketTypes' +import type { SessionEndReason } from '@hapi/protocol' import type { AccessErrorReason, AccessResult } from './types' type SessionAlivePayload = { @@ -26,6 +27,7 @@ type SessionAlivePayload = { type SessionEndPayload = { sid: string time: number + reason?: SessionEndReason } type ResolveSessionAccess = (sessionId: string) => AccessResult diff --git a/hub/src/sync/aliveEvents.test.ts b/hub/src/sync/aliveEvents.test.ts index 6fcfa6dd..c36a455b 100644 --- a/hub/src/sync/aliveEvents.test.ts +++ b/hub/src/sync/aliveEvents.test.ts @@ -1,9 +1,11 @@ import { describe, expect, it } from 'bun:test' import type { SyncEvent } from '@hapi/protocol/types' import { Store } from '../store' +import { RpcRegistry } from '../socket/rpcRegistry' import type { EventPublisher } from './eventPublisher' import { MachineCache } from './machineCache' import { SessionCache } from './sessionCache' +import { SyncEngine } from './syncEngine' function createPublisher(events: SyncEvent[]): EventPublisher { return { @@ -61,4 +63,183 @@ describe('alive incremental events', () => { expect(update.data).toEqual(expect.objectContaining({ id: machine.id, active: true })) }) + + it('marks session thinking immediately when a user message is accepted by the hub', async () => { + const store = new Store(':memory:') + const emittedSocketUpdates: unknown[] = [] + const io = { + of: () => ({ + to: () => ({ + emit: (_event: string, payload: unknown) => { + emittedSocketUpdates.push(payload) + } + }) + }) + } + const engine = new SyncEngine( + store, + io as never, + new RpcRegistry(), + { broadcast() {} } as never + ) + const events: SyncEvent[] = [] + const unsubscribe = engine.subscribe((event) => { + events.push(event) + }) + + try { + const session = engine.getOrCreateSession( + 'session-send-thinking', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + { requests: {}, completedRequests: {} }, + 'default' + ) + + engine.handleSessionAlive({ sid: session.id, time: Date.now(), thinking: false }) + const activeAtBeforeSend = engine.getSession(session.id)?.activeAt + events.length = 0 + + await engine.sendMessage(session.id, { + text: 'hello from web', + sentFrom: 'webapp' + }) + + expect(engine.getSession(session.id)?.thinking).toBe(true) + expect(engine.getSession(session.id)?.activeAt).toBe(activeAtBeforeSend) + expect(emittedSocketUpdates.length).toBeGreaterThan(0) + + const update = events.find((event) => event.type === 'session-updated') + expect(update).toBeDefined() + if (!update || update.type !== 'session-updated') { + return + } + + expect(update.data).toEqual(expect.objectContaining({ thinking: true })) + expect(update.data).not.toHaveProperty('activeAt') + expect((update.data as { updatedAt?: unknown }).updatedAt).toEqual(expect.any(Number)) + } finally { + unsubscribe() + engine.stop() + } + }) + + it('does not revive inactive sessions or refresh liveness when marking queued thinking', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + const now = Date.now() - 30_000 + + const session = cache.getOrCreateSession( + 'session-queued-thinking-inactive', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + { requests: {}, completedRequests: {} }, + 'default' + ) + + cache.handleSessionAlive({ sid: session.id, time: now, thinking: false }) + cache.handleSessionEnd({ sid: session.id, time: now + 1_000 }) + const inactive = cache.getSession(session.id) + expect(inactive?.active).toBe(false) + const inactiveActiveAt = inactive?.activeAt + + events.length = 0 + cache.markMessageQueued(session.id, now + 2_000) + + const updated = cache.getSession(session.id) + expect(updated?.active).toBe(false) + expect(updated?.thinking).toBe(false) + expect(updated?.activeAt).toBe(inactiveActiveAt) + expect(events.find((event) => event.type === 'session-updated')).toBeUndefined() + }) + + it('keeps queued thinking true across false heartbeats during the grace window', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + const now = Date.now() - 30_000 + + const session = cache.getOrCreateSession( + 'session-queued-thinking-grace', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + { requests: {}, completedRequests: {} }, + 'default' + ) + + cache.handleSessionAlive({ sid: session.id, time: now, thinking: false }) + cache.markMessageQueued(session.id, now + 10) + events.length = 0 + + const originalNow = Date.now + Date.now = () => now + 2_000 + try { + cache.handleSessionAlive({ sid: session.id, time: now + 2_000, thinking: false }) + } finally { + Date.now = originalNow + } + + expect(cache.getSession(session.id)?.thinking).toBe(true) + expect(events.find((event) => event.type === 'session-updated')).toBeUndefined() + }) + + it('clears queued thinking after the grace window expires', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + const now = Date.now() - 30_000 + + const session = cache.getOrCreateSession( + 'session-queued-thinking-expire', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + { requests: {}, completedRequests: {} }, + 'default' + ) + + cache.handleSessionAlive({ sid: session.id, time: now, thinking: false }) + cache.markMessageQueued(session.id, now + 10) + events.length = 0 + + cache.handleSessionAlive({ sid: session.id, time: now + 16_000, thinking: false }) + + expect(cache.getSession(session.id)?.thinking).toBe(false) + const update = events.find((event) => event.type === 'session-updated') + expect(update).toBeDefined() + if (!update || update.type !== 'session-updated') { + return + } + expect(update.data).toEqual(expect.objectContaining({ thinking: false })) + }) + + it('expires queued thinking against hub time instead of client heartbeat time', () => { + const store = new Store(':memory:') + const events: SyncEvent[] = [] + const cache = new SessionCache(store, createPublisher(events)) + const now = Date.now() + + const session = cache.getOrCreateSession( + 'session-queued-thinking-clock-skew', + { path: '/tmp/project', host: 'localhost', flavor: 'codex' }, + { requests: {}, completedRequests: {} }, + 'default' + ) + + cache.handleSessionAlive({ sid: session.id, time: now, thinking: false }) + cache.markMessageQueued(session.id, now + 10) + events.length = 0 + + const originalNow = Date.now + Date.now = () => now + 16_000 + try { + cache.handleSessionAlive({ sid: session.id, time: now - 60_000, thinking: false }) + } finally { + Date.now = originalNow + } + + expect(cache.getSession(session.id)?.thinking).toBe(false) + const update = events.find((event) => event.type === 'session-updated') + expect(update).toBeDefined() + if (!update || update.type !== 'session-updated') { + return + } + expect(update.data).toEqual(expect.objectContaining({ thinking: false })) + }) }) diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index ca4ea56f..982c78a0 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -6,12 +6,15 @@ import { EventPublisher } from './eventPublisher' import { extractTodoWriteTodosFromMessageContent, TodosSchema } from './todos' import { extractBackgroundTaskDelta } from './backgroundTasks' +const QUEUED_MESSAGE_THINKING_GRACE_MS = 15_000 + export class SessionCache { private readonly sessions: Map = new Map() private readonly lastBroadcastAtBySessionId: Map = new Map() private readonly todoBackfillAttemptedSessionIds: Set = new Set() private readonly deduplicateInProgress: Set = new Set() private readonly deduplicatePending: Set = new Set() + private readonly pendingThinkingUntilBySessionId: Map = new Map() constructor( private readonly store: Store, @@ -75,6 +78,7 @@ export class SessionCache { let stored = this.store.sessions.getSession(sessionId) if (!stored) { const existed = this.sessions.delete(sessionId) + this.pendingThinkingUntilBySessionId.delete(sessionId) if (existed) { this.publisher.emit({ type: 'session-removed', sessionId }) } @@ -181,11 +185,18 @@ export class SessionCache { const previousModelReasoningEffort = session.modelReasoningEffort const previousEffort = session.effort const previousCollaborationMode = session.collaborationMode + const pendingThinkingUntil = this.pendingThinkingUntilBySessionId.get(session.id) ?? 0 + const requestedThinking = Boolean(payload.thinking) + const hubNow = Date.now() + const preserveQueuedThinking = !requestedThinking && pendingThinkingUntil > hubNow session.active = true session.activeAt = Math.max(session.activeAt, t) - session.thinking = Boolean(payload.thinking) + session.thinking = requestedThinking || preserveQueuedThinking session.thinkingAt = t + if (requestedThinking || pendingThinkingUntil <= hubNow) { + this.pendingThinkingUntilBySessionId.delete(session.id) + } if (payload.permissionMode !== undefined) { session.permissionMode = payload.permissionMode } @@ -248,6 +259,33 @@ export class SessionCache { } } + markMessageQueued(sessionId: string, time: number = Date.now()): void { + const session = this.sessions.get(sessionId) ?? this.refreshSession(sessionId) + if (!session) return + if (!session.active) return + + const nextTime = clampAliveTime(time) ?? Date.now() + const wasThinking = session.thinking + const previousUpdatedAt = session.updatedAt + + session.thinking = true + session.thinkingAt = nextTime + session.updatedAt = Math.max(session.updatedAt, nextTime) + this.pendingThinkingUntilBySessionId.set(session.id, nextTime + QUEUED_MESSAGE_THINKING_GRACE_MS) + + if (!wasThinking || session.updatedAt !== previousUpdatedAt) { + this.lastBroadcastAtBySessionId.set(session.id, Date.now()) + this.publisher.emit({ + type: 'session-updated', + sessionId: session.id, + data: { + thinking: true, + updatedAt: session.updatedAt + } + }) + } + } + applyBackgroundTaskDelta(sessionId: string, delta: { started: number; completed: number }): void { const session = this.sessions.get(sessionId) if (!session) return @@ -312,6 +350,7 @@ export class SessionCache { session.thinking = false session.thinkingAt = t session.backgroundTaskCount = 0 + this.pendingThinkingUntilBySessionId.delete(session.id) this.publisher.emit({ type: 'session-updated', sessionId: session.id, data: { active: false, thinking: false, backgroundTaskCount: 0 } }) } @@ -325,6 +364,7 @@ export class SessionCache { if (now - session.activeAt <= sessionTimeoutMs) continue session.active = false session.thinking = false + this.pendingThinkingUntilBySessionId.delete(session.id) expired.push(session.id) this.publisher.emit({ type: 'session-updated', sessionId: session.id, data: { active: false } }) } @@ -436,6 +476,7 @@ export class SessionCache { this.sessions.delete(sessionId) this.lastBroadcastAtBySessionId.delete(sessionId) this.todoBackfillAttemptedSessionIds.delete(sessionId) + this.pendingThinkingUntilBySessionId.delete(sessionId) this.publisher.emit({ type: 'session-removed', sessionId, namespace: session.namespace }) } diff --git a/hub/src/sync/syncEngine.ts b/hub/src/sync/syncEngine.ts index c75b2222..284160ba 100644 --- a/hub/src/sync/syncEngine.ts +++ b/hub/src/sync/syncEngine.ts @@ -214,8 +214,13 @@ export class SyncEngine { this.triggerDedupIfNeeded(payload.sid) } - handleSessionEnd(payload: { sid: string; time: number }): void { + handleSessionEnd(payload: { sid: string; time: number; reason?: 'completed' | 'terminated' | 'error' }): void { this.sessionCache.handleSessionEnd(payload) + this.eventPublisher.emit({ + type: 'session-ended', + sessionId: payload.sid, + reason: payload.reason + }) // Retry dedup now that this session is inactive — a prior dedup may have // skipped it because it was still active at the time. this.triggerDedupIfNeeded(payload.sid) @@ -285,6 +290,7 @@ export class SyncEngine { } ): Promise { await this.messageService.sendMessage(sessionId, payload) + this.sessionCache.markMessageQueued(sessionId) } async approvePermission( diff --git a/hub/src/telegram/bot.ts b/hub/src/telegram/bot.ts index dc70121d..637975a9 100644 --- a/hub/src/telegram/bot.ts +++ b/hub/src/telegram/bot.ts @@ -10,7 +10,7 @@ import { SyncEngine, Session } from '../sync/syncEngine' import { handleCallback, CallbackContext } from './callbacks' import { formatSessionNotification, createNotificationKeyboard } from './sessionView' import { getAgentName } from '../notifications/sessionInfo' -import type { NotificationChannel } from '../notifications/notificationTypes' +import type { NotificationChannel, TaskNotification } from '../notifications/notificationTypes' import type { Store } from '../store' export interface BotContext extends Context { @@ -241,6 +241,36 @@ export class HappyBot implements NotificationChannel { } } } + + async sendTaskNotification(session: Session, notification: TaskNotification): Promise { + if (!session.active) { + return + } + + const agentName = getAgentName(session) + const status = notification.status?.trim().toLowerCase() + const prefix = status === 'failed' || status === 'error' || status === 'killed' || status === 'aborted' + ? 'Task failed' + : 'Task completed' + const url = buildMiniAppDeepLink(this.publicUrl, `session_${session.id}`) + const keyboard = new InlineKeyboard() + .webApp('Open Session', url) + + const chatIds = this.getBoundChatIds(session.namespace) + if (chatIds.length === 0) { + return + } + + for (const chatId of chatIds) { + try { + await this.bot.api.sendMessage(chatId, `${prefix}\n\n${agentName}: ${notification.summary}`, { + reply_markup: keyboard + }) + } catch (error) { + console.error(`[HAPIBot] Failed to send task notification to chat ${chatId}:`, error) + } + } + } } function buildMiniAppDeepLink(baseUrl: string, startParam: string): string { diff --git a/shared/src/schemas.ts b/shared/src/schemas.ts index a2a7819e..cb9f448d 100644 --- a/shared/src/schemas.ts +++ b/shared/src/schemas.ts @@ -217,6 +217,10 @@ export const SyncEventSchema = z.discriminatedUnion('type', [ SessionChangedSchema.extend({ type: z.literal('messages-invalidated') }), + SessionChangedSchema.extend({ + type: z.literal('session-ended'), + reason: z.enum(['completed', 'terminated', 'error']).optional() + }), MachineChangedSchema.extend({ type: z.literal('machine-updated'), data: z.unknown().optional() diff --git a/shared/src/socket.ts b/shared/src/socket.ts index e4072f1e..cdbe1992 100644 --- a/shared/src/socket.ts +++ b/shared/src/socket.ts @@ -67,6 +67,8 @@ export const TerminalErrorPayloadSchema = z.object({ }) export type TerminalErrorPayload = z.infer +export const SessionEndReasonSchema = z.enum(['completed', 'terminated', 'error']) +export type SessionEndReason = z.infer export const UpdateNewMessageBodySchema = z.object({ t: z.literal('new-message'), @@ -144,7 +146,7 @@ export interface ClientToServerEvents { effort?: string | null collaborationMode?: CodexCollaborationMode }) => void - 'session-end': (data: { sid: string; time: number }) => void + 'session-end': (data: { sid: string; time: number; reason?: SessionEndReason }) => void 'messages-consumed': (data: { sid: string; localIds: string[] }) => void 'update-metadata': (data: { sid: string; expectedVersion: number; metadata: unknown }, cb: (answer: { result: 'error'