Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions apps/website/src/app/api/ingest/route.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
});
});

/**
Expand Down
20 changes: 12 additions & 8 deletions apps/website/src/app/api/ingest/route.ts
Original file line number Diff line number Diff line change
@@ -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',
Expand Down Expand Up @@ -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,
};
}

Expand All @@ -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 });
}
Expand Down Expand Up @@ -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' },
Expand Down
11 changes: 7 additions & 4 deletions docs/growth/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
14 changes: 13 additions & 1 deletion docs/gtm/taxonomy.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand All @@ -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)

Expand Down
19 changes: 19 additions & 0 deletions libs/telemetry/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`.

Expand Down
17 changes: 17 additions & 0 deletions libs/telemetry/src/node/adapter.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
});
});
22 changes: 19 additions & 3 deletions libs/telemetry/src/node/adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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 });
});
}
26 changes: 22 additions & 4 deletions libs/telemetry/src/node/client.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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');
});

Expand All @@ -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 });
});
});
17 changes: 11 additions & 6 deletions libs/telemetry/src/node/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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');
Expand Down Expand Up @@ -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 {
Expand Down
Loading
Loading