Registry

The Registry is the single source of truth for everything in the system. Every other primitive queries it for discovery instead of maintaining its own catalog.

Protocol

Python

class Registry(Protocol):
    async def register(self, entry: RegistryEntry, ctx: ExecContext) -> None: ...
    async def discover(self, kind: ComponentKind, ctx: ExecContext) -> list[RegistryEntry]: ...
    async def resolve(self, name: str, ctx: ExecContext) -> RegistryEntry | None: ...
    async def health(self, name: str) -> HealthStatus: ...
    async def update(self, name: str, patch: RegistryPatch) -> None: ...

TypeScript

export interface Registry {
  register(entry: RegistryEntry, ctx: ExecContext): Promise<void>;
  discover(kind: ComponentKind, ctx: ExecContext): Promise<RegistryEntry[]>;
  resolve(name: string, ctx: ExecContext): Promise<RegistryEntry | null>;
  health(name: string): Promise<HealthStatus>;
  update(name: string, patch: RegistryPatch): Promise<void>;
}

Go

type Registry interface {
    Register(ctx *nctx.ExecContext, entry RegistryEntry) error
    Discover(ctx *nctx.ExecContext, kind ComponentKind) ([]RegistryEntry, error)
    Resolve(ctx *nctx.ExecContext, name string) (*RegistryEntry, error)
    Health(name string) (HealthStatus, error)
    Update(name string, patch RegistryPatch) error
}

Value types

Python

class ComponentKind(StrEnum):
    AGENT = "agent"
    TOOL = "tool"
    SENSE = "sense"
    PLUGIN = "plugin"

class HealthStatus(StrEnum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNAVAILABLE = "unavailable"

@dataclass
class RegistryEntry:
    name: str
    kind: ComponentKind
    description: str
    schema: dict[str, object] | None = None
    metadata: dict[str, str] = field(default_factory=dict)
    health: HealthStatus = HealthStatus.HEALTHY
    stats: InvocationStats = field(default_factory=InvocationStats)
    enabled: bool = True
    requirements: list[str] = field(default_factory=list)
    permissions: list[str] = field(default_factory=list)

@dataclass
class InvocationStats:
    total_calls: int = 0
    successes: int = 0
    failures: int = 0
    last_invoked_at: float | None = None
    avg_duration_ms: float = 0.0   # exponential moving average

TypeScript

export enum ComponentKind {
  AGENT = "agent",
  TOOL = "tool",
  SENSE = "sense",
  PLUGIN = "plugin",
}

export enum HealthStatus {
  HEALTHY = "healthy",
  DEGRADED = "degraded",
  UNAVAILABLE = "unavailable",
}

export interface RegistryEntry {
  readonly name: string;
  readonly kind: ComponentKind;
  description: string;
  readonly schema: Record<string, unknown> | null;
  metadata: Record<string, string>;
  health: HealthStatus;
  readonly stats: InvocationStats;
  enabled: boolean;
  requirements: string[];
  permissions: string[];
}

export class InvocationStats {
  totalCalls: number = 0;
  successes: number = 0;
  failures: number = 0;
  lastInvokedAt: number | null = null;
  avgDurationMs: number = 0;

  recordSuccess(durationMs: number): void { /* ... */ }
  recordFailure(durationMs: number): void { /* ... */ }
}

Go

type ComponentKind string

const (
    KindAgent  ComponentKind = "agent"
    KindTool   ComponentKind = "tool"
    KindSense  ComponentKind = "sense"
    KindPlugin ComponentKind = "plugin"
)

type HealthStatus string

const (
    HealthHealthy     HealthStatus = "healthy"
    HealthDegraded    HealthStatus = "degraded"
    HealthUnavailable HealthStatus = "unavailable"
)

type RegistryEntry struct {
    Name         string
    Kind         ComponentKind
    Description  string
    Schema       map[string]any
    Metadata     map[string]string
    Health       HealthStatus
    Stats        InvocationStats
    Enabled      bool
    Requirements []string
    Permissions  []string
}

type InvocationStats struct {
    TotalCalls    int
    Successes     int
    Failures      int
    LastInvokedAt *time.Time
    AvgDurationMs float64
}

func (s *InvocationStats) RecordSuccess(durationMs float64) { /* ... */ }
func (s *InvocationStats) RecordFailure(durationMs float64) { /* ... */ }

Strategies

InMemoryRegistry

Tip

InMemoryRegistry is perfect for testing and development. No database setup required — everything lives in a dictionary. Switch to SqliteRegistry when you need persistence across restarts.

Dictionary-backed. No persistence. Suitable for testing and single-process deployments.

Python

from nerva.registry.inmemory import InMemoryRegistry

registry = InMemoryRegistry()

await registry.register(RegistryEntry(
    name="weather_agent",
    kind=ComponentKind.AGENT,
    description="Answers weather and forecast questions",
    permissions=["user"],
), ctx)

# Discover all agents visible to the caller
agents = await registry.discover(ComponentKind.AGENT, ctx)

# Resolve a specific component
entry = await registry.resolve("weather_agent", ctx)

TypeScript

import { InMemoryRegistry } from "nerva/registry/inmemory";
import { createRegistryEntry, ComponentKind } from "nerva/registry";

const registry = new InMemoryRegistry();

await registry.register(
  createRegistryEntry("weather_agent", ComponentKind.AGENT, "Answers weather and forecast questions", {
    permissions: ["user"],
  }),
  ctx,
);

// Discover all agents visible to the caller
const agents = await registry.discover(ComponentKind.AGENT, ctx);

// Resolve a specific component
const entry = await registry.resolve("weather_agent", ctx);

Go

import "github.com/otomus/nerva/go/registry"

reg := registry.NewInMemoryRegistry()

err := reg.Register(ctx, registry.RegistryEntry{
    Name:        "weather_agent",
    Kind:        registry.KindAgent,
    Description: "Answers weather and forecast questions",
    Enabled:     true,
    Health:      registry.HealthHealthy,
    Permissions: []string{"user"},
    Metadata:    make(map[string]string),
})

// Discover all agents visible to the caller
agents, err := reg.Discover(ctx, registry.KindAgent)

// Resolve a specific component
entry, err := reg.Resolve(ctx, "weather_agent")

SqliteRegistry

Disk-backed persistence with WAL mode. Survives restarts, single-node only.

Python

from nerva.registry.sqlite import SqliteRegistry

registry = SqliteRegistry(db_path="./data/registry.db")

TypeScript

import { SqliteRegistry } from "nerva/registry/sqlite";
import Database from "better-sqlite3";

const registry = new SqliteRegistry(Database, "./data/registry.db");

Go

import "github.com/otomus/nerva/go/registry"

reg, err := registry.NewSqliteRegistry("./data/registry.db")
if err != nil {
    log.Fatal(err)
}
defer reg.Close()

Health tracking

The registry tracks health per component. The Runtime updates health based on circuit breaker state:

Python

# Mark a handler as degraded after repeated failures
await registry.update("weather_agent", RegistryPatch(health=HealthStatus.DEGRADED))

# Check health before routing
status = await registry.health("weather_agent")
if status == HealthStatus.UNAVAILABLE:
    # Skip this handler
    ...

TypeScript

import { HealthStatus } from "nerva/registry";

// Mark a handler as degraded after repeated failures
await registry.update("weather_agent", { health: HealthStatus.DEGRADED });

// Check health before routing
const status = await registry.health("weather_agent");
if (status === HealthStatus.UNAVAILABLE) {
  // Skip this handler
}

Go

// Mark a handler as degraded after repeated failures
degraded := registry.HealthDegraded
err := reg.Update("weather_agent", registry.RegistryPatch{Health: &degraded})

// Check health before routing
status, err := reg.Health("weather_agent")
if status == registry.HealthUnavailable {
    // Skip this handler
}

discover() automatically filters out disabled and unavailable components.

Invocation stats

Stats are tracked using an exponential moving average for latency:

Python

entry = await registry.resolve("weather_agent", ctx)
print(entry.stats.total_calls)     # 1,234
print(entry.stats.successes)       # 1,200
print(entry.stats.failures)        # 34
print(entry.stats.avg_duration_ms) # 156.3

TypeScript

const entry = await registry.resolve("weather_agent", ctx);
console.log(entry.stats.totalCalls);    // 1234
console.log(entry.stats.successes);     // 1200
console.log(entry.stats.failures);      // 34
console.log(entry.stats.avgDurationMs); // 156.3

Go

entry, err := reg.Resolve(ctx, "weather_agent")
fmt.Println(entry.Stats.TotalCalls)    // 1234
fmt.Println(entry.Stats.Successes)     // 1200
fmt.Println(entry.Stats.Failures)      // 34
fmt.Println(entry.Stats.AvgDurationMs) // 156.3

New observations are blended with a smoothing factor of 0.2, so recent latency is weighted more heavily than historical.

Note

Health checking and invocation stats are automatic. Every invoke() call updates the registry entry’s stats, and the circuit breaker updates health status. You get observability for free just by using the Registry.

Partial updates

Use RegistryPatch to update specific fields without replacing the entire entry:

Python

from nerva.registry import RegistryPatch, HealthStatus

await registry.update("weather_agent", RegistryPatch(
    description="Updated weather agent with radar support",
    health=HealthStatus.HEALTHY,
    enabled=True,
))
# Only the specified fields are changed; stats and metadata are preserved

TypeScript

import { HealthStatus } from "nerva/registry";

await registry.update("weather_agent", {
  description: "Updated weather agent with radar support",
  health: HealthStatus.HEALTHY,
  enabled: true,
});
// Only the specified fields are changed; stats and metadata are preserved

Go

healthy := registry.HealthHealthy
enabled := true
desc := "Updated weather agent with radar support"

err := reg.Update("weather_agent", registry.RegistryPatch{
    Description: &desc,
    Health:      &healthy,
    Enabled:     &enabled,
})
// Only the specified fields are changed; stats and metadata are preserved