diff --git a/.gitignore b/.gitignore index d340dcd..bd804cc 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,8 @@ dist/ # Lobster state .lobster/ +.lobster-cache/ +.pnpm-store/ # Private docs clawdbot_enhancement.md diff --git a/package.json b/package.json index b20bb8b..2ba5b15 100644 --- a/package.json +++ b/package.json @@ -55,6 +55,7 @@ "homepage": "https://github.com/clawdbot/lobster#readme", "license": "MIT", "dependencies": { + "ajv": "^8.17.1", "yaml": "^2.8.2" } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f75fa5f..d67a4c4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -8,6 +8,9 @@ importers: .: dependencies: + ajv: + specifier: ^8.17.1 + version: 8.17.1 yaml: specifier: ^2.8.2 version: 2.8.2 @@ -67,11 +70,27 @@ packages: '@types/node@22.19.7': resolution: {integrity: sha512-MciR4AKGHWl7xwxkBa6xUGxQJ4VBOmPTF7sL+iGzuahOFaO0jHCsuEfS80pan1ef4gWId1oWOweIhrDEYLuaOw==} + ajv@8.17.1: + resolution: {integrity: sha512-B/gBuNg5SiMTrPkC+A2+cW0RszwxYmn6VYxB/inlBStS5nx6xHIt/ehKRhIMhqusl7a8LjQoZnjCs5vhwxOQ1g==} + + fast-deep-equal@3.1.3: + resolution: {integrity: sha512-f3qQ9oQy9j2AhBe/H9VC91wLmKBCCU/gDOnKNAYG5hswO7BLKj09Hc5HYNz9cGI++xlpDCIgDaitVs03ATR84Q==} + + fast-uri@3.1.0: + resolution: {integrity: sha512-iPeeDKJSWf4IEOasVVrknXpaBV0IApz/gp7S2bb7Z4Lljbl2MGJRqInZiUrQwV16cpzw/D3S5j5Julj/gT52AA==} + + json-schema-traverse@1.0.0: + resolution: {integrity: sha512-NM8/P9n3XjXhIZn1lLhkFaACTOURQXjWhV4BA/RnOv8xvgqtqpAX9IO4mRQxSx1Rlo4tqzeqb0sOlruaOy3dug==} + oxlint@0.15.15: resolution: {integrity: sha512-oQNc1mAHrrbKiXyKJMGs9VCZfwGfLy7YiQKa4qupi71X/u4xyWqOh36YKXqWOXnmm2y7vfWFpGZlhJPAa9tMqA==} engines: {node: '>=8.*'} hasBin: true + require-from-string@2.0.2: + resolution: {integrity: sha512-Xf0nWe6RseziFMu+Ap9biiUbmplq6S9/p+7w7YXP/JBHhrUDDUhwa+vANyubuqfZWTveU//DYVGsDG7RKL/vEw==} + engines: {node: '>=0.10.0'} + typescript@5.9.3: resolution: {integrity: sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==} engines: {node: '>=14.17'} @@ -115,6 +134,19 @@ snapshots: dependencies: undici-types: 6.21.0 + ajv@8.17.1: + dependencies: + fast-deep-equal: 3.1.3 + fast-uri: 3.1.0 + json-schema-traverse: 1.0.0 + require-from-string: 2.0.2 + + fast-deep-equal@3.1.3: {} + + fast-uri@3.1.0: {} + + json-schema-traverse@1.0.0: {} + oxlint@0.15.15: optionalDependencies: '@oxlint/darwin-arm64': 0.15.15 @@ -126,6 +158,8 @@ snapshots: '@oxlint/win32-arm64': 0.15.15 '@oxlint/win32-x64': 0.15.15 + require-from-string@2.0.2: {} + typescript@5.9.3: {} undici-types@6.21.0: {} diff --git a/src/commands/registry.ts b/src/commands/registry.ts index b6a860f..74ef5f5 100644 --- a/src/commands/registry.ts +++ b/src/commands/registry.ts @@ -11,6 +11,7 @@ import { mapCommand } from "./stdlib/map.js"; import { groupByCommand } from "./stdlib/group_by.js"; import { approveCommand } from "./stdlib/approve.js"; import { clawdInvokeCommand } from "./stdlib/clawd_invoke.js"; +import { llmTaskInvokeCommand } from "./stdlib/llm_task_invoke.js"; import { stateGetCommand, stateSetCommand } from "./stdlib/state.js"; import { diffLastCommand } from "./stdlib/diff_last.js"; import { workflowsListCommand } from "./workflows/workflows_list.js"; @@ -37,6 +38,7 @@ export function createDefaultRegistry() { groupByCommand, approveCommand, clawdInvokeCommand, + llmTaskInvokeCommand, stateGetCommand, stateSetCommand, diffLastCommand, diff --git a/src/commands/stdlib/llm_task_invoke.ts b/src/commands/stdlib/llm_task_invoke.ts new file mode 100644 index 0000000..fa46b47 --- /dev/null +++ b/src/commands/stdlib/llm_task_invoke.ts @@ -0,0 +1,561 @@ +import path from 'node:path'; +import { promises as fsp } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { Ajv } from 'ajv'; +import type { ErrorObject } from 'ajv'; + +import { readStateJson, writeStateJson, stableStringify } from '../../state/store.js'; + +const ajv = new Ajv({ allErrors: true, strict: false }); + +const artifactSchema = { + type: 'object', + properties: { + kind: { type: 'string' }, + role: { type: 'string' }, + name: { type: 'string' }, + mimeType: { type: 'string' }, + text: { type: 'string' }, + data: {}, + uri: { type: 'string' }, + }, + additionalProperties: true, +}; + +const payloadSchema = { + type: 'object', + properties: { + prompt: { type: 'string', minLength: 1 }, + model: { type: 'string', minLength: 1 }, + artifacts: { type: 'array', items: artifactSchema }, + artifactHashes: { type: 'array', items: { type: 'string', minLength: 10 } }, + schemaVersion: { type: 'string' }, + metadata: { type: 'object', additionalProperties: true }, + outputSchema: { type: 'object', additionalProperties: true }, + temperature: { type: 'number' }, + maxOutputTokens: { type: 'number' }, + retryContext: { + type: 'object', + properties: { + attempt: { type: 'number' }, + validationErrors: { type: 'array', items: { type: 'string' } }, + }, + additionalProperties: false, + }, + }, + required: ['prompt', 'model', 'artifacts', 'artifactHashes'], + additionalProperties: false, +}; + +const responseSchema = { + type: 'object', + properties: { + ok: { type: 'boolean' }, + result: { + type: 'object', + properties: { + runId: { type: 'string' }, + model: { type: 'string' }, + prompt: { type: 'string' }, + status: { type: 'string' }, + output: { + type: 'object', + properties: { + text: { type: 'string' }, + data: {}, + format: { type: 'string' }, + }, + required: [], + additionalProperties: true, + }, + usage: { + type: 'object', + properties: { + inputTokens: { type: 'number' }, + outputTokens: { type: 'number' }, + totalTokens: { type: 'number' }, + }, + additionalProperties: true, + }, + warnings: { type: 'array', items: { type: 'string' } }, + metadata: { type: 'object', additionalProperties: true }, + diagnostics: { type: 'object', additionalProperties: true }, + }, + required: ['output'], + additionalProperties: true, + }, + error: { type: 'object', additionalProperties: true }, + }, + required: ['ok'], + additionalProperties: true, +}; + +const validatePayload = ajv.compile(payloadSchema); +const validateResponseEnvelope = ajv.compile(responseSchema); + +const DEFAULT_MAX_VALIDATION_RETRIES = 1; + +const STATE_VERSION = 1; + +type LlmTaskResponseEnvelope = { + ok: boolean; + result?: LlmTaskResponse | null; + error?: { message?: string } | null; +}; + +type LlmTaskResponse = { + runId?: string | null; + model?: string | null; + prompt?: string | null; + status?: string | null; + output?: { + text?: string | null; + data?: any; + format?: string | null; + } | null; + usage?: Record | null; + warnings?: string[] | null; + metadata?: Record | null; + diagnostics?: Record | null; +}; + +type NormalizedInvocationItem = { + kind: 'llm_task.invoke'; + runId: string | null; + prompt: string | null; + model: string | null; + schemaVersion: string | null; + status: string; + cacheKey: string; + artifactHashes: string[]; + output: { format: string | null; text: string | null; data: any }; + usage: Record | null; + metadata: Record | null; + warnings: string[] | null; + diagnostics: Record | null; + createdAt: string; + source: string; + cached: boolean; + attemptCount: number; +}; + +type CacheEntry = { + items: NormalizedInvocationItem[]; + cacheKey: string; + storedAt: string; +}; + +export const llmTaskInvokeCommand = { + name: 'llm_task.invoke', + meta: { + description: 'Call the llm-task /tool/invoke endpoint with typed payloads and caching', + argsSchema: { + type: 'object', + properties: { + url: { type: 'string', description: 'llm-task base URL (or LLM_TASK_URL)' }, + token: { type: 'string', description: 'Bearer token (or LLM_TASK_TOKEN)' }, + prompt: { type: 'string', description: 'Primary prompt / instructions' }, + model: { type: 'string', description: 'Model identifier (e.g. claude-3-sonnet)' }, + 'artifacts-json': { type: 'string', description: 'JSON array of artifacts to send' }, + 'metadata-json': { type: 'string', description: 'JSON object of metadata to include' }, + 'output-schema': { type: 'string', description: 'JSON schema LLM output must satisfy' }, + 'schema-version': { type: 'string', description: 'Logical schema version for caching' }, + 'max-validation-retries': { type: 'number', description: 'Retries when schema validation fails' }, + temperature: { type: 'number', description: 'Sampling temperature' }, + 'max-output-tokens': { type: 'number', description: 'Max completion tokens' }, + 'state-key': { type: 'string', description: 'Run-state key override (else LOBSTER_RUN_STATE_KEY)' }, + refresh: { type: 'boolean', description: 'Bypass run-state + cache' }, + 'disable-cache': { type: 'boolean', description: 'Skip persistent cache' }, + _: { type: 'array', items: { type: 'string' } }, + }, + required: ['model'], + }, + sideEffects: ['calls_llm_task'], + }, + help() { + return ( + `llm_task.invoke — call llm-task /tool/invoke with caching and schema validation\n\n` + + `Usage:\n` + + ` llm_task.invoke --model claude-3-sonnet --prompt 'Write summary'\n` + + ` cat artifacts.json | llm_task.invoke --model claude-3-sonnet --prompt 'Score each item'\n` + + ` ... | llm_task.invoke --model claude-3-sonnet --prompt 'Plan next steps' --output-schema '{"type":"object"}'\n\n` + + `Features:\n` + + ` - Typed payload validation before invoking remote tool.\n` + + ` - Run-state + file cache so resumes do not re-call the LLM.\n` + + ` - Optional JSON-schema enforcement with bounded retries.\n` + ); + }, + async run({ input, args, ctx }) { + const env = ctx.env ?? process.env; + const baseUrl = String(args.url ?? env.LLM_TASK_URL ?? '').trim(); + if (!baseUrl) throw new Error('llm_task.invoke requires --url or LLM_TASK_URL'); + + const prompt = extractPrompt(args); + if (!prompt) throw new Error('llm_task.invoke requires --prompt or positional text'); + + const model = String(args.model ?? '').trim(); + if (!model) throw new Error('llm_task.invoke requires --model'); + + const schemaVersion = args['schema-version'] + ? String(args['schema-version']).trim() + : env.LLM_TASK_SCHEMA_VERSION + ? String(env.LLM_TASK_SCHEMA_VERSION).trim() + : 'v1'; + + const maxOutputTokens = parseOptionalNumber(args['max-output-tokens']); + const temperature = parseOptionalNumber(args.temperature); + + const providedArtifacts = parseJsonArray(args['artifacts-json'], 'llm_task.invoke --artifacts-json'); + const metadataObject = parseJsonObject(args['metadata-json'], 'llm_task.invoke --metadata-json'); + const userOutputSchema = parseJsonObject(args['output-schema'], 'llm_task.invoke --output-schema'); + + const maxValidationRetriesRaw = args['max-validation-retries'] ?? env.LLM_TASK_VALIDATION_RETRIES; + const maxValidationRetries = userOutputSchema + ? Math.max(0, Number.isFinite(Number(maxValidationRetriesRaw)) + ? Number(maxValidationRetriesRaw) + : DEFAULT_MAX_VALIDATION_RETRIES) + : 0; + + const disableCache = flag(args['disable-cache']); + const forceRefresh = flag(args.refresh ?? env.LLM_TASK_FORCE_REFRESH); + + const stateKey = String(args['state-key'] ?? env.LOBSTER_RUN_STATE_KEY ?? '').trim() || null; + + const inputArtifacts = [] as any[]; + for await (const item of input) inputArtifacts.push(item); + + const normalizedArtifacts = [...inputArtifacts, ...providedArtifacts].map(normalizeArtifact); + const artifactHashes = normalizedArtifacts.map(hashArtifact); + + const cacheKey = computeCacheKey({ prompt, model, schemaVersion, artifactHashes, outputSchema: userOutputSchema }); + + if (stateKey && !forceRefresh) { + const stored = await readStateJson({ env, key: stateKey }).catch(() => null); + const reused = pickReusableState(stored, cacheKey); + if (reused) { + return { + output: streamOf(reused.items.map((item) => ({ ...item, source: 'run_state', cached: true }))), + }; + } + } + + if (!disableCache && !forceRefresh) { + const cache = await readCacheEntry(env, cacheKey); + if (cache) { + return { + output: streamOf(cache.items.map((item: any) => ({ ...item, source: 'cache', cached: true }))), + }; + } + } + + const payload: Record = { + prompt, + model, + artifacts: normalizedArtifacts, + artifactHashes, + }; + + if (metadataObject) payload.metadata = metadataObject; + if (userOutputSchema) payload.outputSchema = userOutputSchema; + if (schemaVersion) payload.schemaVersion = schemaVersion; + if (Number.isFinite(maxOutputTokens ?? NaN)) payload.maxOutputTokens = Number(maxOutputTokens); + if (Number.isFinite(temperature ?? NaN)) payload.temperature = Number(temperature); + + if (!validatePayload(payload)) { + throw new Error(`llm_task.invoke payload invalid: ${ajv.errorsText(validatePayload.errors)}`); + } + + const endpoint = buildEndpoint(baseUrl); + const token = String(args.token ?? env.LLM_TASK_TOKEN ?? '').trim(); + + const validator = userOutputSchema ? ajv.compile(userOutputSchema) : null; + + let attempt = 0; + let lastError: Error | null = null; + let lastValidationErrors: string[] = []; + + while (true) { + attempt++; + if (attempt > 1) { + payload.retryContext = { + attempt, + ...(lastValidationErrors.length ? { validationErrors: lastValidationErrors } : null), + }; + } else { + delete payload.retryContext; + } + + let responseEnvelope: LlmTaskResponseEnvelope; + try { + responseEnvelope = await invokeRemote({ endpoint, token, payload }); + } catch (err: any) { + throw new Error(`llm_task.invoke request failed: ${err?.message ?? String(err)}`); + } + + if (!validateResponseEnvelope(responseEnvelope)) { + throw new Error('llm_task.invoke received invalid response envelope'); + } + + if (responseEnvelope.ok !== true) { + const message = responseEnvelope.error?.message ?? 'llm-task returned an error'; + throw new Error(`llm_task.invoke remote error: ${message}`); + } + + const normalized = normalizeResult({ + envelope: responseEnvelope, + cacheKey, + schemaVersion, + artifactHashes, + source: 'remote', + attempt, + }); + + if (!validator) { + await persistOutputs({ env, stateKey, cacheKey, items: normalized }); + if (!disableCache) await writeCacheEntry(env, cacheKey, normalized); + return { output: streamOf(normalized) }; + } + + const structured = normalized[0]?.output?.data ?? null; + if (validator(structured)) { + await persistOutputs({ env, stateKey, cacheKey, items: normalized }); + if (!disableCache) await writeCacheEntry(env, cacheKey, normalized); + return { output: streamOf(normalized) }; + } + + lastValidationErrors = collectAjvErrors(validator.errors); + lastError = new Error(`llm_task.invoke output failed schema validation: ${lastValidationErrors.join('; ')}`); + if (attempt > maxValidationRetries) { + throw lastError; + } + } + }, +}; + +function extractPrompt(args) { + if (args.prompt) return String(args.prompt); + if (Array.isArray(args._) && args._.length) { + return args._.join(' '); + } + return ''; +} + +function parseJsonArray(raw, label) { + if (!raw) return []; + try { + const parsed = JSON.parse(String(raw)); + if (!Array.isArray(parsed)) throw new Error('must be array'); + return parsed; + } catch { + throw new Error(`${label} must be a JSON array`); + } +} + +function parseJsonObject(raw, label) { + if (!raw) return null; + try { + const parsed = JSON.parse(String(raw)); + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + throw new Error('must be an object'); + } + return parsed; + } catch { + throw new Error(`${label} must be a JSON object`); + } +} + +function parseOptionalNumber(value) { + if (value === undefined || value === null) return null; + const num = Number(value); + return Number.isFinite(num) ? num : null; +} + +function flag(value) { + if (value === undefined || value === null) return false; + if (typeof value === 'boolean') return value; + if (typeof value === 'string') { + const normalized = value.trim().toLowerCase(); + if (['false', '0', 'no'].includes(normalized)) return false; + if (['true', '1', 'yes'].includes(normalized)) return true; + } + return Boolean(value); +} + +function normalizeArtifact(raw) { + if (raw && typeof raw === 'object' && !Array.isArray(raw)) { + return raw; + } + if (typeof raw === 'string') { + return { kind: 'text', text: raw }; + } + if (typeof raw === 'number' || typeof raw === 'boolean') { + return { kind: 'text', text: String(raw) }; + } + return { kind: 'json', data: raw }; +} + +function hashArtifact(artifact) { + const stable = stableStringify(artifact); + return createHash('sha256').update(stable).digest('hex'); +} + +function computeCacheKey({ prompt, model, schemaVersion, artifactHashes, outputSchema }) { + const payload = { + prompt, + model, + schemaVersion, + artifactHashes, + outputSchema: outputSchema ?? null, + }; + return createHash('sha256').update(stableStringify(payload)).digest('hex'); +} + +function buildEndpoint(baseUrl: string) { + const base = new URL(baseUrl); + const cleanBase = base.pathname.endsWith('/') ? base.pathname.slice(0, -1) : base.pathname; + base.pathname = `${cleanBase}/tool/invoke`.replace(/\/+/g, '/'); + return base; +} + +async function invokeRemote({ + endpoint, + token, + payload, +}: { + endpoint: URL; + token: string; + payload: Record; +}): Promise { + const res = await fetch(endpoint, { + method: 'POST', + headers: { + 'content-type': 'application/json', + ...(token ? { authorization: `Bearer ${token}` } : null), + }, + body: JSON.stringify(payload), + }); + + const text = await res.text(); + if (!res.ok) { + throw new Error(`${res.status} ${res.statusText}: ${text.slice(0, 400)}`); + } + + try { + return text + ? (JSON.parse(text) as LlmTaskResponseEnvelope) + : ({ ok: true, result: {} as LlmTaskResponse } satisfies LlmTaskResponseEnvelope); + } catch { + throw new Error('Response was not JSON'); + } +} + +function normalizeResult({ + envelope, + cacheKey, + schemaVersion, + artifactHashes, + source, + attempt, +}: { + envelope: LlmTaskResponseEnvelope; + cacheKey: string; + schemaVersion: string; + artifactHashes: string[]; + source: string; + attempt: number; +}): NormalizedInvocationItem[] { + const result = envelope.result ?? {}; + const output = result.output ?? {}; + const item: NormalizedInvocationItem = { + kind: 'llm_task.invoke', + runId: result.runId ?? null, + prompt: result.prompt ?? null, + model: result.model ?? null, + schemaVersion, + status: result.status ?? 'completed', + cacheKey, + artifactHashes, + output: { + format: output.format ?? (output.data ? 'json' : 'text'), + text: output.text ?? null, + data: output.data ?? null, + }, + usage: (result.usage as Record) ?? null, + metadata: (result.metadata as Record) ?? null, + warnings: (result.warnings as string[]) ?? null, + diagnostics: (result.diagnostics as Record) ?? null, + createdAt: new Date().toISOString(), + source, + cached: source !== 'remote', + attemptCount: attempt, + }; + return [item]; +} + +async function persistOutputs({ + env, + stateKey, + cacheKey, + items, +}: { + env: Record; + stateKey: string | null; + cacheKey: string; + items: NormalizedInvocationItem[]; +}) { + if (!stateKey) return; + const record = { + type: 'llm_task.invoke', + version: STATE_VERSION, + cacheKey, + items, + storedAt: new Date().toISOString(), + }; + await writeStateJson({ env, key: stateKey, value: record }); +} + +function pickReusableState(stored: any, cacheKey: string) { + if (!stored || typeof stored !== 'object') return null; + if (stored.type !== 'llm_task.invoke') return null; + if (stored.cacheKey !== cacheKey) return null; + if (!Array.isArray(stored.items)) return null; + return { items: stored.items }; +} + +function collectAjvErrors(errors: ErrorObject[] | null | undefined) { + if (!errors?.length) return []; + return errors.map((err) => `${err.instancePath || '/'} ${err.message ?? ''}`.trim()); +} + +async function readCacheEntry(env: Record, key: string): Promise { + const filePath = path.join(getCacheDir(env), 'llm_task.invoke', `${key}.json`); + try { + const text = await fsp.readFile(filePath, 'utf8'); + return JSON.parse(text) as CacheEntry; + } catch (err: any) { + if (err?.code === 'ENOENT') return null; + throw err; + } +} + +async function writeCacheEntry( + env: Record, + key: string, + items: NormalizedInvocationItem[], +) { + const dir = path.join(getCacheDir(env), 'llm_task.invoke'); + await fsp.mkdir(dir, { recursive: true }); + const filePath = path.join(dir, `${key}.json`); + await fsp.writeFile(filePath, JSON.stringify({ items, cacheKey: key, storedAt: new Date().toISOString() }, null, 2)); +} + +function getCacheDir(env: Record) { + if (env?.LOBSTER_CACHE_DIR) return env.LOBSTER_CACHE_DIR; + return path.join(process.cwd(), '.lobster-cache'); +} + +async function* streamOf(items: NormalizedInvocationItem[]) { + for (const item of items) { + yield item; + } +} diff --git a/test/llm_task_invoke.test.ts b/test/llm_task_invoke.test.ts new file mode 100644 index 0000000..0341f1e --- /dev/null +++ b/test/llm_task_invoke.test.ts @@ -0,0 +1,276 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import http from 'node:http'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; + +import { createDefaultRegistry } from '../src/commands/registry.js'; + +function streamOf(items: any[]) { + return (async function* () { + for (const item of items) yield item; + })(); +} + +async function collect(iterable: AsyncIterable) { + const items = []; + for await (const item of iterable) items.push(item); + return items; +} + +test('llm_task.invoke posts to /tool/invoke and normalizes result', async () => { + const registry = createDefaultRegistry(); + const cmd = registry.get('llm_task.invoke'); + assert.ok(cmd, 'llm_task.invoke should be registered'); + const cacheDir = await mkdtemp(path.join(tmpdir(), 'lobster-cache-')); + + const bodyLog: any[] = []; + const server = http.createServer((req, res) => { + if (req.method !== 'POST' || req.url !== '/tool/invoke') { + res.writeHead(404); + res.end('nope'); + return; + } + let buf = ''; + req.setEncoding('utf8'); + req.on('data', (d) => (buf += d)); + req.on('end', () => { + const parsed = JSON.parse(buf || '{}'); + bodyLog.push(parsed); + res.writeHead(200, { 'content-type': 'application/json' }); + res.end( + JSON.stringify({ + ok: true, + result: { + runId: 'task_1', + model: parsed.model, + prompt: parsed.prompt, + output: { + text: 'done', + data: { summary: 'hello world' }, + }, + usage: { inputTokens: 12, outputTokens: 2, totalTokens: 14 }, + }, + }), + ); + }); + }); + + await new Promise((resolve) => server.listen(0, resolve)); + const addr = server.address(); + const port = typeof addr === 'object' && addr ? addr.port : 0; + + try { + const result = await cmd.run({ + input: streamOf([{ kind: 'text', text: 'doc' }]), + args: { + _: [], + url: `http://127.0.0.1:${port}`, + token: 'test-token', + model: 'claude-3-sonnet', + prompt: 'Summarize', + }, + ctx: baseCtx({ LOBSTER_CACHE_DIR: cacheDir }, registry), + } as any); + + const items = await collect(result.output!); + assert.equal(items.length, 1); + const payload = items[0]; + assert.equal(payload.kind, 'llm_task.invoke'); + assert.equal(payload.runId, 'task_1'); + assert.equal(payload.output.data.summary, 'hello world'); + assert.equal(payload.model, 'claude-3-sonnet'); + assert.equal(payload.source, 'remote'); + assert.equal(payload.cached, false); + assert.ok(payload.cacheKey); + + assert.equal(bodyLog.length, 1); + assert.equal(bodyLog[0].prompt, 'Summarize'); + assert.equal(bodyLog[0].model, 'claude-3-sonnet'); + assert.equal(bodyLog[0].artifacts.length, 1); + assert.equal(bodyLog[0].artifactHashes.length, 1); + } finally { + await rm(cacheDir, { recursive: true, force: true }); + await closeServer(server); + } +}); + +test('llm_task.invoke retries when schema validation fails', async () => { + const registry = createDefaultRegistry(); + const cmd = registry.get('llm_task.invoke'); + assert.ok(cmd); + const cacheDir = await mkdtemp(path.join(tmpdir(), 'lobster-cache-')); + + let calls = 0; + const server = http.createServer((req, res) => { + if (req.method !== 'POST') { + res.writeHead(404); + res.end(); + return; + } + calls += 1; + const valid = calls >= 2; + const payload = { + ok: true, + result: { + runId: `attempt_${calls}`, + output: valid ? { data: { decision: 'send' } } : { data: { foo: 'bar' } }, + }, + }; + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify(payload)); + }); + + await new Promise((resolve) => server.listen(0, resolve)); + const addr = server.address(); + const port = typeof addr === 'object' && addr ? addr.port : 0; + + try { + const result = await cmd.run({ + input: streamOf([]), + args: { + _: [], + url: `http://127.0.0.1:${port}`, + model: 'claude-3-opus', + prompt: 'Decide', + 'output-schema': '{"type":"object","required":["decision"]}', + 'max-validation-retries': 2, + }, + ctx: baseCtx({ LOBSTER_CACHE_DIR: cacheDir }, registry), + } as any); + + const items = await collect(result.output!); + assert.equal(items.length, 1); + assert.equal(items[0].runId, 'attempt_2'); + assert.equal(items[0].output.data.decision, 'send'); + assert.equal(calls, 2); + } finally { + await rm(cacheDir, { recursive: true, force: true }); + await closeServer(server); + } +}); + +test('llm_task.invoke persists to run state so resume skips remote call', async () => { + const stateDir = await mkdtemp(path.join(tmpdir(), 'lobster-state-')); + const registry = createDefaultRegistry(); + const cmd = registry.get('llm_task.invoke'); + assert.ok(cmd); + + const server = http.createServer((req, res) => { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true, result: { runId: 'state_run', output: { data: { ok: true } } } })); + }); + await new Promise((resolve) => server.listen(0, resolve)); + const addr = server.address(); + const port = typeof addr === 'object' && addr ? addr.port : 0; + + const cacheDir = await mkdtemp(path.join(tmpdir(), 'lobster-cache-')); + const ctxEnv = { LOBSTER_STATE_DIR: stateDir, LOBSTER_CACHE_DIR: cacheDir }; + + try { + const first = await cmd.run({ + input: streamOf([{ foo: 'bar' }]), + args: { + _: [], + url: `http://127.0.0.1:${port}`, + model: 'claude', + prompt: 'Do thing', + 'state-key': 'run123', + }, + ctx: baseCtx(ctxEnv, registry), + } as any); + const firstItems = await collect(first.output!); + assert.equal(firstItems[0].source, 'remote'); + + await closeServer(server); + + const second = await cmd.run({ + input: streamOf([{ foo: 'bar' }]), + args: { + _: [], + url: `http://127.0.0.1:${port}`, + model: 'claude', + prompt: 'Do thing', + 'state-key': 'run123', + }, + ctx: baseCtx(ctxEnv, registry), + } as any); + const secondItems = await collect(second.output!); + assert.equal(secondItems.length, 1); + assert.equal(secondItems[0].source, 'run_state'); + } finally { + await rm(stateDir, { recursive: true, force: true }); + await rm(cacheDir, { recursive: true, force: true }); + await closeServer(server); + } +}); + +test('llm_task.invoke reuses file cache when URL unavailable', async () => { + const cacheDir = await mkdtemp(path.join(tmpdir(), 'lobster-cache-')); + const registry = createDefaultRegistry(); + const cmd = registry.get('llm_task.invoke'); + assert.ok(cmd); + + const server = http.createServer((req, res) => { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ ok: true, result: { runId: 'cache_run', output: { text: 'cached' } } })); + }); + await new Promise((resolve) => server.listen(0, resolve)); + const addr = server.address(); + const port = typeof addr === 'object' && addr ? addr.port : 0; + + const ctxEnv = { LOBSTER_CACHE_DIR: cacheDir }; + + try { + const first = await cmd.run({ + input: streamOf([]), + args: { + _: [], + url: `http://127.0.0.1:${port}`, + model: 'claude', + prompt: 'Cache me', + }, + ctx: baseCtx(ctxEnv, registry), + } as any); + const firstItems = await collect(first.output!); + assert.equal(firstItems[0].source, 'remote'); + + await closeServer(server); + + const second = await cmd.run({ + input: streamOf([]), + args: { + _: [], + url: `http://127.0.0.1:${port}`, + model: 'claude', + prompt: 'Cache me', + }, + ctx: baseCtx(ctxEnv, registry), + } as any); + const secondItems = await collect(second.output!); + assert.equal(secondItems.length, 1); + assert.equal(secondItems[0].source, 'cache'); + assert.equal(secondItems[0].cached, true); + } finally { + await rm(cacheDir, { recursive: true, force: true }); + await closeServer(server); + } +}); + +function baseCtx(envOverrides: Record, registry?) { + return { + stdin: process.stdin, + stdout: process.stdout, + stderr: process.stderr, + env: { ...process.env, ...envOverrides }, + registry: registry ?? null, + mode: 'tool', + render: { json() {}, lines() {} }, + }; +} + +async function closeServer(server: http.Server) { + if (!server.listening) return; + await new Promise((resolve) => server.close(() => resolve())); +} diff --git a/test/workflow_file.test.ts b/test/workflow_file.test.ts index c29476f..aad7b35 100644 --- a/test/workflow_file.test.ts +++ b/test/workflow_file.test.ts @@ -35,16 +35,19 @@ test('workflow file runs with approval and resume', async () => { }; const tmpDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'lobster-workflow-')); + const stateDir = path.join(tmpDir, 'state'); const filePath = path.join(tmpDir, 'workflow.lobster'); await fsp.writeFile(filePath, JSON.stringify(workflow, null, 2), 'utf8'); + const env = { ...process.env, LOBSTER_STATE_DIR: stateDir }; + const first = await runWorkflowFile({ filePath, ctx: { stdin: process.stdin, stdout: process.stdout, stderr: process.stderr, - env: process.env, + env, mode: 'tool', }, }); @@ -62,7 +65,7 @@ test('workflow file runs with approval and resume', async () => { stdin: process.stdin, stdout: process.stdout, stderr: process.stderr, - env: process.env, + env, mode: 'tool', }, resume: payload,