mirror of
https://github.com/openclaw/lobster.git
synced 2026-08-14 00:48:09 +00:00
Initial Lobster MVP: typed pipelines, approvals, gog Gmail
This commit is contained in:
Vignesh Natarajan
@@ -0,0 +1,39 @@
|
||||
# Lobster
|
||||
|
||||
A Clawdbot-native workflow shell: typed (JSON-first) pipelines, jobs, and approval gates.
|
||||
|
||||
This repo is an MVP scaffold focused on the core shell runtime and a first Gmail integration via the `steipete/gog` skill/CLI.
|
||||
|
||||
## Goals
|
||||
|
||||
- Typed pipelines (objects/arrays), not text pipes.
|
||||
- Local-first execution.
|
||||
- No new auth surface: Lobster must not own OAuth/tokens.
|
||||
- Composable macros that Clawdbot can invoke in one step to save tokens.
|
||||
|
||||
## Quick start
|
||||
|
||||
From this folder:
|
||||
|
||||
- `node ./bin/lobster.js --help`
|
||||
- `node ./bin/lobster.js "exec --json 'echo [1,2,3]' | where '0>=0' | json"`
|
||||
|
||||
If you have `gog` installed:
|
||||
|
||||
- `node ./bin/lobster.js "gog.gmail.search --query 'newer_than:7d' --max 5 | table"`
|
||||
|
||||
## Commands
|
||||
|
||||
- `exec`: run OS commands
|
||||
- `gog.gmail.search`: fetch Gmail search results via `gog`
|
||||
- `gog.gmail.send`: send email via `gog` (use approval gates)
|
||||
- `email.triage`: deterministic triage report (rule-based)
|
||||
- `where`, `pick`, `head`: data shaping
|
||||
- `json`, `table`: renderers
|
||||
- `approve`: approval gate (TTY prompt or `--emit` for Clawdbot integration)
|
||||
|
||||
## Next steps
|
||||
|
||||
- Canonical `EmailMessage` schema (normalize gog output predictably).
|
||||
- `email.draft` + `email.send` macros (compose approvals cleanly).
|
||||
- Clawdbot integration: expose Lobster as a first-class tool (`lobster.run`).
|
||||
Executable
+4
@@ -0,0 +1,4 @@
|
||||
#!/usr/bin/env node
|
||||
import { runCli } from '../src/cli.js';
|
||||
|
||||
await runCli(process.argv.slice(2));
|
||||
@@ -0,0 +1,57 @@
|
||||
# Lobster design (draft)
|
||||
|
||||
## What Lobster is
|
||||
|
||||
Lobster is a Clawdbot-native workflow shell.
|
||||
|
||||
- JSON-first typed pipelines (records/arrays) instead of byte streams.
|
||||
- Deterministic composition of tools/skills into reusable macros.
|
||||
- Human-in-the-loop approval gates as language primitives.
|
||||
- Local-first: Lobster should talk to a local Clawdbot runtime and/or local CLIs.
|
||||
|
||||
## What Lobster is not
|
||||
|
||||
- Not a terminal emulator.
|
||||
- Not a POSIX-compatible shell (at least initially).
|
||||
- Not an auth broker: Lobster must not store OAuth tokens.
|
||||
|
||||
## Why it exists
|
||||
|
||||
- Turns multi-step tool orchestration into a single `lobster.run(...)` call.
|
||||
- Saves tokens by moving deterministic orchestration out of the LLM.
|
||||
- Makes automation auditable and safe-by-default.
|
||||
|
||||
## Data model
|
||||
|
||||
- A pipeline is a list of stages.
|
||||
- Each stage consumes an async stream of items and produces an async stream.
|
||||
- Items are arbitrary JSON values, but common shapes should be standardized (e.g. EmailMessage).
|
||||
|
||||
## Safety model
|
||||
|
||||
- Commands declare capabilities (e.g. `email.read`, `email.send`, `fs.write`).
|
||||
- Approval gates must fail closed in non-interactive mode.
|
||||
- When integrated into Clawdbot, approvals are surfaced to the user by Clawdbot.
|
||||
|
||||
## Clawdbot integration (target)
|
||||
|
||||
Preferred: Lobster does not run `gog` directly. Instead it calls Clawdbot tools.
|
||||
|
||||
- Clawdbot exposes `gog` (or `google`) as a tool.
|
||||
- Lobster calls `tools.invoke({ tool: 'gog', action: 'gmail.search', ...})`.
|
||||
|
||||
In the MVP, `gog.gmail.search` shells out to `gog` for fast iteration.
|
||||
|
||||
## MVP scope
|
||||
|
||||
- Parser for `cmd ... | cmd ...` pipelines.
|
||||
- A minimal standard library (where/pick/head/json/table).
|
||||
- An interactive `approve` primitive.
|
||||
- A first Gmail read primitive via `gog.gmail.search`.
|
||||
|
||||
## Next milestones
|
||||
|
||||
1. `email.normalize`: normalize gog output to `EmailMessage`.
|
||||
2. `email.triage`: classify + draft + propose label actions.
|
||||
3. Non-interactive approvals: emit `requiresApproval` objects instead of prompting.
|
||||
4. Clawdbot tool bridge: replace `gog` exec with tool invocation.
|
||||
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"name": "lobster-shell",
|
||||
"version": "0.1.0",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"bin": {
|
||||
"lobster": "./bin/lobster.js"
|
||||
},
|
||||
"scripts": {
|
||||
"test": "node --test"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=20"
|
||||
}
|
||||
}
|
||||
+74
@@ -0,0 +1,74 @@
|
||||
import { parsePipeline } from './parser.js';
|
||||
import { createDefaultRegistry } from './commands/registry.js';
|
||||
import { runPipeline } from './runtime.js';
|
||||
|
||||
export async function runCli(argv) {
|
||||
const registry = createDefaultRegistry();
|
||||
|
||||
if (argv.length === 0 || argv.includes('-h') || argv.includes('--help')) {
|
||||
process.stdout.write(helpText());
|
||||
return;
|
||||
}
|
||||
|
||||
if (argv[0] === 'help') {
|
||||
const topic = argv[1];
|
||||
if (!topic) {
|
||||
process.stdout.write(helpText());
|
||||
return;
|
||||
}
|
||||
const cmd = registry.get(topic);
|
||||
if (!cmd) {
|
||||
process.stderr.write(`Unknown command: ${topic}\n`);
|
||||
process.exitCode = 2;
|
||||
return;
|
||||
}
|
||||
process.stdout.write(cmd.help());
|
||||
return;
|
||||
}
|
||||
|
||||
const inputLine = argv.join(' ');
|
||||
let pipeline;
|
||||
try {
|
||||
pipeline = parsePipeline(inputLine);
|
||||
} catch (err) {
|
||||
process.stderr.write(`Parse error: ${err?.message ?? String(err)}\n`);
|
||||
process.exitCode = 2;
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const output = await runPipeline({
|
||||
pipeline,
|
||||
registry,
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env: process.env,
|
||||
});
|
||||
|
||||
// Default rendering: if the last command didn't render, print JSON.
|
||||
if (!output.rendered) {
|
||||
process.stdout.write(JSON.stringify(output.items, null, 2));
|
||||
process.stdout.write('\n');
|
||||
}
|
||||
} catch (err) {
|
||||
process.stderr.write(`Error: ${err?.message ?? String(err)}\n`);
|
||||
process.exitCode = 1;
|
||||
}
|
||||
}
|
||||
|
||||
function helpText() {
|
||||
return `lobster (v0.1) — Clawdbot-native typed shell\n\n` +
|
||||
`Usage:\n` +
|
||||
` lobster '<pipeline>'\n` +
|
||||
` lobster help <command>\n\n` +
|
||||
`Pipeline basics:\n` +
|
||||
` - Commands are piped with |\n` +
|
||||
` - Data is JSON-first (arrays/objects), not text-first\n` +
|
||||
` - Most commands accept --flag value or --flag=value\n\n` +
|
||||
`Examples:\n` +
|
||||
` lobster 'exec --json "echo [1,2,3]" | json'\n` +
|
||||
` lobster 'gog.gmail.search --query "newer_than:7d" --max 10 | pick id,subject,from | table'\n` +
|
||||
`\nCommands:\n` +
|
||||
` exec, head, json, pick, table, where, approve, gog.gmail.search, gog.gmail.send, email.triage\n`;
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
import { execCommand } from './stdlib/exec.js';
|
||||
import { headCommand } from './stdlib/head.js';
|
||||
import { jsonCommand } from './stdlib/json.js';
|
||||
import { pickCommand } from './stdlib/pick.js';
|
||||
import { tableCommand } from './stdlib/table.js';
|
||||
import { whereCommand } from './stdlib/where.js';
|
||||
import { approveCommand } from './stdlib/approve.js';
|
||||
import { gogGmailSearchCommand } from './stdlib/gog_gmail_search.js';
|
||||
import { gogGmailSendCommand } from './stdlib/gog_gmail_send.js';
|
||||
import { emailTriageCommand } from './stdlib/email_triage.js';
|
||||
|
||||
export function createDefaultRegistry() {
|
||||
const commands = new Map();
|
||||
|
||||
for (const cmd of [
|
||||
execCommand,
|
||||
headCommand,
|
||||
jsonCommand,
|
||||
pickCommand,
|
||||
tableCommand,
|
||||
whereCommand,
|
||||
approveCommand,
|
||||
gogGmailSearchCommand,
|
||||
gogGmailSendCommand,
|
||||
emailTriageCommand,
|
||||
]) {
|
||||
commands.set(cmd.name, cmd);
|
||||
}
|
||||
|
||||
return {
|
||||
get(name) {
|
||||
return commands.get(name);
|
||||
},
|
||||
list() {
|
||||
return [...commands.keys()].sort();
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
function isInteractive(stdin) {
|
||||
return Boolean(stdin.isTTY);
|
||||
}
|
||||
|
||||
export const approveCommand = {
|
||||
name: 'approve',
|
||||
help() {
|
||||
return `approve — require confirmation to continue\n\nUsage:\n ... | approve --prompt "Send these emails?"\n ... | approve --emit --prompt "Send these emails?"\n\nModes:\n - Interactive (default): prompts on TTY and passes items through if approved.\n - Emit (--emit): returns an approval request object and stops the pipeline.\n\nNotes:\n - Default behavior fails closed in non-interactive mode.\n`;
|
||||
},
|
||||
async run({ input, args, ctx }) {
|
||||
const prompt = args.prompt ?? 'Approve?';
|
||||
|
||||
const items = [];
|
||||
for await (const item of input) items.push(item);
|
||||
|
||||
if (args.emit) {
|
||||
return {
|
||||
output: (async function* () {
|
||||
yield {
|
||||
type: 'approval_request',
|
||||
prompt,
|
||||
items,
|
||||
};
|
||||
})(),
|
||||
};
|
||||
}
|
||||
|
||||
if (!isInteractive(ctx.stdin)) {
|
||||
throw new Error('approve requires an interactive TTY (or pass --emit)');
|
||||
}
|
||||
|
||||
ctx.stdout.write(`${prompt} [y/N] `);
|
||||
const answer = await readLine(ctx.stdin);
|
||||
|
||||
if (!/^y(es)?$/i.test(String(answer).trim())) {
|
||||
throw new Error('Not approved');
|
||||
}
|
||||
|
||||
return { output: asStream(items) };
|
||||
},
|
||||
};
|
||||
|
||||
function readLine(stdin) {
|
||||
return new Promise((resolve) => {
|
||||
let buf = '';
|
||||
|
||||
const onData = (chunk) => {
|
||||
buf += chunk.toString('utf8');
|
||||
const idx = buf.indexOf('\n');
|
||||
if (idx !== -1) {
|
||||
stdin.off('data', onData);
|
||||
resolve(buf.slice(0, idx));
|
||||
}
|
||||
};
|
||||
|
||||
stdin.on('data', onData);
|
||||
});
|
||||
}
|
||||
|
||||
async function* asStream(items) {
|
||||
for (const item of items) yield item;
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
function getField(obj, path, fallback = undefined) {
|
||||
if (!obj || typeof obj !== 'object') return fallback;
|
||||
if (!path) return fallback;
|
||||
const parts = String(path).split('.');
|
||||
let cur = obj;
|
||||
for (const p of parts) {
|
||||
if (!cur || typeof cur !== 'object') return fallback;
|
||||
cur = cur[p];
|
||||
}
|
||||
return cur ?? fallback;
|
||||
}
|
||||
|
||||
function normalizeString(v) {
|
||||
if (v === null || v === undefined) return '';
|
||||
return String(v);
|
||||
}
|
||||
|
||||
function parseEmailAddress(from) {
|
||||
// Handles "Name <email@x.com>" or "email@x.com".
|
||||
const s = String(from ?? '').trim();
|
||||
const m = s.match(/<([^>]+)>/);
|
||||
return (m ? m[1] : s).trim();
|
||||
}
|
||||
|
||||
function classifyEmail({ subject, snippet }) {
|
||||
const text = `${subject} ${snippet}`.toLowerCase();
|
||||
if (/(unsubscribe|newsletter|promo|sale|discount)/.test(text)) return { bucket: 'fyi', reason: 'newsletter/promo-ish' };
|
||||
if (/(invoice|receipt|payment|charged|billing)/.test(text)) return { bucket: 'needs_action', reason: 'finance keyword' };
|
||||
if (/(asap|urgent|action required|deadline|due)/.test(text)) return { bucket: 'needs_action', reason: 'urgency keyword' };
|
||||
if (/[?]/.test(text)) return { bucket: 'needs_reply', reason: 'question mark' };
|
||||
return { bucket: 'fyi', reason: 'default' };
|
||||
}
|
||||
|
||||
export const emailTriageCommand = {
|
||||
name: 'email.triage',
|
||||
help() {
|
||||
return `email.triage — deterministic email triage report\n\n` +
|
||||
`Usage:\n` +
|
||||
` <emails> | email.triage [--subject-field subject] [--from-field from] [--snippet-field snippet] [--id-field id]\n\n` +
|
||||
`Output:\n` +
|
||||
` Single object: { summary, items, buckets }\n\n` +
|
||||
`Notes:\n` +
|
||||
` - This is intentionally non-LLM: rule-based classification (fast, predictable).\n` +
|
||||
` - Use --*-field flags to map provider-specific JSON into the triage schema.\n`;
|
||||
},
|
||||
async run({ input, args }) {
|
||||
const idField = args['id-field'] ?? 'id';
|
||||
const threadField = args['thread-field'] ?? 'threadId';
|
||||
const subjectField = args['subject-field'] ?? 'subject';
|
||||
const fromField = args['from-field'] ?? 'from';
|
||||
const snippetField = args['snippet-field'] ?? 'snippet';
|
||||
const dateField = args['date-field'] ?? 'date';
|
||||
|
||||
const items = [];
|
||||
for await (const raw of input) {
|
||||
const subject = normalizeString(getField(raw, subjectField, getField(raw, 'Subject')));
|
||||
const from = normalizeString(getField(raw, fromField, getField(raw, 'From')));
|
||||
const snippet = normalizeString(getField(raw, snippetField, getField(raw, 'Snippet')));
|
||||
|
||||
const classification = classifyEmail({ subject, snippet });
|
||||
|
||||
items.push({
|
||||
id: getField(raw, idField),
|
||||
threadId: getField(raw, threadField),
|
||||
from,
|
||||
fromEmail: parseEmailAddress(from),
|
||||
subject,
|
||||
snippet,
|
||||
date: getField(raw, dateField),
|
||||
bucket: classification.bucket,
|
||||
reason: classification.reason,
|
||||
raw,
|
||||
});
|
||||
}
|
||||
|
||||
const buckets = {
|
||||
needs_reply: items.filter((x) => x.bucket === 'needs_reply'),
|
||||
needs_action: items.filter((x) => x.bucket === 'needs_action'),
|
||||
fyi: items.filter((x) => x.bucket === 'fyi'),
|
||||
};
|
||||
|
||||
const summary = {
|
||||
total: items.length,
|
||||
needs_reply: buckets.needs_reply.length,
|
||||
needs_action: buckets.needs_action.length,
|
||||
fyi: buckets.fyi.length,
|
||||
};
|
||||
|
||||
const report = { summary, items, buckets };
|
||||
|
||||
return { output: (async function* () { yield report; })() };
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,71 @@
|
||||
import { spawn } from 'node:child_process';
|
||||
|
||||
export const execCommand = {
|
||||
name: 'exec',
|
||||
help() {
|
||||
return `exec — run an OS command\n\n` +
|
||||
`Usage:\n` +
|
||||
` exec <command...>\n` +
|
||||
` exec --json <command...>\n` +
|
||||
` exec --shell "<command line>"\n\n` +
|
||||
`Notes:\n` +
|
||||
` - With --json, parses stdout as JSON (single value).\n` +
|
||||
` - With --shell (or a single arg containing spaces), runs via /bin/sh -lc.\n`;
|
||||
},
|
||||
async run({ args, ctx }) {
|
||||
const cmd = args._;
|
||||
if (!cmd.length) throw new Error('exec requires a command');
|
||||
|
||||
const shellLine = typeof args.shell === 'string' ? args.shell : null;
|
||||
const useShell = Boolean(args.shell) || (cmd.length === 1 && /\s/.test(cmd[0]));
|
||||
|
||||
const result = useShell
|
||||
? await runProcess('/bin/sh', ['-lc', shellLine ?? cmd[0] ?? ''], { env: ctx.env, cwd: process.cwd() })
|
||||
: await runProcess(cmd[0], cmd.slice(1), { env: ctx.env, cwd: process.cwd() });
|
||||
|
||||
if (args.json) {
|
||||
let parsed;
|
||||
try {
|
||||
parsed = JSON.parse(result.stdout.trim() || 'null');
|
||||
} catch (err) {
|
||||
throw new Error(`exec --json could not parse stdout as JSON: ${err?.message ?? String(err)}`);
|
||||
}
|
||||
|
||||
return {
|
||||
output: asStream(Array.isArray(parsed) ? parsed : [parsed]),
|
||||
};
|
||||
}
|
||||
|
||||
const lines = result.stdout.split(/\r?\n/).filter(Boolean);
|
||||
return { output: asStream(lines) };
|
||||
},
|
||||
};
|
||||
|
||||
function runProcess(command, argv, { env, cwd }) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn(command, argv, {
|
||||
env,
|
||||
cwd,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
});
|
||||
|
||||
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.on('error', reject);
|
||||
child.on('close', (code) => {
|
||||
if (code === 0) return resolve({ stdout, stderr });
|
||||
reject(new Error(`exec failed (${code}): ${stderr.trim() || stdout.trim() || command}`));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function* asStream(items) {
|
||||
for (const item of items) yield item;
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
import { spawn } from 'node:child_process';
|
||||
|
||||
export const gogGmailSearchCommand = {
|
||||
name: 'gog.gmail.search',
|
||||
help() {
|
||||
return `gog.gmail.search — fetch Gmail messages via steipete/gog\n\n` +
|
||||
`Usage:\n` +
|
||||
` gog.gmail.search --query "newer_than:7d" --max 10 [--account you@gmail.com]\n\n` +
|
||||
`Behavior:\n` +
|
||||
` - Runs: gog gmail search <query> --max <n> --json --no-input\n` +
|
||||
` - Does not handle auth; relies on existing gog auth/credentials locally.\n`;
|
||||
},
|
||||
async run({ args, ctx }) {
|
||||
const query = args.query ?? args._[0];
|
||||
const max = args.max ?? 10;
|
||||
const account = args.account;
|
||||
|
||||
if (!query) throw new Error('gog.gmail.search requires --query');
|
||||
|
||||
const env = { ...ctx.env };
|
||||
if (account) env.GOG_ACCOUNT = String(account);
|
||||
|
||||
const argv = ['gmail', 'search', String(query), '--max', String(max), '--json', '--no-input'];
|
||||
const { stdout } = await runProcess('gog', argv, { env, cwd: process.cwd() });
|
||||
|
||||
let parsed;
|
||||
try {
|
||||
parsed = JSON.parse(stdout.trim() || '[]');
|
||||
} catch (err) {
|
||||
throw new Error(`gog gmail search returned non-JSON output`);
|
||||
}
|
||||
|
||||
// Keep it permissive: pass through what gog returns.
|
||||
// Later we can normalize into a canonical EmailMessage schema.
|
||||
const items = Array.isArray(parsed) ? parsed : [parsed];
|
||||
|
||||
return { output: asStream(items) };
|
||||
},
|
||||
};
|
||||
|
||||
function runProcess(command, argv, { env, cwd }) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn(command, argv, { env, cwd, stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
|
||||
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.on('error', (err) => {
|
||||
if (err?.code === 'ENOENT') {
|
||||
reject(new Error('gog not found on PATH (install steipete/gog from ClawdHub)'));
|
||||
return;
|
||||
}
|
||||
reject(err);
|
||||
});
|
||||
|
||||
child.on('close', (code) => {
|
||||
if (code === 0) return resolve({ stdout, stderr });
|
||||
reject(new Error(`gog failed (${code}): ${stderr.trim() || stdout.trim()}`));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
async function* asStream(items) {
|
||||
for (const item of items) yield item;
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
import { spawn } from 'node:child_process';
|
||||
|
||||
export const gogGmailSendCommand = {
|
||||
name: 'gog.gmail.send',
|
||||
help() {
|
||||
return `gog.gmail.send — send an email via steipete/gog\n\n` +
|
||||
`Usage:\n` +
|
||||
` gog.gmail.send --to a@b.com --subject "Hi" --body "Hello" [--account you@gmail.com]\n\n` +
|
||||
`Notes:\n` +
|
||||
` - Does not handle auth; relies on existing gog auth/credentials locally.\n` +
|
||||
` - Prefer running behind an approval gate (e.g. | approve --prompt "Send?").\n`;
|
||||
},
|
||||
async run({ args, ctx }) {
|
||||
const to = args.to;
|
||||
const subject = args.subject;
|
||||
const body = args.body;
|
||||
const account = args.account;
|
||||
|
||||
if (!to || !subject || body === undefined) {
|
||||
throw new Error('gog.gmail.send requires --to, --subject, and --body');
|
||||
}
|
||||
|
||||
const env = { ...ctx.env };
|
||||
if (account) env.GOG_ACCOUNT = String(account);
|
||||
|
||||
const argv = ['gmail', 'send', '--to', String(to), '--subject', String(subject), '--body', String(body), '--no-input'];
|
||||
|
||||
const { stdout } = await runProcess('gog', argv, { env, cwd: process.cwd() });
|
||||
|
||||
return {
|
||||
output: (async function* () {
|
||||
yield { ok: true, to, subject, result: stdout.trim() };
|
||||
})(),
|
||||
};
|
||||
},
|
||||
};
|
||||
|
||||
function runProcess(command, argv, { env, cwd }) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const child = spawn(command, argv, { env, cwd, stdio: ['ignore', 'pipe', 'pipe'] });
|
||||
|
||||
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.on('error', (err) => {
|
||||
if (err?.code === 'ENOENT') {
|
||||
reject(new Error('gog not found on PATH (install steipete/gog from ClawdHub)'));
|
||||
return;
|
||||
}
|
||||
reject(err);
|
||||
});
|
||||
|
||||
child.on('close', (code) => {
|
||||
if (code === 0) return resolve({ stdout, stderr });
|
||||
reject(new Error(`gog failed (${code}): ${stderr.trim() || stdout.trim()}`));
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
export const headCommand = {
|
||||
name: 'head',
|
||||
help() {
|
||||
return `head — take first N items\n\nUsage:\n head --n 10\n`;
|
||||
},
|
||||
async run({ input, args }) {
|
||||
const n = args.n === undefined ? 10 : Number(args.n);
|
||||
if (!Number.isFinite(n) || n < 0) throw new Error('head --n must be a non-negative number');
|
||||
|
||||
return {
|
||||
output: (async function* () {
|
||||
let i = 0;
|
||||
for await (const item of input) {
|
||||
if (i++ >= n) break;
|
||||
yield item;
|
||||
}
|
||||
})(),
|
||||
};
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,14 @@
|
||||
export const jsonCommand = {
|
||||
name: 'json',
|
||||
help() {
|
||||
return `json — render pipeline output as JSON\n\nUsage:\n ... | json\n`;
|
||||
},
|
||||
async run({ input, ctx }) {
|
||||
const items = [];
|
||||
for await (const item of input) items.push(item);
|
||||
ctx.render.json(items);
|
||||
return { output: emptyStream(), rendered: true };
|
||||
},
|
||||
};
|
||||
|
||||
async function* emptyStream() {}
|
||||
@@ -0,0 +1,25 @@
|
||||
export const pickCommand = {
|
||||
name: 'pick',
|
||||
help() {
|
||||
return `pick — project fields from objects\n\nUsage:\n ... | pick id,subject,from\n`;
|
||||
},
|
||||
async run({ input, args }) {
|
||||
const spec = args._[0];
|
||||
if (!spec) throw new Error('pick requires a comma-separated field list');
|
||||
const fields = spec.split(',').map((s) => s.trim()).filter(Boolean);
|
||||
|
||||
return {
|
||||
output: (async function* () {
|
||||
for await (const item of input) {
|
||||
if (item === null || typeof item !== 'object') {
|
||||
yield item;
|
||||
continue;
|
||||
}
|
||||
const out = {};
|
||||
for (const f of fields) out[f] = item[f];
|
||||
yield out;
|
||||
}
|
||||
})(),
|
||||
};
|
||||
},
|
||||
};
|
||||
@@ -0,0 +1,56 @@
|
||||
function stringifyCell(v) {
|
||||
if (v === null || v === undefined) return '';
|
||||
if (typeof v === 'string') return v;
|
||||
if (typeof v === 'number' || typeof v === 'boolean') return String(v);
|
||||
return JSON.stringify(v);
|
||||
}
|
||||
|
||||
export const tableCommand = {
|
||||
name: 'table',
|
||||
help() {
|
||||
return `table — render items as a simple table\n\nUsage:\n ... | table\n\nNotes:\n - If items are objects, columns are union of keys (first 20 items).\n`;
|
||||
},
|
||||
async run({ input, ctx }) {
|
||||
const items = [];
|
||||
for await (const item of input) items.push(item);
|
||||
|
||||
if (items.length === 0) {
|
||||
ctx.stdout.write('(no results)\n');
|
||||
return { output: emptyStream(), rendered: true };
|
||||
}
|
||||
|
||||
const sample = items.slice(0, 20);
|
||||
const objectItems = sample.filter((x) => x && typeof x === 'object' && !Array.isArray(x));
|
||||
|
||||
if (objectItems.length === sample.length) {
|
||||
const cols = [];
|
||||
const seen = new Set();
|
||||
for (const obj of objectItems) {
|
||||
for (const k of Object.keys(obj)) {
|
||||
if (!seen.has(k)) {
|
||||
seen.add(k);
|
||||
cols.push(k);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const rows = [cols, ...items.map((it) => cols.map((c) => stringifyCell(it?.[c])))]
|
||||
.map((row) => row.map((cell) => cell.replace(/\n/g, ' ')));
|
||||
|
||||
const widths = cols.map((_, i) => Math.max(...rows.map((r) => r[i].length), 3));
|
||||
|
||||
const renderRow = (row) => row.map((cell, i) => cell.padEnd(widths[i])).join(' ');
|
||||
ctx.stdout.write(renderRow(rows[0]) + '\n');
|
||||
ctx.stdout.write(widths.map((w) => '-'.repeat(w)).join(' ') + '\n');
|
||||
for (const row of rows.slice(1)) ctx.stdout.write(renderRow(row) + '\n');
|
||||
|
||||
return { output: emptyStream(), rendered: true };
|
||||
}
|
||||
|
||||
// Fallback: render each item on a line.
|
||||
for (const item of items) ctx.stdout.write(stringifyCell(item) + '\n');
|
||||
return { output: emptyStream(), rendered: true };
|
||||
},
|
||||
};
|
||||
|
||||
async function* emptyStream() {}
|
||||
@@ -0,0 +1,56 @@
|
||||
function parsePredicate(expr) {
|
||||
const m = expr.match(/^([a-zA-Z0-9_\.]+)\s*(==|=|!=|<=|>=|<|>)\s*(.+)$/);
|
||||
if (!m) throw new Error(`Invalid where expression: ${expr}`);
|
||||
const [, path, op, rawValue] = m;
|
||||
|
||||
let value = rawValue;
|
||||
if (rawValue === 'true') value = true;
|
||||
else if (rawValue === 'false') value = false;
|
||||
else if (rawValue === 'null') value = null;
|
||||
else if (!Number.isNaN(Number(rawValue)) && rawValue.trim() !== '') value = Number(rawValue);
|
||||
|
||||
return { path, op: op === '=' ? '==' : op, value };
|
||||
}
|
||||
|
||||
function getPath(obj, path) {
|
||||
const parts = path.split('.');
|
||||
let cur = obj;
|
||||
for (const p of parts) {
|
||||
if (cur === null || typeof cur !== 'object') return undefined;
|
||||
cur = cur[p];
|
||||
}
|
||||
return cur;
|
||||
}
|
||||
|
||||
function compare(left, op, right) {
|
||||
switch (op) {
|
||||
case '==': return left == right; // intentional loose equality for convenience
|
||||
case '!=': return left != right;
|
||||
case '<': return left < right;
|
||||
case '<=': return left <= right;
|
||||
case '>': return left > right;
|
||||
case '>=': return left >= right;
|
||||
default: throw new Error(`Unsupported operator: ${op}`);
|
||||
}
|
||||
}
|
||||
|
||||
export const whereCommand = {
|
||||
name: 'where',
|
||||
help() {
|
||||
return `where — filter objects by a simple predicate\n\nUsage:\n ... | where unread=true\n ... | where minutes>=30\n ... | where sender.domain==example.com\n`;
|
||||
},
|
||||
async run({ input, args }) {
|
||||
const expr = args._[0];
|
||||
if (!expr) throw new Error('where requires an expression (e.g. field=value)');
|
||||
const pred = parsePredicate(expr);
|
||||
|
||||
return {
|
||||
output: (async function* () {
|
||||
for await (const item of input) {
|
||||
const left = getPath(item, pred.path);
|
||||
if (compare(left, pred.op, pred.value)) yield item;
|
||||
}
|
||||
})(),
|
||||
};
|
||||
},
|
||||
};
|
||||
+140
@@ -0,0 +1,140 @@
|
||||
function isWhitespace(ch) {
|
||||
return ch === ' ' || ch === '\t' || ch === '\n' || ch === '\r';
|
||||
}
|
||||
|
||||
function splitPipes(input) {
|
||||
const parts = [];
|
||||
let current = '';
|
||||
let quote = null;
|
||||
|
||||
for (let i = 0; i < input.length; i++) {
|
||||
const ch = input[i];
|
||||
|
||||
if (quote) {
|
||||
if (ch === '\\') {
|
||||
const next = input[i + 1];
|
||||
if (next) {
|
||||
current += next;
|
||||
i++;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (ch === quote) {
|
||||
quote = null;
|
||||
continue;
|
||||
}
|
||||
current += ch;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (ch === '"' || ch === "'") {
|
||||
quote = ch;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (ch === '|') {
|
||||
parts.push(current.trim());
|
||||
current = '';
|
||||
continue;
|
||||
}
|
||||
|
||||
current += ch;
|
||||
}
|
||||
|
||||
if (quote) throw new Error('Unclosed quote');
|
||||
if (current.trim().length > 0) parts.push(current.trim());
|
||||
return parts;
|
||||
}
|
||||
|
||||
function tokenizeCommand(input) {
|
||||
const tokens = [];
|
||||
let current = '';
|
||||
let quote = null;
|
||||
|
||||
const push = () => {
|
||||
if (current.length > 0) tokens.push(current);
|
||||
current = '';
|
||||
};
|
||||
|
||||
for (let i = 0; i < input.length; i++) {
|
||||
const ch = input[i];
|
||||
|
||||
if (quote) {
|
||||
if (ch === '\\') {
|
||||
const next = input[i + 1];
|
||||
if (next) {
|
||||
current += next;
|
||||
i++;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if (ch === quote) {
|
||||
quote = null;
|
||||
continue;
|
||||
}
|
||||
current += ch;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (ch === '"' || ch === "'") {
|
||||
quote = ch;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (isWhitespace(ch)) {
|
||||
push();
|
||||
continue;
|
||||
}
|
||||
|
||||
current += ch;
|
||||
}
|
||||
|
||||
if (quote) throw new Error('Unclosed quote');
|
||||
push();
|
||||
return tokens;
|
||||
}
|
||||
|
||||
function parseArgs(tokens) {
|
||||
const args = { _: [] };
|
||||
|
||||
for (let i = 0; i < tokens.length; i++) {
|
||||
const tok = tokens[i];
|
||||
|
||||
if (tok.startsWith('--')) {
|
||||
const eq = tok.indexOf('=');
|
||||
if (eq !== -1) {
|
||||
const key = tok.slice(2, eq);
|
||||
const value = tok.slice(eq + 1);
|
||||
args[key] = value;
|
||||
continue;
|
||||
}
|
||||
|
||||
const key = tok.slice(2);
|
||||
const next = tokens[i + 1];
|
||||
if (!next || next.startsWith('--')) {
|
||||
args[key] = true;
|
||||
continue;
|
||||
}
|
||||
args[key] = next;
|
||||
i++;
|
||||
continue;
|
||||
}
|
||||
|
||||
args._.push(tok);
|
||||
}
|
||||
|
||||
return args;
|
||||
}
|
||||
|
||||
export function parsePipeline(input) {
|
||||
const stages = splitPipes(input);
|
||||
if (stages.length === 0) throw new Error('Empty pipeline');
|
||||
|
||||
return stages.map((stage) => {
|
||||
const tokens = tokenizeCommand(stage);
|
||||
if (tokens.length === 0) throw new Error('Empty command stage');
|
||||
const name = tokens[0];
|
||||
const args = parseArgs(tokens.slice(1));
|
||||
return { name, args, raw: stage };
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
export function createJsonRenderer(stdout) {
|
||||
return {
|
||||
json(items) {
|
||||
stdout.write(JSON.stringify(items, null, 2));
|
||||
stdout.write('\n');
|
||||
},
|
||||
lines(lines) {
|
||||
for (const line of lines) stdout.write(String(line) + '\n');
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import { createJsonRenderer } from './renderers/json.js';
|
||||
|
||||
export async function runPipeline({ pipeline, registry, stdin, stdout, stderr, env }) {
|
||||
let stream = emptyStream();
|
||||
let rendered = false;
|
||||
|
||||
const ctx = {
|
||||
stdin,
|
||||
stdout,
|
||||
stderr,
|
||||
env,
|
||||
registry,
|
||||
render: createJsonRenderer(stdout),
|
||||
};
|
||||
|
||||
for (const stage of pipeline) {
|
||||
const command = registry.get(stage.name);
|
||||
if (!command) {
|
||||
throw new Error(`Unknown command: ${stage.name}`);
|
||||
}
|
||||
|
||||
const result = await command.run({ input: stream, args: stage.args, ctx });
|
||||
|
||||
if (result && result.rendered) {
|
||||
rendered = true;
|
||||
}
|
||||
|
||||
stream = result?.output ?? emptyStream();
|
||||
}
|
||||
|
||||
const items = [];
|
||||
for await (const item of stream) items.push(item);
|
||||
|
||||
return { items, rendered };
|
||||
}
|
||||
|
||||
async function* emptyStream() {}
|
||||
@@ -0,0 +1,65 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { createDefaultRegistry } from '../src/commands/registry.js';
|
||||
import { runPipeline } from '../src/runtime.js';
|
||||
|
||||
function streamOf(items) {
|
||||
return (async function* () {
|
||||
for (const item of items) yield item;
|
||||
})();
|
||||
}
|
||||
|
||||
test('email.triage classifies by keywords deterministically', async () => {
|
||||
const registry = createDefaultRegistry();
|
||||
const pipeline = [{ name: 'email.triage', args: { _: [] }, raw: 'email.triage' }];
|
||||
|
||||
const output = await runPipeline({
|
||||
pipeline,
|
||||
registry,
|
||||
stdin: process.stdin,
|
||||
stdout: process.stdout,
|
||||
stderr: process.stderr,
|
||||
env: process.env,
|
||||
});
|
||||
|
||||
// Default input is empty stream => report total=0.
|
||||
assert.equal(output.items.length, 1);
|
||||
assert.equal(output.items[0].summary.total, 0);
|
||||
});
|
||||
|
||||
test('email.triage maps custom fields', async () => {
|
||||
const registry = createDefaultRegistry();
|
||||
const pipeline = [
|
||||
{
|
||||
name: 'email.triage',
|
||||
args: {
|
||||
_: [],
|
||||
'subject-field': 'meta.title',
|
||||
'from-field': 'meta.sender',
|
||||
'snippet-field': 'meta.snip',
|
||||
'id-field': 'meta.id',
|
||||
},
|
||||
raw: 'email.triage',
|
||||
},
|
||||
];
|
||||
|
||||
const output = await (async () => {
|
||||
// Inject custom input by monkey-patching runtime: call command directly.
|
||||
const cmd = registry.get('email.triage');
|
||||
const result = await cmd.run({
|
||||
input: streamOf([
|
||||
{ meta: { id: '1', title: 'Invoice due', sender: 'Billing <b@example.com>', snip: 'Pay now' } },
|
||||
]),
|
||||
args: pipeline[0].args,
|
||||
ctx: { stdin: process.stdin, stdout: process.stdout, stderr: process.stderr, env: process.env, registry, render: { json() {}, lines() {} } },
|
||||
});
|
||||
const items = [];
|
||||
for await (const it of result.output) items.push(it);
|
||||
return items;
|
||||
})();
|
||||
|
||||
assert.equal(output.length, 1);
|
||||
assert.equal(output[0].summary.total, 1);
|
||||
assert.equal(output[0].items[0].bucket, 'needs_action');
|
||||
assert.equal(output[0].items[0].fromEmail, 'b@example.com');
|
||||
});
|
||||
@@ -0,0 +1,20 @@
|
||||
import test from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { parsePipeline } from '../src/parser.js';
|
||||
|
||||
test('parsePipeline splits stages and args', () => {
|
||||
const p = parsePipeline("exec echo hi | where a=1 | pick id,subject");
|
||||
assert.equal(p.length, 3);
|
||||
assert.equal(p[0].name, 'exec');
|
||||
assert.deepEqual(p[0].args._, ['echo', 'hi']);
|
||||
assert.equal(p[1].name, 'where');
|
||||
assert.equal(p[1].args._[0], 'a=1');
|
||||
assert.equal(p[2].name, 'pick');
|
||||
assert.equal(p[2].args._[0], 'id,subject');
|
||||
});
|
||||
|
||||
test('parsePipeline keeps quoted pipes', () => {
|
||||
const p = parsePipeline("exec echo 'a|b' | json");
|
||||
assert.equal(p.length, 2);
|
||||
assert.deepEqual(p[0].args._, ['echo', 'a|b']);
|
||||
});
|
||||
Reference in New Issue
Block a user