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 ImageBatchItem
- type ImageBatchJob
- type Pipeline
- func (p *Pipeline) Cfg() config.Config
- func (p *Pipeline) FootprintStore() *fleetnode.Footprints
- func (p *Pipeline) Run(ctx context.Context, req core.Request) core.Result
- func (p *Pipeline) RunImageBatch(ctx context.Context, jobs []ImageBatchJob) ([]ImageBatchItem, error)
- func (p *Pipeline) RunTier(ctx context.Context, req core.Request, model string) (core.Result, bool)
- func (p *Pipeline) StartReloader(interval time.Duration) (stop func())
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func AnswersAgree ¶
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 ¶ added in v0.6.1
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 ImageBatchItem ¶ added in v0.21.1
type ImageBatchItem struct {
Out string `json:"out"`
Seed int `json:"seed"`
OK bool `json:"ok"`
Ms int64 `json:"ms"`
Error string `json:"error,omitempty"`
}
ImageBatchItem is the per-job outcome of a batch, in job order.
type ImageBatchJob ¶ added in v0.21.1
type ImageBatchJob struct {
Prompt string `json:"prompt"`
Negative string `json:"negative,omitempty"`
Width int `json:"width,omitempty"`
Height int `json:"height,omitempty"`
Steps int `json:"steps,omitempty"`
Seed int `json:"seed,omitempty"`
Out string `json:"out,omitempty"`
}
ImageBatchJob is one line of a generate-image --batch JSONL: a prompt plus the per-job overridable request params. Out/Seed are filled by normalizeImageBatch when absent (same invariants as the single-render path).
type Pipeline ¶
type Pipeline struct {
// contains filtered or unexported fields
}
func NewRecordlessPipeline ¶ added in v0.6.1
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 ¶ added in v0.6.0
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) FootprintStore ¶ added in v0.22.0
func (p *Pipeline) FootprintStore() *fleetnode.Footprints
FootprintStore returns the lazily-opened shared footprint store (nil when no path resolves). Exported for the fleet-serve/fleet-measure verbs: health advertises its Entries(), fleet-measure prints its on-disk records.
func (*Pipeline) Run ¶
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) RunImageBatch ¶ added in v0.21.1
func (p *Pipeline) RunImageBatch(ctx context.Context, jobs []ImageBatchJob) ([]ImageBatchItem, error)
RunImageBatch renders N prompts through ONE warm ComfyUI session (the checkpoint loads once) while preserving zero-always-warm AT THE BATCH BOUNDARY: the render script's single teardown + gpugen's deferred /free restore a clean GPU when the batch ends, however it ends. Ledger: one entry per job, same model label as the single-render path (health tiers must not fragment).
func (*Pipeline) RunTier ¶
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 ¶ added in v0.5.0
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).
Source Files