feat(telemetry): implement Telemetry interface — Phase 2

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
 

openspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-04-09

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/TelemetryImpl.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<void>` 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/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 module-level function, not a separate module
+
+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.
+
+## 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.

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/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)
+- 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/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).
+- **Requirements covered**: F-001, F-002, F-003, F-004, F-005, F-006, F-007, F-013

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

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