Function createExecutionStoresForWrite

packages/storage/src/execution-stores.ts:270–523

(
  lease: StorageRootLease<K, 'write'>,
  kind: K,
  extension: E,
)

}

async function createExecutionStoresForWrite<K extends StorageRootKind, E extends object>(
  lease: StorageRootLease<K, 'write'>,
  kind: K,
  extension: E,
): Promise<ExecutionStoresWriterBase<K> & E> {
  const sessionStore = createSessionStore(lease.canonicalPath);
  const agentRunStore = createSqliteAgentRunStore(lease.canonicalPath);
  const interactionStore =
    'interactionStore' in extension
      ? (extension.interactionStore as InteractiveInteractionStoreWriterFacade)
      : undefined;
  const runtimePersistence = await openRuntimeEventPersistence({
    workspaceRoot: lease.canonicalPath,
  }).catch(async (error) => {
    await sessionStore.close?.().catch(() => {});
    agentRunStore.close?.();
    if (interactionStore) closeSqliteInteractionStoreFacade(interactionStore);
    throw error;
  });
  const runtimeEventStore = runtimePersistence.runtimeEventStore;
  let conversationOperationalStateStore: ConversationOperationalStateStore;
  try {
    conversationOperationalStateStore = createConversationOperationalStateStore(
      lease.canonicalPath,
    );
  } catch (error) {
    await closeExecutionStorePersistence(sessionStore, runtimePersistence, {
      agentRunStore,
      interactionStore,
    }).catch(() => {});
    throw error;
  }
  const messageReceiptStore = createSqliteMessageReceiptStore(lease.canonicalPath);
  await Promise.all([agentRunStore.ready?.(), messageReceiptStore.ready()]).catch(async (error) => {
    await closeExecutionStorePersistence(sessionStore, runtimePersistence, {
      agentRunStore,
      conversationOperationalStateStore,
      messageReceiptStore,
      interactionStore,
    }).catch(() => {});
    throw error;
  });
  const run = <T>(operation: () => Promise<T>) =>
    runWithStorageRootLease(lease, kind, 'write', operation);

  const stores: ExecutionStoresWriterBase<K> & E = {
    ...extension,
    kind,
    [executionStoresWriterBrand]: kind,
    purgeConversationOperationalState: (sessionId) =>
      run(() => conversationOperationalStateStore.purge(sessionId)),
    sessionStore: {
      ready: () => run(() => sessionStore.ready()),
      create: (input, initialBoundary) => run(() => sessionStore.create(input, initialBoundary)),
      createImportedSession: (input, messages) =>
        run(() => sessionStore.createImportedSession(input, messages)),
      probeStableSessionCreate: (sessionId, requestFingerprint) =>
        run(() => sessionStore.probeStableSessionCreate(sessionId, requestFingerprint)),