From 10146b64157a36af64895b57cc140fa699864d26 Mon Sep 17 00:00:00 2001 From: jedarden Date: Tue, 21 Apr 2026 11:57:19 -0400 Subject: [PATCH] feat(bd-cj8): adopt NeedleEvent canonical schema + publish docs/schema.md Add NeedleEvent interface as the canonical internal shape versioned at schema-version 1. Parser validates wire format and asserts schema version on incoming events. Legacy LogEvent retained as backward-compatible adapter. docs/schema.md documents all fields, the (worker_id, sequence) ordering contract, and the full event taxonomy cross-referenced with NeedleEventType. Co-Authored-By: Claude Opus 4.7 --- README.md | 1 + docs/schema.md | 190 +++++++++++++++++++++++++++++++++++++++++++++++++ src/parser.ts | 120 +++++++++++++++++++++++++++++-- src/types.ts | 41 +++++++++++ 4 files changed, 346 insertions(+), 6 deletions(-) create mode 100644 docs/schema.md diff --git a/README.md b/README.md index 440bee5..9ef7a34 100644 --- a/README.md +++ b/README.md @@ -119,4 +119,5 @@ Everything stays on your machine — FABRIC is a local collector, not a third-pa ## Documentation +- [NeedleEvent Schema](docs/schema.md) — canonical wire format shared with NEEDLE - [Implementation Plan](docs/plan.md) diff --git a/docs/schema.md b/docs/schema.md new file mode 100644 index 0000000..9995562 --- /dev/null +++ b/docs/schema.md @@ -0,0 +1,190 @@ +# NeedleEvent Schema + +**schema-version: 1** — shared contract between NEEDLE and FABRIC. + +Both projects must agree on this version. NEEDLE emits `schema_version` in its +output; FABRIC asserts compatibility during parse via `NEEDLE_EVENT_SCHEMA_VERSION` +in `src/types.ts`. If the values diverge, `parseNeedleEvent` throws. + +## Wire Format + +Every event emitted by NEEDLE — over JSONL and OTLP logs — conforms to this +shape: + +```json +{ + "schema_version": 1, + "timestamp": "2026-04-21T11:20:19.962811515Z", + "event_type": "worker.started", + "worker_id": "tcb-alpha", + "session_id": "d7261357", + "sequence": 1, + "bead_id": "bd-abc123", + "data": {} +} +``` + +## Fields + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `schema_version` | `number` | recommended | Protocol version. Present in newer NEEDLE output; FABRIC asserts it when present. | +| `timestamp` | `string` | **yes** | RFC3339 timestamp. Display only — **not** authoritative for ordering. | +| `event_type` | `string` | **yes** | Taxonomy string from the event taxonomy table below. | +| `worker_id` | `string` | **yes** | Worker identifier (e.g. `"tcb-alpha"`). | +| `session_id` | `string` | **yes** | Groups a worker's lifetime events into a single session. | +| `sequence` | `number` | **yes** | Per-worker monotonic counter. Authoritative for ordering within a worker. | +| `bead_id` | `string` | no | Present when the event pertains to a specific bead. | +| `data` | `object` | **yes** | Event-specific payload (see taxonomy table for notable fields). May be empty `{}`. | + +## Ordering Contract + +Sort events by **`(worker_id, sequence)`**, not by `timestamp`. + +Wall clocks skew across hosts. `sequence` is the worker's own monotonic counter +and is the only reliable basis for replay and timeline reconstruction within a +single worker. To interleave events from multiple workers, merge-sort on +`sequence` within each `worker_id` partition, then order across partitions by +`timestamp` as a tiebreaker. + +## Event Taxonomy + +Format: `category.action`. Categories group related lifecycle phases. + +### Worker Lifecycle + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `worker.started` | Worker boot | `version`, `worker_name` | +| `worker.idle` | Worker is idle, waiting for work | — | +| `worker.stopped` | Worker exit | `reason` | +| `worker.draining` | Worker is draining before shutdown | — | + +### Bead Lifecycle + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `bead.claimed` | Bead claim succeeded | `bead_id` | +| `bead.prompt_built` | Prompt constructed for agent | `bead_id` | +| `bead.agent_started` | Agent began working on bead | `bead_id` | +| `bead.agent_completed` | Agent finished working on bead | `bead_id`, `duration_ms` | +| `bead.completed` | Bead work fully completed | `bead_id`, `duration_ms` | +| `bead.failed` | Bead work failed | `bead_id`, `error` | +| `bead.released` | Bead released back to queue | `bead_id` | +| `bead.claim_retry` | Claim attempt will be retried | `bead_id`, `attempt` | +| `bead.claim_exhausted` | All claim retries exhausted | `bead_id` | + +### Bead Mitosis + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `bead.mitosis.check` | Checking if bead should split | `bead_id` | +| `bead.mitosis.started` | Mitosis began | `bead_id` | +| `bead.mitosis.child_created` | Child bead created | `bead_id`, `child_id` | +| `bead.mitosis.complete` | Mitosis finished | `bead_id` | +| `bead.mitosis.failed` | Mitosis failed | `bead_id`, `error` | +| `bead.mitosis.skipped` | Mitosis skipped (not needed) | `bead_id` | + +### Strand Lifecycle + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `strand.started` | Strand execution began | `strand` | +| `strand.completed` | Strand execution finished | `strand`, `duration_ms` | +| `strand.fallthrough` | Strand found no work | `strand` | +| `strand.skipped` | Strand was skipped | `strand`, `reason` | + +### Hook Lifecycle + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `hook.started` | Hook execution began | `hook` | +| `hook.completed` | Hook execution finished | `hook`, `duration_ms` | +| `hook.failed` | Hook execution failed | `hook`, `error` | + +### Heartbeat + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `heartbeat.emitted` | Periodic heartbeat | `status` | +| `heartbeat.stuck_detected` | Worker appears stuck | `worker_id`, `since` | +| `heartbeat.recovery` | Worker recovered from stuck state | `worker_id` | + +### Mend (Maintenance) + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `mend.orphan_released` | Orphaned bead released | `bead_id` | +| `mend.heartbeat_cleaned` | Stale heartbeat cleaned | `worker_id` | +| `mend.logs_pruned` | Old logs pruned | `bytes_freed` | +| `mend.completed` | Mend cycle finished | `duration_ms` | + +### Unravel (Alternatives) + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `unravel.alternatives_created` | Alternative beads created | `parent_bead_id`, `count` | +| `unravel.alternative_created` | Single alternative bead created | `parent_bead_id`, `child_bead_id` | +| `unravel.analysis_started` | Alternatives analysis began | `bead_id` | +| `unravel.analysis_completed` | Alternatives analysis finished | `bead_id`, `duration_ms` | + +### Weave (Documentation Gaps) + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `weave.bead_created` | Documentation gap bead created | `bead_id`, `file`, `line` | +| `weave.analysis_started` | Documentation analysis began | — | +| `weave.analysis_completed` | Documentation analysis finished | `gaps_found`, `duration_ms` | + +### Pulse (Health Monitoring) + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `pulse.bead_created` | Health issue bead created | `bead_id`, `detector` | +| `pulse.scan_started` | Health scan began | — | +| `pulse.scan_completed` | Health scan finished | `issues_found`, `duration_ms` | +| `pulse.issue_detected` | Specific issue found | `detector`, `severity` | +| `pulse.detector_started` | Individual detector started | `detector` | +| `pulse.detector_completed` | Individual detector finished | `detector`, `duration_ms` | + +### Error Events + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `error.claim_failed` | Bead claim failed | `bead_id`, `error` | +| `error.agent_crash` | Agent process crashed | `bead_id`, `error`, `exit_code` | +| `error.timeout` | Operation timed out | `bead_id`, `duration_ms` | +| `error.release_failed` | Bead release failed | `bead_id`, `error` | + +### Effort & Budget + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `effort.recorded` | Effort measurement recorded | `bead_id`, `tokens`, `cost` | +| `budget.warning` | Budget approaching limit | `used`, `limit`, `percentage` | +| `budget.exceeded` | Budget exceeded | `used`, `limit` | +| `budget.per_bead_exceeded` | Per-bead budget exceeded | `bead_id`, `used`, `limit` | + +### File Locks + +| `event_type` | Description | Notable `data` fields | +|---|---|---| +| `file.checkout` | File checked out for editing | `path` | +| `file.conflict` | File conflict detected | `path`, `workers` | +| `file.release` | File lock released | `path` | +| `file.stale` | Stale file lock detected | `path` | +| `lock.priority_bump` | Lock priority bumped | `path`, `worker_id` | +| `lock.priority_bump_received` | Received priority bump notification | `path`, `from_worker` | +| `lock.expired` | Lock expired | `path` | + +## TypeScript Reference + +The canonical TypeScript definitions live in `src/types.ts`: + +- `NeedleEvent` — the wire-schema interface +- `NeedleEventType` — union of all known `event_type` strings +- `NEEDLE_EVENT_SCHEMA_VERSION` — the current protocol version constant + +The parser in `src/parser.ts` validates incoming events against the schema and +throws on version mismatch. Legacy `LogEvent` is retained as an adapter for +backward compatibility with existing UI consumers. diff --git a/src/parser.ts b/src/parser.ts index 162801a..e08a16b 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -8,6 +8,8 @@ import { LogEvent, LogLevel, + NeedleEvent, + NEEDLE_EVENT_SCHEMA_VERSION, ConversationEvent, PromptEvent, ResponseEvent, @@ -17,12 +19,105 @@ import { ConversationParseOptions, } from './types.js'; +/** + * Parse a raw JSON object into a canonical NeedleEvent. + * + * Validates the wire-format fields and asserts schema version compatibility. + * Returns null if the object does not match the NeedleEvent shape. + * + * @throws Error if schema_version is present but doesn't match NEEDLE_EVENT_SCHEMA_VERSION + */ +export function parseNeedleEvent(raw: unknown): NeedleEvent | null { + if (typeof raw !== 'object' || raw === null) return null; + const obj = raw as Record; + + // Required fields + if (typeof obj.timestamp !== 'string') return null; + if (typeof obj.event_type !== 'string') return null; + if (typeof obj.worker_id !== 'string') return null; + if (typeof obj.session_id !== 'string') return null; + if (typeof obj.sequence !== 'number') return null; + + // Schema version assertion — present in newer NEEDLE output + if (obj.schema_version !== undefined && obj.schema_version !== NEEDLE_EVENT_SCHEMA_VERSION) { + throw new Error( + `NeedleEvent schema mismatch: got ${obj.schema_version}, expected ${NEEDLE_EVENT_SCHEMA_VERSION}` + ); + } + + const event: NeedleEvent = { + timestamp: obj.timestamp, + event_type: obj.event_type, + worker_id: obj.worker_id, + session_id: obj.session_id, + sequence: obj.sequence, + data: typeof obj.data === 'object' && obj.data !== null + ? obj.data as Record + : {}, + }; + + if (typeof obj.bead_id === 'string') { + event.bead_id = obj.bead_id; + } + + return event; +} + +/** + * Convert a canonical NeedleEvent into the legacy LogEvent shape. + * + * This adapter preserves backward compatibility with existing UI consumers + * that depend on LogEvent's flat structure. + */ +export function needleEventToLogEvent(ne: NeedleEvent): LogEvent { + const ts = new Date(ne.timestamp).getTime(); + const level = inferLogLevel(ne.event_type); + const event: LogEvent = { + ts, + worker: ne.worker_id, + level, + msg: ne.event_type, + session: ne.session_id, + }; + + if (ne.bead_id) event.bead = ne.bead_id; + + const data = ne.data; + if (typeof data.duration_ms === 'number') event.duration_ms = data.duration_ms; + if (typeof data.error === 'string') event.error = data.error; + if (typeof data.tool === 'string') event.tool = data.tool; + if (typeof data.path === 'string') event.path = data.path; + + for (const key of Object.keys(data)) { + if (!isStandardField(key) && !(key in event)) { + event[key] = data[key]; + } + } + + return event; +} + +/** + * Check if a parsed object matches the canonical NeedleEvent wire format. + * Canonical format uses `timestamp`, `event_type`, `worker_id`, `session_id`, `sequence`. + */ +function isCanonicalNeedleFormat(obj: Record): boolean { + return ( + typeof obj.timestamp === 'string' && + typeof obj.event_type === 'string' && + typeof obj.worker_id === 'string' && + typeof obj.session_id === 'string' && + typeof obj.sequence === 'number' + ); +} + /** * Parse a single log line * - * Supports two formats: - * 1. NEEDLE format: {ts: ISO string, event: string, worker: {...}, session: string, data: {...}} - * 2. Legacy format: {ts: Unix ms, worker: string, level: string, msg: string} + * Supports three formats: + * 1. Canonical NeedleEvent: {timestamp, event_type, worker_id, session_id, sequence, data} + * 2. Legacy NEEDLE format: {ts: ISO string, event: string, worker: {...}, session: string, data: {...}} + * 3. Flat legacy format: {ts: Unix ms, worker: string, level: string, msg: string} * * @param line - Raw log line (JSON string) * @returns Parsed LogEvent or null if invalid @@ -36,7 +131,13 @@ export function parseLogLine(line: string): LogEvent | null { try { const parsed = JSON.parse(line); - // Detect format and route to appropriate parser + // Canonical NeedleEvent wire format (top priority) + if (typeof parsed === 'object' && parsed !== null && isCanonicalNeedleFormat(parsed as Record)) { + const ne = parseNeedleEvent(parsed); + if (ne) return needleEventToLogEvent(ne); + } + + // Legacy NEEDLE format if (isNeedleFormat(parsed)) { return parseNeedleFormat(parsed); } @@ -232,12 +333,19 @@ export function parseEventObject(obj: unknown): LogEvent | null { return null; } - // Check for NEEDLE format + // Canonical NeedleEvent wire format (top priority) + const rec = obj as Record; + if (isCanonicalNeedleFormat(rec)) { + const ne = parseNeedleEvent(obj); + if (ne) return needleEventToLogEvent(ne); + } + + // Legacy NEEDLE format if (isNeedleFormat(obj)) { return parseNeedleFormat(obj); } - // Try as legacy format - validate required fields + // Try as flat legacy format - validate required fields const parsed = obj as Record; if (typeof parsed.ts !== 'number') { return null; diff --git a/src/types.ts b/src/types.ts index 281db22..86f651d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -4,6 +4,13 @@ * Core types for NEEDLE log parsing and worker state management. */ +/** + * Schema version for the NEEDLE event wire format. + * Bumped when the canonical NeedleEvent shape changes. + * Both NEEDLE and FABRIC must agree on this version. + */ +export const NEEDLE_EVENT_SCHEMA_VERSION = 1; + export type LogLevel = 'debug' | 'info' | 'warn' | 'error'; export type WorkerStatus = 'active' | 'idle' | 'error'; @@ -94,6 +101,40 @@ export type NeedleEventType = | 'lock.priority_bump_received' | 'lock.expired'; +// ============================================ +// Canonical NeedleEvent (wire schema) +// ============================================ + +/** + * Canonical event shape emitted by NEEDLE over JSONL and OTLP. + * This is the contract between NEEDLE and FABRIC — versioned via NEEDLE_EVENT_SCHEMA_VERSION. + * + * Ordering: sort by (worker_id, sequence), NOT by timestamp. + * Wall clocks skew across hosts; sequence is the worker's monotonic counter. + */ +export interface NeedleEvent { + /** RFC3339 timestamp — display only, NOT authoritative for ordering */ + timestamp: string; + + /** Event taxonomy string, e.g. "worker.started", "bead.claimed" */ + event_type: NeedleEventType | string; + + /** Worker identifier (e.g. "tcb-alpha") */ + worker_id: string; + + /** Session identifier grouping a worker's lifetime events */ + session_id: string; + + /** Per-worker monotonic counter — authoritative for ordering */ + sequence: number; + + /** Present when event pertains to a specific bead */ + bead_id?: string; + + /** Event-specific payload */ + data: Record; +} + // ============================================ // Conversation Event Types // ============================================