stream

Documentation

Overview

Package stream is Recall's reference external adapter: an append-only JSONL event source, spoken over newline-delimited JSON-RPC 2.0 on stdio.

It exists to be copied. Everything an adapter author has to decide once — how to negotiate a version, where an index may be written, what as_of support is honest, how a locator survives a source change, how cancellation is observed — is decided here in one small package, and every decision is recorded in cmd/recall-stream/conformance as a replayable transcript.

Boundary

The adapter owns parsing, its projection, ranking within this source, and locator semantics. It owns no identity: source_uid, the source prior, and the sensitivity floor come from configuration, and the core overwrites the source part of every locator returned here. It writes only inside the workdir supplied at handshake, and only one file: cursor.json.

Lineage

This is the only adapter in the tree that emits derived_from. A stream record that normalizes an upstream system's event carries that system's name and the upstream record's native ref; the settings block maps a system name to a configured source_id, and the pair becomes a display-form locator naming the upstream record. A signal about task td-f62256 therefore declares "tasks:td-f62256", which is character-for-character the locator the Tasks adapter writes for that task — so the two collapse into one lineage root and never corroborate each other. An unmapped system emits no edge at all: guessing a source_id would produce an edge that resolves somewhere, and a wrong lineage root is worse than a missing one.

Freshness and the checkpoint

The projection is memory-resident and brought up to date incrementally by byte offset: an append-only file only ever grows, so catching up costs the bytes appended since the last pass and nothing more. A file that shrank was rewritten, which an append-only stream must not do, so the scan falls back to a full rebuild and health says so.

cursor.json records the last successful boundary — generation, per-file offsets, record and failure counts. It is not a resume point for records: this adapter's index lives in memory and is rebuilt on start. It is what makes generations monotonic across restarts and what a fresh process compares against to notice that the stream was rewritten while it was down. An adapter with a durable index would write its records first and this file second, which is the ordering the spec requires and the reason it is written only after a generation is published.

as_of

recall.AsOfFilter, and deliberately not snapshot. Every record carries an immutable event_time, so restricting to events that happened at or before a boundary is a filter over history the source already stores — the thing the Tasks adapter could not do, because plan dates are not record history.

Snapshot would be a lie. A record describing an early event can be appended at any later time, so the set of records present in the file at a past instant is not the set an event-time filter selects, and this format publishes no append time to reconstruct it from.

Index

• Constants
• Variables
• type Adapter
  • func New(opts Options) *Adapter
  • func (a *Adapter) Close() error
  • func (a *Adapter) Expand(_ context.Context, req recall.ExpandRequest) (recall.ExpandResponse, error)
  • func (a *Adapter) Health(context.Context) (recall.Health, error)
  • func (a *Adapter) Initialize(_ context.Context, cfg adapter.Config) (recall.Manifest, error)
  • func (a *Adapter) Refresh(_ context.Context, p protocol.RefreshParams) (recall.Health, error)
  • func (a *Adapter) Search(ctx context.Context, req recall.SearchRequest) (recall.SearchResponse, error)
• type Options
• type Settings

Constants

const (
	// AdapterID is this implementation's identity in manifests and reports.
	AdapterID = "recall-stream/1"

	// DisplayName is the adapter's name. The instance's name is its
	// configured source_id and arrives at the handshake.
	DisplayName = "JSONL Stream"

	// IndexConfig identifies the retrieval configuration a generation was
	// built under. It changes whenever tokenization or scoring changes, so an
	// evaluation comparing two generations cannot mistake a scoring change for
	// the change under test.
	IndexConfig = "jsonl/1 tokenizer=ident-runs scoring=term-coverage"

	// DefaultMaxCandidates caps one search's candidate list so a long stream
	// cannot flood the fusion pool. The core's per-source limit narrows it
	// further.
	DefaultMaxCandidates = 50
)

Adapter identity and defaults.

const (
	MinSchema = 1
	MaxSchema = 2
)

Schema versions this build can read. A record outside the range is counted as failed rather than guessed at: a shape nobody has written a parser for is not evidence, and silently dropping it would make an incomplete index look complete.

Variables

var ErrNotInitialized = protocol.Errorf(protocol.CodeSourceUnavailable,
	"stream adapter has not completed a handshake")

ErrNotInitialized reports use before a successful handshake. It carries the source_unavailable code, so errors.Is against the sentinel still matches, while keeping a message the wire can show.

Types

type Adapter

type Adapter struct {
	// contains filtered or unexported fields
}

Adapter is the JSONL stream source.

The zero value is not usable; build one with New. It holds a published index generation and the cursors that produced it.

func New

func New(opts Options) *Adapter

New returns an uninitialized adapter.

func (*Adapter) Close

func (a *Adapter) Close() error

Close releases the adapter. Later calls must fail rather than answer from a projection nobody is maintaining any more.

func (*Adapter) Expand

func (a *Adapter) Expand(_ context.Context, req recall.ExpandRequest) (recall.ExpandResponse, error)

Expand retrieves the evidence behind a locator.

The local part is "v<schema>/<id>". The schema version is part of the reference because it is part of what the reference promised: a record rewritten into another version is not the same evidence, and returning it would be the "different revision" the protocol forbids expansion from silently substituting.

func (*Adapter) Health

func (a *Adapter) Health(context.Context) (recall.Health, error)

Health probes the source. A failed probe still reports the generation still published, because that is the one still answering.

func (*Adapter) Initialize

func (a *Adapter) Initialize(_ context.Context, cfg adapter.Config) (recall.Manifest, error)

Initialize negotiates the protocol version, validates the settings block, resolves the stream files, and adopts the workdir's checkpoint. It reads no stream bytes: building an index inside the handshake competes with the handshake timeout, which is why recall/refresh exists.

func (*Adapter) Refresh

func (a *Adapter) Refresh(_ context.Context, p protocol.RefreshParams) (recall.Health, error)

Refresh brings the projection up to date and reports the resulting health. This is what the checkpoint capability means.

func (*Adapter) Search

func (a *Adapter) Search(ctx context.Context, req recall.SearchRequest) (recall.SearchResponse, error)

Search returns this source's own ranked candidates.

Ranking is deliberately simple and explainable:

1. Exact identifier hits first — a query token equal to a record id, an upstream ref, a correlation key, or a locator's local part. This is a partition, not a bonus, mirroring the core's own exact-match promotion.
2. Everything else by term coverage over the title (1.0), the record's own fields — kind, system, actor, ref (0.5) — and the body (0.4).
3. Ties broken by newest event first, then by id, so the order is total and reproducible.

A query with no terms is a time-window browse, which is a real question to ask an event stream: the window's newest events, in order.

type Options

type Options struct {
	// Clock supplies build and probe timestamps. Nil means [time.Now].
	Clock func() time.Time
}

Options are construction-time seams. Instance policy arrives at the handshake, because that is where the spec puts it.

type Settings

type Settings struct {
	// Files are the stream files, absolute or relative to the configured
	// location. Empty means the location itself names the single file.
	Files []string `json:"files"`

	// Upstream maps a record's `system` to the source_id of the Recall source
	// that owns those records. It is what turns a normalized event into a
	// derived_from edge; a system absent here yields no edge.
	Upstream map[string]string `json:"upstream"`

	MaxCandidates int `json:"max_candidates"`

	// StallMS delays every search before it scans. It exists so the
	// cancellation conformance case can be recorded deterministically against
	// a real process, and it is the shortest demonstration of the only thing
	// an adapter owes a cancel: notice the context, return, do not answer.
	// Leave it unset in real configuration.
	StallMS int `json:"debug_stall_ms"`
}

Settings is the adapter-owned settings block, declared by recall.Manifest.SettingsSchema and validated here on every handshake.