wushenghua/hapi
1
0
Fork 0
mirror of https://github.com/wu736139669/hapi.git synced 2026-08-05 06:24:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Structure SSE update patches

Browse Source
This commit is contained in:
weishu
2026-05-21 10:44:46 +08:00
parent 2e1e2d39db
commit 41f6b37d69
10 changed files with 162 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hub/src/socket/handlers/cli/machineHandlers.ts
+2 -2
View File
@@ -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 })
}
}
hub/src/socket/handlers/cli/sessionHandlers.ts
+4 -4
View File
@@ -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 })
}
}
hub/src/sync/machineCache.ts
+8 -24
View File
@@ -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<string, Machine> = new Map()
private readonly lastBroadcastAtByMachineId: Map<string, number> = 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 }
hub/src/sync/sessionCache.ts
+15 -7
View File
@@ -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
shared/src/resume.test.ts
+26
View File
@@ -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)
})
})
shared/src/schemas.ts
+58 -2
View File
@@ -210,6 +210,62 @@ export const SessionSchema = z.object({
export type Session = z.infer<typeof SessionSchema>
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<typeof SessionPatchSchema>
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<typeof MachineSchema>
export const MachinePatchSchema = z.object({
active: z.boolean().optional(),
activeAt: z.number().optional(),
updatedAt: z.number().optional()
}).strict()
export type MachinePatch = z.infer<typeof MachinePatchSchema>
export const SessionUpdatedDataSchema = z.union([SessionSchema, SessionPatchSchema])
export type SessionUpdatedData = z.infer<typeof SessionUpdatedDataSchema>
export const MachineUpdatedDataSchema = z.union([MachineSchema, MachinePatchSchema, z.null()])
export type MachineUpdatedData = z.infer<typeof MachineUpdatedDataSchema>
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'),
shared/src/types.ts
+5
View File
@@ -5,7 +5,12 @@ export type {
AttachmentMetadata,
DecryptedMessage,
Metadata,
Machine,
MachinePatch,
MachineUpdatedData,
Session,
SessionPatch,
SessionUpdatedData,
SyncEvent,
TeamMember,
TeamMessage,
web/src/hooks/useSSE.ts
+23 -91
View File
@@ -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<Pick<Session, 'active' | 'thinking' | 'activeAt' | 'updatedAt' | 'model' | 'modelReasoningEffort' | 'effort' | 'permissionMode' | 'collaborationMode'>>
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<string, unknown> {
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()
}
}
web/src/types/api.ts
+3 -13
View File
@@ -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: {
web/src/utils/formatRunnerSpawnError.ts
+18 -2
View File
@@ -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
}