feat(stdlib): add file I/O and jq-filter commands

.gitignore

@@ -15,6 +15,9 @@ clawdbot_enhancement.md
 # OS files
 .DS_Store
 
+# Package tarballs
+*.tgz
+
 # Logs
 *.log
 

src/cli.ts

@@ -466,5 +466,5 @@ function helpText() {
     `  lobster 'exec --json "echo [1,2,3]" | json'\n` +
     `  lobster run --mode tool 'exec --json "echo [1]" | approve --prompt "ok?"'\n\n` +
     `Commands:\n` +
-    `  exec, head, json, pick, table, where, approve, clawd.invoke, state.get, state.set, diff.last, commands.list, workflows.list, workflows.run\n`;
+    `  exec, file.read, file.write, head, jq.filter, json, pick, table, where, approve, clawd.invoke, state.get, state.set, diff.last, commands.list, workflows.list, workflows.run\n`;
 }

src/commands/registry.ts

@@ -13,6 +13,9 @@ 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 { fileReadCommand } from "./stdlib/file_read.js";
+import { fileWriteCommand } from "./stdlib/file_write.js";
+import { jqFilterCommand } from "./stdlib/jq_filter.js";
 import { diffLastCommand } from "./stdlib/diff_last.js";
 import { workflowsListCommand } from "./workflows/workflows_list.js";
 import { workflowsRunCommand } from "./workflows/workflows_run.js";
@@ -41,6 +44,9 @@ export function createDefaultRegistry() {
     llmTaskInvokeCommand,
     stateGetCommand,
     stateSetCommand,
+    fileReadCommand,
+    fileWriteCommand,
+    jqFilterCommand,
     diffLastCommand,
     workflowsListCommand,
     workflowsRunCommand,

src/commands/stdlib/file_read.ts

@@ -0,0 +1,86 @@
+import { promises as fsp } from 'node:fs';
+import { resolve, isAbsolute } from 'node:path';
+
+export const fileReadCommand = {
+  name: 'file.read',
+  meta: {
+    description: 'Read a file and yield its contents into the pipeline',
+    argsSchema: {
+      type: 'object',
+      properties: {
+        _: { type: 'array', items: { type: 'string' }, description: 'File path' },
+        path: { type: 'string', description: 'File path (alternative to positional)' },
+        format: { type: 'string', enum: ['auto', 'text', 'json', 'jsonl'], description: 'Parse format (default: auto)' },
+      },
+      required: ['_'],
+    },
+    sideEffects: ['reads_fs'],
+  },
+  help() {
+    return `file.read — read a file and yield its contents\n\n` +
+      `Usage:\n` +
+      `  file.read <path> [--format auto|text|json|jsonl]\n\n` +
+      `Formats:\n` +
+      `  auto (default): try JSON parse; if array yield elements; else try JSONL; else text\n` +
+      `  json:  parse as JSON; yield elements if array, else single item\n` +
+      `  jsonl: split lines, parse each as JSON\n` +
+      `  text:  yield entire content as a single string\n\n` +
+      `Notes:\n` +
+      `  - Replaces the pipeline stream; upstream items are discarded.\n\n` +
+      `Security:\n` +
+      `  Paths are unrestricted (same as exec). This command can read any file\n` +
+      `  accessible to the process.\n`;
+  },
+  async run({ input, args }) {
+    // Drain input (file replaces pipeline input).
+    for await (const _item of input) { /* no-op */ }
+
+    const filePath = args._[0] || args.path;
+    if (!filePath) throw new Error('file.read requires a path');
+
+    const resolved = isAbsolute(filePath) ? filePath : resolve(process.cwd(), filePath);
+    const format = (args.format ?? 'auto').toLowerCase();
+    const VALID_FORMATS = ['auto', 'text', 'json', 'jsonl'];
+    if (!VALID_FORMATS.includes(format)) {
+      throw new Error(`file.read: unknown format '${format}'`);
+    }
+
+    const MAX_FILE_SIZE = 50 * 1024 * 1024; // 50 MB
+    const stat = await fsp.stat(resolved);
+    if (stat.size > MAX_FILE_SIZE) {
+      throw new Error(`file.read: file exceeds maximum size (${stat.size} bytes > ${MAX_FILE_SIZE} bytes)`);
+    }
+    const content = await fsp.readFile(resolved, 'utf8');
+
+    if (format === 'text') {
+      return { output: asStream([content]) };
+    }
+
+    if (format === 'json') {
+      const parsed = JSON.parse(content);
+      return { output: asStream(Array.isArray(parsed) ? parsed : [parsed]) };
+    }
+
+    if (format === 'jsonl') {
+      const items = content.split(/\r?\n/).filter(Boolean).map((line) => JSON.parse(line));
+      return { output: asStream(items) };
+    }
+
+    // auto: try JSON, then JSONL, then text.
+    try {
+      const parsed = JSON.parse(content);
+      return { output: asStream(Array.isArray(parsed) ? parsed : [parsed]) };
+    } catch { /* not JSON */ }
+
+    const lines = content.split(/\r?\n/).filter(Boolean);
+    if (lines.length > 0 && lines.every((line) => { try { JSON.parse(line); return true; } catch { return false; } })) {
+      return { output: asStream(lines.map((line) => JSON.parse(line))) };
+    }
+
+    return { output: asStream([content]) };
+  },
+};
+
+async function* asStream(items) {
+  for (const item of items) yield item;
+}

src/commands/stdlib/file_write.ts

@@ -0,0 +1,70 @@
+import { promises as fsp } from 'node:fs';
+import { resolve, isAbsolute, dirname } from 'node:path';
+
+export const fileWriteCommand = {
+  name: 'file.write',
+  meta: {
+    description: 'Write pipeline items to a file and pass them through',
+    argsSchema: {
+      type: 'object',
+      properties: {
+        _: { type: 'array', items: { type: 'string' }, description: 'File path' },
+        path: { type: 'string', description: 'File path (alternative to positional)' },
+        format: { type: 'string', enum: ['json', 'jsonl', 'text'], description: 'Output format (default: json)' },
+        mkdir: { type: 'boolean', description: 'Create parent directories (default: true)' },
+      },
+      required: ['_'],
+    },
+    sideEffects: ['writes_fs'],
+  },
+  help() {
+    return `file.write — write pipeline items to a file\n\n` +
+      `Usage:\n` +
+      `  <items> | file.write <path> [--format json|jsonl|text] [--mkdir true|false]\n\n` +
+      `Formats:\n` +
+      `  json (default): JSON with 2-space indent; single item unwrapped, multiple as array\n` +
+      `  jsonl: one JSON-serialized item per line\n` +
+      `  text:  items joined with newline; non-strings JSON-serialized\n\n` +
+      `Notes:\n` +
+      `  - Tee semantics: all collected items are yielded downstream after write.\n` +
+      `  - --mkdir (default true) creates parent directories if needed.\n\n` +
+      `Security:\n` +
+      `  Paths are unrestricted (same as exec). This command can write to any path\n` +
+      `  accessible to the process.\n`;
+  },
+  async run({ input, args }) {
+    const filePath = args._[0] || args.path;
+    if (!filePath) throw new Error('file.write requires a path');
+
+    const resolved = isAbsolute(filePath) ? filePath : resolve(process.cwd(), filePath);
+    const format = (args.format ?? 'json').toLowerCase();
+    const mkdirEnabled = args.mkdir !== false;
+
+    const items = [];
+    for await (const item of input) items.push(item);
+
+    let content;
+    if (format === 'json') {
+      const value = items.length === 1 ? items[0] : items;
+      content = JSON.stringify(value, null, 2) + '\n';
+    } else if (format === 'jsonl') {
+      content = items.map((item) => JSON.stringify(item)).join('\n') + (items.length ? '\n' : '');
+    } else if (format === 'text') {
+      content = items.map((item) => (typeof item === 'string' ? item : JSON.stringify(item))).join('\n') + (items.length ? '\n' : '');
+    } else {
+      throw new Error(`file.write: unknown format '${format}'`);
+    }
+
+    if (mkdirEnabled) {
+      await fsp.mkdir(dirname(resolved), { recursive: true });
+    }
+
+    await fsp.writeFile(resolved, content, 'utf8');
+
+    return { output: asStream(items) };
+  },
+};
+
+async function* asStream(items) {
+  for (const item of items) yield item;
+}

src/commands/stdlib/jq_filter.ts

@@ -0,0 +1,84 @@
+import { spawn } from 'node:child_process';
+
+export const jqFilterCommand = {
+  name: 'jq.filter',
+  meta: {
+    description: 'Apply a jq expression to each pipeline item',
+    argsSchema: {
+      type: 'object',
+      properties: {
+        _: { type: 'array', items: { type: 'string' }, description: 'jq expression' },
+        expr: { type: 'string', description: 'jq expression (alternative to positional)' },
+        raw: { type: 'boolean', description: 'output raw strings instead of JSON (like jq -r)' },
+      },
+      required: ['_'],
+    },
+    sideEffects: ['local_exec'],
+  },
+  help() {
+    return `jq.filter — apply a jq expression to each pipeline item\n\n` +
+      `Usage:\n` +
+      `  <items> | jq.filter <expr>\n` +
+      `  <items> | jq.filter --expr <expr>\n` +
+      `  <items> | jq.filter --raw <expr>\n\n` +
+      `Options:\n` +
+      `  --raw    Output raw strings instead of JSON (passes -r to jq).\n\n` +
+      `Notes:\n` +
+      `  - Each input item is serialized as JSON and piped to jq -c <expr>.\n` +
+      `  - Each non-empty stdout line is parsed as JSON and yielded.\n` +
+      `  - With --raw, stdout lines are yielded as plain strings (no JSON parse).\n` +
+      `  - Requires jq on PATH.\n`;
+  },
+  async run({ input, args }) {
+    const expr = args._[0] || args.expr;
+    if (!expr) throw new Error('jq.filter requires an expression');
+
+    const raw = Boolean(args.raw);
+    const results = [];
+    for await (const item of input) {
+      const itemJson = JSON.stringify(item);
+      const output = await runJq(expr, itemJson, raw);
+      const lines = output.split(/\r?\n/).filter(Boolean);
+      for (const line of lines) {
+        results.push(raw ? line : JSON.parse(line));
+      }
+    }
+
+    return { output: asStream(results) };
+  },
+};
+
+function runJq(expr, stdin, raw = false) {
+  return new Promise<string>((resolve, reject) => {
+    const jqArgs = ['-c', ...(raw ? ['-r'] : []), expr];
+    const child = spawn('jq', jqArgs, {
+      stdio: ['pipe', 'pipe', 'pipe'],
+      env: { PATH: process.env.PATH || '' },
+    });
+
+    let stdout = '';
+    let stderr = '';
+
+    child.stdout.setEncoding('utf8');
+    child.stderr.setEncoding('utf8');
+
+    child.stdout.on('data', (d) => { stdout += d; });
+    child.stderr.on('data', (d) => { stderr += d; });
+
+    child.stdin.setDefaultEncoding('utf8');
+    child.stdin.write(stdin);
+    child.stdin.end();
+
+    child.on('error', (err) => {
+      reject(new Error(`jq.filter: failed to spawn jq: ${err.message}`));
+    });
+    child.on('close', (code) => {
+      if (code === 0) return resolve(stdout);
+      reject(new Error(`jq.filter failed (exit ${code}): ${stderr.trim() || 'unknown error'}`));
+    });
+  });
+}
+
+async function* asStream(items) {
+  for (const item of items) yield item;
+}

src/commands/stdlib/llm_task_invoke.ts

@@ -146,8 +146,6 @@ type CacheEntry = {
   storedAt: string;
 };
 
-type Transport = 'clawd';
-
 export const llmTaskInvokeCommand = {
   name: 'llm_task.invoke',
   meta: {
@@ -198,7 +196,6 @@ export const llmTaskInvokeCommand = {
     const env = ctx.env ?? process.env;
 
     const clawdUrl = String(env.CLAWD_URL ?? '').trim();
-    const transport: Transport = 'clawd';
     if (!clawdUrl) {
       throw new Error('llm_task.invoke requires CLAWD_URL (run via Clawdbot gateway)');
     }