feat(kap-server): stream raw xstate inspection events over /api/v1/debug/ws #3687
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| import type { InspectionEvent } from 'xstate'; | ||
|
|
||
| export type XstateInspectionEventType = InspectionEvent['type']; | ||
|
|
||
| export interface XstateInspectionEnvelope { | ||
| readonly type: XstateInspectionEventType; | ||
| readonly timestamp: number; | ||
| readonly actorSessionId: string; | ||
| readonly actorId?: string; | ||
| readonly logicId?: string; | ||
| readonly eventType?: string; | ||
| readonly stateValue?: unknown; | ||
| } | ||
|
|
||
| export type XstateInspectionListener = (envelope: XstateInspectionEnvelope) => void; | ||
|
|
||
| export interface XstateInspectionCollector { | ||
| subscribe(listener: XstateInspectionListener): () => void; | ||
| publish(event: InspectionEvent): void; | ||
| } | ||
|
|
||
| function scalar(value: unknown): string | undefined { | ||
| return typeof value === 'string' ? value : undefined; | ||
| } | ||
|
|
||
| function toEnvelope(event: InspectionEvent, now: () => number): XstateInspectionEnvelope { | ||
| const actorRef = event.actorRef as { id?: unknown; logic?: unknown }; | ||
| const logic = actorRef.logic as { id?: unknown } | undefined; | ||
| const snapshot = 'snapshot' in event ? (event.snapshot as { value?: unknown }) : undefined; | ||
| return { | ||
| type: event.type, | ||
| timestamp: now(), | ||
| actorSessionId: event.actorRef.sessionId, | ||
| actorId: scalar(actorRef.id), | ||
| logicId: scalar(logic?.id), | ||
| eventType: | ||
| 'event' in event | ||
| ? event.event.type | ||
| : event.type === '@xstate.action' | ||
| ? event.action.type | ||
| : undefined, | ||
| stateValue: snapshot?.value, | ||
| }; | ||
| } | ||
|
|
||
| export function createXstateInspectionCollector(input?: { | ||
| now?: () => number; | ||
| }): XstateInspectionCollector { | ||
| const now = input?.now ?? Date.now; | ||
| const listeners = new Set<XstateInspectionListener>(); | ||
| return { | ||
| subscribe(listener) { | ||
| listeners.add(listener); | ||
| return () => { | ||
| listeners.delete(listener); | ||
| }; | ||
| }, | ||
| publish(event) { | ||
| if (listeners.size === 0) return; | ||
| const envelope = toEnvelope(event, now); | ||
| for (const listener of listeners) listener(envelope); | ||
| }, | ||
| }; | ||
| } | ||
|
|
||
| export const xstateInspectionCollector = createXstateInspectionCollector(); | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| import type { XstateInspectionCollector } from '@moonshot-ai/agent-core-v2/human/xstateInspection'; | ||
| import { WebSocketServer } from 'ws'; | ||
|
|
||
| import { selectWsBearerProtocol } from '../bearerProtocol'; | ||
| import { WsConnectionDebug } from './wsConnectionDebug'; | ||
|
|
||
| export const WS_DEBUG_PATH = '/api/v1/debug/ws'; | ||
|
|
||
| export interface RegisterWsDebugOptions { | ||
| readonly collector?: XstateInspectionCollector; | ||
| readonly heartbeatIntervalMs?: number; | ||
| readonly flushIntervalMs?: number; | ||
| readonly highWaterMarkBytes?: number; | ||
| } | ||
|
|
||
| export function registerWsDebug(opts: RegisterWsDebugOptions = {}): WebSocketServer { | ||
| const wss = new WebSocketServer({ noServer: true, handleProtocols: selectWsBearerProtocol }); | ||
| const connections = new Set<WsConnectionDebug>(); | ||
|
|
||
| wss.on('connection', (socket) => { | ||
| const conn = new WsConnectionDebug({ | ||
| socket, | ||
| collector: opts.collector, | ||
| heartbeatIntervalMs: opts.heartbeatIntervalMs, | ||
| flushIntervalMs: opts.flushIntervalMs, | ||
| highWaterMarkBytes: opts.highWaterMarkBytes, | ||
| }); | ||
| connections.add(conn); | ||
| socket.on('close', () => { | ||
| connections.delete(conn); | ||
| }); | ||
| }); | ||
|
|
||
| wss.on('close', () => { | ||
| for (const conn of connections) conn.close(); | ||
| }); | ||
|
|
||
| return wss; | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| import { | ||
| xstateInspectionCollector, | ||
| type XstateInspectionCollector, | ||
| type XstateInspectionEnvelope, | ||
| } from '@moonshot-ai/agent-core-v2/human/xstateInspection'; | ||
| import type { WebSocket } from 'ws'; | ||
|
|
||
| const DEFAULT_HEARTBEAT_INTERVAL_MS = 10_000; | ||
| const HEARTBEAT_MISS_LIMIT = 2; | ||
| const DEFAULT_FLUSH_INTERVAL_MS = 16; | ||
| const DEFAULT_HIGH_WATER_MARK_BYTES = 1 << 20; | ||
|
|
||
| export interface WsConnectionDebugOptions { | ||
| readonly socket: WebSocket; | ||
| readonly collector?: XstateInspectionCollector; | ||
| readonly heartbeatIntervalMs?: number; | ||
| readonly flushIntervalMs?: number; | ||
| readonly highWaterMarkBytes?: number; | ||
| } | ||
|
|
||
| export class WsConnectionDebug { | ||
| private readonly socket: WebSocket; | ||
| private readonly heartbeatIntervalMs: number; | ||
| private readonly flushIntervalMs: number; | ||
| private readonly highWaterMarkBytes: number; | ||
| private readonly unsubscribe: () => void; | ||
|
|
||
| private closed = false; | ||
| private outbound: XstateInspectionEnvelope[] = []; | ||
| private flushTimer?: ReturnType<typeof setTimeout>; | ||
| private heartbeatTimer?: ReturnType<typeof setInterval>; | ||
| private lastPongAt = Date.now(); | ||
|
|
||
| constructor(opts: WsConnectionDebugOptions) { | ||
| this.socket = opts.socket; | ||
| this.heartbeatIntervalMs = opts.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS; | ||
| this.flushIntervalMs = opts.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS; | ||
| this.highWaterMarkBytes = opts.highWaterMarkBytes ?? DEFAULT_HIGH_WATER_MARK_BYTES; | ||
|
|
||
| this.socket.on('close', () => this.onClose()); | ||
| this.socket.on('error', () => this.onClose()); | ||
| this.socket.on('pong', () => { | ||
| this.lastPongAt = Date.now(); | ||
| }); | ||
|
|
||
| const collector = opts.collector ?? xstateInspectionCollector; | ||
| this.unsubscribe = collector.subscribe((envelope) => this.onEnvelope(envelope)); | ||
|
|
||
| this.heartbeatTimer = setInterval(() => this.onHeartbeat(), this.heartbeatIntervalMs); | ||
| this.heartbeatTimer.unref?.(); | ||
| } | ||
|
|
||
| private onEnvelope(envelope: XstateInspectionEnvelope): void { | ||
| if (this.closed) return; | ||
| if (this.socket.bufferedAmount > this.highWaterMarkBytes) return; | ||
| this.outbound.push(envelope); | ||
| this.scheduleFlush(); | ||
|
Comment on lines
+53
to
+57
There was a problem hiding this comment.
With a slow or non-reading debug client and a synchronous burst of inspection events, Useful? React with 👍 / 👎. |
||
| } | ||
|
|
||
| private scheduleFlush(): void { | ||
| if (this.flushTimer !== undefined) return; | ||
| this.flushTimer = setTimeout(() => { | ||
| this.flushTimer = undefined; | ||
| this.flush(); | ||
| }, this.flushIntervalMs); | ||
| this.flushTimer.unref?.(); | ||
| } | ||
|
|
||
| private flush(): void { | ||
| if (this.outbound.length === 0) return; | ||
| if (this.closed || this.socket.readyState !== this.socket.OPEN) { | ||
| this.outbound = []; | ||
| return; | ||
| } | ||
| const envelopes = this.outbound; | ||
| this.outbound = []; | ||
| for (const envelope of envelopes) { | ||
| if (this.closed || this.socket.readyState !== this.socket.OPEN) return; | ||
| try { | ||
| this.socket.send(JSON.stringify(envelope)); | ||
| } catch { | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private onHeartbeat(): void { | ||
| if (Date.now() - this.lastPongAt >= this.heartbeatIntervalMs * HEARTBEAT_MISS_LIMIT) { | ||
| this.close(); | ||
| return; | ||
| } | ||
| try { | ||
| this.socket.ping(); | ||
| } catch { | ||
| } | ||
| } | ||
|
|
||
| close(): void { | ||
| if (this.closed) return; | ||
| try { | ||
| this.socket.close(1000); | ||
| } catch { | ||
| } | ||
| this.onClose(); | ||
| } | ||
|
|
||
| private onClose(): void { | ||
| if (this.closed) return; | ||
| this.closed = true; | ||
| if (this.flushTimer !== undefined) clearTimeout(this.flushTimer); | ||
| if (this.heartbeatTimer !== undefined) clearInterval(this.heartbeatTimer); | ||
| this.outbound = []; | ||
| this.unsubscribe(); | ||
| } | ||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When multiple Kimi sessions or agents are active, these fields cannot associate an inspected actor with its owning runtime:
createMachineEnginecreates each root actor without an application session/agent ID, while nested actors reuse IDs such asturnandtool. Because the projection also discards XState's source/parent relationship, consumers cannot reconstruct which child belongs to which root, so concurrent state-machine views will mix unrelated actors. Include an owning session/agent identifier and parent/source actor session ID in the envelope.Useful? React with 👍 / 👎.