diff --git a/hub/src/socket/handlers/cli/machineHandlers.ts b/hub/src/socket/handlers/cli/machineHandlers.ts index d187532d..43c0555f 100644 --- a/hub/src/socket/handlers/cli/machineHandlers.ts +++ b/hub/src/socket/handlers/cli/machineHandlers.ts @@ -89,7 +89,7 @@ export function registerMachineHandlers(socket: CliSocketWithData, deps: Machine } } socket.to(`machine:${id}`).emit('update', update) - onWebappEvent?.({ type: 'machine-updated', machineId: id, data: { id } }) + onWebappEvent?.({ type: 'machine-updated', machineId: id }) } } @@ -134,7 +134,7 @@ export function registerMachineHandlers(socket: CliSocketWithData, deps: Machine } } socket.to(`machine:${id}`).emit('update', update) - onWebappEvent?.({ type: 'machine-updated', machineId: id, data: { id } }) + onWebappEvent?.({ type: 'machine-updated', machineId: id }) } } diff --git a/hub/src/socket/handlers/cli/sessionHandlers.ts b/hub/src/socket/handlers/cli/sessionHandlers.ts index 157fd98e..a36dd82a 100644 --- a/hub/src/socket/handlers/cli/sessionHandlers.ts +++ b/hub/src/socket/handlers/cli/sessionHandlers.ts @@ -111,7 +111,7 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session if (todos) { const updated = store.sessions.setSessionTodos(sid, todos, msg.createdAt, session.namespace) if (updated) { - onWebappEvent?.({ type: 'session-updated', sessionId: sid, data: { sid } }) + onWebappEvent?.({ type: 'session-updated', sessionId: sid }) } } @@ -122,7 +122,7 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session const newTeamState = applyTeamStateDelta(existingTeamState ?? null, teamDelta) const updated = store.sessions.setSessionTeamState(sid, newTeamState, msg.createdAt, session.namespace) if (updated) { - onWebappEvent?.({ type: 'session-updated', sessionId: sid, data: { sid } }) + onWebappEvent?.({ type: 'session-updated', sessionId: sid }) } } @@ -204,7 +204,7 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session } } socket.to(`session:${sid}`).emit('update', update) - onWebappEvent?.({ type: 'session-updated', sessionId: sid, data: { sid } }) + onWebappEvent?.({ type: 'session-updated', sessionId: sid }) } } @@ -251,7 +251,7 @@ export function registerSessionHandlers(socket: CliSocketWithData, deps: Session } } socket.to(`session:${sid}`).emit('update', update) - onWebappEvent?.({ type: 'session-updated', sessionId: sid, data: { sid } }) + onWebappEvent?.({ type: 'session-updated', sessionId: sid }) } } diff --git a/hub/src/sync/machineCache.ts b/hub/src/sync/machineCache.ts index e80c3eb7..da5720b9 100644 --- a/hub/src/sync/machineCache.ts +++ b/hub/src/sync/machineCache.ts @@ -1,4 +1,5 @@ import { z } from 'zod' +import type { Machine, MachinePatch } from '@hapi/protocol/types' import type { Store } from '../store' import { clampAliveTime } from './aliveTime' import { EventPublisher } from './eventPublisher' @@ -15,29 +16,6 @@ const machineMetadataSchema = z.object({ workspaceRoots: z.array(z.string()).optional() }) -export interface Machine { - id: string - namespace: string - seq: number - createdAt: number - updatedAt: number - active: boolean - activeAt: number - metadata: { - host: string - platform: string - happyCliVersion: string - displayName?: string - homeDir?: string - happyHomeDir?: string - happyLibDir?: string - workspaceRoots?: string[] - } | null - metadataVersion: number - runnerState: unknown | null - runnerStateVersion: number -} - export class MachineCache { private readonly machines: Map = new Map() private readonly lastBroadcastAtByMachineId: Map = new Map() @@ -180,7 +158,13 @@ export class MachineCache { if (!machine.active) continue if (now - machine.activeAt <= machineTimeoutMs) continue machine.active = false - this.publisher.emit({ type: 'machine-updated', machineId: machine.id, data: { active: false } }) + this.publisher.emit({ + type: 'machine-updated', + machineId: machine.id, + data: { active: false } satisfies MachinePatch + }) } } } + +export type { Machine } diff --git a/hub/src/sync/sessionCache.ts b/hub/src/sync/sessionCache.ts index 982c78a0..2a1600c6 100644 --- a/hub/src/sync/sessionCache.ts +++ b/hub/src/sync/sessionCache.ts @@ -1,5 +1,5 @@ import { AgentStateSchema, MetadataSchema, TeamStateSchema } from '@hapi/protocol/schemas' -import type { CodexCollaborationMode, PermissionMode, Session } from '@hapi/protocol/types' +import type { CodexCollaborationMode, PermissionMode, Session, SessionPatch } from '@hapi/protocol/types' import type { Store } from '../store' import { clampAliveTime } from './aliveTime' import { EventPublisher } from './eventPublisher' @@ -254,7 +254,7 @@ export class SessionCache { modelReasoningEffort: session.modelReasoningEffort, effort: session.effort, collaborationMode: session.collaborationMode - } + } satisfies SessionPatch }) } } @@ -281,7 +281,7 @@ export class SessionCache { data: { thinking: true, updatedAt: session.updatedAt - } + } satisfies SessionPatch }) } } @@ -298,7 +298,7 @@ export class SessionCache { this.publisher.emit({ type: 'session-updated', sessionId, - data: { backgroundTaskCount: next } + data: { backgroundTaskCount: next } satisfies SessionPatch }) } @@ -332,7 +332,7 @@ export class SessionCache { type: 'session-updated', sessionId, namespace: session.namespace, - data: { updatedAt: session.updatedAt } + data: { updatedAt: session.updatedAt } satisfies SessionPatch }) } @@ -352,7 +352,11 @@ export class SessionCache { session.backgroundTaskCount = 0 this.pendingThinkingUntilBySessionId.delete(session.id) - this.publisher.emit({ type: 'session-updated', sessionId: session.id, data: { active: false, thinking: false, backgroundTaskCount: 0 } }) + this.publisher.emit({ + type: 'session-updated', + sessionId: session.id, + data: { active: false, thinking: false, backgroundTaskCount: 0 } satisfies SessionPatch + }) } expireInactive(now: number = Date.now()): string[] { @@ -366,7 +370,11 @@ export class SessionCache { session.thinking = false this.pendingThinkingUntilBySessionId.delete(session.id) expired.push(session.id) - this.publisher.emit({ type: 'session-updated', sessionId: session.id, data: { active: false } }) + this.publisher.emit({ + type: 'session-updated', + sessionId: session.id, + data: { active: false } satisfies SessionPatch + }) } return expired diff --git a/shared/src/resume.test.ts b/shared/src/resume.test.ts index 471be6d6..8835e9fb 100644 --- a/shared/src/resume.test.ts +++ b/shared/src/resume.test.ts @@ -71,4 +71,30 @@ describe('resume schemas', () => { invokedAt: 123 }).success).toBe(true) }) + + it('validates structured session and machine update patches', () => { + expect(SyncEventSchema.safeParse({ + type: 'session-updated', + sessionId: 'hapi-session-1', + data: { updatedAt: 123, backgroundTaskCount: 1 } + }).success).toBe(true) + + expect(SyncEventSchema.safeParse({ + type: 'session-updated', + sessionId: 'hapi-session-1', + data: { sid: 'hapi-session-1' } + }).success).toBe(false) + + expect(SyncEventSchema.safeParse({ + type: 'machine-updated', + machineId: 'machine-1', + data: { active: false } + }).success).toBe(true) + + expect(SyncEventSchema.safeParse({ + type: 'machine-updated', + machineId: 'machine-1', + data: { id: 'machine-1' } + }).success).toBe(false) + }) }) diff --git a/shared/src/schemas.ts b/shared/src/schemas.ts index cd7206cd..ae071554 100644 --- a/shared/src/schemas.ts +++ b/shared/src/schemas.ts @@ -210,6 +210,62 @@ export const SessionSchema = z.object({ export type Session = z.infer +export const SessionPatchSchema = z.object({ + active: z.boolean().optional(), + thinking: z.boolean().optional(), + activeAt: z.number().optional(), + updatedAt: z.number().optional(), + model: z.string().nullable().optional(), + modelReasoningEffort: z.string().nullable().optional(), + effort: z.string().nullable().optional(), + permissionMode: PermissionModeSchema.optional(), + collaborationMode: CodexCollaborationModeSchema.optional(), + backgroundTaskCount: z.number().optional() +}).strict() + +export type SessionPatch = z.infer + +export const MachineMetadataSchema = z.object({ + host: z.string(), + platform: z.string(), + happyCliVersion: z.string(), + displayName: z.string().optional(), + homeDir: z.string().optional(), + happyHomeDir: z.string().optional(), + happyLibDir: z.string().optional(), + workspaceRoots: z.array(z.string()).optional() +}) + +export const MachineSchema = z.object({ + id: z.string(), + namespace: z.string(), + seq: z.number(), + createdAt: z.number(), + updatedAt: z.number(), + active: z.boolean(), + activeAt: z.number(), + metadata: MachineMetadataSchema.nullable(), + metadataVersion: z.number(), + runnerState: z.unknown().nullable(), + runnerStateVersion: z.number() +}) + +export type Machine = z.infer + +export const MachinePatchSchema = z.object({ + active: z.boolean().optional(), + activeAt: z.number().optional(), + updatedAt: z.number().optional() +}).strict() + +export type MachinePatch = z.infer + +export const SessionUpdatedDataSchema = z.union([SessionSchema, SessionPatchSchema]) +export type SessionUpdatedData = z.infer + +export const MachineUpdatedDataSchema = z.union([MachineSchema, MachinePatchSchema, z.null()]) +export type MachineUpdatedData = z.infer + const SessionEventBaseSchema = z.object({ namespace: z.string().optional() }) @@ -229,7 +285,7 @@ export const SyncEventSchema = z.discriminatedUnion('type', [ }), SessionChangedSchema.extend({ type: z.literal('session-updated'), - data: z.unknown().optional() + data: SessionUpdatedDataSchema.optional() }), SessionEventBaseSchema.extend({ type: z.literal('session-removed'), @@ -248,7 +304,7 @@ export const SyncEventSchema = z.discriminatedUnion('type', [ }), MachineChangedSchema.extend({ type: z.literal('machine-updated'), - data: z.unknown().optional() + data: MachineUpdatedDataSchema.optional() }), SessionEventBaseSchema.extend({ type: z.literal('toast'), diff --git a/shared/src/types.ts b/shared/src/types.ts index cad2b4f0..3d04adbe 100644 --- a/shared/src/types.ts +++ b/shared/src/types.ts @@ -5,7 +5,12 @@ export type { AttachmentMetadata, DecryptedMessage, Metadata, + Machine, + MachinePatch, + MachineUpdatedData, Session, + SessionPatch, + SessionUpdatedData, SyncEvent, TeamMember, TeamMessage, diff --git a/web/src/hooks/useSSE.ts b/web/src/hooks/useSSE.ts index 63cf5671..ddd3bc91 100644 --- a/web/src/hooks/useSSE.ts +++ b/web/src/hooks/useSSE.ts @@ -1,10 +1,12 @@ import { useEffect, useMemo, useRef, useState } from 'react' import { useQueryClient } from '@tanstack/react-query' import { isObject, toSessionSummary } from '@hapi/protocol' +import { MachinePatchSchema, MachineSchema, SessionPatchSchema, SessionSchema } from '@hapi/protocol/schemas' import type { Machine, MachinesResponse, Session, + SessionPatch, SessionResponse, SessionsResponse, SessionSummary, @@ -30,8 +32,6 @@ const RECONNECT_MAX_DELAY_MS = 30_000 const RECONNECT_JITTER_MS = 500 const INVALIDATION_BATCH_MS = 16 -type SessionPatch = Partial> - function sortSessionSummaries(left: SessionSummary, right: SessionSummary): number { if (left.active !== right.active) { return left.active ? -1 : 1 @@ -42,100 +42,28 @@ function sortSessionSummaries(left: SessionSummary, right: SessionSummary): numb return right.updatedAt - left.updatedAt } -function hasRecordShape(value: unknown): value is Record { - return isObject(value) -} - function isSessionRecord(value: unknown): value is Session { - if (!hasRecordShape(value)) { - return false - } - return typeof value.id === 'string' - && typeof value.active === 'boolean' - && typeof value.activeAt === 'number' - && typeof value.updatedAt === 'number' - && typeof value.thinking === 'boolean' + return SessionSchema.safeParse(value).success } function getSessionPatch(value: unknown): SessionPatch | null { - if (!hasRecordShape(value)) { + const parsed = SessionPatchSchema.safeParse(value) + if (!parsed.success) { return null } - - const patch: SessionPatch = {} - let hasKnownPatch = false - - if (typeof value.active === 'boolean') { - patch.active = value.active - hasKnownPatch = true - } - if (typeof value.thinking === 'boolean') { - patch.thinking = value.thinking - hasKnownPatch = true - } - if (typeof value.activeAt === 'number') { - patch.activeAt = value.activeAt - hasKnownPatch = true - } - if (typeof value.updatedAt === 'number') { - patch.updatedAt = value.updatedAt - hasKnownPatch = true - } - if (value.model === null || typeof value.model === 'string') { - patch.model = value.model - hasKnownPatch = true - } - if (value.modelReasoningEffort === null || typeof value.modelReasoningEffort === 'string') { - patch.modelReasoningEffort = value.modelReasoningEffort - hasKnownPatch = true - } - if (value.effort === null || typeof value.effort === 'string') { - patch.effort = value.effort - hasKnownPatch = true - } - if (typeof value.permissionMode === 'string') { - patch.permissionMode = value.permissionMode as Session['permissionMode'] - hasKnownPatch = true - } - if (typeof value.collaborationMode === 'string') { - patch.collaborationMode = value.collaborationMode as Session['collaborationMode'] - hasKnownPatch = true - } - - return hasKnownPatch ? patch : null -} - -function hasUnknownSessionPatchKeys(value: unknown): boolean { - if (!hasRecordShape(value)) { - return false - } - const knownKeys = new Set(['active', 'thinking', 'activeAt', 'updatedAt', 'model', 'modelReasoningEffort', 'effort', 'permissionMode', 'collaborationMode']) - return Object.keys(value).some((key) => !knownKeys.has(key)) -} - -function isMachineMetadata(value: unknown): value is Machine['metadata'] { - if (value === null) { - return true - } - if (!hasRecordShape(value)) { - return false - } - return typeof value.host === 'string' - && typeof value.platform === 'string' - && typeof value.happyCliVersion === 'string' + return Object.keys(parsed.data).length > 0 ? parsed.data : null } function isMachineRecord(value: unknown): value is Machine { - if (!hasRecordShape(value)) { - return false - } - return typeof value.id === 'string' - && typeof value.active === 'boolean' - && isMachineMetadata(value.metadata) + return MachineSchema.safeParse(value).success } -function isInactiveMachinePatch(value: unknown): boolean { - return hasRecordShape(value) && value.active === false +function getMachinePatch(value: unknown): { active?: boolean; activeAt?: number; updatedAt?: number } | null { + const parsed = MachinePatchSchema.safeParse(value) + if (!parsed.success) { + return null + } + return Object.keys(parsed.data).length > 0 ? parsed.data : null } function getVisibilityState(): VisibilityState { @@ -531,10 +459,6 @@ export function useSSE(options: { if (!summaryPatched) { queueSessionListInvalidation() } - if (hasUnknownSessionPatchKeys(event.data)) { - queueSessionDetailInvalidation(event.sessionId) - queueSessionListInvalidation() - } } else { queueSessionDetailInvalidation(event.sessionId) queueSessionListInvalidation() @@ -545,9 +469,17 @@ export function useSSE(options: { if (event.type === 'machine-updated') { if (isMachineRecord(event.data)) { upsertMachine(event.data) - } else if (event.data === null || isInactiveMachinePatch(event.data)) { + } else if (event.data === null) { removeMachine(event.machineId) - } else if (!hasRecordShape(event.data) || typeof event.data.activeAt !== 'number') { + } else { + const patch = getMachinePatch(event.data) + if (patch?.active === false) { + removeMachine(event.machineId) + } else { + queueMachinesInvalidation() + } + } + if (event.data === undefined) { queueMachinesInvalidation() } } diff --git a/web/src/types/api.ts b/web/src/types/api.ts index 14d36172..a7aaf10e 100644 --- a/web/src/types/api.ts +++ b/web/src/types/api.ts @@ -1,5 +1,6 @@ import type { DecryptedMessage as ProtocolDecryptedMessage, + Machine, Session, SessionSummary, SyncEvent as ProtocolSyncEvent, @@ -31,7 +32,9 @@ export type { AttachmentMetadata, CodexCollaborationMode, PermissionMode, + Machine, Session, + SessionPatch, SessionSummary, SessionSummaryMetadata, TeamMember, @@ -84,19 +87,6 @@ export type RunnerState = { } | null } -export type Machine = { - id: string - active: boolean - metadata: { - host: string - platform: string - happyCliVersion: string - displayName?: string - workspaceRoots?: string[] - } | null - runnerState?: RunnerState | null -} - export type AuthResponse = { token: string user: { diff --git a/web/src/utils/formatRunnerSpawnError.ts b/web/src/utils/formatRunnerSpawnError.ts index a0e0c40f..063e716a 100644 --- a/web/src/utils/formatRunnerSpawnError.ts +++ b/web/src/utils/formatRunnerSpawnError.ts @@ -1,8 +1,24 @@ import type { Machine } from '../types/api' +function getLastSpawnError(runnerState: unknown): { message: string; at?: number } | null { + if (!runnerState || typeof runnerState !== 'object' || Array.isArray(runnerState)) { + return null + } + const lastSpawnError = (runnerState as { lastSpawnError?: unknown }).lastSpawnError + if (!lastSpawnError || typeof lastSpawnError !== 'object' || Array.isArray(lastSpawnError)) { + return null + } + const message = (lastSpawnError as { message?: unknown }).message + if (typeof message !== 'string' || message.length === 0) { + return null + } + const at = (lastSpawnError as { at?: unknown }).at + return typeof at === 'number' ? { message, at } : { message } +} + export function formatRunnerSpawnError(machine: Machine | null): string | null { - const lastSpawnError = machine?.runnerState?.lastSpawnError - if (!lastSpawnError?.message) { + const lastSpawnError = getLastSpawnError(machine?.runnerState) + if (!lastSpawnError) { return null }