adapter

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 30, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Overview

Package adapter defines the single adapter contract and supervises the processes that implement it out of band.

Boundary: one interface, two transports. A built-in adapter implements Adapter directly. An external adapter implements the same interface over JSON-RPC via pkg/protocol, reached through Connect for an existing stream or External for a subprocess Recall owns. Serve runs the bridge the other way, exposing any Adapter on a stream. Nothing in this package ranks, fuses, or interprets a locator; it moves requests to a source implementation and reports honestly about what came back.

The core owns spawn, handshake, deadline enforcement, kill, restart policy, and pooling. Adapters own retrieval, ranking within their source, indexing, and locator semantics.

Two rules are load-bearing enough to be worth stating here, because both are invariants rather than behaviors that could be tuned:

  • A source that could not be reached, could not be spawned, could not agree on a protocol version, or ran past its deadline never produces a successful empty result. Every failure path goes through FailedSearch, which cannot return recall.SearchSuccess.
  • Deadline enforcement escalates: the advisory recall/cancel notification first, then SIGTERM, then SIGKILL. An adapter that answers the cancel is kept; one that does not is killed and respawned on next use.

Index

Constants

View Source 
const (
	// DefaultHealthTTL is how long a health probe is reused. The spec's
	// default; probing every source on every query would cost more than the
	// query.
	DefaultHealthTTL = 30 * time.Second

	// DefaultCancelGrace is how long the advisory cancel notification has to be
	// answered before the process is treated as wedged.
	DefaultCancelGrace = protocol.DefaultCancelGrace

	// DefaultTermGrace is how long a clean exit or a SIGTERM has before SIGKILL.
	DefaultTermGrace = 2 * time.Second

	// DefaultHandshakeTimeout bounds spawn plus initialize. Cold start counts
	// against the request budget, so it cannot be unbounded.
	DefaultHandshakeTimeout = 10 * time.Second

	// DefaultProbeTimeout bounds a health probe whose caller supplied no
	// deadline.
	DefaultProbeTimeout = 5 * time.Second

	// DefaultCallTimeout bounds a request whose caller set no deadline. Every
	// request is supposed to carry one; this exists so that forgetting cannot
	// mean "wait forever on a subprocess".
	DefaultCallTimeout = 30 * time.Second
)

Defaults for adapter supervision.

Variables

View Source 
var ErrClosed = errors.New("adapter: closed")

ErrClosed reports use of an adapter that has been closed.

Functions

func Classify 

func Classify(err error) (recall.SearchOutcome, string)

Classify maps a failure to the outcome the core reports for it, plus a short machine-readable reason.

The reasons are contract vocabulary, not prose: they appear in recall.SourceReport.Reason, which a person acts on.

func FailedSearch 

func FailedSearch(err error) recall.SearchResponse

FailedSearch renders a failure as an honest search response.

This is the only bridge from an error to a recall.SearchResponse, and it cannot produce recall.SearchSuccess. That is deliberate: a missing dependency, an unreachable source, or an adapter crash reported as a successful empty result would be indistinguishable from a source that simply had no matches, and the whole ranking layer downstream would believe it.

func Serve 

func Serve(ctx context.Context, r io.Reader, w io.Writer, a Adapter) error

Serve exposes an Adapter on a stream, speaking the same protocol an external adapter does.

This is the other half of "one contract, two transports". A built-in adapter written once is reachable in process through the interface and over the wire through here, so conformance transcripts and the evaluation replay exercise the same implementation the CLI does — rather than a second one written to match.

func Unhealthy 

func Unhealthy(err error) recall.Health

Unhealthy renders a failed probe. An unreachable source is never healthy and never has a known coverage.

func UnsupportedFilters 

func UnsupportedFilters(filters recall.Filters, names ...string) (recall.SearchResponse, bool)

UnsupportedFilters returns the only honest response for a filter this adapter cannot evaluate. It is deliberately called before retrieval: a broader result set is not evidence for the narrower question, even when labeled partial.

Types

type Adapter 

type Adapter interface {
	// Initialize negotiates the protocol version and returns what the adapter
	// can do. Negotiation happens once per instance: a range the adapter
	// cannot satisfy fails here rather than degrading to a version neither
	// side implements.
	Initialize(ctx context.Context, cfg Config) (recall.Manifest, error)

	// Search returns this source's own ranked candidates. A source that could
	// not answer reports it in [recall.SearchResponse.Outcome] and returns an
	// error; it never reports success with no candidates.
	Search(ctx context.Context, req recall.SearchRequest) (recall.SearchResponse, error)

	// Expand retrieves evidence behind a locator. A source that changed
	// incompatibly fails with locator_expired rather than returning a
	// different revision or a nearby record.
	Expand(ctx context.Context, req recall.ExpandRequest) (recall.ExpandResponse, error)

	// Health probes the source. Results may be served from a cache with a TTL;
	// see [DefaultHealthTTL].
	Health(ctx context.Context) (recall.Health, error)

	// Refresh brings an adapter-owned projection up to date and reports the
	// resulting health. An adapter that owns no index returns its health
	// unchanged.
	//
	// This is what [recall.CapCheckpoint] means. Without it the capability was
	// a word an adapter could declare and nothing could invoke, and the only
	// in-contract place to build an index was the handshake — which competes
	// with DefaultHandshakeTimeout on any real corpus.
	//
	// A refresh whose build fails reports that through the health it returns —
	// stale watermark, degraded status, the reason in diagnostics — and not as
	// an error. A JSON-RPC frame carries a result or an error and never both,
	// so an error return would discard the health of the generation that is
	// still published and still answering. The error return means the refresh
	// could not be performed at all.
	Refresh(ctx context.Context, p protocol.RefreshParams) (recall.Health, error)

	// Close releases the adapter. For a subprocess this asks for a clean exit
	// and then guarantees one.
	Close() error
}

Adapter is Recall's one contract with a source implementation.

Built-in adapters implement it directly; external adapters implement it over JSON-RPC. There is one contract with two transports, which is why the signatures speak only in domain types: nothing here is shaped by the fact that a subprocess might be involved.

func WithIdentity 

func WithIdentity(a Adapter, id Identity) Adapter

WithIdentity stamps configuration's identity onto everything an adapter returns, and drops candidates an adapter had no right to emit.

This exists as a wrapper rather than a step in the orchestrator because it must be unskippable. Three separate contracts — the candidate envelope, the locator model, and lineage grouping — are written assuming a candidate arrives already carrying its identity, and lineage grouping resolves the locator prefix as a display name when the identity is missing. So an adapter that returns `{"locator": "tasks:td-f62256"}` while configured as some other source would have its evidence grouped under the Tasks lineage root, and the printed locator would route a later expansion to Tasks. One source would answer as another.

Wrapping closes that by construction: the source part of every locator is replaced with the configured identity regardless of what the adapter wrote, so a forged prefix cannot survive. Only derived_from is left alone — those name other sources by design, and they are resolved against the profile and dropped when unknown, which is a claim about lineage rather than about who is answering.

type Config 

type Config = protocol.InitializeParams

Config is the handshake input: protocol version range, writable workdir, location, and the adapter-owned settings block.

It is an alias, not a copy, of the wire type. The handshake shape is the same fact whether it crosses a process boundary or not, and two structs that had to be kept in sync would eventually drift.

type Conn 

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

Conn is an Adapter reached over the protocol on an already-connected stream.

It holds no process knowledge, so the same request path serves a subprocess and an in-process pipe. External adds supervision on top by installing an escalation hook; without one, a wedged peer still fails as a timeout, there is simply nothing to signal.

func Connect 

func Connect(ctx context.Context, r io.Reader, w io.WriteCloser, cfg Config, opt Options) (*Conn, error)

Connect performs the handshake over an existing stream and returns the adapter behind it.

w is closed by Conn.Close so the peer sees EOF on its stdin. A handshake that cannot agree on a protocol version fails here; there is no partially initialized Conn.

func (*Conn) Close 

func (c *Conn) Close() error

Close asks for a clean exit and releases the stream.

func (*Conn) ColdStart 

func (c *Conn) ColdStart() time.Duration

ColdStart returns how long this connection took to become ready. It counts against the request budget that paid for it and is reported separately from warm latency.

func (*Conn) Diagnostics 

func (c *Conn) Diagnostics() *protocol.Diagnostics

Diagnostics returns the peer's captured stderr and protocol violations.

func (*Conn) Expand 

func (c *Conn) Expand(ctx context.Context, req recall.ExpandRequest) (recall.ExpandResponse, error)

Expand retrieves evidence behind a locator.

func (*Conn) Health 

func (c *Conn) Health(ctx context.Context) (recall.Health, error)

Health probes the source, reusing a recent probe when one is within the TTL.

Probes are serialized: a burst of concurrent queries against one source should cost one probe, not one per query.

func (*Conn) Initialize 

func (c *Conn) Initialize(context.Context, Config) (recall.Manifest, error)

Initialize returns the negotiated manifest.

Negotiation happens once, at connect time: a live connection has already agreed on a version and cannot renegotiate. cfg is therefore ignored here, and is present because the same interface is implemented by built-in adapters that have nothing connected yet.

func (*Conn) Manifest 

func (c *Conn) Manifest() recall.Manifest

Manifest returns what the adapter declared at handshake.

func (*Conn) Refresh 

func (c *Conn) Refresh(ctx context.Context, p protocol.RefreshParams) (recall.Health, error)

Refresh asks the adapter to bring its projection up to date.

The health cache is dropped afterwards whatever the outcome: a refresh is the one operation certain to have changed what a probe would say, so serving a stale entry here would report the generation the refresh replaced.

func (*Conn) Search 

func (c *Conn) Search(ctx context.Context, req recall.SearchRequest) (recall.SearchResponse, error)

Search asks the adapter for candidates.

Every failure returns both an error and a response whose outcome says what happened. A caller that inspects only one of the two still cannot mistake an unreachable source for a source with no matches.

type External 

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

External supervises one out-of-process adapter for one source instance.

The process is spawned on first use and pooled: one process per source instance, reused for its lifetime, with in-flight requests bounded by the manifest's max_concurrency. Two projects using the same adapter binary are two source instances and get two processes, because they are two sources.

Deadline enforcement escalates and is never silent. The advisory recall/cancel notification goes first; an adapter that answers it keeps its process. One that does not is wedged, gets SIGTERM, then SIGKILL, and the request is reported as a timeout. A killed process is respawned on the next use rather than retried inside the failed request: a retry would spend budget the caller did not grant and could re-run work the adapter had already begun.

func NewExternal 

func NewExternal(spec Spec) *External

NewExternal prepares an adapter. Nothing is spawned until the first request: a configured source that is never queried costs nothing.

func (*External) Close 

func (e *External) Close() error

Close asks for a clean exit and then guarantees one. It is idempotent.

func (*External) Diagnostics 

func (e *External) Diagnostics() map[string]any

Diagnostics returns the adapter's captured stderr, protocol violations, and supervision history. Nothing here is parsed as protocol.

func (*External) Expand 

func (e *External) Expand(ctx context.Context, req recall.ExpandRequest) (recall.ExpandResponse, error)

Expand spawns if needed, then retrieves evidence.

func (*External) Health 

func (e *External) Health(ctx context.Context) (recall.Health, error)

Health spawns if needed, then probes. Health is probed on spawn and cached with a TTL, so a burst of queries pays for one probe.

func (*External) Initialize 

func (e *External) Initialize(ctx context.Context, cfg Config) (recall.Manifest, error)

Initialize spawns the adapter if it is not running and returns its manifest.

Negotiation happens once per process. cfg replaces the configured handshake input only when no process is running; an already-negotiated session reports what it agreed to.

func (*External) Manifest 

func (e *External) Manifest() (recall.Manifest, bool)

Manifest returns the manifest of the live session, if there is one.

func (*External) Refresh 

func (e *External) Refresh(ctx context.Context, p protocol.RefreshParams) (recall.Health, error)

Refresh asks the adapter to bring its projection up to date.

func (*External) Search 

func (e *External) Search(ctx context.Context, req recall.SearchRequest) (recall.SearchResponse, error)

Search spawns if needed, then asks. A source that could not be started is unavailable, not empty.

type Identity 

type Identity struct {
	// UID is the immutable configured source identity.
	UID recall.SourceUID
	// ID is the configured display and locator prefix.
	ID string
	// Floor is the least restrictive classification the source may return.
	Floor recall.Sensitivity
}

Identity is what configuration assigned a source instance. An adapter never supplies any of it: the UID is generated at configuration time, the display name is the user's, and the sensitivity floor is policy.

type Options 

type Options struct {
	// HealthTTL bounds how long a probe result is reused.
	HealthTTL time.Duration

	// CancelGrace bounds the wait for an answer to recall/cancel before the
	// adapter is treated as wedged.
	CancelGrace time.Duration

	// TermGrace bounds a clean exit, and then bounds SIGTERM before SIGKILL.
	TermGrace time.Duration

	// HandshakeTimeout bounds spawn plus initialize.
	HandshakeTimeout time.Duration

	// Diagnostics receives the adapter's stderr and any protocol violations. A
	// nil value gets a fresh buffer.
	Diagnostics *protocol.Diagnostics

	// MaxFrame lowers the protocol frame limit.
	MaxFrame int
}

Options tune transport behavior. The zero value uses the defaults above.

type PreparedSearcher 

type PreparedSearcher interface {
	// PrepareSearch checks health and returns request-scoped preparation.
	PrepareSearch(ctx context.Context, req recall.SearchRequest) (recall.Health, SearchPreparation, error)
	// SearchPrepared searches using preparation returned for the same request.
	SearchPrepared(ctx context.Context, req recall.SearchRequest, preparation SearchPreparation) (recall.SearchResponse, error)
}

PreparedSearcher lets a built-in adapter carry one request's health handshake into, or safely perform it beside, its immediately following search.

Preparation is opaque to the core and lives only on the in-memory retrieval plan. It is never serialized, cached globally, or reused by another request. PrepareSearch must make the same eligibility decision as Health. Work overlapped with that decision is speculative until the health and source identity checks admit it, and must be discarded on disagreement. SearchPrepared must preserve Search's result and cancellation contracts.

This seam is optional. It exists for built-in sources whose health and search otherwise repeat or serialize expensive setup. External adapters keep using the ordinary wire contract until that protocol has an equivalent request-scoped token.

type SearchPreparation 

type SearchPreparation struct {
	// State is opaque adapter-owned request-scoped preparation.
	State any
	// Elapsed is the adapter-observed preparation and search duration.
	Elapsed time.Duration
}

SearchPreparation is the opaque result of planning one prepared search. Elapsed is the adapter-observed search duration, including work overlapped with its health handshake.

type SpawnError 

type SpawnError struct {
	// Name identifies the source instance that could not start.
	Name string
	// Command is the executable that could not start.
	Command string
	// Err is the underlying process-start failure.
	Err error
}

SpawnError reports that an external adapter could not be started. It is distinct from a source failure: nothing was ever asked.

func (*SpawnError) Error 

func (e *SpawnError) Error() string

func (*SpawnError) Unwrap 

func (e *SpawnError) Unwrap() error

type Spec 

type Spec struct {
	// Name identifies this source instance in diagnostics. It is not an
	// identity: source_uid comes from configuration.
	Name string

	// Command is the external adapter executable.
	Command string
	// Args are passed to Command.
	Args []string

	// Env replaces the environment when non-nil. A nil value inherits.
	Env []string

	// Dir is the process working directory. It is not the adapter's workdir;
	// that is Config.Workdir, which is where an index may be written.
	Dir string

	// Config is the adapter's initialization input.
	Config Config

	// Options tune subprocess supervision and transport behavior.
	Options
}

Spec is everything needed to run an external adapter.

Command, Args, and Env come only from user-level configuration. A project configuration travels with a cloned repository, so it may never introduce an executable path; see the trust boundary in docs/spec.md.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.