Observations API

Observations let you extract structured facts from versioned documents with automatic staleness detection. For a comprehensive guide with examples, see the Observations Guide.

define_observation_type

function define_observation_type<T>(
  name: string,
  schema: SchemaLike<T>
): ObservationTypeDef<T>

Creates a typed observation definition for use with observations.put().

Parameters

Parameter	Type	Description
name	string	Unique identifier for this observation type
schema	SchemaLike<T>	Schema for validating observation content — a Zod 3 or Zod 4 schema (or any object with compatible parse/safeParse methods)

Example

import { z } from 'zod'
import { define_observation_type } from '@f0rbit/corpus'

const SentimentObservation = define_observation_type(
  'sentiment',
  z.object({
    subject: z.string(),
    score: z.number().min(-1).max(1),
    keywords: z.array(z.string()),
  })
)

---

ObservationsClient

The corpus.observations property provides methods for managing observations.

interface ObservationsClient {
  put<T>(type: ObservationTypeDef<T>, opts: ObservationPutOpts<T>): Promise<Result<Observation<T>, CorpusError>>
  get(id: string): Promise<Result<Observation, CorpusError>>
  query(opts?: ObservationQueryOpts): AsyncIterable<Observation>
  query_meta(opts?: ObservationQueryOpts): AsyncIterable<ObservationMeta>
  delete(id: string): Promise<Result<void, CorpusError>>
  delete_by_source(source: SnapshotPointer): Promise<Result<number, CorpusError>>
  is_stale(pointer: SnapshotPointer): Promise<boolean>
}

Methods

Method	Description
put(type, opts)	Create a new observation
get(id)	Get an observation by ID
query(opts?)	Query observations with filters
query_meta(opts?)	Query metadata only (more efficient)
delete(id)	Delete an observation by ID
delete_by_source(source)	Delete all observations from a source
is_stale(pointer)	Check if a source is stale

---

Types

SnapshotPointer

Universal address to versioned content within a corpus store.

type SnapshotPointer = {
  store_id: string       // Which store
  version: string        // Which version
  path?: string          // JSONPath to nested value (e.g., "$.items[0].text")
  span?: {               // Character range within the resolved value
    start: number
    end: number
  }
}

Observation

A stored observation record linking structured facts to versioned content.

type Observation<T = unknown> = {
  id: string                    // Auto-generated unique ID
  type: string                  // Type discriminator
  source: SnapshotPointer       // Where this came from
  content: T                    // The structured data
  confidence?: number           // 0-1 confidence score
  observed_at?: Date            // When the fact occurred (domain time)
  created_at: Date              // When the observation was created
  derived_from?: SnapshotPointer[]  // Optional provenance chain
}

ObservationMeta

Observation metadata without content payload. Used for efficient listing operations.

type ObservationMeta = Omit<Observation<never>, 'content'>

ObservationTypeDef

Definition for a typed observation schema.

type ObservationTypeDef<T> = {
  readonly name: string
  readonly schema: SchemaLike<T>
}

VersionFilter

Filter for selecting which versions to include in observation queries. Accepts three forms:

type VersionFilter =
  | Set<string>                                                      // Set-based lookup
  | string[]                                                         // Array-based lookup
  | ((store_id: string, version: string) => boolean | Promise<boolean>)  // Dynamic filter

Form	Description	Use Case
Set<string>	O(1) lookup for version membership	Known list of versions, performance-critical
string[]	Array includes check	Simple lists, convenient syntax
Function	Dynamic filtering with async support	Database lookups, external validation

Examples

Set-based filtering - fastest for large version lists:

const publishedVersions = new Set(['v1.0', 'v2.0', 'v2.1'])

for await (const obs of corpus.observations.query({
  source_store: 'documents',
  version_filter: publishedVersions
})) {
  // Only observations from v1.0, v2.0, or v2.1
}

Array-based filtering - convenient for small lists:

for await (const obs of corpus.observations.query({
  version_filter: ['draft', 'review', 'published']
})) {
  // Only observations from these workflow stages
}

Async function filtering - for dynamic version validation:

const version_filter: VersionFilter = async (store_id, version) => {
  const published = await db.query.published_reports.findFirst({
    where: and(
      eq(published_reports.store_id, store_id),
      eq(published_reports.version, version)
    )
  })
  return published !== null
}

for await (const obs of corpus.observations.query({ version_filter })) {
  // Only observations from versions that exist in published_reports table
}

Tip

Use Set<string> for best performance when filtering against a known list of versions. The async function form is useful when version validity depends on external state (e.g., database records, API calls).

ObservationQueryOpts

Query options for filtering observations.

type ObservationQueryOpts = {
  type?: string | string[]     // Filter by observation type(s)
  source_store?: string        // Filter by source store
  source_version?: string      // Filter by exact source version
  exclude_source_version?: string  // Drop observations from this source version
  source_prefix?: string       // Filter by source version prefix
  version_filter?: VersionFilter  // Filter by version (see above)
  after?: Date                 // observed_at after this date
  before?: Date                // observed_at before this date
  created_after?: Date         // created_at after this date
  created_before?: Date        // created_at before this date
  entity_key?: string          // Filter by entity key
  limit?: number               // Max results to return
  cursor?: string              // Pagination cursor
}

entity_key

Observation and ObservationMeta carry an optional entity_key. On put, when entity_key is omitted it is inherited from the source snapshot’s entity_key (observations about unkeyed snapshots stay unkeyed; a source snapshot that isn’t found stores no key and the put still succeeds; any other metadata error, such as a storage failure, fails the put). An explicit entity_key overrides inheritance. query({ entity_key }) returns only observations with that key.

is_stale(pointer) follows the source snapshot’s entity: if that snapshot has an entity_key, it is stale only when a newer snapshot with the same key exists in the store; unkeyed snapshots use the store-wide rule. See the Entities guide.

Current observations only

query() never filters by staleness. To keep only observations whose source is still current:

for await (const obs of observations.query({ type: 'entity_mention' })) {
  if (await observations.is_stale(obs.source)) continue
  // current
}

Unknown keys are rejected

query and query_meta reject keys that are not in ObservationQueryOpts. At compile time the options type is checked even when passed through a variable, so a typo like source_stor is a type error. At runtime (for example after a cast or from untyped input) the query yields nothing and the backend’s on_event handler receives an error event carrying a validation_error, rather than silently returning unfiltered results.

Results are ordered newest-first by created_at; observations with an identical created_at are returned in reverse insertion order (later-inserted first) on every backend.

Newest previous observation

Versions are opaque strings and are not reliably orderable, so there is no “before this version” filter. To find the newest observation for a source from any version other than the one you’re about to write, exclude it and take the first result:

for await (const prev of observations.query({
  type: 'entity_mention',
  source_store: 'hansard',
  exclude_source_version: new_version,
  limit: 1,
})) {
  // newest observation from a different version
}

Cursor-Based Pagination

The cursor field enables efficient pagination through large observation result sets without offset-based iteration:

// First page: no cursor
let cursor: string | undefined = undefined
const page_size = 50

for await (const obs of corpus.observations.query({
  type: 'entity_mention',
  limit: page_size,
  cursor: cursor
})) {
  console.log(obs.id, obs.content)
  cursor = obs.id // Use the last observation's ID as next cursor
}

// Next iteration with new cursor will continue from that point
for await (const obs of corpus.observations.query({
  type: 'entity_mention',
  limit: page_size,
  cursor: cursor  // Resume from last ID
})) {
  console.log(obs.id, obs.content)
  cursor = obs.id
}

The cursor is opaque (typically an observation ID from the previous result set). Pass it in the next query to resume pagination without re-scanning earlier results. This is more efficient than offset-based pagination for large datasets.

ObservationPutOpts

Options for creating a new observation.

type ObservationPutOpts<T> = {
  source: SnapshotPointer       // Required: what this observation is about
  content: T                    // Required: the typed observation data
  confidence?: number           // Optional: 0-1 confidence score
  observed_at?: Date            // Optional: when the fact occurred
  derived_from?: SnapshotPointer[]  // Optional: provenance chain
}

InferObservationContent

Utility type to extract the content type from an ObservationTypeDef.

type InferObservationContent<T> = T extends ObservationTypeDef<infer C> ? C : never

const entity_mention = define_observation_type('entity_mention', EntitySchema)

type EntityMention = InferObservationContent<typeof entity_mention>
// => { entity: string; entity_type: 'person' | 'organization' | ... }

---

Error Handling

Observation operations return Result<T, CorpusError> types. Observation-specific errors:

type ObservationError =
  | { kind: 'validation_error'; message: string }
  | { kind: 'observation_not_found'; id: string }
  | { kind: 'storage_error'; cause: Error; operation: string }

Example

const result = await corpus.observations.put(SentimentObservation, {
  source: pointer,
  content: { subject: 'test', score: 2.0, keywords: [] }, // Invalid: score > 1
})

if (!result.ok) {
  switch (result.error.kind) {
    case 'validation_error':
      console.log('Invalid content:', result.error.message)
      break
    case 'observation_not_found':
      console.log('Observation not found:', result.error.id)
      break
    case 'storage_error':
      console.log('Backend error:', result.error.operation)
      break
  }
}

---

See Also

• Observations Guide - Full guide with examples and use cases
• Core Types - Result, CorpusError
• Testing Patterns - Test observations with memory backend