KastraxStorage

KastraxStorage provides a unified interface for managing:

Suspended Workflows: the serialized state of suspended workflows (so they can be resumed later)
Memory: threads and messages per resourceId in your application
Traces: OpenTelemetry traces from all components of Kastrax
Eval Datasets: scores and scoring reasons from eval runs

Kastrax provides different storage providers, but you can treat them as interchangeable. Eg, you could use libsql in development but postgres in production, and your code will work the same both ways.

Configuration

Kastrax can be configured with a default storage option:


import ai.kastrax.core.kastrax
import ai.kastrax.storage.libsql.LibSQLStore
 
val kastraxInstance = kastrax {
  storage = LibSQLStore(
    url = "file:./kastrax.db"
  )
}

Data Schema

Messages

Stores conversation messages and their metadata. Each message belongs to a thread and contains the actual content along with metadata about the sender role and message type.

uuidv4

PRIMARYKEY

NOT NULL

Unique identifier for the message (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)

thread_id

uuidv4

FK → threads.id

NOT NULL

Parent thread reference

content

text

NOT NULL

role

text

NOT NULL

Enum of system | user | assistant | tool

type

text

NOT NULL

Enum of text | tool-call | tool-result

createdAt

timestamp

NOT NULL

Used for thread message ordering

Threads

Groups related messages together and associates them with a resource. Contains metadata about the conversation.

uuidv4

PRIMARYKEY

NOT NULL

Unique identifier for the thread (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)

resourceId

text

NOT NULL

Primary identifier of the external resource this thread is associated with. Used to group and retrieve related threads.

title

text

NOT NULL

Title of the conversation thread

metadata

text

Custom thread metadata as stringified JSON. Example:


{
  "category": "support",
  "priority": 1
}

createdAt

timestamp

NOT NULL

updatedAt

timestamp

NOT NULL

Used for thread ordering history

Workflows

When suspend is called on a workflow, its state is saved in the following format. When resume is called, that state is rehydrated.

workflow_name

text

NOT NULL

Name of the workflow

run_id

uuidv4

NOT NULL

Unique identifier for the workflow execution. Used to track state across suspend/resume cycles (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)

snapshot

text

NOT NULL

Serialized workflow state as JSON. Example:


{
  "value": {
    "currentState": "running"
  },
  "context": {
    "stepResults": {},
    "attempts": {},
    "triggerData": {}
  },
  "activePaths": [],
  "runId": "550e8400-e29b-41d4-a716-446655440000",
  "timestamp": 1648176000000
}

createdAt

timestamp

NOT NULL

updatedAt

timestamp

NOT NULL

Last modification time, used to track state changes during workflow execution

Eval Datasets

Stores eval results from running metrics against agent outputs.

input

text

NOT NULL

Input provided to the agent

output

text

NOT NULL

Output generated by the agent

result

jsonb

NOT NULL

Eval result data that includes score and details. Example:


{
  "score": 0.95,
  "details": {
    "reason": "Response accurately reflects source material",
    "citations": [
      "page 1",
      "page 3"
    ]
  }
}

agent_name

text

NOT NULL

metric_name

text

NOT NULL

e.g Faithfulness, Hallucination, etc.

instructions

text

NOT NULL

System prompt or instructions for the agent

test_info

jsonb

NOT NULL

Additional test metadata and configuration

global_run_id

uuidv4

NOT NULL

Groups related evaluation runs (e.g. all unit tests in a CI run)

run_id

uuidv4

NOT NULL

Unique identifier for the run being evaluated (format: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx)

created_at

timestamp

NOT NULL

Traces

Captures OpenTelemetry traces for monitoring and debugging.

text

NOT NULL

PRIMARYKEY

Unique trace identifier

parentSpanId

text

ID of the parent span. Null if span is top level

name

text

NOT NULL

Hierarchical operation name (e.g. workflow.myWorkflow.execute, http.request, database.query)

traceId

text

NOT NULL

Root trace identifier that groups related spans

scope

text

NOT NULL

Library/package/service that created the span (e.g. @kastrax/core, express, pg)

kind

integer

NOT NULL

INTERNAL (0, within process), CLIENT (1, outgoing calls), SERVER (2, incoming calls), PRODUCER (3, async job creation), CONSUMER (4, async job processing)

attributes

jsonb

User defined key-value pairs that contain span metadata

status

jsonb

JSON object with code (UNSET=0, ERROR=1, OK=2) and optional message. Example:


{
  "code": 1,
  "message": "HTTP request failed with status 500"
}

events

jsonb

Time-stamped events that occurred during the span

links

jsonb

Links to other related spans

other

text

Additional OpenTelemetry span fields as stringified JSON. Example:


{
  "droppedAttributesCount": 2,
  "droppedEventsCount": 1,
  "instrumentationLibrary": "@opentelemetry/instrumentation-http"
}

startTime

bigint

NOT NULL

Nanoseconds since Unix epoch when span started

endTime

bigint

NOT NULL

Nanoseconds since Unix epoch when span ended

createdAt

timestamp

NOT NULL

Storage Providers

Kastrax supports the following providers:

For local development, check out LibSQL Storage
For production, check out PostgreSQL Storage
For serverless deployments, check out Upstash Storage