diff --git a/packages/runtime/src/v2/runtime/core/__tests__/channel-manager-reconnect.test.ts b/packages/runtime/src/v2/runtime/core/__tests__/channel-manager-reconnect.test.ts index 2868886021..c6bbeec842 100644 --- a/packages/runtime/src/v2/runtime/core/__tests__/channel-manager-reconnect.test.ts +++ b/packages/runtime/src/v2/runtime/core/__tests__/channel-manager-reconnect.test.ts @@ -3,6 +3,7 @@ import { createChannel } from "@copilotkit/channels"; import { CopilotKitIntelligence } from "../../intelligence-platform"; import { ChannelManager } from "../channel-manager"; import type { ActivateChannelEngine, ChannelsHandle } from "../channel-manager"; +import { telemetry } from "../../telemetry"; /* ------------------------------------------------------------------------------------------------ * Reconnection is delegated to the Phoenix connection layer (the launcher's @@ -216,4 +217,136 @@ describe("ChannelManager connection health (onStateChange)", () => { expect(mgr.status().channels.support).toBe("error"); await mgr.stop(); }); + + it("repeats the drop cause on every still-down line (OSS-825)", async () => { + const handle = observableHandle(); + const engine: ActivateChannelEngine = vi.fn(async () => handle); + const logs: string[] = []; + + const mgr = new ChannelManager({ + intelligence: fakeIntelligence(), + channels: [createChannel({ identifyUser: "platform", name: "support" })], + activateChannel: engine, + log: (m: string) => logs.push(m), + reconnectLogIntervalMs: 5, + }); + mgr.activate(); + await mgr.ready(); + + handle.fireState("reconnecting", { + reason: "the gateway host answered HTTP 502", + }); + await new Promise((r) => setTimeout(r, 20)); + + // An operator reading a repeat line hours into an outage must not have to + // scroll back to the first line to learn the cause: prod emitted + // "still down after 233134s; Phoenix is retrying" with no cause at all. + const repeats = logs.filter((m) => m.includes("still down")); + expect(repeats.length).toBeGreaterThan(0); + expect(repeats[0]).toContain("HTTP 502"); + await mgr.stop(); + }); +}); + +/* ------------------------------------------------------------------------------------------------ + * Drop/recovery telemetry (OSS-825). A managed session that loses its gateway + * link is invisible outside the host process: the only trace is the injected + * `log` seam, which for a self-hosted or Railway-hosted runtime reaches nobody + * who can act on it. A 2026-08-12 incident ran 2.7 days on one Channel before a + * customer reported it. These events make an outage — and its end — reportable. + * --------------------------------------------------------------------------------------------- */ +describe("ChannelManager drop/recovery telemetry (OSS-825)", () => { + it("captures a dropped event carrying the transport cause", async () => { + const captureSpy = vi + .spyOn(telemetry, "capture") + .mockResolvedValue(undefined); + try { + const handle = observableHandle(); + const engine: ActivateChannelEngine = vi.fn(async () => handle); + const mgr = new ChannelManager({ + intelligence: fakeIntelligence(), + channels: [ + createChannel({ identifyUser: "platform", name: "support" }), + ], + activateChannel: engine, + }); + mgr.activate(); + await mgr.ready(); + + handle.fireState("reconnecting", { + reason: "the gateway host answered HTTP 502", + }); + + expect(captureSpy).toHaveBeenCalledWith( + "oss.runtime.channel_session_dropped", + { reason: "the gateway host answered HTTP 502" }, + ); + await mgr.stop(); + } finally { + captureSpy.mockRestore(); + } + }); + + it("captures a recovered event carrying the outage duration", async () => { + const captureSpy = vi + .spyOn(telemetry, "capture") + .mockResolvedValue(undefined); + try { + const handle = observableHandle(); + const engine: ActivateChannelEngine = vi.fn(async () => handle); + const mgr = new ChannelManager({ + intelligence: fakeIntelligence(), + channels: [ + createChannel({ identifyUser: "platform", name: "support" }), + ], + activateChannel: engine, + }); + mgr.activate(); + await mgr.ready(); + + handle.fireState("reconnecting"); + await new Promise((r) => setTimeout(r, 10)); + handle.fireState("online"); + + const call = captureSpy.mock.calls.find( + ([event]) => event === "oss.runtime.channel_session_recovered", + ); + expect(call).toBeDefined(); + expect((call![1] as { downForMs: number }).downForMs).toBeGreaterThan(0); + await mgr.stop(); + } finally { + captureSpy.mockRestore(); + } + }); + + it("captures no recovery for an online transition that follows no outage", async () => { + const captureSpy = vi + .spyOn(telemetry, "capture") + .mockResolvedValue(undefined); + try { + const handle = observableHandle(); + const engine: ActivateChannelEngine = vi.fn(async () => handle); + const mgr = new ChannelManager({ + intelligence: fakeIntelligence(), + channels: [ + createChannel({ identifyUser: "platform", name: "support" }), + ], + activateChannel: engine, + }); + mgr.activate(); + await mgr.ready(); + + // A session may report `online` without a preceding drop; that is not a + // recovery and must not be reported as one. + handle.fireState("online"); + + expect(captureSpy).not.toHaveBeenCalledWith( + "oss.runtime.channel_session_recovered", + expect.anything(), + ); + await mgr.stop(); + } finally { + captureSpy.mockRestore(); + } + }); }); diff --git a/packages/runtime/src/v2/runtime/core/channel-manager.ts b/packages/runtime/src/v2/runtime/core/channel-manager.ts index a9dc1da1fc..6f7ea0d7e4 100644 --- a/packages/runtime/src/v2/runtime/core/channel-manager.ts +++ b/packages/runtime/src/v2/runtime/core/channel-manager.ts @@ -5,6 +5,8 @@ import { } from "./channel-activation-config"; import type { ChannelActivationConfig } from "./channel-activation-config"; import type { CopilotKitIntelligence } from "../intelligence-platform"; +import { telemetry } from "../telemetry"; +import type { AnalyticsEvents } from "../telemetry"; import { AbstractAgent, EventType } from "@ag-ui/client"; import type { AgentSubscriber, @@ -351,6 +353,8 @@ interface ChannelEntry { handleStopped: boolean; /** Epoch ms this outage episode began; unset while the session is healthy. */ downSince?: number; + /** Cause of THIS outage episode, replayed on each "still down" reminder. */ + downCause?: string; /** Next "still down" logger for this outage; cleared on recovery/teardown. */ reconnectLogTimer?: ReturnType; /** Delay before the next reminder; doubles after each emitted reminder. */ @@ -1610,15 +1614,34 @@ export class ChannelManager implements ChannelsControl { if (state === "reconnecting") { entry.status = "reconnecting"; entry.downSince ??= Date.now(); + // Remembered for the repeat line below: an operator reading a reminder + // hours into an outage should not have to find the first line to learn + // the cause. + if (cause !== undefined) entry.downCause = cause; this.log?.( `channel "${name}" managed session dropped; reconnecting (Phoenix auto-rejoin)${because}`, ); + this.captureChannelTelemetry("oss.runtime.channel_session_dropped", { + ...(detail?.reason !== undefined ? { reason: detail.reason } : {}), + ...(detail?.code !== undefined ? { code: detail.code } : {}), + }); this.startReconnectLog(name, entry); } else if (state === "online") { entry.status = "online"; this.clearReconnectLog(entry); + // Read BEFORE clearing `downSince`, and only when an outage was + // actually in progress — a session may report `online` with no + // preceding drop, which is not a recovery. + const downSince = entry.downSince; entry.downSince = undefined; + entry.downCause = undefined; this.log?.(`channel "${name}" managed session back online`); + if (downSince !== undefined) { + this.captureChannelTelemetry( + "oss.runtime.channel_session_recovered", + { downForMs: Date.now() - downSince }, + ); + } } else if (state === "gave_up") { // `error` here means "not sendable", NOT "dead": Phoenix keeps retrying // underneath and a successful rejoin restores `online`. Say so, or the @@ -1632,6 +1655,28 @@ export class ChannelManager implements ChannelsControl { }); } + /** + * Report a Channel connection event. A managed session that drops is + * otherwise invisible outside the host process — the `log` seam reaches only + * whoever reads that process's stdout, which for a self-hosted runtime is + * nobody who can act on it. + * + * Fire-and-forget, and failures are swallowed: telemetry must never break a + * live session. Same contract as `fireInstanceCreatedTelemetry`. The `try` + * also covers a `capture` that throws synchronously or returns no promise. + */ + private captureChannelTelemetry< + K extends + | "oss.runtime.channel_session_dropped" + | "oss.runtime.channel_session_recovered", + >(event: K, props: AnalyticsEvents[K]): void { + try { + void telemetry.capture(event, props).catch(() => {}); + } catch { + // Swallow — a telemetry transport must not take the session with it. + } + } + /** Rendered downtime for this outage episode (`"45s"`), or `"unknown"`. */ private downFor(entry: ChannelEntry): string { return entry.downSince === undefined @@ -1656,7 +1701,8 @@ export class ChannelManager implements ChannelsControl { return; } this.log?.( - `channel "${name}" managed session still down after ${this.downFor(entry)}; Phoenix is retrying`, + `channel "${name}" managed session still down after ${this.downFor(entry)}; Phoenix is retrying` + + (entry.downCause !== undefined ? ` — ${entry.downCause}` : ""), ); entry.reconnectLogDelayMs = Math.min( delayMs * 2, diff --git a/packages/runtime/src/v2/runtime/telemetry/events.ts b/packages/runtime/src/v2/runtime/telemetry/events.ts index 617e709eb6..fd24ffe500 100644 --- a/packages/runtime/src/v2/runtime/telemetry/events.ts +++ b/packages/runtime/src/v2/runtime/telemetry/events.ts @@ -13,8 +13,23 @@ export type AnalyticsEvents = { hashedLgcKey?: string; error?: string; }; + /** + * A managed Channel lost its gateway link. Carries only the cause we already + * compute for the log line — never the Channel name, which is a + * customer-chosen identifier, and never message content. + */ + "oss.runtime.channel_session_dropped": ChannelSessionDroppedInfo; + /** A managed Channel's gateway link came back, and how long it was gone. */ + "oss.runtime.channel_session_recovered": { downForMs: number }; }; +export interface ChannelSessionDroppedInfo { + /** Diagnosis of the drop, e.g. `the gateway host answered HTTP 502`. */ + reason?: string; + /** Transport/OS code when the transport named one, e.g. `ECONNRESET`. */ + code?: string; +} + export interface RuntimeInstanceCreatedInfo { actionsAmount: number; endpointTypes: string[];