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

## Summary

An alternative to #4529 that delivers the same session-comments feature,
but built on a **generic, extensible interface** instead of hardcoding
`comments` into the entity schema. Comments becomes one *custom
collection* among potentially many; the mechanism underneath is
reusable.

Three capabilities the agent runtime didn't have before:

1. **Opt-in externally-writable collections.** Entity state is
agent-owned. A collection becomes writable from the HTTP router only
when its definition sets `externallyWritable` — everything else stays
agent-only by default. (Every collection is always writable *by the
agent*; only an opt-in subset is writable *externally*, hence the name.)
2. **Authenticated, auditable writes.** Router writes require
authentication. The server stamps the authenticated principal into the
**change-event header** (provenance, outside the user payload), and the
client materializes it into a read-only **virtual column**
(`_principal`). A client can't spoof it via `value`.
3. **Schema-validated writes.** Each collection carries a schema; router
writes validate `value` server-side via the existing
`validateWriteEvent` before append.

## Design

**Runtime** (`@electric-ax/agents-runtime`)
- `CollectionDefinition.externallyWritable?: boolean`.
- At registration, writable collections are emitted as
`externally_writable_collections` (`{ type }` per collection).
- The entity stream DB materializes `headers.principal` into the fixed
`_principal` virtual column, mirroring `_timeline_order`, and strips it
(with `_seq`/`_timeline_order`) before client write-back.
- `createEntityTimelineQuery` accepts `customSources` —
consumer-provided sources unioned into the timeline query. The runtime
itself knows nothing about specific custom collections, and the agent's
LLM context never includes them.

**Server** (`@electric-ax/agents-server`)
- `externally_writable_collections` persisted as a jsonb column on
`entity_types` (migration 0016) and resolved via `getEffectiveSchemas`.
Legacy `principalColumn` in registration payloads is accepted and
ignored for version-skew tolerance.
- `EntityManager.writeCollection` gates on it (403 if not writable),
enforces entity-status/fork-lock, stamps `headers.principal = { url,
kind, id }`, validates `value`, and appends.
- `POST /:type/:instanceId/collections/:collection` exposes it (auth +
`write` permission, same middleware chain as `send`). Single POST,
`operation` in the body, 201 insert / 200 update·delete.

**Comments consumer** (`@electric-ax/agents` +
`@electric-ax/agents-server-ui`)
- `commentsCollection` (schema ported from #4529, minus
`from_principal`) declared as `state` on Horton and worker.
- The UI projects comments into the timeline by passing a comments
`customSources` entry to the timeline query (author resolved from
`_principal`); comments are a UI-level concern, hardcoded only in
`agents-server-ui`.
- The #4529 comments UI is cloned and re-sourced: writes via an
optimistic TanStack action backed by the authenticated
`/collections/comments` endpoint; reads `comment.from`.


## Test Plan

- [x] `@electric-ax/agents-runtime` — typecheck + 839 tests (incl.
principal virtual-column materialization, write-back stripping,
registration emit, timeline projection)
- [x] `@electric-ax/agents-server` — typecheck + suite (incl.
`writeCollection` 403/409 gating, principal-header stamping, schema
validation, `externally_writable_collections` jsonb round-trip)
- [x] `@electric-ax/agents` — typecheck + 60 tests (comments declared on
Horton/worker)
- [x] `@electric-ax/agents-server-ui` — typecheck + 88 tests (optimistic
comment write to `/collections/comments`, author from `_principal`)
- [ ] Manual: post a comment from the UI, confirm it syncs, right-aligns
for the author, and the principal is recorded in the change-event header

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: balegas

.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. External writes are restricted to a per-collection operations allowlist (insert-only by default), and comments are insert-only.

packages/agents-runtime/src/client.ts

@@ -10,6 +10,7 @@ export {
   getEntityState,
   normalizeEntityTimelineData,
   normalizeTimelineEntities,
+  TIMELINE_ORDER_FALLBACK,
 } from './entity-timeline'
 export {
   canonicalPgSyncOptions,
@@ -90,6 +91,7 @@ export type {
 export type {
   EntityTimelineContentItem,
   EntityTimelineData,
+  EntityTimelineCustomSource,
   EntityTimelineInboxMode,
   EntityTimelineQueryOptions,
   EntityTimelineQueryRow,
@@ -104,4 +106,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,89 @@
+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
+}
+
+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(),
+})
+
+/**
+ * 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,
+  // Insert-only: comments are append-only events. Update/delete would let a
+  // client overwrite or remove another principal's comment by key. Edit and
+  // soft-delete flows are deferred; their ops can be added here when they land.
+  operations: [`insert`],
+}

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,120 @@ 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; operations?: Array<string> }
+  > = {}
+  for (const [collectionName, def] of stateEntries) {
+    if (!def.externallyWritable) continue
+    externallyWritableCollections[collectionName] = {
+      type: def.type ?? `state:${collectionName}`,
+      ...(def.contract && { contract: def.contract }),
+      ...(def.operations && { operations: def.operations }),
+    }
+  }
+
+  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 +528,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)