Documentation
¶
Overview ¶
Package protocol implements the newline-delimited JSON-RPC 2.0 transport and message schemas of the Recall adapter protocol.
Boundary: framing, correlation, error codes, and schema validation. It holds no retrieval or ranking logic and never interprets a locator, a rank, or a sensitivity level — it only checks that the shapes crossing the wire are the shapes the contract declares.
The core is the client (Client); an adapter is the server (Serve). Both ends compile the same embedded schemas and validate in both directions, so a contract break is reported by whichever end introduced it rather than surfacing later as a confusing decode error.
Three rules shape the code here:
- One message per line. A frame never contains a raw newline, so a transcript is diffable and a bad frame resynchronizes at the next newline instead of poisoning the stream.
- stdout carries frames only. An adapter's stderr goes to Diagnostics, which stores it and never parses it. Nothing written to stderr can answer a request.
- A request that misses its deadline is a timeout, never an empty success. Client.Call sends the advisory recall/cancel notification, waits a bounded grace, and then reports CallTimeout with whether the adapter answered — which is what tells a supervisor the process is wedged.
Index ¶
- Constants
- Variables
- func CheckNegotiated(p InitializeParams, got int) error
- func NegotiateVersion(minVersion, maxVersion int) (int, error)
- func Recoverable(err error) bool
- func Serve(ctx context.Context, r io.Reader, w io.Writer, h Handler) error
- type CallTimeout
- type CancelParams
- type Client
- type ClientOptions
- type Code
- type Decoder
- type Diagnostics
- type Encoder
- type Error
- type FrameError
- type Handler
- type HealthParams
- type ID
- type InitializeParams
- type Message
- type RefreshParams
- type SchemaSet
- type ShutdownParams
- type ShutdownResult
- type VersionError
Constants ¶
const ( DefaultDiagnosticLines = 64 DefaultDiagnosticLenCap = 4096 )
Diagnostics defaults. A few dozen recent lines is what an operator reads; more than that belongs in the adapter's own log.
const ( MinVersion = 1 MaxVersion = 1 )
The protocol version range this build speaks. Negotiation happens once, in recall/initialize; an adapter that cannot satisfy the range fails the handshake rather than degrading.
const ( MethodInitialize = "recall/initialize" MethodSearch = "recall/search" MethodExpand = "recall/expand" MethodHealth = "recall/health" MethodRefresh = "recall/refresh" MethodCancel = "recall/cancel" MethodShutdown = "recall/shutdown" )
Protocol methods. recall/cancel is the only notification; everything else is a request that carries a deadline.
const DefaultCancelGrace = 250 * time.Millisecond
DefaultCancelGrace is how long a cancel notification has to be answered before the peer is treated as wedged. It is short: the deadline has already passed, so this only buys the adapter time to say "I stopped".
const DiagStoreIdentity = "store_identity"
DiagStoreIdentity is the health diagnostic under which an adapter names the store one instance actually opened.
It exists because "two sources, one store" is a configuration error that no single source can see. Lineage groups on source_uid plus source_record_id, so one record read twice through two instances arrives as two independent pieces of evidence and collects the corroboration bonus for agreeing with itself. That has now happened three times in this system — overlapping document roots, two catalog instances over one server, and two td sources resolving to one database — reached by three different routes, so the check belongs somewhere that sees every source at once. `recall doctor` reads this key and refuses a profile in which two enabled instances of one adapter report the same value.
Setting it is a CLAIM OF EXCLUSIVITY: this store is mine alone. An adapter for which two instances over one store is a legitimate configuration — one serving different views of a shared catalog, whose candidates collapse on a content fingerprint — must leave it unset, because for that adapter the overlap is the design rather than the defect. Absent means "makes no such claim", never "unknown".
The value is opaque and compared only for equality between instances of the SAME adapter, so an adapter may use whatever names its store precisely: a resolved filesystem path, a database identity, an endpoint plus a namespace. It must name the store the adapter OPENED, never the one configuration asked for — a value copied from configuration would compare equal exactly when the configuration is consistent, which is the case that was never in doubt.
const MaxFrameBytes = 8 << 20
MaxFrameBytes bounds one line. Recall messages are compact by design — candidates are pointers and expansion is budget-bounded — so a frame this large is a runaway adapter, not a big result.
const Version = "2.0"
Version is the JSON-RPC version string every frame carries.
Variables ¶
var ( ErrParse = &Error{Code: CodeParse, Message: CodeParse.String()} ErrInvalidRequest = &Error{Code: CodeInvalidRequest, Message: CodeInvalidRequest.String()} ErrMethodNotFound = &Error{Code: CodeMethodNotFound, Message: CodeMethodNotFound.String()} ErrInvalidParams = &Error{Code: CodeInvalidParams, Message: CodeInvalidParams.String()} ErrInternal = &Error{Code: CodeInternal, Message: CodeInternal.String()} ErrSourceDenied = &Error{Code: CodeSourceDenied, Message: CodeSourceDenied.String()} ErrLocatorUnknown = &Error{Code: CodeLocatorUnknown, Message: CodeLocatorUnknown.String()} ErrLocatorExpired = &Error{Code: CodeLocatorExpired, Message: CodeLocatorExpired.String()} ErrSourceNotConfigured = &Error{Code: CodeSourceNotConfigured, Message: CodeSourceNotConfigured.String()} ErrAsOfUnsupported = &Error{Code: CodeAsOfUnsupported, Message: CodeAsOfUnsupported.String()} ErrBudgetExceeded = &Error{Code: CodeBudgetExceeded, Message: CodeBudgetExceeded.String()} ErrDeadlineExceeded = &Error{Code: CodeDeadlineExceeded, Message: CodeDeadlineExceeded.String()} )
Sentinels for errors.Is. They carry no message: a code is the contract, a message is a diagnostic.
var ErrClosed = errors.New("protocol: client closed")
ErrClosed reports a call on a client whose stream is gone.
var ErrFrameTooLarge = errors.New("protocol frame exceeds the line limit")
ErrFrameTooLarge reports a line past the frame limit. The line is consumed to its newline before the error is returned, so the stream stays framed and the next Decode reads a real message.
Functions ¶
func CheckNegotiated ¶
func CheckNegotiated(p InitializeParams, got int) error
CheckNegotiated verifies the version an adapter reported is one the core asked for. A manifest naming anything else fails the handshake.
func NegotiateVersion ¶
NegotiateVersion picks the highest version both ends support. An adapter calls it from its initialize handler; a range with no overlap is an error, never a downgrade.
func Recoverable ¶
Recoverable reports whether a decode failure left the stream usable. A reader records these and continues; anything else ends the session.
func Serve ¶
Serve reads frames from r, dispatches them to h, and writes replies to w.
Requests run concurrently, each under a context that recall/cancel cancels and the request's own deadline bounds. The encoder serializes writes, so two concurrent replies can never interleave into an unparseable line.
Serve returns when the peer closes the stream or asks for shutdown.
Types ¶
type CallTimeout ¶
type CallTimeout struct {
// Method is the request method that timed out.
Method string
// ID is the request's correlation identity.
ID ID
// Acknowledged reports whether the peer answered recall/cancel.
Acknowledged bool
// Cause is the context error that ended the wait.
Cause error
}
CallTimeout reports a request that outlived its deadline.
Acknowledged is the field that matters to a supervisor. A peer that answered the cancel notification is alive and can be reused; a peer that did not is wedged and must be signalled. Either way this is a timeout, never an empty success.
func (*CallTimeout) Error ¶
func (e *CallTimeout) Error() string
func (*CallTimeout) Unwrap ¶
func (e *CallTimeout) Unwrap() error
type CancelParams ¶
type CancelParams struct {
// ID names the request to abandon.
ID ID `json:"id"`
}
CancelParams names the in-flight request to abandon. Cancellation is advisory: the core still enforces the deadline itself.
type Client ¶
type Client struct {
// contains filtered or unexported fields
}
Client is the core's end of the protocol. It writes requests, correlates responses by id, and enforces deadlines.
Correlation is by the exact id text the request carried, held in a map that is only ever mutated under one mutex. A response whose id has no waiter is counted as a violation and dropped: it can never be handed to a different call.
func NewClient ¶
NewClient starts reading frames from r and writes requests to w. The returned client owns a goroutine until the stream ends or Client.Close runs.
func (*Client) Call ¶
Call sends a request and waits for its response.
On a deadline it sends the advisory recall/cancel notification, waits the configured grace, and returns CallTimeout. It never returns a zero result with a nil error: a caller that ignores the error still cannot mistake a timeout for an answer, because result is left untouched.
func (*Client) Close ¶
Close releases the write side so the peer sees EOF. It does not wait for the peer: a supervisor that must guarantee exit follows this with a signal.
func (*Client) Diagnostics ¶
func (c *Client) Diagnostics() *Diagnostics
Diagnostics returns the buffer holding this peer's stderr and violations.
type ClientOptions ¶
type ClientOptions struct {
// Diagnostics receives stderr and protocol violations. A nil value gets a
// fresh buffer, so violations are always counted somewhere.
Diagnostics *Diagnostics
// CancelGrace bounds the wait for an answer to a cancel notification.
CancelGrace time.Duration
// MaxFrame lowers the frame limit on both directions.
MaxFrame int
// Closer is closed by [Client.Close], typically the peer's stdin, so the
// peer sees EOF and can exit on its own.
Closer io.Closer
}
ClientOptions tunes a client. The zero value is usable.
type Code ¶
type Code int
Code is a JSON-RPC error code. The standard codes are the transport's own; the -32000..-32007 block is Recall's, in the range JSON-RPC reserves for implementation-defined server errors.
const ( CodeParse Code = -32700 CodeInvalidRequest Code = -32600 CodeMethodNotFound Code = -32601 CodeInvalidParams Code = -32602 CodeInternal Code = -32603 )
Standard JSON-RPC codes.
const ( // reported as a search that succeeded with no matches. CodeSourceUnavailable Code = -32000 // CodeSourceDenied means permission was refused. Its diagnostics must not // reveal whether a record exists. CodeSourceDenied Code = -32001 // CodeLocatorUnknown means the locator does not parse for this adapter. CodeLocatorUnknown Code = -32002 // CodeLocatorExpired means the source changed incompatibly. Expansion fails // rather than returning a different revision or a nearby record. CodeLocatorExpired Code = -32003 // CodeSourceNotConfigured means the locator names a source this machine // does not have. CodeSourceNotConfigured Code = -32004 // CodeAsOfUnsupported means the historical boundary cannot be honored. A // source never answers an as_of query from current state instead. CodeAsOfUnsupported Code = -32005 // CodeBudgetExceeded means the adapter declined the request up front: it // cannot be served within the budget offered. CodeBudgetExceeded Code = -32006 // CodeDeadlineExceeded means the request ran out of time, or was abandoned // by the caller while in flight. // // It is distinct from budget_exceeded, which was the only code available // before and made one condition report two different ways: a timed-out // adapter came back as SearchTimeout in process and as SearchFailed over // the wire, purely because a different side of the process boundary // noticed. Evaluation compares source outcomes exactly, so the same case // scored differently depending on the transport. CodeDeadlineExceeded Code = -32007 )
Recall codes. These are the contract in docs/adapter-protocol.md; the numbers are wire values and must not be renumbered.
type Decoder ¶
type Decoder struct {
// contains filtered or unexported fields
}
Decoder reads newline-delimited frames.
func NewDecoder ¶
NewDecoder reads frames from r, rejecting any line longer than MaxFrameBytes.
func (*Decoder) Decode ¶
Decode reads the next frame. Blank lines are skipped: they carry no message and dropping them costs nothing.
func (*Decoder) SetMaxFrame ¶
SetMaxFrame lowers the frame limit. Tests use it to exercise the oversized path without allocating megabytes.
type Diagnostics ¶
type Diagnostics struct {
// contains filtered or unexported fields
}
Diagnostics captures an adapter's stderr and its protocol violations.
stderr is free-form adapter logging. It is stored and never parsed: no line written there can answer a request, complete a handshake, or change an outcome. That separation is the whole reason the protocol lives on stdout alone, so it is enforced by there being no code path from here to a decoder.
The buffer is bounded in both directions. A chatty adapter cannot grow Recall's memory, and a single enormous line cannot either; what was dropped is counted so the report stays honest.
func NewDiagnostics ¶
func NewDiagnostics() *Diagnostics
NewDiagnostics returns a bounded capture buffer.
func (*Diagnostics) Capture ¶
func (d *Diagnostics) Capture(r io.Reader)
Capture reads r to EOF, storing each line. It blocks, so callers run it on its own goroutine for the life of the process.
func (*Diagnostics) Dropped ¶
func (d *Diagnostics) Dropped() int
Dropped returns how many stderr lines were evicted.
func (*Diagnostics) Lines ¶
func (d *Diagnostics) Lines() []string
Lines returns a copy of the captured stderr.
func (*Diagnostics) Map ¶
func (d *Diagnostics) Map() map[string]any
Map renders the capture for a diagnostics field on a result or a health report. It is a snapshot: callers may keep it without holding a lock.
func (*Diagnostics) Record ¶
func (d *Diagnostics) Record(line string)
Record stores one line, truncating it and evicting the oldest as needed.
func (*Diagnostics) RecordViolation ¶
func (d *Diagnostics) RecordViolation(err error)
RecordViolation notes a frame that could not be used. Violations are counted separately from stderr because they mean the adapter's stdout is not clean, which is a contract break rather than logging.
func (*Diagnostics) Violations ¶
func (d *Diagnostics) Violations() int
Violations returns how many frames were unusable.
type Encoder ¶
type Encoder struct {
// contains filtered or unexported fields
}
Encoder writes newline-delimited frames. It is safe for concurrent use: an adapter answers concurrent requests, and two half-written frames interleaved on stdout would be unrecoverable.
func (*Encoder) Encode ¶
Encode writes one frame followed by a newline.
encoding/json compacts embedded raw messages and escapes control characters inside strings, so the result cannot contain a raw newline. The check below asserts that rather than trusting it, because a frame carrying a newline would silently split into two unparseable messages at the peer.
func (*Encoder) SetMaxFrame ¶
SetMaxFrame lowers the frame limit, so this end refuses to write what the other end would refuse to read.
type Error ¶
type Error struct {
// Code is the stable JSON-RPC or Recall error code.
Code Code `json:"code"`
// Message is safe human-readable diagnostic detail.
Message string `json:"message"`
// Data is optional structured error detail.
Data json.RawMessage `json:"data,omitempty"`
}
Error is a JSON-RPC error object and a Go error.
Matching is by code: every sentinel below is an *Error carrying only a code, and Error.Is compares codes, so errors.Is(err, ErrSourceDenied) holds for any denial regardless of the message the adapter attached. errors.As recovers the full value when the caller wants the message or data.
func AsError ¶
AsError renders any handler failure as a protocol error. A handler that returns a typed protocol error keeps its code; anything else is an internal error, because guessing a Recall code from an arbitrary message would put words in the adapter's mouth.
func Errorf ¶
Errorf builds an error with a formatted message. The message is a diagnostic: it must stay safe to show, which for a denied source means it must not reveal whether a record exists.
type FrameError ¶
type FrameError struct {
// Size is the length of the offending line. The line itself is not kept:
// it may contain source content, which is not logged by default.
Size int
// Err is the underlying JSON or message validation error.
Err error
}
FrameError is a line that was read whole but is not a protocol message. It is distinguished from a broken stream because the two need different reactions: a bad frame is recorded and skipped, a broken stream ends the session.
func (*FrameError) Error ¶
func (e *FrameError) Error() string
func (*FrameError) Unwrap ¶
func (e *FrameError) Unwrap() error
type Handler ¶
type Handler interface {
// Initialize negotiates the protocol version and declares what the adapter
// can do. A range it cannot satisfy is an error, never a downgrade.
Initialize(ctx context.Context, p InitializeParams) (recall.Manifest, error)
// Search returns this adapter's ranked candidates.
Search(ctx context.Context, req recall.SearchRequest) (recall.SearchResponse, error)
// Expand resolves one adapter-owned locator into evidence.
Expand(ctx context.Context, req recall.ExpandRequest) (recall.ExpandResponse, error)
// Health reports source availability, freshness, and coverage.
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. A build that fails is reported through the returned health,
// not as an error: a frame carries a result or an error and never both, so
// erroring would discard the health of the generation still answering.
Refresh(ctx context.Context, p RefreshParams) (recall.Health, error)
// Shutdown is asked for a clean exit. Serve returns once in-flight work
// finishes; a handler that never finishes is what SIGTERM is for.
Shutdown(ctx context.Context) error
}
Handler is the adapter end of the contract: six handlers over stdin and stdout. Any language with a JSON library can implement the same thing; this interface exists so a Go adapter, built-in or external, does not have to reimplement framing to be reachable over the wire.
type HealthParams ¶
type HealthParams struct {
// Deadline is the absolute time by which the probe must finish.
Deadline time.Time `json:"deadline"`
}
HealthParams carries the probe's deadline. Every request carries one.
type ID ¶
type ID struct {
// contains filtered or unexported fields
}
ID is a JSON-RPC correlation identity.
The specification allows a string or a number and requires a response to echo the request's id. The raw JSON text is therefore kept verbatim and used as the correlation key: nothing reformats an id, so a large integer or an unusual string cannot round-trip into a different value and correlate a response to the wrong request.
func NumberID ¶
NumberID builds a numeric id. The client numbers its own requests, so this is the form the core uses.
func (ID) IsZero ¶
IsZero reports whether the id names nothing. A message without an id is a notification.
func (ID) MarshalJSON ¶
MarshalJSON preserves the ID's original JSON representation.
func (*ID) UnmarshalJSON ¶
UnmarshalJSON accepts a JSON-RPC string, number, or null ID.
type InitializeParams ¶
type InitializeParams struct {
// ProtocolVersionMin is the oldest protocol version the caller accepts.
ProtocolVersionMin int `json:"protocol_version_min"`
// ProtocolVersionMax is the newest protocol version the caller accepts.
ProtocolVersionMax int `json:"protocol_version_max"`
// Workdir is the adapter's writable state directory.
Workdir string `json:"workdir"`
// SourceID is the display name this instance was configured under. An
// adapter needs it because locator text is "<source_id>:<local>" and it
// writes its own locators. It is NOT identity: the core attaches the
// immutable source_uid and overwrites the source part of every locator an
// adapter returns, so a forged prefix cannot make one source answer as
// another.
SourceID string `json:"source_id"`
// Location is the configured path, endpoint, or connection reference for
// this source instance. Configuration has already resolved a relative one
// against the file that declared it.
Location string `json:"location,omitempty"`
// BaseDir is the directory of the file that declared this source.
//
// It exists because Settings is adapter-owned, so configuration cannot
// resolve paths inside it — an adapter reading settings.root had nothing to
// resolve against and fell back to the process working directory, which
// makes the same configuration read different files depending on where
// Recall was started.
BaseDir string `json:"base_dir,omitempty"`
// Settings is the adapter-owned settings block. Its shape is declared by
// the manifest's settings_schema, which is only knowable after the
// handshake, so the adapter validates it — the core validates it against
// the manifest it already holds when it has one.
Settings map[string]any `json:"settings,omitempty"`
}
InitializeParams is the handshake input. The version range is the whole point of the handshake: an adapter that cannot land inside it fails explicitly.
Workdir is a writable directory under Recall's state directory. An adapter writes indexes and checkpoints only there.
type Message ¶
type Message struct {
// JSONRPC is always [Version].
JSONRPC string `json:"jsonrpc"`
// ID correlates a request and response; it is absent on notifications.
ID *ID `json:"id,omitempty"`
// Method names a request or notification.
Method string `json:"method,omitempty"`
// Params is the method-specific request payload.
Params json.RawMessage `json:"params,omitempty"`
// Result is the method-specific success payload.
Result json.RawMessage `json:"result,omitempty"`
// Error is the failure payload.
Error *Error `json:"error,omitempty"`
}
Message is one JSON-RPC frame. Requests, notifications, and responses share one struct because the wire does: which one a frame is follows from which fields are present, and Message.Validate is the single place that decides.
func NewErrorResponse ¶
NewErrorResponse builds a failure response.
func NewNotification ¶
func NewNotification(method string, params json.RawMessage) *Message
NewNotification builds a frame that expects no response.
func NewRequest ¶
func NewRequest(id ID, method string, params json.RawMessage) *Message
NewRequest builds a request frame.
func NewResult ¶
func NewResult(id ID, result json.RawMessage) *Message
NewResult builds a success response.
func (*Message) IsNotification ¶
IsNotification reports whether the frame is fire-and-forget.
func (*Message) IsResponse ¶
IsResponse reports whether the frame answers an earlier request.
type RefreshParams ¶
type RefreshParams struct {
// Deadline is the absolute time by which the refresh must finish.
Deadline time.Time `json:"deadline"`
// Full asks for a complete rebuild rather than an incremental pass. An
// adapter with no incremental path may treat every refresh as full.
Full bool `json:"full,omitempty"`
}
RefreshParams asks an adapter to bring its projection up to date and returns the resulting health. Only adapters declaring recall.CapCheckpoint serve it.
type SchemaSet ¶
type SchemaSet struct {
// contains filtered or unexported fields
}
SchemaSet is the compiled contract for every payload that crosses the wire.
Both ends hold one. The sender validates before writing and the receiver validates after reading, so a shape that violates the contract is reported by the end that produced it instead of turning into a confusing decode failure three layers later.
func Schemas ¶
Schemas returns the compiled schema set. Compilation happens once per process; the result is immutable and safe to share.
func (*SchemaSet) ValidateParams ¶
func (s *SchemaSet) ValidateParams(method string, raw json.RawMessage) error
ValidateParams checks a request's params against its method's schema.
func (*SchemaSet) ValidateResult ¶
func (s *SchemaSet) ValidateResult(method string, raw json.RawMessage) error
ValidateResult checks a response's result against its method's schema. A method with no result contract, such as the cancel notification, has nothing to check and reports as much.
type VersionError ¶
type VersionError struct {
// Min and Max are the requested version range.
Min, Max int
// Offered is the version the peer named, zero when it named none.
Offered int
// Supported describes what this end can speak.
SupportedMin, SupportedMax int
}
VersionError reports a handshake that could not be satisfied. It is returned rather than silently choosing a version either end does not implement.
func (*VersionError) Error ¶
func (e *VersionError) Error() string
Source Files