src/agents/harness/context-engine-transcript-cursor.test.ts

import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import type { AgentMessage } from "openclaw/plugin-sdk/agent-core";
import {
  readSessionTranscriptVisibleMessageDelta,
  type SessionTranscriptTargetParams,
} from "openclaw/plugin-sdk/session-transcript-runtime";
import { describe, expect, it } from "vitest";
import {
  appendTranscriptMessage,
  replaceTranscriptEvents,
  upsertSessionEntryCore,
} from "../../config/sessions/session-accessor.js";
import type { ContextEngine, ContextEngineSessionTarget } from "../../context-engine/types.js";
import { createUserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.js";
import {
  bootstrapHarnessContextEngine,
  finalizeHarnessContextEngineTurn,
} from "./context-engine-lifecycle.js";

function requireTranscriptTarget(params: {
  sessionId: string;
  sessionKey?: string;
  sessionTarget?: ContextEngineSessionTarget;
}): SessionTranscriptTargetParams {
  const sessionId = params.sessionTarget?.sessionId ?? params.sessionId;
  const sessionKey = params.sessionTarget?.sessionKey ?? params.sessionKey;
  if (!sessionKey) {
    throw new Error("context engine transcript reads require a session key");
  }
  return {
    sessionId,
    sessionKey,
    ...(params.sessionTarget?.agentId ? { agentId: params.sessionTarget.agentId } : {}),
    ...(params.sessionTarget?.storePath ? { storePath: params.sessionTarget.storePath } : {}),
    ...(params.sessionTarget?.threadId !== undefined
      ? { threadId: params.sessionTarget.threadId }
      : {}),
  };
}

function readMessageContent(message: AgentMessage): unknown {
  return "content" in message ? message.content : undefined;
}

describe("context engine transcript cursor contract", () => {
  it("bootstraps, resumes appends, and rebuilds after replacement through the public SDK", async () => {
    const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-context-engine-cursor-"));
    const storePath = path.join(tempDir, "sessions.json");
    const target = {
      agentId: "main",
      sessionId: "context-engine-cursor",
      sessionKey: "agent:main:context-engine-cursor",
      sessionEntry: { sessionId: "context-engine-cursor", updatedAt: 10 },
      storePath,
    };
    const projectedMessages: AgentMessage[] = [];
    let cursor: string | undefined;
    let resetCount = 0;

    const consumeVisibleTranscript = async (params: {
      sessionId: string;
      sessionKey?: string;
      sessionTarget?: ContextEngineSessionTarget;
    }) => {
      const readTarget = requireTranscriptTarget(params);
      for (;;) {
        const result = await readSessionTranscriptVisibleMessageDelta({
          ...readTarget,
          ...(cursor ? { cursor } : {}),
          maxBytes: 10_000,
          maxMessages: 1,
        });
        if (result.kind === "missing" || result.kind === "unavailable") {
          return;
        }
        if (result.kind === "reset") {
          projectedMessages.length = 0;
          cursor = result.cursor;
          resetCount += 1;
          continue;
        }
        projectedMessages.push(...result.entries.map((entry) => entry.message));
        cursor = result.cursor;
        if (!result.hasMore) {
          return;
        }
      }
    };
    const engine: ContextEngine = {
      info: {
        id: "cursor-proof",
        name: "Cursor proof",
        transcriptSemantics: { currentTurnFence: "before-current-turn-entry-v1" },
      },
      bootstrap: async (params) => {
        await consumeVisibleTranscript(params);
        return { bootstrapped: true, importedMessages: projectedMessages.length };
      },
      ingest: async () => ({ ingested: true }),
      assemble: async (params) => ({ messages: params.messages, estimatedTokens: 0 }),
      compact: async () => ({ ok: true, compacted: false }),
      afterTurn: consumeVisibleTranscript,
    };
    const skipMaintenance: NonNullable<
      Parameters<typeof bootstrapHarnessContextEngine>[0]["runMaintenance"]
    > = async () => undefined;

    try {
      await upsertSessionEntryCore(target, { sessionId: target.sessionId, updatedAt: 10 });
      const first = await appendTranscriptMessage(target, {
        message: { role: "user", content: "first" },
        now: 1_000,
      });
      await appendTranscriptMessage(target, {
        message: { role: "assistant", content: "second" },
        parentId: first?.messageId,
        now: 2_000,
      });
      const admitted = await createUserTurnTranscriptRecorder({
        message: { role: "user", content: "third", timestamp: 3_000 },
        target,
        updateMode: "none",
      }).persistApproved();
      if (!admitted) {
        throw new Error("expected admitted user message with transcript admission");
      }
      const admission = admitted.admission;

      await bootstrapHarnessContextEngine({
        hadSessionFile: true,
        contextEngine: engine,
        sessionId: target.sessionId,
        sessionKey: target.sessionKey,
        sessionTarget: target,
        sessionFile: "sqlite://context-engine-cursor",
        transcriptReadFence: admission,
        runMaintenance: skipMaintenance,
        warn: () => {},
      });
      expect(projectedMessages.map(readMessageContent)).toEqual(["first", "second"]);

      await appendTranscriptMessage(target, {
        message: { role: "assistant", content: "fourth" },
        parentId: admitted.messageId,
        now: 4_000,
      });
      await finalizeHarnessContextEngineTurn({
        contextEngine: engine,
        promptError: false,
        aborted: false,
        yieldAborted: false,
        sessionIdUsed: target.sessionId,
        sessionKey: target.sessionKey,
        sessionTarget: target,
        sessionFile: "sqlite://context-engine-cursor",
        messagesSnapshot: [],
        prePromptMessageCount: 0,
        runMaintenance: skipMaintenance,
        warn: () => {},
      });
      expect(projectedMessages.map(readMessageContent)).toEqual([
        "first",
        "second",
        "third",
        "fourth",
      ]);

      await appendTranscriptMessage(target, {
        message: { role: "user", content: "failed turn" },
        now: 5_000,
      });
      await finalizeHarnessContextEngineTurn({
        contextEngine: engine,
        promptError: true,
        aborted: false,
        yieldAborted: false,
        sessionIdUsed: target.sessionId,
        sessionKey: target.sessionKey,
        sessionTarget: target,
        sessionFile: "sqlite://context-engine-cursor",
        messagesSnapshot: [],
        prePromptMessageCount: 0,
        runMaintenance: skipMaintenance,
        warn: () => {},
      });
      expect(projectedMessages.map(readMessageContent)).toEqual([
        "first",
        "second",
        "third",
        "fourth",
      ]);

      await replaceTranscriptEvents(target, [
        {
          type: "message",
          id: "replacement",
          parentId: null,
          timestamp: "1970-01-01T00:00:06.000Z",
          message: { role: "user", content: "replacement" },
        },
      ]);
      await finalizeHarnessContextEngineTurn({
        contextEngine: engine,
        promptError: false,
        aborted: false,
        yieldAborted: false,
        sessionIdUsed: target.sessionId,
        sessionKey: target.sessionKey,
        sessionTarget: target,
        sessionFile: "sqlite://context-engine-cursor",
        messagesSnapshot: [],
        prePromptMessageCount: 0,
        runMaintenance: skipMaintenance,
        warn: () => {},
      });
      expect(resetCount).toBe(1);
      expect(projectedMessages.map(readMessageContent)).toEqual(["replacement"]);
    } finally {
      fs.rmSync(tempDir, { force: true, recursive: true });
    }
  });
});