pipeline

Documentation

Overview

Package pipeline orchestrates one offload request end to end: trivial-check -> context-budget trim -> cache -> build -> generate(grammar) -> parse -> verify -> validate -> (retry|defer|accept) -> cache + ledger.

Index

• func AnswersAgree(task string, candidate string, finalData []byte) (agreed bool, ok bool)
• func NewRecordlessOffload(cfg config.Config, model string, timeout time.Duration) ...
• type Pipeline
  • func New(cfg config.Config, c *llamaclient.Client, ca *cache.Cache, l *ledger.Ledger) *Pipeline
  • func NewRecordlessPipeline(cfg config.Config, timeout time.Duration) *Pipeline
  • func (p *Pipeline) Cfg() config.Config
  • func (p *Pipeline) Run(ctx context.Context, req core.Request) core.Result
  • func (p *Pipeline) RunTier(ctx context.Context, req core.Request, model string) (core.Result, bool)
  • func (p *Pipeline) StartReloader(interval time.Duration) (stop func())

Functions

func AnswersAgree

func AnswersAgree(task string, candidate string, finalData []byte) (agreed bool, ok bool)

AnswersAgree is a thin exported wrapper around answersAgree for use by the shadow-labeling flywheel (which lives in a separate package and cannot call the unexported function directly). task is a task-type string (e.g. "classify"). Behavior is identical to answersAgree.

func NewRecordlessOffload

func NewRecordlessOffload(cfg config.Config, model string, timeout time.Duration) func(ctx context.Context, task, input string, params map[string]any) (string, error)

NewRecordlessOffload builds the in-process offload closure for the local Agent-loop: a FRESH pipeline with nil cache + nil ledger, calling RunTier (record=false), so the agent's offload_* calls cannot write the savings ledger, cache, shadow store, or exemplars. This is the SINGLE place that record=false / nil-store invariant is constructed, so every drive mode (CLI, MCP front door, standalone) shares it and the ledger-pristine guarantee cannot drift.

model is the planner/cascade entry model id. On any non-result (defer / tier miss) it returns a defer JSON the loop can react to — never a fatal error.

Types

type Pipeline

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

func New

func New(cfg config.Config, c *llamaclient.Client, ca *cache.Cache, l *ledger.Ledger) *Pipeline

func NewRecordlessPipeline

func NewRecordlessPipeline(cfg config.Config, timeout time.Duration) *Pipeline

NewRecordlessPipeline builds a fresh pipeline with nil cache + nil ledger, so nothing it runs (RunTier / any offload call) can write the savings ledger, cache, shadow store, or exemplars — and the real ledger is never even opened. This is the SINGLE place the nil-store invariant is constructed; NewRecordlessOffload and the agent-trajectory flywheel (agent-trajectory-label) both use it so it can't drift.

func (*Pipeline) Cfg

func (p *Pipeline) Cfg() config.Config

Cfg exposes the loaded config so callers like the MCP server can build side-channel tools (e.g. the explicit NIM remote tool) from the same configuration without re-loading it. It returns a shallow copy — read-only use only (the slice/map fields share backing with the live config).

func (*Pipeline) Run

func (p *Pipeline) Run(ctx context.Context, req core.Request) core.Result

Run executes req through the Gemma-4 family cascade and always returns a Result (success or structured defer). Fast tasks (triage/classify) enter at the small tier; on a quality failure the request climbs to the next-larger local model before ever deferring to Opus. Infra errors do not escalate.

func (*Pipeline) RunTier

func (p *Pipeline) RunTier(ctx context.Context, req core.Request, model string) (core.Result, bool)

RunTier runs req through exactly the named tier (bypassing modelChain), with the full quality gate (grammar/verify/validate/ground/confidence) that attempt applies. It records NOTHING to the savings ledger — used by the offline shadow-labeling flywheel to evaluate a counterfactual tier without polluting the savings stats. Returns the tier's result and whether it was accepted.

func (*Pipeline) StartReloader

func (p *Pipeline) StartReloader(interval time.Duration) (stop func())

StartReloader launches a background goroutine that polls the watched artifacts every interval and hot-swaps any that changed (via reloadOnce). It returns a stop func that halts the goroutine and blocks until it has exited. It is started ONLY from runMCP (never from New()), so CLI one-shots never leak a goroutine and stay byte-identical. The tick body recovers from any loader panic so a corrupt artifact can never kill the long-running server (P5).