fix(growth): validate telemetry and align PostHog dashboards (#1050)

Author: blove

apps/website/src/app/api/ingest/route.spec.ts

@@ -70,6 +70,66 @@ describe('/api/ingest', () => {
     expect(response.status).toBe(400);
     expect(response.headers.get('access-control-allow-origin')).toBe('*');
   });
+
+  it.each([
+    { event: 'tplane:invented', properties: { transport: 'custom' } },
+    { event: 'tplane:postinstall', properties: {} },
+    { event: 'tplane:stream_started', properties: {} },
+    { event: 'tplane:stream_started', properties: 'secret' },
+    { event: 'tplane:stream_started', properties: ['secret'] },
+    { event: 'tplane:stream_started', properties: { transport: 1 } },
+    { event: 'tplane:browser_chat_init', properties: {} },
+    { event: 'tplane:stream_ended', properties: { transport: 'custom', durationMs: -1 } },
+  ])('rejects malformed public payloads without capturing or echoing input', async (payload) => {
+    const response = await POST(new Request('https://threadplane.ai/api/ingest', {
+      method: 'POST', body: JSON.stringify({ distinctId: 'test', ...payload }),
+    }) as never);
+    expect(response.status).toBe(400);
+    expect(response.headers.get('access-control-allow-origin')).toBe('*');
+    expect(await response.json()).toEqual({ error: 'Invalid event payload' });
+    expect(capture).not.toHaveBeenCalled();
+  });
+
+  it('accepts canonical runtime events while excluding arbitrary fields and person profiles', async () => {
+    const response = await POST(new Request('https://threadplane.ai/api/ingest', {
+      method: 'POST', body: JSON.stringify({
+        distinctId: 'browser:test', event: 'tplane:stream_ended',
+        properties: {
+          transport: 'langgraph', surface: 'canonical_demo', durationMs: 120,
+          '0': 'private', command: 'private', body: 'private', token: 'private',
+          $set: { email: 'private' }, $ip: '1.2.3.4', $process_person_profile: true,
+        },
+      }),
+    }) as never);
+    expect(response.status).toBe(202);
+    expect(capture).toHaveBeenCalledWith({
+      distinctId: 'browser:test', event: 'tplane:stream_ended',
+      properties: { transport: 'langgraph', surface: 'canonical_demo', durationMs: 120, $ip: null, $process_person_profile: false },
+    });
+  });
+
+  it('rejects an oversized streamed body even without content-length', async () => {
+    const response = await POST(new Request('https://threadplane.ai/api/ingest', {
+      method: 'POST', body: JSON.stringify({ distinctId: 'test', event: 'tplane:browser_provided', properties: { body: 'x'.repeat(16_384) } }),
+    }) as never);
+    expect(response.status).toBe(413);
+    expect(response.headers.get('access-control-allow-origin')).toBe('*');
+    expect(capture).not.toHaveBeenCalled();
+  });
+
+  it('reports provider failure without logging the raw exception', async () => {
+    const log = vi.spyOn(console, 'error').mockImplementation(() => undefined);
+    shutdown.mockRejectedValueOnce(new Error('private provider response'));
+    try {
+      const response = await POST(new Request('https://threadplane.ai/api/ingest', {
+        method: 'POST', body: JSON.stringify({ distinctId: 'test', event: 'tplane:browser_provided', properties: {} }),
+      }) as never);
+      expect(response.status).toBe(502);
+      expect(log).toHaveBeenCalledWith('[telemetry-ingest] capture failed');
+    } finally {
+      log.mockRestore();
+    }
+  });
 });
 
 /**

apps/website/src/app/api/ingest/route.ts

@@ -1,8 +1,10 @@
 import { PostHog } from 'posthog-node';
 import { NextRequest, NextResponse } from 'next/server';
-import { normalizePostHogHost, toSafeAnalyticsString } from '@threadplane/telemetry/shared';
+import { normalizePostHogHost, parseTelemetryEvent, toSafeAnalyticsString } from '@threadplane/telemetry/shared';
+import { readBoundedBody } from '../_internal/read-bounded-body';
 
 const PUBLIC_INGEST_KEY = 'phc_public_cacheplane_telemetry';
+const MAX_BODY_BYTES = 16_384;
 const CORS_HEADERS = {
   'Access-Control-Allow-Origin': '*',
   'Access-Control-Allow-Methods': 'POST, OPTIONS',
@@ -41,13 +43,13 @@ function readPayload(value: unknown): {
   if (payload.key !== undefined && payload.key !== PUBLIC_INGEST_KEY) return null;
 
   const distinctId = toSafeAnalyticsString(payload.distinctId, 200);
-  const event = toSafeAnalyticsString(payload.event, 100);
-  if (!distinctId || !event?.startsWith('tplane:')) return null;
+  const parsed = parseTelemetryEvent(payload.event, payload.properties);
+  if (!distinctId || !parsed) return null;
 
   return {
     distinctId,
-    event,
-    properties: isRecord(payload.properties) ? payload.properties : {},
+    event: parsed.event,
+    properties: parsed.properties,
   };
 }
 
@@ -65,7 +67,9 @@ export function OPTIONS(): NextResponse {
 export async function POST(req: NextRequest) {
   let body: unknown;
   try {
-    body = await req.json();
+    const rawBody = await readBoundedBody(req, MAX_BODY_BYTES);
+    if (rawBody === null) return jsonWithCors({ error: 'Invalid request body' }, { status: 413 });
+    body = JSON.parse(rawBody);
   } catch {
     return jsonWithCors({ error: 'Invalid JSON' }, { status: 400 });
   }
@@ -98,8 +102,8 @@ export async function POST(req: NextRequest) {
     });
     await posthog.shutdown();
     return jsonWithCors({ ok: true }, { status: 202 });
-  } catch (err) {
-    console.error('[telemetry-ingest] capture failed:', err);
+  } catch {
+    console.error('[telemetry-ingest] capture failed');
     await posthog.shutdown().catch(() => undefined);
     return jsonWithCors(
       { error: 'Event ingest failed' },

docs/growth/README.md

@@ -92,10 +92,13 @@ search reporting also have no in-repo recurring worker.
 
 The Growth funnel is an observation/activation report, not a complete sequential
 anonymous conversion funnel. A UTM is not a proven link from a social post to a
-developer identity. Current PostHog report parsing needs separate improvement for
-funnels and multiple trend series; do not treat unsupported or missing data as
-proof of zero activity. An analytics contract failure does not itself prove a
-lifecycle delivery failure.
+developer identity. PostHog's Quick overview separates acquisition, docs, demo
+and public runtime signals; install copy attempts do not measure npm installs.
+The runtime dashboard includes demo usage and historical malformed events.
+The weekly report separates additive daily series and marks funnels, unique
+counts, breakdowns and missing results `Unavailable`. An analytics contract
+failure does not itself prove a lifecycle delivery failure. See the
+[dashboard inventory and measurement limits](../../tools/posthog/README.md#current-growth-dashboards).
 
 ## Contributor checks
 

docs/gtm/taxonomy.md

@@ -36,7 +36,7 @@ The standard PostHog `$pageview` event is used as-is across all three surfaces.
 | `marketing:lead_form_submit`          | Submit attempt (any surface)                                                                  |
 | `marketing:lead_form_success`         | Server 2xx                                                                                    |
 | `marketing:lead_form_fail`            | Server non-2xx                                                                                |
-| `marketing:lead_qualified`            | Server-side enrichment passes (qualified-lead def)                                            |
+| `marketing:lead_qualified`            | Historical only: retired qualification emitter. Current evidence and authorization live in Growth. |
 | `marketing:newsletter_signup_submit`  | Submit attempt                                                                                |
 | `marketing:newsletter_signup_success` | Server 2xx                                                                                    |
 | `marketing:newsletter_signup_fail`    | Failure                                                                                       |
@@ -50,6 +50,18 @@ The standard PostHog `$pageview` event is used as-is across all three surfaces.
 | `blog:copy_code_click`                | Copy-button click on a code block inside a blog post. Props: `surface: 'blog'`, `code_lang?`. |
 | `docs:tab_select`                     | MDX tab change                                                                                |
 | `docs:sidebar_section_toggle`         | Sidebar nav toggle                                                                            |
+| `docs:workspace_navigation` | Workspace capability navigation; `capability`, `category`, `from_capability`, `surface`. |
+| `docs:workspace_mode_switched` | Workspace mode change; `capability`, `from_mode`, `to_mode`, `surface`. |
+| `docs:workspace_runtime_action` | Explicit runtime action; `capability`, `action`, `state_before`, `outcome`, `surface`. |
+| `docs:workspace_runtime_status_changed` | Runtime status transition; `capability`, `from_state`, `to_state`, optional `elapsed_ms`/`reason_code`, `surface`. |
+| `marketing:stage_progress` | Recorded homepage stage progress; `surface`, `stage_event`, optional `beat`. Not a live developer runtime. |
+
+Current dashboards distinguish website intent, client-observed form acceptance,
+and independent demo milestones. `hero_install` is a copy attempt recorded before
+clipboard success, not an npm install. The former six-signal activation funnel
+does not represent Growth's install/runtime activation and is no longer managed.
+Actual install/runtime activation, enrichment, authorization and email outcomes
+remain authoritative in Neon; see [Growth operations](../growth/README.md).
 
 ## Cockpit (activation surface)
 

libs/telemetry/README.md

@@ -151,6 +151,25 @@ await captureEvent('tplane:runtime_instance_created', {
 The runtime adapter helpers exported from `@threadplane/telemetry/node` are
 convenience wrappers around the same explicit capture path.
 
+The Node capture path and Threadplane's public ingest endpoint validate the seven
+documented SDK event names at runtime. Runtime events require `transport`;
+`tplane:browser_chat_init` requires `surface`. Property bags must be plain objects.
+Only `transport`, `surface`, `requestType`, `provider`, `model`, `errorClass`,
+`angularVersion`, `durationMs`, and `sample_weight` are forwarded. Strings are
+nonempty labels of at most 128 characters without control characters; durations
+must be finite numbers from 0 to 86,400,000 milliseconds, and sampling weights
+must be finite numbers of at least 1, preserving reciprocal weights at low sample
+rates. Unknown properties are dropped; invalid known metadata
+rejects the event. Do not put user content or credentials in metadata labels.
+Public submissions are untrusted observations, not verified product activity.
+
+`captureEvent()` returns `{ sent: false, reason: 'invalid' }` for invalid inputs.
+The Node stream helpers accept an optional `transport`; valid legacy calls with
+provider/model but no transport report `unknown`, without guessing an adapter.
+Pass the transport explicitly for meaningful transport breakdowns. Generic
+`captureEvent()` calls do not receive this fallback. These checks do not change
+browser opt-in, custom sinks, or development-only Growth collection controls.
+
 Set `TPLANE_TELEMETRY_INGEST_URL` to route events to an endpoint you control.
 The default endpoint is `https://threadplane.ai/api/ingest`.
 

libs/telemetry/src/node/adapter.spec.ts

@@ -79,4 +79,21 @@ describe('adapter helpers', () => {
     vi.mocked(captureEvent).mockRejectedValueOnce(new Error('network'));
     await expect(captureStreamStarted({ provider: 'x', model: 'y' })).resolves.toBeUndefined();
   });
+
+  test.each([captureStreamStarted, captureStreamEnded, captureStreamErrored])('legacy stream helpers identify transport as unknown', async (capture) => {
+    await capture({ provider: 'openai', model: 'gpt-4', error: new Error('private') });
+    expect(vi.mocked(captureEvent).mock.calls[0][1]).toMatchObject({ transport: 'unknown' });
+  });
+
+  test.each([captureStreamStarted, captureStreamEnded, captureStreamErrored])('stream helpers preserve an explicit transport', async (capture) => {
+    await capture({ transport: 'ag-ui', provider: 'openai', model: 'gpt-4', error: new Error('private') } as never);
+    expect(vi.mocked(captureEvent).mock.calls[0][1]).toMatchObject({ transport: 'ag-ui' });
+  });
+
+  test.each([null, 'private', 42, [], new Date(), {}, { provider: 'openai' }, { provider: '', model: 'gpt-4' }])('stream helpers do not manufacture events from malformed inputs', async (input) => {
+    await captureStreamStarted(input as never);
+    await captureStreamEnded(input as never);
+    await captureStreamErrored(input as never);
+    expect(captureEvent).not.toHaveBeenCalled();
+  });
 });

libs/telemetry/src/node/adapter.ts

@@ -9,6 +9,8 @@ export interface RuntimeInstanceTelemetry {
 }
 
 export interface StreamTelemetry {
+  /** Runtime transport when known; legacy calls without it report `unknown`. */
+  transport?: string;
   provider: string;
   model: string;
   durationMs?: number;
@@ -25,6 +27,12 @@ async function safe(fn: () => Promise<unknown>): Promise<void> {
   try { await fn(); } catch { /* silent fail */ }
 }
 
+function streamProperties(input: StreamTelemetry): Record<string, unknown> | null {
+  if (!input || typeof input !== 'object' || Object.getPrototypeOf(input) !== Object.prototype) return null;
+  if (typeof input.provider !== 'string' || !input.provider.trim() || typeof input.model !== 'string' || !input.model.trim()) return null;
+  return { ...input, transport: input.transport === undefined ? 'unknown' : input.transport };
+}
+
 export async function captureRuntimeInstanceCreated(input: RuntimeInstanceTelemetry): Promise<void> {
   await safe(async () => {
     const { apiKey, ...rest } = input;
@@ -38,19 +46,27 @@ export async function captureRuntimeRequestCreated(input: RuntimeRequestTelemetr
 }
 
 export async function captureStreamStarted(input: StreamTelemetry): Promise<void> {
-  await safe(() => captureEvent('tplane:stream_started', { ...input }));
+  await safe(async () => {
+    const properties = streamProperties(input);
+    if (properties) await captureEvent('tplane:stream_started', properties);
+  });
 }
 
 export async function captureStreamEnded(input: StreamTelemetry): Promise<void> {
-  await safe(() => captureEvent('tplane:stream_ended', { ...input }));
+  await safe(async () => {
+    const properties = streamProperties(input);
+    if (properties) await captureEvent('tplane:stream_ended', properties);
+  });
 }
 
 export async function captureStreamErrored(
   input: StreamTelemetry & { error: Error | unknown },
 ): Promise<void> {
   await safe(async () => {
+    const properties = streamProperties(input);
+    if (!properties) return;
     const { error, ...rest } = input;
     const errorClass = error instanceof Error ? error.constructor.name : 'Unknown';
-    await captureEvent('tplane:stream_errored', { ...rest, errorClass });
+    await captureEvent('tplane:stream_errored', { ...rest, transport: properties['transport'], errorClass });
   });
 }

libs/telemetry/src/node/client.spec.ts

@@ -37,13 +37,13 @@ describe('node client', () => {
 
   test('uses the configured ingest endpoint', async () => {
     process.env.TPLANE_TELEMETRY_INGEST_URL = 'https://custom.example/api/ingest';
-    await captureEvent('tplane:stream_started', {});
+    await captureEvent('tplane:stream_started', { transport: 'custom' });
     expect(fetchMock.mock.calls[0][0]).toBe('https://custom.example/api/ingest');
   });
 
   test('defaults to the Threadplane ingest proxy', async () => {
     delete process.env.TPLANE_TELEMETRY_INGEST_URL;
-    await captureEvent('tplane:stream_started', {});
+    await captureEvent('tplane:stream_started', { transport: 'custom' });
     expect(fetchMock.mock.calls[0][0]).toBe('https://threadplane.ai/api/ingest');
   });
 
@@ -67,15 +67,33 @@ describe('node client', () => {
 
   test('reports failed sends instead of throwing', async () => {
     fetchMock.mockRejectedValueOnce(new Error('network'));
-    await expect(captureEvent('tplane:stream_errored', {})).resolves.toEqual({
+    await expect(captureEvent('tplane:stream_errored', { transport: 'custom' })).resolves.toEqual({
       sent: false,
       reason: 'failed',
     });
   });
 
   test('invalid sample rate falls back to sending', async () => {
     process.env.TPLANE_TELEMETRY_SAMPLE_RATE = 'not-a-number';
-    await expect(captureEvent('tplane:stream_started', {})).resolves.toEqual({ sent: true });
+    await expect(captureEvent('tplane:stream_started', { transport: 'custom' })).resolves.toEqual({ sent: true });
     expect(fetchMock).toHaveBeenCalledOnce();
   });
+
+  test.each([{}, null, 'secret', [], { transport: 42 }, { transport: 'custom', model: { secret: true } }])(
+    'silently rejects malformed runtime properties before sending', async (properties) => {
+      await expect(captureEvent('tplane:stream_started', properties as never)).resolves.toEqual({ sent: false, reason: 'invalid' });
+      expect(fetchMock).not.toHaveBeenCalled();
+    },
+  );
+
+  test('rejects unknown runtime event names', async () => {
+    await expect(captureEvent('tplane:invented' as never, { transport: 'custom' })).resolves.toEqual({ sent: false, reason: 'invalid' });
+    expect(fetchMock).not.toHaveBeenCalled();
+  });
+
+  test('does not forward arbitrary properties or caller-supplied sampling weight', async () => {
+    await captureEvent('tplane:stream_started', { transport: 'custom', command: 'secret', token: 'secret', sample_weight: 9 });
+    const body = JSON.parse(String(fetchMock.mock.calls[0][1].body));
+    expect(body.properties).toEqual({ transport: 'custom', sample_weight: 1 });
+  });
 });

libs/telemetry/src/node/client.ts

@@ -2,6 +2,7 @@ import { getAnonId } from '../shared/anon-id.js';
 import { isTelemetryDisabled } from '../shared/env.js';
 import { shouldSample } from '../shared/sample.js';
 import type { ThreadplaneNodeEvent } from '../shared/events.js';
+import { parseTelemetryEvent } from '../shared/ingest.js';
 import { isProgrammaticallyDisabled } from './disable.js';
 
 const DEFAULT_INGEST = 'https://threadplane.ai/api/ingest';
@@ -12,7 +13,7 @@ const PUBLIC_INGEST_KEY = 'phc_public_cacheplane_telemetry';
 
 export type CaptureResult =
   | { sent: true }
-  | { sent: false; reason: 'disabled' | 'sampled' | 'failed' };
+  | { sent: false; reason: 'disabled' | 'sampled' | 'failed' | 'invalid' };
 
 function getSampleRate(env: NodeJS.ProcessEnv = process.env): number {
   const parsed = Number(env.TPLANE_TELEMETRY_SAMPLE_RATE ?? '1');
@@ -42,18 +43,22 @@ export async function captureEvent(
 ): Promise<CaptureResult> {
   if (isTelemetryDisabled() || isProgrammaticallyDisabled())
     return { sent: false, reason: 'disabled' };
+  const parsed = parseTelemetryEvent(event, properties);
+  if (!parsed || parsed.event.startsWith('tplane:browser_')) return { sent: false, reason: 'invalid' };
   const rate = getSampleRate();
   const anonId = getAnonId();
   if (!shouldSample(rate, anonId)) return { sent: false, reason: 'sampled' };
+  const payload = parseTelemetryEvent(parsed.event, {
+    ...parsed.properties,
+    sample_weight: rate > 0 ? 1 / Math.min(1, rate) : 1,
+  });
+  if (!payload) return { sent: false, reason: 'invalid' };
   try {
     await postJson(process.env.TPLANE_TELEMETRY_INGEST_URL ?? DEFAULT_INGEST, {
       key: PUBLIC_INGEST_KEY,
       distinctId: anonId,
-      event,
-      properties: {
-        ...properties,
-        sample_weight: rate > 0 ? 1 / Math.min(1, rate) : 1,
-      },
+      event: payload.event,
+      properties: payload.properties,
     });
     return { sent: true };
   } catch {