From ed3c6d753410e98df5b03694f2a3461ecd55d328 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Tue, 2 Jun 2026 19:21:24 -0700 Subject: [PATCH] feat(backend_meet): integrate with backend gmeet bot via Socket.IO (#3182) --- .../__tests__/socketService.events.test.ts | 122 +++++++++ app/src/services/backendMeetService.test.ts | 121 +++++++++ app/src/services/meetCallService.ts | 65 +++++ app/src/services/socketService.ts | 63 +++++ app/src/store/backendMeetSlice.test.ts | 99 +++++++ app/src/store/backendMeetSlice.ts | 99 +++++++ app/src/store/index.ts | 2 + src/core/all.rs | 8 + src/core/event_bus/events.rs | 44 ++++ src/core/event_bus/mod.rs | 2 +- src/core/socketio.rs | 105 ++++++++ src/openhuman/agent_meetings/mod.rs | 22 ++ src/openhuman/agent_meetings/ops.rs | 242 ++++++++++++++++++ src/openhuman/agent_meetings/schemas.rs | 176 +++++++++++++ src/openhuman/agent_meetings/types.rs | 34 +++ src/openhuman/mod.rs | 1 + src/openhuman/socket/event_handlers.rs | 98 ++++++- 17 files changed, 1301 insertions(+), 2 deletions(-) create mode 100644 app/src/services/backendMeetService.test.ts create mode 100644 app/src/store/backendMeetSlice.test.ts create mode 100644 app/src/store/backendMeetSlice.ts create mode 100644 src/openhuman/agent_meetings/mod.rs create mode 100644 src/openhuman/agent_meetings/ops.rs create mode 100644 src/openhuman/agent_meetings/schemas.rs create mode 100644 src/openhuman/agent_meetings/types.rs diff --git a/app/src/services/__tests__/socketService.events.test.ts b/app/src/services/__tests__/socketService.events.test.ts index 07c126d78..d83ad364a 100644 --- a/app/src/services/__tests__/socketService.events.test.ts +++ b/app/src/services/__tests__/socketService.events.test.ts @@ -212,3 +212,125 @@ describe('socketService — socket event handler dispatches (lines 212, 230, 237 expect((disconnectedCall![0] as { error: string }).error).toBe('connection refused'); }); }); + +describe('socketService — agent_meetings event handlers (lines 428-480)', () => { + beforeEach(() => { + vi.resetModules(); + storeMock.dispatch.mockClear(); + storeMock.getState.mockReturnValue({ + thread: { selectedThreadId: null, activeThreadId: null }, + }); + getCoreRpcUrlMock.mockReset(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('dispatches setBackendMeetJoined on agent_meetings:joined', async () => { + const { handlers, mockSocket } = buildMockSocket(); + vi.doMock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })); + getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.connect('jwt-test-meet-joined'); + + await pollUntil(() => expect(handlers['agent_meetings:joined']).toBeDefined()); + handlers['agent_meetings:joined']!({ meet_url: 'https://meet.google.com/abc' }); + + expect(storeMock.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ payload: { meetUrl: 'https://meet.google.com/abc' } }) + ); + }); + + it('dispatches setBackendMeetLeft on agent_meetings:left', async () => { + const { handlers, mockSocket } = buildMockSocket(); + vi.doMock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })); + getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.connect('jwt-test-meet-left'); + + await pollUntil(() => expect(handlers['agent_meetings:left']).toBeDefined()); + handlers['agent_meetings:left']!({ reason: 'call-ended' }); + + expect(storeMock.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ payload: { reason: 'call-ended' } }) + ); + }); + + it('dispatches setBackendMeetReply on agent_meetings:reply', async () => { + const { handlers, mockSocket } = buildMockSocket(); + vi.doMock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })); + getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.connect('jwt-test-meet-reply'); + + await pollUntil(() => expect(handlers['agent_meetings:reply']).toBeDefined()); + handlers['agent_meetings:reply']!({ transcript: 'hi', reply: 'hello', emotion: 'happy' }); + + expect(storeMock.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ payload: { transcript: 'hi', reply: 'hello', emotion: 'happy' } }) + ); + }); + + it('dispatches setBackendMeetHarness on agent_meetings:harness', async () => { + const { handlers, mockSocket } = buildMockSocket(); + vi.doMock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })); + getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.connect('jwt-test-meet-harness'); + + await pollUntil(() => expect(handlers['agent_meetings:harness']).toBeDefined()); + handlers['agent_meetings:harness']!({ + transcript: 'check email', + instruction: 'read inbox', + emotion: 'thinking', + }); + + expect(storeMock.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + payload: { transcript: 'check email', instruction: 'read inbox', emotion: 'thinking' }, + }) + ); + }); + + it('dispatches setBackendMeetTranscript on agent_meetings:transcript', async () => { + const { handlers, mockSocket } = buildMockSocket(); + vi.doMock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })); + getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.connect('jwt-test-meet-transcript'); + + await pollUntil(() => expect(handlers['agent_meetings:transcript']).toBeDefined()); + handlers['agent_meetings:transcript']!({ + turns: [{ role: 'user', content: 'hi' }], + duration_ms: 5000, + }); + + expect(storeMock.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + payload: { turns: [{ role: 'user', content: 'hi' }], duration_ms: 5000 }, + }) + ); + }); + + it('dispatches setBackendMeetError on agent_meetings:error', async () => { + const { handlers, mockSocket } = buildMockSocket(); + vi.doMock('socket.io-client', () => ({ io: vi.fn(() => mockSocket) })); + getCoreRpcUrlMock.mockResolvedValue('http://127.0.0.1:7788/rpc'); + + const { socketService } = await import('../socketService'); + socketService.connect('jwt-test-meet-error'); + + await pollUntil(() => expect(handlers['agent_meetings:error']).toBeDefined()); + handlers['agent_meetings:error']!({ error: 'bot crashed' }); + + expect(storeMock.dispatch).toHaveBeenCalledWith( + expect.objectContaining({ payload: { error: 'bot crashed' } }) + ); + }); +}); diff --git a/app/src/services/backendMeetService.test.ts b/app/src/services/backendMeetService.test.ts new file mode 100644 index 000000000..1f0e5e793 --- /dev/null +++ b/app/src/services/backendMeetService.test.ts @@ -0,0 +1,121 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { callCoreRpc } from './coreRpcClient'; +import { joinMeetViaBackendBot, leaveBackendMeetBot, sendHarnessResponse } from './meetCallService'; + +vi.mock('./coreRpcClient', () => ({ callCoreRpc: vi.fn() })); + +const mockCallCoreRpc = vi.mocked(callCoreRpc); + +beforeEach(() => { + vi.resetAllMocks(); +}); + +describe('joinMeetViaBackendBot', () => { + it('calls agent_meetings_join with correct params', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ + ok: true, + meet_url: 'https://meet.google.com/abc-defg-hij', + platform: 'gmeet', + }); + + const result = await joinMeetViaBackendBot({ meetUrl: 'https://meet.google.com/abc-defg-hij' }); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.agent_meetings_join', + params: { + meet_url: 'https://meet.google.com/abc-defg-hij', + display_name: undefined, + platform: undefined, + }, + }); + expect(result).toEqual({ meetUrl: 'https://meet.google.com/abc-defg-hij', platform: 'gmeet' }); + }); + + it('trims whitespace from meetUrl', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ + ok: true, + meet_url: 'https://meet.google.com/abc', + platform: 'gmeet', + }); + + await joinMeetViaBackendBot({ meetUrl: ' https://meet.google.com/abc ' }); + + expect(mockCallCoreRpc).toHaveBeenCalledWith( + expect.objectContaining({ + params: expect.objectContaining({ meet_url: 'https://meet.google.com/abc' }), + }) + ); + }); + + it('throws on empty meetUrl', async () => { + await expect(joinMeetViaBackendBot({ meetUrl: ' ' })).rejects.toThrow( + 'Please paste a meeting link.' + ); + expect(mockCallCoreRpc).not.toHaveBeenCalled(); + }); + + it('throws when core rejects', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ ok: false }); + + await expect(joinMeetViaBackendBot({ meetUrl: 'https://meet.google.com/abc' })).rejects.toThrow( + 'Core rejected' + ); + }); + + it('forwards displayName and platform', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ + ok: true, + meet_url: 'https://zoom.us/j/123', + platform: 'zoom', + }); + + await joinMeetViaBackendBot({ + meetUrl: 'https://zoom.us/j/123', + displayName: 'Bot', + platform: 'zoom', + }); + + expect(mockCallCoreRpc).toHaveBeenCalledWith( + expect.objectContaining({ + params: expect.objectContaining({ display_name: 'Bot', platform: 'zoom' }), + }) + ); + }); +}); + +describe('leaveBackendMeetBot', () => { + it('calls agent_meetings_leave', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ ok: true }); + + await leaveBackendMeetBot('user-requested'); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.agent_meetings_leave', + params: { reason: 'user-requested' }, + }); + }); + + it('defaults reason to "requested"', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ ok: true }); + + await leaveBackendMeetBot(); + + expect(mockCallCoreRpc).toHaveBeenCalledWith( + expect.objectContaining({ params: { reason: 'requested' } }) + ); + }); +}); + +describe('sendHarnessResponse', () => { + it('calls agent_meetings_harness_response', async () => { + mockCallCoreRpc.mockResolvedValueOnce({ ok: true }); + + await sendHarnessResponse('tool output here'); + + expect(mockCallCoreRpc).toHaveBeenCalledWith({ + method: 'openhuman.agent_meetings_harness_response', + params: { result: 'tool output here' }, + }); + }); +}); diff --git a/app/src/services/meetCallService.ts b/app/src/services/meetCallService.ts index aa29eb3ce..71ecbc8a9 100644 --- a/app/src/services/meetCallService.ts +++ b/app/src/services/meetCallService.ts @@ -154,6 +154,71 @@ export async function listMeetCalls(limit = 20): Promise { return result.calls ?? []; } +// --------------------------------------------------------------------------- +// Backend Meet Bot via Core Socket.IO bridge +// --------------------------------------------------------------------------- + +export type MeetingPlatform = 'gmeet' | 'zoom' | 'teams' | 'webex'; + +export type BackendMeetJoinInput = { + meetUrl: string; + displayName?: string; + platform?: MeetingPlatform; +}; + +type CoreBackendMeetJoinResponse = { ok: boolean; meet_url: string; platform: string }; + +/** + * Join a meeting via the backend's Recall.ai bot. Supports Google Meet, + * Zoom, Microsoft Teams, and Webex. + * + * Calls the core RPC `openhuman.agent_meetings_join`, which emits `bot:join` + * over the core's persistent Socket.IO connection to the backend. The backend + * streams events back (`bot:reply`, `bot:harness`, `bot:transcript`, `bot:left`) + * which the core bridges to the frontend as `agent_meetings:*` socket events. + */ +export async function joinMeetViaBackendBot( + input: BackendMeetJoinInput +): Promise<{ meetUrl: string; platform: string }> { + const meetUrl = input.meetUrl.trim(); + if (!meetUrl) throw new Error('Please paste a meeting link.'); + + const result = await callCoreRpc({ + method: 'openhuman.agent_meetings_join', + params: { + meet_url: meetUrl, + display_name: input.displayName?.trim() || undefined, + platform: input.platform || undefined, + }, + }); + + if (!result?.ok) { + throw new Error('Core rejected the agent_meetings_join request.'); + } + + return { meetUrl: result.meet_url, platform: result.platform }; +} + +/** + * Ask the backend bot to leave the current meeting. + */ +export async function leaveBackendMeetBot(reason?: string): Promise { + await callCoreRpc<{ ok: boolean }>({ + method: 'openhuman.agent_meetings_leave', + params: { reason: reason || 'requested' }, + }); +} + +/** + * Send a tool execution result back to the backend's meeting LLM. + */ +export async function sendHarnessResponse(result: string): Promise { + await callCoreRpc<{ ok: boolean }>({ + method: 'openhuman.agent_meetings_harness_response', + params: { result }, + }); +} + /** * Backend-driven meet bot join (PR tinyhumansai/backend#773). * diff --git a/app/src/services/socketService.ts b/app/src/services/socketService.ts index 4701d8970..a535caaa3 100644 --- a/app/src/services/socketService.ts +++ b/app/src/services/socketService.ts @@ -4,6 +4,14 @@ import { type Socket } from 'socket.io-client'; import { getCoreStateSnapshot } from '../lib/coreState/store'; import { SocketIOMCPTransportImpl } from '../lib/mcp'; import { store } from '../store'; +import { + setBackendMeetError, + setBackendMeetHarness, + setBackendMeetJoined, + setBackendMeetLeft, + setBackendMeetReply, + setBackendMeetTranscript, +} from '../store/backendMeetSlice'; import { upsertChannelConnection } from '../store/channelConnectionsSlice'; import { type CompanionStateChangedEvent, setCompanionState } from '../store/companionSlice'; import { setBackend } from '../store/connectivitySlice'; @@ -416,6 +424,61 @@ class SocketService { store.dispatch(setCompanionState(event)); }); + // Backend Meet bot events — forwarded from core's DomainEvent bus + this.socket.on('agent_meetings:joined', (data: unknown) => { + const obj = data as Record | null; + const meetUrl = typeof obj?.meet_url === 'string' ? obj.meet_url : ''; + socketLog('agent_meetings:joined meet_url_len=%d', meetUrl.length); + store.dispatch(setBackendMeetJoined({ meetUrl })); + }); + this.socket.on('agent_meetings:left', (data: unknown) => { + const obj = data as Record | null; + const reason = typeof obj?.reason === 'string' ? obj.reason : 'unknown'; + socketLog('agent_meetings:left reason=%s', reason); + store.dispatch(setBackendMeetLeft({ reason })); + }); + this.socket.on('agent_meetings:reply', (data: unknown) => { + const obj = data as Record | null; + if (!obj) return; + socketLog('agent_meetings:reply'); + store.dispatch( + setBackendMeetReply({ + transcript: typeof obj.transcript === 'string' ? obj.transcript : '', + reply: typeof obj.reply === 'string' ? obj.reply : '', + emotion: typeof obj.emotion === 'string' ? obj.emotion : 'neutral', + }) + ); + }); + this.socket.on('agent_meetings:harness', (data: unknown) => { + const obj = data as Record | null; + if (!obj) return; + socketLog('agent_meetings:harness'); + store.dispatch( + setBackendMeetHarness({ + transcript: typeof obj.transcript === 'string' ? obj.transcript : '', + instruction: typeof obj.instruction === 'string' ? obj.instruction : '', + emotion: typeof obj.emotion === 'string' ? obj.emotion : 'neutral', + }) + ); + }); + this.socket.on('agent_meetings:transcript', (data: unknown) => { + const obj = data as Record | null; + if (!obj) return; + socketLog('agent_meetings:transcript'); + store.dispatch( + setBackendMeetTranscript({ + turns: Array.isArray(obj.turns) ? obj.turns : [], + duration_ms: typeof obj.duration_ms === 'number' ? obj.duration_ms : 0, + }) + ); + }); + this.socket.on('agent_meetings:error', (data: unknown) => { + const obj = data as Record | null; + const error = typeof obj?.error === 'string' ? obj.error : 'Unknown error'; + socketError('agent_meetings:error %s', error); + store.dispatch(setBackendMeetError({ error })); + }); + this.socket.connect(); } diff --git a/app/src/store/backendMeetSlice.test.ts b/app/src/store/backendMeetSlice.test.ts new file mode 100644 index 000000000..f9aecd8d9 --- /dev/null +++ b/app/src/store/backendMeetSlice.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, it } from 'vitest'; + +import backendMeetReducer, { + resetBackendMeet, + setBackendMeetError, + setBackendMeetHarness, + setBackendMeetJoined, + setBackendMeetJoining, + setBackendMeetLeft, + setBackendMeetReply, + setBackendMeetTranscript, +} from './backendMeetSlice'; + +const initial = backendMeetReducer(undefined, { type: 'init' }); + +describe('backendMeetSlice', () => { + it('starts in idle state', () => { + expect(initial.status).toBe('idle'); + expect(initial.meetUrl).toBeNull(); + expect(initial.lastReply).toBeNull(); + expect(initial.transcript).toBeNull(); + }); + + it('transitions to joining', () => { + const state = backendMeetReducer( + initial, + setBackendMeetJoining({ meetUrl: 'https://meet.google.com/abc-defg-hij' }) + ); + expect(state.status).toBe('joining'); + expect(state.meetUrl).toBe('https://meet.google.com/abc-defg-hij'); + }); + + it('transitions to active on joined', () => { + const joining = backendMeetReducer( + initial, + setBackendMeetJoining({ meetUrl: 'https://meet.google.com/abc-defg-hij' }) + ); + const state = backendMeetReducer( + joining, + setBackendMeetJoined({ meetUrl: 'https://meet.google.com/abc-defg-hij' }) + ); + expect(state.status).toBe('active'); + }); + + it('transitions to ended on left', () => { + const active = backendMeetReducer( + backendMeetReducer(initial, setBackendMeetJoined({ meetUrl: 'x' })), + setBackendMeetLeft({ reason: 'call-ended' }) + ); + expect(active.status).toBe('ended'); + }); + + it('stores reply events', () => { + const state = backendMeetReducer( + initial, + setBackendMeetReply({ transcript: 'Hey bot', reply: 'Hello!', emotion: 'happy' }) + ); + expect(state.lastReply).toEqual({ transcript: 'Hey bot', reply: 'Hello!', emotion: 'happy' }); + }); + + it('stores harness events', () => { + const state = backendMeetReducer( + initial, + setBackendMeetHarness({ + transcript: 'Check my email', + instruction: 'read 5 latest emails', + emotion: 'thinking', + }) + ); + expect(state.lastHarness?.instruction).toBe('read 5 latest emails'); + }); + + it('stores transcript on close', () => { + const state = backendMeetReducer( + initial, + setBackendMeetTranscript({ + turns: [ + { role: 'user', content: 'Hello' }, + { role: 'assistant', content: 'Hi there!' }, + ], + duration_ms: 120000, + }) + ); + expect(state.transcript?.turns).toHaveLength(2); + expect(state.transcript?.duration_ms).toBe(120000); + }); + + it('stores error', () => { + const state = backendMeetReducer(initial, setBackendMeetError({ error: 'connection failed' })); + expect(state.status).toBe('error'); + expect(state.error).toBe('connection failed'); + }); + + it('resets to initial state', () => { + const active = backendMeetReducer(initial, setBackendMeetJoined({ meetUrl: 'x' })); + const state = backendMeetReducer(active, resetBackendMeet()); + expect(state).toEqual(initial); + }); +}); diff --git a/app/src/store/backendMeetSlice.ts b/app/src/store/backendMeetSlice.ts new file mode 100644 index 000000000..cc44b61c4 --- /dev/null +++ b/app/src/store/backendMeetSlice.ts @@ -0,0 +1,99 @@ +import { createSlice, type PayloadAction } from '@reduxjs/toolkit'; + +import { resetUserScopedState } from './resetActions'; + +export type BackendMeetStatus = 'idle' | 'joining' | 'active' | 'ended' | 'error'; + +export interface BackendMeetTurn { + role: string; + content: string; +} + +export interface BackendMeetReplyEvent { + transcript: string; + reply: string; + emotion: string; +} + +export interface BackendMeetHarnessEvent { + transcript: string; + instruction: string; + emotion: string; +} + +export interface BackendMeetTranscriptEvent { + turns: BackendMeetTurn[]; + duration_ms: number; +} + +interface BackendMeetState { + status: BackendMeetStatus; + meetUrl: string | null; + lastReply: BackendMeetReplyEvent | null; + lastHarness: BackendMeetHarnessEvent | null; + transcript: BackendMeetTranscriptEvent | null; + error: string | null; +} + +const initialState: BackendMeetState = { + status: 'idle', + meetUrl: null, + lastReply: null, + lastHarness: null, + transcript: null, + error: null, +}; + +const backendMeetSlice = createSlice({ + name: 'backendMeet', + initialState, + reducers: { + setBackendMeetJoining(state, action: PayloadAction<{ meetUrl: string }>) { + state.status = 'joining'; + state.meetUrl = action.payload.meetUrl; + state.error = null; + state.lastReply = null; + state.lastHarness = null; + state.transcript = null; + }, + setBackendMeetJoined(state, action: PayloadAction<{ meetUrl: string }>) { + state.status = 'active'; + state.meetUrl = action.payload.meetUrl; + }, + setBackendMeetLeft(state, _action: PayloadAction<{ reason: string }>) { + state.status = 'ended'; + }, + setBackendMeetReply(state, action: PayloadAction) { + state.lastReply = action.payload; + }, + setBackendMeetHarness(state, action: PayloadAction) { + state.lastHarness = action.payload; + }, + setBackendMeetTranscript(state, action: PayloadAction) { + state.transcript = action.payload; + }, + setBackendMeetError(state, action: PayloadAction<{ error: string }>) { + state.status = 'error'; + state.error = action.payload.error; + }, + resetBackendMeet() { + return initialState; + }, + }, + extraReducers: builder => { + builder.addCase(resetUserScopedState, () => initialState); + }, +}); + +export const { + setBackendMeetJoining, + setBackendMeetJoined, + setBackendMeetLeft, + setBackendMeetReply, + setBackendMeetHarness, + setBackendMeetTranscript, + setBackendMeetError, + resetBackendMeet, +} = backendMeetSlice.actions; + +export default backendMeetSlice.reducer; diff --git a/app/src/store/index.ts b/app/src/store/index.ts index a052ec696..b47701888 100644 --- a/app/src/store/index.ts +++ b/app/src/store/index.ts @@ -20,6 +20,7 @@ import { filterArtifactsForPersist, rehydrateArtifactsFromPersist, } from './artifactsPersistFilter'; +import backendMeetReducer from './backendMeetSlice'; import channelConnectionsReducer from './channelConnectionsSlice'; import chatRuntimeReducer from './chatRuntimeSlice'; import companionReducer from './companionSlice'; @@ -184,6 +185,7 @@ const persistedChatRuntimeReducer = persistReducer(chatRuntimePersistConfig, cha export const store = configureStore({ reducer: { + backendMeet: backendMeetReducer, socket: socketReducer, connectivity: connectivityReducer, thread: persistedThreadReducer, diff --git a/src/core/all.rs b/src/core/all.rs index ce36f9d85..89a6fd8e2 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -264,6 +264,9 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::notifications::all_notifications_registered_controllers()); // Google Meet call-join request validation (shell handles the webview) controllers.extend(crate::openhuman::meet::all_meet_registered_controllers()); + // Agent meetings — backend-delegated Meet bot via Socket.IO + controllers + .extend(crate::openhuman::agent_meetings::all_agent_meetings_registered_controllers()); // Live meet-agent loop: STT/LLM/TTS over the open call's audio. controllers.extend(crate::openhuman::meet_agent::all_meet_agent_registered_controllers()); // Desktop companion — Clicky-style interaction loop. @@ -383,6 +386,8 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::notifications::all_notifications_controller_schemas()); // Google Meet call-join request validation schemas.extend(crate::openhuman::meet::all_meet_controller_schemas()); + // Agent meetings — backend-delegated Meet bot via Socket.IO + schemas.extend(crate::openhuman::agent_meetings::all_agent_meetings_controller_schemas()); // Live meet-agent listening + speaking loop schemas.extend(crate::openhuman::meet_agent::all_meet_agent_controller_schemas()); // Desktop companion — Clicky-style interaction loop. @@ -514,6 +519,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "Live agent loop for an open Google Meet call: shell streams inbound PCM, \ core runs VAD-segmented STT → LLM → TTS, shell pulls synthesized PCM back.", ), + "agent_meetings" => Some( + "Backend-delegated meeting bot (Google Meet, Zoom, Teams, Webex) via Socket.IO — join, leave, and harness response.", + ), "devices" => Some( "Paired mobile device management — pairing channel creation, listing, and revocation.", ), diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 502636d2d..7f2961d9e 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -813,6 +813,37 @@ pub enum DomainEvent { /// deliberate follow-up; emitting the event now lets that bridge attach /// without a schema change. TaskPlanAwaitingApproval { card_id: String, thread_id: String }, + + // ── Backend Meet Bot ────────────────────────────────────────────── + /// Backend gmeet bot successfully joined the meeting. + BackendMeetJoined { meet_url: String }, + /// Backend gmeet bot left the meeting. + BackendMeetLeft { reason: String }, + /// Backend gmeet bot produced a spoken reply. + BackendMeetReply { + transcript: String, + reply: String, + emotion: String, + }, + /// Backend gmeet bot needs the harness to execute a tool instruction. + BackendMeetHarness { + transcript: String, + instruction: String, + emotion: String, + }, + /// Backend gmeet bot sent the full meeting transcript on close. + BackendMeetTranscript { + turns: Vec, + duration_ms: u64, + }, + /// Backend gmeet bot emitted an error. + BackendMeetError { error: String }, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct BackendMeetTurn { + pub role: String, + pub content: String, } impl DomainEvent { @@ -925,6 +956,13 @@ impl DomainEvent { | Self::McpClientToolExecuted { .. } | Self::McpSetupSecretRequested { .. } | Self::McpToolRejected { .. } => "mcp_client", + + Self::BackendMeetJoined { .. } + | Self::BackendMeetLeft { .. } + | Self::BackendMeetReply { .. } + | Self::BackendMeetHarness { .. } + | Self::BackendMeetTranscript { .. } + | Self::BackendMeetError { .. } => "agent_meetings", } } @@ -1021,6 +1059,12 @@ impl DomainEvent { Self::TaskSourceTaskIngested { .. } => "TaskSourceTaskIngested", Self::TaskSourceFetchFailed { .. } => "TaskSourceFetchFailed", Self::TaskPlanAwaitingApproval { .. } => "TaskPlanAwaitingApproval", + Self::BackendMeetJoined { .. } => "BackendMeetJoined", + Self::BackendMeetLeft { .. } => "BackendMeetLeft", + Self::BackendMeetReply { .. } => "BackendMeetReply", + Self::BackendMeetHarness { .. } => "BackendMeetHarness", + Self::BackendMeetTranscript { .. } => "BackendMeetTranscript", + Self::BackendMeetError { .. } => "BackendMeetError", } } diff --git a/src/core/event_bus/mod.rs b/src/core/event_bus/mod.rs index 1dc7c85d8..637631105 100644 --- a/src/core/event_bus/mod.rs +++ b/src/core/event_bus/mod.rs @@ -61,7 +61,7 @@ pub mod testing; mod tracing; pub use bus::{global, init_global, publish_global, subscribe_global, EventBus, DEFAULT_CAPACITY}; -pub use events::DomainEvent; +pub use events::{BackendMeetTurn, DomainEvent}; pub use native_request::{ init_native_registry, native_registry, register_native_global, request_native_global, NativeRegistry, NativeRequestError, diff --git a/src/core/socketio.rs b/src/core/socketio.rs index 3e0c8374c..0894d6679 100644 --- a/src/core/socketio.rs +++ b/src/core/socketio.rs @@ -552,6 +552,7 @@ pub fn spawn_web_channel_bridge(io: SocketIo) { let io_companion = io.clone(); let io_mcp_setup = io.clone(); let io_memory_sync = io.clone(); + let io_agent_meetings = io.clone(); // 2. Dictation hotkey events → broadcast to all connected clients. tokio::spawn(async move { @@ -921,6 +922,110 @@ pub fn spawn_web_channel_bridge(io: SocketIo) { } log::debug!("[socketio] memory_sync bridge stopped"); }); + + // 9. Backend Meet bot events → broadcast to all connected frontend sockets. + tokio::spawn(async move { + let bus = { + const RETRY_INTERVAL_MS: u64 = 250; + const MAX_WAIT_SECS: u64 = 30; + let max_attempts = (MAX_WAIT_SECS * 1000) / RETRY_INTERVAL_MS; + let mut attempts: u64 = 0; + loop { + if let Some(bus) = crate::core::event_bus::global() { + break bus; + } + attempts += 1; + if attempts > max_attempts { + log::warn!( + "[socketio] event_bus not initialised after {}s — agent_meetings bridge giving up", + MAX_WAIT_SECS + ); + return; + } + tokio::time::sleep(std::time::Duration::from_millis(RETRY_INTERVAL_MS)).await; + } + }; + let mut rx = bus.raw_receiver(); + loop { + let event = match rx.recv().await { + Ok(event) => event, + Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => { + log::warn!( + "[socketio] dropped {} event_bus events due to lag (agent_meetings bridge)", + skipped + ); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + }; + match event { + crate::core::event_bus::DomainEvent::BackendMeetJoined { meet_url } => { + let payload = serde_json::json!({ "meet_url": meet_url }); + log::debug!("[socketio] broadcast agent_meetings:joined"); + let _ = io_agent_meetings.emit("agent_meetings:joined", &payload); + } + crate::core::event_bus::DomainEvent::BackendMeetLeft { reason } => { + let payload = serde_json::json!({ "reason": reason }); + log::debug!("[socketio] broadcast agent_meetings:left reason={}", reason); + let _ = io_agent_meetings.emit("agent_meetings:left", &payload); + } + crate::core::event_bus::DomainEvent::BackendMeetReply { + transcript, + reply, + emotion, + } => { + let payload = serde_json::json!({ + "transcript": transcript, + "reply": reply, + "emotion": emotion, + }); + log::debug!( + "[socketio] broadcast agent_meetings:reply reply_len={}", + reply.len() + ); + let _ = io_agent_meetings.emit("agent_meetings:reply", &payload); + } + crate::core::event_bus::DomainEvent::BackendMeetHarness { + transcript, + instruction, + emotion, + } => { + let payload = serde_json::json!({ + "transcript": transcript, + "instruction": instruction, + "emotion": emotion, + }); + log::debug!( + "[socketio] broadcast agent_meetings:harness instruction_len={}", + instruction.len() + ); + let _ = io_agent_meetings.emit("agent_meetings:harness", &payload); + } + crate::core::event_bus::DomainEvent::BackendMeetTranscript { + turns, + duration_ms, + } => { + let payload = serde_json::json!({ + "turns": turns, + "duration_ms": duration_ms, + }); + log::debug!( + "[socketio] broadcast agent_meetings:transcript turns={} duration_ms={}", + turns.len(), + duration_ms + ); + let _ = io_agent_meetings.emit("agent_meetings:transcript", &payload); + } + crate::core::event_bus::DomainEvent::BackendMeetError { error } => { + let payload = serde_json::json!({ "error": error }); + log::debug!("[socketio] broadcast agent_meetings:error"); + let _ = io_agent_meetings.emit("agent_meetings:error", &payload); + } + _ => {} + } + } + log::debug!("[socketio] agent_meetings bridge stopped"); + }); } /// Join `socket` to `room`, logging the result. diff --git a/src/openhuman/agent_meetings/mod.rs b/src/openhuman/agent_meetings/mod.rs new file mode 100644 index 000000000..09538370e --- /dev/null +++ b/src/openhuman/agent_meetings/mod.rs @@ -0,0 +1,22 @@ +//! Agent Meetings integration domain. +//! +//! Delegates Google Meet bot joining/leaving to the TinyHumans backend +//! via the existing Socket.IO connection (`SocketManager`). The backend +//! runs a Camoufox headless browser that joins the meeting, captures +//! captions, and streams LLM decisions back over Socket.IO events +//! (`bot:reply`, `bot:harness`, `bot:transcript`). +//! +//! ## Module layout +//! +//! - [`types`] — request/response types +//! - [`ops`] — RPC handlers that emit Socket.IO events +//! - [`schemas`] — controller schema + registered handler wrappers + +pub mod ops; +pub mod schemas; +pub mod types; + +pub use schemas::{ + all_controller_schemas as all_agent_meetings_controller_schemas, + all_registered_controllers as all_agent_meetings_registered_controllers, +}; diff --git a/src/openhuman/agent_meetings/ops.rs b/src/openhuman/agent_meetings/ops.rs new file mode 100644 index 000000000..d9ba65085 --- /dev/null +++ b/src/openhuman/agent_meetings/ops.rs @@ -0,0 +1,242 @@ +//! RPC handlers for the `agent_meetings` domain. +//! +//! Each handler emits a Socket.IO event to the backend via the global +//! `SocketManager`. The backend's meeting bot handler picks these up and +//! drives the Recall.ai (or Camoufox) session. + +use serde_json::{json, Map, Value}; + +use crate::openhuman::meet::ops::validate_display_name; +use crate::openhuman::socket::global_socket_manager; +use crate::rpc::RpcOutcome; + +use super::types::{ + BackendMeetHarnessResponseRequest, BackendMeetJoinRequest, BackendMeetJoinResponse, + BackendMeetLeaveRequest, +}; + +const ALLOWED_HOSTS: &[(&str, &str)] = &[ + ("meet.google.com", "gmeet"), + ("zoom.us", "zoom"), + ("teams.microsoft.com", "teams"), + ("webex.com", "webex"), +]; + +fn validate_meeting_url(raw: &str) -> Result { + let url = url::Url::parse(raw.trim()).map_err(|e| format!("invalid meeting URL: {e}"))?; + + if url.scheme() != "https" && url.scheme() != "http" { + return Err(format!( + "invalid meeting URL: scheme `{}` not allowed", + url.scheme() + )); + } + + let host = url + .host_str() + .ok_or_else(|| "invalid meeting URL: missing host".to_string())?; + + let is_allowed = ALLOWED_HOSTS + .iter() + .any(|(allowed, _)| host == *allowed || host.ends_with(&format!(".{allowed}"))); + + if !is_allowed { + return Err(format!( + "invalid meeting URL: host `{host}` not recognized (supported: Google Meet, Zoom, Teams, Webex)" + )); + } + + Ok(url) +} + +fn infer_platform(url: &url::Url) -> &'static str { + let host = url.host_str().unwrap_or(""); + for (allowed, platform) in ALLOWED_HOSTS { + if host == *allowed || host.ends_with(&format!(".{allowed}")) { + return platform; + } + } + "gmeet" +} + +/// Handle `openhuman.agent_meetings_join`. +pub async fn handle_join(params: Map) -> Result { + let req: BackendMeetJoinRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("[agent_meetings] invalid join params: {e}"))?; + + let normalized_url = + validate_meeting_url(&req.meet_url).map_err(|e| format!("[agent_meetings] {e}"))?; + + let display_name = match &req.display_name { + Some(name) => validate_display_name(name).map_err(|e| format!("[agent_meetings] {e}"))?, + None => "OpenHuman".to_string(), + }; + + let inferred = infer_platform(&normalized_url); + let platform = match req.platform.as_deref() { + Some(p) if p != inferred => { + return Err(format!( + "[agent_meetings] platform mismatch: URL implies `{inferred}` but `{p}` was supplied" + )); + } + Some(p) => p, + None => inferred, + }; + + let mgr = global_socket_manager() + .ok_or_else(|| "[agent_meetings] socket not connected to backend".to_string())?; + + if !mgr.is_connected() { + return Err("[agent_meetings] socket not connected to backend".to_string()); + } + + tracing::info!( + meet_url_host = %normalized_url.host_str().unwrap_or(""), + platform = %platform, + display_name_len = display_name.len(), + "[agent_meetings] emitting bot:join" + ); + + mgr.emit( + "bot:join", + json!({ + "meetUrl": normalized_url.as_str(), + "displayName": display_name, + "platform": platform, + }), + ) + .await + .map_err(|e| format!("[agent_meetings] emit failed: {e}"))?; + + let response = BackendMeetJoinResponse { + ok: true, + meet_url: normalized_url.to_string(), + platform: platform.to_string(), + }; + let outcome = RpcOutcome::new( + serde_json::to_value(response).map_err(|e| format!("[agent_meetings] serialize: {e}"))?, + vec![], + ); + outcome.into_cli_compatible_json() +} + +/// Handle `openhuman.agent_meetings_leave`. +pub async fn handle_leave(params: Map) -> Result { + let req: BackendMeetLeaveRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("[agent_meetings] invalid leave params: {e}"))?; + + let mgr = global_socket_manager() + .ok_or_else(|| "[agent_meetings] socket not connected to backend".to_string())?; + + if !mgr.is_connected() { + return Err("[agent_meetings] socket not connected to backend".to_string()); + } + + let reason = req.reason.unwrap_or_else(|| "requested".to_string()); + + tracing::info!(reason = %reason, "[agent_meetings] emitting bot:leave"); + + mgr.emit("bot:leave", json!({ "reason": reason })) + .await + .map_err(|e| format!("[agent_meetings] emit failed: {e}"))?; + + let outcome = RpcOutcome::new(json!({ "ok": true }), vec![]); + outcome.into_cli_compatible_json() +} + +/// Handle `openhuman.agent_meetings_harness_response`. +pub async fn handle_harness_response(params: Map) -> Result { + let req: BackendMeetHarnessResponseRequest = serde_json::from_value(Value::Object(params)) + .map_err(|e| format!("[agent_meetings] invalid harness_response params: {e}"))?; + + if req.result.trim().is_empty() { + return Err("[agent_meetings] result must not be empty".to_string()); + } + + let mgr = global_socket_manager() + .ok_or_else(|| "[agent_meetings] socket not connected to backend".to_string())?; + + if !mgr.is_connected() { + return Err("[agent_meetings] socket not connected to backend".to_string()); + } + + tracing::info!( + result_len = req.result.len(), + "[agent_meetings] emitting bot:harness:response" + ); + + mgr.emit("bot:harness:response", json!({ "result": req.result })) + .await + .map_err(|e| format!("[agent_meetings] emit failed: {e}"))?; + + let outcome = RpcOutcome::new(json!({ "ok": true }), vec![]); + outcome.into_cli_compatible_json() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_google_meet_url() { + validate_meeting_url("https://meet.google.com/abc-defg-hij").unwrap(); + } + + #[test] + fn accepts_zoom_url() { + validate_meeting_url("https://zoom.us/j/123456789").unwrap(); + validate_meeting_url("https://company.zoom.us/j/123456789").unwrap(); + } + + #[test] + fn accepts_teams_url() { + validate_meeting_url("https://teams.microsoft.com/l/meetup-join/abc").unwrap(); + } + + #[test] + fn accepts_webex_url() { + validate_meeting_url("https://meet.webex.com/meet/abc").unwrap(); + validate_meeting_url("https://company.webex.com/meet/abc").unwrap(); + } + + #[test] + fn rejects_unknown_host() { + assert!(validate_meeting_url("https://example.com/meeting").is_err()); + } + + #[test] + fn infers_platform_from_host() { + let url = url::Url::parse("https://meet.google.com/abc-defg-hij").unwrap(); + assert_eq!(infer_platform(&url), "gmeet"); + + let url = url::Url::parse("https://zoom.us/j/123").unwrap(); + assert_eq!(infer_platform(&url), "zoom"); + + let url = url::Url::parse("https://teams.microsoft.com/l/meetup").unwrap(); + assert_eq!(infer_platform(&url), "teams"); + + let url = url::Url::parse("https://meet.webex.com/meet/abc").unwrap(); + assert_eq!(infer_platform(&url), "webex"); + + let url = url::Url::parse("https://company.zoom.us/j/123").unwrap(); + assert_eq!(infer_platform(&url), "zoom"); + } + + #[tokio::test] + async fn join_fails_when_socket_not_connected() { + let params: Map = + serde_json::from_value(json!({"meet_url": "https://meet.google.com/abc-defg-hij"})) + .unwrap(); + let result = handle_join(params).await; + assert!(result.is_err()); + assert!(result.unwrap_err().contains("socket not connected")); + } + + #[tokio::test] + async fn harness_response_rejects_empty_result() { + let params: Map = serde_json::from_value(json!({"result": " "})).unwrap(); + let result = handle_harness_response(params).await; + assert!(result.is_err()); + assert!(result.unwrap_err().contains("must not be empty")); + } +} diff --git a/src/openhuman/agent_meetings/schemas.rs b/src/openhuman/agent_meetings/schemas.rs new file mode 100644 index 000000000..690837847 --- /dev/null +++ b/src/openhuman/agent_meetings/schemas.rs @@ -0,0 +1,176 @@ +//! Controller schema definitions and registered handlers for the +//! `agent_meetings` domain. + +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; + +type SchemaBuilder = fn() -> ControllerSchema; +type ControllerHandler = fn(Map) -> ControllerFuture; + +struct BackendMeetControllerDef { + function: &'static str, + schema: SchemaBuilder, + handler: ControllerHandler, +} + +const DEFS: &[BackendMeetControllerDef] = &[ + BackendMeetControllerDef { + function: "join", + schema: schema_join, + handler: handle_join_wrap, + }, + BackendMeetControllerDef { + function: "leave", + schema: schema_leave, + handler: handle_leave_wrap, + }, + BackendMeetControllerDef { + function: "harness_response", + schema: schema_harness_response, + handler: handle_harness_response_wrap, + }, +]; + +pub fn all_controller_schemas() -> Vec { + DEFS.iter().map(|def| (def.schema)()).collect() +} + +pub fn all_registered_controllers() -> Vec { + DEFS.iter() + .map(|def| RegisteredController { + schema: (def.schema)(), + handler: def.handler, + }) + .collect() +} + +fn schema_join() -> ControllerSchema { + ControllerSchema { + namespace: "agent_meetings", + function: "join", + description: "Ask the backend to join a meeting via Recall.ai bot. Supports \ + Google Meet, Zoom, Teams, and Webex. Emits bot:join over Socket.IO; \ + the backend streams events back (bot:reply, bot:harness, bot:transcript, bot:left).", + inputs: vec![ + FieldSchema { + name: "meet_url", + ty: TypeSchema::String, + comment: "Meeting URL (Google Meet, Zoom, Teams, or Webex).", + required: true, + }, + FieldSchema { + name: "display_name", + ty: TypeSchema::String, + comment: "Display name for the bot in the meeting. Defaults to OpenHuman.", + required: false, + }, + FieldSchema { + name: "platform", + ty: TypeSchema::String, + comment: "Platform: gmeet, zoom, teams, or webex. Auto-detected from URL if omitted.", + required: false, + }, + ], + outputs: vec![ + FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the join request was emitted.", + required: true, + }, + FieldSchema { + name: "meet_url", + ty: TypeSchema::String, + comment: "Normalized meeting URL.", + required: true, + }, + FieldSchema { + name: "platform", + ty: TypeSchema::String, + comment: "Resolved platform: gmeet, zoom, teams, or webex.", + required: true, + }, + ], + } +} + +fn schema_leave() -> ControllerSchema { + ControllerSchema { + namespace: "agent_meetings", + function: "leave", + description: "Ask the backend bot to leave the current meeting.", + inputs: vec![FieldSchema { + name: "reason", + ty: TypeSchema::String, + comment: "Optional leave reason. Defaults to 'requested'.", + required: false, + }], + outputs: vec![FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the leave request was emitted.", + required: true, + }], + } +} + +fn schema_harness_response() -> ControllerSchema { + ControllerSchema { + namespace: "agent_meetings", + function: "harness_response", + description: "Send a tool execution result back to the backend's meeting LLM so \ + it can incorporate the result in the next conversation turn.", + inputs: vec![FieldSchema { + name: "result", + ty: TypeSchema::String, + comment: "The tool execution result text.", + required: true, + }], + outputs: vec![FieldSchema { + name: "ok", + ty: TypeSchema::Bool, + comment: "True when the response was emitted.", + required: true, + }], + } +} + +fn handle_join_wrap(params: Map) -> ControllerFuture { + Box::pin(async move { super::ops::handle_join(params).await }) +} + +fn handle_leave_wrap(params: Map) -> ControllerFuture { + Box::pin(async move { super::ops::handle_leave(params).await }) +} + +fn handle_harness_response_wrap(params: Map) -> ControllerFuture { + Box::pin(async move { super::ops::handle_harness_response(params).await }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registered_controllers_match_schemas() { + let schema_fns: Vec<_> = all_controller_schemas() + .into_iter() + .map(|s| s.function) + .collect(); + let handler_fns: Vec<_> = all_registered_controllers() + .into_iter() + .map(|c| c.schema.function) + .collect(); + assert_eq!(schema_fns, handler_fns); + assert_eq!(schema_fns, vec!["join", "leave", "harness_response"]); + } + + #[test] + fn join_schema_has_correct_namespace() { + let s = schema_join(); + assert_eq!(s.namespace, "agent_meetings"); + assert_eq!(s.function, "join"); + } +} diff --git a/src/openhuman/agent_meetings/types.rs b/src/openhuman/agent_meetings/types.rs new file mode 100644 index 000000000..f17937aa6 --- /dev/null +++ b/src/openhuman/agent_meetings/types.rs @@ -0,0 +1,34 @@ +//! Request / response types for the `agent_meetings` domain. + +use serde::{Deserialize, Serialize}; + +/// Inputs to `openhuman.agent_meetings_join`. +#[derive(Debug, Clone, Deserialize)] +pub struct BackendMeetJoinRequest { + pub meet_url: String, + #[serde(default)] + pub display_name: Option, + #[serde(default)] + pub platform: Option, +} + +/// Outputs from `openhuman.agent_meetings_join`. +#[derive(Debug, Clone, Serialize)] +pub struct BackendMeetJoinResponse { + pub ok: bool, + pub meet_url: String, + pub platform: String, +} + +/// Inputs to `openhuman.agent_meetings_leave`. +#[derive(Debug, Clone, Deserialize)] +pub struct BackendMeetLeaveRequest { + #[serde(default)] + pub reason: Option, +} + +/// Inputs to `openhuman.agent_meetings_harness_response`. +#[derive(Debug, Clone, Deserialize)] +pub struct BackendMeetHarnessResponseRequest { + pub result: String, +} diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index d27a0b1a3..52a6c53d8 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -18,6 +18,7 @@ pub mod about_app; pub mod accessibility; pub mod agent; pub mod agent_experience; +pub mod agent_meetings; pub mod agent_orchestration; pub mod agent_registry; pub mod agent_tool_policy; diff --git a/src/openhuman/socket/event_handlers.rs b/src/openhuman/socket/event_handlers.rs index 74645ff12..2415cfe87 100644 --- a/src/openhuman/socket/event_handlers.rs +++ b/src/openhuman/socket/event_handlers.rs @@ -10,7 +10,7 @@ use serde_json::json; use tokio::sync::mpsc; use crate::api::models::socket::ConnectionStatus; -use crate::core::event_bus::{publish_global, DomainEvent}; +use crate::core::event_bus::{publish_global, BackendMeetTurn, DomainEvent}; use crate::openhuman::webhooks::WebhookRequest; use super::manager::{emit_server_event, emit_state_change, SharedState}; @@ -248,6 +248,102 @@ pub(super) fn handle_sio_event( } } + // ── Backend Meet Bot events ────────────────────────────────────── + "bot:joined" => { + let meet_url = data + .get("meetUrl") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + log::info!("[socket] bot:joined meet_url_len={}", meet_url.len()); + publish_global(DomainEvent::BackendMeetJoined { meet_url }); + } + "bot:left" => { + let reason = data + .get("reason") + .and_then(|v| v.as_str()) + .unwrap_or("unknown") + .to_string(); + log::info!("[socket] bot:left reason={}", reason); + publish_global(DomainEvent::BackendMeetLeft { reason }); + } + "bot:reply" => { + let transcript = data + .get("transcript") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let reply = data + .get("reply") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let emotion = data + .get("emotion") + .and_then(|v| v.as_str()) + .unwrap_or("neutral") + .to_string(); + log::info!( + "[socket] bot:reply reply_len={} emotion={}", + reply.len(), + emotion + ); + publish_global(DomainEvent::BackendMeetReply { + transcript, + reply, + emotion, + }); + } + "bot:harness" => { + let transcript = data + .get("transcript") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let instruction = data + .get("instruction") + .and_then(|v| v.as_str()) + .unwrap_or("") + .to_string(); + let emotion = data + .get("emotion") + .and_then(|v| v.as_str()) + .unwrap_or("neutral") + .to_string(); + log::info!( + "[socket] bot:harness instruction_len={} emotion={}", + instruction.len(), + emotion + ); + publish_global(DomainEvent::BackendMeetHarness { + transcript, + instruction, + emotion, + }); + } + "bot:transcript" => { + let turns: Vec = data + .get("turns") + .and_then(|v| serde_json::from_value(v.clone()).ok()) + .unwrap_or_default(); + let duration_ms = data.get("durationMs").and_then(|v| v.as_u64()).unwrap_or(0); + log::info!( + "[socket] bot:transcript turns={} duration_ms={}", + turns.len(), + duration_ms + ); + publish_global(DomainEvent::BackendMeetTranscript { turns, duration_ms }); + } + "bot:error" => { + let error = data + .get("error") + .and_then(|v| v.as_str()) + .unwrap_or("unknown error") + .to_string(); + log::error!("[socket] bot:error: {}", error); + publish_global(DomainEvent::BackendMeetError { error }); + } + // Channel inbound message — publish to event bus for ChannelInboundSubscriber _ if event_name.ends_with(":message") => { log::info!(