shadow

Documentation

Overview

Package shadow holds the capture-queue for the nightly shadow-labeling flywheel. At request time a small sample of offload calls is appended here (input + the entry tier's output); the nightly drain replays a counterfactual tier and writes training labels. The queue is purged on drain.

Index

• func Enqueue(path string, it Item) error
• func LabelQueue(ctx context.Context, items []Item, cap int, d LabelDeps) (written int)
• type Item
  • func Drain(path string) ([]Item, error)
• type LabelDeps

Functions

func Enqueue

func Enqueue(path string, it Item) error

Enqueue appends one item as a JSON line (concurrency-safe).

func LabelQueue

func LabelQueue(ctx context.Context, items []Item, cap int, d LabelDeps) (written int)

LabelQueue drains up to cap items from the shadow queue, runs a counterfactual escalation tier on each, judges the result, and appends a confhead label row. Fail-safe: a single item error is skipped, never aborts the batch. Returns the count of labels successfully written.

Types

type Item

type Item struct {
	TS          int64              `json:"ts"`
	Task        string             `json:"task"`
	Input       string             `json:"input"`
	Params      map[string]any     `json:"params,omitempty"`
	EntryTier   string             `json:"entry_tier"`
	EntryOutput string             `json:"entry_output"`
	Feat        map[string]float64 `json:"feat"`
}

func Drain

func Drain(path string) ([]Item, error)

Drain atomically claims the queue (rename aside), reads it, and deletes the claimed file. The rename is the cross-process cursor: any Enqueue after the claim writes to a fresh queue file and is processed by the next Drain, so an item is never lost to a read/truncate race even across processes. A missing or empty queue returns (nil, nil); a corrupt line is skipped; on a leftover claim from a crashed prior drain, that claim is recovered first.

type LabelDeps

type LabelDeps struct {
	// Escalation is the model alias to run each item through (counterfactual tier).
	Escalation string
	// E2B is the entry-tier E2B model alias ("gemma4-e2b"). Used for B1 router labels.
	E2B string
	// RunTier runs req on the named model and returns (result, ok).
	// It must NOT write to the savings ledger or shadow queue (record=false path).
	RunTier func(ctx context.Context, req core.Request, model string) (core.Result, bool)
	// AnswersAgree judges whether the entry-tier candidate and the escalation
	// tier's output agree on the class field. ok=false means un-judgeable (skip).
	// candidate is the raw EntryOutput string; finalData is the escalation result's Data bytes.
	AnswersAgree func(task string, candidate string, finalData []byte) (bool, bool)
	// Ground checks whether the escalation result's structured output is grounded
	// in the source input. Used for extract tasks. ok=false means not applicable.
	Ground func(task core.TaskType, input string, data []byte) (grounded bool, ok bool)
	// Similar computes the semantic similarity between two strings (0..1).
	// Used for the B2 summarize judge. Injected from judge.Embedder.Similar.
	Similar func(a, b string) (float64, error)
	// SummarizeSimThreshold is the minimum similarity to count as "agreed".
	// Caller sets it; default is 0.80.
	SummarizeSimThreshold float64
	// AppendLabel appends a label entry to the confhead sidecar at path.
	// Must NOT touch the main ledger.
	AppendLabel func(path string, e ledger.Entry) error
	// LabelsPath is the confhead sidecar file (confhead-labels.jsonl).
	LabelsPath string
	// RouterLabelsPath is the router sidecar file for B1 E2B counterfactual labels.
	RouterLabelsPath string
	// Embed returns the embedding vector for an item input (injected from
	// judge.Embedder.Embed). Used to build the kNN entry-tier substrate. nil =
	// kNN substrate building disabled.
	Embed func(text string) ([]float64, error)
	// AppendKNN appends one kNN substrate row (task, vec, E2B-accept). nil =
	// disabled. Must NOT touch the savings ledger.
	AppendKNN func(task string, vec []float64, accept bool) error
}

LabelDeps holds the injected functions for LabelQueue, so the logic is fully unit-testable with fakes and no model or network.