feat(storage): make SQLite a profile the app can run on

apps/server/src/modules/agent/agenetes/conversation-stores.ts

@@ -0,0 +1,132 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+/**
+ * Which Agenetes conversation stores this deployment runs on.
+ *
+ * Agenetes takes its three storage ports at mount, once, while the storage
+ * profile is only known at runtime and the active Workspace can change under
+ * a running process. So the mounted stores are dispatchers: each call picks
+ * the implementation that suits the namespace it was handed.
+ *
+ * The choice is made per namespace rather than per process because that is
+ * where the answer actually lives. A namespace carries a `storage.root` when
+ * the Space it belongs to is a directory, and does not when it is rows — the
+ * same fact the Space facade reports, arriving here through Agenetes's own
+ * vocabulary.
+ *
+ * The in-memory fall-through is not a backend choice. It is what an *unnamed*
+ * namespace has always got: a conversation with no Space to belong to, which
+ * Agenetes explicitly treats as non-persistent.
+ */
+
+import {
+  FileEventLogStore,
+  FileThreadStore,
+  FileTurnStore,
+  InMemoryEventLogStore,
+  InMemoryThreadStore,
+  InMemoryTurnStore,
+} from '@agenetes/agenetes';
+
+import {
+  conversationTables,
+  SqliteEventLogStore,
+  SqliteThreadStore,
+  SqliteTurnStore,
+} from './sqlite-stores.js';
+
+import type {
+  EventLogEntry,
+  EventLogRecord,
+  EventLogStore,
+  PersistedTurn,
+  ThreadRecord,
+  ThreadStore,
+  TurnStartLogEntry,
+  TurnStore,
+} from '@agenetes/agenetes';
+import type { AgentSubmission, Namespace } from '@agenetes/protocol';
+
+interface Backing {
+  readonly threads: ThreadStore;
+  readonly events: EventLogStore;
+  readonly turns: TurnStore;
+}
+
+const file: Backing = {
+  threads: new FileThreadStore(),
+  events: new FileEventLogStore(),
+  turns: new FileTurnStore(),
+};
+
+const sqlite: Backing = {
+  threads: new SqliteThreadStore(),
+  events: new SqliteEventLogStore(),
+  turns: new SqliteTurnStore(),
+};
+
+/**
+ * Shared, so an unnamed namespace keeps one conversation for the life of the
+ * process instead of a fresh empty one per port.
+ */
+const memory: Backing = {
+  threads: new InMemoryThreadStore(),
+  events: new InMemoryEventLogStore(),
+  turns: new InMemoryTurnStore(),
+};
+
+/** The stores that own this namespace's durable conversation state. */
+function backingFor(namespace: Namespace): Backing {
+  // A directory to write into settles it: that is the Disk profile, and the
+  // file stores are what wrote whatever is already there.
+  if (namespace.storage?.root) return file;
+  if (namespace.name && conversationTables(namespace) !== null) return sqlite;
+  return memory;
+}
+
+export const conversationThreadStore: ThreadStore = {
+  upsert: (namespace, threadId, record: ThreadRecord) =>
+    backingFor(namespace).threads.upsert(namespace, threadId, record),
+  get: (namespace, threadId) =>
+    backingFor(namespace).threads.get(namespace, threadId),
+  list: (namespace) => backingFor(namespace).threads.list(namespace),
+  delete: (namespace, threadId) =>
+    backingFor(namespace).threads.delete(namespace, threadId),
+};
+
+export const conversationEventLogStore: EventLogStore = {
+  appendTurnStart: (
+    namespace,
+    threadId,
+    request: AgentSubmission | null,
+  ): TurnStartLogEntry =>
+    backingFor(namespace).events.appendTurnStart(namespace, threadId, request),
+  append: (namespace, threadId, event): EventLogEntry =>
+    backingFor(namespace).events.append(namespace, threadId, event),
+  read: (namespace, threadId, sinceSeq) =>
+    backingFor(namespace).events.read(namespace, threadId, sinceSeq),
+  readRecords: (namespace, threadId, sinceSeq) =>
+    backingFor(namespace).events.readRecords(namespace, threadId, sinceSeq),
+  maxSeq: (namespace, threadId) =>
+    backingFor(namespace).events.maxSeq(namespace, threadId),
+  replace: (namespace, threadId, records: readonly EventLogRecord[]) =>
+    backingFor(namespace).events.replace(namespace, threadId, records),
+  delete: (namespace, threadId) =>
+    backingFor(namespace).events.delete(namespace, threadId),
+};
+
+export const conversationTurnStore: TurnStore = {
+  append: (namespace, threadId, persisted: PersistedTurn) =>
+    backingFor(namespace).turns.append(namespace, threadId, persisted),
+  list: (namespace, threadId) =>
+    backingFor(namespace).turns.list(namespace, threadId),
+  count: (namespace, threadId) =>
+    backingFor(namespace).turns.count(namespace, threadId),
+  fence: (namespace, threadId) =>
+    backingFor(namespace).turns.fence(namespace, threadId),
+  replace: (namespace, threadId, persisted: readonly PersistedTurn[]) =>
+    backingFor(namespace).turns.replace(namespace, threadId, persisted),
+  delete: (namespace, threadId) =>
+    backingFor(namespace).turns.delete(namespace, threadId),
+};

apps/server/src/modules/agent/agenetes/drivers.ts

@@ -7,15 +7,15 @@ import {
   type AcpCreateSpec,
   type AcpTurnCtx,
 } from '@agenetes/acp-driver';
-import {
-  FileEventLogStore,
-  FileThreadStore,
-  FileTurnStore,
-  mountAgenetes,
-} from '@agenetes/agenetes';
+import { mountAgenetes } from '@agenetes/agenetes';
 import { getAgentTeamRegistry } from '@agenetes/agentlet-host';
 import { piDriverFactory, type PiTurnCtx } from '@agenetes/pi-driver';
 
+import {
+  conversationEventLogStore,
+  conversationThreadStore,
+  conversationTurnStore,
+} from './conversation-stores.js';
 import { type AgentHandle } from './handle.js';
 import { HISTORY_LOAD_SANITY_LIMIT } from './history-replay.js';
 import { huabuPiDriverPorts } from './pi-driver.js';
@@ -62,9 +62,11 @@ export const agenetes: Agenetes = mountAgenetes({
     [INTERNAL_DRIVER_KIND]: piDriverFactory({ ports: huabuPiDriverPorts }),
     [EXTERNAL_DRIVER_KIND]: externalDriver,
   },
-  threadStore: new FileThreadStore(),
-  eventLogStore: new FileEventLogStore(),
-  turnStore: new FileTurnStore(),
+  // Dispatchers, not one backing: which store owns a conversation depends on
+  // where its Space lives, and that is a runtime fact (`conversation-stores`).
+  threadStore: conversationThreadStore,
+  eventLogStore: conversationEventLogStore,
+  turnStore: conversationTurnStore,
   // Corruption guard, not a context budget: replay restores whatever the
   // live handle would still be holding, and trimming that is the
   // conversation's problem, not recovery's.

apps/server/src/modules/agent/agenetes/sqlite-stores.test.ts

@@ -0,0 +1,218 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+/**
+ * The Agenetes conversation stores against a real SQLite profile.
+ *
+ * The claim under test is the one a user would notice: a conversation held in
+ * a Space that has no directory survives a restart, and goes away with its
+ * Space. Everything is driven through the mounted profile rather than a stub,
+ * so a broken extension substrate or a missing cascade fails here.
+ */
+
+import { afterEach, describe, expect, it } from 'vitest';
+
+import {
+  conversationEventLogStore,
+  conversationThreadStore,
+  conversationTurnStore,
+} from './conversation-stores.js';
+import { deleteSpace } from '../../storage/index.js';
+import {
+  mountTestWorkspace,
+  type MountedTestStorage,
+} from '../../storage/testing.js';
+import { canvasAcpNamespace } from '../../workspace/paths.js';
+
+import type { StorageProfile } from '../../storage/profile.js';
+import type { ThreadRecord } from '@agenetes/agenetes';
+import type { AgentStateSnapshot, WorkloadSpec } from '@agenetes/protocol';
+
+const SQLITE: StorageProfile = {
+  structured: { kind: 'sqlite' },
+  blobs: { kind: 'disk' },
+};
+
+const CANVAS_ID = 'canvas-conversation';
+const THREAD_ID = 'thread-1';
+
+let mounted: MountedTestStorage | null = null;
+
+afterEach(async () => {
+  await mounted?.close();
+  mounted = null;
+});
+
+/** Open the profile and create the Space the conversation belongs to. */
+async function openWithSpace(): Promise<MountedTestStorage> {
+  const opened = await mountTestWorkspace(SQLITE, 'huabu-agenetes-sqlite-');
+  mounted = opened;
+  const created = await opened.storage.structured
+    .spaces()
+    .create({ canvasId: CANVAS_ID, title: 'Conversation Space' });
+  if (!created.ok) throw new Error('Expected to create the Space');
+  return opened;
+}
+
+function threadRecord(threadId = THREAD_ID): ThreadRecord {
+  return {
+    driverSchemaVersion: 1,
+    spec: {
+      kind: 'internal',
+      threadId,
+      namespace: { name: CANVAS_ID },
+    } as unknown as WorkloadSpec,
+    state: { status: 'idle' } as unknown as AgentStateSnapshot,
+  };
+}
+
+describe('Agenetes conversation stores on SQLite', () => {
+  it('keeps a Space with no directory out of the file stores', async () => {
+    await openWithSpace();
+    const namespace = canvasAcpNamespace(CANVAS_ID);
+
+    // The absence of `storage.root` is the whole signal: it is what tells the
+    // dispatcher this Space is not a folder.
+    expect(namespace.storage).toBeUndefined();
+    expect(namespace.name).toBe(CANVAS_ID);
+  });
+
+  it('round-trips threads, events, and folded turns', async () => {
+    await openWithSpace();
+    const namespace = canvasAcpNamespace(CANVAS_ID);
+
+    conversationThreadStore.upsert(namespace, THREAD_ID, threadRecord());
+    expect(conversationThreadStore.get(namespace, THREAD_ID)).toEqual(
+      threadRecord(),
+    );
+    expect(conversationThreadStore.list(namespace)).toHaveLength(1);
+
+    const start = conversationEventLogStore.appendTurnStart(
+      namespace,
+      THREAD_ID,
+      null,
+    );
+    expect(start).toMatchObject({ seq: 1, kind: 'turn_start', request: null });
+    const appended = conversationEventLogStore.append(namespace, THREAD_ID, {
+      type: 'text',
+      text: 'hello',
+    } as never);
+    expect(appended.seq).toBe(2);
+    expect(conversationEventLogStore.maxSeq(namespace, THREAD_ID)).toBe(2);
+
+    // `read` is the streamed frames only; `readRecords` includes the internal
+    // turn boundary.
+    expect(conversationEventLogStore.read(namespace, THREAD_ID)).toHaveLength(
+      1,
+    );
+    expect(
+      conversationEventLogStore.readRecords(namespace, THREAD_ID),
+    ).toHaveLength(2);
+    expect(
+      conversationEventLogStore.read(namespace, THREAD_ID, 2),
+    ).toHaveLength(0);
+
+    conversationTurnStore.append(namespace, THREAD_ID, {
+      turn: { id: 'turn-1' } as never,
+      seqStart: 1,
+      seqEnd: 2,
+    });
+    expect(conversationTurnStore.count(namespace, THREAD_ID)).toBe(1);
+    expect(conversationTurnStore.fence(namespace, THREAD_ID)).toBe(2);
+    expect(conversationTurnStore.list(namespace, THREAD_ID)).toEqual([
+      { turn: { id: 'turn-1' }, seqStart: 1, seqEnd: 2 },
+    ]);
+  });
+
+  it('isolates one Space from another', async () => {
+    const opened = await openWithSpace();
+    const other = 'canvas-conversation-other';
+    const created = await opened.storage.structured
+      .spaces()
+      .create({ canvasId: other, title: 'Other Space' });
+    if (!created.ok) throw new Error('Expected to create the second Space');
+
+    conversationThreadStore.upsert(
+      canvasAcpNamespace(CANVAS_ID),
+      THREAD_ID,
+      threadRecord(),
+    );
+
+    expect(
+      conversationThreadStore.get(canvasAcpNamespace(other), THREAD_ID),
+    ).toBeUndefined();
+    expect(conversationThreadStore.list(canvasAcpNamespace(other))).toEqual([]);
+  });
+
+  it('destroys a conversation with the Space that held it', async () => {
+    await openWithSpace();
+    const namespace = canvasAcpNamespace(CANVAS_ID);
+    conversationThreadStore.upsert(namespace, THREAD_ID, threadRecord());
+    conversationEventLogStore.append(namespace, THREAD_ID, {
+      type: 'text',
+      text: 'hello',
+    } as never);
+    conversationTurnStore.append(namespace, THREAD_ID, {
+      turn: { id: 'turn-1' } as never,
+      seqStart: 1,
+      seqEnd: 1,
+    });
+
+    await expect(deleteSpace(CANVAS_ID)).resolves.toEqual({
+      ok: true,
+      reason: 'deleted',
+    });
+
+    // The Space is gone, so there is no substrate to answer from — which is
+    // the port's rule, and is also what the foreign-key cascade leaves behind.
+    expect(conversationThreadStore.get(namespace, THREAD_ID)).toBeUndefined();
+    expect(conversationEventLogStore.maxSeq(namespace, THREAD_ID)).toBe(0);
+    expect(conversationTurnStore.count(namespace, THREAD_ID)).toBe(0);
+  });
+
+  it('survives a restart', async () => {
+    const opened = await openWithSpace();
+    const namespace = canvasAcpNamespace(CANVAS_ID);
+    conversationThreadStore.upsert(namespace, THREAD_ID, threadRecord());
+    conversationEventLogStore.appendTurnStart(namespace, THREAD_ID, null);
+    conversationEventLogStore.append(namespace, THREAD_ID, {
+      type: 'text',
+      text: 'hello',
+    } as never);
+    conversationTurnStore.append(namespace, THREAD_ID, {
+      turn: { id: 'turn-1' } as never,
+      seqStart: 1,
+      seqEnd: 2,
+    });
+
+    await opened.reopen();
+
+    // Same namespace, new connection: this is the whole reason these stores
+    // exist rather than the in-memory defaults.
+    const after = canvasAcpNamespace(CANVAS_ID);
+    expect(conversationThreadStore.get(after, THREAD_ID)).toEqual(
+      threadRecord(),
+    );
+    expect(conversationEventLogStore.maxSeq(after, THREAD_ID)).toBe(2);
+    expect(
+      conversationEventLogStore.readRecords(after, THREAD_ID),
+    ).toHaveLength(2);
+    expect(conversationTurnStore.fence(after, THREAD_ID)).toBe(2);
+  });
+
+  it('reports an unnamed namespace as having no durable place', async () => {
+    await openWithSpace();
+    const anonymous = canvasAcpNamespace('');
+
+    // Agenetes's own rule: a namespace with no name is non-persistent. It must
+    // not fall through to some other Space's tables.
+    expect(conversationThreadStore.list(anonymous)).toEqual([]);
+    conversationThreadStore.upsert(anonymous, THREAD_ID, threadRecord());
+    expect(conversationThreadStore.get(anonymous, THREAD_ID)).toEqual(
+      threadRecord(),
+    );
+    expect(
+      conversationThreadStore.get(canvasAcpNamespace(CANVAS_ID), THREAD_ID),
+    ).toBeUndefined();
+  });
+});