pi/packages/coding-agent/test/session-sync-payload.test.ts

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",
		]);
	});
});