feat: add llm_task.invoke primitive

.gitignore

@@ -6,6 +6,8 @@ dist/
 
 # Lobster state
 .lobster/
+.lobster-cache/
+.pnpm-store/
 
 # Private docs
 clawdbot_enhancement.md

package.json

@@ -55,6 +55,7 @@
   "homepage": "https://github.com/clawdbot/lobster#readme",
   "license": "MIT",
   "dependencies": {
+    "ajv": "^8.17.1",
     "yaml": "^2.8.2"
   }
 }

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: {}

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,

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<string, unknown> | null;
+  warnings?: string[] | null;
+  metadata?: Record<string, unknown> | null;
+  diagnostics?: Record<string, unknown> | 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<string, unknown> | null;
+  metadata: Record<string, unknown> | null;
+  warnings: string[] | null;
+  diagnostics: Record<string, unknown> | 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<string, any> = {
+      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<string, unknown>;
+}): Promise<LlmTaskResponseEnvelope> {
+  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<string, unknown>) ?? null,
+    metadata: (result.metadata as Record<string, unknown>) ?? null,
+    warnings: (result.warnings as string[]) ?? null,
+    diagnostics: (result.diagnostics as Record<string, unknown>) ?? null,
+    createdAt: new Date().toISOString(),
+    source,
+    cached: source !== 'remote',
+    attemptCount: attempt,
+  };
+  return [item];
+}
+
+async function persistOutputs({
+  env,
+  stateKey,
+  cacheKey,
+  items,
+}: {
+  env: Record<string, string | undefined>;
+  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<string, string | undefined>, key: string): Promise<CacheEntry | null> {
+  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<string, string | undefined>,
+  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<string, string | undefined>) {
+  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;
+  }
+}

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<any>) {
+  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<void>((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<void>((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<void>((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<void>((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<string, string>, 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<void>((resolve) => server.close(() => resolve()));
+}

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,