Versions

• Client: plain HTTP (durable-streams appends to an entity's main stream; agents-runtime produces them with @durable-streams/client 0.2.6 IdempotentProducer)
• Electric: @electric-ax/agents-server 0.6.4 (main at 140a5f4e7)

Bug description

The Agents Server's stream-append route validates a typed append one event at a time, and each validation re-reads the entity type from Postgres. An append of N events therefore costs 1 + N sequential registry round trips before the request is forwarded to the durable-streams server, and all N type reads return the same row.

Where it happens (packages/agents-server, main at 140a5f4e7):

• src/routing/stream-append.ts:92 resolves the entity once: const entity = await manager.registry.getEntityByStream(path) (one SELECT on entities, src/entity-registry.ts:886-895).

• src/routing/stream-append.ts:120-135 then loops over the events:

  if (event) {
    const events = Array.isArray(event) ? event : [event]
    for (const eventItem of events) {
      const validationError = await manager.validateWriteEvent(
        entity,
        eventItem
      )
      if (validationError) {
        return apiError(
          validationError.status,
          validationError.code,
          validationError.message
        )
      }
    }
  }

• EntityManager.validateWriteEvent (src/entity-manager.ts:3710-3756) calls this.getEffectiveSchemas(entity) (:3716) for every event. For a typed entity that method does const latestType = await this.registry.getEntityType(entity.type) (:4035), a SELECT ... FROM entity_types ... LIMIT 1 (src/entity-registry.ts:712-720). Nothing caches the result between events.

Registry round trips per append of a typed entity:

Why it matters: with Postgres on the same host the extra reads are sub-millisecond and invisible. With Postgres remote they dominate the append path. For an Agents Server whose Postgres is ~60 ms away, a 10-event append spends about 11 x 60 ms = 660 ms in the route before it is forwarded, against about 120 ms for the two reads that are actually needed. Appends of that size are the normal case: agents-runtime batches state events with the client's 5 ms linger and pipelines up to 5 batches (@durable-streams/client 0.2.6 defaults), so a busy handler sends a stream of batches of around ten events, each one paying the per-event reads.

A smaller correctness consequence: because the type is re-read per event, the events of one append can be validated against different revisions of the type if it is re-registered while the batch is being checked.

Reproduction, as an executable test case (no Docker needed). It drives the real electricAgentsStreamAppendRouter with a real EntityManager over a mocked registry and expects registry.getEntityType to be called once for a 10-event append:

// packages/agents-server/test/stream-append-route.test.ts (abridged)
const registry = {
  getEntityByStream: vi.fn().mockResolvedValue({
    url: `/coder/session-1`,
    type: `coder`,
    status: `running`,
    write_token: `token-1`,
    streams: { main: `/coder/session-1/main` },
  }),
  getEntityType: vi.fn().mockResolvedValue({
    name: `coder`,
    state_schemas: { observed_item: observedItemSchema },
  }),
  close: vi.fn(),
}
const runtime: ElectricAgentsStreamAppendRuntime = {
  manager: new EntityManager({
    registry: registry as any,
    streamClient: {} as any,
    validator: new SchemaValidator(),
    wakeRegistry: {
      setTimeoutCallback: vi.fn(),
      setDebounceCallback: vi.fn(),
    } as any,
  }),
  evaluateWakePayload: vi.fn(async () => {}),
  checkRunFinished: vi.fn(),
  syncManifestWakes: vi.fn(async () => {}),
  syncManifestEntitySources: vi.fn(async () => {}),
  syncManifestSchedules: vi.fn(async () => {}),
}
const forward = vi.fn(async () => new Response(null, { status: 200 }))
const events = Array.from({ length: 10 }, (_, i) => ({
  type: `observed_item`,
  key: `item-${i}`,
  headers: { operation: `insert` },
  value: { key: `item-${i}`, value: `value-${i}` },
}))

const response = await electricAgentsStreamAppendRouter.fetch(
  createStreamAppendRouteRequest(
    new Request(`http://agents.test/coder/session-1/main`, {
      method: `POST`,
      headers: {
        authorization: `Bearer token-1`,
        'content-type': `application/json`,
      },
      body: JSON.stringify(events),
    })
  ),
  runtime,
  forward
)

expect(response?.status).toBe(200)
expect(forward).toHaveBeenCalledTimes(1)
expect(registry.getEntityByStream).toHaveBeenCalledTimes(1)
expect(registry.getEntityType).toHaveBeenCalledTimes(1) // on main: called 10 times

The full file, with four more cases covering error order, the untyped-entity path and the single-snapshot check, is at packages/agents-server/test/stream-append-route.test.ts on my fork. On main with only the test file added:

cd packages/agents-server && pnpm exec vitest run test/stream-append-route.test.ts
 × resolves the entity type once for a multi-event append
 ✓ rejects the first invalid event in order without forwarding
 ✓ reports a schema failure at its position in the batch
 ✓ skips validation for an untyped entity
 × validates every event of an append against one schema snapshot

 FAIL  test/stream-append-route.test.ts > stream append route > resolves the entity type once for a multi-event append
AssertionError: expected "vi.fn()" to be called 1 times, but got 10 times
 FAIL  test/stream-append-route.test.ts > stream append route > validates every event of an append against one schema snapshot
AssertionError: expected 422 to be 200 // Object.is equality

 Test Files  1 failed (1)
      Tests  2 failed | 3 passed (5)

The same can be seen against a real stack: set log_statement = 'all' on the Agents Server's Postgres, register an entity type with state_schemas, spawn an entity, and POST a JSON array of 10 events to /{type}/{id}/main with the entity's write token. The log shows one SELECT on entities followed by ten identical SELECTs on entity_types before the append reaches the durable-streams server.

Minimal fix: resolve the effective schemas once per append and validate every event in memory against that snapshot. EntityManager gets validateWriteEvents(entity, events), which does the single lookup and returns the first invalid event's error in order; validateWriteEvent(entity, event) becomes return this.validateWriteEvents(entity, [event]) so its other caller (writeCollection) is unchanged; and the route replaces the loop with one call. No SQL, registry API or error code changes. The change is on the same fork branch, fix/agents-server-stream-append-schema-lookup; PR to follow.

An alternative that was considered and not taken: a joined entities LEFT JOIN entity_types registry query would get an append to a single round trip, but it needs a new registry method that duplicates getEntityByStream's /main parsing and getEffectiveSchemas' merge, a larger review surface for one more saved read. It could be a follow-up on top of this change.

Related, out of scope here: writeCollection (src/entity-manager.ts:2466-2467 and :2530) calls getEffectiveSchemas and then validateWriteEvent, so a single collection write reads the type twice.

Expected behavior

An append of N events to a typed entity resolves the entity's schemas once and validates all N events against that one snapshot: two registry round trips regardless of N, with unchanged per-event checks and error codes. The first invalid event in order is still reported with the same error (422 UNKNOWN_EVENT_TYPE or SCHEMA_VALIDATION_FAILED) and the append is not forwarded; untyped entities still skip validation without a type lookup; a missing type row still falls back to the entity's own schemas.