From 73dc5ad043e3ad247853e080f27834db5eefb5da Mon Sep 17 00:00:00 2001 From: Vegard Stikbakke Date: Thu, 4 Jun 2026 09:41:42 +0200 Subject: [PATCH] feat(coding-agent): add session sync --- packages/coding-agent/CHANGELOG.md | 1 + packages/coding-agent/src/core/index.ts | 76 ++++ .../src/core/session-analytics-reader.ts | 141 ++++++++ .../src/core/session-analytics.ts | 331 ++++++++++++++++++ .../src/core/session-discovery.ts | 185 ++++++++++ .../coding-agent/src/core/session-sync-api.ts | 239 +++++++++++++ .../src/core/session-sync-payload.ts | 209 +++++++++++ .../src/core/session-sync-state.ts | 113 ++++++ .../coding-agent/src/core/session-sync.ts | 202 +++++++++++ .../coding-agent/src/core/settings-manager.ts | 34 ++ .../coding-agent/src/core/slash-commands.ts | 12 + packages/coding-agent/src/index.ts | 78 +++++ .../src/modes/interactive/interactive-mode.ts | 129 ++++++- .../test/session-analytics-reader.test.ts | 83 +++++ .../test/session-analytics.test.ts | 276 +++++++++++++++ .../test/session-discovery.test.ts | 98 ++++++ .../test/session-sync-api.test.ts | 129 +++++++ .../test/session-sync-payload.test.ts | 86 +++++ .../test/session-sync-state.test.ts | 58 +++ .../coding-agent/test/session-sync.test.ts | 151 ++++++++ .../coding-agent/test/slash-commands.test.ts | 16 + 21 files changed, 2644 insertions(+), 3 deletions(-) create mode 100644 packages/coding-agent/src/core/session-analytics-reader.ts create mode 100644 packages/coding-agent/src/core/session-analytics.ts create mode 100644 packages/coding-agent/src/core/session-discovery.ts create mode 100644 packages/coding-agent/src/core/session-sync-api.ts create mode 100644 packages/coding-agent/src/core/session-sync-payload.ts create mode 100644 packages/coding-agent/src/core/session-sync-state.ts create mode 100644 packages/coding-agent/src/core/session-sync.ts create mode 100644 packages/coding-agent/test/session-analytics-reader.test.ts create mode 100644 packages/coding-agent/test/session-analytics.test.ts create mode 100644 packages/coding-agent/test/session-discovery.test.ts create mode 100644 packages/coding-agent/test/session-sync-api.test.ts create mode 100644 packages/coding-agent/test/session-sync-payload.test.ts create mode 100644 packages/coding-agent/test/session-sync-state.test.ts create mode 100644 packages/coding-agent/test/session-sync.test.ts create mode 100644 packages/coding-agent/test/slash-commands.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index a0c747813..4ba6aff52 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -4,6 +4,7 @@ ### Added +- Added core session analytics sync client modules, opt-in background sync, and a `/session-sync` command for reading local session analytics, building zstd NDJSON payloads, sending idempotency-keyed zstd NDJSON uploads, and calling the pi.dev sync API. - Added containerization documentation and a Gondolin extension example for routing built-in tools into a local micro-VM. - Added Ant Ling provider selection and setup documentation. - Added NVIDIA NIM provider selection, setup documentation, and direct NIM request attribution headers. diff --git a/packages/coding-agent/src/core/index.ts b/packages/coding-agent/src/core/index.ts index 71c45e9c8..95dbb8175 100644 --- a/packages/coding-agent/src/core/index.ts +++ b/packages/coding-agent/src/core/index.ts @@ -74,4 +74,80 @@ export { type TurnStartEvent, type WorkingIndicatorOptions, } from "./extensions/index.ts"; +export { + hashSessionAnalyticsString, + type ProjectSessionAnalyticsOptions, + type ProjectSessionHeaderAnalyticsOptions, + projectSessionEntryForAnalytics, + projectSessionForAnalytics, + projectSessionHeaderForAnalytics, + SESSION_ANALYTICS_SCHEMA_VERSION, + type SessionAnalyticsContentStats, + type SessionAnalyticsEntryRecord, + type SessionAnalyticsRecord, + type SessionAnalyticsSessionRecord, + type SessionAnalyticsUsage, +} from "./session-analytics.ts"; +export { + type BuildSessionAnalyticsUploadOptions, + type BuildSessionAnalyticsUploadResult, + buildSessionAnalyticsUpload, +} from "./session-analytics-reader.ts"; +export { + type DiscoveredSession, + type DiscoverSessionFilesOptions, + type DiscoverSessionsOptions, + discoverSessionFiles, + discoverSessions, + type SessionDiscoveryPhase, + type SessionDiscoveryProgress, + type SessionDiscoveryProgressCallback, +} from "./session-discovery.ts"; +export { + type SessionSyncResult, + type SessionSyncStatus, + type SyncSessionAnalyticsOptions, + syncSessionAnalytics, +} from "./session-sync.ts"; +export { + DEFAULT_PI_DEV_URL, + getSessionSyncWatermark, + pollSessionSyncDeviceToken, + refreshSessionSyncAccessToken, + SESSION_SYNC_CLIENT_ID, + SESSION_SYNC_SCOPE, + SessionSyncApiError, + type SessionSyncApiOptions, + type SessionSyncDeviceFlowResponse, + type SessionSyncFetch, + type SessionSyncTokenResponse, + type SessionSyncUploadResponse, + type SessionSyncWatermarkResponse, + startSessionSyncDeviceFlow, + type UploadSessionAnalyticsOptions, + uploadSessionAnalytics, +} from "./session-sync-api.ts"; +export { + type BuildSessionSyncPayloadsOptions, + buildSessionSyncPayloads, + compareSessionAnalyticsRecords, + getSessionAnalyticsRecordTimestamp, + SESSION_SYNC_CONTENT_ENCODING, + SESSION_SYNC_MAX_COMPRESSED_BYTES, + SESSION_SYNC_MAX_DECOMPRESSED_BYTES, + type SessionSyncPayload, + serializeSessionAnalyticsNdjson, + sortSessionAnalyticsRecords, +} from "./session-sync-payload.ts"; +export { + getSessionSyncStatePaths, + getStableSessionSyncDeviceId, + loadSessionSyncState, + type SessionSyncLockResult, + type SessionSyncState, + type SessionSyncStatePaths, + saveSessionSyncState, + updateSessionSyncState, + withSessionSyncLock, +} from "./session-sync-state.ts"; export { createSyntheticSourceInfo } from "./source-info.ts"; diff --git a/packages/coding-agent/src/core/session-analytics-reader.ts b/packages/coding-agent/src/core/session-analytics-reader.ts new file mode 100644 index 000000000..7f3f15bd4 --- /dev/null +++ b/packages/coding-agent/src/core/session-analytics-reader.ts @@ -0,0 +1,141 @@ +import { createReadStream } from "node:fs"; +import { createInterface } from "node:readline"; +import { projectSessionForAnalytics, type SessionAnalyticsRecord } from "./session-analytics.ts"; +import { discoverSessions, type SessionDiscoveryProgressCallback } from "./session-discovery.ts"; +import type { FileEntry, SessionEntry, SessionHeader } from "./session-manager.ts"; + +export interface BuildSessionAnalyticsUploadOptions { + /** Server watermark from GET /analytics/sessions/:deviceId. */ + serverWatermark: string | null; + /** Root sessions directory. Defaults to ~/.pi/agent/sessions. */ + sessionsRoot?: string; + scanCutoff?: Date; + signal?: AbortSignal; + onDiscoveryProgress?: SessionDiscoveryProgressCallback; +} + +export interface BuildSessionAnalyticsUploadResult { + records: SessionAnalyticsRecord[]; + scanCutoff: string; + filesScanned: number; + malformedFiles: number; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function isSessionHeader(value: unknown): value is SessionHeader { + if (!isRecord(value)) return false; + return ( + value.type === "session" && + typeof value.id === "string" && + value.id.length > 0 && + typeof value.timestamp === "string" && + value.timestamp.length > 0 && + typeof value.cwd === "string" && + (value.version === undefined || typeof value.version === "number") && + (value.parentSession === undefined || typeof value.parentSession === "string") + ); +} + +function isSessionEntry(value: unknown): value is SessionEntry { + if (!isRecord(value)) return false; + return ( + typeof value.type === "string" && + value.type !== "session" && + typeof value.id === "string" && + value.id.length > 0 && + (value.parentId === null || typeof value.parentId === "string") && + typeof value.timestamp === "string" && + value.timestamp.length > 0 + ); +} + +function parseIsoTime(value: string): number | undefined { + const time = new Date(value).getTime(); + return Number.isNaN(time) ? undefined : time; +} + +function getRecordTimestamp(record: SessionAnalyticsRecord): string | undefined { + if (record.recordType === "entry") return record.timestamp; + return record.createdAt ?? record.modifiedAt; +} + +function recordIsBeforeScanCutoff(record: SessionAnalyticsRecord, scanCutoffTime: number): boolean { + const timestamp = getRecordTimestamp(record); + if (!timestamp) return false; + const recordTime = parseIsoTime(timestamp); + return recordTime !== undefined && recordTime < scanCutoffTime; +} + +async function readSessionFile(path: string): Promise<{ header: SessionHeader; entries: SessionEntry[] } | undefined> { + const stream = createReadStream(path, { encoding: "utf8" }); + const lines = createInterface({ input: stream, crlfDelay: Infinity }); + let header: SessionHeader | undefined; + const entries: SessionEntry[] = []; + + try { + for await (const line of lines) { + if (!line.trim()) continue; + let parsed: unknown; + try { + parsed = JSON.parse(line) as FileEntry; + } catch { + return undefined; + } + + if (!header) { + if (!isSessionHeader(parsed)) return undefined; + header = parsed; + continue; + } + + if (!isSessionEntry(parsed)) return undefined; + entries.push(parsed); + } + } finally { + lines.close(); + stream.destroy(); + } + + return header ? { header, entries } : undefined; +} + +export async function buildSessionAnalyticsUpload( + options: BuildSessionAnalyticsUploadOptions, +): Promise { + const scanCutoff = options.scanCutoff ?? new Date(); + const scanCutoffTime = scanCutoff.getTime(); + const serverWatermarkTime = options.serverWatermark ? parseIsoTime(options.serverWatermark) : undefined; + const sessions = await discoverSessions({ + sessionsRoot: options.sessionsRoot, + signal: options.signal, + onProgress: options.onDiscoveryProgress, + }); + const records: SessionAnalyticsRecord[] = []; + let filesScanned = 0; + let malformedFiles = 0; + + for (const session of sessions) { + if (options.signal?.aborted) break; + if (serverWatermarkTime !== undefined && session.modifiedAt.getTime() <= serverWatermarkTime) continue; + filesScanned++; + const parsed = await readSessionFile(session.path).catch(() => undefined); + if (!parsed) { + malformedFiles++; + continue; + } + const projectedRecords = projectSessionForAnalytics(parsed.header, parsed.entries, { + modifiedAt: session.modifiedAt, + }); + records.push(...projectedRecords.filter((record) => recordIsBeforeScanCutoff(record, scanCutoffTime))); + } + + return { + records, + scanCutoff: scanCutoff.toISOString(), + filesScanned, + malformedFiles, + }; +} diff --git a/packages/coding-agent/src/core/session-analytics.ts b/packages/coding-agent/src/core/session-analytics.ts new file mode 100644 index 000000000..211cccb2f --- /dev/null +++ b/packages/coding-agent/src/core/session-analytics.ts @@ -0,0 +1,331 @@ +import { createHash } from "crypto"; +import type { SessionEntry, SessionHeader } from "./session-manager.ts"; + +export const SESSION_ANALYTICS_SCHEMA_VERSION = 1; + +export interface SessionAnalyticsUsage { + input: number; + output: number; + cacheRead: number; + cacheWrite: number; + totalTokens: number; + costInput: number; + costOutput: number; + costCacheRead: number; + costCacheWrite: number; + costTotal: number; +} + +export interface SessionAnalyticsContentStats { + stringContent: boolean; + textBlocks: number; + imageBlocks: number; + thinkingBlocks: number; + redactedThinkingBlocks: number; + toolCallBlocks: number; + otherBlocks: number; +} + +export interface SessionAnalyticsSessionRecord { + recordType: "session"; + schemaVersion: typeof SESSION_ANALYTICS_SCHEMA_VERSION; + sessionId: string; + version?: number; + createdAt?: string; + modifiedAt?: string; + parentSessionHash?: string; +} + +export interface SessionAnalyticsEntryRecord { + recordType: "entry"; + schemaVersion: typeof SESSION_ANALYTICS_SCHEMA_VERSION; + sessionId: string; + entryId: string; + parentEntryId: string | null; + entryType: string; + timestamp: string; + + // Message-level metadata. Raw content, tool arguments, thinking text, and errors are intentionally omitted. + role?: string; + api?: string; + provider?: string; + model?: string; + responseModel?: string; + stopReason?: string; + hasError?: boolean; + usage?: SessionAnalyticsUsage; + contentStats?: SessionAnalyticsContentStats; + isError?: boolean; + exitCode?: number; + cancelled?: boolean; + truncated?: boolean; + excludeFromContext?: boolean; + + // Non-message entry metadata. Raw summaries, labels, names, custom data, and details are intentionally omitted. + modelId?: string; + thinkingLevel?: string; + activeToolCount?: number; + firstKeptEntryId?: string; + tokensBefore?: number; + fromHook?: boolean; + hasDetails?: boolean; + fromId?: string; + customType?: string; + display?: boolean; + hasData?: boolean; + targetId?: string | null; + hasLabel?: boolean; + hasName?: boolean; +} + +export type SessionAnalyticsRecord = SessionAnalyticsSessionRecord | SessionAnalyticsEntryRecord; + +export interface ProjectSessionHeaderAnalyticsOptions { + modifiedAt?: Date | string; + hashString?: (value: string) => string; +} + +export interface ProjectSessionAnalyticsOptions extends ProjectSessionHeaderAnalyticsOptions {} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function maybeString(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function maybeNumber(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) ? value : undefined; +} + +function asNumber(value: unknown): number { + return maybeNumber(value) ?? 0; +} + +function modifiedAtToString(value: Date | string | undefined): string | undefined { + if (value instanceof Date) return value.toISOString(); + return maybeString(value); +} + +function cleanRecord(record: T): T { + const mutableRecord = record as Record; + for (const key of Object.keys(mutableRecord)) { + if (mutableRecord[key] === undefined) delete mutableRecord[key]; + } + return record; +} + +export function hashSessionAnalyticsString(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +export function projectSessionHeaderForAnalytics( + header: SessionHeader, + options: ProjectSessionHeaderAnalyticsOptions = {}, +): SessionAnalyticsSessionRecord { + const hash = options.hashString ?? hashSessionAnalyticsString; + return cleanRecord({ + recordType: "session", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: header.id, + version: header.version, + createdAt: maybeString(header.timestamp), + modifiedAt: modifiedAtToString(options.modifiedAt), + parentSessionHash: header.parentSession ? hash(header.parentSession) : undefined, + }); +} + +function projectUsageForAnalytics(usage: unknown): SessionAnalyticsUsage | undefined { + if (!isRecord(usage)) return undefined; + const cost = isRecord(usage.cost) ? usage.cost : undefined; + return { + input: asNumber(usage.input), + output: asNumber(usage.output), + cacheRead: asNumber(usage.cacheRead), + cacheWrite: asNumber(usage.cacheWrite), + totalTokens: asNumber(usage.totalTokens), + costInput: asNumber(cost?.input), + costOutput: asNumber(cost?.output), + costCacheRead: asNumber(cost?.cacheRead), + costCacheWrite: asNumber(cost?.cacheWrite), + costTotal: asNumber(cost?.total), + }; +} + +function projectContentStatsForAnalytics(content: unknown): SessionAnalyticsContentStats { + const stats: SessionAnalyticsContentStats = { + stringContent: false, + textBlocks: 0, + imageBlocks: 0, + thinkingBlocks: 0, + redactedThinkingBlocks: 0, + toolCallBlocks: 0, + otherBlocks: 0, + }; + + if (typeof content === "string") { + stats.stringContent = true; + if (content.length > 0) stats.textBlocks = 1; + return stats; + } + + if (!Array.isArray(content)) { + stats.otherBlocks = 1; + return stats; + } + + for (const block of content) { + if (!isRecord(block)) { + stats.otherBlocks++; + continue; + } + switch (block.type) { + case "text": + stats.textBlocks++; + break; + case "image": + stats.imageBlocks++; + break; + case "thinking": + stats.thinkingBlocks++; + if (block.redacted === true) stats.redactedThinkingBlocks++; + break; + case "toolCall": + stats.toolCallBlocks++; + break; + default: + stats.otherBlocks++; + break; + } + } + + return stats; +} + +function createEntryBase(sessionId: string, entry: SessionEntry): SessionAnalyticsEntryRecord { + return { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId, + entryId: entry.id, + parentEntryId: entry.parentId, + entryType: entry.type, + timestamp: entry.timestamp, + }; +} + +function projectMessageEntryForAnalytics(sessionId: string, entry: Extract) { + const base = createEntryBase(sessionId, entry); + const message = isRecord(entry.message) ? entry.message : undefined; + const role = maybeString(message?.role) ?? "unknown"; + + if (role === "assistant") { + return cleanRecord({ + ...base, + role, + api: maybeString(message?.api), + provider: maybeString(message?.provider), + model: maybeString(message?.model), + responseModel: maybeString(message?.responseModel), + stopReason: maybeString(message?.stopReason), + hasError: + Boolean(message?.errorMessage) || message?.stopReason === "error" || message?.stopReason === "aborted", + usage: projectUsageForAnalytics(message?.usage), + contentStats: projectContentStatsForAnalytics(message?.content), + }); + } + + if (role === "user") return { ...base, role, contentStats: projectContentStatsForAnalytics(message?.content) }; + + if (role === "toolResult") { + return { + ...base, + role, + isError: message?.isError === true, + contentStats: projectContentStatsForAnalytics(message?.content), + }; + } + + if (role === "bashExecution") { + return cleanRecord({ + ...base, + role, + exitCode: maybeNumber(message?.exitCode), + cancelled: message?.cancelled === true, + truncated: message?.truncated === true, + excludeFromContext: message?.excludeFromContext === true, + }); + } + + if (role === "custom") { + return cleanRecord({ + ...base, + role, + customType: maybeString(message?.customType), + display: typeof message?.display === "boolean" ? message.display : undefined, + contentStats: projectContentStatsForAnalytics(message?.content), + }); + } + + return { ...base, role }; +} + +export function projectSessionEntryForAnalytics(sessionId: string, entry: SessionEntry): SessionAnalyticsEntryRecord { + switch (entry.type) { + case "message": + return projectMessageEntryForAnalytics(sessionId, entry); + case "model_change": + return { ...createEntryBase(sessionId, entry), provider: entry.provider, modelId: entry.modelId }; + case "thinking_level_change": + return { ...createEntryBase(sessionId, entry), thinkingLevel: entry.thinkingLevel }; + case "compaction": + return cleanRecord({ + ...createEntryBase(sessionId, entry), + firstKeptEntryId: entry.firstKeptEntryId, + tokensBefore: entry.tokensBefore, + fromHook: entry.fromHook, + hasDetails: entry.details !== undefined, + }); + case "branch_summary": + return cleanRecord({ + ...createEntryBase(sessionId, entry), + fromId: entry.fromId, + fromHook: entry.fromHook, + hasDetails: entry.details !== undefined, + }); + case "custom": + return cleanRecord({ + ...createEntryBase(sessionId, entry), + customType: entry.customType, + hasData: entry.data !== undefined, + }); + case "custom_message": + return cleanRecord({ + ...createEntryBase(sessionId, entry), + customType: entry.customType, + display: entry.display, + hasDetails: entry.details !== undefined, + contentStats: projectContentStatsForAnalytics(entry.content), + }); + case "label": + return cleanRecord({ + ...createEntryBase(sessionId, entry), + targetId: entry.targetId, + hasLabel: typeof entry.label === "string" && entry.label.length > 0, + }); + case "session_info": + return { ...createEntryBase(sessionId, entry), hasName: Boolean(entry.name?.trim()) }; + default: + return createEntryBase(sessionId, entry); + } +} + +export function projectSessionForAnalytics( + header: SessionHeader, + entries: SessionEntry[], + options: ProjectSessionAnalyticsOptions = {}, +): SessionAnalyticsRecord[] { + const session = projectSessionHeaderForAnalytics(header, options); + return [session, ...entries.map((entry) => projectSessionEntryForAnalytics(session.sessionId, entry))]; +} diff --git a/packages/coding-agent/src/core/session-discovery.ts b/packages/coding-agent/src/core/session-discovery.ts new file mode 100644 index 000000000..a8751b537 --- /dev/null +++ b/packages/coding-agent/src/core/session-discovery.ts @@ -0,0 +1,185 @@ +import type { Dirent } from "fs"; +import { createReadStream } from "fs"; +import { readdir, stat } from "fs/promises"; +import { basename, dirname, join, relative, resolve } from "path"; +import { createInterface } from "readline"; +import { getSessionsDir } from "../config.ts"; +import type { SessionHeader } from "./session-manager.ts"; + +export type SessionDiscoveryPhase = "scan" | "read"; + +export interface SessionDiscoveryProgress { + phase: SessionDiscoveryPhase; + foundFiles: number; + processedFiles: number; + sessions: number; + currentFile?: string; +} + +export type SessionDiscoveryProgressCallback = (progress: SessionDiscoveryProgress) => void; + +export interface DiscoverSessionFilesOptions { + /** Root sessions directory. Defaults to ~/.pi/agent/sessions. */ + sessionsRoot?: string; + signal?: AbortSignal; + onProgress?: SessionDiscoveryProgressCallback; +} + +export interface DiscoverSessionsOptions extends DiscoverSessionFilesOptions {} + +export interface DiscoveredSession { + path: string; + relativePath: string; + sessionDir: string; + sessionDirName: string; + header: SessionHeader; + sessionId: string; + cwd: string; + createdAt?: Date; + modifiedAt: Date; + sizeBytes: number; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function parseSessionHeader(value: unknown): SessionHeader | undefined { + if (!isRecord(value)) return undefined; + if (value.type !== "session") return undefined; + if (typeof value.id !== "string" || !value.id) return undefined; + if (typeof value.timestamp !== "string" || !value.timestamp) return undefined; + if (typeof value.cwd !== "string") return undefined; + if (value.version !== undefined && typeof value.version !== "number") return undefined; + if (value.parentSession !== undefined && typeof value.parentSession !== "string") return undefined; + return { + type: "session", + version: value.version, + id: value.id, + timestamp: value.timestamp, + cwd: value.cwd, + parentSession: value.parentSession, + }; +} + +function parseDate(value: string): Date | undefined { + const date = new Date(value); + return Number.isNaN(date.getTime()) ? undefined : date; +} + +function emitProgress( + onProgress: SessionDiscoveryProgressCallback | undefined, + progress: SessionDiscoveryProgress, +): void { + onProgress?.({ ...progress }); +} + +async function walkSessionFiles( + dir: string, + files: string[], + progress: SessionDiscoveryProgress, + signal: AbortSignal | undefined, + onProgress: SessionDiscoveryProgressCallback | undefined, +): Promise { + if (signal?.aborted) return; + let entries: Dirent[]; + try { + entries = await readdir(dir, { withFileTypes: true }); + } catch { + return; + } + + for (const entry of entries) { + if (signal?.aborted) return; + const path = join(dir, entry.name); + if (entry.isDirectory()) { + await walkSessionFiles(path, files, progress, signal, onProgress); + continue; + } + if (!entry.isFile() || !entry.name.endsWith(".jsonl")) continue; + files.push(path); + progress.foundFiles = files.length; + emitProgress(onProgress, progress); + } +} + +export async function discoverSessionFiles(options: DiscoverSessionFilesOptions = {}): Promise { + const sessionsRoot = resolve(options.sessionsRoot ?? getSessionsDir()); + const files: string[] = []; + const progress: SessionDiscoveryProgress = { + phase: "scan", + foundFiles: 0, + processedFiles: 0, + sessions: 0, + }; + emitProgress(options.onProgress, progress); + await walkSessionFiles(sessionsRoot, files, progress, options.signal, options.onProgress); + files.sort((a, b) => a.localeCompare(b)); + return files; +} + +async function readSessionHeader(filePath: string): Promise { + const stream = createReadStream(filePath, { encoding: "utf8" }); + const lines = createInterface({ input: stream, crlfDelay: Infinity }); + try { + for await (const line of lines) { + if (!line.trim()) return undefined; + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + return undefined; + } + return parseSessionHeader(parsed); + } + return undefined; + } finally { + lines.close(); + stream.destroy(); + } +} + +async function discoverSessionFromFile(sessionsRoot: string, filePath: string): Promise { + const [fileStats, header] = await Promise.all([stat(filePath), readSessionHeader(filePath)]); + if (!header) return undefined; + const sessionDir = dirname(filePath); + return { + path: filePath, + relativePath: relative(sessionsRoot, filePath), + sessionDir, + sessionDirName: basename(sessionDir), + header, + sessionId: header.id, + cwd: header.cwd, + createdAt: parseDate(header.timestamp), + modifiedAt: fileStats.mtime, + sizeBytes: fileStats.size, + }; +} + +export async function discoverSessions(options: DiscoverSessionsOptions = {}): Promise { + const sessionsRoot = resolve(options.sessionsRoot ?? getSessionsDir()); + const files = await discoverSessionFiles({ sessionsRoot, signal: options.signal, onProgress: options.onProgress }); + const sessions: DiscoveredSession[] = []; + const progress: SessionDiscoveryProgress = { + phase: "read", + foundFiles: files.length, + processedFiles: 0, + sessions: 0, + }; + emitProgress(options.onProgress, progress); + + for (const filePath of files) { + if (options.signal?.aborted) break; + emitProgress(options.onProgress, { ...progress, currentFile: filePath }); + const session = await discoverSessionFromFile(sessionsRoot, filePath).catch(() => undefined); + progress.processedFiles++; + if (session) { + sessions.push(session); + progress.sessions = sessions.length; + } + emitProgress(options.onProgress, { ...progress, currentFile: filePath }); + } + + return sessions; +} diff --git a/packages/coding-agent/src/core/session-sync-api.ts b/packages/coding-agent/src/core/session-sync-api.ts new file mode 100644 index 000000000..3b956be7e --- /dev/null +++ b/packages/coding-agent/src/core/session-sync-api.ts @@ -0,0 +1,239 @@ +import type { Buffer } from "node:buffer"; + +export const SESSION_SYNC_CLIENT_ID = "pi-coding-agent"; +export const SESSION_SYNC_SCOPE = "session_sync offline_access"; +export const DEFAULT_PI_DEV_URL = "https://pi.dev"; + +export interface SessionSyncDeviceFlowResponse { + device_code: string; + user_code: string; + verification_uri: string; + verification_uri_complete: string; + expires_in: number; + interval: number; +} + +export interface SessionSyncTokenResponse { + token_type: "Bearer"; + access_token: string; + refresh_token: string; + expires_in: number; + scope: string; +} + +export interface SessionSyncWatermarkResponse { + ok: true; + watermark: string | null; +} + +export interface SessionSyncUploadResponse { + ok: true; + records_received: number; + first_record_timestamp: string; + last_record_timestamp: string; + received_bytes: number; + watermark: string; +} + +export type SessionSyncFetch = typeof fetch; + +export interface SessionSyncApiOptions { + baseUrl?: string; + fetch?: SessionSyncFetch; +} + +export interface UploadSessionAnalyticsOptions extends SessionSyncApiOptions { + accessToken: string; + deviceId: string; + watermark: string; + idempotencyKey: string; + body: Buffer; + contentEncoding: "zstd"; +} + +export class SessionSyncApiError extends Error { + status: number; + errorCode?: string; + description?: string; + + constructor(status: number, errorCode: string | undefined, description: string | undefined) { + super(description ? `${errorCode ?? "session_sync_error"}: ${description}` : (errorCode ?? `HTTP ${status}`)); + this.name = "SessionSyncApiError"; + this.status = status; + this.errorCode = errorCode; + this.description = description; + } +} + +function getBaseUrl(baseUrl: string | undefined): string { + return (baseUrl ?? process.env.PI_DEV_URL ?? DEFAULT_PI_DEV_URL).replace(/\/$/, ""); +} + +function getFetch(fetchImpl: SessionSyncFetch | undefined): SessionSyncFetch { + return fetchImpl ?? fetch; +} + +function formBody(fields: Record): URLSearchParams { + const body = new URLSearchParams(); + for (const [key, value] of Object.entries(fields)) { + body.set(key, value); + } + return body; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +async function readJson(response: Response): Promise { + const text = await response.text(); + if (!text) return undefined; + try { + return JSON.parse(text) as unknown; + } catch { + return undefined; + } +} + +async function throwIfNotOk(response: Response): Promise { + if (response.ok) return; + const json = await readJson(response); + const errorCode = isRecord(json) && typeof json.error === "string" ? json.error : undefined; + const description = isRecord(json) && typeof json.description === "string" ? json.description : undefined; + throw new SessionSyncApiError(response.status, errorCode, description); +} + +function requireString(record: Record, key: string): string { + const value = record[key]; + if (typeof value !== "string" || value.length === 0) + throw new Error(`Invalid session sync response: missing ${key}`); + return value; +} + +function requireNumber(record: Record, key: string): number { + const value = record[key]; + if (typeof value !== "number" || !Number.isFinite(value)) { + throw new Error(`Invalid session sync response: missing ${key}`); + } + return value; +} + +function parseDeviceFlowResponse(json: unknown): SessionSyncDeviceFlowResponse { + if (!isRecord(json)) throw new Error("Invalid session sync device flow response"); + return { + device_code: requireString(json, "device_code"), + user_code: requireString(json, "user_code"), + verification_uri: requireString(json, "verification_uri"), + verification_uri_complete: requireString(json, "verification_uri_complete"), + expires_in: requireNumber(json, "expires_in"), + interval: requireNumber(json, "interval"), + }; +} + +function parseTokenResponse(json: unknown): SessionSyncTokenResponse { + if (!isRecord(json)) throw new Error("Invalid session sync token response"); + const tokenType = requireString(json, "token_type"); + if (tokenType !== "Bearer") throw new Error(`Invalid session sync token type: ${tokenType}`); + return { + token_type: "Bearer", + access_token: requireString(json, "access_token"), + refresh_token: requireString(json, "refresh_token"), + expires_in: requireNumber(json, "expires_in"), + scope: requireString(json, "scope"), + }; +} + +function parseWatermarkResponse(json: unknown): SessionSyncWatermarkResponse { + if (!isRecord(json) || json.ok !== true || (json.watermark !== null && typeof json.watermark !== "string")) { + throw new Error("Invalid session sync watermark response"); + } + return { ok: true, watermark: json.watermark }; +} + +function parseUploadResponse(json: unknown): SessionSyncUploadResponse { + if (!isRecord(json) || json.ok !== true) throw new Error("Invalid session sync upload response"); + return { + ok: true, + records_received: requireNumber(json, "records_received"), + first_record_timestamp: requireString(json, "first_record_timestamp"), + last_record_timestamp: requireString(json, "last_record_timestamp"), + received_bytes: requireNumber(json, "received_bytes"), + watermark: requireString(json, "watermark"), + }; +} + +export async function startSessionSyncDeviceFlow( + deviceId: string, + options: SessionSyncApiOptions = {}, +): Promise { + const response = await getFetch(options.fetch)(`${getBaseUrl(options.baseUrl)}/api/oauth/device`, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: formBody({ client_id: SESSION_SYNC_CLIENT_ID, scope: SESSION_SYNC_SCOPE, device_id: deviceId }), + }); + await throwIfNotOk(response); + return parseDeviceFlowResponse(await readJson(response)); +} + +export async function pollSessionSyncDeviceToken( + deviceCode: string, + options: SessionSyncApiOptions = {}, +): Promise { + const response = await getFetch(options.fetch)(`${getBaseUrl(options.baseUrl)}/api/oauth/token`, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: formBody({ + grant_type: "urn:ietf:params:oauth:grant-type:device_code", + client_id: SESSION_SYNC_CLIENT_ID, + device_code: deviceCode, + }), + }); + await throwIfNotOk(response); + return parseTokenResponse(await readJson(response)); +} + +export async function refreshSessionSyncAccessToken( + refreshToken: string, + options: SessionSyncApiOptions = {}, +): Promise { + const response = await getFetch(options.fetch)(`${getBaseUrl(options.baseUrl)}/api/oauth/token`, { + method: "POST", + headers: { "Content-Type": "application/x-www-form-urlencoded" }, + body: formBody({ grant_type: "refresh_token", client_id: SESSION_SYNC_CLIENT_ID, refresh_token: refreshToken }), + }); + await throwIfNotOk(response); + return parseTokenResponse(await readJson(response)); +} + +export async function getSessionSyncWatermark( + accessToken: string, + deviceId: string, + options: SessionSyncApiOptions = {}, +): Promise { + const response = await getFetch(options.fetch)(`${getBaseUrl(options.baseUrl)}/analytics/sessions/${deviceId}`, { + headers: { Authorization: `Bearer ${accessToken}` }, + }); + await throwIfNotOk(response); + return parseWatermarkResponse(await readJson(response)); +} + +export async function uploadSessionAnalytics( + options: UploadSessionAnalyticsOptions, +): Promise { + const response = await getFetch(options.fetch)( + `${getBaseUrl(options.baseUrl)}/analytics/sessions/${options.deviceId}`, + { + method: "POST", + headers: { + Authorization: `Bearer ${options.accessToken}`, + "Content-Type": "application/x-ndjson", + "Content-Encoding": options.contentEncoding, + "Pi-Sync-Watermark": options.watermark, + "Idempotency-Key": options.idempotencyKey, + }, + body: options.body, + }, + ); + await throwIfNotOk(response); + return parseUploadResponse(await readJson(response)); +} diff --git a/packages/coding-agent/src/core/session-sync-payload.ts b/packages/coding-agent/src/core/session-sync-payload.ts new file mode 100644 index 000000000..2e2df6ca1 --- /dev/null +++ b/packages/coding-agent/src/core/session-sync-payload.ts @@ -0,0 +1,209 @@ +import { Buffer } from "node:buffer"; +import { promisify } from "node:util"; +import { zstdCompress } from "node:zlib"; +import type { SessionAnalyticsRecord } from "./session-analytics.ts"; + +export const SESSION_SYNC_CONTENT_ENCODING = "zstd"; +export const SESSION_SYNC_MAX_COMPRESSED_BYTES = 25 * 1024 * 1024; +export const SESSION_SYNC_MAX_DECOMPRESSED_BYTES = 75 * 1024 * 1024; + +export interface BuildSessionSyncPayloadsOptions { + records: SessionAnalyticsRecord[]; + scanCutoff: string; + serverWatermark: string | null; + maxCompressedBytes?: number; + maxDecompressedBytes?: number; + compress?: (input: Buffer) => Promise; +} + +export interface SessionSyncPayload { + records: SessionAnalyticsRecord[]; + recordCount: number; + firstRecordTimestamp: string; + lastRecordTimestamp: string; + watermark: string; + contentEncoding: typeof SESSION_SYNC_CONTENT_ENCODING; + body: Buffer; + decompressedBytes: number; + compressedBytes: number; +} + +function parseIsoTime(value: string): number { + const time = new Date(value).getTime(); + if (Number.isNaN(time)) throw new Error(`Invalid session analytics timestamp: ${value}`); + return time; +} + +export function getSessionAnalyticsRecordTimestamp(record: SessionAnalyticsRecord): string { + const timestamp = record.recordType === "entry" ? record.timestamp : (record.createdAt ?? record.modifiedAt); + if (!timestamp) { + throw new Error(`Session analytics ${record.recordType} record is missing a timestamp`); + } + parseIsoTime(timestamp); + return timestamp; +} + +export function compareSessionAnalyticsRecords(a: SessionAnalyticsRecord, b: SessionAnalyticsRecord): number { + const aTimestamp = getSessionAnalyticsRecordTimestamp(a); + const bTimestamp = getSessionAnalyticsRecordTimestamp(b); + const byTime = parseIsoTime(aTimestamp) - parseIsoTime(bTimestamp); + if (byTime !== 0) return byTime; + if (a.recordType !== b.recordType) return a.recordType === "session" ? -1 : 1; + const aSessionId = a.sessionId; + const bSessionId = b.sessionId; + const bySession = aSessionId.localeCompare(bSessionId); + if (bySession !== 0) return bySession; + const aEntryId = a.recordType === "entry" ? a.entryId : ""; + const bEntryId = b.recordType === "entry" ? b.entryId : ""; + return aEntryId.localeCompare(bEntryId); +} + +export function sortSessionAnalyticsRecords(records: SessionAnalyticsRecord[]): SessionAnalyticsRecord[] { + return [...records].sort(compareSessionAnalyticsRecords); +} + +export function serializeSessionAnalyticsNdjson(records: SessionAnalyticsRecord[]): Buffer { + if (records.length === 0) throw new Error("Session analytics upload has no records"); + return Buffer.from(`${records.map((record) => JSON.stringify(record)).join("\n")}\n`, "utf8"); +} + +async function compressSessionSyncNdjson( + input: Buffer, + compress: ((input: Buffer) => Promise) | undefined, +): Promise { + if (compress) return compress(input); + if (typeof zstdCompress !== "function") { + throw new Error("Session sync requires Node zstd compression support, but node:zlib.zstdCompress is unavailable"); + } + const compressed = await promisify(zstdCompress)(input); + return Buffer.from(compressed); +} + +function getRecordGroups(records: SessionAnalyticsRecord[]): SessionAnalyticsRecord[][] { + const groups: SessionAnalyticsRecord[][] = []; + for (const record of records) { + const timestamp = getSessionAnalyticsRecordTimestamp(record); + const previousGroup = groups.at(-1); + if (previousGroup && getSessionAnalyticsRecordTimestamp(previousGroup[0]) === timestamp) { + previousGroup.push(record); + } else { + groups.push([record]); + } + } + return groups; +} + +function getPayloadWatermark( + records: SessionAnalyticsRecord[], + isOnlyPayload: boolean, + isFinalPayload: boolean, + scanCutoff: string, + serverWatermark: string | null, +): string { + if (isOnlyPayload || isFinalPayload) return scanCutoff; + const maxTimestamp = getSessionAnalyticsRecordTimestamp(records[records.length - 1]); + if (serverWatermark && parseIsoTime(maxTimestamp) <= parseIsoTime(serverWatermark)) return serverWatermark; + return maxTimestamp; +} + +async function createPayload( + records: SessionAnalyticsRecord[], + watermark: string, + maxCompressedBytes: number, + maxDecompressedBytes: number, + compress: ((input: Buffer) => Promise) | undefined, +): Promise { + const ndjson = serializeSessionAnalyticsNdjson(records); + if (ndjson.byteLength > maxDecompressedBytes) { + throw new Error( + `Session analytics payload exceeds decompressed size limit (${ndjson.byteLength} > ${maxDecompressedBytes} bytes)`, + ); + } + const body = await compressSessionSyncNdjson(ndjson, compress); + if (body.byteLength > maxCompressedBytes) { + throw new Error( + `Session analytics payload exceeds compressed size limit (${body.byteLength} > ${maxCompressedBytes} bytes)`, + ); + } + return { + records, + recordCount: records.length, + firstRecordTimestamp: getSessionAnalyticsRecordTimestamp(records[0]), + lastRecordTimestamp: getSessionAnalyticsRecordTimestamp(records[records.length - 1]), + watermark, + contentEncoding: SESSION_SYNC_CONTENT_ENCODING, + body, + decompressedBytes: ndjson.byteLength, + compressedBytes: body.byteLength, + }; +} + +async function payloadFits( + records: SessionAnalyticsRecord[], + maxCompressedBytes: number, + maxDecompressedBytes: number, + compress: ((input: Buffer) => Promise) | undefined, +): Promise { + const ndjson = serializeSessionAnalyticsNdjson(records); + if (ndjson.byteLength > maxDecompressedBytes) return false; + const body = await compressSessionSyncNdjson(ndjson, compress); + return body.byteLength <= maxCompressedBytes; +} + +export async function buildSessionSyncPayloads( + options: BuildSessionSyncPayloadsOptions, +): Promise { + if (options.records.length === 0) throw new Error("Session analytics upload has no records"); + const maxCompressedBytes = options.maxCompressedBytes ?? SESSION_SYNC_MAX_COMPRESSED_BYTES; + const maxDecompressedBytes = options.maxDecompressedBytes ?? SESSION_SYNC_MAX_DECOMPRESSED_BYTES; + const sortedRecords = sortSessionAnalyticsRecords(options.records); + + if (await payloadFits(sortedRecords, maxCompressedBytes, maxDecompressedBytes, options.compress)) { + return [ + await createPayload( + sortedRecords, + options.scanCutoff, + maxCompressedBytes, + maxDecompressedBytes, + options.compress, + ), + ]; + } + + const batches: SessionAnalyticsRecord[][] = []; + let current: SessionAnalyticsRecord[] = []; + for (const group of getRecordGroups(sortedRecords)) { + const candidate = [...current, ...group]; + if ( + current.length > 0 && + !(await payloadFits(candidate, maxCompressedBytes, maxDecompressedBytes, options.compress)) + ) { + batches.push(current); + current = []; + } + const next = [...current, ...group]; + if (!(await payloadFits(next, maxCompressedBytes, maxDecompressedBytes, options.compress))) { + throw new Error("Session analytics records with the same timestamp exceed the upload size limit"); + } + current = next; + } + if (current.length > 0) batches.push(current); + + return Promise.all( + batches.map((batch, index) => + createPayload( + batch, + getPayloadWatermark( + batch, + batches.length === 1, + index === batches.length - 1, + options.scanCutoff, + options.serverWatermark, + ), + maxCompressedBytes, + maxDecompressedBytes, + options.compress, + ), + ), + ); +} diff --git a/packages/coding-agent/src/core/session-sync-state.ts b/packages/coding-agent/src/core/session-sync-state.ts new file mode 100644 index 000000000..fd00ae49a --- /dev/null +++ b/packages/coding-agent/src/core/session-sync-state.ts @@ -0,0 +1,113 @@ +import { randomUUID } from "node:crypto"; +import { existsSync } from "node:fs"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { dirname, join } from "node:path"; +import lockfile from "proper-lockfile"; +import { getAgentDir } from "../config.ts"; +import { normalizePath } from "../utils/paths.ts"; +import type { SettingsManager } from "./settings-manager.ts"; + +export interface SessionSyncState { + refreshToken?: string; + lastAttemptAt?: string; + lastSuccessAt?: string; +} + +export type SessionSyncLockResult = { status: "acquired"; result: T } | { status: "already_running" }; + +export interface SessionSyncStatePaths { + agentDir: string; + statePath: string; + lockPath: string; +} + +export function getSessionSyncStatePaths(agentDir: string = getAgentDir()): SessionSyncStatePaths { + const resolvedAgentDir = normalizePath(agentDir); + return { + agentDir: resolvedAgentDir, + statePath: join(resolvedAgentDir, "session-sync.json"), + lockPath: join(resolvedAgentDir, "session-sync.lock"), + }; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function isString(value: unknown): value is string { + return typeof value === "string" && value.length > 0; +} + +function parseSessionSyncState(value: unknown): SessionSyncState { + if (!isRecord(value)) return {}; + return { + refreshToken: isString(value.refreshToken) ? value.refreshToken : undefined, + lastAttemptAt: isString(value.lastAttemptAt) ? value.lastAttemptAt : undefined, + lastSuccessAt: isString(value.lastSuccessAt) ? value.lastSuccessAt : undefined, + }; +} + +export async function loadSessionSyncState(agentDir?: string): Promise { + const { statePath } = getSessionSyncStatePaths(agentDir); + try { + return parseSessionSyncState(JSON.parse(await readFile(statePath, "utf8"))); + } catch (error) { + const code = error && typeof error === "object" && "code" in error ? String(error.code) : undefined; + if (code === "ENOENT") return {}; + throw error; + } +} + +export async function saveSessionSyncState(state: SessionSyncState, agentDir?: string): Promise { + const { statePath } = getSessionSyncStatePaths(agentDir); + await mkdir(dirname(statePath), { recursive: true, mode: 0o700 }); + await writeFile(statePath, JSON.stringify(state, null, 2), { encoding: "utf8", mode: 0o600 }); +} + +export async function updateSessionSyncState( + updater: (state: SessionSyncState) => SessionSyncState, + agentDir?: string, +): Promise { + const next = updater(await loadSessionSyncState(agentDir)); + await saveSessionSyncState(next, agentDir); + return next; +} + +function isLockError(error: unknown): boolean { + return typeof error === "object" && error !== null && "code" in error && String(error.code) === "ELOCKED"; +} + +export async function withSessionSyncLock( + fn: () => Promise, + agentDir?: string, +): Promise> { + const { lockPath } = getSessionSyncStatePaths(agentDir); + await mkdir(dirname(lockPath), { recursive: true, mode: 0o700 }); + if (!existsSync(lockPath)) await writeFile(lockPath, "", { mode: 0o600 }); + let release: (() => Promise) | undefined; + try { + release = await lockfile.lock(lockPath, { + stale: 10 * 60 * 1000, + update: 30 * 1000, + retries: 0, + realpath: false, + }); + } catch (error) { + if (isLockError(error)) return { status: "already_running" }; + throw error; + } + + try { + return { status: "acquired", result: await fn() }; + } finally { + if (release) await release(); + } +} + +export function getStableSessionSyncDeviceId(settingsManager: SettingsManager): string { + const existing = settingsManager.getSessionSyncDeviceId(); + if (existing) return existing; + const deviceId = randomUUID(); + settingsManager.setSessionSyncDeviceId(deviceId); + return deviceId; +} diff --git a/packages/coding-agent/src/core/session-sync.ts b/packages/coding-agent/src/core/session-sync.ts new file mode 100644 index 000000000..4a86aa1c6 --- /dev/null +++ b/packages/coding-agent/src/core/session-sync.ts @@ -0,0 +1,202 @@ +import { randomUUID } from "node:crypto"; +import { buildSessionAnalyticsUpload } from "./session-analytics-reader.ts"; +import { + getSessionSyncWatermark, + refreshSessionSyncAccessToken, + SessionSyncApiError, + type SessionSyncFetch, + uploadSessionAnalytics, +} from "./session-sync-api.ts"; +import { buildSessionSyncPayloads, type SessionSyncPayload } from "./session-sync-payload.ts"; +import { + getStableSessionSyncDeviceId, + loadSessionSyncState, + type SessionSyncState, + saveSessionSyncState, + withSessionSyncLock, +} from "./session-sync-state.ts"; +import { SettingsManager } from "./settings-manager.ts"; + +export type SessionSyncStatus = "uploaded" | "no_changes" | "not_authenticated" | "already_running" | "failed"; + +export interface SessionSyncResult { + status: SessionSyncStatus; + recordsSent?: number; + compressedBytes?: number; + decompressedBytes?: number; + watermark?: string; + filesScanned?: number; + error?: string; +} + +export interface SyncSessionAnalyticsOptions { + agentDir?: string; + sessionsRoot?: string; + settingsManager?: SettingsManager; + baseUrl?: string; + fetch?: SessionSyncFetch; + signal?: AbortSignal; + now?: Date; +} + +interface AccessTokenState { + accessToken: string; + refreshToken: string; +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +async function refreshAccessToken( + state: SessionSyncState, + options: SyncSessionAnalyticsOptions, +): Promise { + if (!state.refreshToken) return undefined; + const token = await refreshSessionSyncAccessToken(state.refreshToken, { + baseUrl: options.baseUrl, + fetch: options.fetch, + }); + state.refreshToken = token.refresh_token; + await saveSessionSyncState(state, options.agentDir); + return { accessToken: token.access_token, refreshToken: token.refresh_token }; +} + +async function uploadWithRefreshRetry( + state: SessionSyncState, + access: AccessTokenState, + payload: Pick, + metadata: { deviceId: string; idempotencyKey: string }, + options: SyncSessionAnalyticsOptions, +): Promise<{ watermark: string; access: AccessTokenState }> { + try { + const response = await uploadSessionAnalytics({ + baseUrl: options.baseUrl, + fetch: options.fetch, + accessToken: access.accessToken, + deviceId: metadata.deviceId, + watermark: payload.watermark, + idempotencyKey: metadata.idempotencyKey, + body: payload.body, + contentEncoding: payload.contentEncoding, + }); + return { watermark: response.watermark, access }; + } catch (error) { + if (!(error instanceof SessionSyncApiError) || error.status !== 401) throw error; + const refreshed = await refreshAccessToken(state, options); + if (!refreshed) throw error; + const response = await uploadSessionAnalytics({ + baseUrl: options.baseUrl, + fetch: options.fetch, + accessToken: refreshed.accessToken, + deviceId: metadata.deviceId, + watermark: payload.watermark, + idempotencyKey: metadata.idempotencyKey, + body: payload.body, + contentEncoding: payload.contentEncoding, + }); + return { watermark: response.watermark, access: refreshed }; + } +} + +async function getWatermarkWithRefreshRetry( + state: SessionSyncState, + access: AccessTokenState, + deviceId: string, + options: SyncSessionAnalyticsOptions, +): Promise<{ watermark: string | null; access: AccessTokenState }> { + try { + const response = await getSessionSyncWatermark(access.accessToken, deviceId, { + baseUrl: options.baseUrl, + fetch: options.fetch, + }); + return { watermark: response.watermark, access }; + } catch (error) { + if (!(error instanceof SessionSyncApiError) || error.status !== 401) throw error; + const refreshed = await refreshAccessToken(state, options); + if (!refreshed) throw error; + const response = await getSessionSyncWatermark(refreshed.accessToken, deviceId, { + baseUrl: options.baseUrl, + fetch: options.fetch, + }); + return { watermark: response.watermark, access: refreshed }; + } +} + +async function syncSessionAnalyticsUnlocked(options: SyncSessionAnalyticsOptions): Promise { + const settingsManager = options.settingsManager ?? SettingsManager.create(process.cwd(), options.agentDir); + const deviceId = getStableSessionSyncDeviceId(settingsManager); + await settingsManager.flush(); + const state = await loadSessionSyncState(options.agentDir); + state.lastAttemptAt = (options.now ?? new Date()).toISOString(); + await saveSessionSyncState(state, options.agentDir); + + let access = await refreshAccessToken(state, options); + if (!access) return { status: "not_authenticated" }; + + const watermarkResponse = await getWatermarkWithRefreshRetry(state, access, deviceId, options); + access = watermarkResponse.access; + // The server watermark means: the server has accepted everything this client had fully scanned/prepared through this local time. + const upload = await buildSessionAnalyticsUpload({ + serverWatermark: watermarkResponse.watermark, + sessionsRoot: options.sessionsRoot, + signal: options.signal, + }); + + if (upload.records.length === 0) { + await saveSessionSyncState(state, options.agentDir); + return { + status: "no_changes", + filesScanned: upload.filesScanned, + watermark: watermarkResponse.watermark ?? undefined, + }; + } + + const payloads = await buildSessionSyncPayloads({ + records: upload.records, + scanCutoff: upload.scanCutoff, + serverWatermark: watermarkResponse.watermark, + }); + let recordsSent = 0; + let compressedBytes = 0; + let decompressedBytes = 0; + let watermark = watermarkResponse.watermark ?? undefined; + + for (const payload of payloads) { + const uploaded = await uploadWithRefreshRetry( + state, + access, + payload, + { deviceId, idempotencyKey: randomUUID() }, + options, + ); + access = uploaded.access; + watermark = uploaded.watermark; + recordsSent += payload.recordCount; + compressedBytes += payload.compressedBytes; + decompressedBytes += payload.decompressedBytes; + state.lastSuccessAt = (options.now ?? new Date()).toISOString(); + await saveSessionSyncState(state, options.agentDir); + } + + return { + status: "uploaded", + recordsSent, + compressedBytes, + decompressedBytes, + watermark, + filesScanned: upload.filesScanned, + }; +} + +export async function syncSessionAnalytics(options: SyncSessionAnalyticsOptions = {}): Promise { + const locked = await withSessionSyncLock(async () => { + try { + return await syncSessionAnalyticsUnlocked(options); + } catch (error) { + return { status: "failed", error: errorMessage(error) } satisfies SessionSyncResult; + } + }, options.agentDir); + if (locked.status === "already_running") return { status: "already_running" }; + return locked.result; +} diff --git a/packages/coding-agent/src/core/settings-manager.ts b/packages/coding-agent/src/core/settings-manager.ts index 910b68aab..2d78a72fa 100644 --- a/packages/coding-agent/src/core/settings-manager.ts +++ b/packages/coding-agent/src/core/settings-manager.ts @@ -57,6 +57,15 @@ export interface WarningSettings { anthropicExtraUsage?: boolean; // default: true } +export interface TelemetrySettings { + sessionSyncDeviceId?: string; +} + +export interface SessionSyncSettings { + enabled?: boolean; + intervalHours?: number; +} + export type TransportSetting = Transport; /** @@ -110,6 +119,8 @@ export interface Settings { showHardwareCursor?: boolean; // Show terminal cursor while still positioning it for IME markdown?: MarkdownSettings; warnings?: WarningSettings; + telemetry?: TelemetrySettings; + sessionSync?: SessionSyncSettings; sessionDir?: string; // Custom session storage directory (same format as --session-dir CLI flag) httpIdleTimeoutMs?: number; // HTTP header/body idle timeout in milliseconds; 0 disables it websocketConnectTimeoutMs?: number; // WebSocket connect/open handshake timeout in milliseconds; 0 disables it @@ -1088,4 +1099,27 @@ export class SettingsManager { this.markModified("warnings"); this.save(); } + + getSessionSyncDeviceId(): string | undefined { + const deviceId = this.settings.telemetry?.sessionSyncDeviceId; + return deviceId && /^[0-9a-fA-F-]{36}$/.test(deviceId) ? deviceId : undefined; + } + + setSessionSyncDeviceId(deviceId: string): void { + if (!this.globalSettings.telemetry) { + this.globalSettings.telemetry = {}; + } + this.globalSettings.telemetry.sessionSyncDeviceId = deviceId; + this.markModified("telemetry", "sessionSyncDeviceId"); + this.save(); + } + + getSessionSyncSettings(): { enabled: boolean; intervalHours: number } { + const intervalHours = this.settings.sessionSync?.intervalHours; + return { + enabled: this.settings.sessionSync?.enabled ?? false, + intervalHours: + typeof intervalHours === "number" && Number.isFinite(intervalHours) ? Math.max(1, intervalHours) : 24, + }; + } } diff --git a/packages/coding-agent/src/core/slash-commands.ts b/packages/coding-agent/src/core/slash-commands.ts index e52892c81..22b51dd71 100644 --- a/packages/coding-agent/src/core/slash-commands.ts +++ b/packages/coding-agent/src/core/slash-commands.ts @@ -15,6 +15,10 @@ export interface BuiltinSlashCommand { description: string; } +export function isSessionSyncFeatureEnabled(sessionSyncEnv: string | undefined = process.env.PI_SESSION_SYNC): boolean { + return sessionSyncEnv === "1"; +} + export const BUILTIN_SLASH_COMMANDS: ReadonlyArray = [ { name: "settings", description: "Open settings menu" }, { name: "model", description: "Select model (opens selector UI)" }, @@ -25,6 +29,7 @@ export const BUILTIN_SLASH_COMMANDS: ReadonlyArray = [ { name: "copy", description: "Copy last agent message to clipboard" }, { name: "name", description: "Set session display name" }, { name: "session", description: "Show session info and stats" }, + { name: "session-sync", description: "Log in and sync session analytics with pi.dev" }, { name: "changelog", description: "Show changelog entries" }, { name: "hotkeys", description: "Show all keyboard shortcuts" }, { name: "fork", description: "Create a new fork from a previous user message" }, @@ -38,3 +43,10 @@ export const BUILTIN_SLASH_COMMANDS: ReadonlyArray = [ { name: "reload", description: "Reload keybindings, extensions, skills, prompts, and themes" }, { name: "quit", description: `Quit ${APP_NAME}` }, ]; + +export function getVisibleBuiltinSlashCommands( + sessionSyncEnv: string | undefined = process.env.PI_SESSION_SYNC, +): ReadonlyArray { + if (isSessionSyncFeatureEnabled(sessionSyncEnv)) return BUILTIN_SLASH_COMMANDS; + return BUILTIN_SLASH_COMMANDS.filter((command) => command.name !== "session-sync"); +} diff --git a/packages/coding-agent/src/index.ts b/packages/coding-agent/src/index.ts index 0cdfc7260..4e1bfbbcb 100644 --- a/packages/coding-agent/src/index.ts +++ b/packages/coding-agent/src/index.ts @@ -190,6 +190,35 @@ export { createWriteTool, type PromptTemplate, } from "./core/sdk.ts"; +export { + hashSessionAnalyticsString, + type ProjectSessionAnalyticsOptions, + type ProjectSessionHeaderAnalyticsOptions, + projectSessionEntryForAnalytics, + projectSessionForAnalytics, + projectSessionHeaderForAnalytics, + SESSION_ANALYTICS_SCHEMA_VERSION, + type SessionAnalyticsContentStats, + type SessionAnalyticsEntryRecord, + type SessionAnalyticsRecord, + type SessionAnalyticsSessionRecord, + type SessionAnalyticsUsage, +} from "./core/session-analytics.ts"; +export { + type BuildSessionAnalyticsUploadOptions, + type BuildSessionAnalyticsUploadResult, + buildSessionAnalyticsUpload, +} from "./core/session-analytics-reader.ts"; +export { + type DiscoveredSession, + type DiscoverSessionFilesOptions, + type DiscoverSessionsOptions, + discoverSessionFiles, + discoverSessions, + type SessionDiscoveryPhase, + type SessionDiscoveryProgress, + type SessionDiscoveryProgressCallback, +} from "./core/session-discovery.ts"; export { type BranchSummaryEntry, buildSessionContext, @@ -213,12 +242,61 @@ export { type SessionMessageEntry, type ThinkingLevelChangeEntry, } from "./core/session-manager.ts"; +export { + type SessionSyncResult, + type SessionSyncStatus, + type SyncSessionAnalyticsOptions, + syncSessionAnalytics, +} from "./core/session-sync.ts"; +export { + DEFAULT_PI_DEV_URL, + getSessionSyncWatermark, + pollSessionSyncDeviceToken, + refreshSessionSyncAccessToken, + SESSION_SYNC_CLIENT_ID, + SESSION_SYNC_SCOPE, + SessionSyncApiError, + type SessionSyncApiOptions, + type SessionSyncDeviceFlowResponse, + type SessionSyncFetch, + type SessionSyncTokenResponse, + type SessionSyncUploadResponse, + type SessionSyncWatermarkResponse, + startSessionSyncDeviceFlow, + type UploadSessionAnalyticsOptions, + uploadSessionAnalytics, +} from "./core/session-sync-api.ts"; +export { + type BuildSessionSyncPayloadsOptions, + buildSessionSyncPayloads, + compareSessionAnalyticsRecords, + getSessionAnalyticsRecordTimestamp, + SESSION_SYNC_CONTENT_ENCODING, + SESSION_SYNC_MAX_COMPRESSED_BYTES, + SESSION_SYNC_MAX_DECOMPRESSED_BYTES, + type SessionSyncPayload, + serializeSessionAnalyticsNdjson, + sortSessionAnalyticsRecords, +} from "./core/session-sync-payload.ts"; +export { + getSessionSyncStatePaths, + getStableSessionSyncDeviceId, + loadSessionSyncState, + type SessionSyncLockResult, + type SessionSyncState, + type SessionSyncStatePaths, + saveSessionSyncState, + updateSessionSyncState, + withSessionSyncLock, +} from "./core/session-sync-state.ts"; export { type CompactionSettings, type ImageSettings, type PackageSource, type RetrySettings, + type SessionSyncSettings, SettingsManager, + type TelemetrySettings, } from "./core/settings-manager.ts"; // Skills export { diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index b874495b6..532c5c9ad 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -81,7 +81,22 @@ import { BUILT_IN_PROVIDER_DISPLAY_NAMES } from "../../core/provider-display-nam import type { ResourceDiagnostic } from "../../core/resource-loader.ts"; import { formatMissingSessionCwdPrompt, MissingSessionCwdError } from "../../core/session-cwd.ts"; import { type SessionContext, SessionManager } from "../../core/session-manager.ts"; -import { BUILTIN_SLASH_COMMANDS } from "../../core/slash-commands.ts"; +import { syncSessionAnalytics } from "../../core/session-sync.ts"; +import { + pollSessionSyncDeviceToken, + SessionSyncApiError, + startSessionSyncDeviceFlow, +} from "../../core/session-sync-api.ts"; +import { + getStableSessionSyncDeviceId, + loadSessionSyncState, + saveSessionSyncState, +} from "../../core/session-sync-state.ts"; +import { + BUILTIN_SLASH_COMMANDS, + getVisibleBuiltinSlashCommands, + isSessionSyncFeatureEnabled, +} from "../../core/slash-commands.ts"; import type { SourceInfo } from "../../core/source-info.ts"; import { isInstallTelemetryEnabled } from "../../core/telemetry.ts"; import type { TruncationResult } from "../../core/tools/truncate.ts"; @@ -89,9 +104,11 @@ import { getChangelogPath, getNewEntries, parseChangelog } from "../../utils/cha import { copyToClipboard } from "../../utils/clipboard.ts"; import { extensionForImageMimeType, readClipboardImage } from "../../utils/clipboard-image.ts"; import { parseGitUrl } from "../../utils/git.ts"; +import { openBrowser } from "../../utils/open-browser.ts"; import { getCwdRelativePath } from "../../utils/paths.ts"; import { getPiUserAgent } from "../../utils/pi-user-agent.ts"; import { killTrackedDetachedChildren } from "../../utils/shell.ts"; +import { sleep } from "../../utils/sleep.ts"; import { ensureTool } from "../../utils/tools-manager.ts"; import { checkForNewPiVersion, type LatestPiRelease } from "../../utils/version-check.ts"; import { ArminComponent } from "./components/armin.ts"; @@ -471,7 +488,7 @@ export class InteractiveMode { private createBaseAutocompleteProvider(): AutocompleteProvider { // Define commands for autocomplete - const slashCommands: SlashCommand[] = BUILTIN_SLASH_COMMANDS.map((command) => ({ + const slashCommands: SlashCommand[] = getVisibleBuiltinSlashCommands().map((command) => ({ name: command.name, description: command.description, })); @@ -515,7 +532,7 @@ export class InteractiveMode { })); // Convert extension commands to SlashCommand format - const builtinCommandNames = new Set(slashCommands.map((c) => c.name)); + const builtinCommandNames = new Set(BUILTIN_SLASH_COMMANDS.map((c) => c.name)); const extensionCommands: SlashCommand[] = this.session.extensionRunner .getRegisteredCommands() .filter((cmd) => !builtinCommandNames.has(cmd.name)) @@ -761,6 +778,8 @@ export class InteractiveMode { } }); + this.maybeRunBackgroundSessionSync(); + // Show startup warnings const { migratedProviders, modelFallbackMessage, initialMessage, initialImages, initialMessages } = this.options; @@ -908,6 +927,25 @@ export class InteractiveMode { return undefined; } + private maybeRunBackgroundSessionSync(): void { + const settings = this.settingsManager.getSessionSyncSettings(); + if (!isSessionSyncFeatureEnabled() || !settings.enabled || process.env.PI_OFFLINE) return; + + void loadSessionSyncState(getAgentDir()) + .then((state) => { + const lastAttemptTime = state.lastAttemptAt ? new Date(state.lastAttemptAt).getTime() : 0; + if ( + Number.isFinite(lastAttemptTime) && + Date.now() - lastAttemptTime < settings.intervalHours * 60 * 60 * 1000 + ) { + return undefined; + } + return syncSessionAnalytics({ settingsManager: this.settingsManager }); + }) + .then(() => undefined) + .catch(() => undefined); + } + private reportInstallTelemetry(version: string): void { if (process.env.PI_OFFLINE) { return; @@ -2539,6 +2577,15 @@ export class InteractiveMode { this.editor.setText(""); return; } + if (text === "/session-sync" || text.startsWith("/session-sync ")) { + this.editor.setText(""); + if (!isSessionSyncFeatureEnabled()) { + this.showWarning("Session sync is in early access. Set PI_SESSION_SYNC=1 to use /session-sync."); + return; + } + await this.handleSessionSyncCommand(text); + return; + } if (text === "/changelog") { this.handleChangelogCommand(); this.editor.setText(""); @@ -5265,6 +5312,82 @@ export class InteractiveMode { this.ui.requestRender(); } + private async handleSessionSyncCommand(text: string): Promise { + if (!isSessionSyncFeatureEnabled()) { + this.showWarning("Session sync is in early access. Set PI_SESSION_SYNC=1 to use /session-sync."); + return; + } + if (text !== "/session-sync") { + this.showWarning("Usage: /session-sync"); + return; + } + if (!(await this.ensureSessionSyncAuthenticated())) return; + await this.runSessionSyncNow(); + } + + private async runSessionSyncNow(): Promise { + this.showStatus("Running session sync..."); + const result = await syncSessionAnalytics({ settingsManager: this.settingsManager }); + if (result.status === "uploaded") { + this.showStatus( + `Session sync uploaded ${result.recordsSent?.toLocaleString() ?? 0} records (${result.compressedBytes?.toLocaleString() ?? 0} compressed bytes)`, + ); + return; + } + if (result.status === "no_changes") { + this.showStatus("Session sync found no changes"); + return; + } + if (result.status === "not_authenticated") { + this.showWarning("Session sync is not authenticated. Run /session-sync to log in and sync."); + return; + } + if (result.status === "already_running") { + this.showWarning("Session sync is already running in another process."); + return; + } + this.showError(result.error ?? "Session sync failed"); + } + + private async ensureSessionSyncAuthenticated(): Promise { + const existingState = await loadSessionSyncState(getAgentDir()); + if (existingState.refreshToken) return true; + + const deviceId = getStableSessionSyncDeviceId(this.settingsManager); + await this.settingsManager.flush(); + this.showStatus("Starting session sync login..."); + const flow = await startSessionSyncDeviceFlow(deviceId).catch((error: unknown) => { + this.showError(error instanceof Error ? error.message : String(error)); + return undefined; + }); + if (!flow) return false; + openBrowser(flow.verification_uri_complete); + this.showStatus(`Opened ${flow.verification_uri_complete}. Waiting for approval...`); + let intervalMs = Math.max(1, flow.interval) * 1000; + const deadline = Date.now() + flow.expires_in * 1000; + while (Date.now() < deadline) { + await sleep(intervalMs); + try { + const token = await pollSessionSyncDeviceToken(flow.device_code); + const state = await loadSessionSyncState(getAgentDir()); + state.refreshToken = token.refresh_token; + await saveSessionSyncState(state, getAgentDir()); + this.showStatus("Session sync login complete"); + return true; + } catch (error) { + if (error instanceof SessionSyncApiError && error.errorCode === "authorization_pending") continue; + if (error instanceof SessionSyncApiError && error.errorCode === "slow_down") { + intervalMs += 1000; + continue; + } + this.showError(error instanceof Error ? error.message : String(error)); + return false; + } + } + this.showWarning("Session sync login expired. Run /session-sync again."); + return false; + } + private handleChangelogCommand(): void { const changelogPath = getChangelogPath(); const allEntries = parseChangelog(changelogPath); diff --git a/packages/coding-agent/test/session-analytics-reader.test.ts b/packages/coding-agent/test/session-analytics-reader.test.ts new file mode 100644 index 000000000..7b661569d --- /dev/null +++ b/packages/coding-agent/test/session-analytics-reader.test.ts @@ -0,0 +1,83 @@ +import { mkdirSync, mkdtempSync, rmSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { buildSessionAnalyticsUpload } from "../src/core/session-analytics-reader.ts"; + +const tempDirs: string[] = []; + +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "pi-session-analytics-reader-")); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +function writeJsonl(path: string, lines: unknown[]): void { + writeFileSync(path, `${lines.map((line) => JSON.stringify(line)).join("\n")}\n`); +} + +function sessionHeader(id: string, timestamp: string): unknown { + return { type: "session", version: 3, id, timestamp, cwd: `/work/${id}` }; +} + +function entry(id: string, timestamp: string): unknown { + return { type: "model_change", id, parentId: null, timestamp, provider: "anthropic", modelId: "model" }; +} + +describe("buildSessionAnalyticsUpload", () => { + it("selects files by mtime and includes old records before the scan cutoff", async () => { + const root = createTempDir(); + const project = join(root, "--project--"); + mkdirSync(project, { recursive: true }); + const oldFile = join(project, "old.jsonl"); + const changedFile = join(project, "changed.jsonl"); + writeJsonl(oldFile, [ + sessionHeader("old", "2026-01-01T00:00:00.000Z"), + entry("old-entry", "2026-01-01T00:00:01.000Z"), + ]); + writeJsonl(changedFile, [ + sessionHeader("changed", "2026-01-01T00:00:00.000Z"), + entry("included-old-entry", "2026-01-01T00:00:01.000Z"), + entry("future-entry", "2026-01-03T00:00:00.000Z"), + ]); + utimesSync(oldFile, new Date("2026-01-01T00:10:00.000Z"), new Date("2026-01-01T00:10:00.000Z")); + utimesSync(changedFile, new Date("2026-01-02T00:10:00.000Z"), new Date("2026-01-02T00:10:00.000Z")); + + const result = await buildSessionAnalyticsUpload({ + sessionsRoot: root, + serverWatermark: "2026-01-02T00:00:00.000Z", + scanCutoff: new Date("2026-01-02T12:00:00.000Z"), + }); + + expect(result.filesScanned).toBe(1); + expect(result.malformedFiles).toBe(0); + expect(result.scanCutoff).toBe("2026-01-02T12:00:00.000Z"); + expect( + result.records.map((record) => (record.recordType === "entry" ? record.entryId : record.sessionId)), + ).toEqual(["changed", "included-old-entry"]); + }); + + it("skips malformed selected files", async () => { + const root = createTempDir(); + const project = join(root, "--project--"); + mkdirSync(project, { recursive: true }); + const malformed = join(project, "malformed.jsonl"); + writeFileSync(malformed, `${JSON.stringify(sessionHeader("bad", "2026-01-01T00:00:00.000Z"))}\nnot json\n`); + + const result = await buildSessionAnalyticsUpload({ + sessionsRoot: root, + serverWatermark: null, + scanCutoff: new Date("2026-01-02T00:00:00.000Z"), + }); + + expect(result.records).toEqual([]); + expect(result.filesScanned).toBe(1); + expect(result.malformedFiles).toBe(1); + }); +}); diff --git a/packages/coding-agent/test/session-analytics.test.ts b/packages/coding-agent/test/session-analytics.test.ts new file mode 100644 index 000000000..2368eeb65 --- /dev/null +++ b/packages/coding-agent/test/session-analytics.test.ts @@ -0,0 +1,276 @@ +import { describe, expect, it } from "vitest"; +import { + projectSessionForAnalytics, + SESSION_ANALYTICS_SCHEMA_VERSION, + type SessionAnalyticsRecord, +} from "../src/core/session-analytics.ts"; +import type { SessionEntry, SessionHeader } from "../src/core/session-manager.ts"; + +const header: SessionHeader = { + type: "session", + version: 3, + id: "session-1", + timestamp: "2026-01-02T03:04:05.000Z", + cwd: "/tmp/project", + parentSession: "/tmp/parent.jsonl", +}; + +const entries: SessionEntry[] = [ + { + type: "model_change", + id: "model-1", + parentId: null, + timestamp: "2026-01-02T03:04:06.000Z", + provider: "anthropic", + modelId: "claude-sonnet-4-5", + }, + { + type: "message", + id: "user-1", + parentId: "model-1", + timestamp: "2026-01-02T03:04:07.000Z", + message: { + role: "user", + content: [ + { type: "text", text: "secret user prompt" }, + { type: "image", data: "secret-image-data", mimeType: "image/png" }, + ], + timestamp: 1767323047000, + }, + }, + { + type: "message", + id: "assistant-1", + parentId: "user-1", + timestamp: "2026-01-02T03:04:08.000Z", + message: { + role: "assistant", + content: [ + { type: "text", text: "secret assistant answer" }, + { type: "thinking", thinking: "secret reasoning" }, + { type: "thinking", thinking: "", thinkingSignature: "secret-signature", redacted: true }, + { type: "toolCall", id: "call-1", name: "read", arguments: { path: "secret/path.ts" } }, + ], + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + responseModel: "claude-sonnet-4-5-20260101", + responseId: "secret-response-id", + usage: { + input: 100, + output: 20, + cacheRead: 30, + cacheWrite: 40, + totalTokens: 190, + cost: { + input: 0.1, + output: 0.2, + cacheRead: 0.03, + cacheWrite: 0.04, + total: 0.37, + }, + }, + stopReason: "toolUse", + timestamp: 1767323048000, + }, + }, + { + type: "message", + id: "tool-result-1", + parentId: "assistant-1", + timestamp: "2026-01-02T03:04:09.000Z", + message: { + role: "toolResult", + toolCallId: "call-1", + toolName: "read", + content: [{ type: "text", text: "secret tool result" }], + isError: true, + timestamp: 1767323049000, + }, + }, + { + type: "compaction", + id: "compaction-1", + parentId: "tool-result-1", + timestamp: "2026-01-02T03:04:10.000Z", + summary: "secret compaction summary", + firstKeptEntryId: "user-1", + tokensBefore: 1234, + details: { secret: "compaction details" }, + fromHook: true, + }, + { + type: "label", + id: "label-1", + parentId: "compaction-1", + timestamp: "2026-01-02T03:04:11.000Z", + targetId: "assistant-1", + label: "secret label", + }, + { + type: "session_info", + id: "session-info-1", + parentId: "label-1", + timestamp: "2026-01-02T03:04:12.000Z", + name: "secret session name", + }, +]; + +describe("projectSessionForAnalytics", () => { + it("projects a complete session into ordered analytics records", () => { + const records = projectSessionForAnalytics(header, entries, { + modifiedAt: "2026-01-03T04:05:06.000Z", + hashString: (value) => `hashed:${value}`, + }); + + expect(records).toEqual([ + { + recordType: "session", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + version: 3, + createdAt: "2026-01-02T03:04:05.000Z", + modifiedAt: "2026-01-03T04:05:06.000Z", + parentSessionHash: "hashed:/tmp/parent.jsonl", + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "model-1", + parentEntryId: null, + entryType: "model_change", + timestamp: "2026-01-02T03:04:06.000Z", + provider: "anthropic", + modelId: "claude-sonnet-4-5", + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "user-1", + parentEntryId: "model-1", + entryType: "message", + timestamp: "2026-01-02T03:04:07.000Z", + role: "user", + contentStats: { + stringContent: false, + textBlocks: 1, + imageBlocks: 1, + thinkingBlocks: 0, + redactedThinkingBlocks: 0, + toolCallBlocks: 0, + otherBlocks: 0, + }, + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "assistant-1", + parentEntryId: "user-1", + entryType: "message", + timestamp: "2026-01-02T03:04:08.000Z", + role: "assistant", + api: "anthropic-messages", + provider: "anthropic", + model: "claude-sonnet-4-5", + responseModel: "claude-sonnet-4-5-20260101", + stopReason: "toolUse", + hasError: false, + usage: { + input: 100, + output: 20, + cacheRead: 30, + cacheWrite: 40, + totalTokens: 190, + costInput: 0.1, + costOutput: 0.2, + costCacheRead: 0.03, + costCacheWrite: 0.04, + costTotal: 0.37, + }, + contentStats: { + stringContent: false, + textBlocks: 1, + imageBlocks: 0, + thinkingBlocks: 2, + redactedThinkingBlocks: 1, + toolCallBlocks: 1, + otherBlocks: 0, + }, + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "tool-result-1", + parentEntryId: "assistant-1", + entryType: "message", + timestamp: "2026-01-02T03:04:09.000Z", + role: "toolResult", + isError: true, + contentStats: { + stringContent: false, + textBlocks: 1, + imageBlocks: 0, + thinkingBlocks: 0, + redactedThinkingBlocks: 0, + toolCallBlocks: 0, + otherBlocks: 0, + }, + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "compaction-1", + parentEntryId: "tool-result-1", + entryType: "compaction", + timestamp: "2026-01-02T03:04:10.000Z", + firstKeptEntryId: "user-1", + tokensBefore: 1234, + fromHook: true, + hasDetails: true, + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "label-1", + parentEntryId: "compaction-1", + entryType: "label", + timestamp: "2026-01-02T03:04:11.000Z", + targetId: "assistant-1", + hasLabel: true, + }, + { + recordType: "entry", + schemaVersion: SESSION_ANALYTICS_SCHEMA_VERSION, + sessionId: "session-1", + entryId: "session-info-1", + parentEntryId: "label-1", + entryType: "session_info", + timestamp: "2026-01-02T03:04:12.000Z", + hasName: true, + }, + ] satisfies SessionAnalyticsRecord[]); + }); + + it("omits raw private payload fields", () => { + const serialized = JSON.stringify(projectSessionForAnalytics(header, entries)); + + expect(serialized).not.toContain("secret user prompt"); + expect(serialized).not.toContain("secret assistant answer"); + expect(serialized).not.toContain("secret reasoning"); + expect(serialized).not.toContain("secret-image-data"); + expect(serialized).not.toContain("secret/path.ts"); + expect(serialized).not.toContain("secret tool result"); + expect(serialized).not.toContain("secret compaction summary"); + expect(serialized).not.toContain("compaction details"); + expect(serialized).not.toContain("secret label"); + expect(serialized).not.toContain("secret session name"); + expect(serialized).not.toContain("secret-response-id"); + expect(serialized).not.toContain("/tmp/project"); + }); +}); diff --git a/packages/coding-agent/test/session-discovery.test.ts b/packages/coding-agent/test/session-discovery.test.ts new file mode 100644 index 000000000..ef48ea0c6 --- /dev/null +++ b/packages/coding-agent/test/session-discovery.test.ts @@ -0,0 +1,98 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "fs"; +import { tmpdir } from "os"; +import { basename, join, relative } from "path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + discoverSessionFiles, + discoverSessions, + type SessionDiscoveryProgress, +} from "../src/core/session-discovery.ts"; + +const tempDirs: string[] = []; + +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "pi-session-discovery-")); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +function writeJsonl(path: string, lines: unknown[]): void { + writeFileSync(path, `${lines.map((line) => JSON.stringify(line)).join("\n")}\n`); +} + +describe("session discovery", () => { + it("finds session jsonl files recursively", async () => { + const root = createTempDir(); + const projectA = join(root, "--project-a--"); + const projectB = join(root, "--project-b--", "nested"); + mkdirSync(projectA, { recursive: true }); + mkdirSync(projectB, { recursive: true }); + writeJsonl(join(projectA, "a.jsonl"), [ + { type: "session", version: 3, id: "a", timestamp: "2026-01-01T00:00:00.000Z", cwd: "/a" }, + ]); + writeJsonl(join(projectB, "b.jsonl"), [ + { type: "session", version: 3, id: "b", timestamp: "2026-01-02T00:00:00.000Z", cwd: "/b" }, + ]); + writeFileSync(join(projectA, "notes.txt"), "not a session"); + + const files = await discoverSessionFiles({ sessionsRoot: root }); + + expect(files.map((file) => relative(root, file))).toEqual([ + join("--project-a--", "a.jsonl"), + join("--project-b--", "nested", "b.jsonl"), + ]); + }); + + it("returns metadata for valid sessions and skips invalid jsonl files", async () => { + const root = createTempDir(); + const projectA = join(root, "--project-a--"); + const projectB = join(root, "--project-b--"); + mkdirSync(projectA, { recursive: true }); + mkdirSync(projectB, { recursive: true }); + writeJsonl(join(projectA, "a.jsonl"), [ + { + type: "session", + version: 3, + id: "session-a", + timestamp: "2026-01-01T00:00:00.000Z", + cwd: "/work/a", + parentSession: "/parent/session.jsonl", + }, + ]); + writeJsonl(join(projectB, "b.jsonl"), [ + { type: "session", version: 3, id: "session-b", timestamp: "2026-01-02T00:00:00.000Z", cwd: "/work/b" }, + ]); + writeFileSync(join(projectB, "invalid.jsonl"), "not json\n"); + + const progress: SessionDiscoveryProgress[] = []; + const sessions = await discoverSessions({ sessionsRoot: root, onProgress: (update) => progress.push(update) }); + + expect(sessions.map((session) => session.sessionId)).toEqual(["session-a", "session-b"]); + expect(sessions[0]).toMatchObject({ + path: join(projectA, "a.jsonl"), + relativePath: join("--project-a--", "a.jsonl"), + sessionDir: projectA, + sessionDirName: basename(projectA), + sessionId: "session-a", + cwd: "/work/a", + header: { + type: "session", + version: 3, + id: "session-a", + timestamp: "2026-01-01T00:00:00.000Z", + cwd: "/work/a", + parentSession: "/parent/session.jsonl", + }, + }); + expect(sessions[0].createdAt?.toISOString()).toBe("2026-01-01T00:00:00.000Z"); + expect(sessions[0].modifiedAt).toBeInstanceOf(Date); + expect(sessions[0].sizeBytes).toBeGreaterThan(0); + expect(progress.at(-1)).toMatchObject({ phase: "read", foundFiles: 3, processedFiles: 3, sessions: 2 }); + }); +}); diff --git a/packages/coding-agent/test/session-sync-api.test.ts b/packages/coding-agent/test/session-sync-api.test.ts new file mode 100644 index 000000000..1c944052a --- /dev/null +++ b/packages/coding-agent/test/session-sync-api.test.ts @@ -0,0 +1,129 @@ +import { Buffer } from "node:buffer"; +import { describe, expect, it } from "vitest"; +import { + getSessionSyncWatermark, + refreshSessionSyncAccessToken, + SessionSyncApiError, + startSessionSyncDeviceFlow, + uploadSessionAnalytics, +} from "../src/core/session-sync-api.ts"; + +function jsonResponse(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { status, headers: { "Content-Type": "application/json" } }); +} + +describe("session sync api", () => { + it("starts the OAuth device flow with the session sync scope", async () => { + let request: Request | undefined; + const fetchMock: typeof fetch = async (input, init) => { + request = new Request(input, init); + return jsonResponse({ + device_code: "pigd_1", + user_code: "ABCD-EFGH", + verification_uri: "https://pi.dev/pair", + verification_uri_complete: "https://pi.dev/pair?code=ABCD-EFGH", + expires_in: 300, + interval: 2, + }); + }; + + const response = await startSessionSyncDeviceFlow("00000000-0000-4000-8000-000000000000", { + baseUrl: "https://example.test/", + fetch: fetchMock, + }); + + expect(response.device_code).toBe("pigd_1"); + expect(request?.url).toBe("https://example.test/api/oauth/device"); + expect(request?.method).toBe("POST"); + const body = new URLSearchParams(await request?.text()); + expect(body.get("client_id")).toBe("pi-coding-agent"); + expect(body.get("scope")).toBe("session_sync offline_access"); + expect(body.get("device_id")).toBe("00000000-0000-4000-8000-000000000000"); + }); + + it("refreshes tokens and reads watermarks", async () => { + const urls: string[] = []; + const fetchMock: typeof fetch = async (input, init) => { + const request = new Request(input, init); + urls.push(request.url); + if (request.url.endsWith("/api/oauth/token")) { + return jsonResponse({ + token_type: "Bearer", + access_token: "access-2", + refresh_token: "refresh-2", + expires_in: 86400, + scope: "session_sync offline_access", + }); + } + expect(request.headers.get("Authorization")).toBe("Bearer access-2"); + return jsonResponse({ ok: true, watermark: "2026-01-01T00:00:00.000Z" }); + }; + + const token = await refreshSessionSyncAccessToken("refresh-1", { + baseUrl: "https://example.test", + fetch: fetchMock, + }); + const watermark = await getSessionSyncWatermark(token.access_token, "device-1", { + baseUrl: "https://example.test", + fetch: fetchMock, + }); + + expect(token.refresh_token).toBe("refresh-2"); + expect(watermark.watermark).toBe("2026-01-01T00:00:00.000Z"); + expect(urls).toEqual([ + "https://example.test/api/oauth/token", + "https://example.test/analytics/sessions/device-1", + ]); + }); + + it("uploads compressed NDJSON with sync headers and surfaces API errors", async () => { + let request: Request | undefined; + const fetchMock: typeof fetch = async (input, init) => { + request = new Request(input, init); + return jsonResponse( + { + ok: true, + records_received: 1, + first_record_timestamp: "2026-01-01T00:00:00.000Z", + last_record_timestamp: "2026-01-01T00:00:00.000Z", + received_bytes: 10, + watermark: "2026-01-02T00:00:00.000Z", + }, + 201, + ); + }; + + const response = await uploadSessionAnalytics({ + baseUrl: "https://example.test", + fetch: fetchMock, + accessToken: "access-1", + deviceId: "device-1", + watermark: "2026-01-02T00:00:00.000Z", + idempotencyKey: "retry-key", + body: Buffer.from("payload"), + contentEncoding: "zstd", + }); + + expect(response.records_received).toBe(1); + expect(request?.headers.get("Authorization")).toBe("Bearer access-1"); + expect(request?.headers.get("Content-Type")).toBe("application/x-ndjson"); + expect(request?.headers.get("Content-Encoding")).toBe("zstd"); + expect(request?.headers.get("Pi-Sync-Watermark")).toBe("2026-01-02T00:00:00.000Z"); + expect(request?.headers.get("Idempotency-Key")).toBe("retry-key"); + + const failingFetch: typeof fetch = async () => + jsonResponse({ ok: false, error: "invalid_payload", description: "bad line" }, 400); + await expect( + uploadSessionAnalytics({ + baseUrl: "https://example.test", + fetch: failingFetch, + accessToken: "access-1", + deviceId: "device-1", + watermark: "2026-01-02T00:00:00.000Z", + idempotencyKey: "retry-key", + body: Buffer.from("payload"), + contentEncoding: "zstd", + }), + ).rejects.toMatchObject(new SessionSyncApiError(400, "invalid_payload", "bad line")); + }); +}); diff --git a/packages/coding-agent/test/session-sync-payload.test.ts b/packages/coding-agent/test/session-sync-payload.test.ts new file mode 100644 index 000000000..cc113cf9d --- /dev/null +++ b/packages/coding-agent/test/session-sync-payload.test.ts @@ -0,0 +1,86 @@ +import { promisify } from "node:util"; +import { zstdDecompress } from "node:zlib"; +import { describe, expect, it } from "vitest"; +import type { SessionAnalyticsRecord } from "../src/core/session-analytics.ts"; +import { + buildSessionSyncPayloads, + serializeSessionAnalyticsNdjson, + sortSessionAnalyticsRecords, +} from "../src/core/session-sync-payload.ts"; + +const zstdDecompressAsync = promisify(zstdDecompress); + +function session(id: string, createdAt: string): SessionAnalyticsRecord { + return { recordType: "session", schemaVersion: 1, sessionId: id, createdAt }; +} + +function entry(id: string, timestamp: string): SessionAnalyticsRecord { + return { + recordType: "entry", + schemaVersion: 1, + sessionId: "session-1", + entryId: id, + parentEntryId: null, + entryType: "model_change", + timestamp, + }; +} + +describe("session sync payloads", () => { + it("sorts records oldest-first with sessions before entries at the same timestamp", () => { + const records = [ + entry("entry-b", "2026-01-02T00:00:00.000Z"), + session("session-1", "2026-01-01T00:00:00.000Z"), + entry("entry-a", "2026-01-01T00:00:00.000Z"), + ]; + + expect(sortSessionAnalyticsRecords(records)).toEqual([ + session("session-1", "2026-01-01T00:00:00.000Z"), + entry("entry-a", "2026-01-01T00:00:00.000Z"), + entry("entry-b", "2026-01-02T00:00:00.000Z"), + ]); + }); + + it("globally sorts, serializes, and zstd-compresses NDJSON", async () => { + const records = [entry("entry-1", "2026-01-01T00:00:01.000Z"), session("session-1", "2026-01-01T00:00:00.000Z")]; + const [payload] = await buildSessionSyncPayloads({ + records, + scanCutoff: "2026-01-02T00:00:00.000Z", + serverWatermark: null, + }); + + expect(payload.contentEncoding).toBe("zstd"); + expect(payload.watermark).toBe("2026-01-02T00:00:00.000Z"); + expect((await zstdDecompressAsync(payload.body)).toString("utf8")).toBe( + serializeSessionAnalyticsNdjson(sortSessionAnalyticsRecords(records)).toString("utf8"), + ); + }); + + it("globally sorts split payloads and uses scanCutoff only for the final batch watermark", async () => { + const records = [ + entry("new", "2026-01-03T00:00:00.000Z"), + entry("same-b", "2026-01-02T00:00:00.000Z"), + entry("old", "2026-01-01T00:00:00.000Z"), + entry("same-a", "2026-01-02T00:00:00.000Z"), + ]; + const payloads = await buildSessionSyncPayloads({ + records, + scanCutoff: "2026-01-04T00:00:00.000Z", + serverWatermark: "2026-01-01T12:00:00.000Z", + maxCompressedBytes: 400, + maxDecompressedBytes: 400, + compress: async (input) => input, + }); + + expect( + payloads.map((payload) => + payload.records.map((record) => (record.recordType === "entry" ? record.entryId : record.sessionId)), + ), + ).toEqual([["old"], ["same-a", "same-b"], ["new"]]); + expect(payloads.map((payload) => payload.watermark)).toEqual([ + "2026-01-01T12:00:00.000Z", + "2026-01-02T00:00:00.000Z", + "2026-01-04T00:00:00.000Z", + ]); + }); +}); diff --git a/packages/coding-agent/test/session-sync-state.test.ts b/packages/coding-agent/test/session-sync-state.test.ts new file mode 100644 index 000000000..7b079b964 --- /dev/null +++ b/packages/coding-agent/test/session-sync-state.test.ts @@ -0,0 +1,58 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + getStableSessionSyncDeviceId, + loadSessionSyncState, + saveSessionSyncState, + withSessionSyncLock, +} from "../src/core/session-sync-state.ts"; +import { SettingsManager } from "../src/core/settings-manager.ts"; + +const tempDirs: string[] = []; + +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "pi-session-sync-state-")); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +describe("session sync state", () => { + it("loads and saves sync state", async () => { + const agentDir = createTempDir(); + await saveSessionSyncState({ refreshToken: "refresh-1", lastAttemptAt: "2026-01-01T00:00:00.000Z" }, agentDir); + + expect(await loadSessionSyncState(agentDir)).toEqual({ + refreshToken: "refresh-1", + lastAttemptAt: "2026-01-01T00:00:00.000Z", + lastSuccessAt: undefined, + }); + }); + + it("stores a stable device id in telemetry settings", () => { + const settings = SettingsManager.inMemory(); + const first = getStableSessionSyncDeviceId(settings); + const second = getStableSessionSyncDeviceId(settings); + + expect(first).toBe(second); + expect(first).toMatch(/^[0-9a-f-]{36}$/); + expect(settings.getGlobalSettings().telemetry?.sessionSyncDeviceId).toBe(first); + }); + + it("returns already_running when the sync lock is held", async () => { + const agentDir = createTempDir(); + const result = await withSessionSyncLock( + async () => withSessionSyncLock(async () => "inner", agentDir), + agentDir, + ); + + expect(result).toEqual({ status: "acquired", result: { status: "already_running" } }); + }); +}); diff --git a/packages/coding-agent/test/session-sync.test.ts b/packages/coding-agent/test/session-sync.test.ts new file mode 100644 index 000000000..f055426bc --- /dev/null +++ b/packages/coding-agent/test/session-sync.test.ts @@ -0,0 +1,151 @@ +import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { syncSessionAnalytics } from "../src/core/session-sync.ts"; +import { loadSessionSyncState, saveSessionSyncState } from "../src/core/session-sync-state.ts"; +import { SettingsManager } from "../src/core/settings-manager.ts"; + +const tempDirs: string[] = []; + +function createTempDir(): string { + const dir = mkdtempSync(join(tmpdir(), "pi-session-sync-")); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0)) { + rmSync(dir, { recursive: true, force: true }); + } +}); + +function jsonResponse(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { status, headers: { "Content-Type": "application/json" } }); +} + +function writeSessionFile(sessionsRoot: string): void { + const sessionDir = join(sessionsRoot, "default"); + mkdirSync(sessionDir, { recursive: true }); + writeFileSync( + join(sessionDir, "session-1.jsonl"), + `${[ + { + type: "session", + version: 3, + id: "session-1", + timestamp: "2026-01-01T00:00:00.000Z", + cwd: "/tmp", + }, + { + type: "model_change", + id: "entry-1", + parentId: null, + timestamp: "2026-01-01T00:00:01.000Z", + provider: "openai", + modelId: "gpt-4.1", + }, + ] + .map((record) => JSON.stringify(record)) + .join("\n")}\n`, + ); +} + +describe("syncSessionAnalytics", () => { + it("returns not_authenticated after recording lastAttemptAt", async () => { + const agentDir = createTempDir(); + const result = await syncSessionAnalytics({ + agentDir, + settingsManager: SettingsManager.inMemory(), + now: new Date("2026-01-01T00:00:00.000Z"), + }); + + expect(result).toEqual({ status: "not_authenticated" }); + expect(await loadSessionSyncState(agentDir)).toMatchObject({ lastAttemptAt: "2026-01-01T00:00:00.000Z" }); + }); + + it("updates lastAttemptAt on no_changes", async () => { + const agentDir = createTempDir(); + const sessionsRoot = createTempDir(); + await saveSessionSyncState({ refreshToken: "refresh-1" }, agentDir); + const fetchMock: typeof fetch = async (input, init) => { + const request = new Request(input, init); + if (request.url.endsWith("/api/oauth/token")) { + return jsonResponse({ + token_type: "Bearer", + access_token: "access-1", + refresh_token: "refresh-2", + expires_in: 86400, + scope: "session_sync offline_access", + }); + } + return jsonResponse({ ok: true, watermark: null }); + }; + + const result = await syncSessionAnalytics({ + agentDir, + sessionsRoot, + settingsManager: SettingsManager.inMemory(), + fetch: fetchMock, + now: new Date("2026-01-02T00:00:00.000Z"), + }); + + expect(result).toMatchObject({ status: "no_changes", filesScanned: 0 }); + expect(await loadSessionSyncState(agentDir)).toMatchObject({ + refreshToken: "refresh-2", + lastAttemptAt: "2026-01-02T00:00:00.000Z", + }); + }); + + it("uploads with an idempotency key without persisting payload files", async () => { + const agentDir = createTempDir(); + const sessionsRoot = createTempDir(); + writeSessionFile(sessionsRoot); + await saveSessionSyncState({ refreshToken: "refresh-1" }, agentDir); + const idempotencyKeys: string[] = []; + const fetchMock: typeof fetch = async (input, init) => { + const request = new Request(input, init); + if (request.url.endsWith("/api/oauth/token")) { + return jsonResponse({ + token_type: "Bearer", + access_token: "access-1", + refresh_token: "refresh-2", + expires_in: 86400, + scope: "session_sync offline_access", + }); + } + if (request.method === "GET") return jsonResponse({ ok: true, watermark: null }); + idempotencyKeys.push(request.headers.get("Idempotency-Key") ?? ""); + expect((await request.arrayBuffer()).byteLength).toBeGreaterThan(0); + return jsonResponse({ + ok: true, + records_received: 2, + first_record_timestamp: "2026-01-01T00:00:00.000Z", + last_record_timestamp: "2026-01-01T00:00:01.000Z", + received_bytes: 21, + watermark: "2026-01-02T00:00:00.000Z", + }); + }; + + const result = await syncSessionAnalytics({ + agentDir, + sessionsRoot, + settingsManager: SettingsManager.inMemory(), + fetch: fetchMock, + now: new Date("2026-01-03T00:00:00.000Z"), + }); + + expect(result).toMatchObject({ + status: "uploaded", + recordsSent: 2, + watermark: "2026-01-02T00:00:00.000Z", + }); + expect(idempotencyKeys).toHaveLength(1); + expect(idempotencyKeys[0]).toMatch(/^[0-9a-f-]{36}$/); + expect(existsSync(join(agentDir, "session-sync-payloads"))).toBe(false); + expect(await loadSessionSyncState(agentDir)).toMatchObject({ + refreshToken: "refresh-2", + lastSuccessAt: "2026-01-03T00:00:00.000Z", + }); + }); +}); diff --git a/packages/coding-agent/test/slash-commands.test.ts b/packages/coding-agent/test/slash-commands.test.ts new file mode 100644 index 000000000..7f3387d8a --- /dev/null +++ b/packages/coding-agent/test/slash-commands.test.ts @@ -0,0 +1,16 @@ +import { describe, expect, it } from "vitest"; +import { getVisibleBuiltinSlashCommands, isSessionSyncFeatureEnabled } from "../src/core/slash-commands.ts"; + +function visibleCommandNames(sessionSyncEnv?: string): string[] { + return getVisibleBuiltinSlashCommands(sessionSyncEnv).map((command) => command.name); +} + +describe("slash commands", () => { + it("hides session sync unless the early access flag is set", () => { + expect(isSessionSyncFeatureEnabled(undefined)).toBe(false); + expect(isSessionSyncFeatureEnabled("true")).toBe(false); + expect(isSessionSyncFeatureEnabled("1")).toBe(true); + expect(visibleCommandNames("")).not.toContain("session-sync"); + expect(visibleCommandNames("1")).toContain("session-sync"); + }); +});