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 <noreply@anthropic.com>
This commit is contained in:
jedarden
parent
adb3d68366
commit
10146b6415
4 changed files with 346 additions and 6 deletions
|
|
@ -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)
|
||||
|
|
|
|||
190
docs/schema.md
Normal file
190
docs/schema.md
Normal file
|
|
@ -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.
|
||||
120
src/parser.ts
120
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<string, unknown>;
|
||||
|
||||
// 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<string, unknown>
|
||||
: {},
|
||||
};
|
||||
|
||||
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<string, unknown>): 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<string, unknown>)) {
|
||||
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<string, unknown>;
|
||||
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<string, unknown>;
|
||||
if (typeof parsed.ts !== 'number') {
|
||||
return null;
|
||||
|
|
|
|||
41
src/types.ts
41
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<string, unknown>;
|
||||
}
|
||||
|
||||
// ============================================
|
||||
// Conversation Event Types
|
||||
// ============================================
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue