feat(storage): activate Postgres profiles with async agent persistence

apps/server/src/modules/agent/acp/external-agent-realization.ts

@@ -78,8 +78,10 @@ interface RealizationDependencies {
   readRecord: (
     namespace: Namespace,
     threadId: string,
-  ) => ReturnType<typeof agenetes.record>;
-  createHandle: (spec: AcpWorkloadSpec) => AcpHandle;
+  ) =>
+    | ReturnType<typeof agenetes.record>
+    | Awaited<ReturnType<typeof agenetes.record>>;
+  createHandle: (spec: AcpWorkloadSpec) => AcpHandle | Promise<AcpHandle>;
   buildSpec: typeof buildAcpWorkloadSpec;
   subscribeProfileCache: typeof ensureProfileCacheSubscription;
   ensureSession: (
@@ -107,7 +109,7 @@ async function ensureSessionFromCanonicalSpec(
     resolvedEnvironment || spec.spec.env
       ? { ...resolvedEnvironment, ...spec.spec.env }
       : undefined;
-  const record = agenetes.record(spec.namespace, spec.threadId);
+  const record = await agenetes.record(spec.namespace, spec.threadId);
   return ensureAcpSession({
     agentletId: resolveAcpAgentletId(spec),
     threadId: spec.threadId,
@@ -135,8 +137,9 @@ const DEFAULT_DEPENDENCIES: RealizationDependencies = {
   resolveFixedAgentNode: (canvasId, threadId) =>
     agentThreadResolver.resolveFixedAgentNode(canvasId, threadId),
   collectSpacePrompt: resolveSpacePrompt,
-  readRecord: (namespace, threadId) => agenetes.record(namespace, threadId),
-  createHandle: (spec) => agenetes.create(spec) as AcpHandle,
+  readRecord: async (namespace, threadId) =>
+    await agenetes.record(namespace, threadId),
+  createHandle: async (spec) => (await agenetes.create(spec)) as AcpHandle,
   buildSpec: buildAcpWorkloadSpec,
   subscribeProfileCache: ensureProfileCacheSubscription,
   ensureSession: ensureSessionFromCanonicalSpec,
@@ -201,7 +204,10 @@ export class ExternalAgentRealizationService {
               )
             : null))
         : options.agentTarget;
-    const record = this.dependencies.readRecord(namespace, options.threadId);
+    const record = await this.dependencies.readRecord(
+      namespace,
+      options.threadId,
+    );
 
     if (record) {
       if (record.spec.kind !== EXTERNAL_DRIVER_KIND) {
@@ -216,7 +222,7 @@ export class ExternalAgentRealizationService {
         binding,
         fixedTarget,
         spec,
-        handle: this.dependencies.createHandle(spec),
+        handle: await this.dependencies.createHandle(spec),
       };
       this.dependencies.subscribeProfileCache(
         options.threadId,
@@ -293,7 +299,7 @@ export class ExternalAgentRealizationService {
       binding,
       fixedTarget,
       spec,
-      handle: this.dependencies.createHandle(spec),
+      handle: await this.dependencies.createHandle(spec),
     };
     this.dependencies.subscribeProfileCache(
       options.threadId,

apps/server/src/modules/agent/acp/threads.route.ts

@@ -82,9 +82,15 @@ async function realizeControlThread(
   }
 }
 
-function resolveThreadAgentletId(threadId: string, canvasId?: string): string {
+async function resolveThreadAgentletId(
+  threadId: string,
+  canvasId?: string,
+): Promise<string> {
   if (canvasId) {
-    const record = agenetes.record(canvasAcpNamespace(canvasId), threadId);
+    const record = await agenetes.record(
+      canvasAcpNamespace(canvasId),
+      threadId,
+    );
     const driverSpec = record?.spec.spec;
     if (
       driverSpec &&
@@ -223,7 +229,7 @@ const acpThreadsRoutes: FastifyPluginAsync = async (app) => {
       });
     }
     const { canvasId, profileId } = parsed.data;
-    const agentletId = resolveThreadAgentletId(threadId, canvasId);
+    const agentletId = await resolveThreadAgentletId(threadId, canvasId);
     const live = acpSessionRegistry.get(agentletId, threadId);
     if (live) {
       return {
@@ -234,7 +240,10 @@ const acpThreadsRoutes: FastifyPluginAsync = async (app) => {
       };
     }
     if (canvasId) {
-      const record = agenetes.record(canvasAcpNamespace(canvasId), threadId);
+      const record = await agenetes.record(
+        canvasAcpNamespace(canvasId),
+        threadId,
+      );
       const persistedMeta = record?.state?.metadata;
       if (persistedMeta) {
         return {

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

@@ -30,20 +30,23 @@ import {
 } from '@agenetes/agenetes';
 
 import {
-  conversationTables,
+  PostgresThreadStore,
+  PostgresEventLogStore,
+  PostgresTurnStore,
+} from './postgres-stores.js';
+import {
   SqliteEventLogStore,
   SqliteThreadStore,
   SqliteTurnStore,
 } from './sqlite-stores.js';
+import { getStructuredStore } from '../../storage/index.js';
 
 import type {
-  EventLogEntry,
   EventLogRecord,
   EventLogStore,
   PersistedTurn,
   ThreadRecord,
   ThreadStore,
-  TurnStartLogEntry,
   TurnStore,
 } from '@agenetes/agenetes';
 import type { AgentSubmission, Namespace } from '@agenetes/protocol';
@@ -60,6 +63,12 @@ const file: Backing = {
   turns: new FileTurnStore(),
 };
 
+const postgres: Backing = {
+  threads: new PostgresThreadStore(),
+  events: new PostgresEventLogStore(),
+  turns: new PostgresTurnStore(),
+};
+
 const sqlite: Backing = {
   threads: new SqliteThreadStore(),
   events: new SqliteEventLogStore(),
@@ -81,8 +90,11 @@ 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;
+  if (!namespace.name) return memory;
+  const kind = getStructuredStore().kind;
+  if (kind === 'postgres') return postgres;
+  if (kind === 'sqlite') return sqlite;
+  throw new Error('A named Disk conversation requires a storage root');
 }
 
 export const conversationThreadStore: ThreadStore = {
@@ -96,13 +108,9 @@ export const conversationThreadStore: ThreadStore = {
 };
 
 export const conversationEventLogStore: EventLogStore = {
-  appendTurnStart: (
-    namespace,
-    threadId,
-    request: AgentSubmission | null,
-  ): TurnStartLogEntry =>
+  appendTurnStart: (namespace, threadId, request: AgentSubmission | null) =>
     backingFor(namespace).events.appendTurnStart(namespace, threadId, request),
-  append: (namespace, threadId, event): EventLogEntry =>
+  append: (namespace, threadId, event) =>
     backingFor(namespace).events.append(namespace, threadId, event),
   read: (namespace, threadId, sinceSeq) =>
     backingFor(namespace).events.read(namespace, threadId, sinceSeq),