diff --git a/.changeset/unified-transcript-layer.md b/.changeset/unified-transcript-layer.md new file mode 100644 index 0000000000..0b407289de --- /dev/null +++ b/.changeset/unified-transcript-layer.md @@ -0,0 +1,7 @@ +--- +"@moonshot-ai/transcript": patch +"@moonshot-ai/kap-server": patch +"@moonshot-ai/agent-core-v2": patch +--- + +Add a unified, agent-granular transcript rendering data layer and serve it from the v2 server: clients can fetch turn-paginated transcripts via `GET /sessions/{id}/transcript` and subscribe to per-agent transcript updates over the v1 WebSocket with per-connection granularity control (off / turn / block / delta). All transcript wire types are owned by the transcript package itself. `turn.started` now carries the turn's prompt text so live transcripts render the user input as soon as the turn opens. diff --git a/AGENTS.md b/AGENTS.md index 6715558877..fd32638c1a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,15 +17,17 @@ This is a TypeScript monorepo built for agent-assisted development. Keep the roo - `apps/kimi-code`: the CLI / TUI application. It consumes core capabilities through `@moonshot-ai/kimi-code-sdk` and must not depend directly on `@moonshot-ai/agent-core`. When writing or modifying its terminal UI, use the `write-tui` skill (`.agents/skills/write-tui/SKILL.md`). - `apps/kimi-web`: the browser web UI, a peer to the TUI. Vue 3 + Vite + vue-i18n; talks to the server over REST + WebSocket under `/api/v1`. It must not depend on `@moonshot-ai/agent-core` (wire types are re-implemented locally). Debug against the two engines via the root `pnpm dev:v1` / `pnpm dev:v2` backend scripts — the dev Sidebar shows the active backend and switches it at runtime. See `apps/kimi-web/AGENTS.md`. - `apps/vis`, `apps/vis/server`, `apps/vis/web`: visual debugging tools for sessions and replays. -- `apps/kimi-inspect`: web inspector for the v2 (kap-server) `/api/v2` surface — workspace/session browser, per-session chat, and live Service panels (data + trigger buttons) for the Session and Agent scopes. Built on its own old-klient-style channel layer (`src/channel/`: the VS Code `ProxyChannel` model — service-bound `IChannel`, HTTP `ProxyChannel` for calls routed to `/api/v2`, `WsChannel` over the shared `/api/v2/ws` socket for events), typed by `agent-core-v2` Service interfaces; `GET {rpcBasePath}/channels` loads every wire protocol 1:1 — probing `/api/v1/debug` first (dev, whitelist-free) and falling back to `/api/v2`. The Vite dev server proxies `/api` to a running kap-server (`KIMI_SERVER_URL`, default `http://127.0.0.1:58627`) and exposes `GET /__inspect/servers` (`vite/serverDiscovery.ts`), which scans the local kap-server instance registry (`~/.kimi-code/server/instances` + legacy `lock`) and the home token so the app can zero-config auto-connect and switch servers from the header dropdown at runtime. +- `apps/kimi-inspect`: web inspector for the kap-server `/api/v1/debug` RPC surface — workspace/session browser, per-session chat, and Service panels (data + trigger buttons) for the Session and Agent scopes. Built on its own old-klient-style channel layer (`src/channel/`: the VS Code `ProxyChannel` model — service-bound `IChannel`, HTTP `ProxyChannel` for calls routed to `/api/v1/debug`), typed by `agent-core-v2` Service interfaces; `GET /api/v1/debug/channels` loads the whole wire protocol 1:1 (every scoped Service, no whitelist). There is no Service-event push channel: panels fetch/refresh on demand (`Sidebar` polls react-query on a 15 s interval), and a connection failure shows a blocking "Debug surface unavailable" screen instead of falling back anywhere. The Vite dev server proxies `/api` to a running kap-server (`KIMI_SERVER_URL`, default `http://127.0.0.1:58627`) and exposes `GET /__inspect/servers` (`vite/serverDiscovery.ts`), which scans the local kap-server instance registry (`~/.kimi-code/server/instances` + legacy `lock`) and the home token so the app can zero-config auto-connect and switch servers from the header dropdown at runtime. The per-session chat (`src/components/ChatView.tsx`) renders turn-granularly from the **transcript** surface instead of context memory: full state is read from `GET /api/v1/sessions/{id}/transcript` (initial load = newest page, refreshes re-read from the tail backwards, "load earlier" pages with `before_turn`), while `/api/v1/ws` is a delta-only channel (`transcript.ops`, grade `delta`; `transcript.reset` is ignored). Convergence reuses `@moonshot-ai/transcript`'s L2 reducer (`src/transcript/`: REST/WS clients + store; the data model and reducer come from the package, nothing is re-implemented locally). - `packages/agent-core`: the unified agent engine, including Agent, Session, profile, skills, tools, plan, permission, background, records, the in-process DI service layer (`src/services/`), and other core capabilities. - `packages/node-sdk`: the public TypeScript SDK and harness. - `packages/kosong`: the LLM / provider abstraction layer. - `packages/kaos`: the execution environment and file/process abstractions. - `packages/oauth`: Kimi OAuth and managed auth utilities. - `packages/telemetry`: shared client-side telemetry infrastructure. -- `packages/kap-server`: the Kimi Code server, backed by the DI × Scope agent engine (`@moonshot-ai/agent-core-v2`). Exposes sessions over REST + WebSocket (`/api/v1` and the native `/api/v2` RPC surface); bootstrapped from `src/start.ts` and consumed by `apps/kimi-code`. With `--debug-endpoints` on a loopback bind it additionally mounts `/api/v1/debug/*` — the same reflection dispatcher as `/api/v2` but without the channel whitelist (every scoped Service callable, `src/transport/registerDebugRoutes.ts`); internal only, repo dev scripts pass the flag. -- `packages/klient`: the client SDK — a contract-driven facade over agent-core-v2 with aggregated `global.*` / `session(id).*` / `agent(id).*` methods, zod validation on every call, and klient-level typed event forwarding. Transport is chosen once at creation via subpath entry (`@moonshot-ai/klient/http|ipc|memory`); all three return the same `Klient`. The package also hosts the e2e suites: dual-backend session/agent suites (`test/e2e/dual/`, in-memory + in-process server), `/api/v2` wire tests (`test/e2e/v2/`), the legacy `/api/v1` live suites (`test/e2e/legacy/`), and the docker e2e runner (`pnpm --filter @moonshot-ai/klient docker:e2e`). See `packages/klient/AGENTS.md`. +- `packages/transcript`: the isomorphic transcript rendering data layer — agent-granular L1 store, idempotent L2 operations, `off/turn/block/delta` L3 subscription granularity, framework-free L4 view registry, and turn-cursor pagination. Pure TypeScript (browser-safe, no engine imports) and the sole owner of all transcript wire types; consumed by `packages/kap-server` (engine events → transcript, REST + WS surface; live stores backfill history from the persisted per-agent wire records — main on first attach, any agent on demand, cold sessions rebuild any agent — with 0-based turn ordinals matching the engine's). +- `packages/kap-server`: the Kimi Code server, backed by the DI × Scope agent engine (`@moonshot-ai/agent-core-v2`). Exposes sessions over REST + WebSocket (`/api/v1` + `/api/v1/ws`); bootstrapped from `src/start.ts` and consumed by `apps/kimi-code`. The RPC surface is `/api/v1/debug/*` — a reflection dispatcher over the ENTIRE scoped DI registry (every Service callable, no whitelist; `src/transport/registerDebugRoutes.ts` + `serviceDispatcherRoutes.ts`), mounted only with `--debug-endpoints` on a loopback bind and gated by the global bearer auth; repo dev scripts pass the flag. +- `packages/klient`: the client SDK — a contract-driven facade over agent-core-v2 with aggregated `global.*` / `session(id).*` / `agent(id).*` methods, zod validation on every call, and klient-level typed event forwarding. Transport is chosen once at creation via subpath entry (`@moonshot-ai/klient/ipc|memory`); both return the same `Klient`. The package also hosts the e2e suites: the legacy `/api/v1` live suites (`test/e2e/legacy/`) and the docker e2e runner (`pnpm --filter @moonshot-ai/klient docker:e2e`). See `packages/klient/AGENTS.md`. +- `packages/server-e2e`: live e2e tests and scenarios against a running server (`KIMI_SERVER_URL`, default `http://127.0.0.1:58627`). See `packages/server-e2e/AGENTS.md`. ## Environment Requirements diff --git a/apps/kimi-inspect/package.json b/apps/kimi-inspect/package.json index 63acb6f06c..c347485a4d 100644 --- a/apps/kimi-inspect/package.json +++ b/apps/kimi-inspect/package.json @@ -23,6 +23,7 @@ }, "dependencies": { "@moonshot-ai/agent-core-v2": "workspace:^", + "@moonshot-ai/transcript": "workspace:^", "@tanstack/react-query": "^5.74.4", "react": "^19.1.0", "react-dom": "^19.1.0" diff --git a/apps/kimi-inspect/src/App.tsx b/apps/kimi-inspect/src/App.tsx index 57a5c3dcc4..4e1528d5e4 100644 --- a/apps/kimi-inspect/src/App.tsx +++ b/apps/kimi-inspect/src/App.tsx @@ -1,13 +1,12 @@ /** - * App shell — selection state, session resume, and the three WS event - * subscriptions that feed the live bus: - * core `events` — process-wide domain events - * session `interactions` — pending approvals / questions - * agent `events` — the active agent's live event stream + * App shell — selection state and session resume. There is no live event + * push anymore: the v2 socket (`/api/v2/ws`) that fed the core/session/agent + * event streams was removed server-side, so Service panels and the pending + * interactions card fetch on demand and the sidebar polls. * Layout: header / left sidebar (workspaces + sessions) / chat / inspector. */ -import { useEffect, useRef, useState } from 'react'; +import { useEffect, useState } from 'react'; import { ISessionLifecycleService } from '@moonshot-ai/agent-core-v2/app/sessionLifecycle/sessionLifecycle'; @@ -16,47 +15,14 @@ import { Inspector } from './components/Inspector'; import { ServerSwitcher } from './components/ServerSwitcher'; import { Sidebar } from './components/Sidebar'; import { useConnection } from './connection'; -import { LiveBusProvider, type Emit, type LiveEvent } from './live'; -import { Badge, errorMessage } from './ui'; +import { errorMessage } from './ui'; export function App() { - const { klient, wsState, baseUrl, disconnect } = useConnection(); + const { klient, baseUrl, disconnect } = useConnection(); const [sessionId, setSessionId] = useState(null); const [agentId, setAgentId] = useState('main'); const [ready, setReady] = useState(false); const [resumeError, setResumeError] = useState(null); - const emitRef = useRef(null); - const publish = (source: LiveEvent['source'], data: unknown) => - emitRef.current?.({ source, data, at: Date.now() }); - - // Core events — for the lifetime of the connection. - useEffect(() => { - const sub = klient.ws().listen('events', (data) => publish('core', data)); - return () => sub.dispose(); - }, [klient]); - - // Session interactions — re-subscribe when the session changes. - useEffect(() => { - if (sessionId === null || !ready) return; - const session = klient.ws().session(sessionId); - const a = session.listen('interactions', (data) => publish('session', data)); - const b = session.listen('interactions:resolved', (data) => publish('session', data)); - return () => { - a.dispose(); - b.dispose(); - }; - }, [klient, sessionId, ready]); - - // Agent events — re-subscribe when session or agent changes. - useEffect(() => { - if (sessionId === null || !ready) return; - const sub = klient - .ws() - .session(sessionId) - .agent(agentId) - .listen('events', (data) => publish('agent', data)); - return () => sub.dispose(); - }, [klient, sessionId, agentId, ready]); // Resume (materialize) the session on the server when it is selected, so // session / agent scoped Services become reachable. @@ -80,47 +46,41 @@ export function App() { }, [klient, sessionId]); // Switching servers invalidates every session/agent selection: sessions - // belong to the server they were listed from. The client and its WS - // subscriptions rebuild from the new config on their own. + // belong to the server they were listed from. useEffect(() => { setSessionId(null); setAgentId('main'); }, [baseUrl]); return ( - -
-
- KIMI INSPECT - - - ws: {wsState} - -
- -
-
- - {resumeError !== null ? ( -
- Failed to open session: {errorMessage(resumeError)} -
- ) : ( - - )} - -
+
+
+ KIMI INSPECT + +
+ +
+
+ + {resumeError !== null ? ( +
+ Failed to open session: {errorMessage(resumeError)} +
+ ) : ( + + )} +
- +
); } diff --git a/apps/kimi-inspect/src/channel/channel.test.ts b/apps/kimi-inspect/src/channel/channel.test.ts index fa1b249665..55c0f0d3bc 100644 --- a/apps/kimi-inspect/src/channel/channel.test.ts +++ b/apps/kimi-inspect/src/channel/channel.test.ts @@ -1,17 +1,16 @@ /** - * Channel layer unit tests — `ProxyChannel` URL/envelope semantics, `makeProxy` - * routing, and `WsChannel`'s ref-counted `listen`. The WS wire protocol itself - * is covered by the kap-server contract and klient's e2e suites. + * Channel layer unit tests — `ProxyChannel` URL/envelope semantics, + * `makeProxy` routing, the HTTP-only `listen` failure, and the debug-surface + * probe (`/api/v1/debug` is the only RPC surface; there is no v2 fallback). */ -import { describe, expect, it, vi } from 'vitest'; +import { afterEach, describe, expect, it, vi } from 'vitest'; import type { Event, IChannel } from './channel'; +import { probeDebugSurface } from './channels'; import { RPCError } from './errors'; import { makeProxy } from './proxy'; import { ProxyChannel } from './proxyChannel'; -import { WsChannel } from './wsChannel'; -import type { WsSocket } from './wsSocket'; const ok = (data: unknown) => ({ code: 0, msg: 'success', data, request_id: 'r1' }); @@ -24,44 +23,23 @@ function fakeFetch(envelope: unknown) { return { calls, fetchImpl }; } -function stubSocket() { - const state = { - listens: 0, - disposed: 0, - handler: undefined as ((data: unknown) => void) | undefined, - }; - const raw = { - call: vi.fn(async () => 'ws-ret'), - listen: ( - _scope: string, - _event: string, - _ids: unknown, - handler: (data: unknown) => void, - _service?: string, - ) => { - state.listens += 1; - state.handler = handler; - return { - dispose: () => { - state.disposed += 1; - }, - }; - }, - }; - return { raw, state, socket: raw as unknown as WsSocket }; -} +afterEach(() => { + vi.unstubAllGlobals(); +}); describe('ProxyChannel.call', () => { it('POSTs the command to the service base URL; no body and no header without args/token', async () => { const { calls, fetchImpl } = fakeFetch(ok({ id: 's1' })); const channel = new ProxyChannel({ - baseUrl: 'http://h:1/api/v2/session/s%201/agent/main/agentRPCService', + baseUrl: 'http://h:1/api/v1/debug/session/s%201/agent/main/agentRPCService', fetch: fetchImpl, }); const result = await channel.call('getModel', []); expect(result).toEqual({ id: 's1' }); expect(calls).toHaveLength(1); - expect(calls[0]!.url).toBe('http://h:1/api/v2/session/s%201/agent/main/agentRPCService/getModel'); + expect(calls[0]!.url).toBe( + 'http://h:1/api/v1/debug/session/s%201/agent/main/agentRPCService/getModel', + ); expect(calls[0]!.init?.method).toBe('POST'); expect(calls[0]!.init?.body).toBeUndefined(); }); @@ -69,7 +47,7 @@ describe('ProxyChannel.call', () => { it('sends the complete argument array as the JSON body, plus the bearer token', async () => { const { calls, fetchImpl } = fakeFetch(ok(null)); const channel = new ProxyChannel({ - baseUrl: 'http://h:2/api/v2/configService', + baseUrl: 'http://h:2/api/v1/debug/configService', token: 'tok', fetch: fetchImpl, }); @@ -89,7 +67,10 @@ describe('ProxyChannel.call', () => { request_id: 'r2', details: { id: 's9' }, }); - const channel = new ProxyChannel({ baseUrl: 'http://h:3/api/v2/sessionIndex', fetch: fetchImpl }); + const channel = new ProxyChannel({ + baseUrl: 'http://h:3/api/v1/debug/sessionIndex', + fetch: fetchImpl, + }); const err: unknown = await channel.call('get', ['s9']).catch((error: unknown) => error); expect(err).toBeInstanceOf(RPCError); expect((err as RPCError).code).toBe(40401); @@ -125,71 +106,54 @@ describe('makeProxy', () => { }); }); -describe('WsChannel', () => { - it('forwards calls over the socket with scope + service + ids', async () => { - const { raw, socket } = stubSocket(); - const channel = new WsChannel({ - socket, - scope: 'agent', - service: 'agentRPCService', - sessionId: 's1', - agentId: 'main', - }); - const result = await channel.call('getModel', [{}]); - expect(result).toBe('ws-ret'); - expect(raw.call).toHaveBeenCalledWith('agent', 'agentRPCService', 'getModel', [{}], { - sessionId: 's1', - agentId: 'main', +describe('ProxyChannel.listen', () => { + it('throws: the debug surface is HTTP-only, there is no event transport', () => { + const channel = new ProxyChannel({ + baseUrl: 'http://h:4/api/v1/debug/configService', + fetch: fakeFetch(ok(null)).fetchImpl, }); + expect(() => channel.listen('onDidChangeConfiguration')).toThrow(/events are not supported/); }); +}); - it('multiplexes local listeners onto one remote subscription', () => { - const { state, socket } = stubSocket(); - const channel = new WsChannel({ - socket, - scope: 'session', - service: 'sessionMetadata', - sessionId: 's1', +describe('probeDebugSurface', () => { + function stubProbeFetch(impl: (url: string, init?: RequestInit) => unknown) { + const calls: { url: string; init?: RequestInit }[] = []; + vi.stubGlobal('fetch', async (url: string | URL, init?: RequestInit) => { + calls.push({ url: String(url), init }); + return impl(String(url), init); }); - const onDidChange = channel.listen('onDidChangeMetadata'); - const seen: unknown[] = []; - const sub1 = onDidChange((e) => seen.push(['l1', e])); - const sub2 = onDidChange((e) => seen.push(['l2', e])); - expect(state.listens).toBe(1); - state.handler!({ title: 't' }); - expect(seen).toEqual([ - ['l1', { title: 't' }], - ['l2', { title: 't' }], - ]); - sub1.dispose(); - expect(state.disposed).toBe(0); - sub2.dispose(); - expect(state.disposed).toBe(1); + return calls; + } + + it('resolves when /api/v1/debug/channels answers a zero-code envelope (with bearer header)', async () => { + const calls = stubProbeFetch(() => ({ ok: true, json: async () => ({ code: 0 }) })); + await expect( + probeDebugSurface({ baseUrl: 'http://h:5/', token: 'tok' }), + ).resolves.toBeUndefined(); + expect(calls[0]!.url).toBe('http://h:5/api/v1/debug/channels'); + expect(calls[0]!.init?.headers).toEqual({ authorization: 'Bearer tok' }); }); -}); -describe('ProxyChannel.listen', () => { - it('throws without a WS binding', () => { - const channel = new ProxyChannel({ - baseUrl: 'http://h:4/api/v2/configService', - fetch: fakeFetch(ok(null)).fetchImpl, + it('throws a --debug-endpoints hint when the surface is not mounted (HTTP 404)', async () => { + stubProbeFetch(() => ({ ok: false, status: 404 })); + await expect(probeDebugSurface({ baseUrl: 'http://h:6' })).rejects.toThrow( + /--debug-endpoints/, + ); + }); + + it('throws an unreachable-server error when fetch itself fails', async () => { + vi.stubGlobal('fetch', async () => { + throw new Error('ECONNREFUSED'); }); - expect(() => channel.listen('onDidChangeConfiguration')).toThrow(/events are not supported/); + await expect(probeDebugSurface({ baseUrl: 'http://h:7' })).rejects.toThrow(/cannot reach/); }); - it('delegates to one lazily-created WsChannel when a WS binding is provided', () => { - const { state, socket } = stubSocket(); - const factory = vi.fn(() => new WsChannel({ socket, scope: 'core', service: 'configService' })); - const channel = new ProxyChannel( - { baseUrl: 'http://h:5/api/v2/configService', fetch: fakeFetch(ok(null)).fetchImpl }, - factory, - ); - const sub = channel.listen('onDidChangeConfiguration')(() => {}); - expect(factory).toHaveBeenCalledTimes(1); - expect(state.listens).toBe(1); - channel.listen('onDidSectionChange'); - expect(factory).toHaveBeenCalledTimes(1); - sub.dispose(); - expect(state.disposed).toBe(1); + it('throws a token hint when the envelope carries a non-zero code', async () => { + stubProbeFetch(() => ({ + ok: true, + json: async () => ({ code: 40101, msg: 'unauthorized' }), + })); + await expect(probeDebugSurface({ baseUrl: 'http://h:8' })).rejects.toThrow(/bearer token/); }); }); diff --git a/apps/kimi-inspect/src/channel/channel.ts b/apps/kimi-inspect/src/channel/channel.ts index fe87242509..1d71e28912 100644 --- a/apps/kimi-inspect/src/channel/channel.ts +++ b/apps/kimi-inspect/src/channel/channel.ts @@ -1,10 +1,13 @@ /** - * Transport-agnostic channel contract for the `/api/v2` client — the + * Transport-agnostic channel contract for the debug-RPC client — the * old-klient / VS Code `ProxyChannel` model: the channel is bound to one * Service (the URL carries the scope + the Service's decorator id) and * `command` is the method name, invoked by reflection on the server. - * `listen` is for the Service's `onXxx` emitter events over the persistent - * `/api/v2/ws` transport. + * + * `listen` is kept for contract completeness (Service `onXxx` emitters map to + * it), but the `/api/v1/debug` surface is HTTP-only — the v2 event socket + * (`/api/v2/ws`) that used to serve it was removed — so the HTTP channel's + * `listen` throws and panels fetch on demand instead. */ import type { ServiceIdentifier } from '@moonshot-ai/agent-core-v2/_base/di/instantiation'; @@ -31,7 +34,7 @@ export type ServiceRef = ServiceIdentifier | string; * Remote view of a Service contract: every method becomes an async wire call; * `onXxx` event members (`Event` — callables returning `IDisposable`) stay * subscribable events; plain non-function members become zero-arg property - * reads (the `/api/v2` dispatcher returns non-function members as-is). + * reads (the dispatcher returns non-function members as-is). */ export type ServiceProxy = { [K in keyof T]: T[K] extends (...args: infer A) => infer R diff --git a/apps/kimi-inspect/src/channel/channels.ts b/apps/kimi-inspect/src/channel/channels.ts index 1810f6c830..f36c6f984e 100644 --- a/apps/kimi-inspect/src/channel/channels.ts +++ b/apps/kimi-inspect/src/channel/channels.ts @@ -1,23 +1,27 @@ /** - * Protocol loading — the server's `{rpcBasePath}/channels` endpoint is its - * self-description of every wire-callable Service (name, scope, domain, - * methods + properties). On dev servers that's the whitelist-free - * `/api/v1/debug`; older/production servers fall back to the `/api/v2` - * whitelist set (`probeRpcBasePath` decides). Paired with `serviceByName`, - * each descriptor materializes 1:1 into a typed proxy of the channel layer: - * same channel name, same scope route, methods invoked by reflection. + * Protocol loading — the debug surface's `GET /api/v1/debug/channels` + * endpoint is the server's self-description of every wire-callable Service + * (name, scope, domain, methods + properties), whitelist-free. Paired with + * `serviceByName`, each descriptor materializes 1:1 into a typed proxy of + * the channel layer: same channel name, same scope route, methods invoked by + * reflection. + * + * `/api/v1/debug` is the ONLY RPC surface this app talks to (mounted by + * kap-server with `--debug-endpoints` on a loopback bind); the v2 surface + * (`/api/v2` + `/api/v2/ws`) was removed server-side, so there is no + * fallback — `probeDebugSurface` fails the connection with a clear error. */ import { createDecorator } from '@moonshot-ai/agent-core-v2/_base/di/instantiation'; +import { DEBUG_RPC_BASE, type InspectClient } from './client'; import { RPCError } from './errors'; -import type { InspectClient } from './client'; import type { ServiceProxy } from './channel'; /** Wire scope kinds reported by the channels endpoint (`app` ≡ the core route). */ export type ChannelScope = 'app' | 'session' | 'agent'; -/** Mirror of `ChannelDescriptor` in kap-server (`GET /api/v2/channels`). */ +/** Mirror of `ChannelDescriptor` in kap-server (`GET /api/v1/debug/channels`). */ export interface ChannelDescriptor { readonly name: string; readonly scope: ChannelScope; @@ -30,8 +34,7 @@ export interface ChannelDescriptor { }[]; } -/** Fetch the dynamic channel list (unwrapped from the project envelope), - * from whichever RPC surface the connection probed (`rpcBasePath`). */ +/** Fetch the dynamic channel list (unwrapped from the project envelope). */ export async function fetchChannelDescriptors( client: InspectClient, ): Promise { @@ -39,7 +42,7 @@ export async function fetchChannelDescriptors( if (client.token !== undefined && client.token !== '') { headers['authorization'] = `Bearer ${client.token}`; } - const res = await fetch(`${client.baseUrl}${client.rpcBasePath}/channels`, { headers }); + const res = await fetch(`${client.baseUrl}${DEBUG_RPC_BASE}/channels`, { headers }); const envelope = (await res.json()) as { code: number; msg: string; @@ -49,39 +52,44 @@ export async function fetchChannelDescriptors( return envelope.data; } -/** The dev server's whitelist-free debug surface (`--debug-endpoints`). */ -export const DEBUG_RPC_BASE = '/api/v1/debug' as const; -/** The stable whitelist RPC surface — fallback when debug is not mounted. */ -export const V2_RPC_BASE = '/api/v2' as const; - -export type RpcBasePath = typeof DEBUG_RPC_BASE | typeof V2_RPC_BASE; - /** - * Probe which RPC surface a server offers: dev servers started with - * `--debug-endpoints` answer `/api/v1/debug/channels`; older/production - * servers only the whitelisted `/api/v2`. Always resolves (fallback `/api/v2`). + * Verify the server mounts the debug RPC surface before the client is built. + * Resolves silently when `GET /api/v1/debug/channels` answers with a + * zero-code envelope; otherwise throws an `Error` whose message tells the + * user exactly what is wrong (unreachable server, surface not mounted → + * start kap-server with `--debug-endpoints`, or a rejected probe → check the + * token). */ -export async function probeRpcBasePath(options: { +export async function probeDebugSurface(options: { readonly baseUrl: string; readonly token?: string; -}): Promise { +}): Promise { + const headers: Record = {}; + if (options.token !== undefined && options.token !== '') { + headers['authorization'] = `Bearer ${options.token}`; + } + const url = `${options.baseUrl.replace(/\/$/, '')}${DEBUG_RPC_BASE}/channels`; + let res: Response; try { - const headers: Record = {}; - if (options.token !== undefined && options.token !== '') { - headers['authorization'] = `Bearer ${options.token}`; - } - const res = await fetch( - `${options.baseUrl.replace(/\/$/, '')}${DEBUG_RPC_BASE}/channels`, - { headers }, + res = await fetch(url, { headers }); + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + throw new Error(`cannot reach ${options.baseUrl} — is kap-server running? (${reason})`); + } + if (!res.ok) { + throw new Error( + `GET ${DEBUG_RPC_BASE}/channels answered HTTP ${res.status} — this server does not ` + + 'mount the debug RPC surface. Start kap-server with --debug-endpoints on a loopback bind.', + ); + } + const envelope = (await res.json()) as { code?: number; msg?: string }; + if (envelope.code !== 0) { + throw new Error( + `the debug surface rejected the probe (code ${envelope.code ?? '?'}: ${ + envelope.msg ?? 'no message' + }) — check the bearer token.`, ); - if (res.ok) { - const envelope = (await res.json()) as { code?: number }; - if (envelope.code === 0) return DEBUG_RPC_BASE; - } - } catch { - // fall through to the v2 fallback } - return V2_RPC_BASE; } export interface ServiceTarget { diff --git a/apps/kimi-inspect/src/channel/client.ts b/apps/kimi-inspect/src/channel/client.ts index 2f1149dd23..7ed5a3097e 100644 --- a/apps/kimi-inspect/src/channel/client.ts +++ b/apps/kimi-inspect/src/channel/client.ts @@ -1,9 +1,8 @@ /** - * Inspect client — the app's `/api/v2` entry point, in the old-klient VS Code - * `ProxyChannel` model: a three-level scope entry (`core` / `session` / - * `agent`) whose every Service handle is a `makeProxy`-materialized typed - * proxy over a service-bound channel, plus the one shared `/api/v2/ws` socket - * for scope event streams and connection state. + * Inspect client — the app's `/api/v1/debug` entry point, in the old-klient + * VS Code `ProxyChannel` model: a three-level scope entry (`core` / + * `session` / `agent`) whose every Service handle is a + * `makeProxy`-materialized typed proxy over a service-bound HTTP channel. * * const client = createInspectClient({ url: 'http://127.0.0.1:58627' }); * await client.core(ISessionIndex).list({}); @@ -12,22 +11,19 @@ * * The `agent-core-v2` service token is the whole key: its type parameter `T` * types the returned proxy, and its decorator id (`String(id)`) is the channel - * name in the URL. Calls ride HTTP (`ProxyChannel`); the proxy's `onXxx` - * emitter events and the scope streams (`events` / `interactions` / - * `interactions:resolved`) ride the one shared `WsSocket`. + * name in the URL. Calls ride HTTP (`ProxyChannel`). There is no event + * transport: the v2 socket (`/api/v2/ws`) that used to carry Service `onXxx` + * emitters and scope event streams was removed server-side, so the UI reads + * Service state on demand. (The transcript's own `/api/v1/ws` delta channel + * lives in `src/transcript/` and is unrelated to this client.) */ import type { ServiceProxy, ServiceRef } from './channel'; import { makeProxy } from './proxy'; import { ProxyChannel } from './proxyChannel'; -import { WsChannel } from './wsChannel'; -import { - WsSocket, - type WsScopeIds, - type WsScopeKind, - type WsSocketState, - type WsSubscription, -} from './wsSocket'; + +/** The dev server's whitelist-free debug surface (`--debug-endpoints` + loopback). */ +export const DEBUG_RPC_BASE = '/api/v1/debug' as const; export interface InspectAgentHandle { service(id: ServiceRef): ServiceProxy; @@ -37,34 +33,13 @@ export interface InspectSessionHandle extends InspectAgentHandle { agent(agentId: string): InspectAgentHandle; } -export interface InspectWsAgent { - listen(stream: string, handler: (data: unknown) => void): WsSubscription; -} - -export interface InspectWsSession extends InspectWsAgent { - agent(agentId: string): InspectWsAgent; -} - -/** The one owned WebSocket: connection state plus per-scope stream subscriptions. */ -export interface InspectWs { - readonly state: WsSocketState; - onDidChangeState(listener: (state: WsSocketState) => void): WsSubscription; - listen(stream: string, handler: (data: unknown) => void): WsSubscription; - session(sessionId: string): InspectWsSession; - close(): void; -} - export interface InspectClient { /** Absolute server base URL, e.g. `http://127.0.0.1:58627`. */ readonly baseUrl: string; /** Bearer token in use, when any. */ readonly token?: string; - /** RPC base path for calls and `/channels`: `/api/v1/debug` on dev servers - * (whitelist-free), `/api/v2` otherwise. Resolved by the connection probe. */ - readonly rpcBasePath: string; core(id: ServiceRef): ServiceProxy; session(sessionId: string): InspectSessionHandle; - ws(): InspectWs; } export interface InspectClientOptions { @@ -72,78 +47,34 @@ export interface InspectClientOptions { readonly url: string; /** Optional bearer token. */ readonly token?: string; - /** RPC base path for service calls + `/channels` introspection. Default - * `/api/v2`; the connection layer probes and passes `/api/v1/debug` when - * the server mounts the dev debug surface (`--debug-endpoints`). */ - readonly rpcBasePath?: string; } export function createInspectClient(options: InspectClientOptions): InspectClient { const url = options.url.replace(/\/$/, ''); - const rpcBasePath = options.rpcBasePath ?? '/api/v2'; - const socket = new WsSocket(options); /** Materialize a typed proxy for one Service on one scope binding. */ - function proxy( - scopePath: string, - scope: WsScopeKind, - ids: WsScopeIds, - id: ServiceRef, - ): ServiceProxy { + function proxy(scopePath: string, id: ServiceRef): ServiceProxy { const service = String(id); return makeProxy( - new ProxyChannel( - { baseUrl: `${url}${rpcBasePath}${scopePath}/${service}`, token: options.token }, - () => new WsChannel({ socket, scope, service, ...ids }), - ), + new ProxyChannel({ + baseUrl: `${url}${DEBUG_RPC_BASE}${scopePath}/${service}`, + token: options.token, + }), ); } - function wsListen(ids: WsScopeIds): InspectWsAgent { - return { - listen: (stream, handler) => - socket.listen( - ids.agentId !== undefined ? 'agent' : ids.sessionId !== undefined ? 'session' : 'core', - stream, - ids, - handler, - ), - }; - } - - const ws: InspectWs = { - get state() { - return socket.currentState; - }, - onDidChangeState: (listener) => socket.onDidChangeState(listener), - ...wsListen({}), - session: (sessionId) => ({ - ...wsListen({ sessionId }), - agent: (agentId) => wsListen({ sessionId, agentId }), - }), - close: () => { - socket.close(); - }, - }; - return { baseUrl: url, token: options.token, - rpcBasePath, - core: (id) => proxy('', 'core', {}, id), + core: (id) => proxy('', id), session: (sessionId) => { const scopePath = `/session/${encodeURIComponent(sessionId)}`; return { - service: (id) => proxy(scopePath, 'session', { sessionId }, id), + service: (id) => proxy(scopePath, id), agent: (agentId) => ({ - service: (subId) => - proxy(`${scopePath}/agent/${encodeURIComponent(agentId)}`, 'agent', { - sessionId, - agentId, - }, subId), + service: (subId) => proxy(`${scopePath}/agent/${encodeURIComponent(agentId)}`, subId), }), }; }, - ws: () => ws, }; } diff --git a/apps/kimi-inspect/src/channel/errors.ts b/apps/kimi-inspect/src/channel/errors.ts index b655802de5..1748e9ac59 100644 --- a/apps/kimi-inspect/src/channel/errors.ts +++ b/apps/kimi-inspect/src/channel/errors.ts @@ -1,7 +1,8 @@ /** - * Client-side RPC error surfaced when the `/api/v2` envelope carries a non-zero - * `code`. Mirrors the server envelope (`{ code, msg, data, request_id }`) — the - * numeric `code` is the stable branch key across the wire, not `instanceof`. + * Client-side RPC error surfaced when the debug-RPC envelope carries a + * non-zero `code`. Mirrors the server envelope (`{ code, msg, data, + * request_id }`) — the numeric `code` is the stable branch key across the + * wire, not `instanceof`. */ export class RPCError extends Error { constructor( diff --git a/apps/kimi-inspect/src/channel/index.ts b/apps/kimi-inspect/src/channel/index.ts index bcb9186399..4dadf528b8 100644 --- a/apps/kimi-inspect/src/channel/index.ts +++ b/apps/kimi-inspect/src/channel/index.ts @@ -4,5 +4,4 @@ export * from './client'; export * from './errors'; export * from './proxy'; export * from './proxyChannel'; -export * from './wsChannel'; -export * from './wsSocket'; +export * from './wsLike'; diff --git a/apps/kimi-inspect/src/channel/proxyChannel.ts b/apps/kimi-inspect/src/channel/proxyChannel.ts index e88143f8eb..8ab9795d51 100644 --- a/apps/kimi-inspect/src/channel/proxyChannel.ts +++ b/apps/kimi-inspect/src/channel/proxyChannel.ts @@ -1,20 +1,18 @@ /** * `ProxyChannel` — an `IChannel` bound to one Service, routing `call`s to - * kap-server's `/api/v2` HTTP surface. Every call `POST`s the method name to - * the Service base URL with the complete argument array as the JSON body, - * then unwraps the project envelope: a non-zero `code` throws `RPCError`, - * otherwise `data` is returned. Non-function members answer as property reads - * through the same route (the dispatcher returns them as-is). + * kap-server's `/api/v1/debug` HTTP surface. Every call `POST`s the method + * name to the Service base URL with the complete argument array as the JSON + * body, then unwraps the project envelope: a non-zero `code` throws + * `RPCError`, otherwise `data` is returned. Non-function members answer as + * property reads through the same route (the dispatcher returns them as-is). * - * `listen` cannot be served by HTTP; when the client supplies a WS binding - * (`events` factory) it is delegated to a lazily-created `WsChannel` bound to - * the same scope + Service, so the Service's `onXxx` emitter events work 1:1. - * Without a factory `listen` throws, matching the old HTTP-only channel. + * `listen` cannot be served by HTTP: the v2 event socket (`/api/v2/ws`) that + * used to back Service emitter events was removed, so `listen` throws and + * the UI fetches Service state on demand instead. */ import type { Event, IChannel } from './channel'; import { RPCError } from './errors'; -import type { WsChannel } from './wsChannel'; interface Envelope { readonly code: number; @@ -25,7 +23,7 @@ interface Envelope { } export interface ProxyChannelOptions { - /** Service base URL, e.g. `http://127.0.0.1:58627/api/v2[/session/:sid[/agent/:aid]]/:service`. */ + /** Service base URL, e.g. `http://127.0.0.1:58627/api/v1/debug[/session/:sid[/agent/:aid]]/:service`. */ readonly baseUrl: string; /** Optional bearer token. */ readonly token?: string; @@ -37,16 +35,13 @@ export class ProxyChannel implements IChannel { private readonly baseUrl: string; private readonly token?: string; private readonly fetchImpl: typeof fetch; - private readonly eventsFactory?: () => WsChannel; - private eventsChannel: WsChannel | undefined; - constructor(opts: ProxyChannelOptions, events?: () => WsChannel) { + constructor(opts: ProxyChannelOptions) { this.baseUrl = opts.baseUrl.replace(/\/$/, ''); this.token = opts.token; // Bind the global fetch: browsers throw "Illegal invocation" when the // native function is invoked with a non-Window receiver. this.fetchImpl = opts.fetch ?? fetch.bind(globalThis); - this.eventsFactory = events; } async call(command: string, args: unknown[] = []): Promise { @@ -71,11 +66,9 @@ export class ProxyChannel implements IChannel { return envelope.data; } - listen(event: string): Event { - if (this.eventsFactory === undefined) { - throw new Error('events are not supported on this channel (no WS binding)'); - } - this.eventsChannel ??= this.eventsFactory(); - return this.eventsChannel.listen(event); + listen(_event: string): Event { + throw new Error( + 'events are not supported on this channel (HTTP-only; the /api/v2/ws event socket was removed)', + ); } } diff --git a/apps/kimi-inspect/src/channel/wsChannel.ts b/apps/kimi-inspect/src/channel/wsChannel.ts deleted file mode 100644 index 7166a1ee87..0000000000 --- a/apps/kimi-inspect/src/channel/wsChannel.ts +++ /dev/null @@ -1,79 +0,0 @@ -/** - * `WsChannel` — an `IChannel` bound to one Service that forwards `call`s and - * `listen`s over the shared `/api/v2/ws` socket instead of HTTP. Same VS Code - * shape as `ProxyChannel` (the URL equivalent is the `{scope, service, ids}` - * triple the socket puts on each frame). `listen` multiplexes local listeners - * onto one remote subscription: the first local listener opens it, the last - * `dispose()` tears it down, and it survives reconnects until then. - */ - -import type { Event, IChannel } from './channel'; -import type { WsScopeIds, WsScopeKind, WsSocket } from './wsSocket'; - -export interface WsChannelOptions { - readonly socket: WsSocket; - readonly scope: WsScopeKind; - /** Service channel name (the decorator id, `String(id)`). */ - readonly service: string; - readonly sessionId?: string; - readonly agentId?: string; -} - -interface SharedEvent { - readonly listeners: Set<{ listener: (data: unknown) => unknown; thisArg: unknown }>; - remote?: { dispose(): void }; -} - -export class WsChannel implements IChannel { - private readonly socket: WsSocket; - private readonly scope: WsScopeKind; - private readonly service: string; - private readonly ids: WsScopeIds; - private readonly events = new Map(); - - constructor(opts: WsChannelOptions) { - this.socket = opts.socket; - this.scope = opts.scope; - this.service = opts.service; - this.ids = { sessionId: opts.sessionId, agentId: opts.agentId }; - } - - call(command: string, args: unknown[] = []): Promise { - return this.socket.call(this.scope, this.service, command, args, this.ids); - } - - listen(event: string): Event { - let shared = this.events.get(event); - if (shared === undefined) { - shared = { listeners: new Set() }; - this.events.set(event, shared); - } - return (listener, thisArg, disposables) => { - const entry = { listener: listener as (data: unknown) => unknown, thisArg }; - shared.listeners.add(entry); - shared.remote ??= this.socket.listen( - this.scope, - event, - this.ids, - (data) => { - for (const current of shared.listeners) current.listener.call(current.thisArg, data); - }, - this.service, - ); - let disposed = false; - const subscription = { - dispose: (): void => { - if (disposed) return; - disposed = true; - shared.listeners.delete(entry); - if (shared.listeners.size === 0) { - shared.remote?.dispose(); - shared.remote = undefined; - } - }, - }; - disposables?.push(subscription); - return subscription; - }; - } -} diff --git a/apps/kimi-inspect/src/channel/wsLike.ts b/apps/kimi-inspect/src/channel/wsLike.ts new file mode 100644 index 0000000000..f489ce76bb --- /dev/null +++ b/apps/kimi-inspect/src/channel/wsLike.ts @@ -0,0 +1,17 @@ +/** + * Minimal DOM-compatible WebSocket surface shared by the app's socket + * clients (today only the transcript `/api/v1/ws` client). Coding against + * this structural type keeps the clients testable with an injected fake; + * the default is the global `WebSocket` (browsers, Node ≥ 21). + */ +export interface WsLike { + readonly readyState: number; + send(data: string): void; + close(code?: number, reason?: string): void; + addEventListener(type: 'open' | 'message' | 'close' | 'error', listener: (event: never) => void): void; +} + +export interface WsLikeCtor { + new (url: string, protocols?: string | string[]): WsLike; + readonly OPEN: number; +} diff --git a/apps/kimi-inspect/src/channel/wsSocket.ts b/apps/kimi-inspect/src/channel/wsSocket.ts deleted file mode 100644 index 98eca01650..0000000000 --- a/apps/kimi-inspect/src/channel/wsSocket.ts +++ /dev/null @@ -1,456 +0,0 @@ -/** - * `/api/v2/ws` socket — the persistent WebSocket transport behind the inspect - * client's `WsChannel` (Service emitter `listen`s) and scope event streams. - * - * Speaks the kap-server v2 JSON protocol: one socket multiplexes RPC `call`s - * and event `listen`s, correlated by client-chosen ids. Adds the client-side - * safety features a long-lived devtool connection needs: `hello` handshake, - * `ping`→`pong` heartbeat answers, per-call timeouts, and opt-out automatic - * reconnect (active `listen`s are re-subscribed after a reconnect; in-flight - * calls reject on close — the server cannot resume them). - * - * The bearer token is presented at the upgrade through the - * `kimi-code.bearer.` subprotocol (the only credential channel a browser - * WebSocket has) and again in the `hello` frame for the present-only handshake - * check. Works against the DOM WebSocket (browsers, Node ≥ 21); any compatible - * implementation can be injected for tests. - */ - -import { RPCError } from './errors'; - -/** Wire scope kinds, mirroring kap-server's `ScopeKind`. */ -export type WsScopeKind = 'core' | 'session' | 'agent'; - -/** Scope coordinates carried on `call` / `listen` frames. */ -export interface WsScopeIds { - readonly sessionId?: string; - readonly agentId?: string; -} - -export type WsSocketState = 'connecting' | 'open' | 'closed'; - -export interface WsSubscription { - dispose(): void; -} - -/** Minimal DOM-compatible WebSocket surface this module codes against. */ -export interface WsLike { - readonly readyState: number; - send(data: string): void; - close(code?: number, reason?: string): void; - addEventListener(type: 'open' | 'message' | 'close' | 'error', listener: (event: never) => void): void; -} - -export interface WsLikeCtor { - new (url: string, protocols?: string | string[]): WsLike; - readonly OPEN: number; -} - -export interface WsSocketOptions { - /** Server base URL (`http(s)://host:port`) or a full `ws(s)://…/api/v2/ws` URL. */ - readonly url: string; - /** Optional bearer token. */ - readonly token?: string; - /** WebSocket implementation; defaults to the global `WebSocket`. */ - readonly WebSocketImpl?: WsLikeCtor; - /** Reconnect after an unexpected close. Default `true`. */ - readonly autoReconnect?: boolean; - /** Base delay (ms) for the reconnect backoff. Default `500`. */ - readonly reconnectDelayMs?: number; - /** Per-call deadline (ms). Default `30000`. */ - readonly callTimeoutMs?: number; -} - -interface PendingCall { - readonly resolve: (data: unknown) => void; - readonly reject: (err: Error) => void; - readonly timer: ReturnType | undefined; -} - -interface ActiveListen { - readonly scope: WsScopeKind; - readonly service?: string; - readonly event: string; - readonly ids: WsScopeIds; - readonly handler: (data: unknown) => void; - readonly onError?: (error: Error) => void; - acknowledged: boolean; -} - -export interface WsListenError { - readonly scope: WsScopeKind; - readonly service?: string; - readonly event: string; - readonly error: Error; -} - -interface ServerFrame { - readonly type: string; - readonly id?: string; - readonly data?: unknown; - readonly code?: number; - readonly msg?: string; - readonly eventId?: string; -} - -const WS_BEARER_PROTOCOL_PREFIX = 'kimi-code.bearer.'; -const DEFAULT_CALL_TIMEOUT_MS = 30_000; - -export class WsSocket { - private readonly wsUrl: string; - private readonly token?: string; - private readonly WsCtor: WsLikeCtor; - private readonly autoReconnect: boolean; - private readonly reconnectDelayMs: number; - private readonly callTimeoutMs: number; - - private ws: WsLike | undefined; - private state: WsSocketState = 'connecting'; - private manualClose = false; - private reconnectAttempt = 0; - private reconnectTimer: ReturnType | undefined; - private readyWaiters: { resolve: () => void; reject: (err: Error) => void }[] = []; - - private readonly pending = new Map(); - private readonly listens = new Map(); - private readonly eventControllers = new Map(); - private readonly stateListeners = new Set<(state: WsSocketState) => void>(); - private readonly listenErrorListeners = new Set<(event: WsListenError) => void>(); - - private seq = 0; - private readonly idPrefix = `k${Date.now().toString(36)}`; - - constructor(opts: WsSocketOptions) { - this.wsUrl = toWsUrl(opts.url); - this.token = opts.token; - const ctor = opts.WebSocketImpl ?? (globalThis.WebSocket as unknown as WsLikeCtor | undefined); - if (ctor === undefined) { - throw new Error('no WebSocket implementation available; pass WebSocketImpl'); - } - this.WsCtor = ctor; - this.autoReconnect = opts.autoReconnect ?? true; - this.reconnectDelayMs = opts.reconnectDelayMs ?? 500; - this.callTimeoutMs = opts.callTimeoutMs ?? DEFAULT_CALL_TIMEOUT_MS; - this.connect(); - } - - get currentState(): WsSocketState { - return this.state; - } - - onDidChangeState(listener: (state: WsSocketState) => void): WsSubscription { - this.stateListeners.add(listener); - return { dispose: () => this.stateListeners.delete(listener) }; - } - - onDidListenError(listener: (event: WsListenError) => void): WsSubscription { - this.listenErrorListeners.add(listener); - return { dispose: () => this.listenErrorListeners.delete(listener) }; - } - - /** RPC call over the socket; rejects on `error` frame, timeout, or close. */ - async call( - scope: WsScopeKind, - service: string, - method: string, - arg?: unknown, - ids?: WsScopeIds, - ): Promise { - await this.whenReady(); - // The socket may have dropped between `whenReady` resolving and this - // continuation running; never register a call we cannot send. - if (this.state !== 'open') { - throw new Error('ws closed'); - } - const id = this.nextId(); - const promise = new Promise((resolve, reject) => { - const timer = - this.callTimeoutMs > 0 - ? setTimeout(() => { - this.pending.delete(id); - reject(new RPCError(50001, `call timed out after ${this.callTimeoutMs}ms`)); - }, this.callTimeoutMs) - : undefined; - this.pending.set(id, { - resolve: resolve as (data: unknown) => void, - reject, - timer, - }); - }); - this.send({ type: 'call', id, scope, service, method, arg, ...ids }); - return promise; - } - - /** - * Subscribe to a scope event stream. The subscription survives reconnects - * (re-sent after each reconnect) until `dispose()`d. - */ - listen( - scope: WsScopeKind, - event: string, - ids: WsScopeIds, - handler: (data: unknown) => void, - service?: string, - onError?: (error: Error) => void, - ): WsSubscription { - const id = this.nextId(); - this.listens.set(id, { scope, service, event, ids, handler, onError, acknowledged: false }); - if (this.state === 'open') { - this.send({ type: 'listen', id, scope, service, event, ...ids }); - } - return { - dispose: () => { - if (!this.listens.delete(id)) return; - if (this.state === 'open') { - this.send({ type: 'unlisten', id }); - } - }, - }; - } - - /** Tear the socket down permanently; rejects in-flight calls. */ - close(): void { - this.manualClose = true; - if (this.reconnectTimer !== undefined) { - clearTimeout(this.reconnectTimer); - this.reconnectTimer = undefined; - } - this.setState('closed'); - this.ws?.close(); - this.ws = undefined; - this.failAll(new Error('ws closed')); - this.rejectReadyWaiters(new Error('ws closed')); - } - - // ------------------------------------------------------------------------- - // Internals - // ------------------------------------------------------------------------- - - private nextId(): string { - this.seq += 1; - return `${this.idPrefix}_${this.seq}`; - } - - private connect(): void { - this.setState('connecting'); - const protocols = - this.token !== undefined && this.token.length > 0 - ? [`${WS_BEARER_PROTOCOL_PREFIX}${this.token}`] - : undefined; - let ws: WsLike; - try { - ws = new this.WsCtor(this.wsUrl, protocols); - } catch (error) { - this.scheduleReconnect(error); - return; - } - this.ws = ws; - ws.addEventListener('open', () => { - this.onOpen(); - }); - ws.addEventListener('message', (event: { data: unknown }) => { - this.onMessage(event.data); - }); - ws.addEventListener('close', () => { - this.onClose(); - }); - ws.addEventListener('error', () => { - // The 'close' event always follows 'error'; reconnect logic lives there. - }); - } - - private onOpen(): void { - this.reconnectAttempt = 0; - this.setState('open'); - this.send({ type: 'hello', token: this.token }); - for (const [id, sub] of this.listens) { - this.send({ - type: 'listen', - id, - scope: sub.scope, - service: sub.service, - event: sub.event, - ...sub.ids, - }); - } - const waiters = this.readyWaiters; - this.readyWaiters = []; - for (const w of waiters) w.resolve(); - } - - private onMessage(raw: unknown): void { - let frame: ServerFrame; - try { - frame = JSON.parse(typeof raw === 'string' ? raw : String(raw)) as ServerFrame; - } catch { - return; - } - switch (frame.type) { - case 'ready': - case 'server_hello': - return; - case 'ping': - this.send({ type: 'pong' }); - return; - case 'result': { - const p = this.take(frame.id); - p?.resolve(frame.data); - return; - } - case 'error': { - const p = this.take(frame.id); - if (p !== undefined) { - p.reject(new RPCError(frame.code ?? 50001, frame.msg ?? 'error')); - } else { - const sub = this.listens.get(frame.id ?? ''); - if (sub !== undefined) { - this.listens.delete(frame.id ?? ''); - const error = new RPCError(frame.code ?? 50001, frame.msg ?? 'error'); - sub.onError?.(error); - queueMicrotask(() => { - for (const listener of this.listenErrorListeners) { - listener({ scope: sub.scope, service: sub.service, event: sub.event, error }); - } - }); - } - } - return; - } - case 'listen_result': { - const sub = this.listens.get(frame.id ?? ''); - if (sub !== undefined) sub.acknowledged = true; - return; - } - case 'event': { - const sub = this.listens.get(frame.id ?? ''); - if (sub === undefined) return; - if (frame.eventId === undefined) { - sub.handler(frame.data); - return; - } - const controller = new AbortController(); - const eventKey = `${frame.id}:${frame.eventId}`; - this.eventControllers.set(eventKey, controller); - const waits: Promise[] = []; - const data = { - ...(frame.data as object), - signal: controller.signal, - waitUntil: (promise: Promise) => waits.push(promise), - }; - try { - sub.handler(data); - } catch { - // Listener failures are fail-open for the server-side event. - } - void Promise.allSettled(waits).finally(() => { - if (!this.eventControllers.delete(eventKey)) return; - this.send({ type: 'event_result', id: frame.id, eventId: frame.eventId }); - }); - return; - } - case 'event_cancel': { - const key = `${frame.id}:${frame.eventId}`; - const controller = this.eventControllers.get(key); - if (controller !== undefined) { - this.eventControllers.delete(key); - controller.abort(); - } - return; - } - } - } - - private onClose(): void { - this.ws = undefined; - for (const controller of this.eventControllers.values()) controller.abort(); - this.eventControllers.clear(); - this.failAll(new Error('ws closed')); - if (this.manualClose || !this.autoReconnect) { - this.setState('closed'); - this.rejectReadyWaiters(new Error('ws closed')); - return; - } - // Transient drop: queued calls keep waiting for the reconnect. - this.scheduleReconnect(undefined); - } - - private scheduleReconnect(_cause: unknown): void { - if (this.manualClose || !this.autoReconnect) { - this.setState('closed'); - return; - } - this.reconnectAttempt += 1; - const delay = Math.min(this.reconnectDelayMs * 2 ** (this.reconnectAttempt - 1), 10_000); - this.setState('connecting'); - this.reconnectTimer = setTimeout(() => { - this.reconnectTimer = undefined; - this.connect(); - }, delay); - this.reconnectTimer.unref?.(); - } - - private whenReady(): Promise { - if (this.state === 'open') return Promise.resolve(); - if (this.state === 'closed' && this.manualClose) { - return Promise.reject(new Error('ws closed')); - } - return new Promise((resolve, reject) => { - this.readyWaiters.push({ resolve, reject }); - }); - } - - private rejectReadyWaiters(err: Error): void { - const waiters = this.readyWaiters; - this.readyWaiters = []; - for (const w of waiters) w.reject(err); - } - - private take(id: string | undefined): PendingCall | undefined { - const p = this.pending.get(id ?? ''); - if (p !== undefined) { - this.pending.delete(id ?? ''); - if (p.timer !== undefined) clearTimeout(p.timer); - } - return p; - } - - private failAll(err: Error): void { - for (const p of this.pending.values()) { - if (p.timer !== undefined) clearTimeout(p.timer); - p.reject(err); - } - this.pending.clear(); - } - - private send(frame: Record): void { - const ws = this.ws; - if (ws === undefined || ws.readyState !== this.WsCtor.OPEN) return; - try { - ws.send(JSON.stringify(frame)); - } catch { - // best-effort; the close handler handles teardown - } - } - - private setState(next: WsSocketState): void { - if (this.state === next) return; - this.state = next; - for (const listener of this.stateListeners) listener(next); - } -} - -/** Derive the `/api/v2/ws` WebSocket URL from a server base URL (or pass a full ws URL through). */ -function toWsUrl(base: string): string { - const url = new URL(base); - if (url.protocol === 'http:') url.protocol = 'ws:'; - else if (url.protocol === 'https:') url.protocol = 'wss:'; - if (url.protocol !== 'ws:' && url.protocol !== 'wss:') { - throw new Error(`unsupported URL scheme for WS transport: ${base}`); - } - if (!url.pathname.endsWith('/api/v2/ws')) { - url.pathname = `${url.pathname.replace(/\/$/, '')}/api/v2/ws`; - } - url.search = ''; - url.hash = ''; - return url.toString(); -} diff --git a/apps/kimi-inspect/src/components/ChatView.tsx b/apps/kimi-inspect/src/components/ChatView.tsx index ac6169b8a5..2f3b3d136b 100644 --- a/apps/kimi-inspect/src/components/ChatView.tsx +++ b/apps/kimi-inspect/src/components/ChatView.tsx @@ -1,80 +1,183 @@ /** - * Main view — the conversation of the active session + agent. + * Main view — the conversation of the active session + agent, rendered from + * the transcript surface (`/api/v1`): * - * History comes from `IAgentContextMemoryService.get()` (the authoritative - * context); live turns stream in through the agent `events` subscription - * (`assistant.delta` / `thinking.delta` / `tool.*`) as ephemeral entries. On - * `turn.ended` the history is refetched and the ephemeral layer is dropped, so - * the view always converges to the authoritative context. Prompts go out via - * `IAgentRPCService.prompt`; `cancel` aborts the running turn. + * - FULL state comes from the REST transcript API only: the initial load + * reads the newest page, a full refresh re-reads from the tail backwards + * until the previously loaded window is re-covered, and "Load earlier + * turns" pages further with a `before_turn` cursor. + * - The WS channel (`/api/v1/ws`) is a DELTA channel only: `transcript.ops` + * at `delta` grade; `transcript.reset` snapshots are ignored. Ops are + * buffered while a REST refresh is in flight and flushed onto the fresh + * pages — idempotent upserts and offset-placed appends make that converge. + * - Loss signals (`resync_required`, append gap, socket reconnect) trigger + * a full REST refresh; nothing is resynced from the socket itself. + * + * Rendering is turn-granular (turn → step → frame) and typed entirely by the + * transcript data model. Prompts/cancels go through the `IAgentRPCService` + * over the debug RPC surface (`/api/v1/debug`); the running indicator + * derives from transcript state (`meta.activity` / running turns). */ -import { useQuery, useQueryClient } from '@tanstack/react-query'; -import { useEffect, useMemo, useRef, useState } from 'react'; +import { + useCallback, + useEffect, + useLayoutEffect, + useRef, + useState, + useSyncExternalStore, +} from 'react'; -import { IAgentContextMemoryService } from '@moonshot-ai/agent-core-v2/agent/contextMemory/contextMemory'; import { IAgentRPCService } from '@moonshot-ai/agent-core-v2/agent/rpc/rpc'; +import { + EMPTY_AGENT_STATE, + itemId, + type AgentState, + type InteractionFrame, + type NoticeFrame, + type ToolCallFrame, + type TranscriptAttachment, + type TranscriptFrame, + type TranscriptInteraction, + type TranscriptItem, + type TranscriptMarker, + type TranscriptOperation, + type TranscriptTask, + type TranscriptTaskRef, + type TranscriptTurn, + type TranscriptUsage, + type TurnOrigin, + type TurnState, +} from '@moonshot-ai/transcript'; import { useConnection } from '../connection'; -import { eventType, payloadField, useLiveEvent } from '../live'; -import { ActionButton, Badge, ErrorLine } from '../ui'; - -interface ChatEntry { - readonly id: string; - readonly kind: 'user' | 'assistant' | 'thinking' | 'tool' | 'error'; - text: string; - name?: string; - args?: string; - output?: string; - isError?: boolean; -} +import { fetchTranscriptPage, TRANSCRIPT_PAGE_SIZE } from '../transcript/api'; +import { + createCoalescedRunner, + oldestTurnId, + recoverLoadedWindow, + TranscriptChatStore, +} from '../transcript/store'; +import { TranscriptWs } from '../transcript/ws'; +import { ActionButton, Badge, ErrorLine, JsonView, relTime } from '../ui'; + +const noopSubscribe = () => () => {}; -interface HistoryMessage { - readonly role?: string; - readonly content?: readonly { type?: string; text?: string; think?: string }[]; - readonly toolCalls?: readonly { id?: string; name?: string; arguments?: string | null }[]; - readonly isError?: boolean; +interface TranscriptChannel { + /** Null until the effect has created the store (pre-ready / no session). */ + readonly store: TranscriptChatStore | null; + readonly state: AgentState; + /** True once the initial REST page load succeeded. */ + readonly loaded: boolean; + /** Set when the initial/refresh load failed (e.g. server without transcript). */ + readonly loadError: unknown; } -function mapHistory(messages: readonly HistoryMessage[]): ChatEntry[] { - const entries: ChatEntry[] = []; - messages.forEach((m, i) => { - const parts = m.content ?? []; - const text = parts - .filter((p) => p.type === 'text') - .map((p) => p.text ?? '') - .join('\n'); - const think = parts - .filter((p) => p.type === 'think') - .map((p) => p.think ?? p.text ?? '') - .join('\n'); - if (m.role === 'user') { - entries.push({ id: `h${i}`, kind: 'user', text: text || '[non-text content]' }); - } else if (m.role === 'assistant') { - if (think !== '') entries.push({ id: `h${i}t`, kind: 'thinking', text: think }); - if (text !== '') entries.push({ id: `h${i}a`, kind: 'assistant', text }); - for (const call of m.toolCalls ?? []) { - entries.push({ - id: `h${i}c${call.id ?? ''}`, - kind: 'tool', - text: '', - name: call.name ?? 'tool', - args: call.arguments ?? undefined, +/** + * Owns the store, the REST load/refresh pipeline, and the WS delta + * subscription for one (sessionId, agentId) pair. + */ +function useTranscriptChannel( + sessionId: string | null, + agentId: string, + ready: boolean, + captureAnchor: () => void, +): TranscriptChannel { + const { baseUrl, config } = useConnection(); + const token = config.token.trim(); + const [channel, setChannel] = useState<{ store: TranscriptChatStore } | null>(null); + const [loaded, setLoaded] = useState(false); + const [loadError, setLoadError] = useState(null); + + useEffect(() => { + if (!ready || sessionId === null) return; + const store = new TranscriptChatStore(); + const authToken = token === '' ? undefined : token; + let disposed = false; + /** While a REST refresh is in flight, WS ops are buffered, then flushed. */ + let fetching = true; + let buffer: TranscriptOperation[] = []; + + /** Full-state (re)load: newest page first, then backwards over `before_turn`. */ + const refresh = createCoalescedRunner(async (): Promise => { + fetching = true; + buffer = []; + // The window's oldest turn is the re-cover anchor: after a refresh the + // server window may have shifted, and only re-loading up to THIS turn + // preserves the previously loaded history. + const prevOldest = oldestTurnId(store.getState().items); + if (prevOldest !== undefined) captureAnchor(); + try { + const newest = await fetchTranscriptPage({ + baseUrl, + token: authToken, + sessionId, + agentId, + pageSize: TRANSCRIPT_PAGE_SIZE, }); + if (disposed) return; + store.applyPage(newest, { replace: true }); + // Re-cover the previously loaded window for refreshes (a no-op on the + // initial load, where there is no previous oldest turn). + await recoverLoadedWindow( + store, + prevOldest, + (beforeTurn) => + fetchTranscriptPage({ + baseUrl, + token: authToken, + sessionId, + agentId, + beforeTurn, + pageSize: TRANSCRIPT_PAGE_SIZE, + }), + () => disposed, + ); + if (!disposed) { + setLoaded(true); + setLoadError(null); + } + } catch (error) { + if (!disposed) setLoadError(error); + } finally { + fetching = false; + if (buffer.length > 0) store.applyOps(buffer); + buffer = []; } - } else if (m.role === 'tool') { - entries.push({ - id: `h${i}r`, - kind: 'tool', - text: '', - name: 'result', - output: text, - isError: m.isError, - }); - } - // system / developer messages (the system prompt) are not shown. - }); - return entries; + }); + + const ws = new TranscriptWs({ + url: baseUrl, + token: authToken, + sessionId, + agentId, + handlers: { + onOps: (aid, ops) => { + if (aid !== agentId) return; + if (fetching) buffer.push(...ops); + else store.applyOps(ops); + }, + onResyncRequired: refresh, + onReconnected: refresh, + }, + }); + store.onGap = refresh; + setChannel({ store }); + setLoaded(false); + setLoadError(null); + refresh(); + return () => { + disposed = true; + ws.close(); + setChannel(null); + }; + }, [sessionId, agentId, ready, baseUrl, token, captureAnchor]); + + const state = useSyncExternalStore( + channel?.store.subscribe ?? noopSubscribe, + () => channel?.store.getState() ?? EMPTY_AGENT_STATE, + ); + return { store: channel?.store ?? null, state, loaded, loadError }; } export function ChatView({ @@ -86,120 +189,90 @@ export function ChatView({ agentId: string; ready: boolean; }) { - const { klient } = useConnection(); - const queryClient = useQueryClient(); - const [stream, setStream] = useState([]); + const { klient, baseUrl, config } = useConnection(); const [input, setInput] = useState(''); - const [running, setRunning] = useState(false); const [sendError, setSendError] = useState(null); - const bottomRef = useRef(null); - - const enabled = sessionId !== null && ready; - const history = useQuery({ - queryKey: ['history', sessionId, agentId], - queryFn: () => - klient - .session(sessionId as string) - .agent(agentId) - .service(IAgentContextMemoryService) - .get(), - enabled, - }); - - const refetchHistory = useRef | undefined>(undefined); - const scheduleRefetch = () => { - if (refetchHistory.current !== undefined) clearTimeout(refetchHistory.current); - refetchHistory.current = setTimeout(() => { - void queryClient.invalidateQueries({ queryKey: ['history', sessionId, agentId] }); - setStream([]); - setRunning(false); - }, 500); + const [loadingOlder, setLoadingOlder] = useState(false); + const [olderError, setOlderError] = useState(null); + const scrollRef = useRef(null); + /** Distance from the scroll bottom captured before a prepend (restore anchor). */ + const anchorRef = useRef(null); + /** Whether the viewport was pinned to the bottom before the last update. */ + const stickBottomRef = useRef(true); + + const captureAnchor = useCallback(() => { + const el = scrollRef.current; + if (el !== null) anchorRef.current = el.scrollHeight - el.scrollTop; + }, []); + + const { store, state, loaded, loadError } = useTranscriptChannel( + sessionId, + agentId, + ready, + captureAnchor, + ); + const items = state.items; + + useLayoutEffect(() => { + const el = scrollRef.current; + if (el === null) return; + if (anchorRef.current !== null) { + el.scrollTop = el.scrollHeight - anchorRef.current; + anchorRef.current = null; + return; + } + if (stickBottomRef.current) el.scrollTop = el.scrollHeight; + }, [items]); + + const onScroll = () => { + const el = scrollRef.current; + if (el === null) return; + stickBottomRef.current = el.scrollHeight - el.scrollTop - el.clientHeight < 80; }; - useLiveEvent((event) => { - if (event.source !== 'agent') return; - const type = eventType(event); - const data = event.data as Record; - switch (type) { - case 'turn.started': - setRunning(true); - return; - case 'assistant.delta': - case 'thinking.delta': { - const turnId = payloadField(data, 'turnId', '?'); - const kind = type === 'assistant.delta' ? 'assistant' : 'thinking'; - const id = `stream:${turnId}:${kind}`; - const delta = payloadField(data, 'delta', ''); - setStream((prev) => { - const found = prev.find((e) => e.id === id); - if (found === undefined) return [...prev, { id, kind, text: delta }]; - return prev.map((e) => (e.id === id ? { ...e, text: e.text + delta } : e)); - }); - return; - } - case 'tool.call.started': { - const callId = payloadField(data, 'toolCallId', String(Math.random())); - setStream((prev) => [ - ...prev, - { - id: `stream:tool:${callId}`, - kind: 'tool', - text: '', - name: payloadField(data, 'name', 'tool'), - args: typeof data['args'] === 'string' ? data['args'] : JSON.stringify(data['args'] ?? ''), - }, - ]); - return; - } - case 'tool.result': { - const callId = payloadField(data, 'toolCallId', ''); - const output = typeof data['output'] === 'string' ? data['output'] : JSON.stringify(data['output']); - const id = `stream:tool:${callId}`; - setStream((prev) => { - const found = prev.find((e) => e.id === id); - if (found === undefined) { - return [...prev, { id, kind: 'tool', text: '', name: 'result', output, isError: Boolean(data['isError']) }]; - } - return prev.map((e) => (e.id === id ? { ...e, output, isError: Boolean(data['isError']) } : e)); - }); - return; - } - case 'turn.ended': - case 'prompt.completed': - case 'prompt.aborted': - case 'compaction.completed': - scheduleRefetch(); - return; - case 'error': { - const message = - typeof data['message'] === 'string' - ? data['message'] - : JSON.stringify(data).slice(0, 300); - setStream((prev) => [ - ...prev, - { id: `stream:err:${Date.now()}`, kind: 'error', text: message }, - ]); - setRunning(false); - return; - } + const loadOlder = async () => { + if (sessionId === null || loadingOlder || store === null) return; + const oldest = oldestTurnId(items); + if (oldest === undefined) return; + captureAnchor(); + setLoadingOlder(true); + setOlderError(null); + try { + const token = config.token.trim(); + const page = await fetchTranscriptPage({ + baseUrl, + token: token === '' ? undefined : token, + sessionId, + agentId, + beforeTurn: oldest, + pageSize: TRANSCRIPT_PAGE_SIZE, + }); + store.applyPage(page); + } catch (error) { + anchorRef.current = null; + setOlderError(error); + } finally { + setLoadingOlder(false); } - }); + }; - const entries = useMemo( - () => [...mapHistory((history.data ?? []) as readonly HistoryMessage[]), ...stream], - [history.data, stream], - ); + const running = + state.meta.activity === 'turn' || + items.some((item) => item.kind === 'turn' && item.state === 'running'); - useEffect(() => { - bottomRef.current?.scrollIntoView({ block: 'end' }); - }, [entries.length, stream]); + // Interactions render inline at their anchor tool frame; entities whose + // anchor frame is outside the loaded window collect here. + const anchoredToolCallIds = collectToolCallIds(items); + const unanchoredInteractions = [...state.interactions.values()].filter( + (interaction) => !anchoredToolCallIds.has(interaction.toolCallId), + ); + const latestTodo = [...state.todos.values()].at(-1); const send = async () => { if (sessionId === null || input.trim() === '' || running) return; const text = input.trim(); setInput(''); setSendError(null); - setStream((prev) => [...prev, { id: `optimistic:${Date.now()}`, kind: 'user', text }]); try { await klient .session(sessionId) @@ -241,21 +314,73 @@ export function ChatView({ {sessionId} agent: {agentId} {running ? turn running : idle} + {state.pendingInteractions.size > 0 ? ( + {state.pendingInteractions.size} pending + ) : null}
-
- {history.isError ? : null} - {entries.length === 0 && !history.isLoading ? ( -
Empty context — send a prompt below.
+
+ {state.hasMoreOlder ? ( +
+ void loadOlder()} disabled={loadingOlder}> + {loadingOlder ? 'Loading…' : 'Load earlier turns'} + +
+ ) : null} + {olderError !== null ? ( +
+ +
+ ) : null} + {loadError !== null ? ( +
+ +
+ Failed to load the transcript — the server may be too old to expose the transcript + API. +
+
+ ) : null} + {items.length === 0 && loadError === null ? ( +
+ {loaded ? 'Empty transcript — send a prompt below.' : 'Loading transcript…'} +
) : null} - {entries.map((entry) => ( - + {latestTodo !== undefined && latestTodo.items.length > 0 ? ( +
+
todo (latest)
+ {latestTodo.items.map((entry, i) => ( +
+ + {entry.status === 'done' ? '✔' : entry.status === 'in_progress' ? '◐' : '□'} + + + {entry.title} + +
+ ))} +
+ ) : null} + {items.map((item) => ( + + ))} + {unanchoredInteractions.map((interaction) => ( + ))} -
- {sendError !== null ?
: null} + {sendError !== null ? ( +
+ +
+ ) : null}