diff --git a/packages/kap-server/src/routes/files.ts b/packages/kap-server/src/routes/files.ts index abae647c6..240339357 100644 --- a/packages/kap-server/src/routes/files.ts +++ b/packages/kap-server/src/routes/files.ts @@ -137,11 +137,11 @@ export function registerFilesRoutes(app: FilesRouteHost, core: Scope): void { return; } reply.send(okEnvelope(meta, req.id)); - } catch (err) { - sendMappedError(reply as unknown as FilesReply, req.id, err); + } catch (error) { + sendMappedError(reply as unknown as FilesReply, req.id, error); } - } catch (err) { - sendMappedError(reply as unknown as FilesReply, req.id, err); + } catch (error) { + sendMappedError(reply as unknown as FilesReply, req.id, error); } }, ); @@ -171,14 +171,25 @@ export function registerFilesRoutes(app: FilesRouteHost, core: Scope): void { const store = core.accessor.get(IFileService); const { meta, stream } = await store.get(file_id); const r = reply as unknown as FilesReply; + const size = meta.size; r.type(meta.media_type) - .header('content-disposition', buildContentDisposition(meta.name)) - .header('content-length', meta.size) - .header('etag', `"${meta.id}-${meta.size}"`) - .code(200); + .header('content-disposition', buildContentDisposition(meta.name, meta.media_type)) + .header('accept-ranges', 'bytes') + .header('etag', `"${meta.id}-${size}"`); + + const range = parseRange((req as unknown as FastifyRequestLike).headers['range'], size); + if (range) { + const data = await readStream(stream); + r.header('content-range', `bytes ${range.start}-${range.end}/${size}`) + .header('content-length', range.length) + .code(206); + return r.send(data.subarray(range.start, range.end + 1)) as unknown as void; + } + + r.header('content-length', size).code(200); return r.send(stream) as unknown as void; - } catch (err) { - sendMappedError(reply as unknown as FilesReply, req.id, err); + } catch (error) { + sendMappedError(reply as unknown as FilesReply, req.id, error); return; } }, @@ -204,8 +215,8 @@ export function registerFilesRoutes(app: FilesRouteHost, core: Scope): void { const store = core.accessor.get(IFileService); await store.delete(file_id); reply.send(okEnvelope({ deleted: true as const }, req.id)); - } catch (err) { - sendMappedError(reply as unknown as FilesReply, req.id, err); + } catch (error) { + sendMappedError(reply as unknown as FilesReply, req.id, error); } }, ); @@ -266,9 +277,33 @@ function readFieldNumber(field: unknown): number | undefined { return undefined; } -function buildContentDisposition(name: string): string { - if (/^[\w. ()+[\]-]+$/.test(name)) { - return `attachment; filename="${name}"`; +async function readStream(stream: NodeJS.ReadableStream): Promise { + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.from(chunk as string | Uint8Array)); } - return 'attachment'; + return Buffer.concat(chunks); +} + +function buildContentDisposition(name: string, mediaType?: string): string { + const disposition = /^(image|video|audio)\//.test(mediaType ?? '') ? 'inline' : 'attachment'; + if (/^[\w. ()+[\]-]+$/.test(name)) { + return `${disposition}; filename="${name}"`; + } + return disposition; +} + +function parseRange( + value: string | string[] | undefined, + size: number, +): { start: number; end: number; length: number } | undefined { + const header = Array.isArray(value) ? value[0] : value; + const match = size > 0 ? /^bytes=(\d*)-(\d*)$/i.exec(header?.trim() ?? '') : null; + if (!match || (match[1] === '' && match[2] === '')) return undefined; + + const start = match[1] === '' ? Math.max(size - Number(match[2]), 0) : Number(match[1]); + let end = match[1] === '' || match[2] === '' ? size - 1 : Number(match[2]); + if (![start, end].every(Number.isSafeInteger) || start < 0 || start >= size || start > end) return undefined; + end = Math.min(end, size - 1); + return { start, end, length: end - start + 1 }; } diff --git a/packages/kap-server/src/routes/prompts.ts b/packages/kap-server/src/routes/prompts.ts index 462aa7002..e1b98e817 100644 --- a/packages/kap-server/src/routes/prompts.ts +++ b/packages/kap-server/src/routes/prompts.ts @@ -5,12 +5,20 @@ * working against server-v2. */ +import { createWriteStream } from 'node:fs'; +import { mkdir, stat } from 'node:fs/promises'; +import { extname, join } from 'node:path'; +import { pipeline } from 'node:stream/promises'; + import { + IBootstrapService, IAgentLifecycleService, IAgentPromptLegacyService, + IFileService, ISessionLifecycleService, isKimiError, KimiError, + type GetResult, type Scope, } from '@moonshot-ai/agent-core-v2'; import { @@ -21,6 +29,7 @@ import { promptSteerResultSchema, promptSubmissionSchema, promptSubmitResultSchema, + type PromptSubmission, } from '@moonshot-ai/protocol'; import { z } from 'zod'; @@ -55,6 +64,14 @@ const sessionIdParamSchema = z.object({ const validationDetailsSchema = z.array(z.object({ path: z.string(), message: z.string() })); const authProviderDetailsSchema = z.object({ provider_id: z.string() }); const authModelDetailsSchema = z.object({ model_id: z.string(), provider_id: z.string() }).partial(); +const VIDEO_EXT_BY_MIME: Record = { + 'video/mp4': '.mp4', + 'video/quicktime': '.mov', + 'video/webm': '.webm', + 'video/x-msvideo': '.avi', + 'video/x-matroska': '.mkv', + 'video/mpeg': '.mpeg', +}; async function resolveLegacy( core: Scope, @@ -131,8 +148,13 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { async (req, reply) => { try { const { session_id } = req.params; - const legacy = await resolveLegacy(core, session_id, req.body.agent_id); - const result = await legacy.submit(req.body); + const resolvedBody = await resolvePromptMediaFiles( + req.body, + core.accessor.get(IFileService), + core.accessor.get(IBootstrapService).cacheDir, + ); + const legacy = await resolveLegacy(core, session_id, resolvedBody.agent_id); + const result = await legacy.submit(resolvedBody); reply.send(okEnvelope(result, req.id)); } catch (error) { sendMappedError(reply, req.id, error); @@ -212,6 +234,50 @@ export function registerPromptsRoutes(app: PromptRouteHost, core: Scope): void { app.post(actionRoute.path, actionRoute.options, actionRoute.handler as Parameters[2]); } +async function resolvePromptMediaFiles( + body: PromptSubmission, + store: IFileService, + cacheDir: string, +): Promise { + const content: PromptSubmission['content'] = []; + for (const part of body.content) { + if (part.type !== 'video' || part.source.kind !== 'file') { + content.push(part); + continue; + } + + const file = await store.get(part.source.file_id); + if (!file.meta.media_type.toLowerCase().startsWith('video/')) { + throw new KimiError( + 'validation.failed', + `file ${file.meta.id} is ${file.meta.media_type}, not a video`, + ); + } + const cachePath = await materializeVideoToCache(file, cacheDir); + content.push({ type: 'text', text: `` }); + } + return { ...body, content }; +} + +async function materializeVideoToCache(file: GetResult, cacheDir: string): Promise { + await mkdir(cacheDir, { recursive: true }); + const ext = extname(file.meta.name) || (VIDEO_EXT_BY_MIME[file.meta.media_type.toLowerCase()] ?? '.bin'); + const target = join(cacheDir, `${file.meta.id}${ext}`); + const info = await stat(target).catch(() => undefined); + if (info?.size === file.meta.size) return target; + + await pipeline(file.stream, createWriteStream(target)); + return target; +} + +function escapeAttribute(value: string): string { + return value + .replaceAll('&', '&') + .replaceAll('"', '"') + .replaceAll('<', '<') + .replaceAll('>', '>'); +} + function sendMappedError( reply: { send(payload: unknown): unknown }, requestId: string, @@ -223,6 +289,9 @@ function sendMappedError( case 'agent.not_found': reply.send(errEnvelope(ErrorCode.SESSION_NOT_FOUND, err.message, requestId, err.stack)); return; + case 'file.not_found': + reply.send(errEnvelope(ErrorCode.FILE_NOT_FOUND, err.message, requestId, err.stack)); + return; case 'prompt.not_found': reply.send(errEnvelope(ErrorCode.PROMPT_NOT_FOUND, err.message, requestId, err.stack)); return; diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index a2e74dbcf..413444dd1 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts @@ -120,10 +120,17 @@ interface SessionState { export const DEFAULT_MAX_BUFFER_SIZE = 1000; const GLOBAL_SESSION_ID = '__global__'; +async function disposeSessionState(state: SessionState): Promise { + for (const d of state.lifecycleDisposables) d.dispose(); + for (const d of state.agentDisposables.values()) d.dispose(); + await state.journal.close(); +} + export class SessionEventBroadcaster { private readonly sessions = new Map(); private readonly maxBufferSize: number; private readonly coreEventSubscription: IDisposable; + private closed = false; constructor( private readonly opts: { @@ -253,16 +260,17 @@ export class SessionEventBroadcaster { } async close(): Promise { + if (this.closed) return; + this.closed = true; this.coreEventSubscription.dispose(); for (const state of this.sessions.values()) { - for (const d of state.lifecycleDisposables) d.dispose(); - for (const d of state.agentDisposables.values()) d.dispose(); - await state.journal.close(); + await disposeSessionState(state); } this.sessions.clear(); } private async ensureState(sessionId: string): Promise { + if (this.closed) return undefined; let state = this.sessions.get(sessionId); if (state !== undefined) return state; @@ -273,6 +281,10 @@ export class SessionEventBroadcaster { sessionJournalPath(this.opts.eventsDir, sessionId), this.opts.logger, ); + if (this.closed) { + await journal.close(); + return undefined; + } state = { sessionId, journal, @@ -285,8 +297,15 @@ export class SessionEventBroadcaster { knownInteractions: new Map(), }; this.sessions.set(sessionId, state); - this.attachAgents(sessionId, session, state); - this.attachInteractions(sessionId, session, state); + try { + this.attachAgents(sessionId, session, state); + this.attachInteractions(sessionId, session, state); + } catch (error) { + this.sessions.delete(sessionId); + await disposeSessionState(state); + if (error instanceof Error && error.message === 'InstantiationService has been disposed') return undefined; + throw error; + } return state; } diff --git a/packages/kap-server/test/files.test.ts b/packages/kap-server/test/files.test.ts index 0b2f5b131..6a9060d1d 100644 --- a/packages/kap-server/test/files.test.ts +++ b/packages/kap-server/test/files.test.ts @@ -254,6 +254,55 @@ describe('POST /api/v1/files (server-v2)', () => { expect((res.json() as Envelope<{ name: string }>).data?.name).toBe('overridden.txt'); }); + it('serves byte ranges with 206 Partial Content for video playback', async () => { + const r = await boot(); + const data = Buffer.from('0123456789abcdefghijklmnopqrstuvwxyz'); + const mp = buildMultipart({ + file: { + fieldName: 'file', + filename: 'clip.mp4', + contentType: 'video/mp4', + data, + }, + }); + const upRes = await appOf(r).inject({ + method: 'POST', + url: '/api/v1/files', + payload: mp.body, + headers: { 'content-type': mp.contentType }, + }); + const meta = (upRes.json() as Envelope<{ id: string; size: number }>).data!; + + const full = await appOf(r).inject({ + method: 'GET', + url: `/api/v1/files/${meta.id}`, + }); + expect(full.statusCode).toBe(200); + expect(full.headers['accept-ranges']).toBe('bytes'); + expect(full.headers['content-type']).toBe('video/mp4'); + expect(String(full.headers['content-disposition'])).toMatch(/^inline;/); + expect(full.rawPayload).toEqual(data); + + const part = await appOf(r).inject({ + method: 'GET', + url: `/api/v1/files/${meta.id}`, + headers: { range: 'bytes=4-9' }, + }); + expect(part.statusCode).toBe(206); + expect(part.headers['content-range']).toBe(`bytes 4-9/${data.length}`); + expect(part.headers['content-length']).toBe('6'); + expect(part.rawPayload).toEqual(data.subarray(4, 10)); + + const tail = await appOf(r).inject({ + method: 'GET', + url: `/api/v1/files/${meta.id}`, + headers: { range: 'bytes=30-' }, + }); + expect(tail.statusCode).toBe(206); + expect(tail.headers['content-range']).toBe(`bytes 30-${data.length - 1}/${data.length}`); + expect(tail.rawPayload).toEqual(data.subarray(30)); + }); + it('missing file part → 40001 validation error', async () => { const r = await boot(); const boundary = '------WebKitFormBoundaryNoFile'; diff --git a/packages/kap-server/test/prompts.test.ts b/packages/kap-server/test/prompts.test.ts index 2a53f5830..67156352a 100644 --- a/packages/kap-server/test/prompts.test.ts +++ b/packages/kap-server/test/prompts.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -130,6 +130,40 @@ describe('server-v2 /api/v1 prompts', () => { expect(Array.isArray(list.body.data.queued)).toBe(true); }); + it('materializes uploaded video prompts into cache path tags', async () => { + const id = await createSession(home as string); + await createMainAgent(id); + const videoBytes = Buffer.from('tiny fake mp4 bytes'); + const form = new FormData(); + form.set('file', new Blob([videoBytes], { type: 'video/mp4' }), 'clip.mp4'); + const uploadRes = await fetch(`${base}/api/v1/files`, { + method: 'POST', + headers: authHeaders(server as RunningServer), + body: form, + } as never); + const uploaded = (await uploadRes.json()) as Envelope<{ id: string }>; + expect(uploaded.code).toBe(0); + + const submitted = await call('POST', `/api/v1/sessions/${id}/prompts`, { + content: [ + { type: 'text', text: 'what happens in this video?' }, + { type: 'video', source: { kind: 'file', file_id: uploaded.data.id } }, + ], + }); + expect(submitted.body.code).toBe(0); + + const content = submitted.body.data.content as Array<{ type: string; text?: string }>; + expect(content).toHaveLength(2); + expect(content[0]).toEqual({ type: 'text', text: 'what happens in this video?' }); + expect(content[1]?.type).toBe('text'); + const match = /