feat(agents): generic externally-writable custom collections (comments as first consumer)

.changeset/generic-externally-writable-collections.md

@@ -0,0 +1,8 @@
+---
+'@electric-ax/agents-runtime': patch
+'@electric-ax/agents-server': patch
+'@electric-ax/agents-server-ui': patch
+'@electric-ax/agents': patch
+---
+
+Add generic externally-writable custom collections for agent entity state: collections opt in via `externallyWritable`, writes go through an authenticated schema-validated endpoint that stamps the principal into a read-only `_principal` column, and `createEntityTimelineQuery` can project them into the timeline via `customSources`. Comments are reimplemented as one such collection, gated per agent type through a reserved `comments/v1` contract that the UI keys its comment affordances on.

packages/agents-runtime/src/client.ts

@@ -10,6 +10,7 @@ export {
   getEntityState,
   normalizeEntityTimelineData,
   normalizeTimelineEntities,
+  TIMELINE_ORDER_FALLBACK,
 } from './entity-timeline'
 export {
   canonicalPgSyncOptions,
@@ -82,6 +83,7 @@ export type {
 export type {
   EntityTimelineContentItem,
   EntityTimelineData,
+  EntityTimelineCustomSource,
   EntityTimelineInboxMode,
   EntityTimelineQueryOptions,
   EntityTimelineQueryRow,
@@ -96,4 +98,9 @@ export type {
   IncludesInboxMessage,
   IncludesRun,
 } from './entity-timeline'
+export { COMMENTS_CONTRACT, commentsCollection } from './comments-collection'
+export type {
+  CommentSnapshotValue as CommentSnapshot,
+  CommentTargetValue as CommentTarget,
+} from './comments-collection'
 export type { EntityTimelineEntry } from './use-chat'

packages/agents-runtime/src/comments-collection.ts

@@ -0,0 +1,91 @@
+import { z } from 'zod'
+import type { CollectionDefinition } from './types'
+
+export type CommentTargetValue =
+  | { kind: `comment`; key: string }
+  | {
+      kind: `timeline`
+      collection:
+        | `inbox`
+        | `run`
+        | `text`
+        | `tool_call`
+        | `wake`
+        | `signal`
+        | `manifest`
+      key: string
+      run_id?: string
+    }
+
+export type CommentSnapshotValue = {
+  label: string
+  text?: string
+  from?: string
+  timestamp?: string
+  collection?: string
+}
+
+export type CommentValue = {
+  key?: string
+  body: string
+  timestamp: string
+  reply_to?: CommentTargetValue
+  target_snapshot?: CommentSnapshotValue
+  edited_at?: string
+  deleted_at?: string
+  deleted_by?: string
+}
+
+const commentTargetSchema = z.union([
+  z.object({ kind: z.literal(`comment`), key: z.string() }),
+  z.object({
+    kind: z.literal(`timeline`),
+    collection: z.enum([
+      `inbox`,
+      `run`,
+      `text`,
+      `tool_call`,
+      `wake`,
+      `signal`,
+      `manifest`,
+    ]),
+    key: z.string(),
+    run_id: z.string().optional(),
+  }),
+])
+
+const commentSnapshotSchema = z.object({
+  label: z.string(),
+  text: z.string().optional(),
+  from: z.string().optional(),
+  timestamp: z.string().optional(),
+  collection: z.string().optional(),
+})
+
+export const commentSchema = z.object({
+  key: z.string().optional(),
+  body: z.string(),
+  timestamp: z.string(),
+  reply_to: commentTargetSchema.optional(),
+  target_snapshot: commentSnapshotSchema.optional(),
+  edited_at: z.string().optional(),
+  deleted_at: z.string().optional(),
+  deleted_by: z.string().optional(),
+})
+
+/**
+ * Contract identifier for the canonical comments collection. The server
+ * reserves the `comments` collection name for this contract, and the UI
+ * only surfaces comment affordances for entity types whose registration
+ * advertises it — so an agent's unrelated `comments` state can never be
+ * mistaken for platform comments.
+ */
+export const COMMENTS_CONTRACT = `comments/v1`
+
+export const commentsCollection: CollectionDefinition = {
+  schema: commentSchema,
+  type: `state:comments`,
+  primaryKey: `key`,
+  externallyWritable: true,
+  contract: COMMENTS_CONTRACT,
+}

packages/agents-runtime/src/create-handler.ts

@@ -17,6 +17,7 @@ import type { IncomingMessage, ServerResponse } from 'node:http'
 import type { WebhookSignatureVerifierConfig } from './webhook-signature'
 import type {
   AgentTool,
+  AnyEntityDefinition,
   EntityStreamDBWithActions,
   HeadersProvider,
   ProcessWakeConfig,
@@ -207,6 +208,119 @@ export interface RuntimeDebugState {
 export type RuntimeHandlerConfig = RuntimeRouterConfig
 export type RuntimeHandlerResult = RuntimeHandler
 
+const JSON_SCHEMA_KEYWORDS = [
+  `type`,
+  `properties`,
+  `items`,
+  `enum`,
+  `oneOf`,
+  `anyOf`,
+  `allOf`,
+  `additionalProperties`,
+] as const
+
+function stripSchemaKeyword(
+  jsonSchema: Record<string, unknown>
+): Record<string, unknown> {
+  const { $schema: _schema, ...rest } = jsonSchema
+  return rest
+}
+
+function toJsonSchema(schema: unknown): Record<string, unknown> {
+  if (!schema || typeof schema !== `object` || Array.isArray(schema)) {
+    return {}
+  }
+
+  const standardSchema = schema as {
+    [`~standard`]?: {
+      jsonSchema?: {
+        input?: () => unknown
+      }
+    }
+    toJSONSchema?: () => Record<string, unknown>
+  }
+
+  const standardJsonSchema = standardSchema[`~standard`]?.jsonSchema?.input?.()
+  if (standardJsonSchema) {
+    return stripSchemaKeyword(standardJsonSchema as Record<string, unknown>)
+  }
+
+  if (typeof standardSchema.toJSONSchema === `function`) {
+    return stripSchemaKeyword(standardSchema.toJSONSchema())
+  }
+
+  if (`~standard` in standardSchema) {
+    return {}
+  }
+
+  const jsonSchemaLike = schema as Record<string, unknown>
+  if (JSON_SCHEMA_KEYWORDS.some((keyword) => keyword in jsonSchemaLike)) {
+    return stripSchemaKeyword(jsonSchemaLike)
+  }
+
+  return zodToJsonSchema(schema as any, { target: `jsonSchema7` })
+}
+
+function mapSchemas(
+  schemas: Record<string, unknown>
+): Record<string, Record<string, unknown>> {
+  return Object.fromEntries(
+    Object.entries(schemas).map(([k, v]) => [k, toJsonSchema(v)])
+  )
+}
+
+export function buildEntityTypeRegistrationBody(
+  name: string,
+  definition: AnyEntityDefinition
+): Record<string, unknown> {
+  const stateEntries = definition.state ? Object.entries(definition.state) : []
+
+  const stateSchemas = Object.fromEntries(
+    stateEntries.map(([collectionName, def]) => [
+      def.type ?? `state:${collectionName}`,
+      toJsonSchema(def.schema ?? passthrough()),
+    ])
+  )
+
+  const externallyWritableCollections: Record<
+    string,
+    { type: string; contract?: string }
+  > = {}
+  for (const [collectionName, def] of stateEntries) {
+    if (!def.externallyWritable) continue
+    externallyWritableCollections[collectionName] = {
+      type: def.type ?? `state:${collectionName}`,
+      ...(def.contract && { contract: def.contract }),
+    }
+  }
+
+  const body: Record<string, unknown> = {
+    name,
+    description: definition.description ?? `${name} entity`,
+    ...(definition.creationSchema && {
+      creation_schema: toJsonSchema(definition.creationSchema),
+    }),
+    ...(definition.inboxSchemas && {
+      inbox_schemas: mapSchemas(definition.inboxSchemas),
+    }),
+    ...(definition.slashCommands && {
+      slash_commands: definition.slashCommands,
+    }),
+    state_schemas: {
+      ...DEFAULT_STATE_SCHEMAS,
+      ...stateSchemas,
+      ...(definition.stateSchemas ? mapSchemas(definition.stateSchemas) : {}),
+    },
+    ...(Object.keys(externallyWritableCollections).length > 0 && {
+      externally_writable_collections: externallyWritableCollections,
+    }),
+    ...(definition.permissionGrants && {
+      permission_grants: definition.permissionGrants,
+    }),
+  }
+  return body
+}
+
 export function createRuntimeRouter(
   config: RuntimeRouterConfig
 ): RuntimeRouter {
@@ -413,110 +527,18 @@ export function createRuntimeRouter(
     return handleWebhookRequest(request)
   }
 
-  const stripSchemaKeyword = (
-    jsonSchema: Record<string, unknown>
-  ): Record<string, unknown> => {
-    const { $schema: _schema, ...rest } = jsonSchema
-    return rest
-  }
-
-  const JSON_SCHEMA_KEYWORDS = [
-    `type`,
-    `properties`,
-    `items`,
-    `enum`,
-    `oneOf`,
-    `anyOf`,
-    `allOf`,
-    `additionalProperties`,
-  ] as const
-
-  const toJsonSchema = (schema: unknown): Record<string, unknown> => {
-    if (!schema || typeof schema !== `object` || Array.isArray(schema)) {
-      return {}
-    }
-
-    const standardSchema = schema as {
-      [`~standard`]?: {
-        jsonSchema?: {
-          input?: () => unknown
-        }
-      }
-      toJSONSchema?: () => Record<string, unknown>
-    }
-
-    const standardJsonSchema =
-      standardSchema[`~standard`]?.jsonSchema?.input?.()
-    if (standardJsonSchema) {
-      return stripSchemaKeyword(standardJsonSchema as Record<string, unknown>)
-    }
-
-    if (typeof standardSchema.toJSONSchema === `function`) {
-      return stripSchemaKeyword(standardSchema.toJSONSchema())
-    }
-
-    if (`~standard` in standardSchema) {
-      return {}
-    }
-
-    const jsonSchemaLike = schema as Record<string, unknown>
-    if (JSON_SCHEMA_KEYWORDS.some((keyword) => keyword in jsonSchemaLike)) {
-      return stripSchemaKeyword(jsonSchemaLike)
-    }
-
-    return zodToJsonSchema(schema as any, { target: `jsonSchema7` })
-  }
-
   const registerTypes = async (): Promise<void> => {
     const types = getRegisteredTypes()
     const registered: Array<string> = []
     const failed: Array<string> = []
     const totalStart = performance.now()
     const effectiveConcurrency = Math.max(1, registrationConcurrency ?? 8)
 
-    const mapSchemas = (
-      schemas: Record<string, unknown>
-    ): Record<string, Record<string, unknown>> =>
-      Object.fromEntries(
-        Object.entries(schemas).map(([k, v]) => [k, toJsonSchema(v)])
-      )
-
     await forEachWithConcurrency(types, effectiveConcurrency, async (entry) => {
       const registrationStart = performance.now()
       const { name, definition } = entry
 
-      const stateSchemas = definition.state
-        ? Object.fromEntries(
-            Object.entries(definition.state).map(([collectionName, def]) => [
-              def.type ?? `state:${collectionName}`,
-              toJsonSchema(def.schema ?? passthrough()),
-            ])
-          )
-        : {}
-
-      const body: Record<string, unknown> = {
-        name,
-        description: definition.description ?? `${name} entity`,
-        ...(definition.creationSchema && {
-          creation_schema: toJsonSchema(definition.creationSchema),
-        }),
-        ...(definition.inboxSchemas && {
-          inbox_schemas: mapSchemas(definition.inboxSchemas),
-        }),
-        ...(definition.slashCommands && {
-          slash_commands: definition.slashCommands,
-        }),
-        state_schemas: {
-          ...DEFAULT_STATE_SCHEMAS,
-          ...stateSchemas,
-          ...(definition.stateSchemas
-            ? mapSchemas(definition.stateSchemas)
-            : {}),
-        },
-        ...(definition.permissionGrants && {
-          permission_grants: definition.permissionGrants,
-        }),
-      }
+      const body = buildEntityTypeRegistrationBody(name, definition)
 
       const defaultDispatchPolicy = defaultDispatchPolicyForType?.(name)