feat(workflow): enrich workflow/agent metadata from workflowProgress;
lightweight workflowTree, build debounce, minor fixes
- Index phase, label, model, state, duration, tokens per workflow agent
- Index duration, total_tokens, status, name per workflow run
- workflowTree returns parsed result + agent summaries instead of
dumping all messages
- 30s debounce on buildIndex to avoid repeated directory scans
- Fix broken BASH_EXIT_PAT (SQLite LIKE has no character classes)
- Fix SKILL.md step numbering, document FTS5 hyphen limitation
This commit is contained in:
tommy0103
@@ -84,8 +84,8 @@ All list-returning functions accept a common filter opts object: `{ project, aft
|
||||
- `trace(uuid)` -- full parent chain from root to message
|
||||
- `thread(sessionId)` -- all messages in a session, ordered by time
|
||||
- `subagents(opts?)` -- subagent metadata + message counts. opts: `{ sessionId, project, limit }`
|
||||
- `workflows(opts?)` -- workflow runs. opts: `{ sessionId, project, after, before, limit }`
|
||||
- `workflowTree(runId)` -- workflow + its agents + all their messages
|
||||
- `workflows(opts?)` -- workflow runs with duration, tokens, status. opts: `{ sessionId, project, after, before, limit }`
|
||||
- `workflowTree(runId)` -- workflow metadata + parsed result + agents with phase/label/tokens/duration (no messages; use `sql()` with `agent_id` to drill into a specific agent)
|
||||
- `fileHistory(filePath, opts?)` -- every Edit/Write/Read on a file. opts: `{ after, before, limit }`
|
||||
- `failures(opts?)` -- tool calls that returned errors, with surrounding context. opts: `{ sessionId, project, after, before, limit }`
|
||||
- `summaries(opts?)` -- session summaries (away recaps, compaction summaries). opts: `{ sessionId, project, after, before, limit, sessions }`. Returns: `[{ id, session_id, timestamp, source, content, session_title, project }]`. Use `source` for values like `away_summary`; use `content` for the summary text.
|
||||
@@ -102,7 +102,7 @@ All list-returning functions accept a common filter opts object: `{ project, aft
|
||||
1. `sessions({ project: '...' })` or `recent()` — find relevant sessions
|
||||
2. `summaries({ project: '...' })` — read session summaries to judge relevance (cheapest)
|
||||
3. `search()` — find specific messages matching a query
|
||||
3. When you find a relevant message and want more context, expand from that point:
|
||||
4. When you find a relevant message and want more context, expand from that point:
|
||||
- **Horizontally**: use `sql()` to fetch neighboring messages by timestamp
|
||||
```js
|
||||
sql('SELECT uuid,role,text FROM messages WHERE session_id=? AND timestamp>? ORDER BY timestamp LIMIT 5', sid, msg.timestamp)
|
||||
@@ -194,3 +194,4 @@ return context(hits[0].message.uuid)
|
||||
- Query scripts run in a sandboxed VM context -- no file system or network access from inside scripts.
|
||||
- Text is truncated to 10k chars per message during indexing.
|
||||
- FTS5 search supports standard SQLite FTS syntax: `"exact phrase"`, `term1 AND term2`, `term1 OR term2`, `term1 NOT term2`.
|
||||
- FTS5 tokenizes on hyphens. To search for `SkillOpt-outputs`, use `"skillopt outputs"` (replace hyphen with space, wrap in quotes for phrase match). For exact match on hyphenated strings, use `sql()` with LIKE instead.
|
||||
|
||||
+17
-6
@@ -132,7 +132,11 @@ CREATE TABLE workflows (
|
||||
script TEXT, -- workflow script content (truncated)
|
||||
result_json TEXT, -- JSON-serialized workflow result
|
||||
timestamp TEXT, -- ISO 8601 execution time
|
||||
agent_count INTEGER DEFAULT 0 -- number of agents in this workflow
|
||||
agent_count INTEGER DEFAULT 0, -- number of agents in this workflow
|
||||
duration_ms INTEGER, -- wall-clock duration of the workflow run
|
||||
total_tokens INTEGER, -- total tokens across all agents
|
||||
status TEXT, -- "completed", "failed", etc.
|
||||
workflow_name TEXT -- name from the workflow script meta
|
||||
);
|
||||
```
|
||||
|
||||
@@ -144,11 +148,18 @@ Individual agents within a workflow run.
|
||||
|
||||
```sql
|
||||
CREATE TABLE workflow_agents (
|
||||
agent_id TEXT PRIMARY KEY, -- agent UUID
|
||||
agent_id TEXT PRIMARY KEY, -- agent UUID (prefixed with "agent-")
|
||||
run_id TEXT, -- FK -> workflows.run_id
|
||||
session_id TEXT, -- FK -> sessions.id
|
||||
agent_type TEXT, -- agent type label
|
||||
description TEXT -- task description
|
||||
description TEXT, -- task description
|
||||
phase TEXT, -- workflow phase title (e.g. "Review", "Verify")
|
||||
label TEXT, -- agent label from workflow script
|
||||
model TEXT, -- model used (e.g. "claude-opus-4-6[1m]")
|
||||
state TEXT, -- "done", "error", etc.
|
||||
duration_ms INTEGER, -- wall-clock duration of this agent
|
||||
tokens INTEGER, -- total tokens used by this agent
|
||||
tool_calls INTEGER -- number of tool calls made
|
||||
);
|
||||
```
|
||||
|
||||
@@ -305,13 +316,13 @@ return wfs.map(w => ({ run: w.run_id, agents: w.agent_count, time: w.timestamp }
|
||||
|
||||
#### `workflowTree(runId)`
|
||||
|
||||
Full execution tree for a workflow: the workflow record plus all its agents and their messages.
|
||||
Lightweight execution tree for a workflow: metadata, parsed result, and agent summaries with phase/label/performance data. Does not load agent messages — use `sql()` with `agent_id` to drill into a specific agent.
|
||||
|
||||
**Returns:** `{ ...workflow_row, agents: Array<{ ...agent_row, messages: Array<message> }> }` or `null`.
|
||||
**Returns:** `{ ...workflow_row, result: object, agents: Array<{ ...agent_row, messageCount }> }` or `null`.
|
||||
|
||||
```js
|
||||
const tree = workflowTree('run-uuid');
|
||||
return tree?.agents.map(a => ({ type: a.agent_type, msgs: a.messages.length }));
|
||||
return tree?.agents.map(a => ({ phase: a.phase, label: a.label, tokens: a.tokens, msgs: a.messageCount }));
|
||||
```
|
||||
|
||||
#### `fileHistory(filePath, opts?)`
|
||||
|
||||
+5
-2
@@ -31,10 +31,13 @@ CREATE TABLE IF NOT EXISTS subagents (
|
||||
agent_type TEXT, description TEXT, duration_ms INTEGER, total_tokens INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS workflows (
|
||||
run_id TEXT PRIMARY KEY, session_id TEXT, task_id TEXT,
|
||||
script TEXT, result_json TEXT, timestamp TEXT, agent_count INTEGER DEFAULT 0);
|
||||
script TEXT, result_json TEXT, timestamp TEXT, agent_count INTEGER DEFAULT 0,
|
||||
duration_ms INTEGER, total_tokens INTEGER, status TEXT, workflow_name TEXT);
|
||||
CREATE TABLE IF NOT EXISTS workflow_agents (
|
||||
agent_id TEXT PRIMARY KEY, run_id TEXT, session_id TEXT,
|
||||
agent_type TEXT, description TEXT);
|
||||
agent_type TEXT, description TEXT,
|
||||
phase TEXT, label TEXT, model TEXT, state TEXT,
|
||||
duration_ms INTEGER, tokens INTEGER, tool_calls INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS index_state (
|
||||
jsonl_path TEXT PRIMARY KEY, mtime REAL, lines_processed INTEGER);
|
||||
CREATE TABLE IF NOT EXISTS summaries (
|
||||
|
||||
+20
-4
@@ -148,7 +148,7 @@ function indexSubagentMeta(db, fi) {
|
||||
const ts = db.prepare('SELECT MIN(timestamp) as t0, MAX(timestamp) as t1 FROM messages WHERE agent_id=?').get(fi.agentId);
|
||||
const dur = ts?.t0 && ts?.t1 ? new Date(ts.t1).getTime() - new Date(ts.t0).getTime() : null;
|
||||
if (fi.workflowRunId) {
|
||||
db.prepare('INSERT OR REPLACE INTO workflow_agents VALUES(?,?,?,?,?)').run(fi.agentId, fi.workflowRunId, fi.sessionId, meta.agentType||null, meta.description||null);
|
||||
db.prepare('INSERT OR REPLACE INTO workflow_agents (agent_id,run_id,session_id,agent_type,description) VALUES(?,?,?,?,?)').run(fi.agentId, fi.workflowRunId, fi.sessionId, meta.agentType||null, meta.description||null);
|
||||
} else {
|
||||
db.prepare('INSERT OR REPLACE INTO subagents VALUES(?,?,?,?,?,?,?)').run(fi.agentId, fi.sessionId, meta.toolUseId||null, meta.agentType||null, meta.description||null, dur, tok?.t||0);
|
||||
}
|
||||
@@ -175,9 +175,17 @@ function indexWorkflows(db) {
|
||||
const wf = JSON.parse(fs.readFileSync(path.join(wd, f), 'utf8'));
|
||||
if (!wf.runId) continue;
|
||||
const ac = db.prepare('SELECT COUNT(*) as c FROM workflow_agents WHERE run_id=?').get(wf.runId);
|
||||
db.prepare('INSERT OR REPLACE INTO workflows VALUES(?,?,?,?,?,?,?)').run(
|
||||
db.prepare('INSERT OR REPLACE INTO workflows (run_id,session_id,task_id,script,result_json,timestamp,agent_count,duration_ms,total_tokens,status,workflow_name) VALUES(?,?,?,?,?,?,?,?,?,?,?)').run(
|
||||
wf.runId, sd, wf.taskId||null, wf.script||null,
|
||||
wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0);
|
||||
wf.result ? JSON.stringify(wf.result) : null, wf.timestamp||null, ac?.c||0,
|
||||
wf.durationMs||null, wf.totalTokens||null, wf.status||null, wf.workflowName||null);
|
||||
const progress = wf.workflowProgress || [];
|
||||
for (const item of progress) {
|
||||
if (item.type !== 'workflow_agent' || !item.agentId) continue;
|
||||
db.prepare('UPDATE workflow_agents SET phase=?, label=?, model=?, state=?, duration_ms=?, tokens=?, tool_calls=? WHERE agent_id=?').run(
|
||||
item.phaseTitle||null, item.label||null, item.model||null, item.state||null,
|
||||
item.durationMs||null, item.tokens||null, item.toolCalls||null, 'agent-' + item.agentId);
|
||||
}
|
||||
} catch (e) { process.stderr.write(`Warning: failed to index workflow ${f}: ${e.message}\n`); }
|
||||
}
|
||||
}
|
||||
@@ -194,8 +202,15 @@ function indexHistory(db) {
|
||||
});
|
||||
}
|
||||
|
||||
function buildIndex() {
|
||||
const BUILD_DEBOUNCE_MS = 30000;
|
||||
|
||||
function buildIndex({ force = false } = {}) {
|
||||
const db = openDb();
|
||||
if (!force) {
|
||||
const last = db.prepare("SELECT mtime FROM index_state WHERE jsonl_path='__last_build__'").get();
|
||||
if (last && Date.now() - last.mtime < BUILD_DEBOUNCE_MS) { db.close(); return; }
|
||||
}
|
||||
|
||||
const files = discoverJsonlFiles();
|
||||
for (const f of files) {
|
||||
db.exec('BEGIN');
|
||||
@@ -213,6 +228,7 @@ function buildIndex() {
|
||||
indexWorkflows(db);
|
||||
indexHistory(db);
|
||||
db.exec("INSERT INTO messages_fts(messages_fts) VALUES('rebuild')");
|
||||
db.prepare("INSERT OR REPLACE INTO index_state (jsonl_path, mtime, lines_processed) VALUES ('__last_build__', ?, 0)").run(Date.now());
|
||||
db.exec('COMMIT');
|
||||
} catch (e) {
|
||||
db.exec('ROLLBACK');
|
||||
|
||||
+8
-5
@@ -22,7 +22,7 @@ function buildWhere(opts, aliases) {
|
||||
return { where: clauses.length ? clauses.join(' AND ') : '1=1', params };
|
||||
}
|
||||
|
||||
const BASH_EXIT_PAT = 'Exit code [1-9]%';
|
||||
const BASH_EXIT_PAT = 'Exit code %';
|
||||
|
||||
function createQueryApi(db) {
|
||||
const q = (sql, ...p) => db.prepare(sql).all(...p);
|
||||
@@ -107,10 +107,13 @@ function createQueryApi(db) {
|
||||
const workflowTree = (runId) => {
|
||||
const wf = db.prepare('SELECT * FROM workflows WHERE run_id=?').get(runId);
|
||||
if (!wf) return null;
|
||||
const agents = db.prepare('SELECT * FROM workflow_agents WHERE run_id=?').all(runId).map(a => ({
|
||||
...a, messages: db.prepare('SELECT * FROM messages WHERE agent_id=? ORDER BY timestamp').all(a.agent_id),
|
||||
}));
|
||||
return { ...wf, agents };
|
||||
let result = null;
|
||||
try { result = JSON.parse(wf.result_json); } catch {}
|
||||
const agents = db.prepare('SELECT * FROM workflow_agents WHERE run_id=?').all(runId).map(a => {
|
||||
const mc = db.prepare('SELECT COUNT(*) as c FROM messages WHERE agent_id=?').get(a.agent_id);
|
||||
return { ...a, messageCount: mc?.c || 0 };
|
||||
});
|
||||
return { ...wf, result, agents };
|
||||
};
|
||||
|
||||
const fileHistory = (fp, opts = {}) => {
|
||||
|
||||
+1
-1
@@ -22,7 +22,7 @@ function executeQuery(db, scriptContent) {
|
||||
function main() {
|
||||
const args = process.argv.slice(2);
|
||||
if (args[0] === '--build') {
|
||||
buildIndex();
|
||||
buildIndex({ force: true });
|
||||
process.stdout.write(JSON.stringify({ ok: true, db: DB_PATH }) + '\n');
|
||||
return;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user