packages/storage/src/runtime-event-persistence.ts

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import { join } from 'node:path';
import type { RuntimeEvent } from '@maka/core/runtime-event';
import type {
  RuntimeInvocationPageInput,
  RuntimeInvocationPageResult,
  RuntimeInvocationRecord,
  RuntimeInvocationSearchResult,
} from '@maka/core/runtime-invocation';
import type { BoundedEvidenceReadResult, EvidenceReadBudget } from './agent-run-store.js';
import { createSqliteRuntimeStore, type SqliteRuntimeStore } from './sqlite-runtime-store.js';
import {
  acquireOperationalStateDatabase,
  OPERATIONAL_STATE_DATABASE_NAME,
} from './operational-state-store.js';

export type RuntimeEventPersistence = {
  kind: 'sqlite';
  runtimeEventStore: SqliteRuntimeStore;
  runtimeCommitStore: SqliteRuntimeStore;
  close(): void;
};

export type RuntimeEventReadPersistence = {
  kind: 'sqlite';
  runtimeEventStore: RuntimeEventReadStore;
  close(): void;
};

export interface RuntimeEventReadStore {
  listSessionInvocations(sessionId: string): Promise<RuntimeInvocationRecord[]>;
  readRunInvocation(sessionId: string, runId: string): Promise<RuntimeInvocationRecord | undefined>;
  listSessionInvocationsBounded(
    sessionId: string,
    limit: number,
  ): Promise<RuntimeInvocationSearchResult>;
  listSessionInvocationsPage(
    sessionId: string,
    input: RuntimeInvocationPageInput,
  ): Promise<RuntimeInvocationPageResult>;
  readInvocation(sessionId: string, invocationId: string): Promise<RuntimeInvocationRecord>;
  readRuntimeEvents(sessionId: string, runId: string): Promise<RuntimeEvent[]>;
  readRuntimeEventsBounded(
    sessionId: string,
    runId: string,
    budget: EvidenceReadBudget,
  ): Promise<BoundedEvidenceReadResult<RuntimeEvent>>;
  readImmutableRuntimeEvents(sessionId: string, runId: string): Promise<RuntimeEvent[]>;
  readSessionRuntimeEvents(sessionId: string): Promise<RuntimeEvent[]>;
  /** Session-wide events with the ordinal that fixes their transcript order. */
  readSessionRuntimeEventEntries(
    sessionId: string,
  ): Promise<ReadonlyArray<{ ordinal: number; event: RuntimeEvent }>>;
  /** Recall's narrowing over the ledger; see `RuntimeEventStore`. */
  listSessionsWithRuntimeEventText(
    sessionIds: readonly string[],
    terms: readonly string[],
  ): Promise<string[]>;
  countRuntimeEventMessages(sessionIds: readonly string[]): Promise<number>;
}

export async function openRuntimeEventPersistence(input: {
  workspaceRoot: string;
}): Promise<RuntimeEventPersistence> {
  const store = createWorkspaceRuntimeStore(input.workspaceRoot);
  return {
    kind: 'sqlite',
    runtimeEventStore: store,
    runtimeCommitStore: store,
    close: () => store.close(),
  };
}

export function createWorkspaceRuntimeStore(workspaceRoot: string): SqliteRuntimeStore {
  const databaseLease = acquireOperationalStateDatabase(workspaceRoot);
  return createSqliteRuntimeStore(join(workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME), {
    databaseLease,
  });
}

export async function openRuntimeEventReadPersistence(input: {
  workspaceRoot: string;
}): Promise<RuntimeEventReadPersistence> {
  const store = createSqliteRuntimeStore(
    join(input.workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME),
    { readOnly: true },
  );
  return {
    kind: 'sqlite',
    runtimeEventStore: Object.freeze({
      listSessionInvocations: (sessionId: string) => store.listSessionInvocations(sessionId),
      readRunInvocation: (sessionId: string, runId: string) =>
        store.readRunInvocation(sessionId, runId),
      listSessionInvocationsBounded: (sessionId: string, limit: number) =>
        store.listSessionInvocationsBounded(sessionId, limit),
      listSessionInvocationsPage: (sessionId: string, input: RuntimeInvocationPageInput) =>
        store.listSessionInvocationsPage(sessionId, input),
      readInvocation: (sessionId: string, invocationId: string) =>
        store.readInvocation(sessionId, invocationId),
      readRuntimeEvents: (sessionId: string, runId: string) =>
        store.readRuntimeEvents(sessionId, runId),
      readRuntimeEventsBounded: (sessionId: string, runId: string, budget: EvidenceReadBudget) =>
        store.readRuntimeEventsBounded(sessionId, runId, budget),
      readImmutableRuntimeEvents: (sessionId: string, runId: string) =>
        store.readImmutableRuntimeEvents(sessionId, runId),
      readSessionRuntimeEvents: (sessionId: string) => store.readSessionRuntimeEvents(sessionId),
      readSessionRuntimeEventEntries: (sessionId: string) =>
        store.readSessionRuntimeEventEntries(sessionId),
      listSessionsWithRuntimeEventText: (sessionIds: readonly string[], terms: readonly string[]) =>
        store.listSessionsWithRuntimeEventText(sessionIds, terms),
      countRuntimeEventMessages: (sessionIds: readonly string[]) =>
        store.countRuntimeEventMessages(sessionIds),
    }),
    close: () => store.close(),
  };
}