feat(telemetry): implement Telemetry interface — Phase 2#28
Merged
Conversation
…, getPipelineSummary) 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).
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (4)
✅ Files skipped from review due to trivial changes (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughIntroduces Telemetry Phase 2: adds Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant TelemetryImpl
participant StorageBackend
participant Aggregator
Client->>TelemetryImpl: emit(event)
TelemetryImpl->>TelemetryImpl: validate required fields
TelemetryImpl->>StorageBackend: append(event)
StorageBackend-->>TelemetryImpl: success / throw
alt append throws
TelemetryImpl->>TelemetryImpl: log and swallow error
end
TelemetryImpl-->>Client: resolve
Client->>TelemetryImpl: query(filter)
TelemetryImpl->>StorageBackend: query(filter)
StorageBackend-->>TelemetryImpl: events[]
TelemetryImpl-->>Client: events[]
Client->>TelemetryImpl: getPipelineSummary(pipelineId)
TelemetryImpl->>StorageBackend: query({pipelineId})
StorageBackend-->>TelemetryImpl: events[]
TelemetryImpl->>Aggregator: aggregate events (duration, tokens, stages, retries, status)
Aggregator-->>TelemetryImpl: PipelineSummary
TelemetryImpl-->>Client: PipelineSummary
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
|
View your CI Pipeline Execution ↗ for commit 887256a
☁️ Nx Cloud last updated this comment at |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8ec0e9a705
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Comment on lines
+36
to
+40
| if (!event.pipelineId) { | ||
| throw new Error('PipelineEvent.pipelineId is required'); | ||
| } | ||
| if (event.stage === undefined) { | ||
| throw new Error('PipelineEvent.stage is required'); |
There was a problem hiding this comment.
phase before appending events
emit() is intended to reject events missing required fields, but validateEvent() checks timestamp, pipelineId, stage, and action while never validating phase. This allows malformed runtime inputs (for example, JS callers or deserialized payloads with phase: undefined) to be persisted, which violates the required-event contract and can corrupt downstream summary/constraint logic that assumes phase is always present.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Pull request overview
This PR delivers Phase 2 of the telemetry package by adding a concrete TelemetryImpl that composes existing storage backends to support event emission, querying, and pipeline summaries, along with updated docs/spec artifacts.
Changes:
- Added
TelemetryImplimplementingemit,query, andgetPipelineSummaryon top ofStorageBackend. - Added a comprehensive Vitest suite covering emission validation/delegation/error handling, query delegation, and summary aggregation/status derivation.
- Updated package exports and documentation/specs to reflect Telemetry Phase 2 completion.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Updates repository status table to reflect telemetry core implementation. |
| packages/telemetry/src/TelemetryImpl.ts | New Telemetry implementation (emit/query/summary aggregation). |
| packages/telemetry/tests/telemetry.test.ts | New tests covering TelemetryImpl behavior. |
| packages/telemetry/src/index.ts | Exports TelemetryImpl from the package entrypoint. |
| packages/telemetry/README.md | Replaces Nx boilerplate with usage docs and API tables. |
| packages/telemetry/ROADMAP.md | Marks Phase 2 tasks as complete and updates deliverable references. |
| openspec/specs/event-emission/spec.md | Adds spec for emit() validation/delegation/error handling behavior. |
| openspec/specs/event-query/spec.md | Adds spec for query() delegation and filter support. |
| openspec/specs/pipeline-summary/spec.md | Adds spec for summary aggregation fields and status precedence. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yaml | Archives spec-driven change metadata. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/tasks.md | Archives the Phase 2 task checklist and quality gates. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md | Archives the Phase 2 proposal and claimed file-level impact. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md | Archives the Phase 2 design decisions and trade-offs. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-emission/spec.md | Archived copy of event-emission spec. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-query/spec.md | Archived copy of event-query spec. |
| openspec/changes/archive/2026-04-09-telemetry-phase-2/specs/pipeline-summary/spec.md | Archived copy of pipeline-summary spec. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+32
to
+45
| 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'); | ||
| } | ||
| } |
There was a problem hiding this comment.
emit() validation is missing the phase required-field check. Specs and docs state phase is required, but validateEvent() currently only checks timestamp, pipelineId, stage, and action, so JS callers can pass phase: undefined and the event will be persisted.
Comment on lines
+109
to
+143
| 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'; | ||
|
|
There was a problem hiding this comment.
currentStage/status derivation treats only strictly-greater timestamps as newer, and deriveStatus() uses find() by timestamp. If multiple events share the same timestamp, the "last event by timestamp" becomes ambiguous and the implementation will pick the first matching event rather than the latest appended one. Consider tracking the latest event itself during accumulation and using >= (or a tie-breaker) so equal-timestamp events resolve deterministically.
Comment on lines
+22
to
+69
| 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); | ||
| }); |
There was a problem hiding this comment.
The emit() validation test suite doesn’t cover the spec-required phase field. Add a case where phase is missing/undefined and assert emit() throws and the event is not persisted (mirrors the timestamp/pipelineId tests).
Comment on lines
+5
to
+12
| ## 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 |
There was a problem hiding this comment.
This archive proposal still references src/telemetry.ts, but the implementation in this PR is src/TelemetryImpl.ts. Update the file path references so the archived change artifacts accurately reflect what was merged.
|
|
||
| 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. |
There was a problem hiding this comment.
This design doc refers to the implementation file as src/telemetry.ts (and mentions a private validateEvent() method on the class), but the PR adds src/TelemetryImpl.ts with validateEvent as a helper function. Update these references to match the actual implementation for traceability.
Suggested change
| 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. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (1)
packages/telemetry/src/TelemetryImpl.ts (1)
13-20: Error suppression is intentional per NF-011, but consider logging.The catch block suppresses storage errors to prevent pipeline crashes (NF-011 requirement). While this is the specified behavior, consider adding a comment or future enhancement for an
onErrorcallback (as mentioned in the design doc) so telemetry failures aren't completely silent in production.💡 Optional: Add comment documenting intentional suppression
async emit(event: PipelineEvent): Promise<void> { validateEvent(event); try { await this.storage.append(event); - } catch { + } catch { + // NF-011: Telemetry failure must not crash pipeline - error suppressed intentionally return; } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/telemetry/src/TelemetryImpl.ts` around lines 13 - 20, The catch in async emit(event: PipelineEvent) intentionally suppresses errors from this.storage.append to satisfy NF-011, but add an explicit comment above the try/catch in TelemetryImpl.emit referencing NF-011 and noting the suppression is intentional; also add a TODO or short note about implementing the proposed onError callback from the design doc for non-silent telemetry failures (mention emit, validateEvent, and storage.append so reviewers can find the code).
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@openspec/changes/archive/2026-04-09-telemetry-phase-2/design.md`:
- Around line 76-80: The doc references the wrong filename casing for the
Telemetry implementation; update the mention of `src/telemetry.ts` to the
correct `src/TelemetryImpl.ts` wherever it appears (specifically in the sentence
describing the `TelemetryImpl` class) so the file name matches the actual
class/file naming convention and other references like `MemoryStorageBackend`.
- Line 7: The design doc incorrectly references src/telemetry.ts while the
implementation is named src/TelemetryImpl.ts; update the document to reference
the actual implementation filename (TelemetryImpl.ts) consistently wherever
src/telemetry.ts appears and, if applicable, also align casing/module-name
references with the exported class or symbol (e.g., TelemetryImpl) so readers
can locate the implementation accurately.
In `@openspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.md`:
- Line 7: The proposal references src/telemetry.ts but the implementation lives
in src/TelemetryImpl.ts; update the proposal to reference the actual filename
(TelemetryImpl.ts) or rename the implementation to telemetry.ts for consistency,
and ensure references to the Telemetry interface and its methods (Telemetry,
emit, query, getPipelineSummary) match the chosen filename throughout the
document (lines noting the file at 7, 11, 28) so readers can locate the
implementation unambiguously.
In `@openspec/specs/event-emission/spec.md`:
- Line 11: The validateEvent() function in TelemetryImpl.ts is missing
validation for PipelineEvent.phase; update validateEvent() to check if
event.phase is undefined (or otherwise invalid) and throw an Error (e.g.,
"PipelineEvent.phase is required") so emit() enforces the spec's required fields
(timestamp, pipelineId, phase, stage, action).
---
Nitpick comments:
In `@packages/telemetry/src/TelemetryImpl.ts`:
- Around line 13-20: The catch in async emit(event: PipelineEvent) intentionally
suppresses errors from this.storage.append to satisfy NF-011, but add an
explicit comment above the try/catch in TelemetryImpl.emit referencing NF-011
and noting the suppression is intentional; also add a TODO or short note about
implementing the proposed onError callback from the design doc for non-silent
telemetry failures (mention emit, validateEvent, and storage.append so reviewers
can find the code).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: a582bc51-92b6-4734-a107-06a3c791f64e
📒 Files selected for processing (16)
README.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/.openspec.yamlopenspec/changes/archive/2026-04-09-telemetry-phase-2/design.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/proposal.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-emission/spec.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/specs/event-query/spec.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/specs/pipeline-summary/spec.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/tasks.mdopenspec/specs/event-emission/spec.mdopenspec/specs/event-query/spec.mdopenspec/specs/pipeline-summary/spec.mdpackages/telemetry/README.mdpackages/telemetry/ROADMAP.mdpackages/telemetry/src/TelemetryImpl.tspackages/telemetry/src/index.tspackages/telemetry/tests/telemetry.test.ts
- 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)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Telemetryinterface (emit,query,getPipelineSummary) inTelemetryImplclass, building on Phase 0 (types) and Phase 1 (storage)emit()validates required event fields and delegates to pluggableStorageBackend, catching storage failures per NF-011 (no pipeline crashes from telemetry)query()provides pure delegation to storage backend with fullEventFiltersupport (pipelineId, stage, action, time range)getPipelineSummary()aggregates events intoPipelineSummarywith single-pass computation of totalDurationMs, totalTokens, stagesCompleted, currentStage, retryCount, and status derivation (halted > failed > completed > running)Changes
packages/telemetry/src/TelemetryImpl.tspackages/telemetry/tests/telemetry.test.tspackages/telemetry/src/index.tsTelemetryImplexportpackages/telemetry/README.mdpackages/telemetry/ROADMAP.mdREADME.mdopenspec/specs/{event-emission,event-query,pipeline-summary}/spec.mdopenspec/changes/archive/2026-04-09-telemetry-phase-2/Test plan
npx nx build telemetrycleannpx nx lint telemetryclean (0 errors)Summary by CodeRabbit
New Features
Documentation
Tests