feat(agent): export cloud run telemetry over OTLP

packages/agent/README.md

@@ -171,6 +171,16 @@ Required environment variables (validated by zod in `src/server/bin.ts`):
 - `POSTHOG_PERSONAL_API_KEY` — API key for PostHog requests
 - `POSTHOG_PROJECT_ID` — numeric project ID
 
+Optional run telemetry (the logs pair must both be set, otherwise telemetry stays off):
+
+- `POSTHOG_AGENT_OTEL_LOGS_URL` — full OTLP logs URL for run metadata, e.g. `https://us.i.posthog.com/i/v1/logs`
+- `POSTHOG_AGENT_OTEL_LOGS_TOKEN` — project API key of the telemetry project
+- `POSTHOG_AGENT_OTEL_TRACES_URL` — full OTLP traces URL, e.g. `https://us.i.posthog.com/i/v1/traces`; additionally enables one APM trace per run
+
+When set, `AgentServer` ships an allowlisted metadata subset of the session log (run/turn/tool lifecycle, usage, error provenance — never message content, tool arguments, or raw error text; see `src/otel-telemetry.ts`) to PostHog Logs, tagged with `service.name=posthog-code-agent` and `run_id`/`task_id`/`team_id`/`user_id`/`distinct_id` resource attributes so cloud runs are filterable per user. With the traces URL set, each run also produces an APM trace (`task_run` root span, a `turn` span per prompt, a `tool_call:<kind>` span per tool call; see `src/otel-trace-builder.ts`), and log records carry the matching trace/span ids so Logs and APM cross-link.
+
+The `task_run` root span is ended and exported at the run's in-process terminal point — a background run's prompt settling, a terminal failure, or session cleanup (`close`). It cannot wait for teardown: agent-server is an exec'd process inside the sandbox, so `docker stop` / Modal terminate kill it without SIGTERM ever arriving, and a span still open at that moment is lost. Interactive sessions that end via hard teardown (e.g. inactivity timeout) therefore lose the root span; turn/tool spans and logs still assemble under the same trace id.
+
 ## Agent SDK
 
 The `Agent` class (`src/agent.ts`) is the entrypoint for local/programmatic usage. It handles LLM gateway configuration, log writer setup, and model filtering — then delegates to `createAcpConnection()`.
@@ -209,12 +219,9 @@ Logs serve two purposes: real-time observability and session resume. Every ACP m
 
 ### Writing logs
 
-`SessionLogWriter` (`src/session-log-writer.ts`) is a per-session multiplexer that buffers raw ndJson lines. On flush (auto-scheduled 500ms after writes, or explicit), it dispatches to whichever backend is configured:
-
-- **OTEL** (`src/otel-log-writer.ts`) — preferred path. Creates an OpenTelemetry `LoggerProvider` per session with resource attributes (`task_id`, `run_id`, `device_type`) set once and indexed via `resource_fingerprint`. Each ndJson line is emitted as an OTEL log record with an `event_type` attribute (the ACP method name) and exported via OTLP HTTP to PostHog's `/i/v1/agent-logs` endpoint. Batch flush interval defaults to 500ms.
-- **Legacy S3** — falls back to `PostHogAPIClient.appendTaskRunLog()`, which POSTs batched `StoredNotification` entries to the Django API. The API stores them as the task run's `log_url`.
+`SessionLogWriter` (`src/session-log-writer.ts`) is a per-session multiplexer that buffers raw ndJson lines. On flush (auto-scheduled 500ms after writes, or explicit), it POSTs batched `StoredNotification` entries to the Django API via `PostHogAPIClient.appendTaskRunLog()`; the API stores them in S3 as the task run's `log_url`. This S3 log is the source of truth for the full transcript and for session resume.
 
-Both backends can be active simultaneously — OTEL for fast indexed queries, S3 for full log download.
+Independently of that flush cycle, the writer tees every parsed non-chunk entry to optional `SessionLogSink`s at append time. In cloud, `OtelRunTelemetry` (`src/otel-telemetry.ts`) is wired as a sink and exports an allowlisted metadata subset (run/turn/tool lifecycle, usage, error provenance — never message content, tool arguments, or raw error text) to PostHog Logs, plus an APM trace per run when configured. See "Optional run telemetry" above for the env vars and behavior. Sink failures can never break S3 log persistence.
 
 ### Resuming from logs
 

packages/agent/package.json

@@ -163,10 +163,13 @@
     "@earendil-works/pi-coding-agent": "catalog:",
     "@hono/node-server": "^1.19.9",
     "@openai/codex": "0.144.0",
+    "@opentelemetry/api": "^1.9.1",
     "@opentelemetry/api-logs": "^0.208.0",
     "@opentelemetry/exporter-logs-otlp-http": "^0.208.0",
+    "@opentelemetry/exporter-trace-otlp-http": "^0.208.0",
     "@opentelemetry/resources": "^2.0.0",
     "@opentelemetry/sdk-logs": "^0.208.0",
+    "@opentelemetry/sdk-trace-base": "^2.8.0",
     "@opentelemetry/semantic-conventions": "^1.28.0",
     "@types/jsonwebtoken": "^9.0.10",
     "commander": "^14.0.2",

packages/agent/src/otel-attributes.ts

@@ -0,0 +1,84 @@
+const MAX_BODY_CHARS = 2000;
+// PostHog Logs only facets attribute key/value pairs shorter than 256 chars,
+// so free-text attribute values are capped well below that.
+const MAX_ATTR_CHARS = 200;
+// The SDK default export timeout is 30s; a hanging endpoint must not hold up
+// session cleanup (the sandbox can be torn down right after), so keep it short.
+const EXPORT_TIMEOUT_MS = 5000;
+
+// Batch flush cadence shared by the log and span processors.
+const DEFAULT_FLUSH_INTERVAL_MS = 2000;
+
+export type AttributeValue = string | number | boolean;
+export type Attributes = Record<string, AttributeValue>;
+
+export {
+  DEFAULT_FLUSH_INTERVAL_MS,
+  EXPORT_TIMEOUT_MS,
+  MAX_ATTR_CHARS,
+  MAX_BODY_CHARS,
+};
+
+/**
+ * extNotification() can double-prefix custom methods (see matchesExt in
+ * acp-extensions.ts); normalize so both spellings map identically.
+ */
+export function normalizeMethod(method: string): string {
+  return method.startsWith("__posthog/") ? method.slice(1) : method;
+}
+
+export function asString(value: unknown): string | undefined {
+  return typeof value === "string" ? value : undefined;
+}
+
+export function truncate(value: string, max: number): string {
+  return value.length <= max ? value : `${value.slice(0, max)}…`;
+}
+
+export function asRecord(value: unknown): Record<string, unknown> | undefined {
+  return typeof value === "object" && value !== null && !Array.isArray(value)
+    ? (value as Record<string, unknown>)
+    : undefined;
+}
+
+export function strAttr(
+  attrs: Attributes,
+  key: string,
+  value: unknown,
+  max = MAX_ATTR_CHARS,
+): string | undefined {
+  if (typeof value !== "string" || value.length === 0) return undefined;
+  const truncated = truncate(value, max);
+  attrs[key] = truncated;
+  return truncated;
+}
+
+export function numAttr(attrs: Attributes, key: string, value: unknown): void {
+  if (typeof value === "number" && Number.isFinite(value)) {
+    attrs[key] = value;
+  }
+}
+
+export function usageAttributes(params: Record<string, unknown>): Attributes {
+  const attrs: Attributes = {};
+  const used = asRecord(params.used);
+  if (used) {
+    numAttr(attrs, "tokens_input", used.inputTokens);
+    numAttr(attrs, "tokens_output", used.outputTokens);
+    numAttr(attrs, "tokens_cached_read", used.cachedReadTokens);
+    numAttr(attrs, "tokens_cached_write", used.cachedWriteTokens);
+  }
+  // Claude sends a plain number; other shapes carry { amount }.
+  const cost =
+    typeof params.cost === "number"
+      ? params.cost
+      : asRecord(params.cost)?.amount;
+  numAttr(attrs, "cost_usd", cost);
+  return attrs;
+}
+
+/** Timestamp of a stored entry as a Date, falling back to now when invalid. */
+export function entryTime(timestamp: string): Date {
+  const parsed = new Date(timestamp);
+  return Number.isNaN(parsed.getTime()) ? new Date() : parsed;
+}