pi-conductor-analytics-plugin

Pi extension that reports pi-conductor run telemetry to an external HTTP endpoint.

Overview

This plugin hooks into pi-conductor's pi.events.on("conductor:record", ...) event to receive persisted run records (session_started, session_ended, transition_accepted, transition_rejected, checkpoint_snapshot, etc.) and forwards them to a configured HTTP(S) endpoint in a non-blocking, best-effort fashion.

File-mutation telemetry

The analytics contract uses pi-conductor's persisted file_mutation record. The plugin forwards this record unchanged in the envelope's records array; it does not emit a duplicate display record.

A successful write or edit is delivered in this shape:

{
  "type": "file_mutation",
  "run_id": "run-123",
  "role": "worker",
  "session_id": "session-456",
  "session_file": "/tmp/session-456.jsonl",
  "tool_name": "edit",
  "files": [
    {
      "path": "src/index.ts",
      "additions": 12,
      "deletions": 4,
      "hunks": [
        { "lineNumber": 10, "content": "+const ready = true;", "kind": "add" }
      ]
    }
  ],
  "ts": 1735689600000
}

files[*].additions and files[*].deletions are character counts. The optional hunks array contains { lineNumber, content, kind } entries, where kind is add, del, or context. Hunk generation can be unavailable (for example, when the host cannot read the pre-mutation file); consumers should still use the path and character counts. Failed mutations and malformed mutation arguments do not produce a file_mutation record.

Consumers should select records where type === "file_mutation" and use tool_name to distinguish write from edit. These records retain the plugin's normal at-least-once backstop and delivery semantics.

Role-turn telemetry

The plugin forwards role_turn records from pi-conductor's conductor:record bridge without parsing or inspecting the record payload. Only the type field is validated at the admission boundary.

Opaque record boundary

The plugin treats every role_turn value as an opaque persisted analytics record:

• The admission boundary accepts any non-null object with a string type field
• Payload fields (text, thinking, etc.) are never parsed, rewritten, or inspected
• Records are forwarded byte-for-byte unchanged inside envelope records[]
• Record order from a single run is preserved in outgoing envelopes

Example role_turn record shape

{
  "type": "role_turn",
  "run_id": "run-abc",
  "session_id": "session-123",
  "role": "worker",
  "turn_index": 0,
  "ts": 1735689600000,
  "text_tokens": 234,
  "thinking_tokens": 89,
  "message_count": 3
}

Consumers should select records where type === "role_turn". These records retain the plugin's normal at-least-once backstop and delivery semantics.

High-volume role-turn streams

The plugin is designed for high-volume unattended role-turn telemetry:

Aspect	Behavior
Queue capacity	500 pending records (hard cap)
Overflow handling	Oldest-record drop with diagnostic callback
Batching	Up to 25 records per POST (configurable)
Flush interval	1 second for partial batches (configurable)
Retry policy	Exponential backoff (100ms–2000ms, max 2 retries)
At-least-once	Watermarks advance only after confirmed delivery

For high-volume streams, consider tuning batch.maxRecords (e.g., 50–100) and batch.flushIntervalMs (e.g., 500) to reduce per-request overhead while staying within endpoint rate limits.

Installation

# From the plugin directory
pi install /path/to/pi-conductor-analytics-plugin

# Or via npm/pnpm
pnpm add pi-conductor-analytics-plugin
pi install ./node_modules/pi-conductor-analytics-plugin

Prerequisites: pi-conductor must also be installed as a pi extension. The plugin listens for events emitted by pi-conductor's extension; if pi-conductor is not loaded, the plugin will be inactive.

Programmatic API

For library integrations that want to report analytics without the full Pi extension lifecycle, use createAnalyticsReporter():

import { createAnalyticsReporter } from "pi-conductor-analytics-plugin";
import { join } from "node:path";

const reporter = createAnalyticsReporter({
  cwd: process.cwd(),
  runsDir: join(process.cwd(), ".pi-conductor/runs"),
  configPath: "./analytics.json",        // optional explicit config
  source: "library:conductor:record",    // optional custom source label
});

// Enqueue records for delivery
reporter.enqueue({ type: "my_event", value: 42 });

// Flush pending records
await reporter.flush();

// Stop the reporter (flushes remaining records with 2-second timeout)
await reporter.shutdown();

// Get delivery statistics
const stats = reporter.stats();
// { enqueued: 1, delivered: 1, failed: 0, dropped: 0, pending: 0 }

All methods are safe to call when the reporter is disabled (missing config, no endpoint):

• enqueue() silently drops invalid records
• backfill(), flush(), and shutdown() resolve immediately
• stats() returns a snapshot with all counters at zero

JSONL Backfill

The backfill() method scans the runs directory for JSONL files and enqueues records past each run's committed watermark:

// Scan for unprocessed records from disk
const count = await reporter.backfill();
// Returns the number of records enqueued

Configuration

Config lookup order (first existing file wins):

1. <cwd>/.pi-conductor-analytics.json
2. <cwd>/.pi/conductor-analytics.json
3. $HOME/.pi-conductor-analytics.json
4. $HOME/.config/pi-conductor/analytics.json

Example config

{
  "enabled": true,
  "endpoint": "https://analytics.example.com/pi-conductor/events",
  "headers": {
    "Authorization": "Bearer ${PI_CONDUCTOR_ANALYTICS_TOKEN}"
  },
  "batch": {
    "enabled": true,
    "maxRecords": 25,
    "flushIntervalMs": 1000
  },
  "request": {
    "timeoutMs": 5000,
    "maxRetries": 2
  }
}

Config fields

Field	Default	Description
enabled	true	Master switch
endpoint	required	HTTP(S) URL to POST records to
headers	{}	Custom HTTP headers with ${ENV_VAR} interpolation
batch.enabled	true	Enable batch delivery
batch.maxRecords	25	Max records per batch POST
batch.flushIntervalMs	1000	Flush interval (ms) for partial batches
request.timeoutMs	5000	Request timeout (ms)
request.maxRetries	2	Max retries per failed request
request.retry.baseDelayMs	100	Base delay (ms) for exponential backoff
request.retry.maxDelayMs	2000	Maximum delay (ms) cap for backoff
request.retry.jitterFactor	0	Jitter factor (0-1) for thundering herd prevention

Delivery behavior

Non-blocking delivery

Record handlers only enqueue work synchronously. Network I/O happens in the background via a bounded delivery queue. The queue flushes:

• When the batch size (batch.maxRecords) is reached
• On the configured interval (batch.flushIntervalMs)
• On session_shutdown

Request deadline with abort semantics

Each HTTP request has an overall deadline (configured via request.timeoutMs, default 5,000 ms). The deadline:

• Starts when the first attempt begins
• Applies to all retries collectively
• Aborts the in-flight attempt when reached
• Stops retries after abort or deadline expiry

On shutdown, a 2-second deadline is used to ensure the process exits cleanly even if the network is unresponsive.

Deterministic retry policy

Failed requests are retried with deterministic capped exponential backoff:

Attempt	Backoff delay
1	100 ms
2	200 ms
3	400 ms
4+	800 ms (capped at 2,000 ms)

The maximum delay is capped at 2,000 ms. If the remaining deadline is less than the calculated backoff, the retry is skipped.

Retryable statuses: 408 Request Timeout, 429 Too Many Requests, 5xx Server Errors

Non-retryable: All other HTTP responses (including 4xx client errors) are accepted and dropped after one attempt.

Queue overflow behavior

The delivery queue is bounded to 500 pending records to prevent unbounded memory growth.

• Oldest-drop: When the queue is full, the oldest pending record is dropped
• Immediate first diagnostic: The overflow callback is invoked synchronously on the first overflow event
• Rate-limited subsequent diagnostics: After the first overflow, callbacks are rate-limited to once per 5 seconds while aggregating suppressed drops
• Recovery: After pending records drain, the rate-limit window resets so a subsequent overflow is reported immediately

Use queue.setOverflowCallback() to subscribe to overflow events:

const reporter = createAnalyticsReporter(options, (dropped, pending, suppressed) => {
  console.log(`Overflow: ${dropped} dropped, ${pending} pending, ${suppressed} suppressed since last callback`);
});

Statistics

Call stats() to get a snapshot of delivery metrics:

const { enqueued, delivered, failed, dropped, pending } = reporter.stats();

• enqueued: Total records added to the queue
• delivered: Records successfully POSTed and confirmed
• failed: Records that failed after exhausting retries
• dropped: Records dropped due to queue overflow
• pending: Records currently in the queue awaiting delivery

JSONL backstop / watermark

On session_start, the plugin scans <cwd>/.pi-conductor/runs/*.jsonl for records that haven't been sent yet (tracked via per-run .watermark.json sidecar files). This ensures records are not lost if:

• The plugin loads after a run has already started
• There was a network outage during the run
• The plugin was installed after previous runs

At-least-once semantics

The backstop provides at-least-once delivery for persisted records:

• Watermarks advance only after successful contiguous delivery (2xx response)
• Failed or timed-out batches leave the watermark unchanged
• Records replay after restart if delivery was not confirmed
• Duplicates are possible after a crash: if the plugin crashes between successful delivery and watermark update, the same records will be resent

Atomic sidecar updates

Watermark updates use atomic file operations:

1. Write to a temporary file in the same directory
2. Flush and close the temporary file
3. Rename the temp file to the watermark file

This ensures that a crash during update leaves either the old or new watermark intact.

Backfill behavior

• Only complete newline-terminated JSONL lines are processed
• Malformed lines are skipped without blocking valid records
• A trailing incomplete line (partial write) is not included and does not advance the watermark
• On the next backfill (after restart), the partial line is retried

Privacy

• This plugin sends only pi-conductor's persisted run telemetry records.
• It does not read or upload full role-session transcript files (the JSONL files at <session_file> paths).
• No role or session secrets are logged.
• Custom HTTP headers with ${ENV_VAR} interpolation are expanded at config load time. The expanded values are never logged.

Development

# Install dependencies
pnpm install

# Build
pnpm build

# Typecheck
pnpm typecheck

# Test
pnpm test

# Lint
pnpm lint

Project structure

├── package.json             Package metadata, pi extension registration
├── extensions/analytics.ts  Extension entrypoint (loaded by pi)
├── src/
│   ├── index.ts             Public API exports
│   ├── types.ts             TypeScript types
│   ├── config.ts            Config discovery, validation, env interpolation
│   ├── envelope.ts          Envelope creation
│   ├── queue.ts             Bounded async delivery queue, POST logic
│   ├── reporter.ts          Programmatic reporter factory
│   └── watermark.ts        JSONL backstop / per-run watermark tracking
└── tests/
    ├── config.test.ts
    ├── envelope.test.ts
    ├── queue.test.ts
    ├── reporter.test.ts
    ├── extension.test.ts
    └── watermark.test.ts

Behavior notes

• Warning surfaces: Factory/setup context uses console.warn (no ctx available). session_start / session_shutdown handlers use ctx.ui.notify. The conductor:record event handler receives only data, no ctx; use console.warn.
• No pi-conductor peer dependency: The plugin depends only on @earendil-works/pi-coding-agent. It uses pi.events (the shared event bus) for the conductor:record bridge, avoiding a direct import of pi-conductor internals.
• session_shutdown: This is a first-class typed event (ExtensionAPI.on("session_shutdown", ...)) and is used for the typed handler form.