From 8ec0e9a705154834f97f78c8e751aca2059911c2 Mon Sep 17 00:00:00 2001 From: rsimpson2 Date: Thu, 9 Apr 2026 12:05:32 -0400 Subject: [PATCH 1/2] =?UTF-8?q?feat(telemetry):=20implement=20Telemetry=20?= =?UTF-8?q?interface=20=E2=80=94=20Phase=202=20(emit,=20query,=20getPipeli?= =?UTF-8?q?neSummary)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the core Telemetry interface in @open-forge/telemetry, building on the completed Phase 0 (types) and Phase 1 (storage) foundations. TelemetryImpl class: - emit(): validates required event fields, delegates to StorageBackend, catches storage failures per NF-011 (no pipeline crashes from telemetry) - query(): pure delegation to storage backend with EventFilter support - getPipelineSummary(): single-pass aggregation computing totalDurationMs, totalTokens, stagesCompleted, currentStage, retryCount, and status derivation (halted > failed > completed > running) 39 new tests covering all spec scenarios (72 total pass). Build and lint clean. Docs updated (README, ROADMAP, root status table). Unblocks Phase 3 (Constraint Evaluator) and Phase 4 (Integration API). --- README.md | 12 +- .../.openspec.yaml | 2 + .../2026-04-09-telemetry-phase-2/design.md | 96 ++++ .../2026-04-09-telemetry-phase-2/proposal.md | 32 ++ .../specs/event-emission/spec.md | 76 +++ .../specs/event-query/spec.md | 64 +++ .../specs/pipeline-summary/spec.md | 141 +++++ .../2026-04-09-telemetry-phase-2/tasks.md | 33 ++ openspec/specs/event-emission/spec.md | 76 +++ openspec/specs/event-query/spec.md | 64 +++ openspec/specs/pipeline-summary/spec.md | 141 +++++ packages/telemetry/README.md | 104 +++- packages/telemetry/ROADMAP.md | 14 +- packages/telemetry/src/TelemetryImpl.ts | 145 +++++ packages/telemetry/src/index.ts | 1 + packages/telemetry/tests/telemetry.test.ts | 494 ++++++++++++++++++ 16 files changed, 1479 insertions(+), 16 deletions(-) create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yaml create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-emission/spec.md create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-query/spec.md create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/pipeline-summary/spec.md create mode 100644 openspec/changes/archive/2026-04-09-telemetry-phase-2/tasks.md create mode 100644 openspec/specs/event-emission/spec.md create mode 100644 openspec/specs/event-query/spec.md create mode 100644 openspec/specs/pipeline-summary/spec.md create mode 100644 packages/telemetry/src/TelemetryImpl.ts create mode 100644 packages/telemetry/tests/telemetry.test.ts diff --git a/README.md b/README.md index c4c9275..acea19c 100644 --- a/README.md +++ b/README.md @@ -20,12 +20,12 @@ The differentiator is not the agents. It's the harness: constraints that prevent > **Early development.** The architecture is defined and core packages are scaffolded, but most implementations are in progress. Contributions and feedback are welcome — expect breaking changes. -| Package | Status | -| ------------------------- | --------------------------------------------------------------------------------------------------------------- | -| `@open-forge/pipeline` | Core types, roadmap generation, handoff management, and helper utilities implemented. Orchestrator in progress. | -| `@open-forge/mcp` | Functional — MCP server with spec tools, prompts, and change tracking. | -| `@open-forge/telemetry` | Type definitions exported. Implementations (event emitting, constraint evaluation) not yet started. | -| `@open-forge/evaluations` | Type definitions exported. Implementations planned — blocked on telemetry for §3.6/§3.7 integration. | +| Package | Status | +| ------------------------- | ------------------------------------------------------------------------------------------------------------------------------- | +| `@open-forge/pipeline` | Core types, roadmap generation, handoff management, and helper utilities implemented. Orchestrator in progress. | +| `@open-forge/mcp` | Functional — MCP server with spec tools, prompts, and change tracking. | +| `@open-forge/telemetry` | Types, storage backends, and telemetry core (emit, query, pipeline summary) implemented. Constraint evaluation not yet started. | +| `@open-forge/evaluations` | Type definitions exported. Implementations planned — blocked on telemetry for §3.6/§3.7 integration. | ## Packages diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yaml b/openspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yaml new file mode 100644 index 0000000..98d7681 --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-04-09 diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md new file mode 100644 index 0000000..64bd63b --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md @@ -0,0 +1,96 @@ +## Context + +The telemetry package has completed Phase 0 (types) and Phase 1 (storage backends). The `Telemetry` interface is defined in `src/types.ts` with three methods — `emit()`, `query()`, `getPipelineSummary()` — but has no implementation. Storage backends (`MemoryStorageBackend`, `FileStorageBackend`) implement the `StorageBackend` contract (`append`, `query`, `count`) and are ready to consume. + +The `Telemetry` implementation is the composition layer: it accepts a `StorageBackend` at construction, validates events before persisting, delegates queries to storage, and computes `PipelineSummary` by aggregating stored events. No new types are needed — all interfaces already exist. + +This is a single-module change (`src/telemetry.ts`) with no external dependencies. + +## Goals / Non-Goals + +**Goals:** + +- Implement `emit()` that validates required event fields and delegates to `StorageBackend.append()` +- Implement `query()` that delegates directly to `StorageBackend.query()` (filtering is already handled by storage + `matchesFilter`) +- Implement `getPipelineSummary()` that aggregates all events for a pipeline into a `PipelineSummary` +- Graceful error handling in `emit()` — telemetry failure must not crash the pipeline (NF-011) +- Export `TelemetryImpl` from `src/index.ts` + +**Non-Goals:** + +- Constraint evaluation (Phase 3) +- Factory function / `createTelemetry()` (Phase 4) +- Validation of optional fields (`durationMs`, `tokenUsage`, etc.) beyond type-checking +- Event batching or write coalescing +- Retry logic on storage failure (storage backends handle their own errors) + +## Decisions + +### 1. Constructor injection of `StorageBackend` + +`TelemetryImpl` takes a `StorageBackend` instance via constructor. No default backend selection — the factory (`createTelemetry`, Phase 4) handles that. + +**Rationale**: Keeps `TelemetryImpl` pure and testable. Consumers inject the backend they want. The existing `describeStorageContract` test pattern in `tests/storage.test.ts` demonstrates this same injection approach works well. + +**Alternative considered**: Accept `StorageBackend` per-method. Rejected — the backend is a lifecycle dependency, not per-call configuration. + +### 2. Synchronous validation in `emit()` before async storage + +`emit()` validates required fields (`timestamp`, `pipelineId`, `phase`, `stage`, `action`) synchronously, then calls `storage.append()` asynchronously. Validation errors throw immediately (caller's responsibility). Storage errors are caught and logged per NF-011. + +**Rationale**: Required field validation is deterministic and should fail fast. Storage I/O is where async errors occur and where NF-011 (don't crash pipeline) applies. + +### 3. `emit()` error handling: catch-and-swallow with optional error callback + +On storage failure, `emit()` catches the error and does NOT re-throw. The method resolves normally so the pipeline continues. A future enhancement can add an optional `onError` callback, but for v1 the contract is: telemetry failure is silent to the caller. + +**Rationale**: NF-011 explicitly states the system must not crash the pipeline on telemetry failure. If callers need to know about failures, they can observe via `query()` returning fewer events than expected. + +**Alternative considered**: Return a result type `{ success: boolean; error?: Error }`. Rejected — this would change the `Telemetry` interface signature, which returns `Promise` and is already defined. + +### 4. `query()` is a pure delegation + +`query()` passes the `EventFilter` directly to `storage.query()` and returns the result. No additional filtering, transformation, or caching. + +**Rationale**: The storage layer already implements filtering via `matchesFilter()`. Adding another filter layer would duplicate logic. If future needs require client-side filtering (e.g., on fields not in `EventFilter`), that's a separate concern. + +### 5. `getPipelineSummary()` aggregation strategy + +Query ALL events for the pipeline (filter by `pipelineId` only), then aggregate in memory: + +- `totalDurationMs`: Sum of `durationMs` from events where `action === 'complete'` +- `totalTokens`: Sum of `tokenUsage.prompt + tokenUsage.completion` from all events with `tokenUsage` +- `stagesCompleted`: Unique stage names from events where `action === 'complete'`, preserving insertion order +- `currentStage`: Stage name from the chronologically last event (by `timestamp`) +- `retryCount`: Count of events where `action === 'retry'` +- `status`: Derived from event actions: + - `'halted'` if any event has `action === 'escalate'` + - `'failed'` if any event has `action === 'fail'` (and no escalation) + - `'completed'` if the last event has `action === 'complete'` for the final stage + - `'running'` otherwise + +**Rationale**: Single query, single pass aggregation. No complex joins or multi-query flows. The event set per pipeline is bounded (NF-003: 10,000+ events supported), so in-memory aggregation is fine. + +**Alternative considered**: Multiple targeted queries (one per metric). Rejected — would be slower and more complex for bounded datasets. + +### 6. Single class, single file + +All three methods live in one `TelemetryImpl` class in `src/telemetry.ts`. + +**Rationale**: The class is small (3 methods + constructor + validation helper). No need to split until complexity warrants it. Follows the pattern of `MemoryStorageBackend` — one class per file. + +### 7. Validation helper as a private method, not a separate module + +A private `validateEvent()` method on the class checks required fields. Not extracted to a separate utility. + +**Rationale**: Validation logic is ~10 lines and specific to `TelemetryImpl`. If it grows or needs reuse (e.g., in the factory), extract then. YAGNI. + +## Risks / Trade-offs + +- **Silent emit failures** → Callers won't know if events weren't persisted. Mitigation: the storage backends are reliable (memory never fails, file handles OS errors). If telemetry is critical, Phase 4 can add an `onError` callback. + +- **Full event scan for summary** → `getPipelineSummary()` loads all events for a pipeline into memory. For 10,000 events this is fine (~1-5MB). For future scale (100K+), consider a streaming/accumulator approach. Not a concern for v1. + +- **Timestamp string comparison** → `currentStage` is determined by sorting on `timestamp` (ISO 8601 strings). ISO 8601 is lexicographically sortable, so string comparison is correct. No need for `Date` parsing. + +- **Status derivation edge case** → If a pipeline has both `fail` and `escalate` events, status is `halted` (escalation takes precedence). This matches the pipeline semantics: escalation means human intervention, which overrides any transient failure. diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md new file mode 100644 index 0000000..d9e2470 --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md @@ -0,0 +1,32 @@ +## Why + +Phase 0 (types) and Phase 1 (storage) are complete, but the `Telemetry` interface defined in `src/types.ts` has no implementation. Pipeline stages have no way to emit events, query history, or retrieve aggregated summaries. This is the core value layer — without it, the storage backends have nothing to store and downstream consumers (constraint evaluator, integration API) are blocked. + +## What Changes + +- Add `src/telemetry.ts` implementing the `Telemetry` interface (`emit`, `query`, `getPipelineSummary`) +- `emit()` validates incoming events, delegates to a `StorageBackend`, and handles async errors without crashing the pipeline (NF-011) +- `query()` accepts an `EventFilter` and delegates filtering to the storage backend +- `getPipelineSummary()` aggregates stored events for a pipeline into a `PipelineSummary` (total duration, token usage, completed stages, retry count, status) +- Add `tests/telemetry.test.ts` covering all three methods with both storage backends +- Update `src/index.ts` to export the new `TelemetryImpl` class + +## Capabilities + +### New Capabilities + +- `event-emission`: Validates and persists pipeline events via `emit()`. Enforces required fields (timestamp, pipelineId, phase, stage, action), delegates to pluggable `StorageBackend`, and provides async-safe error handling per NF-011. +- `event-query`: Retrieves filtered pipeline events via `query()`. Accepts `EventFilter` with pipelineId (required), stage, action, and time range — delegates to storage backend's existing `query()` method. +- `pipeline-summary`: Aggregates events into a `PipelineSummary` via `getPipelineSummary()`. Computes totalDurationMs from `complete` events, totalTokens from tokenUsage, stagesCompleted from unique stages, currentStage from latest event, retryCount from `retry` actions, and derives status (`running`/`completed`/`failed`/`halted`) from event actions. + +### Modified Capabilities + +_(No existing specs are modified — this is additive on top of the storage layer.)_ + +## Impact + +- **Code**: New file `src/telemetry.ts`, new test file `tests/telemetry.test.ts`, updated exports in `src/index.ts` +- **Dependencies**: Depends on `StorageBackend` interface and existing implementations (MemoryStorageBackend, FileStorageBackend). No new runtime dependencies. +- **API**: Adds concrete `TelemetryImpl` class implementing the already-defined `Telemetry` interface. No breaking changes — the interface was defined but unimplemented. +- **Downstream**: Unblocks Phase 3 (Constraint Evaluator) and Phase 4 (Integration API / `createTelemetry` factory). +- **Requirements covered**: F-001, F-002, F-003, F-004, F-005, F-006, F-007, F-013 diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-emission/spec.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-emission/spec.md new file mode 100644 index 0000000..08dddab --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-emission/spec.md @@ -0,0 +1,76 @@ +# event-emission Specification + +## Purpose + +Defines how the `TelemetryImpl.emit()` method validates and persists pipeline events. Ensures required fields are present before storage and that storage failures do not crash the calling pipeline. + +## Requirements + +### Requirement: emit validates required event fields + +The `emit()` method SHALL validate that the `PipelineEvent` contains all required fields (`timestamp`, `pipelineId`, `phase`, `stage`, `action`) before attempting storage. If any required field is missing or empty, `emit()` SHALL throw an error. + +#### Scenario: Emit with all required fields + +- **WHEN** `emit()` is called with a `PipelineEvent` containing `timestamp`, `pipelineId`, `phase`, `stage`, and `action` +- **THEN** the event SHALL be persisted to the storage backend + +#### Scenario: Emit rejects missing timestamp + +- **WHEN** `emit()` is called with a `PipelineEvent` where `timestamp` is missing or empty +- **THEN** `emit()` SHALL throw an error +- **AND** the event SHALL NOT be persisted + +#### Scenario: Emit rejects missing pipelineId + +- **WHEN** `emit()` is called with a `PipelineEvent` where `pipelineId` is missing or empty +- **THEN** `emit()` SHALL throw an error +- **AND** the event SHALL NOT be persisted + +#### Scenario: Emit rejects missing stage + +- **WHEN** `emit()` is called with a `PipelineEvent` where `stage` is missing +- **THEN** `emit()` SHALL throw an error + +#### Scenario: Emit rejects missing action + +- **WHEN** `emit()` is called with a `PipelineEvent` where `action` is missing +- **THEN** `emit()` SHALL throw an error + +#### Scenario: Emit accepts event with only required fields + +- **WHEN** `emit()` is called with a `PipelineEvent` containing only the required fields and no optional fields +- **THEN** the event SHALL be persisted successfully + +### Requirement: emit delegates to storage backend + +The `emit()` method SHALL delegate persistence to the injected `StorageBackend.append()` method. + +#### Scenario: Emit appends to injected storage + +- **WHEN** `emit()` is called with a valid event and a `MemoryStorageBackend` +- **THEN** the event SHALL be retrievable via the storage backend's `query()` method + +#### Scenario: Emit works with file storage backend + +- **WHEN** `emit()` is called with a valid event and a `FileStorageBackend` +- **THEN** the event SHALL be persisted to the file and retrievable after recreation + +### Requirement: emit handles storage failures gracefully + +If the storage backend's `append()` method throws an error, `emit()` SHALL catch the error, suppress it, and resolve normally. The pipeline SHALL NOT crash due to a telemetry storage failure. + +#### Scenario: Emit suppresses storage error + +- **WHEN** `emit()` is called and the storage backend's `append()` throws an error +- **THEN** `emit()` SHALL resolve without throwing +- **AND** the pipeline SHALL continue unaffected + +### Requirement: emit preserves optional fields + +When a `PipelineEvent` includes optional fields (`durationMs`, `tokenUsage`, `agentId`, `model`, `filesModified`, `filesCreated`, `error`, `metadata`), `emit()` SHALL pass them through to storage without modification. + +#### Scenario: Emit preserves all optional fields + +- **WHEN** `emit()` is called with an event containing `durationMs`, `tokenUsage`, `agentId`, `model`, `filesModified`, `filesCreated`, `error`, and `metadata` +- **THEN** the persisted event SHALL retain all optional fields with their original values diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-query/spec.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-query/spec.md new file mode 100644 index 0000000..19eb897 --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-query/spec.md @@ -0,0 +1,64 @@ +# event-query Specification + +## Purpose + +Defines how the `TelemetryImpl.query()` method retrieves filtered pipeline events from the storage backend. Query is a pure delegation layer — it accepts an `EventFilter` and returns matching events without additional processing. + +## Requirements + +### Requirement: query delegates to storage backend + +The `query()` method SHALL pass the `EventFilter` directly to the injected `StorageBackend.query()` method and return the results without modification. + +#### Scenario: Query returns matching events + +- **WHEN** events have been emitted and `query()` is called with a matching `EventFilter` +- **THEN** all events matching the filter SHALL be returned + +#### Scenario: Query returns empty for no matches + +- **WHEN** `query()` is called with a filter that matches no events +- **THEN** an empty array SHALL be returned + +### Requirement: query supports all filter fields + +The `query()` method SHALL support filtering by all fields defined in `EventFilter`: `pipelineId` (required), `stage`, `action`, `from`, and `to`. + +#### Scenario: Query by pipelineId only + +- **WHEN** `query()` is called with only `pipelineId` in the filter +- **THEN** all events for that pipeline SHALL be returned + +#### Scenario: Query by pipelineId and stage + +- **WHEN** `query()` is called with `pipelineId` and `stage` +- **THEN** only events matching both SHALL be returned + +#### Scenario: Query by pipelineId and action + +- **WHEN** `query()` is called with `pipelineId` and `action` +- **THEN** only events matching both SHALL be returned + +#### Scenario: Query by time range + +- **WHEN** `query()` is called with `from` and/or `to` timestamps +- **THEN** only events within the specified time range SHALL be returned + +#### Scenario: Query with all filter fields + +- **WHEN** `query()` is called with `pipelineId`, `stage`, `action`, `from`, and `to` +- **THEN** only events matching all criteria SHALL be returned + +### Requirement: query works with any storage backend + +The `query()` method SHALL work identically regardless of which `StorageBackend` implementation is injected. + +#### Scenario: Query with memory storage + +- **WHEN** `TelemetryImpl` is constructed with a `MemoryStorageBackend` +- **THEN** `query()` SHALL return filtered events from memory + +#### Scenario: Query with file storage + +- **WHEN** `TelemetryImpl` is constructed with a `FileStorageBackend` +- **THEN** `query()` SHALL return filtered events from the JSONL file diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/pipeline-summary/spec.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/pipeline-summary/spec.md new file mode 100644 index 0000000..976e258 --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/pipeline-summary/spec.md @@ -0,0 +1,141 @@ +# pipeline-summary Specification + +## Purpose + +Defines how the `TelemetryImpl.getPipelineSummary()` method aggregates stored events into a `PipelineSummary`. Computes total duration, token usage, completed stages, current stage, retry count, and derives pipeline status from event actions. + +## Requirements + +### Requirement: getPipelineSummary computes totalDurationMs + +The `getPipelineSummary()` method SHALL sum the `durationMs` field from all events where `action` is `'complete'`. + +#### Scenario: Duration from complete events + +- **WHEN** events include `complete` actions with `durationMs` of 100, 200, and 300 +- **THEN** `totalDurationMs` SHALL be 600 + +#### Scenario: Zero duration with no complete events + +- **WHEN** no events have `action` equal to `'complete'` +- **THEN** `totalDurationMs` SHALL be 0 + +#### Scenario: Duration excludes non-complete events + +- **WHEN** events include `start`, `fail`, and `retry` actions with `durationMs` values +- **THEN** only `complete` events SHALL contribute to `totalDurationMs` + +### Requirement: getPipelineSummary computes totalTokens + +The `getPipelineSummary()` method SHALL sum `tokenUsage.prompt + tokenUsage.completion` from all events that include `tokenUsage`. + +#### Scenario: Token sum from multiple events + +- **WHEN** events have `tokenUsage` of `{ prompt: 100, completion: 50 }` and `{ prompt: 200, completion: 100 }` +- **THEN** `totalTokens` SHALL be 450 + +#### Scenario: Zero tokens with no token usage + +- **WHEN** no events include `tokenUsage` +- **THEN** `totalTokens` SHALL be 0 + +#### Scenario: Tokens from events without tokenUsage are skipped + +- **WHEN** some events have `tokenUsage` and others do not +- **THEN** only events with `tokenUsage` SHALL contribute to `totalTokens` + +### Requirement: getPipelineSummary computes stagesCompleted + +The `getPipelineSummary()` method SHALL collect unique stage names from events where `action` is `'complete'`, preserving insertion order. + +#### Scenario: Completed stages from complete events + +- **WHEN** events show `complete` actions for stages `'plan'`, `'test'`, and `'implement'` +- **THEN** `stagesCompleted` SHALL be `['plan', 'test', 'implement']` + +#### Scenario: Deduplicates repeated stage completions + +- **WHEN** events show `complete` for `'plan'` twice and `'test'` once +- **THEN** `stagesCompleted` SHALL be `['plan', 'test']` + +#### Scenario: Empty stages with no completions + +- **WHEN** no events have `action` equal to `'complete'` +- **THEN** `stagesCompleted` SHALL be an empty array + +### Requirement: getPipelineSummary determines currentStage + +The `getPipelineSummary()` method SHALL set `currentStage` to the `stage` value of the chronologically last event (by `timestamp`). + +#### Scenario: Current stage is last event's stage + +- **WHEN** the last event by timestamp has `stage` equal to `'implement'` +- **THEN** `currentStage` SHALL be `'implement'` + +#### Scenario: Current stage undefined with no events + +- **WHEN** no events exist for the pipeline +- **THEN** `currentStage` SHALL be `undefined` + +### Requirement: getPipelineSummary computes retryCount + +The `getPipelineSummary()` method SHALL count events where `action` is `'retry'`. + +#### Scenario: Retry count from retry events + +- **WHEN** events include 3 events with `action` equal to `'retry'` +- **THEN** `retryCount` SHALL be 3 + +#### Scenario: Zero retries + +- **WHEN** no events have `action` equal to `'retry'` +- **THEN** `retryCount` SHALL be 0 + +### Requirement: getPipelineSummary derives pipeline status + +The `getPipelineSummary()` method SHALL derive `status` from event actions using this precedence: + +1. `'halted'` if any event has `action === 'escalate'` +2. `'failed'` if any event has `action === 'fail'` (and no escalation) +3. `'completed'` if the last event by timestamp has `action === 'complete'` +4. `'running'` otherwise + +#### Scenario: Status halted on escalation + +- **WHEN** any event has `action` equal to `'escalate'` +- **THEN** `status` SHALL be `'halted'` + +#### Scenario: Status failed on failure without escalation + +- **WHEN** any event has `action` equal to `'fail'` and no event has `action` equal to `'escalate'` +- **THEN** `status` SHALL be `'failed'` + +#### Scenario: Status completed on final complete + +- **WHEN** the last event by timestamp has `action` equal to `'complete'` +- **AND** no `fail` or `escalate` events exist +- **THEN** `status` SHALL be `'completed'` + +#### Scenario: Status running for in-progress pipeline + +- **WHEN** events exist but none are `escalate`, `fail`, or a final `complete` +- **THEN** `status` SHALL be `'running'` + +#### Scenario: Escalation takes precedence over failure + +- **WHEN** events include both `fail` and `escalate` actions +- **THEN** `status` SHALL be `'halted'` + +#### Scenario: Status running with no events + +- **WHEN** no events exist for the pipeline +- **THEN** `status` SHALL be `'running'` + +### Requirement: getPipelineSummary returns pipelineId + +The `getPipelineSummary()` method SHALL set `pipelineId` to the value passed as the method argument. + +#### Scenario: PipelineId matches argument + +- **WHEN** `getPipelineSummary('pipeline-42')` is called +- **THEN** the returned `PipelineSummary.pipelineId` SHALL be `'pipeline-42'` diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/tasks.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/tasks.md new file mode 100644 index 0000000..70d96f2 --- /dev/null +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/tasks.md @@ -0,0 +1,33 @@ +## 1. emit() — TDD Implementation + +- [x] 1.1 Write failing tests for `emit()` validation: rejects missing timestamp, missing pipelineId, missing stage, missing action, empty timestamp, empty pipelineId — all throw and do not persist +- [x] 1.2 Write failing tests for `emit()` delegation: valid event persists to MemoryStorageBackend, preserves all optional fields (durationMs, tokenUsage, agentId, model, filesModified, filesCreated, error, metadata), accepts event with only required fields +- [x] 1.3 Write failing test for `emit()` error handling: storage backend append() throws → emit() resolves without throwing (NF-011) +- [x] 1.4 Implement `TelemetryImpl` class in `src/TelemetryImpl.ts` — constructor accepting `StorageBackend`, private `validateEvent()` method, `emit()` with sync validation + async storage delegation + try/catch on append +- [x] 1.5 Verify all emit tests pass, existing tests still pass + +## 2. query() — TDD Implementation + +- [x] 2.1 Write failing tests for `query()` delegation: returns matching events, returns empty for no matches +- [x] 2.2 Write failing tests for `query()` filter fields: filter by pipelineId only, by pipelineId+stage, by pipelineId+action, by time range (from/to), with all fields combined +- [x] 2.3 Implement `query()` on `TelemetryImpl` — pure delegation to `storage.query(filter)` +- [x] 2.4 Verify all query tests pass, all prior tests still pass + +## 3. getPipelineSummary() — TDD Implementation + +- [x] 3.1 Write failing tests for `totalDurationMs`: sums durationMs from complete events, returns 0 with no complete events, excludes non-complete events +- [x] 3.2 Write failing tests for `totalTokens`: sums prompt+completion across events with tokenUsage, returns 0 with no tokenUsage, skips events without tokenUsage +- [x] 3.3 Write failing tests for `stagesCompleted`: unique stages from complete events, deduplicates repeated completions, empty array with no completions +- [x] 3.4 Write failing tests for `currentStage`: last event's stage by timestamp, undefined with no events +- [x] 3.5 Write failing tests for `retryCount`: counts retry actions, returns 0 with no retries +- [x] 3.6 Write failing tests for `status` derivation: halted on escalate, failed on fail without escalate, completed on last complete, running for in-progress, escalation precedes failure, running with no events +- [x] 3.7 Write failing test for `pipelineId`: returned summary has pipelineId matching argument +- [x] 3.8 Implement `getPipelineSummary()` on `TelemetryImpl` — query all events by pipelineId, single-pass aggregation computing all PipelineSummary fields +- [x] 3.9 Verify all summary tests pass, all prior tests still pass + +## 4. Exports and Quality Gates + +- [x] 4.1 Update `src/index.ts` to export `TelemetryImpl` from `./TelemetryImpl.js` +- [x] 4.2 Run `npx nx build telemetry` — verify build passes with no errors +- [x] 4.3 Run `npx nx test telemetry` — verify all tests pass (existing + new) +- [x] 4.4 Run `npx nx lint telemetry` — verify no lint errors diff --git a/openspec/specs/event-emission/spec.md b/openspec/specs/event-emission/spec.md new file mode 100644 index 0000000..08dddab --- /dev/null +++ b/openspec/specs/event-emission/spec.md @@ -0,0 +1,76 @@ +# event-emission Specification + +## Purpose + +Defines how the `TelemetryImpl.emit()` method validates and persists pipeline events. Ensures required fields are present before storage and that storage failures do not crash the calling pipeline. + +## Requirements + +### Requirement: emit validates required event fields + +The `emit()` method SHALL validate that the `PipelineEvent` contains all required fields (`timestamp`, `pipelineId`, `phase`, `stage`, `action`) before attempting storage. If any required field is missing or empty, `emit()` SHALL throw an error. + +#### Scenario: Emit with all required fields + +- **WHEN** `emit()` is called with a `PipelineEvent` containing `timestamp`, `pipelineId`, `phase`, `stage`, and `action` +- **THEN** the event SHALL be persisted to the storage backend + +#### Scenario: Emit rejects missing timestamp + +- **WHEN** `emit()` is called with a `PipelineEvent` where `timestamp` is missing or empty +- **THEN** `emit()` SHALL throw an error +- **AND** the event SHALL NOT be persisted + +#### Scenario: Emit rejects missing pipelineId + +- **WHEN** `emit()` is called with a `PipelineEvent` where `pipelineId` is missing or empty +- **THEN** `emit()` SHALL throw an error +- **AND** the event SHALL NOT be persisted + +#### Scenario: Emit rejects missing stage + +- **WHEN** `emit()` is called with a `PipelineEvent` where `stage` is missing +- **THEN** `emit()` SHALL throw an error + +#### Scenario: Emit rejects missing action + +- **WHEN** `emit()` is called with a `PipelineEvent` where `action` is missing +- **THEN** `emit()` SHALL throw an error + +#### Scenario: Emit accepts event with only required fields + +- **WHEN** `emit()` is called with a `PipelineEvent` containing only the required fields and no optional fields +- **THEN** the event SHALL be persisted successfully + +### Requirement: emit delegates to storage backend + +The `emit()` method SHALL delegate persistence to the injected `StorageBackend.append()` method. + +#### Scenario: Emit appends to injected storage + +- **WHEN** `emit()` is called with a valid event and a `MemoryStorageBackend` +- **THEN** the event SHALL be retrievable via the storage backend's `query()` method + +#### Scenario: Emit works with file storage backend + +- **WHEN** `emit()` is called with a valid event and a `FileStorageBackend` +- **THEN** the event SHALL be persisted to the file and retrievable after recreation + +### Requirement: emit handles storage failures gracefully + +If the storage backend's `append()` method throws an error, `emit()` SHALL catch the error, suppress it, and resolve normally. The pipeline SHALL NOT crash due to a telemetry storage failure. + +#### Scenario: Emit suppresses storage error + +- **WHEN** `emit()` is called and the storage backend's `append()` throws an error +- **THEN** `emit()` SHALL resolve without throwing +- **AND** the pipeline SHALL continue unaffected + +### Requirement: emit preserves optional fields + +When a `PipelineEvent` includes optional fields (`durationMs`, `tokenUsage`, `agentId`, `model`, `filesModified`, `filesCreated`, `error`, `metadata`), `emit()` SHALL pass them through to storage without modification. + +#### Scenario: Emit preserves all optional fields + +- **WHEN** `emit()` is called with an event containing `durationMs`, `tokenUsage`, `agentId`, `model`, `filesModified`, `filesCreated`, `error`, and `metadata` +- **THEN** the persisted event SHALL retain all optional fields with their original values diff --git a/openspec/specs/event-query/spec.md b/openspec/specs/event-query/spec.md new file mode 100644 index 0000000..19eb897 --- /dev/null +++ b/openspec/specs/event-query/spec.md @@ -0,0 +1,64 @@ +# event-query Specification + +## Purpose + +Defines how the `TelemetryImpl.query()` method retrieves filtered pipeline events from the storage backend. Query is a pure delegation layer — it accepts an `EventFilter` and returns matching events without additional processing. + +## Requirements + +### Requirement: query delegates to storage backend + +The `query()` method SHALL pass the `EventFilter` directly to the injected `StorageBackend.query()` method and return the results without modification. + +#### Scenario: Query returns matching events + +- **WHEN** events have been emitted and `query()` is called with a matching `EventFilter` +- **THEN** all events matching the filter SHALL be returned + +#### Scenario: Query returns empty for no matches + +- **WHEN** `query()` is called with a filter that matches no events +- **THEN** an empty array SHALL be returned + +### Requirement: query supports all filter fields + +The `query()` method SHALL support filtering by all fields defined in `EventFilter`: `pipelineId` (required), `stage`, `action`, `from`, and `to`. + +#### Scenario: Query by pipelineId only + +- **WHEN** `query()` is called with only `pipelineId` in the filter +- **THEN** all events for that pipeline SHALL be returned + +#### Scenario: Query by pipelineId and stage + +- **WHEN** `query()` is called with `pipelineId` and `stage` +- **THEN** only events matching both SHALL be returned + +#### Scenario: Query by pipelineId and action + +- **WHEN** `query()` is called with `pipelineId` and `action` +- **THEN** only events matching both SHALL be returned + +#### Scenario: Query by time range + +- **WHEN** `query()` is called with `from` and/or `to` timestamps +- **THEN** only events within the specified time range SHALL be returned + +#### Scenario: Query with all filter fields + +- **WHEN** `query()` is called with `pipelineId`, `stage`, `action`, `from`, and `to` +- **THEN** only events matching all criteria SHALL be returned + +### Requirement: query works with any storage backend + +The `query()` method SHALL work identically regardless of which `StorageBackend` implementation is injected. + +#### Scenario: Query with memory storage + +- **WHEN** `TelemetryImpl` is constructed with a `MemoryStorageBackend` +- **THEN** `query()` SHALL return filtered events from memory + +#### Scenario: Query with file storage + +- **WHEN** `TelemetryImpl` is constructed with a `FileStorageBackend` +- **THEN** `query()` SHALL return filtered events from the JSONL file diff --git a/openspec/specs/pipeline-summary/spec.md b/openspec/specs/pipeline-summary/spec.md new file mode 100644 index 0000000..976e258 --- /dev/null +++ b/openspec/specs/pipeline-summary/spec.md @@ -0,0 +1,141 @@ +# pipeline-summary Specification + +## Purpose + +Defines how the `TelemetryImpl.getPipelineSummary()` method aggregates stored events into a `PipelineSummary`. Computes total duration, token usage, completed stages, current stage, retry count, and derives pipeline status from event actions. + +## Requirements + +### Requirement: getPipelineSummary computes totalDurationMs + +The `getPipelineSummary()` method SHALL sum the `durationMs` field from all events where `action` is `'complete'`. + +#### Scenario: Duration from complete events + +- **WHEN** events include `complete` actions with `durationMs` of 100, 200, and 300 +- **THEN** `totalDurationMs` SHALL be 600 + +#### Scenario: Zero duration with no complete events + +- **WHEN** no events have `action` equal to `'complete'` +- **THEN** `totalDurationMs` SHALL be 0 + +#### Scenario: Duration excludes non-complete events + +- **WHEN** events include `start`, `fail`, and `retry` actions with `durationMs` values +- **THEN** only `complete` events SHALL contribute to `totalDurationMs` + +### Requirement: getPipelineSummary computes totalTokens + +The `getPipelineSummary()` method SHALL sum `tokenUsage.prompt + tokenUsage.completion` from all events that include `tokenUsage`. + +#### Scenario: Token sum from multiple events + +- **WHEN** events have `tokenUsage` of `{ prompt: 100, completion: 50 }` and `{ prompt: 200, completion: 100 }` +- **THEN** `totalTokens` SHALL be 450 + +#### Scenario: Zero tokens with no token usage + +- **WHEN** no events include `tokenUsage` +- **THEN** `totalTokens` SHALL be 0 + +#### Scenario: Tokens from events without tokenUsage are skipped + +- **WHEN** some events have `tokenUsage` and others do not +- **THEN** only events with `tokenUsage` SHALL contribute to `totalTokens` + +### Requirement: getPipelineSummary computes stagesCompleted + +The `getPipelineSummary()` method SHALL collect unique stage names from events where `action` is `'complete'`, preserving insertion order. + +#### Scenario: Completed stages from complete events + +- **WHEN** events show `complete` actions for stages `'plan'`, `'test'`, and `'implement'` +- **THEN** `stagesCompleted` SHALL be `['plan', 'test', 'implement']` + +#### Scenario: Deduplicates repeated stage completions + +- **WHEN** events show `complete` for `'plan'` twice and `'test'` once +- **THEN** `stagesCompleted` SHALL be `['plan', 'test']` + +#### Scenario: Empty stages with no completions + +- **WHEN** no events have `action` equal to `'complete'` +- **THEN** `stagesCompleted` SHALL be an empty array + +### Requirement: getPipelineSummary determines currentStage + +The `getPipelineSummary()` method SHALL set `currentStage` to the `stage` value of the chronologically last event (by `timestamp`). + +#### Scenario: Current stage is last event's stage + +- **WHEN** the last event by timestamp has `stage` equal to `'implement'` +- **THEN** `currentStage` SHALL be `'implement'` + +#### Scenario: Current stage undefined with no events + +- **WHEN** no events exist for the pipeline +- **THEN** `currentStage` SHALL be `undefined` + +### Requirement: getPipelineSummary computes retryCount + +The `getPipelineSummary()` method SHALL count events where `action` is `'retry'`. + +#### Scenario: Retry count from retry events + +- **WHEN** events include 3 events with `action` equal to `'retry'` +- **THEN** `retryCount` SHALL be 3 + +#### Scenario: Zero retries + +- **WHEN** no events have `action` equal to `'retry'` +- **THEN** `retryCount` SHALL be 0 + +### Requirement: getPipelineSummary derives pipeline status + +The `getPipelineSummary()` method SHALL derive `status` from event actions using this precedence: + +1. `'halted'` if any event has `action === 'escalate'` +2. `'failed'` if any event has `action === 'fail'` (and no escalation) +3. `'completed'` if the last event by timestamp has `action === 'complete'` +4. `'running'` otherwise + +#### Scenario: Status halted on escalation + +- **WHEN** any event has `action` equal to `'escalate'` +- **THEN** `status` SHALL be `'halted'` + +#### Scenario: Status failed on failure without escalation + +- **WHEN** any event has `action` equal to `'fail'` and no event has `action` equal to `'escalate'` +- **THEN** `status` SHALL be `'failed'` + +#### Scenario: Status completed on final complete + +- **WHEN** the last event by timestamp has `action` equal to `'complete'` +- **AND** no `fail` or `escalate` events exist +- **THEN** `status` SHALL be `'completed'` + +#### Scenario: Status running for in-progress pipeline + +- **WHEN** events exist but none are `escalate`, `fail`, or a final `complete` +- **THEN** `status` SHALL be `'running'` + +#### Scenario: Escalation takes precedence over failure + +- **WHEN** events include both `fail` and `escalate` actions +- **THEN** `status` SHALL be `'halted'` + +#### Scenario: Status running with no events + +- **WHEN** no events exist for the pipeline +- **THEN** `status` SHALL be `'running'` + +### Requirement: getPipelineSummary returns pipelineId + +The `getPipelineSummary()` method SHALL set `pipelineId` to the value passed as the method argument. + +#### Scenario: PipelineId matches argument + +- **WHEN** `getPipelineSummary('pipeline-42')` is called +- **THEN** the returned `PipelineSummary.pipelineId` SHALL be `'pipeline-42'` diff --git a/packages/telemetry/README.md b/packages/telemetry/README.md index 2267563..e357494 100644 --- a/packages/telemetry/README.md +++ b/packages/telemetry/README.md @@ -1,6 +1,104 @@ -# telemetry +# @open-forge/telemetry -This library was generated with [Nx](https://nx.dev). +Structured event capture, querying, and constraint evaluation for open-forge pipeline stages. + +## Installation + +```bash +npm install @open-forge/telemetry +``` + +## Quick Start + +```typescript +import { TelemetryImpl, MemoryStorageBackend } from '@open-forge/telemetry'; + +const storage = new MemoryStorageBackend(); +const telemetry = new TelemetryImpl(storage); + +// Emit events from pipeline stages +await telemetry.emit({ + timestamp: new Date().toISOString(), + pipelineId: 'my-pipeline', + phase: 1, + stage: 'plan', + action: 'start', +}); + +await telemetry.emit({ + timestamp: new Date().toISOString(), + pipelineId: 'my-pipeline', + phase: 1, + stage: 'plan', + action: 'complete', + durationMs: 1500, + tokenUsage: { prompt: 100, completion: 200 }, +}); + +// Query events +const events = await telemetry.query({ + pipelineId: 'my-pipeline', + stage: 'plan', +}); + +// Get aggregated summary +const summary = await telemetry.getPipelineSummary('my-pipeline'); +console.log(summary); +// { +// pipelineId: 'my-pipeline', +// totalDurationMs: 1500, +// totalTokens: 300, +// stagesCompleted: ['plan'], +// currentStage: 'plan', +// retryCount: 0, +// status: 'completed' +// } +``` + +## Storage Backends + +Two storage backends are included: + +- **`MemoryStorageBackend`** — in-memory, ephemeral. Good for testing and lightweight use. +- **`FileStorageBackend`** — JSONL append-only file. Zero-config production backend; persists events across restarts. + +```typescript +import { FileStorageBackend } from '@open-forge/telemetry'; + +const storage = new FileStorageBackend('./telemetry-events.jsonl'); +const telemetry = new TelemetryImpl(storage); +``` + +## API + +### `TelemetryImpl` + +Implements the `Telemetry` interface. Construct with a `StorageBackend` instance. + +| Method | Description | +| -------------------------------- | ------------------------------------------------------------------------------------------------------------------------- | +| `emit(event)` | Validate and persist a `PipelineEvent`. Throws on missing required fields. Storage failures are caught silently (NF-011). | +| `query(filter)` | Retrieve events matching an `EventFilter` (pipelineId, stage, action, time range). | +| `getPipelineSummary(pipelineId)` | Aggregate all events for a pipeline into a `PipelineSummary` (duration, tokens, stages, retries, status). | + +### `PipelineEvent` Required Fields + +| Field | Type | +| ------------ | ---------------------------------------------------------- | +| `timestamp` | `string` (ISO 8601) | +| `pipelineId` | `string` | +| `phase` | `number` | +| `stage` | `StageName` | +| `action` | `'start' \| 'complete' \| 'fail' \| 'retry' \| 'escalate'` | + +### `PipelineSummary` Status Derivation + +| Status | Condition | +| ----------- | ------------------------------------------- | +| `halted` | Any `escalate` event exists | +| `failed` | Any `fail` event (no escalation) | +| `completed` | Last event is `complete` (no fail/escalate) | +| `running` | Otherwise | ## Building @@ -8,4 +106,4 @@ Run `nx build telemetry` to build the library. ## Running unit tests -Run `nx test telemetry` to execute the unit tests via [Vitest](https://vitest.dev/). +Run `nx test telemetry` to execute the unit tests via [Vitest](https://vitest.dev). diff --git a/packages/telemetry/ROADMAP.md b/packages/telemetry/ROADMAP.md index df65461..a26c7fa 100644 --- a/packages/telemetry/ROADMAP.md +++ b/packages/telemetry/ROADMAP.md @@ -45,16 +45,16 @@ This roadmap breaks down the [telemetry REQUIREMENTS](../../docs/open-forge-tele --- -## Phase 2: Telemetry Core +## Phase 2: Telemetry Core (COMPLETE) **Goal**: Implement event emission and querying — the `Telemetry` interface. ### Tasks -- [ ] 2.1 Implement `emit()` [deps: 1.2] [deliverable: `src/telemetry.ts` — validates event, writes to storage, async-safe] -- [ ] 2.2 Implement `query()` [deps: 2.1] [deliverable: `src/telemetry.ts` — filter by pipelineId, stage, action, time range] -- [ ] 2.3 Implement `getPipelineSummary()` [deps: 2.2] [deliverable: `src/telemetry.ts` — aggregate events into PipelineSummary] -- [ ] 2.4 Write telemetry tests [deps: 2.1, 2.2, 2.3] [deliverable: `tests/telemetry.test.ts`] +- [x] 2.1 Implement `emit()` [deps: 1.2] [deliverable: `src/TelemetryImpl.ts` — validates event, writes to storage, async-safe] +- [x] 2.2 Implement `query()` [deps: 2.1] [deliverable: `src/TelemetryImpl.ts` — filter by pipelineId, stage, action, time range] +- [x] 2.3 Implement `getPipelineSummary()` [deps: 2.2] [deliverable: `src/TelemetryImpl.ts` — aggregate events into PipelineSummary] +- [x] 2.4 Write telemetry tests [deps: 2.1, 2.2, 2.3] [deliverable: `tests/telemetry.test.ts` — 39 tests] **Parallel Groups**: @@ -129,9 +129,9 @@ This roadmap breaks down the [telemetry REQUIREMENTS](../../docs/open-forge-tele ``` Phase 0 (Foundation) ✅ │ - └──→ Phase 1 (Storage) + └──→ Phase 1 (Storage) ✅ │ - └──→ Phase 2 (Telemetry Core) + └──→ Phase 2 (Telemetry Core) ✅ │ └──→ Phase 3 (Constraints) │ diff --git a/packages/telemetry/src/TelemetryImpl.ts b/packages/telemetry/src/TelemetryImpl.ts new file mode 100644 index 0000000..46263b8 --- /dev/null +++ b/packages/telemetry/src/TelemetryImpl.ts @@ -0,0 +1,145 @@ +import type { + PipelineEvent, + EventFilter, + PipelineSummary, + StageName, + Telemetry, +} from './types.js'; +import type { StorageBackend } from './storage/StorageBackend.js'; + +export class TelemetryImpl implements Telemetry { + constructor(private readonly storage: StorageBackend) {} + + async emit(event: PipelineEvent): Promise { + validateEvent(event); + try { + await this.storage.append(event); + } catch { + return; + } + } + + async query(filter: EventFilter): Promise { + return this.storage.query(filter); + } + + async getPipelineSummary(pipelineId: string): Promise { + const events = await this.storage.query({ pipelineId }); + return aggregateSummary(pipelineId, events); + } +} + +function validateEvent(event: PipelineEvent): void { + if (!event.timestamp) { + throw new Error('PipelineEvent.timestamp is required'); + } + if (!event.pipelineId) { + throw new Error('PipelineEvent.pipelineId is required'); + } + if (event.stage === undefined) { + throw new Error('PipelineEvent.stage is required'); + } + if (event.action === undefined) { + throw new Error('PipelineEvent.action is required'); + } +} + +interface AggregationState { + totalDurationMs: number; + totalTokens: number; + stagesSeen: Set; + stagesCompleted: StageName[]; + currentStage: StageName | undefined; + latestTimestamp: string; + retryCount: number; + hasEscalate: boolean; + hasFail: boolean; +} + +function aggregateSummary( + pipelineId: string, + events: PipelineEvent[] +): PipelineSummary { + const state: AggregationState = { + totalDurationMs: 0, + totalTokens: 0, + stagesSeen: new Set(), + stagesCompleted: [], + currentStage: undefined, + latestTimestamp: '', + retryCount: 0, + hasEscalate: false, + hasFail: false, + }; + + for (const event of events) { + accumulateEvent(state, event); + } + + return { + pipelineId, + totalDurationMs: state.totalDurationMs, + totalTokens: state.totalTokens, + stagesCompleted: state.stagesCompleted, + currentStage: state.currentStage, + retryCount: state.retryCount, + status: deriveStatus({ + events, + hasEscalate: state.hasEscalate, + hasFail: state.hasFail, + latestTimestamp: state.latestTimestamp, + }), + }; +} + +function accumulateEvent(state: AggregationState, event: PipelineEvent): void { + if (event.action === 'complete' && event.durationMs !== undefined) { + state.totalDurationMs += event.durationMs; + } + + if (event.tokenUsage !== undefined) { + state.totalTokens += event.tokenUsage.prompt + event.tokenUsage.completion; + } + + if (event.action === 'complete' && !state.stagesSeen.has(event.stage)) { + state.stagesSeen.add(event.stage); + state.stagesCompleted.push(event.stage); + } + + if (event.timestamp > state.latestTimestamp) { + state.latestTimestamp = event.timestamp; + state.currentStage = event.stage; + } + + if (event.action === 'retry') { + state.retryCount++; + } + + if (event.action === 'escalate') { + state.hasEscalate = true; + } + + if (event.action === 'fail') { + state.hasFail = true; + } +} + +interface StatusInput { + events: PipelineEvent[]; + hasEscalate: boolean; + hasFail: boolean; + latestTimestamp: string; +} + +function deriveStatus(input: StatusInput): PipelineSummary['status'] { + if (input.events.length === 0) return 'running'; + if (input.hasEscalate) return 'halted'; + if (input.hasFail) return 'failed'; + + const lastEvent = input.events.find( + (e) => e.timestamp === input.latestTimestamp + ); + if (lastEvent?.action === 'complete') return 'completed'; + + return 'running'; +} diff --git a/packages/telemetry/src/index.ts b/packages/telemetry/src/index.ts index cc9da6a..db12a01 100644 --- a/packages/telemetry/src/index.ts +++ b/packages/telemetry/src/index.ts @@ -16,3 +16,4 @@ export type { export type { StorageBackend } from './storage/StorageBackend.js'; export { MemoryStorageBackend } from './storage/MemoryStorageBackend.js'; export { FileStorageBackend } from './storage/FileStorageBackend.js'; +export { TelemetryImpl } from './TelemetryImpl.js'; diff --git a/packages/telemetry/tests/telemetry.test.ts b/packages/telemetry/tests/telemetry.test.ts new file mode 100644 index 0000000..2297379 --- /dev/null +++ b/packages/telemetry/tests/telemetry.test.ts @@ -0,0 +1,494 @@ +import { describe, it, expect, beforeEach } from 'vitest'; +import type { PipelineEvent } from '../src/types.js'; +import type { StorageBackend } from '../src/storage/StorageBackend.js'; +import { MemoryStorageBackend } from '../src/storage/MemoryStorageBackend.js'; +import { TelemetryImpl } from '../src/TelemetryImpl.js'; + +function makeEvent(overrides: Partial = {}): PipelineEvent { + return { + timestamp: '2024-06-15T10:00:00.000Z', + pipelineId: 'pipeline-1', + phase: 1, + stage: 'plan', + action: 'start', + ...overrides, + }; +} + +// --------------------------------------------------------------------------- +// 1. emit() — Validation +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — emit() validation', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should throw on missing timestamp', async () => { + const event = makeEvent({ timestamp: undefined as unknown as string }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + + it('should throw on empty timestamp', async () => { + const event = makeEvent({ timestamp: '' }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + + it('should throw on missing pipelineId', async () => { + const event = makeEvent({ pipelineId: undefined as unknown as string }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + + it('should throw on empty pipelineId', async () => { + const event = makeEvent({ pipelineId: '' }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + + it('should throw on missing stage', async () => { + const event = makeEvent({ + stage: undefined as unknown as PipelineEvent['stage'], + }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + + it('should throw on missing action', async () => { + const event = makeEvent({ + action: undefined as unknown as PipelineEvent['action'], + }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + + it('should not persist event when validation fails', async () => { + const badEvent = makeEvent({ pipelineId: '' }); + await expect(telemetry.emit(badEvent)).rejects.toThrow(); + + const results = await telemetry.query({ pipelineId: 'pipeline-1' }); + expect(results).toHaveLength(0); + }); +}); + +// --------------------------------------------------------------------------- +// 2. emit() — Delegation to storage +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — emit() delegation', () => { + let storage: StorageBackend; + let telemetry: TelemetryImpl; + + beforeEach(() => { + storage = new MemoryStorageBackend(); + telemetry = new TelemetryImpl(storage); + }); + + it('should persist a valid event to storage', async () => { + const event = makeEvent(); + await telemetry.emit(event); + + const results = await storage.query({ pipelineId: 'pipeline-1' }); + expect(results).toHaveLength(1); + expect(results[0]).toEqual(event); + }); + + it('should persist event with only required fields', async () => { + const event: PipelineEvent = { + timestamp: '2024-06-15T10:00:00.000Z', + pipelineId: 'pipeline-1', + phase: 1, + stage: 'plan', + action: 'start', + }; + await telemetry.emit(event); + + const results = await storage.query({ pipelineId: 'pipeline-1' }); + expect(results).toHaveLength(1); + expect(results[0]).toEqual(event); + }); + + it('should preserve all optional fields', async () => { + const event = makeEvent({ + durationMs: 1500, + tokenUsage: { prompt: 100, completion: 200 }, + agentId: 'agent-1', + model: 'gpt-4', + filesModified: ['src/foo.ts'], + filesCreated: ['src/bar.ts'], + error: 'some error', + metadata: { key: 'value' }, + }); + await telemetry.emit(event); + + const results = await storage.query({ pipelineId: 'pipeline-1' }); + expect(results).toHaveLength(1); + expect(results[0]).toEqual(event); + }); + + it('should persist multiple events', async () => { + await telemetry.emit(makeEvent({ action: 'start' })); + await telemetry.emit(makeEvent({ action: 'complete' })); + + const results = await storage.query({ pipelineId: 'pipeline-1' }); + expect(results).toHaveLength(2); + }); +}); + +// --------------------------------------------------------------------------- +// 3. emit() — Error handling (NF-011) +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — emit() error handling', () => { + it('should not throw when storage append fails', async () => { + const failingStorage: StorageBackend = { + append: async () => { + throw new Error('disk full'); + }, + query: async () => [], + count: async () => 0, + }; + const telemetry = new TelemetryImpl(failingStorage); + + // Should resolve without throwing + await expect(telemetry.emit(makeEvent())).resolves.toBeUndefined(); + }); +}); + +// --------------------------------------------------------------------------- +// 4. query() — Delegation +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — query() delegation', () => { + let telemetry: TelemetryImpl; + + beforeEach(async () => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + await telemetry.emit( + makeEvent({ pipelineId: 'p1', stage: 'plan', action: 'start' }) + ); + await telemetry.emit( + makeEvent({ pipelineId: 'p1', stage: 'plan', action: 'complete' }) + ); + await telemetry.emit( + makeEvent({ pipelineId: 'p1', stage: 'test', action: 'start' }) + ); + await telemetry.emit( + makeEvent({ pipelineId: 'p2', stage: 'plan', action: 'start' }) + ); + }); + + it('should return matching events by pipelineId', async () => { + const results = await telemetry.query({ pipelineId: 'p1' }); + expect(results).toHaveLength(3); + }); + + it('should return empty array for no matches', async () => { + const results = await telemetry.query({ pipelineId: 'nonexistent' }); + expect(results).toEqual([]); + }); + + it('should filter by pipelineId and stage', async () => { + const results = await telemetry.query({ pipelineId: 'p1', stage: 'plan' }); + expect(results).toHaveLength(2); + }); + + it('should filter by pipelineId and action', async () => { + const results = await telemetry.query({ + pipelineId: 'p1', + action: 'complete', + }); + expect(results).toHaveLength(1); + expect(results[0]?.action).toBe('complete'); + }); + + it('should filter by time range (from)', async () => { + const results = await telemetry.query({ + pipelineId: 'p1', + from: '2024-06-15T10:00:00.000Z', + }); + expect(results).toHaveLength(3); + }); + + it('should filter by time range (to)', async () => { + const results = await telemetry.query({ + pipelineId: 'p1', + to: '2024-06-15T10:00:00.000Z', + }); + // All events have the same timestamp, so to is inclusive + expect(results).toHaveLength(3); + }); + + it('should filter with all fields combined', async () => { + const results = await telemetry.query({ + pipelineId: 'p1', + stage: 'plan', + action: 'start', + from: '2024-06-15T00:00:00.000Z', + to: '2024-06-16T00:00:00.000Z', + }); + expect(results).toHaveLength(1); + }); +}); + +// --------------------------------------------------------------------------- +// 5. getPipelineSummary() — totalDurationMs +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() totalDurationMs', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should sum durationMs from complete events', async () => { + await telemetry.emit(makeEvent({ action: 'start', stage: 'plan' })); + await telemetry.emit( + makeEvent({ action: 'complete', stage: 'plan', durationMs: 100 }) + ); + await telemetry.emit(makeEvent({ action: 'start', stage: 'test' })); + await telemetry.emit( + makeEvent({ action: 'complete', stage: 'test', durationMs: 200 }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.totalDurationMs).toBe(300); + }); + + it('should return 0 when no complete events', async () => { + await telemetry.emit(makeEvent({ action: 'start' })); + await telemetry.emit(makeEvent({ action: 'fail' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.totalDurationMs).toBe(0); + }); + + it('should exclude durationMs from non-complete events', async () => { + await telemetry.emit(makeEvent({ action: 'start', durationMs: 999 })); + await telemetry.emit(makeEvent({ action: 'fail', durationMs: 888 })); + await telemetry.emit(makeEvent({ action: 'retry', durationMs: 777 })); + await telemetry.emit( + makeEvent({ action: 'complete', stage: 'plan', durationMs: 100 }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.totalDurationMs).toBe(100); + }); +}); + +// --------------------------------------------------------------------------- +// 6. getPipelineSummary() — totalTokens +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() totalTokens', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should sum tokenUsage across events', async () => { + await telemetry.emit( + makeEvent({ tokenUsage: { prompt: 100, completion: 50 } }) + ); + await telemetry.emit( + makeEvent({ tokenUsage: { prompt: 200, completion: 100 } }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.totalTokens).toBe(450); + }); + + it('should return 0 when no events have tokenUsage', async () => { + await telemetry.emit(makeEvent()); + await telemetry.emit(makeEvent()); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.totalTokens).toBe(0); + }); + + it('should skip events without tokenUsage', async () => { + await telemetry.emit( + makeEvent({ tokenUsage: { prompt: 100, completion: 50 } }) + ); + await telemetry.emit(makeEvent()); // no tokenUsage + await telemetry.emit( + makeEvent({ tokenUsage: { prompt: 50, completion: 25 } }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.totalTokens).toBe(225); + }); +}); + +// --------------------------------------------------------------------------- +// 7. getPipelineSummary() — stagesCompleted +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() stagesCompleted', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should list unique stages from complete events', async () => { + await telemetry.emit(makeEvent({ action: 'complete', stage: 'plan' })); + await telemetry.emit(makeEvent({ action: 'complete', stage: 'test' })); + await telemetry.emit(makeEvent({ action: 'complete', stage: 'implement' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.stagesCompleted).toEqual(['plan', 'test', 'implement']); + }); + + it('should deduplicate repeated completions', async () => { + await telemetry.emit(makeEvent({ action: 'complete', stage: 'plan' })); + await telemetry.emit(makeEvent({ action: 'complete', stage: 'plan' })); + await telemetry.emit(makeEvent({ action: 'complete', stage: 'test' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.stagesCompleted).toEqual(['plan', 'test']); + }); + + it('should return empty array with no completions', async () => { + await telemetry.emit(makeEvent({ action: 'start' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.stagesCompleted).toEqual([]); + }); +}); + +// --------------------------------------------------------------------------- +// 8. getPipelineSummary() — currentStage +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() currentStage', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should return stage of chronologically last event', async () => { + await telemetry.emit( + makeEvent({ stage: 'plan', timestamp: '2024-01-01T00:00:00.000Z' }) + ); + await telemetry.emit( + makeEvent({ stage: 'test', timestamp: '2024-01-02T00:00:00.000Z' }) + ); + await telemetry.emit( + makeEvent({ stage: 'implement', timestamp: '2024-01-03T00:00:00.000Z' }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.currentStage).toBe('implement'); + }); + + it('should return undefined when no events exist', async () => { + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.currentStage).toBeUndefined(); + }); +}); + +// --------------------------------------------------------------------------- +// 9. getPipelineSummary() — retryCount +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() retryCount', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should count retry events', async () => { + await telemetry.emit(makeEvent({ action: 'retry' })); + await telemetry.emit(makeEvent({ action: 'retry' })); + await telemetry.emit(makeEvent({ action: 'retry' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.retryCount).toBe(3); + }); + + it('should return 0 with no retries', async () => { + await telemetry.emit(makeEvent({ action: 'start' })); + await telemetry.emit(makeEvent({ action: 'complete' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.retryCount).toBe(0); + }); +}); + +// --------------------------------------------------------------------------- +// 10. getPipelineSummary() — status derivation +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() status', () => { + let telemetry: TelemetryImpl; + + beforeEach(() => { + telemetry = new TelemetryImpl(new MemoryStorageBackend()); + }); + + it('should be halted when escalate action exists', async () => { + await telemetry.emit(makeEvent({ action: 'fail' })); + await telemetry.emit(makeEvent({ action: 'escalate' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.status).toBe('halted'); + }); + + it('should be failed when fail action exists without escalate', async () => { + await telemetry.emit(makeEvent({ action: 'start' })); + await telemetry.emit(makeEvent({ action: 'fail' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.status).toBe('failed'); + }); + + it('should be completed when last event is complete', async () => { + await telemetry.emit( + makeEvent({ action: 'start', timestamp: '2024-01-01T00:00:00.000Z' }) + ); + await telemetry.emit( + makeEvent({ action: 'complete', timestamp: '2024-01-02T00:00:00.000Z' }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.status).toBe('completed'); + }); + + it('should be running when in-progress', async () => { + await telemetry.emit( + makeEvent({ action: 'start', timestamp: '2024-01-01T00:00:00.000Z' }) + ); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.status).toBe('running'); + }); + + it('should have escalation take precedence over failure', async () => { + await telemetry.emit(makeEvent({ action: 'fail' })); + await telemetry.emit(makeEvent({ action: 'escalate' })); + await telemetry.emit(makeEvent({ action: 'complete' })); + + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.status).toBe('halted'); + }); + + it('should be running with no events', async () => { + const summary = await telemetry.getPipelineSummary('pipeline-1'); + expect(summary.status).toBe('running'); + }); +}); + +// --------------------------------------------------------------------------- +// 11. getPipelineSummary() — pipelineId +// --------------------------------------------------------------------------- + +describe('TelemetryImpl — getPipelineSummary() pipelineId', () => { + it('should return the pipelineId from the argument', async () => { + const telemetry = new TelemetryImpl(new MemoryStorageBackend()); + const summary = await telemetry.getPipelineSummary('pipeline-42'); + expect(summary.pipelineId).toBe('pipeline-42'); + }); +}); From 887256a111f890751de657167ad545a4c99dd3ea Mon Sep 17 00:00:00 2001 From: rsimpson2 Date: Thu, 9 Apr 2026 12:19:02 -0400 Subject: [PATCH 2/2] fix(telemetry): address PR #28 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add phase validation to validateEvent() (Codex + Copilot) - Fix equal-timestamp ambiguity: use >= instead of > for currentStage derivation so last-appended event wins (Copilot) - Add test for missing phase field (Copilot) - Fix stale src/telemetry.ts references in proposal.md and design.md to src/TelemetryImpl.ts (Copilot) - Fix design.md Decision #7: private method → module-level function (Copilot) --- .../archive/2026-04-09-telemetry-phase-2/design.md | 8 ++++---- .../archive/2026-04-09-telemetry-phase-2/proposal.md | 4 ++-- packages/telemetry/src/TelemetryImpl.ts | 5 ++++- packages/telemetry/tests/telemetry.test.ts | 7 +++++++ 4 files changed, 17 insertions(+), 7 deletions(-) diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md index 64bd63b..9a22ef3 100644 --- a/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md @@ -4,7 +4,7 @@ The telemetry package has completed Phase 0 (types) and Phase 1 (storage backend The `Telemetry` implementation is the composition layer: it accepts a `StorageBackend` at construction, validates events before persisting, delegates queries to storage, and computes `PipelineSummary` by aggregating stored events. No new types are needed — all interfaces already exist. -This is a single-module change (`src/telemetry.ts`) with no external dependencies. +This is a single-module change (`src/TelemetryImpl.ts`) with no external dependencies. ## Goals / Non-Goals @@ -75,13 +75,13 @@ Query ALL events for the pipeline (filter by `pipelineId` only), then aggregate ### 6. Single class, single file -All three methods live in one `TelemetryImpl` class in `src/telemetry.ts`. +All three methods live in one `TelemetryImpl` class in `src/TelemetryImpl.ts`. **Rationale**: The class is small (3 methods + constructor + validation helper). No need to split until complexity warrants it. Follows the pattern of `MemoryStorageBackend` — one class per file. -### 7. Validation helper as a private method, not a separate module +### 7. Validation helper as a module-level function, not a separate module -A private `validateEvent()` method on the class checks required fields. Not extracted to a separate utility. +A module-level `validateEvent()` function checks required fields. Not extracted to a separate utility. **Rationale**: Validation logic is ~10 lines and specific to `TelemetryImpl`. If it grows or needs reuse (e.g., in the factory), extract then. YAGNI. diff --git a/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md b/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md index d9e2470..6f0dea0 100644 --- a/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md +++ b/openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md @@ -4,7 +4,7 @@ Phase 0 (types) and Phase 1 (storage) are complete, but the `Telemetry` interfac ## What Changes -- Add `src/telemetry.ts` implementing the `Telemetry` interface (`emit`, `query`, `getPipelineSummary`) +- Add `src/TelemetryImpl.ts` implementing the `Telemetry` interface (`emit`, `query`, `getPipelineSummary`) - `emit()` validates incoming events, delegates to a `StorageBackend`, and handles async errors without crashing the pipeline (NF-011) - `query()` accepts an `EventFilter` and delegates filtering to the storage backend - `getPipelineSummary()` aggregates stored events for a pipeline into a `PipelineSummary` (total duration, token usage, completed stages, retry count, status) @@ -25,7 +25,7 @@ _(No existing specs are modified — this is additive on top of the storage laye ## Impact -- **Code**: New file `src/telemetry.ts`, new test file `tests/telemetry.test.ts`, updated exports in `src/index.ts` +- **Code**: New file `src/TelemetryImpl.ts`, new test file `tests/telemetry.test.ts`, updated exports in `src/index.ts` - **Dependencies**: Depends on `StorageBackend` interface and existing implementations (MemoryStorageBackend, FileStorageBackend). No new runtime dependencies. - **API**: Adds concrete `TelemetryImpl` class implementing the already-defined `Telemetry` interface. No breaking changes — the interface was defined but unimplemented. - **Downstream**: Unblocks Phase 3 (Constraint Evaluator) and Phase 4 (Integration API / `createTelemetry` factory). diff --git a/packages/telemetry/src/TelemetryImpl.ts b/packages/telemetry/src/TelemetryImpl.ts index 46263b8..fa752e4 100644 --- a/packages/telemetry/src/TelemetryImpl.ts +++ b/packages/telemetry/src/TelemetryImpl.ts @@ -39,6 +39,9 @@ function validateEvent(event: PipelineEvent): void { if (event.stage === undefined) { throw new Error('PipelineEvent.stage is required'); } + if (event.phase === undefined) { + throw new Error('PipelineEvent.phase is required'); + } if (event.action === undefined) { throw new Error('PipelineEvent.action is required'); } @@ -106,7 +109,7 @@ function accumulateEvent(state: AggregationState, event: PipelineEvent): void { state.stagesCompleted.push(event.stage); } - if (event.timestamp > state.latestTimestamp) { + if (event.timestamp >= state.latestTimestamp) { state.latestTimestamp = event.timestamp; state.currentStage = event.stage; } diff --git a/packages/telemetry/tests/telemetry.test.ts b/packages/telemetry/tests/telemetry.test.ts index 2297379..ca1e9bf 100644 --- a/packages/telemetry/tests/telemetry.test.ts +++ b/packages/telemetry/tests/telemetry.test.ts @@ -60,6 +60,13 @@ describe('TelemetryImpl — emit() validation', () => { await expect(telemetry.emit(event)).rejects.toThrow(); }); + it('should throw on missing phase', async () => { + const event = makeEvent({ + phase: undefined as unknown as number, + }); + await expect(telemetry.emit(event)).rejects.toThrow(); + }); + it('should not persist event when validation fails', async () => { const badEvent = makeEvent({ pipelineId: '' }); await expect(telemetry.emit(badEvent)).rejects.toThrow();