marketing-shibata50/multica
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge pull request #224 from multica-ai/codex/fix-multi-session-chat-sync

Browse source 
fix(chat): stabilize multi-session sync and tool flow
This commit is contained in:
Jiayuan Zhang 2026-02-17 16:05:53 +08:00 • committed by GitHub
commit 7387216482
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 354 additions and 268 deletions
Download patch file Download diff file Expand all files Collapse all files

23
apps/desktop/src/main/electron-env.d.ts vendored
View file

@ -194,11 +194,12 @@ interface ElectronAPI {
revokeDevice: (deviceId: string) => Promise<{ ok: boolean }>
onConnectionStateChanged: (callback: (state: string) => void) => void
offConnectionStateChanged: () => void
onDevicesChanged: (callback: () => void) => void
offDevicesChanged: () => void
onInboundMessage: (callback: (event: InboundMessageEvent) => void) => void
offInboundMessage: () => void
}
onDevicesChanged: (callback: () => void) => void
offDevicesChanged: () => void
onInboundMessage: (callback: (event: InboundMessageEvent) => void) => () => void
onConversationsChanged: (callback: () => void) => () => void
offInboundMessage: () => void
}
tools: {
list: () => Promise<ToolInfo[]>
toggle: (name: string) => Promise<unknown>
@ -253,18 +254,18 @@ interface ElectronAPI {
wake: (reason?: string) => Promise<{ ok: boolean; result?: unknown; error?: string }>
}
localChat: {
subscribe: (conversationId: string) => Promise<{ ok?: boolean; error?: string; alreadySubscribed?: boolean }>
unsubscribe: (conversationId: string) => Promise<{ ok: boolean }>
getHistory: (conversationId: string, options?: { offset?: number; limit?: number }) => Promise<{ messages: unknown[]; total: number; offset: number; limit: number; contextWindowTokens?: number }>
subscribe: (conversationId: string) => Promise<{ ok?: boolean; error?: string; alreadySubscribed?: boolean; token?: number; isRunning?: boolean }>
unsubscribe: (conversationId: string, token?: number) => Promise<{ ok: boolean; skipped?: boolean; alreadyUnsubscribed?: boolean }>
getHistory: (conversationId: string, options?: { offset?: number; limit?: number }) => Promise<{ messages: unknown[]; total: number; offset: number; limit: number; contextWindowTokens?: number; isRunning?: boolean }>
send: (conversationId: string, content: string) => Promise<{ ok?: boolean; error?: string }>
abort: (conversationId: string) => Promise<{ ok?: boolean; error?: string }>
resolveExecApproval: (approvalId: string, decision: string) => Promise<{ ok: boolean }>
onEvent: (callback: (event: LocalChatEvent) => void) => void
onEvent: (callback: (event: LocalChatEvent) => void) => () => void
offEvent: () => void
onApproval: (callback: (approval: LocalChatApproval) => void) => void
onApproval: (callback: (approval: LocalChatApproval) => void) => () => void
offApproval: () => void
}
}
}
// Used in Renderer process, expose in `preload.ts`
interface Window {

84
apps/desktop/src/main/ipc/hub.ts
View file

@ -13,9 +13,21 @@ let hub: Hub | null = null
let defaultConversationId: string | null = null
let mainWindowRef: BrowserWindow | null = null
function isConversationBusy(conversation: AsyncAgent): boolean {
return conversation.isRunning
|| conversation.isStreaming
|| conversation.hasQueuedMessages()
|| conversation.getPendingWrites() > 0
}
interface IpcAgentSubscription {
token: number
unsubscribe: () => void
}
// Track which agents have active IPC subscriptions (for local direct chat)
// Value is the unsubscribe function returned by agent.subscribe()
const ipcAgentSubscriptions = new Map<string, () => void>()
const ipcAgentSubscriptions = new Map<string, IpcAgentSubscription>()
let nextIpcSubscriptionToken = 1
// Resolve gateway URL: GATEWAY_URL env > MAIN_VITE_GATEWAY_URL (.env file)
const gatewayUrl =
@ -258,9 +270,19 @@ export function registerHubIpcHandlers(): void {
}
const logicalAgentId = h.getConversationAgentId(conversationId) ?? conversationId
// Already subscribed?
if (ipcAgentSubscriptions.has(conversationId)) {
return { ok: true, alreadySubscribed: true }
const existingSubscription = ipcAgentSubscriptions.get(conversationId)
if (existingSubscription) {
const refreshedToken = nextIpcSubscriptionToken++
ipcAgentSubscriptions.set(conversationId, {
token: refreshedToken,
unsubscribe: existingSubscription.unsubscribe,
})
return {
ok: true,
alreadySubscribed: true,
token: refreshedToken,
isRunning: isConversationBusy(conversation),
}
}
// Track current stream ID for message grouping
@ -318,7 +340,8 @@ export function registerHubIpcHandlers(): void {
}
})
ipcAgentSubscriptions.set(conversationId, unsubscribe)
const token = nextIpcSubscriptionToken++
ipcAgentSubscriptions.set(conversationId, { token, unsubscribe })
// Register local approval handler so exec approval requests route via IPC
h.setLocalApprovalHandler(conversationId, (payload) => {
@ -329,17 +352,24 @@ export function registerHubIpcHandlers(): void {
safeLog(`[IPC] Local chat subscribed to conversation: ${conversationId}`)
return { ok: true }
return { ok: true, token, isRunning: isConversationBusy(conversation) }
})
/**
* Unsubscribe from local agent events.
*/
ipcMain.handle('localChat:unsubscribe', async (_event, conversationId: string) => {
const unsubscribe = ipcAgentSubscriptions.get(conversationId)
if (unsubscribe) {
unsubscribe()
ipcMain.handle('localChat:unsubscribe', async (_event, conversationId: string, token?: number) => {
const subscription = ipcAgentSubscriptions.get(conversationId)
if (!subscription) {
return { ok: true, alreadyUnsubscribed: true }
}
if (typeof token === 'number' && token !== subscription.token) {
safeLog(`[IPC] Skip stale local chat unsubscribe: conversation=${conversationId}, token=${token}`)
return { ok: true, skipped: true }
}
subscription.unsubscribe()
ipcAgentSubscriptions.delete(conversationId)
getHub().removeLocalApprovalHandler(conversationId)
safeLog(`[IPC] Local chat unsubscribed from conversation: ${conversationId}`)
@ -362,21 +392,36 @@ export function registerHubIpcHandlers(): void {
const h = getHub()
const agent = h.getConversation(conversationId)
if (!agent) {
return { messages: [], total: 0, offset: 0, limit: 0, contextWindowTokens: undefined }
return {
messages: [],
total: 0,
offset: 0,
limit: 0,
contextWindowTokens: undefined,
isRunning: false,
}
}
try {
await agent.ensureInitialized()
const allMessages = agent.loadSessionMessagesForDisplay()
const contextWindowTokens = agent.getContextWindowTokens()
const isRunning = isConversationBusy(agent)
const total = allMessages.length
// Must match DEFAULT_MESSAGES_LIMIT from @multica/sdk/actions/rpc
const limit = options?.limit ?? 200
const offset = options?.offset ?? Math.max(0, total - limit)
const sliced = allMessages.slice(offset, offset + limit)
return { messages: sliced, total, offset, limit, contextWindowTokens }
return { messages: sliced, total, offset, limit, contextWindowTokens, isRunning }
} catch {
return { messages: [], total: 0, offset: 0, limit: 0, contextWindowTokens: undefined }
return {
messages: [],
total: 0,
offset: 0,
limit: 0,
contextWindowTokens: undefined,
isRunning: false,
}
}
})
@ -528,6 +573,13 @@ export function setupDeviceConfirmation(mainWindow: Electron.BrowserWindow): voi
mainWindow.webContents.send('hub:inbound-message', event)
}
})
// Forward conversation list changes (e.g. created from Telegram / RPC).
h.onConversationsChanged(() => {
if (!mainWindow.isDestroyed()) {
mainWindow.webContents.send('hub:conversations-changed')
}
})
}
/**
@ -535,8 +587,8 @@ export function setupDeviceConfirmation(mainWindow: Electron.BrowserWindow): voi
*/
export function cleanupHub(): void {
// Unsubscribe all IPC listeners
for (const unsubscribe of ipcAgentSubscriptions.values()) {
unsubscribe()
for (const subscription of ipcAgentSubscriptions.values()) {
subscription.unsubscribe()
}
ipcAgentSubscriptions.clear()

59
apps/desktop/src/preload/index.ts
View file

@ -207,13 +207,28 @@ const electronAPI = {
offDevicesChanged: () => {
ipcRenderer.removeAllListeners('hub:devices-changed')
},
onInboundMessage: (callback: (event: InboundMessageEvent) => void) => {
ipcRenderer.on('hub:inbound-message', (_event, data: InboundMessageEvent) => callback(data))
},
offInboundMessage: () => {
ipcRenderer.removeAllListeners('hub:inbound-message')
},
},
onInboundMessage: (callback: (event: InboundMessageEvent) => void) => {
const listener = (_event: Electron.IpcRendererEvent, data: InboundMessageEvent): void => {
callback(data)
}
ipcRenderer.on('hub:inbound-message', listener)
return (): void => {
ipcRenderer.removeListener('hub:inbound-message', listener)
}
},
onConversationsChanged: (callback: () => void) => {
const listener = (): void => {
callback()
}
ipcRenderer.on('hub:conversations-changed', listener)
return (): void => {
ipcRenderer.removeListener('hub:conversations-changed', listener)
}
},
offInboundMessage: () => {
ipcRenderer.removeAllListeners('hub:inbound-message')
},
},
// Tools management
tools: {
@ -334,7 +349,7 @@ const electronAPI = {
/** Subscribe to conversation events for local direct chat */
subscribe: (conversationId: string) => ipcRenderer.invoke('localChat:subscribe', conversationId),
/** Unsubscribe from conversation events */
unsubscribe: (conversationId: string) => ipcRenderer.invoke('localChat:unsubscribe', conversationId),
unsubscribe: (conversationId: string, token?: number) => ipcRenderer.invoke('localChat:unsubscribe', conversationId, token),
/** Get message history for local chat with pagination (returns raw AgentMessageItem[]) */
getHistory: (conversationId: string, options?: { offset?: number; limit?: number }) =>
ipcRenderer.invoke('localChat:getHistory', conversationId, options),
@ -347,18 +362,30 @@ const electronAPI = {
/** Resolve an exec approval request */
resolveExecApproval: (approvalId: string, decision: string) =>
ipcRenderer.invoke('localChat:resolveExecApproval', approvalId, decision),
/** Listen for agent events */
onEvent: (callback: (event: LocalChatEvent) => void) => {
ipcRenderer.on('localChat:event', (_event, data: LocalChatEvent) => callback(data))
},
/** Listen for agent events */
onEvent: (callback: (event: LocalChatEvent) => void) => {
const listener = (_event: Electron.IpcRendererEvent, data: LocalChatEvent): void => {
callback(data)
}
ipcRenderer.on('localChat:event', listener)
return (): void => {
ipcRenderer.removeListener('localChat:event', listener)
}
},
/** Remove event listener */
offEvent: () => {
ipcRenderer.removeAllListeners('localChat:event')
},
/** Listen for exec approval requests */
onApproval: (callback: (approval: LocalChatApproval) => void) => {
ipcRenderer.on('localChat:approval', (_event, data: LocalChatApproval) => callback(data))
},
/** Listen for exec approval requests */
onApproval: (callback: (approval: LocalChatApproval) => void) => {
const listener = (_event: Electron.IpcRendererEvent, data: LocalChatApproval): void => {
callback(data)
}
ipcRenderer.on('localChat:approval', listener)
return (): void => {
ipcRenderer.removeListener('localChat:approval', listener)
}
},
/** Remove approval listener */
offApproval: () => {
ipcRenderer.removeAllListeners('localChat:approval')

115
apps/desktop/src/renderer/src/hooks/use-local-chat.ts
View file

@ -15,10 +15,34 @@ export interface QueuedLocalMessage {
createdAt: number
}
interface QueuedInboundMessage {
content: string
agentId: string
conversationId: string
source: MessageSource
}
interface UseLocalChatOptions {
conversationId?: string
}
interface LocalChatSubscribeResult {
ok?: boolean
error?: string
alreadySubscribed?: boolean
token?: number
isRunning?: boolean
}
interface LocalChatHistoryResult {
messages?: AgentMessageItem[]
total: number
offset: number
limit: number
contextWindowTokens?: number
isRunning?: boolean
}
function makeQueueId(): string {
return globalThis.crypto?.randomUUID?.() ?? `queued-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`
}
@ -35,6 +59,7 @@ export function useLocalChat(options: UseLocalChatOptions = {}) {
const [isLoadingMore, setIsLoadingMore] = useState(false)
const isLoadingMoreRef = useRef(false)
const [queuedMessages, setQueuedMessages] = useState<QueuedLocalMessage[]>([])
const [queuedInboundMessages, setQueuedInboundMessages] = useState<QueuedInboundMessage[]>([])
const [initError, setInitError] = useState<string | null>(null)
const initRef = useRef(false)
const offsetRef = useRef<number | null>(null)
@ -68,20 +93,31 @@ export function useLocalChat(options: UseLocalChatOptions = {}) {
// Subscribe to events + fetch history once conversation is available
useEffect(() => {
if (!activeConversationId) return
let disposed = false
setQueuedMessages([])
setQueuedInboundMessages([])
offsetRef.current = null
setIsLoading(false)
setIsLoadingHistory(true)
chatRef.current.reset()
// Subscribe to agent events
window.electronAPI.localChat.subscribe(activeConversationId).catch(() => {})
const subscribePromise = window.electronAPI.localChat.subscribe(activeConversationId)
.then((result) => {
const typed = result as LocalChatSubscribeResult
if (!disposed && typed.isRunning) {
setIsLoading(true)
}
return typed
})
.catch(() => null)
// Listen for stream events
window.electronAPI.localChat.onEvent((data) => {
const unsubscribeEvent = window.electronAPI.localChat.onEvent((data) => {
// Cast IPC event to StreamPayload (same shape: { agentId, streamId, event })
const payload = data as unknown as StreamPayload
if (!payload.event) return
if (payload.conversationId !== activeConversationId) return
// Handle agent error events
if (payload.event.type === 'agent_error') {
@ -111,21 +147,33 @@ export function useLocalChat(options: UseLocalChatOptions = {}) {
})
// Listen for exec approval requests
window.electronAPI.localChat.onApproval((approval) => {
const unsubscribeApproval = window.electronAPI.localChat.onApproval((approval) => {
if (approval.conversationId !== activeConversationId) return
chatRef.current.addApproval(approval as ExecApprovalRequestPayload)
})
// Listen for inbound messages from all sources (gateway, channel)
// This allows the local UI to display messages from other sources
window.electronAPI.hub.onInboundMessage((event: InboundMessageEvent) => {
const unsubscribeInbound = window.electronAPI.hub.onInboundMessage((event: InboundMessageEvent) => {
const eventConversationId = event.conversationId
// Only add non-local messages (local messages are added by sendMessage)
if (event.source.type !== 'local' && eventConversationId === activeConversationId) {
const queuedInbound: QueuedInboundMessage = {
content: event.content,
agentId: event.agentId,
conversationId: eventConversationId,
source: event.source as MessageSource,
}
if (isLoadingRef.current) {
setQueuedInboundMessages((prev) => [...prev, queuedInbound])
return
}
chatRef.current.addUserMessage(
event.content,
event.agentId,
eventConversationId,
event.source as MessageSource,
queuedInbound.content,
queuedInbound.agentId,
queuedInbound.conversationId,
queuedInbound.source,
)
setIsLoading(true)
}
@ -136,6 +184,11 @@ export function useLocalChat(options: UseLocalChatOptions = {}) {
limit: DEFAULT_MESSAGES_LIMIT,
})
.then((result) => {
if (disposed) return
const typed = result as LocalChatHistoryResult
if (typed.isRunning) {
setIsLoading(true)
}
console.log('[LocalChat] getHistory result:', result.messages?.length, 'messages, total:', result.total)
if (result.messages?.length) {
chatRef.current.setHistory(result.messages as AgentMessageItem[], activeConversationId, activeConversationId, {
@ -153,13 +206,23 @@ export function useLocalChat(options: UseLocalChatOptions = {}) {
}
})
.catch(() => {})
.finally(() => setIsLoadingHistory(false))
.finally(() => {
if (!disposed) {
setIsLoadingHistory(false)
}
})
return () => {
window.electronAPI.localChat.offEvent()
window.electronAPI.localChat.offApproval()
window.electronAPI.hub.offInboundMessage()
window.electronAPI.localChat.unsubscribe(activeConversationId).catch(() => {})
disposed = true
unsubscribeEvent?.()
unsubscribeApproval?.()
unsubscribeInbound?.()
void subscribePromise
.then((result) => {
if (typeof result?.token !== 'number') return
return window.electronAPI.localChat.unsubscribe(activeConversationId, result.token)
})
.catch(() => {})
}
}, [activeConversationId])
@ -213,12 +276,28 @@ export function useLocalChat(options: UseLocalChatOptions = {}) {
}, [])
useEffect(() => {
if (!activeConversationId || isLoading || queuedMessages.length === 0) return
const next = queuedMessages[0]
if (!next) return
if (!activeConversationId || isLoading) return
// Inbound channel/gateway messages are already queued in backend.
// Render them first to keep frontend ordering aligned with agent run order.
const nextInbound = queuedInboundMessages[0]
if (nextInbound) {
setQueuedInboundMessages((prev) => prev.slice(1))
chatRef.current.addUserMessage(
nextInbound.content,
nextInbound.agentId,
nextInbound.conversationId,
nextInbound.source,
)
setIsLoading(true)
return
}
const nextLocal = queuedMessages[0]
if (!nextLocal) return
setQueuedMessages((prev) => prev.slice(1))
dispatchMessageNow(next.text)
}, [activeConversationId, isLoading, queuedMessages, dispatchMessageNow])
dispatchMessageNow(nextLocal.text)
}, [activeConversationId, isLoading, queuedInboundMessages, queuedMessages, dispatchMessageNow])
const abortGeneration = useCallback(() => {
if (!activeConversationId) return

6
apps/desktop/src/renderer/src/stores/hub.ts
View file

@ -63,6 +63,12 @@ export const useHubStore = create<HubStore>()((set, get) => ({
: prev.hubInfo,
}))
})
// Refresh conversation list when backend conversations change
// (e.g. Telegram / RPC creates a new session).
window.electronAPI.hub.onConversationsChanged(() => {
void get().refresh()
})
} catch (err) {
const message = err instanceof Error ? err.message : String(err)
set({ error: message })