feat(agent): add OTEL log transport with resource attributes

chore: cleaner

chore: clean

Author: tatoalo

apps/twig/src/main/services/agent/service.ts

@@ -430,12 +430,23 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
     const mockNodeDir = this.setupMockNodeEnvironment(taskRunId);
     this.setupEnvironment(credentials, mockNodeDir);
 
+    // Route agent logs to dedicated agent_logs Kafka topic via capture-logs-agent service
+    // In local dev, use Caddy proxy (port 8010) which routes /i/v1/agent-logs to capture-logs-agent
+    // In prod, use the main API host which proxies /i/v1/agent-logs to capture-logs-agent
+    const otelHost = credentials.apiHost;
+    const otelPath = "/i/v1/agent-logs";
+
     const agent = new Agent({
       posthog: {
         apiUrl: credentials.apiHost,
         getApiKey: () => this.getToken(credentials.apiKey),
         projectId: credentials.projectId,
       },
+      otelTransport: {
+        host: otelHost,
+        apiKey: this.getToken(credentials.apiKey),
+        logsPath: otelPath,
+      },
       debug: !app.isPackaged,
       onLog: onAgentLog,
     });
@@ -577,7 +588,9 @@ export class AgentService extends TypedEventEmitter<AgentServiceEvents> {
         return this.getOrCreateSession(config, isReconnect, true);
       }
       log.error(
-        `Failed to ${isReconnect ? "reconnect" : "create"} session${isRetry ? " after retry" : ""}`,
+        `Failed to ${isReconnect ? "reconnect" : "create"} session${
+          isRetry ? " after retry" : ""
+        }`,
         err,
       );
       if (isReconnect) return null;

apps/twig/src/main/services/git/service.ts

@@ -64,7 +64,7 @@ export class GitService extends TypedEventEmitter<GitServiceEvents> {
     const remoteUrl = await getRemoteUrl(directoryPath);
     if (!remoteUrl) return null;
 
-    const repo = await parseGitHubUrl(remoteUrl);
+    const repo = parseGitHubUrl(remoteUrl);
     if (!repo) return null;
 
     const branch = await getCurrentBranch(directoryPath);

packages/agent/package.json

@@ -82,6 +82,11 @@
     "vitest": "^2.1.8"
   },
   "dependencies": {
+    "@opentelemetry/api-logs": "^0.208.0",
+    "@opentelemetry/exporter-logs-otlp-http": "^0.208.0",
+    "@opentelemetry/resources": "^2.0.0",
+    "@opentelemetry/sdk-logs": "^0.208.0",
+    "@opentelemetry/semantic-conventions": "^1.28.0",
     "@agentclientprotocol/sdk": "^0.13.1",
     "@anthropic-ai/claude-agent-sdk": "0.2.12",
     "@anthropic-ai/sdk": "^0.71.0",

packages/agent/src/adapters/acp-connection.ts

@@ -18,6 +18,8 @@ export type AcpConnectionConfig = {
   logWriter?: SessionLogWriter;
   sessionId?: string;
   taskId?: string;
+  /** Deployment environment - "local" for desktop, "cloud" for cloud sandbox */
+  deviceType?: "local" | "cloud";
   logger?: Logger;
   processCallbacks?: ClaudeAcpAgentOptions;
 };
@@ -52,6 +54,7 @@ export function createAcpConnection(
       logWriter.register(config.sessionId, {
         taskId: config.taskId ?? config.sessionId,
         runId: config.sessionId,
+        deviceType: config.deviceType,
       });
     }
 

packages/agent/src/adapters/claude/claude-agent.ts

@@ -28,10 +28,8 @@ import {
   type SDKUserMessage,
 } from "@anthropic-ai/claude-agent-sdk";
 import { v7 as uuidv7 } from "uuid";
-import type {
-  SessionLogConfig,
-  SessionLogWriter,
-} from "@/session-log-writer.js";
+import type { SessionContext } from "@/otel-log-writer.js";
+import type { SessionLogWriter } from "@/session-log-writer.js";
 import { unreachable } from "@/utils/common.js";
 import { Logger } from "@/utils/logger.js";
 import { Pushable } from "@/utils/streams.js";
@@ -388,7 +386,7 @@ export class ClaudeAcpAgent extends BaseAcpAgent {
     sessionId: string,
     meta: Record<string, unknown> | undefined,
   ) {
-    const persistence = meta?.persistence as SessionLogConfig | undefined;
+    const persistence = meta?.persistence as SessionContext | undefined;
     if (persistence && this.logWriter) {
       this.logWriter.register(sessionId, persistence);
     }

packages/agent/src/agent.ts

@@ -25,10 +25,17 @@ export class Agent {
 
     if (config.posthog) {
       this.posthogAPI = new PostHogAPIClient(config.posthog);
-      this.sessionLogWriter = new SessionLogWriter(
-        this.posthogAPI,
-        this.logger.child("SessionLogWriter"),
-      );
+    }
+
+    if (config.otelTransport) {
+      this.sessionLogWriter = new SessionLogWriter({
+        otelConfig: {
+          posthogHost: config.otelTransport.host,
+          apiKey: config.otelTransport.apiKey,
+          logsPath: config.otelTransport.logsPath,
+        },
+        logger: this.logger.child("SessionLogWriter"),
+      });
     }
   }
 
@@ -62,6 +69,7 @@ export class Agent {
       logWriter: this.sessionLogWriter,
       sessionId: taskRunId,
       taskId,
+      deviceType: "local",
       logger: this.logger,
       processCallbacks: options.processCallbacks,
     });

packages/agent/src/index.ts

@@ -22,6 +22,8 @@ export type {
 } from "./adapters/acp-connection.js";
 export { createAcpConnection } from "./adapters/acp-connection.js";
 export { Agent } from "./agent.js";
+export type { OtelLogConfig, SessionContext } from "./otel-log-writer.js";
+export { OtelLogWriter } from "./otel-log-writer.js";
 export { PostHogAPIClient } from "./posthog-api.js";
 export type {
   ConversationTurn,
@@ -30,6 +32,8 @@ export type {
   ToolCallInfo,
 } from "./resume.js";
 export { conversationToPromptHistory, resumeFromLog } from "./resume.js";
+export type { SessionLogWriterOptions } from "./session-log-writer.js";
+export { SessionLogWriter } from "./session-log-writer.js";
 export type { TreeSnapshot, TreeTrackerConfig } from "./tree-tracker.js";
 export {
   isCommitOnRemote,
@@ -44,6 +48,7 @@ export type {
   FileStatus,
   LogLevel,
   OnLogCallback,
+  OtelTransportConfig,
   StoredEntry,
   StoredNotification,
   Task,

packages/agent/src/otel-log-writer.test.ts

@@ -0,0 +1,105 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { OtelLogWriter } from "./otel-log-writer.js";
+import type { StoredNotification } from "./types.js";
+
+// Mock the OTEL exporter
+const mockExport = vi.fn((_logs, callback) => {
+  callback({ code: 0 }); // Success
+});
+
+vi.mock("@opentelemetry/exporter-logs-otlp-http", () => ({
+  OTLPLogExporter: vi.fn().mockImplementation(() => ({
+    export: mockExport,
+    shutdown: vi.fn().mockResolvedValue(undefined),
+  })),
+}));
+
+describe("OtelLogWriter", () => {
+  let writer: OtelLogWriter;
+
+  beforeEach(() => {
+    mockExport.mockClear();
+    // Session context (taskId, runId) is now passed in constructor as resource attributes
+    writer = new OtelLogWriter(
+      {
+        posthogHost: "https://us.i.posthog.com",
+        apiKey: "phc_test_key",
+        flushIntervalMs: 100,
+      },
+      {
+        taskId: "task-123",
+        runId: "run-456",
+      },
+    );
+  });
+
+  afterEach(async () => {
+    await writer.shutdown();
+  });
+
+  it("should emit a log entry with event_type as regular attribute", async () => {
+    const notification: StoredNotification = {
+      type: "notification",
+      timestamp: new Date().toISOString(),
+      notification: {
+        jsonrpc: "2.0",
+        method: "_posthog/test_event",
+        params: { foo: "bar" },
+      },
+    };
+
+    // taskId and runId are now resource attributes set in constructor,
+    // only notification is passed per-emit
+    writer.emit({ notification });
+
+    // Force flush to trigger export
+    await writer.flush();
+
+    // Verify export was called
+    expect(mockExport).toHaveBeenCalled();
+
+    // Get the logs that were exported
+    const exportedLogs = mockExport.mock.calls[0][0];
+    expect(exportedLogs.length).toBe(1);
+
+    const log = exportedLogs[0];
+    // task_id and run_id are now resource attributes, not regular attributes
+    expect(log.attributes.task_id).toBeUndefined();
+    expect(log.attributes.run_id).toBeUndefined();
+    // event_type is still a regular attribute (varies per log entry)
+    expect(log.attributes.event_type).toBe("_posthog/test_event");
+    expect(log.body).toBe(JSON.stringify(notification));
+
+    // Verify resource attributes contain task_id and run_id
+    expect(log.resource.attributes.task_id).toBe("task-123");
+    expect(log.resource.attributes.run_id).toBe("run-456");
+    expect(log.resource.attributes["service.name"]).toBe("twig-agent");
+  });
+
+  it("should batch multiple log entries", async () => {
+    const makeNotification = (method: string): StoredNotification => ({
+      type: "notification",
+      timestamp: new Date().toISOString(),
+      notification: {
+        jsonrpc: "2.0",
+        method,
+      },
+    });
+
+    writer.emit({ notification: makeNotification("event_1") });
+    writer.emit({ notification: makeNotification("event_2") });
+    writer.emit({ notification: makeNotification("event_3") });
+
+    await writer.flush();
+
+    expect(mockExport).toHaveBeenCalled();
+    const exportedLogs = mockExport.mock.calls[0][0];
+    expect(exportedLogs.length).toBe(3);
+
+    // All logs should share the same resource attributes
+    for (const log of exportedLogs) {
+      expect(log.resource.attributes.task_id).toBe("task-123");
+      expect(log.resource.attributes.run_id).toBe("run-456");
+    }
+  });
+});

packages/agent/src/otel-log-writer.ts

@@ -0,0 +1,106 @@
+import { SeverityNumber } from "@opentelemetry/api-logs";
+import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
+import { resourceFromAttributes } from "@opentelemetry/resources";
+import {
+  BatchLogRecordProcessor,
+  LoggerProvider,
+} from "@opentelemetry/sdk-logs";
+import { ATTR_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
+import type { StoredNotification } from "./types.js";
+import { Logger } from "./utils/logger.js";
+
+export interface OtelLogConfig {
+  /** PostHog ingest host, e.g., "https://us.i.posthog.com" */
+  posthogHost: string;
+  /** Project API key, e.g., "phc_xxx" */
+  apiKey: string;
+  /** Batch flush interval in ms (default: 500) */
+  flushIntervalMs?: number;
+  /** Override the logs endpoint path (default: /i/v1/agent-logs) */
+  logsPath?: string;
+}
+
+/**
+ * Session context for resource attributes.
+ * These are set once per OTEL logger instance and indexed via resource_fingerprint
+ */
+export interface SessionContext {
+  /** Parent task grouping - all runs for a task share this */
+  taskId: string;
+  /** Primary conversation identifier - all events in a run share this */
+  runId: string;
+  /** Deployment environment - "local" for desktop, "cloud" for cloud sandbox */
+  deviceType?: "local" | "cloud";
+}
+
+export class OtelLogWriter {
+  private loggerProvider: LoggerProvider;
+  private logger: ReturnType<LoggerProvider["getLogger"]>;
+  private debugLogger: Logger;
+  private sessionContext: SessionContext;
+
+  constructor(
+    config: OtelLogConfig,
+    sessionContext: SessionContext,
+    debugLogger?: Logger,
+  ) {
+    this.debugLogger =
+      debugLogger ?? new Logger({ debug: false, prefix: "[OtelLogWriter]" });
+    this.sessionContext = sessionContext;
+
+    const logsPath = config.logsPath ?? "/i/v1/agent-logs";
+    const exporter = new OTLPLogExporter({
+      url: `${config.posthogHost}${logsPath}`,
+      headers: { Authorization: `Bearer ${config.apiKey}` },
+    });
+
+    const processor = new BatchLogRecordProcessor(exporter, {
+      scheduledDelayMillis: config.flushIntervalMs ?? 500,
+    });
+
+    // Resource attributes are set ONCE per session and indexed via resource_fingerprint
+    // So we have fast queries by run_id/task_id in PostHog Logs UI
+    this.loggerProvider = new LoggerProvider({
+      resource: resourceFromAttributes({
+        [ATTR_SERVICE_NAME]: "twig-agent",
+        run_id: sessionContext.runId,
+        task_id: sessionContext.taskId,
+        device_type: sessionContext.deviceType ?? "local",
+      }),
+      processors: [processor],
+    });
+
+    this.logger = this.loggerProvider.getLogger("agent-session");
+  }
+
+  /**
+   * Emit an agent event to PostHog Logs via OTEL.
+   */
+  emit(entry: { notification: StoredNotification }): void {
+    const { notification } = entry;
+    const eventType = notification.notification.method;
+
+    this.logger.emit({
+      severityNumber: SeverityNumber.INFO,
+      severityText: "INFO",
+      body: JSON.stringify(notification),
+      attributes: {
+        event_type: eventType,
+      },
+    });
+
+    this.debugLogger.debug("Emitted OTEL log", {
+      taskId: this.sessionContext.taskId,
+      runId: this.sessionContext.runId,
+      eventType,
+    });
+  }
+
+  async flush(): Promise<void> {
+    await this.loggerProvider.forceFlush();
+  }
+
+  async shutdown(): Promise<void> {
+    await this.loggerProvider.shutdown();
+  }
+}

packages/agent/src/server/agent-server.ts

@@ -434,20 +434,25 @@ export class AgentServer {
       logger: new Logger({ debug: true, prefix: "[TreeTracker]" }),
     });
 
-    const posthogAPI = new PostHogAPIClient({
+    const _posthogAPI = new PostHogAPIClient({
       apiUrl: this.config.apiUrl,
       projectId: this.config.projectId,
       getApiKey: () => this.config.apiKey,
     });
 
-    const logWriter = new SessionLogWriter(
-      posthogAPI,
-      new Logger({ debug: true, prefix: "[SessionLogWriter]" }),
-    );
+    const logWriter = new SessionLogWriter({
+      otelConfig: {
+        posthogHost: this.config.apiUrl,
+        apiKey: this.config.apiKey,
+        logsPath: "/i/v1/agent-logs",
+      },
+      logger: new Logger({ debug: true, prefix: "[SessionLogWriter]" }),
+    });
 
     const acpConnection = createAcpConnection({
       sessionId: payload.run_id,
       taskId: payload.task_id,
+      deviceType: deviceInfo.type,
       logWriter,
     });