spb/khaelor Public

1/**2 * KHAELOR3 * File: src/session/store.ts4 * Description: Append-only JSONL session log — serialized atomic appends, replay, torn-last-line recovery.5 *6 * Author: Simon-Pierre Boucher7 * Contact: contact@spboucher.ai8 */910import { appendFile, mkdir, readFile, truncate, writeFile } from "node:fs/promises";11import { homedir } from "node:os";12import { join } from "node:path";13import { KhaelorError, ulid } from "../shared/index.js";14import { EVENT_SCHEMA_VERSION, isDurableEventType } from "./events.js";15import type { DurableEvent, DurableEventInput } from "./events.js";16import type { DurableAppender } from "./bus.js";1718/** Default sessions root (ARCHITECTURE.md §5.1). */19export function defaultSessionsDir(): string {20  return join(homedir(), ".khaelor", "sessions");21}2223export interface SessionLogCreateOptions {24  projectHash: string;25  sessionsDir?: string;26  sessionId?: string; // defaults to a fresh ULID27}2829export interface SessionLogOpenOptions {30  projectHash: string;31  sessionId: string;32  sessionsDir?: string;33}3435export interface TornLineRecovery {36  /** Byte offset the file was truncated to. */37  truncatedTo: number;38  /** Diagnostics file containing the torn bytes. */39  tornFile: string;40}4142/**43 * Canonicalize a JSON-compatible value: object keys sorted recursively.44 * Applied to `ToolRequested.input` so replayed LLM history reproduces45 * identical bytes on every rebuild (EVENT_MODEL.md §5.1).46 */47export function canonicalizeJsonValue(value: unknown): unknown {48  if (Array.isArray(value)) return value.map(canonicalizeJsonValue);49  if (value !== null && typeof value === "object") {50    const source = value as Record<string, unknown>;51    const out: Record<string, unknown> = {};52    for (const key of Object.keys(source).sort()) {53      out[key] = canonicalizeJsonValue(source[key]);54    }55    return out;56  }57  return value;58}5960interface ParsedEnvelope {61  v: number;62  id: string;63  sessionId: string;64  seq: number;65  ts: number;66  type: string;67  payload: unknown;68}6970function validateEnvelope(value: unknown): ParsedEnvelope | null {71  if (value === null || typeof value !== "object" || Array.isArray(value)) return null;72  const o = value as Record<string, unknown>;73  if (typeof o["v"] !== "number") return null;74  if (typeof o["id"] !== "string" || o["id"].length === 0) return null;75  if (typeof o["sessionId"] !== "string" || o["sessionId"].length === 0) return null;76  if (typeof o["seq"] !== "number" || !Number.isInteger(o["seq"]) || (o["seq"] as number) < 1) return null;77  if (typeof o["ts"] !== "number") return null;78  if (typeof o["type"] !== "string" || o["type"].length === 0) return null;79  if (!("payload" in o)) return null;80  return o as unknown as ParsedEnvelope;81}8283interface PairingEntry {84  requestedSeq: number;85  closedSeq: number | null;86}8788/**89 * One append-only JSONL file per session — THE source of truth (ADR-3).90 * - `append` assigns the envelope synchronously (seq gapless, at enqueue)91 *   and serializes the write through a per-session queue: one complete92 *   line per write call, never interleaved (EVENT_MODEL.md §5.2).93 * - `open` replays with torn-last-line recovery (EVENT_MODEL.md §5.3).94 */95export class SessionLog implements DurableAppender {96  readonly sessionId: string;97  readonly filePath: string;98  /** Events replayed by `open()`; empty for a freshly created log. */99  readonly replayedEvents: readonly DurableEvent[];100  /** Set when `open()` recovered from a torn final line. */101  readonly recovery: TornLineRecovery | null;102103  private nextSeq: number;104  private queue: Promise<void> = Promise.resolve();105  private writeError: unknown = null;106  private closed = false;107  /** toolUseId → pairing bookkeeping for compaction-cut validation (EVENT_MODEL.md §6.5). */108  private readonly pairing = new Map<string, PairingEntry>();109110  private constructor(args: {111    sessionId: string;112    filePath: string;113    replayedEvents: DurableEvent[];114    recovery: TornLineRecovery | null;115  }) {116    this.sessionId = args.sessionId;117    this.filePath = args.filePath;118    this.replayedEvents = args.replayedEvents;119    this.recovery = args.recovery;120    const last = args.replayedEvents[args.replayedEvents.length - 1];121    this.nextSeq = last ? last.seq + 1 : 1;122    for (const event of args.replayedEvents) this.track(event);123  }124125  /** Next seq that will be assigned (last seq + 1). */126  get seqCursor(): number {127    return this.nextSeq;128  }129130  static async create(options: SessionLogCreateOptions): Promise<SessionLog> {131    const sessionsDir = options.sessionsDir ?? defaultSessionsDir();132    const sessionId = options.sessionId ?? ulid();133    const dir = join(sessionsDir, options.projectHash);134    await mkdir(dir, { recursive: true });135    const filePath = join(dir, `${sessionId}.jsonl`);136    // "wx": refuse to clobber an existing session log.137    await writeFile(filePath, "", { flag: "wx", encoding: "utf8" });138    return new SessionLog({ sessionId, filePath, replayedEvents: [], recovery: null });139  }140141  static async open(options: SessionLogOpenOptions): Promise<SessionLog> {142    const sessionsDir = options.sessionsDir ?? defaultSessionsDir();143    const filePath = join(sessionsDir, options.projectHash, `${options.sessionId}.jsonl`);144    let buf: Buffer;145    try {146      buf = await readFile(filePath);147    } catch (error) {148      throw new KhaelorError("session-log-io", `Cannot open session log: ${filePath}`, {149        cause: String(error),150      });151    }152153    const events: DurableEvent[] = [];154    let offset = 0;155    let lastValidEnd = 0;156    let tornStart = -1;157158    while (offset < buf.length) {159      const nl = buf.indexOf(0x0a, offset);160      if (nl === -1) {161        // Final bytes lack a trailing newline — torn write.162        tornStart = offset;163        break;164      }165      const line = buf.subarray(offset, nl).toString("utf8");166      const parsed = SessionLog.parseLine(line);167      if (parsed === null) {168        if (nl === buf.length - 1) {169          // Invalid final complete line — treated as torn (EVENT_MODEL.md §5.3.1).170          tornStart = offset;171          break;172        }173        // Corruption anywhere other than the final line: refuse (no silent repair).174        throw new KhaelorError(175          "session-log-corrupted",176          `Session log corrupted before the final line (byte offset ${offset}): ${filePath}. ` +177            "Refusing to resume; inspect the file manually.",178          { filePath, offset },179        );180      }181      const expectedSeq = events.length + 1;182      if (parsed.seq !== expectedSeq) {183        throw new KhaelorError(184          "session-log-corrupted",185          `Session log seq gap at line ${events.length + 1}: expected seq ${expectedSeq}, found ${parsed.seq}: ${filePath}`,186          { filePath, expectedSeq, foundSeq: parsed.seq },187        );188      }189      events.push(parsed as unknown as DurableEvent);190      lastValidEnd = nl + 1;191      offset = nl + 1;192    }193194    let recovery: TornLineRecovery | null = null;195    if (tornStart >= 0) {196      const tornFile = `${filePath}.torn`;197      await writeFile(tornFile, buf.subarray(tornStart));198      await truncate(filePath, lastValidEnd);199      recovery = { truncatedTo: lastValidEnd, tornFile };200    }201202    return new SessionLog({ sessionId: options.sessionId, filePath, replayedEvents: events, recovery });203  }204205  private static parseLine(line: string): ParsedEnvelope | null {206    if (line.length === 0) return null;207    let value: unknown;208    try {209      value = JSON.parse(line);210    } catch {211      return null;212    }213    return validateEnvelope(value);214  }215216  /**217   * Assign the envelope and enqueue the append. Synchronous by design:218   * seq is assigned at enqueue and is gapless; the write itself is219   * serialized behind all prior writes (write-ahead of bus publication).220   */221  append(input: DurableEventInput): DurableEvent {222    if (this.closed) {223      throw new KhaelorError("store-write-failed", "Session log is closed");224    }225    if (!isDurableEventType(input.type)) {226      throw new KhaelorError("invalid-event", `Not a durable event type: ${String(input.type)}`);227    }228    const payload =229      input.type === "tool.requested"230        ? { ...input.payload, input: canonicalizeJsonValue(input.payload.input) }231        : input.payload;232233    if (input.type === "context.compacted") {234      this.validateCompactionCut(input.payload.cut);235    }236237    // Envelope key order is fixed: v,id,sessionId,seq,ts,type,payload (EVENT_MODEL.md §5.1).238    const event = {239      v: EVENT_SCHEMA_VERSION,240      id: ulid(),241      sessionId: this.sessionId,242      seq: this.nextSeq++,243      ts: Date.now(),244      type: input.type,245      payload,246    } as DurableEvent;247248    this.track(event);249250    const line = JSON.stringify(event) + "\n";251    this.queue = this.queue252      .then(() => appendFile(this.filePath, line, "utf8"))253      .catch((error: unknown) => {254        if (this.writeError === null) this.writeError = error;255      });256    return event;257  }258259  /** Await all enqueued appends; throws if any write failed. */260  async flush(): Promise<void> {261    await this.queue;262    if (this.writeError !== null) {263      throw new KhaelorError("store-write-failed", `Failed writing session log: ${this.filePath}`, {264        cause: String(this.writeError),265      });266    }267  }268269  async close(): Promise<void> {270    this.closed = true;271    await this.flush();272  }273274  /**275   * Pairing bookkeeping: track open/closed tool_use ids so a compaction276   * cut can be validated before it is appended (EVENT_MODEL.md §6.5.3).277   */278  private track(event: DurableEvent): void {279    switch (event.type) {280      case "tool.requested":281        this.pairing.set(event.payload.toolUseId, { requestedSeq: event.seq, closedSeq: null });282        return;283      case "tool.completed":284      case "tool.failed":285      case "tool.cancelled": {286        const entry = this.pairing.get(event.payload.toolUseId);287        if (entry) entry.closedSeq = event.seq;288        return;289      }290      default:291        return;292    }293  }294295  private validateCompactionCut(cut: { fromSeq: number; toSeq: number }): void {296    if (!Number.isInteger(cut.fromSeq) || !Number.isInteger(cut.toSeq) || cut.fromSeq > cut.toSeq) {297      throw new KhaelorError("pairing-violation", "context.compacted: invalid cut range", { cut });298    }299    for (const [toolUseId, entry] of this.pairing) {300      const requestedInOrBefore = entry.requestedSeq <= cut.toSeq;301      const closedInOrBefore = entry.closedSeq !== null && entry.closedSeq <= cut.toSeq;302      // A tool_use at seq ≤ toSeq must have its tool_result at seq ≤ toSeq.303      if (requestedInOrBefore && !closedInOrBefore) {304        throw new KhaelorError(305          "pairing-violation",306          `context.compacted: cut would orphan tool_use ${toolUseId} (requested at seq ${entry.requestedSeq}, not closed within cut)`,307          { toolUseId, cut },308        );309      }310      // A tool_use before the cut start must not have its result consumed by the cut.311      if (312        entry.requestedSeq < cut.fromSeq &&313        entry.closedSeq !== null &&314        entry.closedSeq >= cut.fromSeq &&315        entry.closedSeq <= cut.toSeq316      ) {317        throw new KhaelorError(318          "pairing-violation",319          `context.compacted: cut would consume the tool_result of ${toolUseId} while keeping its tool_use`,320          { toolUseId, cut },321        );322      }323    }324  }325}326