From c4f1005f43a940ee5005a87d6410cd75e5572565 Mon Sep 17 00:00:00 2001 From: eleith Date: Thu, 1 Oct 2026 12:38:17 -0700 Subject: [PATCH 1/2] [web] share progress console and SSE framing Extract the preview presentation into ProgressConsole with caller-provided wording and request-local rendering types. Preserve preview copy/styles through a thin wrapper and isolate log IDs per component instance. Share database-free SSE encoding and headers without changing publish/preview job polling, framing, or cursor behavior. Add Node SSR component tests and framing/job-stream regressions using existing Svelte tooling. Verified on Node v26.4.0 and pnpm 11.2.2: pnpm lint, pnpm check, and pnpm test pass. Web has 133 passing tests, up from 95. The owner manually ran preview/publish and confirmed they still work. Known baseline failure: pnpm build stops in @luzzle/web prebuild because apps/web/config.yaml is missing. This remains a failing root build gate, not a passing check. No generation endpoint, token output, worker execution, or proxy changes are included. --- .../src/lib/components/ProgressConsole.svelte | 478 +++++++++++++++++ .../lib/components/ProgressConsole.test.ts | 147 ++++++ apps/web/src/lib/components/progress.ts | 21 + apps/web/src/lib/server/sse.test.ts | 24 + apps/web/src/lib/server/sse.ts | 13 + .../src/lib/server/workflow/stream.test.ts | 179 +++++++ apps/web/src/lib/server/workflow/stream.ts | 17 +- .../preview/[jobId]/WorkflowConsole.svelte | 499 +----------------- apps/web/vitest.config.ts | 5 + 9 files changed, 883 insertions(+), 500 deletions(-) create mode 100644 apps/web/src/lib/components/ProgressConsole.svelte create mode 100644 apps/web/src/lib/components/ProgressConsole.test.ts create mode 100644 apps/web/src/lib/components/progress.ts create mode 100644 apps/web/src/lib/server/sse.test.ts create mode 100644 apps/web/src/lib/server/sse.ts create mode 100644 apps/web/src/lib/server/workflow/stream.test.ts diff --git a/apps/web/src/lib/components/ProgressConsole.svelte b/apps/web/src/lib/components/ProgressConsole.svelte new file mode 100644 index 00000000..4fc07c52 --- /dev/null +++ b/apps/web/src/lib/components/ProgressConsole.svelte @@ -0,0 +1,478 @@ + + +
+ + {#if status === 'completed'} + + {:else if status === 'running'} + + {:else if status === 'enqueued'} + + {:else if status === 'failed' || status === 'expired'} + + {:else} + + {/if} + + +
+ {#if statusText} +
{statusText.title}
+
+ {#if statusText.description} + {statusText.description} + {:else if statusText.durationLabel} + {formatDuration(overallDuration)} {statusText.durationLabel} + {/if} +
+ {/if} +
+
+ +{#if errorMessage && status === 'failed'} +
+ + {errorMessage} +
+{/if} + +{#if phases.length > 0} +
+ {#each phases as phase (phase.phase)} + {@const hasLogs = (logs[phase.phase]?.length ?? 0) > 0} +
+
+ + {#if phase.status === 'completed'} + + {:else if phase.status === 'running'} + + {:else if phase.status === 'failed'} + + {:else if phase.status === 'skipped'} + + {:else} + + {/if} + + {phase.phase} + {#if phase.message} + + {phase.message} + + {/if} + + {#if phase.started_at} + + {formatDuration((phase.finished_at || Date.now()) - phase.started_at)} + + {/if} + +
+ + {#if hasLogs} +
+
+ {logs[phase.phase]?.length || 0} lines +
+ {#if phase.started_at} + + {formatDuration((phase.finished_at || Date.now()) - phase.started_at)} + + {/if} + +
+
+
handleScroll(phase.phase, e)} + > + {#each logs[phase.phase] ?? [] as log (log.line_number)} +
+ + {new Date(log.ts).toISOString().split('T')[1].slice(0, -5)} + + {log.message} +
+ {/each} +
+
+ {/if} +
+ {/each} +
+{/if} + + diff --git a/apps/web/src/lib/components/ProgressConsole.test.ts b/apps/web/src/lib/components/ProgressConsole.test.ts new file mode 100644 index 00000000..9ce8bd45 --- /dev/null +++ b/apps/web/src/lib/components/ProgressConsole.test.ts @@ -0,0 +1,147 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { render } from 'svelte/server' +import ProgressConsole from './ProgressConsole.svelte' +import WorkflowConsole from '../../routes/admin/piece/[...path]/preview/[jobId]/WorkflowConsole.svelte' +import type { ProgressPhase, ProgressLog } from './progress.js' + +const started = Date.parse('2026-01-02T03:04:00Z') +const phases: ProgressPhase[] = [ + { + phase: 'render', + status: 'completed', + started_at: started, + finished_at: started + 65_000, + message: 'Rendered successfully' + } +] +const logs: Record = { + render: [ + { phase: 'render', line_number: 1, ts: started, level: 'stdout', message: 'Output ready' } + ] +} + +function text(html: string) { + return html + .replace(/<[^>]+>/g, '') + .replace(/\s+/g, ' ') + .trim() +} + +beforeEach(() => { + vi.spyOn(Date, 'now').mockReturnValue(started + 75_000) +}) +afterEach(() => vi.restoreAllMocks()) + +describe('preview compatibility wrapper', () => { + test.each([ + [ + 'expired', + 'Preview expired', + 'Assets are purged after 2 days. Re-run the preview from the editor.' + ], + ['enqueued', 'Enqueued', 'Waiting for worker…'], + ['running', 'Rendering preview', '1m 5s elapsed'], + ['failed', 'Preview failed', '1m 5s stopped at'], + ['completed', 'Preview completed', '1m 5s total time'] + ])('retains wording and duration for %s', (status, title, subtitle) => { + const result = render(WorkflowConsole, { props: { status, phases, logs } }) + expect(text(result.body)).toContain(`${title} ${subtitle}`) + expect(result.body).toContain(`status-${status}`) + expect(result.body).toContain('phase-icon') + expect(result.body).toContain('log-viewport') + expect(text(result.body)).toContain('Rendered successfully') + expect(text(result.body)).toContain('03:04:00') + expect(text(result.body)).toContain('Output ready') + }) + + test('preserves an empty summary for an unrecognized status', () => { + const result = render(WorkflowConsole, { props: { status: 'unknown', phases: [], logs: {} } }) + expect(result.body).not.toContain('status-title') + expect(result.body).not.toContain('status-sub') + }) + + test.each(['failed', 'running', 'completed', 'expired'])( + 'shows the existing error strip only for failed status: %s', + (status) => { + const result = render(WorkflowConsole, { + props: { status, phases: [], logs: {}, errorMessage: 'Render failed' } + }) + expect(result.body.includes('error-strip')).toBe(status === 'failed') + expect(text(result.body).includes('Render failed')).toBe(status === 'failed') + } + ) +}) + +describe('shared progress presentation', () => { + test('accepts caller wording and progress/log rows without job IDs', () => { + const result = render(ProgressConsole, { + props: { + status: 'completed', + statusText: { title: 'Content ready', durationLabel: 'finished in' }, + phases, + logs + } + }) + expect(text(result.body)).toContain('Content ready 1m 5s finished in') + expect(text(result.body)).not.toContain('Preview') + expect(text(result.body)).not.toContain('worker') + expect(text(result.body)).toContain('1 lines') + expect(result.body).toContain('Auto-scroll locked') + }) + + test.each([ + [[], '0s'], + [[{ ...phases[0], started_at: 0 }], '0s'], + [[{ ...phases[0], finished_at: null }], '1m 15s'], + [[{ ...phases[0], finished_at: started + 5000 }], '5s'], + [[{ ...phases[0], finished_at: started - 1000 }], '0s'] + ])('retains elapsed-time formatting for %j', (items, duration) => { + const result = render(ProgressConsole, { + props: { + status: 'running', + statusText: { title: 'Working', durationLabel: 'elapsed' }, + phases: items, + logs: {} + } + }) + expect(text(result.body)).toContain(`Working ${duration} elapsed`) + }) + + test('retains phase statuses and stderr/error highlighting', () => { + const items = ['running', 'failed', 'skipped', 'waiting'].map((status) => ({ + ...phases[0], + phase: status, + status + })) + const result = render(ProgressConsole, { + props: { + status: 'failed', + phases: items, + logs: { + failed: [ + { ...logs.render[0], phase: 'failed', level: 'stderr', message: 'Failure details' } + ] + } + } + }) + for (const { status } of items) expect(result.body).toContain(`status-${status}`) + expect(result.body).toContain('spin') + expect(result.body).toContain('is-error') + expect(text(result.body)).toContain('Failure details') + }) + + test('escapes caller titles, messages, and logs', () => { + const unsafe = '' + const result = render(ProgressConsole, { + props: { + status: 'failed', + statusText: { title: unsafe, description: unsafe }, + errorMessage: unsafe, + phases: [{ ...phases[0], message: unsafe }], + logs: { render: [{ ...logs.render[0], message: unsafe }] } + } + }) + expect(result.body).not.toContain(' -
- - {#if status === 'completed'} - - {:else if status === 'running'} - - {:else if status === 'enqueued'} - - {:else if status === 'failed' || status === 'expired'} - - {:else} - - {/if} - - -
- {#if status === 'expired'} -
Preview expired
-
- Assets are purged after 2 days. Re-run the preview from the editor. -
- {:else if status === 'enqueued'} -
Enqueued
-
Waiting for worker…
- {:else if status === 'running'} -
Rendering preview
-
{formatDuration(overallDuration)} elapsed
- {:else if status === 'failed'} -
Preview failed
-
{formatDuration(overallDuration)} stopped at
- {:else if status === 'completed'} -
Preview completed
-
{formatDuration(overallDuration)} total time
- {/if} -
-
- -{#if errorMessage && status === 'failed'} -
- - {errorMessage} -
-{/if} - -{#if phases.length > 0} -
- {#each phases as phase (phase.phase)} - {@const hasLogs = (logs[phase.phase]?.length ?? 0) > 0} -
-
- - {#if phase.status === 'completed'} - - {:else if phase.status === 'running'} - - {:else if phase.status === 'failed'} - - {:else if phase.status === 'skipped'} - - {:else} - - {/if} - - {phase.phase} - {#if phase.message} - - {phase.message} - - {/if} - - {#if phase.started_at} - - {formatDuration((phase.finished_at || Date.now()) - phase.started_at)} - - {/if} - -
- - {#if hasLogs} -
-
- {logs[phase.phase]?.length || 0} lines -
- {#if phase.started_at} - - {formatDuration((phase.finished_at || Date.now()) - phase.started_at)} - - {/if} - -
-
-
handleScroll(phase.phase, e)} - > - {#each logs[phase.phase] ?? [] as log (log.line_number)} -
- - {new Date(log.ts).toISOString().split('T')[1].slice(0, -5)} - - {log.message} -
- {/each} -
-
- {/if} -
- {/each} -
-{/if} - - + diff --git a/apps/web/vitest.config.ts b/apps/web/vitest.config.ts index ccd969f1..6cac2936 100644 --- a/apps/web/vitest.config.ts +++ b/apps/web/vitest.config.ts @@ -1,6 +1,11 @@ import { defineConfig } from 'vitest/config' +import { svelte } from '@sveltejs/vite-plugin-svelte' +import Icons from 'unplugin-icons/vite' +import { fileURLToPath } from 'node:url' export default defineConfig({ + plugins: [svelte({ configFile: false, hot: false }), Icons({ compiler: 'svelte' })], + resolve: { alias: { $lib: fileURLToPath(new URL('./src/lib', import.meta.url)) } }, test: { include: ['src/**/*.{test,spec}.{ts,js}', 'scripts/**/*.{test,spec}.{ts,js}'], exclude: ['build/**', '.svelte-kit/**', 'node_modules/**'] -- 2.54.0 From 459269c055edd15039234012cb9a25557c720492 Mon Sep 17 00:00:00 2001 From: eleith Date: Thu, 1 Oct 2026 17:31:48 -0700 Subject: [PATCH 2/2] [web] stream generation progress and validated results Handle create and edit requests with explicit form readers and server-owned schemas. Share SSE transport with publish and preview while keeping their producers separate. Keep accepted generation running through cleanup after disconnect, and keep budgets in code rather than user configuration. --- apps/core/src/llm/google/client.ts | 72 +--- apps/core/src/llm/google/constants.ts | 12 + apps/core/src/llm/google/generation.test.ts | 142 ++++--- apps/core/src/llm/google/index.ts | 5 +- apps/web/src/hooks.server.test.ts | 62 +++ apps/web/src/lib/generation/types.ts | 17 + apps/web/src/lib/server/constants.ts | 6 + apps/web/src/lib/server/generation.test.ts | 387 ++++++++++++++++++ apps/web/src/lib/server/generation.ts | 36 ++ .../src/lib/server/generation/generator.ts | 103 +++++ .../src/lib/server/generation/input.test.ts | 194 +++++++++ apps/web/src/lib/server/generation/input.ts | 84 ++++ .../src/lib/server/generation/stream.test.ts | 372 +++++++++++++++++ apps/web/src/lib/server/generation/stream.ts | 75 ++++ apps/web/src/lib/server/sse.test.ts | 239 ++++++++++- apps/web/src/lib/server/sse.ts | 57 +++ .../src/lib/server/workflow/stream.test.ts | 230 ++++++++++- apps/web/src/lib/server/workflow/stream.ts | 73 ++-- .../src/routes/api/admin/generate/+server.ts | 4 + 19 files changed, 1988 insertions(+), 182 deletions(-) create mode 100644 apps/core/src/llm/google/constants.ts create mode 100644 apps/web/src/hooks.server.test.ts create mode 100644 apps/web/src/lib/generation/types.ts create mode 100644 apps/web/src/lib/server/constants.ts create mode 100644 apps/web/src/lib/server/generation.test.ts create mode 100644 apps/web/src/lib/server/generation.ts create mode 100644 apps/web/src/lib/server/generation/generator.ts create mode 100644 apps/web/src/lib/server/generation/input.test.ts create mode 100644 apps/web/src/lib/server/generation/input.ts create mode 100644 apps/web/src/lib/server/generation/stream.test.ts create mode 100644 apps/web/src/lib/server/generation/stream.ts create mode 100644 apps/web/src/routes/api/admin/generate/+server.ts diff --git a/apps/core/src/llm/google/client.ts b/apps/core/src/llm/google/client.ts index f5b10385..6d15e822 100644 --- a/apps/core/src/llm/google/client.ts +++ b/apps/core/src/llm/google/client.ts @@ -10,41 +10,22 @@ import { } from '@google/genai' import type { File as GeminiFile, GenerateContentResponse, Part } from '@google/genai' import type { PieceFrontmatter, PieceFrontmatterSchema } from '../../pieces/utils/frontmatter.js' - -const MODEL_NAME = 'gemini-3.8-flash' -const REQUEST_TIMEOUT_MS = 5 * 60 * 1000 -const FILE_POLL_MS = 5000 -const CLEANUP_TIMEOUT_MS = 10_000 +import { + MODEL_NAME, + REQUEST_TIMEOUT_MS, + FILE_POLL_MS, + CLEANUP_TIMEOUT_MS, + DEFAULT_GENERATION_LIMITS, +} from './constants.js' export type GenerationProgress = { phase: 'preparation' | 'generation' | 'validation' message: string } -export type GenerationLimits = { - maxFiles: number - maxFileBytes: number - maxTotalFileBytes: number - /** Maximum accepted result text; checked after the SDK receives the response. */ - maxOutputBytes: number - /** Stop starting new status checks after this wait; active SDK requests may finish. */ - maxFileProcessingMs: number -} - -/** Core safety budgets, not measured web deployment limits. */ -export const DEFAULT_GENERATION_LIMITS: Readonly = { - maxFiles: 10, - maxFileBytes: 50_000_000, - maxTotalFileBytes: 100_000_000, - maxOutputBytes: 1_000_000, - maxFileProcessingMs: 5 * 60 * 1000, -} - export type GenerationOptions = { files?: Array - limits?: Partial onProgress?: (progress: GenerationProgress) => void | Promise - onWarning?: (message: string) => void | Promise } type GenerationTask = { @@ -56,7 +37,6 @@ type GenerationTask = { type GenerationContext = { client: GoogleGenAI - limits: GenerationLimits options: GenerationOptions uploaded: Set } @@ -76,16 +56,6 @@ export async function validateApiKey( } } -function generationLimits(options: GenerationOptions, fileCount: number): GenerationLimits { - const limits = { ...DEFAULT_GENERATION_LIMITS, ...options.limits } - for (const [key, value] of Object.entries(limits)) { - if (!Number.isSafeInteger(value) || value <= 0) - throw new Error(`Invalid generation limit: ${key}.`) - } - if (fileCount > limits.maxFiles) throw new Error('Too many generation attachments.') - return limits -} - function progress(context: GenerationContext, phase: GenerationProgress['phase'], message: string) { // Losing a progress subscriber must not interrupt accepted work. try { @@ -132,7 +102,7 @@ async function uploadFile( } async function waitForFile(name: string, context: GenerationContext): Promise { - const stopPollingAt = Date.now() + context.limits.maxFileProcessingMs + const stopPollingAt = Date.now() + DEFAULT_GENERATION_LIMITS.maxFileProcessingMs while (Date.now() < stopPollingAt) { const file = await context.client.files.get({ name }) if (file.state === 'ACTIVE' && file.uri && file.mimeType) @@ -147,7 +117,7 @@ async function waitForFile(name: string, context: GenerationContext): Promise) { - const { limits } = context + const limits = DEFAULT_GENERATION_LIMITS const parts: Array = [] let totalBytes = 0 for (const [index, file] of files.entries()) { @@ -173,16 +143,6 @@ async function prepareAttachments(context: GenerationContext, files: Array console.warn(message)) - } catch { - console.warn(message) - } -} - async function cleanupFiles(context: GenerationContext) { await Promise.all( [...context.uploaded].map(async (name) => { @@ -193,13 +153,15 @@ async function cleanupFiles(context: GenerationContext) { }) } catch { // Provider errors may include credentials or user content; don't expose them. - warnAboutCleanup(context.options) + console.warn( + 'Could not delete a temporary generation file; provider expiry still applies.' + ) } }) ) } -function completedText(response: GenerateContentResponse, maxOutputBytes: number): string { +function completedText(response: GenerateContentResponse): string { if ( response.promptFeedback?.blockReason && response.promptFeedback.blockReason !== 'BLOCKED_REASON_UNSPECIFIED' @@ -225,7 +187,7 @@ function completedText(response: GenerateContentResponse, maxOutputBytes: number text += part.text } if (!text.trim()) throw new Error('Generation returned empty output.') - if (Buffer.byteLength(text, 'utf8') > maxOutputBytes) + if (Buffer.byteLength(text, 'utf8') > DEFAULT_GENERATION_LIMITS.maxOutputBytes) throw new Error('Generated output exceeds the byte limit.') return text } @@ -257,7 +219,7 @@ async function generateText( ], }, }) - return completedText(response, context.limits.maxOutputBytes) + return completedText(response) } export async function runGeneration( @@ -266,10 +228,10 @@ export async function runGeneration( options: GenerationOptions ): Promise { const files = [...(options.files ?? [])] - const limits = generationLimits(options, files.length) + if (files.length > DEFAULT_GENERATION_LIMITS.maxFiles) + throw new Error('Too many generation attachments.') const context: GenerationContext = { client: getClient(apiKey), - limits, options, uploaded: new Set(), } diff --git a/apps/core/src/llm/google/constants.ts b/apps/core/src/llm/google/constants.ts new file mode 100644 index 00000000..1e20b385 --- /dev/null +++ b/apps/core/src/llm/google/constants.ts @@ -0,0 +1,12 @@ +export const MODEL_NAME = 'gemini-3.8-flash' +export const REQUEST_TIMEOUT_MS = 5 * 60 * 1000 +export const FILE_POLL_MS = 5000 +export const CLEANUP_TIMEOUT_MS = 10_000 + +export const DEFAULT_GENERATION_LIMITS = { + maxFiles: 10, + maxFileBytes: 50_000_000, + maxTotalFileBytes: 100_000_000, + maxOutputBytes: 1_000_000, + maxFileProcessingMs: 5 * 60 * 1000, +} as const diff --git a/apps/core/src/llm/google/generation.test.ts b/apps/core/src/llm/google/generation.test.ts index 58838075..98556938 100644 --- a/apps/core/src/llm/google/generation.test.ts +++ b/apps/core/src/llm/google/generation.test.ts @@ -80,6 +80,8 @@ async function until(condition: () => boolean) { } throw new Error('Expected asynchronous operation did not start') } +// Small byte budgets exercise boundaries without allocating deployment-sized fixtures. +const originalLimits = { ...DEFAULT_GENERATION_LIMITS } const run = (options: GenerationOptions = {}) => generatePieceMetadata('test-key', schema(), 'My instructions', options) @@ -98,6 +100,7 @@ beforeEach(() => { mocks.list.mockResolvedValue({} as never) }) afterEach(() => { + Object.assign(DEFAULT_GENERATION_LIMITS, originalLimits) vi.useRealTimers() vi.restoreAllMocks() }) @@ -545,82 +548,62 @@ describe('input preparation and cleanup', () => { expect(mocks.delete).toHaveBeenCalledTimes(1) }) - test('reports cleanup failures without exposing provider errors or changing the valid result', async () => { + test('logs safe cleanup warnings without changing the valid result', async () => { binary() mocks.delete.mockRejectedValue(new Error('API key secret; private prompt')) - const onWarning = vi.fn() - await expect(run({ files: [Buffer.from('one')], onWarning })).resolves.toEqual({ + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + await expect(run({ files: [Buffer.from('one')] })).resolves.toEqual({ title: 'generated', }) - expect(onWarning).toHaveBeenCalledExactlyOnceWith( - 'Could not delete a temporary generation file; provider expiry still applies.' - ) - }) - - test('uses safe console warnings by default and if the warning callback throws', async () => { - binary() - mocks.delete.mockRejectedValue(new Error('secret')) - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) - await run({ files: [Buffer.from('one')] }) - await run({ - files: [Buffer.from('two')], - onWarning: () => { - throw new Error('observer failed') - }, - }) - expect(warn).toHaveBeenCalledTimes(2) - expect(JSON.stringify(warn.mock.calls)).not.toContain('secret') - }) - test('falls back to a safe warning when an asynchronous warning callback rejects', async () => { - binary() - mocks.delete.mockRejectedValue(new Error('private provider error')) - const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) - const onWarning = vi.fn().mockRejectedValue(new Error('private observer error')) - await expect(run({ files: [Buffer.from('one')], onWarning })).resolves.toEqual({ - title: 'generated', - }) - expect(onWarning).toHaveBeenCalledOnce() expect(warn).toHaveBeenCalledExactlyOnceWith( 'Could not delete a temporary generation file; provider expiry still applies.' ) }) - test('does not wait for warning callbacks or skip cleanup of other known files', async () => { + test.each(['provider', 'validation'])( + 'logs safe cleanup warnings without masking a %s error', + async (failure) => { + binary() + const providerError = new Error('provider unavailable') + if (failure === 'provider') mocks.generate.mockRejectedValue(providerError) + else respond('{"title":42}') + mocks.delete.mockRejectedValue(new Error('API key secret; private prompt')) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + const generating = run({ files: [Buffer.from('one')] }) + if (failure === 'provider') await expect(generating).rejects.toBe(providerError) + else await expect(generating).rejects.toThrow('does not match schema') + expect(warn).toHaveBeenCalledExactlyOnceWith( + 'Could not delete a temporary generation file; provider expiry still applies.' + ) + } + ) + + test('does not skip cleanup of other known files after a deletion fails', async () => { binary() mocks.upload .mockResolvedValueOnce({ name: 'files/first' }) .mockResolvedValueOnce({ name: 'files/second' }) mocks.delete.mockRejectedValueOnce(new Error('cleanup failed')) - const onWarning = vi.fn(() => new Promise(() => {})) - await expect( - run({ files: [Buffer.from('one'), Buffer.from('two')], onWarning }) - ).resolves.toEqual({ title: 'generated' }) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) + await expect(run({ files: [Buffer.from('one'), Buffer.from('two')] })).resolves.toEqual({ + title: 'generated', + }) expect(mocks.delete.mock.calls.map(([arg]) => arg.name)).toEqual([ 'files/first', 'files/second', ]) - expect(onWarning).toHaveBeenCalledOnce() + expect(warn).toHaveBeenCalledExactlyOnceWith( + 'Could not delete a temporary generation file; provider expiry still applies.' + ) }) }) describe('input and output bounds', () => { - test.each([0, -1, NaN, Infinity, 1.5, Number.MAX_SAFE_INTEGER + 1])( - 'rejects invalid limits before provider work: %s', - async (maxFiles) => { - await expect(run({ limits: { maxFiles } })).rejects.toThrow('Invalid generation limit') - await expect(run({ limits: { maxFileProcessingMs: maxFiles } })).rejects.toThrow( - 'Invalid generation limit' - ) - expect(GoogleGenAI).not.toHaveBeenCalled() - } - ) - - test('rejects excess file counts and retains the default processing wait limit', async () => { - await expect( - run({ files: [Buffer.from('1'), Buffer.from('2')], limits: { maxFiles: 1 } }) - ).rejects.toThrow('Too many') + test('accepts the file count boundary and rejects excess files before provider work', async () => { + const files = Array.from({ length: DEFAULT_GENERATION_LIMITS.maxFiles }, () => Buffer.from('1')) + await expect(run({ files: [...files, Buffer.from('2')] })).rejects.toThrow('Too many') expect(GoogleGenAI).not.toHaveBeenCalled() - expect(DEFAULT_GENERATION_LIMITS.maxFileProcessingMs).toBe(300000) + await expect(run({ files })).resolves.toEqual({ title: 'generated' }) }) test('rejects invalid schema or unknown field before provider work', async () => { @@ -634,22 +617,34 @@ describe('input and output bounds', () => { }) test('bounds buffers and combined input bytes before starting further uploads', async () => { - await expect(run({ files: [Buffer.from('123')], limits: { maxFileBytes: 2 } })).rejects.toThrow( + Object.assign(DEFAULT_GENERATION_LIMITS, { maxFileBytes: 2, maxTotalFileBytes: 3 }) + await expect(run({ files: [Buffer.from('123')] })).rejects.toThrow('byte limit') + await expect(run({ files: [Buffer.from('12'), Buffer.from('34')] })).rejects.toThrow( 'byte limit' ) - await expect( - run({ files: [Buffer.from('12'), Buffer.from('34')], limits: { maxTotalFileBytes: 3 } }) - ).rejects.toThrow('byte limit') expect(mocks.upload).not.toHaveBeenCalled() expect(mocks.generate).not.toHaveBeenCalled() + await expect(run({ files: [Buffer.from('12'), Buffer.from('3')] })).resolves.toEqual({ + title: 'generated', + }) + }) + + test('cleans earlier uploads when a later attachment exceeds the total byte budget', async () => { + Object.assign(DEFAULT_GENERATION_LIMITS, { maxFileBytes: 2, maxTotalFileBytes: 3 }) + binary() + await expect(run({ files: [Buffer.from('12'), Buffer.from('34')] })).rejects.toThrow( + 'byte limit' + ) + expect(mocks.upload).toHaveBeenCalledOnce() + expect(mocks.delete).toHaveBeenCalledOnce() + expect(mocks.generate).not.toHaveBeenCalled() }) test('stops reading an oversized path rather than buffering the entire file', async () => { + Object.assign(DEFAULT_GENERATION_LIMITS, { maxFileBytes: 3 }) const stream = Readable.from([Buffer.from('12'), Buffer.from('34'), Buffer.from('56')]) mocks.readStream.mockReturnValue(stream as ReadStream) - await expect(run({ files: ['/file'], limits: { maxFileBytes: 3 } })).rejects.toThrow( - 'byte limit' - ) + await expect(run({ files: ['/file'] })).rejects.toThrow('byte limit') expect(stream.destroyed).toBe(true) expect(mocks.fileType).not.toHaveBeenCalled() }) @@ -658,8 +653,10 @@ describe('input and output bounds', () => { const text = '{"title":"🧩"}' respond(text) const bytes = Buffer.byteLength(text, 'utf8') - await expect(run({ limits: { maxOutputBytes: bytes } })).resolves.toEqual({ title: '🧩' }) - await expect(run({ limits: { maxOutputBytes: bytes - 1 } })).rejects.toThrow('output exceeds') + Object.assign(DEFAULT_GENERATION_LIMITS, { maxOutputBytes: bytes }) + await expect(run()).resolves.toEqual({ title: '🧩' }) + Object.assign(DEFAULT_GENERATION_LIMITS, { maxOutputBytes: bytes - 1 }) + await expect(run()).rejects.toThrow('output exceeds') }) }) @@ -668,6 +665,8 @@ describe('disconnected progress observers', () => { 'finishes validation and cleanup when the progress sink disconnects (%s)', async (failure) => { binary() + mocks.delete.mockRejectedValue(new Error('API key secret; private prompt')) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) const pending = deferred() mocks.generate.mockReturnValue(pending.promise) let disconnected = false @@ -684,6 +683,9 @@ describe('disconnected progress observers', () => { await expect(generating).resolves.toEqual({ title: 'finished' }) expect(onProgress.mock.calls.map(([event]) => event.phase)).toContain('validation') expect(mocks.delete).toHaveBeenCalledOnce() + expect(warn).toHaveBeenCalledExactlyOnceWith( + 'Could not delete a temporary generation file; provider expiry still applies.' + ) } ) @@ -705,6 +707,8 @@ describe('disconnected progress observers', () => { 'preserves a %s error after disconnect and still cleans known files', async (failure) => { binary() + mocks.delete.mockRejectedValue(new Error('API key secret; private prompt')) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}) const pending = deferred() mocks.generate.mockReturnValue(pending.promise) let disconnected = false @@ -722,6 +726,9 @@ describe('disconnected progress observers', () => { else pending.resolve(response('{"title":42}')) await rejected expect(mocks.delete).toHaveBeenCalledOnce() + expect(warn).toHaveBeenCalledExactlyOnceWith( + 'Could not delete a temporary generation file; provider expiry still applies.' + ) } ) }) @@ -731,13 +738,15 @@ describe('bounded file processing polling', () => { { maxFileProcessingMs: 1, polls: 1 }, { maxFileProcessingMs: 5000, polls: 1 }, { maxFileProcessingMs: 6000, polls: 2 }, + { maxFileProcessingMs: DEFAULT_GENERATION_LIMITS.maxFileProcessingMs, polls: 60 }, ])( 'stops polling at the wait limit: $maxFileProcessingMs ms', async ({ maxFileProcessingMs, polls }) => { vi.useFakeTimers() binary() mocks.get.mockResolvedValue({ state: FileState.PROCESSING }) - const generating = run({ files: [Buffer.from('one')], limits: { maxFileProcessingMs } }) + Object.assign(DEFAULT_GENERATION_LIMITS, { maxFileProcessingMs }) + const generating = run({ files: [Buffer.from('one')] }) const rejected = expect(generating).rejects.toThrow( 'Attachment processing exceeded the wait limit.' ) @@ -761,13 +770,10 @@ describe('bounded file processing polling', () => { const pending = deferred>>() mocks.get.mockReturnValue(pending.promise) const settled = vi.fn() - const generating = run({ - files: [Buffer.from('one')], - limits: { maxFileProcessingMs: 100 }, - }) + const generating = run({ files: [Buffer.from('one')] }) void generating.then(settled, settled) await until(() => mocks.get.mock.calls.length === 1) - await vi.advanceTimersByTimeAsync(101) + await vi.advanceTimersByTimeAsync(DEFAULT_GENERATION_LIMITS.maxFileProcessingMs + 1) expect(settled).not.toHaveBeenCalled() pending.resolve({ state, uri: 'https://provider/ready', mimeType: 'application/pdf' }) if (state === FileState.ACTIVE) { diff --git a/apps/core/src/llm/google/index.ts b/apps/core/src/llm/google/index.ts index 7a74307f..4cec6330 100644 --- a/apps/core/src/llm/google/index.ts +++ b/apps/core/src/llm/google/index.ts @@ -1,3 +1,4 @@ export * from './generation.js' -export { DEFAULT_GENERATION_LIMITS, validateApiKey } from './client.js' -export type { GenerationOptions, GenerationLimits, GenerationProgress } from './client.js' +export { DEFAULT_GENERATION_LIMITS } from './constants.js' +export { validateApiKey } from './client.js' +export type { GenerationOptions, GenerationProgress } from './client.js' diff --git a/apps/web/src/hooks.server.test.ts b/apps/web/src/hooks.server.test.ts new file mode 100644 index 00000000..ded310e8 --- /dev/null +++ b/apps/web/src/hooks.server.test.ts @@ -0,0 +1,62 @@ +import { afterEach, expect, test, vi } from 'vitest' +import type { Handle } from '@sveltejs/kit' + +const fixture = vi.hoisted(() => ({ + config: { + auth: { enabled: true, type: 'credentials', secret: 'test-secret' }, + url: { app_assets: '' } + } +})) +vi.mock('$lib/server/config', () => ({ config: fixture.config })) +vi.mock('$lib/server/database', () => ({})) +// Exercise the application guard, not Auth.js or SvelteKit's request-store machinery. +vi.mock('@sveltejs/kit/hooks', () => ({ sequence: (_auth: Handle, guard: Handle) => guard })) +vi.mock('@auth/sveltekit', () => ({ + SvelteKitAuth: () => ({ + handle: (({ event, resolve }) => resolve(event)) satisfies Handle + }) +})) + +afterEach(() => { + fixture.config.auth.enabled = true + vi.resetModules() +}) + +function event(session: object | null) { + return { + url: new URL('http://localhost/api/admin/generate'), + locals: { auth: vi.fn().mockResolvedValue(session) } + } as unknown as Parameters[0]['event'] +} + +test('the generation namespace requires a session before resolving its handler', async () => { + const { handle } = await import('./hooks.server') + const resolve = vi.fn() + await expect(handle({ event: event(null), resolve })).rejects.toMatchObject({ + status: 302, + location: '/signin?redirectTo=/api/admin/generate' + }) + expect(resolve).not.toHaveBeenCalled() +}) + +test('generation remains inaccessible when admin authentication is disabled', async () => { + fixture.config.auth.enabled = false + const { handle } = await import('./hooks.server') + const resolve = vi.fn() + await expect(handle({ event: event({ user: { name: 'test' } }), resolve })).rejects.toMatchObject( + { + status: 302, + location: '/' + } + ) + expect(resolve).not.toHaveBeenCalled() +}) + +test('an authenticated generation request reaches its handler', async () => { + const { handle } = await import('./hooks.server') + const response = new Response('resolved') + const resolve = vi.fn().mockResolvedValue(response) + await expect(handle({ event: event({ user: { name: 'test' } }), resolve })).resolves.toBe( + response + ) +}) diff --git a/apps/web/src/lib/generation/types.ts b/apps/web/src/lib/generation/types.ts new file mode 100644 index 00000000..9eeda16c --- /dev/null +++ b/apps/web/src/lib/generation/types.ts @@ -0,0 +1,17 @@ +import type { ProgressPhase } from '$lib/components/progress' + +export type GenerationTarget = { kind: 'field'; key: string } | { kind: 'body' } + +export type GenerationResult = + | { kind: 'creation'; markdown: string } + | { kind: 'field'; key: string; value: unknown } + | { kind: 'body'; value: string } + +export type GenerationEvent = + | { type: 'state'; data: { state: 'running' } } + | { type: 'phase'; data: ProgressPhase[] } + | { type: 'error'; data: { message: string } } + | { + type: 'done' + data: { state: 'completed'; result: GenerationResult } | { state: 'failed' } + } diff --git a/apps/web/src/lib/server/constants.ts b/apps/web/src/lib/server/constants.ts new file mode 100644 index 00000000..1e812f99 --- /dev/null +++ b/apps/web/src/lib/server/constants.ts @@ -0,0 +1,6 @@ +// Generation and progress-stream budgets, tuned in code rather than exposed as user settings. +export const MAX_SOURCE_CHARS = 1_000_000 +export const MAX_INSTRUCTION_CHARS = 100_000 +export const MAX_CONCURRENT_GENERATIONS = 1 +export const HEARTBEAT_MS = 15_000 +export const WORKFLOW_POLL_INTERVAL_MS = 350 diff --git a/apps/web/src/lib/server/generation.test.ts b/apps/web/src/lib/server/generation.test.ts new file mode 100644 index 00000000..00bcfc94 --- /dev/null +++ b/apps/web/src/lib/server/generation.test.ts @@ -0,0 +1,387 @@ +import { afterEach, beforeEach, expect, test, vi } from 'vitest' +import { + Piece, + extractFullMarkdown, + generateBody, + generateFieldValue, + generatePieceMetadata, + type GenerationOptions, + type PieceFrontmatter, + type PieceFrontmatterSchema +} from '@luzzle/core' +import type { GenerationEvent } from '$lib/generation/types' +import { POST } from '../../routes/api/admin/generate/+server' + +const fixture = vi.hoisted(() => { + const limits = { + maxFiles: 2, + maxFileBytes: 50, + maxTotalFileBytes: 75, + maxOutputBytes: 1000, + maxFileProcessingMs: 300000 + } + const ai = { provider: 'google' as const, api_key: 'server-secret' } + return { + limits, + ai, + config: { + ai: ai as typeof ai | undefined, + pieces: [{ type: 'books', fields: { title: 'title' } }] + }, + storage: { + stat: vi.fn(), + readFile: vi.fn(), + writeFile: vi.fn(), + createFile: vi.fn(), + makeDirectory: vi.fn(), + delete: vi.fn() + } + } +}) + +vi.mock('./config', () => ({ config: fixture.config })) +vi.mock('./storage', () => ({ getStorage: () => fixture.storage })) +vi.mock('./constants', () => ({ + MAX_SOURCE_CHARS: 1000, + MAX_INSTRUCTION_CHARS: 100, + MAX_CONCURRENT_GENERATIONS: 1, + HEARTBEAT_MS: 15000 +})) +vi.mock('@luzzle/core', async (original) => ({ + ...(await original>()), + DEFAULT_GENERATION_LIMITS: fixture.limits, + generatePieceMetadata: vi.fn(), + generateFieldValue: vi.fn(), + generateBody: vi.fn() +})) +vi.mock('$lib/server/database', () => { + throw new Error('Generation must not import the database') +}) +vi.mock('@luzzle/web.jobs', () => { + throw new Error('Generation must not import durable jobs') +}) + +let schema: PieceFrontmatterSchema + +beforeEach(() => { + vi.clearAllMocks() + fixture.config.ai = fixture.ai + fixture.limits.maxOutputBytes = 1000 + fixture.config.pieces = [{ type: 'books', fields: { title: 'title' } }] + schema = { + title: 'books', + type: 'object', + properties: { + title: { type: 'string', minLength: 2 }, + body: { type: 'string' }, + cover: { type: 'string' } + }, + required: ['title'], + additionalProperties: false + } + fixture.storage.readFile.mockImplementation(async (file: string) => { + expect(file).toBe('.luzzle/schemas/books.json') + return JSON.stringify(schema) + }) + fixture.storage.stat.mockImplementation(async (file: string) => { + if (file === '.' || file === 'books') return { type: 'directory' } + if (file === 'books/disk.books.md') return { type: 'file' } + throw new Error('Not in the archive') + }) + vi.mocked(generatePieceMetadata).mockImplementation(async (_key, _schema, _prompt, options) => { + await options?.onProgress?.({ phase: 'generation', message: 'Generating metadata' }) + await options?.onProgress?.({ phase: 'validation', message: 'Validating metadata' }) + return { title: 'Model title', cover: 'https://invalid.example/asset.jpg' } + }) + vi.mocked(generateFieldValue).mockResolvedValue('Generated field') + vi.mocked(generateBody).mockResolvedValue('# Generated body') + vi.spyOn(Piece.prototype, 'setField').mockRejectedValue(new Error('No asset downloads')) + vi.spyOn(console, 'error').mockImplementation(() => {}) +}) + +afterEach(() => { + fixture.limits.maxOutputBytes = 1000 + for (const method of ['writeFile', 'createFile', 'makeDirectory', 'delete'] as const) { + expect(fixture.storage[method]).not.toHaveBeenCalled() + } + expect(Piece.prototype.setField).not.toHaveBeenCalled() + vi.restoreAllMocks() +}) + +function form(mode: 'create' | 'edit' = 'create') { + const data = new FormData() + data.set('mode', mode) + data.set('instructions', 'Find accurate details') + if (mode === 'create') { + data.set('name', 'Entered title') + data.set('type', 'books') + data.set('directory', 'books') + } else { + data.set('file', 'books/disk.books.md') + data.set('source', JSON.stringify('---\ntitle: Unsaved\n---\n\nActual unsaved body\n')) + data.set('target', 'body') + } + return data +} + +function request(data = form(), signal?: AbortSignal) { + return new Request('http://localhost/api/admin/generate', { method: 'POST', body: data, signal }) +} + +function post(req = request()) { + return POST({ request: req } as Parameters[0]) +} + +async function events(response: Response): Promise { + return (await response.text()) + .trim() + .split('\n\n') + .filter((frame) => frame.startsWith('event:')) + .map((frame) => { + const [name, data] = frame.split('\n') + return { type: name.slice(7), data: JSON.parse(data.slice(6)) } as GenerationEvent + }) +} + +function noProviderCalls() { + expect(generatePieceMetadata).not.toHaveBeenCalled() + expect(generateFieldValue).not.toHaveBeenCalled() + expect(generateBody).not.toHaveBeenCalled() +} + +test('creation returns validated metadata with the entered title and an empty body, without writes', async () => { + const data = form() + data.append('files', new File(['attachment'], 'source.txt')) + data.set('api_key', 'untrusted-browser-key') + data.set('schema', JSON.stringify({ type: 'object' })) + const response = await post(request(data)) + expect(response.headers.get('content-type')).toBe('text/event-stream') + const frames = await events(response) + const done = frames.at(-1) + expect(done).toMatchObject({ type: 'done', data: { state: 'completed' } }) + if (done?.type !== 'done' || done.data.state !== 'completed') throw new Error('Missing result') + const result = done.data.result + if (result.kind !== 'creation') throw new Error('Wrong result') + expect(await extractFullMarkdown(result.markdown)).toMatchObject({ + frontmatter: { title: 'Entered title', cover: 'https://invalid.example/asset.jpg' }, + markdown: '' + }) + expect(generatePieceMetadata).toHaveBeenCalledWith( + 'server-secret', + schema, + expect.stringContaining('Entered title: "Entered title"'), + { + files: [Buffer.from('attachment')], + onProgress: expect.any(Function) + } + ) + expect(JSON.stringify(frames)).not.toContain('server-secret') + // No examples/defaults are needed: creation does not initialize a placeholder piece. + expect(schema.properties.title).not.toHaveProperty('examples') +}) + +test('body generation receives the raw unsaved source, never disk content', async () => { + const data = form('edit') + const frames = await events(await post(request(data))) + expect(generateBody).toHaveBeenCalledWith( + 'server-secret', + { source: JSON.parse(data.get('source') as string), instructions: data.get('instructions') }, + { files: [], onProgress: expect.any(Function) } + ) + expect(frames.at(-1)).toEqual({ + type: 'done', + data: { state: 'completed', result: { kind: 'body', value: '# Generated body' } } + }) + expect(fixture.storage.readFile).toHaveBeenCalledTimes(1) +}) + +test('a metadata field named body is not the Markdown body target', async () => { + const data = form('edit') + data.set('target', 'field') + data.set('key', 'body') + const frames = await events(await post(request(data))) + expect(generateBody).not.toHaveBeenCalled() + expect(generateFieldValue).toHaveBeenCalledWith( + 'server-secret', + { + key: 'body', + schema, + source: JSON.parse(data.get('source') as string), + instructions: data.get('instructions') + }, + { files: [], onProgress: expect.any(Function) } + ) + expect(frames.at(-1)).toMatchObject({ + type: 'done', + data: { result: { kind: 'field', key: 'body', value: 'Generated field' } } + }) +}) + +test.each([null, '', 'delete', new File(['create'], 'mode.txt')])( + 'rejects invalid mode %s before choosing a generator and releases capacity', + async (mode) => { + const data = form() + if (mode === null) data.delete('mode') + else data.set('mode', mode) + await expect(post(request(data))).rejects.toMatchObject({ status: 400 }) + expect(fixture.storage.readFile).not.toHaveBeenCalled() + expect(fixture.storage.stat).not.toHaveBeenCalled() + noProviderCalls() + expect((await events(await post())).at(-1)).toMatchObject({ data: { state: 'completed' } }) + } +) + +test.each(['unknown', 'toString', '__proto__', 'title.nested'])( + 'rejects field %s before provider work', + async (key) => { + const data = form('edit') + data.set('target', 'field') + data.set('key', key) + await expect(post(request(data))).rejects.toMatchObject({ status: 400 }) + noProviderCalls() + } +) + +test.each([ + 'missing.books.md', + '../outside.books.md', + 'books/disk.unknown.md', + 'assets/disk.books.md', + 'file.books.txt' +])('rejects invalid existing identity %s', async (file) => { + const data = form('edit') + data.set('file', file) + await expect(post(request(data))).rejects.toMatchObject({ status: 400 }) + noProviderCalls() +}) + +test.each([ + ['type', 'unknown'], + ['directory', '../outside'], + ['directory', 'books/disk.books.md'] +])('rejects bad creation %s', async (field, value) => { + const data = form() + data.set(field, value) + await expect(post(request(data))).rejects.toMatchObject({ status: 400 }) + noProviderCalls() +}) + +test('rejects a missing schema and releases capacity', async () => { + fixture.storage.readFile.mockRejectedValueOnce(new Error('missing schema')) + await expect(post()).rejects.toMatchObject({ status: 400 }) + noProviderCalls() + expect((await events(await post())).at(-1)).toMatchObject({ data: { state: 'completed' } }) +}) + +test.each(['missing', '__proto__.title'])( + 'rejects unsupported title configuration %s', + async (title) => { + fixture.config.pieces[0].fields.title = title + await expect(post()).rejects.toMatchObject({ status: 400 }) + noProviderCalls() + } +) + +test('supports a configured nested title without asset-setting operations', async () => { + schema = { + title: 'books', + type: 'object', + properties: { + details: { type: 'object', properties: { name: { type: 'string' } }, required: ['name'] } + }, + required: ['details'] + } + fixture.config.pieces[0].fields.title = 'details.name' + vi.mocked(generatePieceMetadata).mockResolvedValue({ details: { name: 'Model name' } }) + const frames = await events(await post()) + expect(JSON.stringify(frames.at(-1))).toContain('Entered title') + expect(JSON.stringify(frames.at(-1))).not.toContain('Model name') +}) + +test('revalidates the entered title against the completed metadata', async () => { + const data = form() + data.set('name', 'X') // Too short for the schema. + const frames = await events(await post(request(data))) + expect(frames.at(-1)).toEqual({ type: 'done', data: { state: 'failed' } }) + expect(frames.some((frame) => frame.type === 'error')).toBe(true) + expect(JSON.stringify(frames)).not.toContain('cover') +}) + +test('caps the final serialized result as well as the core completion', async () => { + fixture.limits.maxOutputBytes = 10 + const frames = await events(await post()) + expect(frames.at(-1)).toEqual({ type: 'done', data: { state: 'failed' } }) +}) + +test('invalid core completions are terminal failures, never partial results', async () => { + vi.mocked(generateFieldValue).mockImplementation(async (_key, _request, options) => { + await options?.onProgress?.({ phase: 'validation', message: 'Validating generated field' }) + throw new Error('Generated field is invalid; secret provider request details') + }) + const data = form('edit') + data.set('target', 'field') + data.set('key', 'title') + const frames = await events(await post(request(data))) + expect(frames.filter((event) => event.type === 'done')).toEqual([ + { type: 'done', data: { state: 'failed' } } + ]) + expect(JSON.stringify(frames)).not.toContain('secret provider request details') +}) + +test('unconfigured AI rejects without reading the body', async () => { + fixture.config.ai = undefined + const req = request() + await expect(post(req)).rejects.toMatchObject({ status: 503 }) + expect(req.bodyUsed).toBe(false) + noProviderCalls() +}) + +test('reserves capacity before multipart parsing and releases it after malformed input', async () => { + let stream!: ReadableStreamDefaultController + const req = new Request('http://localhost/api/admin/generate', { + method: 'POST', + headers: { 'Content-Type': 'multipart/form-data; boundary=test' }, + body: new ReadableStream({ + start(value) { + stream = value + } + }), + duplex: 'half' + } as RequestInit) + const pending = post(req) + const rejected = request() + await expect(post(rejected)).rejects.toMatchObject({ status: 429 }) + expect(rejected.bodyUsed).toBe(false) + stream.error(new Error('Incomplete upload')) + await expect(pending).rejects.toThrow('Incomplete upload') + noProviderCalls() + expect((await events(await post())).at(-1)).toMatchObject({ data: { state: 'completed' } }) +}) + +test.each(['cancel', 'abort', 'detached failure'] as const)( + 'keeps the slot through provider cleanup after %s', + async (disconnect) => { + let settle!: (value: string) => void + let fail!: (cause: Error) => void + let received: GenerationOptions | undefined + vi.mocked(generateBody).mockImplementationOnce(async (_key, _context, options) => { + received = options + return new Promise((resolve, reject) => { + settle = resolve + fail = reject + }) + }) + const abort = new AbortController() + const response = await post(request(form('edit'), abort.signal)) + if (disconnect === 'abort') abort.abort() + else await response.body!.cancel() + await received?.onProgress?.({ phase: 'generation', message: 'Still working' }) + await expect(post()).rejects.toMatchObject({ status: 429 }) + // Settling this fake core promise represents generation AND provider-file cleanup finishing. + if (disconnect === 'detached failure') fail(new Error('provider failed after disconnect')) + else settle('Abandoned result') + await vi.waitFor(async () => { + expect((await events(await post())).at(-1)).toMatchObject({ data: { state: 'completed' } }) + }) + } +) diff --git a/apps/web/src/lib/server/generation.ts b/apps/web/src/lib/server/generation.ts new file mode 100644 index 00000000..7c3de0c2 --- /dev/null +++ b/apps/web/src/lib/server/generation.ts @@ -0,0 +1,36 @@ +import { error } from '@sveltejs/kit' +import { config } from './config' +import { readCreateInput, readEditInput } from './generation/input' +import { createDraftGenerator, createEditorGenerator, type Generate } from './generation/generator' +import { generationStream } from './generation/stream' +import { MAX_CONCURRENT_GENERATIONS } from './constants' + +let active = 0 + +export async function generateResponse(request: Request): Promise { + const ai = config.ai + if (!ai) error(503, 'AI is not configured.') + if (active >= MAX_CONCURRENT_GENERATIONS) error(429, 'Generation is busy. Please try again.') + + // Reserve before consuming multipart data; disconnected accepted work still owns its slot. + active++ + const release = () => active-- + try { + const form = await request.formData() + const mode = form.get('mode') + let generate: Generate + if (mode === 'create') { + const input = await readCreateInput(form) + generate = await createDraftGenerator(input, ai.api_key) + } else if (mode === 'edit') { + const input = await readEditInput(form) + generate = await createEditorGenerator(input, ai.api_key) + } else { + error(400, 'Choose creation or editing.') + } + return generationStream(request, generate, release) + } catch (cause) { + release() + throw cause + } +} diff --git a/apps/web/src/lib/server/generation/generator.ts b/apps/web/src/lib/server/generation/generator.ts new file mode 100644 index 00000000..f70a854c --- /dev/null +++ b/apps/web/src/lib/server/generation/generator.ts @@ -0,0 +1,103 @@ +import { + findFrontmatterField, + generateBody, + generateFieldValue, + generatePieceMetadata, + makePieceMarkdown, + makePieceMarkdownString, + setFrontmatterValue, + type GenerationProgress +} from '@luzzle/core' +import { error } from '@sveltejs/kit' +import type { GenerationResult } from '$lib/generation/types' +import { config } from '../config' +import { getPieces } from '../pieces' +import { getStorage } from '../storage' +import type { CreateGenerationInput, EditGenerationInput } from './input' + +export type Generate = (progress: (event: GenerationProgress) => void) => Promise + +export async function createDraftGenerator( + input: CreateGenerationInput, + apiKey: string +): Promise { + const piece = await getConfiguredPiece(input.type) + await archiveEntry(input.directory || '.', 'directory') + const title = config.pieces.find((entry) => entry.type === input.type)?.fields.title + if ( + !title || + title.split('.').some((part) => ['__proto__', 'constructor', 'prototype'].includes(part)) || + findFrontmatterField(piece.fields, title)?.type !== 'string' + ) { + error(400, 'This piece type needs a string title field.') + } + return async (progress) => { + const frontmatter = await generatePieceMetadata( + apiKey, + piece.schema, + `${input.instructions}\n\nEntered title: ${JSON.stringify(input.name)}`, + { files: input.files, onProgress: progress } + ) + // Do not use setField(): generation must not download assets or persist anything. + setFrontmatterValue(frontmatter, title, input.name) + // The destination is derived afresh on explicit Create, not reserved during generation. + const draft = makePieceMarkdown('', input.type, '', frontmatter) + if (!piece.validate(draft).isValid) throw new Error('Invalid generated draft.') + return { kind: 'creation', markdown: makePieceMarkdownString(draft) } + } +} + +export async function createEditorGenerator( + input: EditGenerationInput, + apiKey: string +): Promise { + const pieces = getPieces() + const file = pieces.parseFilename(input.file) + if (file.format !== '.md' || pieces.isAsset(input.file)) { + error(400, 'Choose an archive piece.') + } + const piece = await getConfiguredPiece(file.type) + await archiveEntry(input.file, 'file') + const target = input.target + if ( + target.kind === 'field' && + (target.key === '__proto__' || !Object.hasOwn(piece.schema.properties, target.key)) + ) { + error(400, 'Choose a top-level metadata field.') + } + return async (progress) => { + const context = { source: input.source, instructions: input.instructions } + const generationOptions = { files: input.files, onProgress: progress } + if (target.kind === 'body') { + return { kind: 'body', value: await generateBody(apiKey, context, generationOptions) } + } + return { + kind: 'field', + key: target.key, + value: await generateFieldValue( + apiKey, + { ...context, schema: piece.schema, key: target.key }, + generationOptions + ) + } + } +} + +async function archiveEntry(path: string, type: 'file' | 'directory') { + try { + if ((await getStorage().stat(path)).type !== type) throw new Error('Wrong entry type') + } catch { + error(400, `Choose an existing archive ${type}.`) + } +} + +async function getConfiguredPiece(type: string | null) { + if (!type || !config.pieces.some((entry) => entry.type === type)) { + error(400, 'Choose a configured piece type.') + } + try { + return await getPieces().getPiece(type) + } catch { + error(400, 'Could not prepare the piece schema.') + } +} diff --git a/apps/web/src/lib/server/generation/input.test.ts b/apps/web/src/lib/server/generation/input.test.ts new file mode 100644 index 00000000..d61ce1a3 --- /dev/null +++ b/apps/web/src/lib/server/generation/input.test.ts @@ -0,0 +1,194 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { readCreateInput, readEditInput } from './input' +import { MAX_SOURCE_CHARS, MAX_INSTRUCTION_CHARS } from '../constants' + +vi.mock('@luzzle/core', () => ({ + DEFAULT_GENERATION_LIMITS: { maxFiles: 2, maxFileBytes: 8, maxTotalFileBytes: 12 } +})) +vi.mock('../constants', () => ({ MAX_SOURCE_CHARS: 128, MAX_INSTRUCTION_CHARS: 32 })) + +function createForm(): FormData { + const form = new FormData() + form.set('name', 'Dune') + form.set('type', 'book') + return form +} + +function editForm(): FormData { + const form = new FormData() + form.set('file', 'library/dune.book.md') + form.set('source', JSON.stringify('')) + form.set('target', 'body') + return form +} + +function attach(form: FormData, size: number, name = 'reference.bin') { + form.append('files', new File([new Uint8Array(size).fill(255)], name)) +} + +afterEach(() => vi.restoreAllMocks()) + +describe('readCreateInput', () => { + it('accepts create input with optional fields omitted', async () => { + await expect(readCreateInput(createForm())).resolves.toEqual({ + name: 'Dune', + type: 'book', + directory: '', + instructions: '', + files: [] + }) + }) + + it('preserves create text, returns binary Buffers and ignores unrelated fields', async () => { + const form = createForm() + form.set('directory', 'library/science fiction') + form.set('instructions', ' Keep the original tone. ') + form.set('unrelated', 'ignored') + attach(form, 8) + attach(form, 4) + const result = await readCreateInput(form) + expect(result).toEqual({ + name: 'Dune', + type: 'book', + directory: 'library/science fiction', + instructions: ' Keep the original tone. ', + files: [Buffer.alloc(8, 255), Buffer.alloc(4, 255)] + }) + expect(result.files.every(Buffer.isBuffer)).toBe(true) + }) + + it.each(['name', 'type'])('requires nonblank text for %s', async (field) => { + const form = createForm() + form.delete(field) + await expect(readCreateInput(form)).rejects.toMatchObject({ status: 400 }) + for (const value of [' \t ', new File(['value'], 'value.txt')]) { + form.set(field, value) + await expect(readCreateInput(form)).rejects.toMatchObject({ status: 400 }) + } + }) + + it('requires optional directory to be text', async () => { + const form = createForm() + form.set('directory', new File(['value'], 'value.txt')) + await expect(readCreateInput(form)).rejects.toMatchObject({ status: 400 }) + }) + + it.each([ + ['attachment type', [1], 400], + ['per-file bytes', [1, 9], 413], + ['combined bytes', [8, 5], 413], + ['file count', [1, 1, 1], 413] + ] as const)('checks %s before reading any attachment buffers', async (reason, sizes, status) => { + const form = createForm() + for (const size of sizes) attach(form, size) + const reads = form.getAll('files').map((file) => vi.spyOn(file as File, 'arrayBuffer')) + if (reason === 'attachment type') form.append('files', 'not a file') + await expect(readCreateInput(form)).rejects.toMatchObject({ status }) + for (const bufferRead of reads) expect(bufferRead).not.toHaveBeenCalled() + }) + + it('ignores empty files without counting them toward attachment limits', async () => { + const form = createForm() + for (let i = 0; i < 3; i++) attach(form, 0, 'empty.txt') + attach(form, 0, '') + attach(form, 8) + attach(form, 4) + await expect(readCreateInput(form)).resolves.toMatchObject({ + files: [Buffer.alloc(8, 255), Buffer.alloc(4, 255)] + }) + }) +}) + +describe('readEditInput', () => { + it.each(['', '---\ntitle: Dune\n---\n\n body\n', ' body\r\nsecond line\r\n'])( + 'preserves JSON-encoded editor source exactly: %j', + async (source) => { + const form = editForm() + form.set('source', JSON.stringify(source)) + await expect(readEditInput(form)).resolves.toEqual({ + file: 'library/dune.book.md', + source, + target: { kind: 'body' }, + instructions: '', + files: [] + }) + } + ) + + it.each(['title', 'body'])('keeps field target %s distinct from the body target', async (key) => { + const form = editForm() + form.set('target', 'field') + form.set('key', key) + form.set('unrelated', 'ignored') + await expect(readEditInput(form)).resolves.toMatchObject({ target: { kind: 'field', key } }) + }) + + it.each(['file', 'source', 'target'])('requires text for %s', async (field) => { + const form = editForm() + form.delete(field) + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + form.set(field, new File(['value'], 'value.txt')) + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + }) + + it('rejects a blank file', async () => { + const form = editForm() + form.set('file', ' \t ') + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + }) + + it('rejects unsupported targets and requires a text key for field edits', async () => { + const form = editForm() + form.set('target', 'unknown') + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + form.set('target', 'field') + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + for (const key of ['', new File(['title'], 'key.txt')]) { + form.set('key', key) + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + } + }) + + it.each(['not json', 'null', '{}', '123'])( + 'rejects nonstring source JSON: %s', + async (source) => { + const form = editForm() + form.set('source', source) + await expect(readEditInput(form)).rejects.toMatchObject({ status: 400 }) + } + ) + + it('bounds decoded source characters, not UTF-8 bytes', async () => { + const form = editForm() + const source = 'é'.repeat(MAX_SOURCE_CHARS) + form.set('source', JSON.stringify(source)) + await expect(readEditInput(form)).resolves.toMatchObject({ source }) + form.set('source', JSON.stringify(source + 'é')) + await expect(readEditInput(form)).rejects.toMatchObject({ status: 413 }) + }) +}) + +describe('instructions in both readers', () => { + it('requires optional instructions to be text', async () => { + const create = createForm() + const edit = editForm() + create.set('instructions', new File(['value'], 'value.txt')) + edit.set('instructions', new File(['value'], 'value.txt')) + await expect(readCreateInput(create)).rejects.toMatchObject({ status: 400 }) + await expect(readEditInput(edit)).rejects.toMatchObject({ status: 400 }) + }) + + it('preserves whitespace and bounds characters, not UTF-8 bytes', async () => { + const create = createForm() + const edit = editForm() + const instructions = ' ' + 'é'.repeat(MAX_INSTRUCTION_CHARS - 2) + ' ' + create.set('instructions', instructions) + edit.set('instructions', instructions) + await expect(readCreateInput(create)).resolves.toMatchObject({ instructions }) + await expect(readEditInput(edit)).resolves.toMatchObject({ instructions }) + create.set('instructions', instructions + 'é') + edit.set('instructions', instructions + 'é') + await expect(readCreateInput(create)).rejects.toMatchObject({ status: 413 }) + await expect(readEditInput(edit)).rejects.toMatchObject({ status: 413 }) + }) +}) diff --git a/apps/web/src/lib/server/generation/input.ts b/apps/web/src/lib/server/generation/input.ts new file mode 100644 index 00000000..975728c9 --- /dev/null +++ b/apps/web/src/lib/server/generation/input.ts @@ -0,0 +1,84 @@ +import { error } from '@sveltejs/kit' +import { DEFAULT_GENERATION_LIMITS } from '@luzzle/core' +import type { GenerationTarget } from '$lib/generation/types' +import { MAX_SOURCE_CHARS, MAX_INSTRUCTION_CHARS } from '../constants' + +export type CreateGenerationInput = { + name: string + type: string + directory: string + instructions: string + files: Buffer[] +} + +export type EditGenerationInput = { + file: string + source: string + target: GenerationTarget + instructions: string + files: Buffer[] +} + +export async function readCreateInput(form: FormData): Promise { + const name = form.get('name') + const type = form.get('type') + const directory = form.get('directory') ?? '' + const instructions = form.get('instructions') ?? '' + if (typeof name !== 'string' || !name.trim()) error(400, 'Name is required.') + if (typeof type !== 'string' || !type.trim()) error(400, 'Piece type is required.') + if (typeof directory !== 'string') error(400, 'Directory must be text.') + if (typeof instructions !== 'string') error(400, 'Instructions must be text.') + if (instructions.length > MAX_INSTRUCTION_CHARS) error(413, 'Instructions are too long.') + return { name, type, directory, instructions, files: await readAttachments(form) } +} + +export async function readEditInput(form: FormData): Promise { + const file = form.get('file') + const encodedSource = form.get('source') + const targetKind = form.get('target') + const key = form.get('key') + const instructions = form.get('instructions') ?? '' + if (typeof file !== 'string' || !file.trim()) error(400, 'Piece file is required.') + if (typeof encodedSource !== 'string') error(400, 'Editor source is required.') + if (typeof instructions !== 'string') error(400, 'Instructions must be text.') + if (instructions.length > MAX_INSTRUCTION_CHARS) error(413, 'Instructions are too long.') + + let target: GenerationTarget + if (targetKind === 'body') { + target = { kind: 'body' } + } else if (targetKind === 'field' && typeof key === 'string' && key) { + target = { kind: 'field', key } + } else { + error(400, 'Choose a field or the body.') + } + + // JSON preserves the exact editor buffer; multipart text normalizes line endings. + let source: unknown + try { + source = JSON.parse(encodedSource) + } catch { + error(400, 'Invalid editor source.') + } + if (typeof source !== 'string') error(400, 'Editor source must be text.') + if (source.length > MAX_SOURCE_CHARS) error(413, 'Editor source is too long.') + return { file, source, target, instructions, files: await readAttachments(form) } +} + +async function readAttachments(form: FormData): Promise { + const files: File[] = [] + let bytes = 0 + for (const file of form.getAll('files')) { + if (typeof file === 'string') error(400, 'Attachments must be files.') + if (file.size === 0) continue + if (file.size > DEFAULT_GENERATION_LIMITS.maxFileBytes) error(413, 'Attachment is too large.') + bytes += file.size + if (bytes > DEFAULT_GENERATION_LIMITS.maxTotalFileBytes) + error(413, 'Attachments are too large.') + files.push(file) + } + if (files.length > DEFAULT_GENERATION_LIMITS.maxFiles) error(413, 'Too many attachments.') + + const buffers: Buffer[] = [] + for (const file of files) buffers.push(Buffer.from(await file.arrayBuffer())) + return buffers +} diff --git a/apps/web/src/lib/server/generation/stream.test.ts b/apps/web/src/lib/server/generation/stream.test.ts new file mode 100644 index 00000000..f872a5f5 --- /dev/null +++ b/apps/web/src/lib/server/generation/stream.test.ts @@ -0,0 +1,372 @@ +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import type { GenerationEvent, GenerationResult } from '$lib/generation/types' +import type { Generate } from './generator' +import { generationStream } from './stream' + +const result: GenerationResult = { kind: 'body', value: 'café\n🧩 日本語' } +const decoder = new TextDecoder() + +function deferred() { + let resolve!: (value: T) => void + let reject!: (reason: unknown) => void + const promise = new Promise((yes, no) => { + resolve = yes + reject = no + }) + return { promise, resolve, reject } +} + +function events(text: string): GenerationEvent[] { + return text + .split('\n\n') + .filter((frame) => frame.startsWith('event:')) + .map((frame) => { + const [type, data] = frame.split('\n') + return { type: type.slice(7), data: JSON.parse(data.slice(6)) } as GenerationEvent + }) +} + +async function read(reader: ReadableStreamDefaultReader) { + const chunk = await reader.read() + expect(chunk.done).toBe(false) + expect(chunk.value).toBeInstanceOf(Uint8Array) + return decoder.decode(chunk.value) +} + +async function drain(reader: ReadableStreamDefaultReader) { + let text = '' + while (true) { + const { done, value } = await reader.read() + if (done) return text + text += decoder.decode(value) + } +} + +function setup(generate?: Generate, abort = new AbortController()) { + const completion = deferred() + let progress!: Parameters[0] + const fake = vi.fn((notify) => { + progress = notify + return generate ? generate(notify) : completion.promise + }) + const request = new Request('http://localhost/api/admin/generate', { signal: abort.signal }) + const removeListener = vi.spyOn(request.signal, 'removeEventListener') + const release = vi.fn() + const response = generationStream(request, fake, release) + return { completion, progress, fake, request, abort, removeListener, release, response } +} + +beforeEach(() => { + vi.useFakeTimers() + vi.setSystemTime(1_000) +}) + +afterEach(() => { + vi.clearAllTimers() + vi.useRealTimers() + vi.restoreAllMocks() +}) + +describe('generationStream', () => { + test('streams early state and progress, then exactly one validated UTF-8 result', async () => { + const s = setup() + const reader = s.response.body!.getReader() + expect(s.response).toBeInstanceOf(Response) + expect(Object.fromEntries(s.response.headers)).toEqual({ + 'content-type': 'text/event-stream', + 'cache-control': 'no-cache', + connection: 'keep-alive', + 'x-accel-buffering': 'no' + }) + let text = await read(reader) + expect(events(text)).toEqual([{ type: 'state', data: { state: 'running' } }]) + text += await read(reader) + expect(events(text)[1]).toEqual({ + type: 'phase', + data: [ + { + phase: 'preparation', + status: 'running', + started_at: 1_000, + finished_at: null, + message: 'Preparing generation' + } + ] + }) + + vi.setSystemTime(1_100) + s.progress({ phase: 'preparation', message: 'Reading attachments' }) + text += await read(reader) + vi.setSystemTime(1_200) + s.progress({ phase: 'generation', message: 'Waiting for provider' }) + text += await read(reader) + vi.setSystemTime(1_300) + s.progress({ phase: 'validation', message: 'Checking output' }) + text += await read(reader) + expect(events(text).map((event) => event.type)).toEqual([ + 'state', + 'phase', + 'phase', + 'phase', + 'phase' + ]) + expect(text).not.toMatch(/token|result|café/) + expect(s.release).not.toHaveBeenCalled() + expect(events(text).at(-1)).toEqual({ + type: 'phase', + data: [ + { + phase: 'preparation', + status: 'completed', + started_at: 1_000, + finished_at: 1_200, + message: 'Reading attachments' + }, + { + phase: 'generation', + status: 'completed', + started_at: 1_200, + finished_at: 1_300, + message: 'Waiting for provider' + }, + { + phase: 'validation', + status: 'running', + started_at: 1_300, + finished_at: null, + message: 'Checking output' + } + ] + }) + + const next = reader.read() + const received = vi.fn() + void next.then(received) + await vi.advanceTimersByTimeAsync(0) + expect(received).not.toHaveBeenCalled() + vi.setSystemTime(1_400) + s.completion.resolve(result) + text += decoder.decode((await next).value) + text += await drain(reader) + const frames = events(text) + expect(frames.map((event) => event.type)).toEqual([ + 'state', + 'phase', + 'phase', + 'phase', + 'phase', + 'phase', + 'done' + ]) + expect(frames.filter((event) => event.type === 'done')).toEqual([ + { type: 'done', data: { state: 'completed', result } } + ]) + expect(frames.at(-2)).toMatchObject({ + type: 'phase', + data: [ + { status: 'completed' }, + { status: 'completed' }, + { status: 'completed', started_at: 1_300, finished_at: 1_400 } + ] + }) + expect(frames.at(-1)?.type).toBe('done') + expect(s.release).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + expect(s.removeListener).toHaveBeenCalledWith('abort', expect.any(Function)) + }) + + test('supports native Response.text() with multibyte output', async () => { + const s = setup(async () => result) + const text = await s.response.text() + expect(text).toContain('café\\n🧩 日本語') + expect(events(text).at(-1)).toEqual({ + type: 'done', + data: { state: 'completed', result } + }) + expect(s.release).toHaveBeenCalledTimes(1) + }) + + test.each(['reject', 'throw'] as const)( + '%s sends one failed done, a safe error and no result', + async (mode) => { + const secret = 'api-key=secret prompt=private filename=/private/scan.pdf provider-stack' + const log = vi.spyOn(console, 'error').mockImplementation(() => {}) + const s = setup( + mode === 'throw' + ? () => { + throw new Error(secret) + } + : undefined + ) + if (mode === 'reject') { + s.progress({ phase: 'validation', message: 'Checking output' }) + vi.setSystemTime(1_200) + s.completion.reject(new Error(secret)) + } + const text = await s.response.text() + const frames = events(text) + expect(frames.filter((event) => event.type === 'done')).toEqual([ + { type: 'done', data: { state: 'failed' } } + ]) + expect(frames.filter((event) => event.type === 'error')).toEqual([ + { type: 'error', data: { message: 'Generation failed. Please try again.' } } + ]) + const phase = frames.filter((event) => event.type === 'phase').at(-1)! + expect(phase.data.at(-1)).toMatchObject({ + status: 'failed', + finished_at: mode === 'reject' ? 1_200 : 1_000 + }) + expect(text).not.toMatch(/result|api-key|private|provider-stack/) + expect(log.mock.calls).toEqual([['AI generation failed.']]) + expect(s.release).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + } + ) + + test('heartbeats every 15 seconds through and beyond 300 seconds, not an operation deadline', async () => { + const s = setup() + const reader = s.response.body!.getReader() + await read(reader) + await read(reader) + for (let tick = 0; tick < 21; tick++) { + const next = reader.read() + const received = vi.fn() + void next.then(received) + await vi.advanceTimersByTimeAsync(14_999) + expect(received).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + const chunk = await next + expect(chunk.done).toBe(false) + expect(decoder.decode(chunk.value)).toBe(': heartbeat\n\n') + expect(s.release).not.toHaveBeenCalled() + } + s.completion.resolve(result) + expect(events(await drain(reader)).at(-1)).toEqual({ + type: 'done', + data: { state: 'completed', result } + }) + expect(vi.getTimerCount()).toBe(0) + }) + + test('does not accumulate queued heartbeats behind an unread response', async () => { + const s = setup() + await vi.advanceTimersByTimeAsync(300_000) + const reader = s.response.body!.getReader() + expect(events(await read(reader))[0].type).toBe('state') + expect(events(await read(reader))[0].type).toBe('phase') + // Once the initial frames drain, at most one unread heartbeat can queue. + await vi.advanceTimersByTimeAsync(300_000) + expect(await read(reader)).toBe(': heartbeat\n\n') + const received = vi.fn() + const next = reader.read() + void next.then(received) + await vi.advanceTimersByTimeAsync(14_999) + expect(received).not.toHaveBeenCalled() + s.completion.resolve(result) + expect(events(decoder.decode((await next).value))[0].type).toBe('phase') + expect(events(await drain(reader))).toEqual([ + { type: 'done', data: { state: 'completed', result } } + ]) + }) + + test.each([ + ['abort', 'success'], + ['abort', 'provider failure'], + ['abort', 'cleanup failure'], + ['cancel', 'success'], + ['cancel', 'provider failure'], + ['cancel', 'cleanup failure'] + ] as const)( + '%s detaches but retains the slot through %s and cleanup', + async (disconnect, outcome) => { + const provider = deferred() + const cleanup = deferred() + const cleaning = vi.fn() + const log = vi.spyOn(console, 'error').mockImplementation(() => {}) + const unhandled = vi.fn() + process.on('unhandledRejection', unhandled) + try { + const s = setup(async (progress) => { + try { + const value = await provider.promise + progress({ phase: 'validation', message: 'Checking abandoned output' }) + return value + } finally { + cleaning() + await cleanup.promise + } + }) + const reader = s.response.body!.getReader() + await read(reader) + await read(reader) + if (disconnect === 'abort') s.abort.abort() + else await reader.cancel() + expect(s.request.signal.aborted).toBe(disconnect === 'abort') + expect(await reader.read()).toEqual({ done: true, value: undefined }) + expect(s.removeListener).toHaveBeenCalledWith('abort', expect.any(Function)) + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(315_000) + expect(s.fake).toHaveBeenCalledTimes(1) + expect(cleaning).not.toHaveBeenCalled() + expect(s.release).not.toHaveBeenCalled() + + if (outcome === 'provider failure') provider.reject(new Error('secret provider details')) + else provider.resolve(result) + await vi.advanceTimersByTimeAsync(0) + expect(cleaning).toHaveBeenCalledTimes(1) + expect(s.release).not.toHaveBeenCalled() + if (outcome === 'cleanup failure') cleanup.reject(new Error('secret cleanup details')) + else cleanup.resolve() + await vi.advanceTimersByTimeAsync(0) + expect(s.release).toHaveBeenCalledTimes(1) + expect(await reader.read()).toEqual({ done: true, value: undefined }) + expect(log.mock.calls).toEqual(outcome === 'success' ? [] : [['AI generation failed.']]) + expect(unhandled).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + } finally { + process.off('unhandledRejection', unhandled) + } + } + ) + + test('an already-aborted accepted request still runs and releases only after settling', async () => { + const abort = new AbortController() + abort.abort() + const s = setup(undefined, abort) + expect(s.fake).toHaveBeenCalledTimes(1) + expect(await s.response.body!.getReader().read()).toEqual({ done: true, value: undefined }) + expect(s.removeListener).toHaveBeenCalledWith('abort', expect.any(Function)) + expect(vi.getTimerCount()).toBe(0) + expect(s.release).not.toHaveBeenCalled() + s.completion.resolve(result) + await vi.advanceTimersByTimeAsync(0) + expect(s.release).toHaveBeenCalledTimes(1) + }) + + test.each(['initial', 'heartbeat'] as const)( + '%s enqueue failure detaches listeners and timers, not work', + async (when) => { + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + const fail = () => + enqueue.mockImplementationOnce(() => { + throw new Error('closed transport') + }) + if (when === 'initial') fail() + const s = setup() + const reader = s.response.body!.getReader() + if (when === 'heartbeat') { + await read(reader) + await read(reader) + fail() + await vi.advanceTimersByTimeAsync(15_000) + } + expect(await reader.read()).toEqual({ done: true, value: undefined }) + expect(s.removeListener).toHaveBeenCalledWith('abort', expect.any(Function)) + expect(vi.getTimerCount()).toBe(0) + expect(s.release).not.toHaveBeenCalled() + s.completion.resolve(result) + await vi.advanceTimersByTimeAsync(0) + expect(s.release).toHaveBeenCalledTimes(1) + } + ) +}) diff --git a/apps/web/src/lib/server/generation/stream.ts b/apps/web/src/lib/server/generation/stream.ts new file mode 100644 index 00000000..ec38521c --- /dev/null +++ b/apps/web/src/lib/server/generation/stream.ts @@ -0,0 +1,75 @@ +import { DEFAULT_GENERATION_LIMITS, type GenerationProgress } from '@luzzle/core' +import type { ProgressPhase } from '$lib/components/progress' +import type { GenerationEvent } from '$lib/generation/types' +import { createEventStream } from '../sse' +import type { Generate } from './generator' + +export function generationStream( + request: Request, + generate: Generate, + release: () => void +): Response { + const events = createEventStream(request) + const phases: ProgressPhase[] = [] + + function emit(event: GenerationEvent) { + events.emit(event.type, event.data) + } + + function finishPhase(status: 'completed' | 'failed') { + const current = phases.at(-1) + if (current?.status === 'running') { + current.status = status + current.finished_at = Date.now() + } + } + + function progress(event: GenerationProgress) { + if (events.signal.aborted) return + const current = phases.at(-1) + if (current?.phase === event.phase) { + current.message = event.message + } else { + finishPhase('completed') + phases.push({ + phase: event.phase, + status: 'running', + started_at: Date.now(), + finished_at: null, + message: event.message + }) + } + emit({ type: 'phase', data: phases }) + } + + async function run() { + try { + const result = await generate(progress) + if ( + Buffer.byteLength(JSON.stringify(result), 'utf8') > DEFAULT_GENERATION_LIMITS.maxOutputBytes + ) { + throw new Error('Generated result is too large.') + } + finishPhase('completed') + emit({ type: 'phase', data: phases }) + emit({ type: 'done', data: { state: 'completed', result } }) + } catch { + // Provider exceptions may contain request details. Never send or log their raw text. + console.error('AI generation failed.') + finishPhase('failed') + emit({ type: 'phase', data: phases }) + emit({ type: 'error', data: { message: 'Generation failed. Please try again.' } }) + emit({ type: 'done', data: { state: 'failed' } }) + } finally { + // The core promise settles only after provider-file cleanup, even without a reader. + release() + events.close() + } + } + + emit({ type: 'state', data: { state: 'running' } }) + progress({ phase: 'preparation', message: 'Preparing generation' }) + // Accepted work runs even if the observer has already disconnected. + void run() + return events.response +} diff --git a/apps/web/src/lib/server/sse.test.ts b/apps/web/src/lib/server/sse.test.ts index 1c26c978..a9f60b28 100644 --- a/apps/web/src/lib/server/sse.test.ts +++ b/apps/web/src/lib/server/sse.test.ts @@ -1,5 +1,6 @@ -import { describe, expect, test } from 'vitest' -import { encodeEvent } from './sse.js' +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { HEARTBEAT_MS } from './constants.js' +import { createEventStream, encodeEvent } from './sse.js' describe('encodeEvent', () => { test('preserves unicode and JSON-escapes newlines in a single data line', () => { @@ -22,3 +23,237 @@ describe('encodeEvent', () => { expect(encodeEvent('state', [], '0')).toBe('event: state\nid: 0\ndata: []\n\n') }) }) + +describe('createEventStream', () => { + const streams: ReturnType[] = [] + + function open(request = new Request('http://localhost/events')) { + const stream = createEventStream(request) + streams.push(stream) + return stream + } + + beforeEach(() => { + vi.useFakeTimers() + }) + + afterEach(() => { + for (const stream of streams.splice(0)) stream.close() + vi.restoreAllMocks() + vi.useRealTimers() + }) + + test('returns a native response with the existing SSE headers', () => { + const { response } = open() + + expect(response).toBeInstanceOf(Response) + expect(response.status).toBe(200) + expect(Object.fromEntries(response.headers)).toEqual({ + 'cache-control': 'no-cache', + connection: 'keep-alive', + 'content-type': 'text/event-stream', + 'x-accel-buffering': 'no' + }) + }) + + test('encodes unicode as bytes consumable by native Response.text()', async () => { + const { response, emit, close } = open() + emit('log', { message: 'café 🧩 日本語\nnext\r\n"quoted"' }) + close() + + await expect(response.text()).resolves.toBe( + 'event: log\ndata: {"message":"café 🧩 日本語\\nnext\\r\\n\\"quoted\\""}\n\n' + ) + }) + + test('preserves cursor ids and event framing unchanged on the wire', async () => { + const { response, emit, close } = open() + emit('cursor', { sync: 7, publish: 2 }, '{"sync":7,"publish":2}') + emit('state', [], '0') + emit('done', null) + close() + + await expect(response.text()).resolves.toBe( + 'event: cursor\nid: {"sync":7,"publish":2}\ndata: {"sync":7,"publish":2}\n\n' + + 'event: state\nid: 0\ndata: []\n\n' + + 'event: done\ndata: null\n\n' + ) + }) + + test('keeps a quiet observer alive with a byte heartbeat every 15 seconds', async () => { + const { response, signal } = open() + const reader = response.body!.getReader() + const received = vi.fn() + const first = reader.read().then((chunk) => { + received(chunk) + return chunk + }) + + expect(HEARTBEAT_MS).toBe(15_000) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS - 1) + expect(received).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + await expect(first).resolves.toEqual({ + done: false, + value: new TextEncoder().encode(': heartbeat\n\n') + }) + + const second = reader.read() + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS) + await expect(second).resolves.toEqual({ + done: false, + value: new TextEncoder().encode(': heartbeat\n\n') + }) + expect(signal.aborted).toBe(false) + await reader.cancel() + }) + + test('has no overall stream timeout even after a day without producer events', async () => { + const { response, emit, signal } = open() + const reader = response.body!.getReader() + + await vi.advanceTimersByTimeAsync(24 * 60 * 60 * 1000) + expect(signal.aborted).toBe(false) + expect(vi.getTimerCount()).toBe(1) + await expect(reader.read()).resolves.toEqual({ + done: false, + value: new TextEncoder().encode(': heartbeat\n\n') + }) + + emit('done', { ok: true }) + await expect(reader.read()).resolves.toEqual({ + done: false, + value: new TextEncoder().encode('event: done\ndata: {"ok":true}\n\n') + }) + await reader.cancel() + }) + + test('does not accumulate unread heartbeats behind events or other heartbeats', async () => { + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + const { response, emit } = open() + const reader = response.body!.getReader() + + emit('state', { ready: true }) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 100) + expect(enqueue).toHaveBeenCalledTimes(1) + await expect(reader.read()).resolves.toEqual({ + done: false, + value: new TextEncoder().encode('event: state\ndata: {"ready":true}\n\n') + }) + + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 100) + expect(enqueue).toHaveBeenCalledTimes(2) + await expect(reader.read()).resolves.toEqual({ + done: false, + value: new TextEncoder().encode(': heartbeat\n\n') + }) + + // Draining the queue allows future heartbeats, not a backlog of old ones. + const received = vi.fn() + const next = reader.read().then(received) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS - 1) + expect(received).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + await next + expect(enqueue).toHaveBeenCalledTimes(3) + await reader.cancel() + }) + + test.each([ + 'explicit close', + 'request abort', + 'reader cancellation', + 'event enqueue failure', + 'heartbeat enqueue failure' + ])('%s ends only observation and cleans up exactly once', async (cause) => { + const abort = new AbortController() + const request = new Request('http://localhost/events', { signal: abort.signal }) + const removeListener = vi.spyOn(request.signal, 'removeEventListener') + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + const closeController = vi.spyOn(ReadableStreamDefaultController.prototype, 'close') + const { response, emit, close, signal } = open(request) + const reader = response.body!.getReader() + const observerClosed = vi.fn() + signal.addEventListener('abort', observerClosed) + const pending = reader.read() + + expect(signal).not.toBe(request.signal) + expect(signal.aborted).toBe(false) + expect(vi.getTimerCount()).toBe(1) + + switch (cause) { + case 'explicit close': + close() + break + case 'request abort': + abort.abort() + break + case 'reader cancellation': + await reader.cancel() + break + default: + // Keep native Request/Response/streams; fail only the synchronous enqueue. + enqueue.mockImplementationOnce(() => { + throw new TypeError('observer is no longer readable') + }) + if (cause === 'event enqueue failure') { + expect(() => emit('state', {})).not.toThrow() + } else { + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS) + } + } + + expect(signal.aborted).toBe(true) + expect(request.signal.aborted).toBe(cause === 'request abort') + expect(observerClosed).toHaveBeenCalledTimes(1) + expect(removeListener).toHaveBeenCalledExactlyOnceWith('abort', close) + expect(closeController).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + await expect(pending).resolves.toEqual({ done: true, value: undefined }) + + const enqueueCalls = enqueue.mock.calls.length + const toJSON = vi.fn(() => { + throw new Error('closed observers must not serialize frames') + }) + expect(() => emit('late', { toJSON }, 'late-cursor')).not.toThrow() + close() + close() + abort.abort() + await reader.cancel() + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 100) + + expect(toJSON).not.toHaveBeenCalled() + expect(enqueue).toHaveBeenCalledTimes(enqueueCalls) + expect(observerClosed).toHaveBeenCalledTimes(1) + expect(removeListener).toHaveBeenCalledTimes(1) + expect(closeController).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + await expect(reader.read()).resolves.toEqual({ done: true, value: undefined }) + }) + + test('an already-aborted request closes immediately without starting any heartbeat', async () => { + const abort = new AbortController() + abort.abort() + const request = new Request('http://localhost/events', { signal: abort.signal }) + const removeListener = vi.spyOn(request.signal, 'removeEventListener') + const interval = vi.spyOn(globalThis, 'setInterval') + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + const { response, emit, close, signal } = open(request) + const toJSON = vi.fn(() => { + throw new Error('already-closed observers must not serialize frames') + }) + + expect(signal.aborted).toBe(true) + expect(removeListener).toHaveBeenCalledExactlyOnceWith('abort', close) + expect(interval).not.toHaveBeenCalled() + expect(() => emit('late', { toJSON })).not.toThrow() + close() + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 100) + + expect(toJSON).not.toHaveBeenCalled() + expect(enqueue).not.toHaveBeenCalled() + expect(removeListener).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + await expect(response.text()).resolves.toBe('') + }) +}) diff --git a/apps/web/src/lib/server/sse.ts b/apps/web/src/lib/server/sse.ts index 9d586d89..9a146d24 100644 --- a/apps/web/src/lib/server/sse.ts +++ b/apps/web/src/lib/server/sse.ts @@ -1,3 +1,5 @@ +import { HEARTBEAT_MS } from './constants' + export const SSE_HEADERS = { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', @@ -11,3 +13,58 @@ export function encodeEvent(event: string, data: unknown, id?: string): string { lines.push(`data: ${JSON.stringify(data)}`) return lines.join('\n') + '\n\n' } + +/** Owns the response observer, not the work producing its events. */ +export function createEventStream(request: Request) { + const observer = new AbortController() + const encoder = new TextEncoder() + let controller: ReadableStreamDefaultController + let heartbeat: ReturnType | undefined + + function close() { + if (observer.signal.aborted) return + clearInterval(heartbeat) + request.signal.removeEventListener('abort', close) + observer.abort() + try { + controller.close() + } catch { + // Reader cancellation or an enqueue failure may already have closed it. + } + } + + function send(text: string) { + if (observer.signal.aborted) return + try { + controller.enqueue(encoder.encode(text)) + } catch { + close() + } + } + + const stream = new ReadableStream({ + start(value) { + controller = value + request.signal.addEventListener('abort', close, { once: true }) + if (request.signal.aborted) { + close() + return + } + heartbeat = setInterval(() => { + // Don't accumulate heartbeats behind an unread response. + if ((controller.desiredSize ?? 0) > 0) send(': heartbeat\n\n') + }, HEARTBEAT_MS) + }, + cancel: close + }) + + return { + response: new Response(stream, { headers: SSE_HEADERS }), + // Only observation ends here. Producers must not pass this signal to accepted work. + signal: observer.signal, + emit(event: string, data: unknown, id?: string) { + if (!observer.signal.aborted) send(encodeEvent(event, data, id)) + }, + close + } +} diff --git a/apps/web/src/lib/server/workflow/stream.test.ts b/apps/web/src/lib/server/workflow/stream.test.ts index 0de41686..45c3d216 100644 --- a/apps/web/src/lib/server/workflow/stream.test.ts +++ b/apps/web/src/lib/server/workflow/stream.test.ts @@ -1,4 +1,5 @@ import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { HEARTBEAT_MS, WORKFLOW_POLL_INTERVAL_MS } from '../constants.js' import { streamJobProgress } from './stream.js' const mocks = vi.hoisted(() => { @@ -12,6 +13,7 @@ const mocks = vi.hoisted(() => { query, selectFrom: vi.fn(() => query), getOpenWorkflowDb: vi.fn(() => ({})), + getOpenWorkflow: vi.fn(), getWorkflowRun: vi.fn(), getStepAttempts: vi.fn() } @@ -20,7 +22,10 @@ const mocks = vi.hoisted(() => { vi.mock('$lib/server/database/index.js', () => ({ db: { selectFrom: mocks.selectFrom } })) -vi.mock('./index.js', () => ({ getOpenWorkflowDb: mocks.getOpenWorkflowDb })) +vi.mock('./index.js', () => ({ + getOpenWorkflowDb: mocks.getOpenWorkflowDb, + getOpenWorkflow: mocks.getOpenWorkflow +})) vi.mock('@luzzle/web.jobs', () => ({ getWorkflowRun: mocks.getWorkflowRun, getStepAttempts: mocks.getStepAttempts @@ -46,12 +51,19 @@ const phaseFrame = function openStream({ header, cursor, - jobClass = 'Publish' -}: { header?: string; cursor?: string; jobClass?: string | string[] } = {}) { + jobClass = 'Publish', + signal +}: { + header?: string + cursor?: string + jobClass?: string | string[] + signal?: AbortSignal +} = {}) { const url = new URL('http://localhost/jobs/job-1/stream') if (cursor !== undefined) url.searchParams.set('cursor', cursor) const request = new Request(url, { - headers: header === undefined ? {} : { 'Last-Event-ID': header } + headers: header === undefined ? {} : { 'Last-Event-ID': header }, + signal }) return streamJobProgress({ jobId: 'job-1', jobClass, request, url }) } @@ -62,20 +74,42 @@ async function readChunks(response: Response): Promise { while (true) { const { value, done } = await reader.read() if (done) return chunks - // The existing stream enqueues strings, not the bytes Response.text() expects. - expect(typeof value).toBe('string') - chunks.push(value as unknown as string) + expect(value).toBeInstanceOf(Uint8Array) + chunks.push(new TextDecoder().decode(value)) } } +async function readFrame(reader: ReadableStreamDefaultReader): Promise { + const { value, done } = await reader.read() + expect(done).toBe(false) + expect(value).toBeInstanceOf(Uint8Array) + return new TextDecoder().decode(value) +} + +function deferredLogs() { + let resolve!: (logs: (typeof log)[]) => void + const promise = new Promise<(typeof log)[]>((done) => { + resolve = done + }) + return { promise, resolve } +} + beforeEach(() => { - vi.clearAllMocks() + vi.resetAllMocks() + mocks.selectFrom.mockReturnValue(mocks.query) + mocks.query.selectAll.mockReturnThis() + mocks.query.where.mockReturnThis() + mocks.query.orderBy.mockReturnThis() + mocks.getOpenWorkflowDb.mockReturnValue({}) mocks.getWorkflowRun.mockReturnValue(run) mocks.getStepAttempts.mockReturnValue([step]) mocks.query.execute.mockResolvedValue([]) }) afterEach(() => { + // Observing or disconnecting must never acquire a worker client to cancel its job. + expect(mocks.getOpenWorkflow).not.toHaveBeenCalled() + vi.restoreAllMocks() vi.useRealTimers() }) @@ -125,7 +159,7 @@ describe('streamJobProgress', () => { mocks.getWorkflowRun.mockReturnValueOnce({ ...run, status: 'running' }) mocks.query.execute.mockResolvedValueOnce([log]) const chunksPromise = readChunks(openStream()) - await vi.advanceTimersByTimeAsync(350) + await vi.advanceTimersByTimeAsync(WORKFLOW_POLL_INTERVAL_MS) const chunks = await chunksPromise expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(2) expect(mocks.query.where.mock.calls.filter(([column]) => column === 'line_number')).toEqual([ @@ -146,6 +180,184 @@ describe('streamJobProgress', () => { ) }) + test('keeps a quiet, pending database read alive every 15 seconds and cleans up on completion', async () => { + vi.useFakeTimers() + expect(HEARTBEAT_MS).toBe(15_000) + expect(WORKFLOW_POLL_INTERVAL_MS).toBe(350) + const pending = deferredLogs() + mocks.query.execute.mockReturnValueOnce(pending.promise) + const reader = openStream().body!.getReader() + expect(await readFrame(reader)).toBe( + 'event: state\ndata: {"state":"completed","result":"ok","errors":null}\n\n' + ) + expect(await readFrame(reader)).toBe(phaseFrame) + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + + for (let heartbeat = 0; heartbeat < 2; heartbeat++) { + const frame = readFrame(reader) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS - 1) + expect(enqueue).toHaveBeenCalledTimes(heartbeat) + await vi.advanceTimersByTimeAsync(1) + expect(await frame).toBe(': heartbeat\n\n') + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1) + expect(mocks.query.execute).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(1) + } + + pending.resolve([]) + expect(await readFrame(reader)).toBe( + 'event: done\ndata: {"state":"completed","result":"ok","errors":null}\n\n' + ) + expect(await reader.read()).toEqual({ done: true, value: undefined }) + expect(vi.getTimerCount()).toBe(0) + }) + + test.each(['request abort', 'reader cancel'] as const)( + '%s stops polling and removes sleep listeners on both timeout and cancellation', + async (disconnect) => { + vi.useFakeTimers() + mocks.getWorkflowRun.mockReturnValue({ ...run, status: 'running' }) + const abort = new AbortController() + const add = vi.spyOn(AbortSignal.prototype, 'addEventListener') + const remove = vi.spyOn(AbortSignal.prototype, 'removeEventListener') + const reader = openStream({ signal: abort.signal }).body!.getReader() + await readFrame(reader) + await readFrame(reader) + await vi.advanceTimersByTimeAsync(0) + + // The last registration is the current poll sleep, on the observer's signal. + const sleepIndex = add.mock.calls.findLastIndex(([type]) => type === 'abort') + const observerSignal = add.mock.contexts[sleepIndex] as AbortSignal + expect(observerSignal).toBeInstanceOf(AbortSignal) + expect(observerSignal).not.toBe(abort.signal) + const addedListeners = () => + add.mock.calls + .filter( + ([type], index) => type === 'abort' && add.mock.contexts[index] === observerSignal + ) + .map(([, listener]) => listener) + const removedListeners = () => + remove.mock.calls + .filter( + ([type], index) => type === 'abort' && remove.mock.contexts[index] === observerSignal + ) + .map(([, listener]) => listener) + + for (let poll = 1; poll <= 20; poll++) { + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(poll) + expect(addedListeners()).toHaveLength(poll) + expect(removedListeners()).toEqual(addedListeners().slice(0, -1)) + expect(vi.getTimerCount()).toBe(2) + if (poll === 20) break + await vi.advanceTimersByTimeAsync(WORKFLOW_POLL_INTERVAL_MS - 1) + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(poll) + await vi.advanceTimersByTimeAsync(1) + await readFrame(reader) + await readFrame(reader) + } + + if (disconnect === 'request abort') abort.abort() + else await reader.cancel() + expect(abort.signal.aborted).toBe(disconnect === 'request abort') + expect(observerSignal.aborted).toBe(true) + expect(removedListeners()).toEqual(addedListeners()) + expect(vi.getTimerCount()).toBe(0) + expect(await reader.read()).toEqual({ done: true, value: undefined }) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2) + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(20) + expect(mocks.query.execute).toHaveBeenCalledTimes(20) + expect(vi.getTimerCount()).toBe(0) + } + ) + + test('an enqueue failure during poll sleep clears both timers and stops future polls', async () => { + vi.useFakeTimers() + mocks.getWorkflowRun.mockReturnValue({ ...run, status: 'running' }) + const abort = new AbortController() + const chunksPromise = readChunks(openStream({ signal: abort.signal })) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS - 1) + const polls = Math.floor((HEARTBEAT_MS - 1) / WORKFLOW_POLL_INTERVAL_MS) + 1 + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(polls) + expect(vi.getTimerCount()).toBe(2) + const enqueue = vi + .spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + .mockImplementationOnce(() => { + throw new TypeError('Cannot enqueue into a closed stream') + }) + + await vi.advanceTimersByTimeAsync(1) + expect(vi.getTimerCount()).toBe(0) + expect(abort.signal.aborted).toBe(false) + const chunks = await chunksPromise + expect(chunks).toHaveLength(polls * 2) + expect(chunks.every((chunk) => /^(event: state|event: phase)\n/.test(chunk))).toBe(true) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2) + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(polls) + expect(mocks.query.execute).toHaveBeenCalledTimes(polls) + expect(enqueue).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + }) + + test.each(['request abort', 'reader cancel', 'enqueue failure'] as const)( + '%s closes immediately during an in-flight DB read, with no later polling or frames', + async (disconnect) => { + vi.useFakeTimers() + mocks.getWorkflowRun.mockReturnValue({ ...run, status: 'running' }) + const pending = deferredLogs() + mocks.query.execute.mockReturnValueOnce(pending.promise) + const abort = new AbortController() + const reader = openStream({ signal: abort.signal }).body!.getReader() + await readFrame(reader) + await readFrame(reader) + expect(vi.getTimerCount()).toBe(1) + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + + if (disconnect === 'request abort') abort.abort() + else if (disconnect === 'reader cancel') await reader.cancel() + else { + enqueue.mockImplementationOnce(() => { + throw new TypeError('Cannot enqueue into a closed stream') + }) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS) + } + + // Neither closing the response nor clearing timers may wait for the DB promise. + expect(abort.signal.aborted).toBe(disconnect === 'request abort') + expect(vi.getTimerCount()).toBe(0) + expect(await reader.read()).toEqual({ done: true, value: undefined }) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2) + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1) + expect(mocks.query.execute).toHaveBeenCalledTimes(1) + + pending.resolve([log]) + await vi.advanceTimersByTimeAsync(0) + // A DB read finishing after disconnect must not start another polling wait. + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2) + expect(mocks.getWorkflowRun).toHaveBeenCalledTimes(1) + expect(mocks.query.execute).toHaveBeenCalledTimes(1) + expect(enqueue).toHaveBeenCalledTimes(disconnect === 'enqueue failure' ? 1 : 0) + expect(vi.getTimerCount()).toBe(0) + expect(await reader.read()).toEqual({ done: true, value: undefined }) + } + ) + + test('an already-aborted request never polls, emits, or installs timers', async () => { + vi.useFakeTimers() + const abort = new AbortController() + abort.abort() + const enqueue = vi.spyOn(ReadableStreamDefaultController.prototype, 'enqueue') + expect(await readChunks(openStream({ signal: abort.signal }))).toEqual([]) + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(HEARTBEAT_MS * 2) + expect(mocks.getOpenWorkflowDb).not.toHaveBeenCalled() + expect(mocks.getWorkflowRun).not.toHaveBeenCalled() + expect(mocks.getStepAttempts).not.toHaveBeenCalled() + expect(mocks.query.execute).not.toHaveBeenCalled() + expect(enqueue).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + }) + test.each([ ['completed', 'completed', '"ok"'], ['succeeded', 'completed', '"ok"'], diff --git a/apps/web/src/lib/server/workflow/stream.ts b/apps/web/src/lib/server/workflow/stream.ts index 279af485..dfecca97 100644 --- a/apps/web/src/lib/server/workflow/stream.ts +++ b/apps/web/src/lib/server/workflow/stream.ts @@ -1,9 +1,9 @@ import { db, type JobProgressRow } from '$lib/server/database/index.js' -import { encodeEvent, SSE_HEADERS } from '../sse.js' +import { createEventStream } from '../sse.js' +import { WORKFLOW_POLL_INTERVAL_MS } from '../constants.js' import { getOpenWorkflowDb } from './index.js' import { getWorkflowRun, getStepAttempts } from '@luzzle/web.jobs' -const POLL_INTERVAL_MS = 350 const TERMINAL_STATES = new Set(['completed', 'failed', 'canceled']) type Cursors = Record @@ -25,17 +25,17 @@ function parseCursors(raw: string | null): Cursors { } } -function sleep(ms: number, signal: AbortSignal): Promise { +function waitForNextPoll(signal: AbortSignal): Promise { return new Promise((resolve) => { - const timer = setTimeout(resolve, ms) - signal.addEventListener( - 'abort', - () => { - clearTimeout(timer) - resolve() - }, - { once: true } - ) + // The observer may have closed while the previous database read was pending. + if (signal.aborted) return resolve() + const finish = () => { + clearTimeout(timer) + signal.removeEventListener('abort', finish) + resolve() + } + const timer = setTimeout(finish, WORKFLOW_POLL_INTERVAL_MS) + signal.addEventListener('abort', finish, { once: true }) }) } @@ -162,42 +162,23 @@ export function streamJobProgress({ request.headers.get('last-event-id') ?? url.searchParams.get('cursor') ) - const stream = new ReadableStream({ - async start(controller) { - const { signal } = request - let closed = false + const events = createEventStream(request) - const close = () => { - if (closed) return - closed = true - try { - controller.close() - } catch { - // already closed - } - } - - const emit: Emit = (event, data, id) => { - if (closed) return - try { - controller.enqueue(encodeEvent(event, data, id)) - } catch { - close() - } - } - - signal.addEventListener('abort', close, { once: true }) - - while (!closed && !signal.aborted) { - const terminal = await pollOnce(jobId, jobClass, cursors, emit) - if (terminal) { - close() - return - } - await sleep(POLL_INTERVAL_MS, signal) + async function poll() { + try { + while (!events.signal.aborted) { + if (await pollOnce(jobId, jobClass, cursors, events.emit)) break + await waitForNextPoll(events.signal) } + } catch (error) { + console.error('SSE stream error:', error) + events.emit('error', { message: 'Internal stream error' }) + } finally { + events.close() } - }) + } - return new Response(stream, { headers: SSE_HEADERS }) + // Only polling ends when the observer disconnects; the worker job is independent. + void poll() + return events.response } diff --git a/apps/web/src/routes/api/admin/generate/+server.ts b/apps/web/src/routes/api/admin/generate/+server.ts new file mode 100644 index 00000000..5e6cb237 --- /dev/null +++ b/apps/web/src/routes/api/admin/generate/+server.ts @@ -0,0 +1,4 @@ +import type { RequestHandler } from './$types' +import { generateResponse } from '$lib/server/generation' + +export const POST: RequestHandler = ({ request }) => generateResponse(request) -- 2.54.0