Documentation
¶
Overview ¶
* ChatCLI - Lesson Queue: idempotency key derivation. * * Idempotency is enforced via a content-addressable JobID. Re- * enqueueing the same (task, trigger, attempt) combination yields * the same JobID and is treated as a duplicate by the queue. * * The hash inputs are normalized (trimmed + lowercase for free-text * fields) so minor whitespace churn between triggers doesn't inflate * the DLQ with near-duplicates.
* ChatCLI - Lesson Queue: Dead Letter Queue operations. * * The DLQ is its own WAL directory. Entries live there until an * operator explicitly purges them or replays them back to the * active queue via the /reflect slash commands. * * Layout: the DLQ shares the WAL file format with the active queue * so the same reader code works for both. The only difference is * that workers never pull from the DLQ — it's read-only to the * process, mutated only by DLQ operations in this file.
* ChatCLI - Lesson Queue: Prometheus metrics. * * Follows the existing pattern from chatcli's metrics/ package: a * struct with per-metric fields registered once against the shared * Registry. A singleton holder is used so both the Runner and the * ReflexionHook can emit without passing plumbing around. * * Registration is once-per-process, guarded by sync.Once. Re-creating * a Runner (tests, hot reload) reuses the same metric collectors — * registering the same CounterVec twice would panic otherwise.
* ChatCLI - Lesson Queue: default Processor adapter. * * Converts a quality.LessonLLM + quality.PersistLessonFunc pair into * a Processor the Runner can call. Classifies errors: * * - ctx cancellation / timeout → Transient (provider likely OK) * - LLM returned skip sentinel → Skipped (valid terminal state) * - LLM returned "no blocks" → Permanent (parser failure) * - LLM network-ish errors → Transient (retry with backoff) * - Persist errors → Transient (disk/memory temp) * * Classification is intentionally tolerant — we prefer a retry over a * DLQ for ambiguous failures because the DLQ needs operator action.
* ChatCLI - Lesson Queue: in-memory bounded queue. * * The in-memory layer is a min-heap keyed on NextAttemptAt so workers * naturally pick up ready-to-retry jobs in scheduled order. A * map[JobID]*heapEntry fronts the heap for O(1) dedupe + removal. * * Invariants: * - Each JobID appears at most once in the heap at any time. * - Removing a job requires invalidating the heap entry (we use the * "lazy deletion" trick: entry.dead=true, Pop skips dead entries). * - The WAL is the source of truth; the heap is a scheduling index * over it. On crash/reboot, Drain rebuilds the heap from the WAL.
* ChatCLI - Lesson Queue: retry policy. * * Computes back-off with full jitter (per AWS Architecture blog, 2015): * uniform random in [delay * (1-frac), delay * (1+frac)] instead of a * flat exponential. This prevents thundering-herd when many lessons * fail simultaneously (e.g. provider 429 storm).
* ChatCLI - Lesson Queue: Runner (public entry point). * * The Runner composes WAL + Queue + DLQ + worker pool + retry policy * into the single object the rest of the codebase interacts with. * ReflexionHook.Enqueue → Runner.Enqueue; /reflect drain → Runner.Drain; * process startup → Runner.Start + Runner.Replay(WAL). * * Ownership: * - Runner is the only component that mutates both the WAL and the * in-memory queue under the durability contract (WAL written * BEFORE queue visible). * - Workers read via Queue.Dequeue and call the injected Processor. * - Runner classifies the ProcessResult and updates WAL/queue/DLQ. * * Lifecycle: * * rnr, _ := lessonq.NewRunner(cfg, metrics, logger) * rnr.Start(ctx, processor) // spins up workers * pending := rnr.Replay(ctx) // drain WAL → queue * rnr.Enqueue(ctx, request) // from the hook * // ... * rnr.Shutdown(30 * time.Second) // on process exit
* ChatCLI - Lesson Queue: core types. * * Package lessonq provides a durable, crash-consistent queue for * Reflexion lesson generation. It sits between the ReflexionHook * (which emits LessonRequests as quality triggers fire) and the * LLM+persistence layer (which materializes and stores Lessons). * * Durability model: * 1. Enqueue writes a WAL record with CRC32 + fsync + atomic rename. * 2. Worker dequeues and processes (LLM call → Lesson → persist). * 3. On success, WAL ACKs the entry (deletion or segment rotation). * 4. On retryable failure, reschedule with exponential backoff + jitter. * 5. On permanent failure or MaxAttempts, move to DLQ (another WAL). * 6. Drain-on-boot scans WAL and re-enqueues pending entries. * * All types here are value types so the package stays easy to mock * and feed in tests. Behavior lives in queue.go, worker.go, wal.go.
* ChatCLI - Lesson Queue: Write-Ahead Log. * * The WAL is the source of durability for the queue. Every accepted * Enqueue synchronously appends a record here before the job is * visible to workers; every ACK (success, skip, or DLQ move) removes * it. This means a process crash at any point leaves the on-disk * state recoverable: * * - Record present, job unprocessed → re-enqueue on boot. * - Record present, job finished → caller forgot to ACK; safe * to re-process because Processor is idempotent per JobID. * - Record absent, job in flight → processing pre-crash, we * lose nothing: the record was already ACKed, which means the * lesson persisted successfully (or was explicitly DLQ-moved). * * File layout: one file per record, named by JobID. This trades * slightly more filesystem overhead for O(1) ACK (single unlink) and * trivial forensics — operators can `ls` the directory and see what's * pending. No background compaction needed. * * Record format (per file): * * [4B magic 'LSN1'][4B length-BE][4B CRC32 of payload][N bytes payload][4B trailing CRC32] * * Double CRC guards against torn writes: if the OS crashes mid-fsync * we might see a truncated file; the trailing CRC ensures we only * accept complete records. * * Atomic append: * 1. Write to <dir>/<id>.wal.tmp.<pid>.<monotonic> * 2. f.Sync() * 3. os.Rename(tmp, <dir>/<id>.wal) [atomic on POSIX] * 4. syncDir(<dir>) [durability of the rename itself] * * The trailing CRC is identical to the leading CRC — we repeat it so * the only way for a reader to see both is a fully-flushed write.
Index ¶
- Variables
- type DLQ
- type JobID
- type LLMCaller
- type LessonJob
- type Metrics
- type OverflowPolicy
- type PersistFn
- type ProcessOutcome
- type ProcessResult
- type Processor
- type Queue
- type RetryPolicy
- type Runner
- func (r *Runner) DLQCount() int
- func (r *Runner) DLQList() ([]LessonJob, error)
- func (r *Runner) DLQPurge(id JobID) error
- func (r *Runner) DLQReplay(ctx context.Context, id JobID) error
- func (r *Runner) DrainAndShutdown(timeout time.Duration)
- func (r *Runner) Enqueue(ctx context.Context, req quality.LessonRequest) error
- func (r *Runner) PendingSnapshot() []LessonJob
- func (r *Runner) QueueDepth() int
- func (r *Runner) Replay(ctx context.Context) (int, error)
- func (r *Runner) Start(ctx context.Context, processor Processor) error
- type RunnerConfig
- type WAL
Constants ¶
This section is empty.
Variables ¶
var ErrDuplicate = errors.New("lessonq: duplicate job")
ErrDuplicate is returned when Enqueue is called with a JobID that is already in the queue (or a job with the same key that is in- flight or in the WAL). Caller should treat this as success — the work is already scheduled.
var ErrQueueClosed = errors.New("lessonq: queue closed")
ErrQueueClosed is returned from Enqueue after Close.
var ErrQueueFull = errors.New("lessonq: queue full")
ErrQueueFull is returned when OverflowBlock times out without a slot opening up and OverflowDropOldest is not in effect.
var ErrWALClosed = errors.New("lessonq: WAL closed")
ErrWALClosed is returned from Append after Close has been called.
Functions ¶
This section is empty.
Types ¶
type DLQ ¶
type DLQ struct {
// contains filtered or unexported fields
}
DLQ wraps a WAL-backed dead letter queue. Not safe for concurrent mutation from outside (ops like Replay+Purge are admin-only and rare); internal WAL locking handles the file-side concurrency.
func NewDLQ ¶
NewDLQ opens a DLQ rooted at dir. dir is typically <base>/dlq next to the active WAL. A nil logger is upgraded to a no-op so the caller never has to nil-check.
func (*DLQ) Pop ¶
Pop returns and removes a DLQ entry by ID. Used by /reflect retry, which hands the job back to the active queue. Returns (job, false) if the ID is unknown.
type JobID ¶
type JobID string
JobID uniquely identifies a lesson job across the entire lifecycle (queue, in-flight, DLQ, replay). It is the idempotency key — re- enqueueing the same JobID is a no-op.
JobID is derived from sha256(task | trigger | attempt_hash), truncated to 16 hex chars for log readability. See dedupe.go for the builder.
func DeriveJobID ¶
func DeriveJobID(req quality.LessonRequest) JobID
DeriveJobID returns the 16-hex-char idempotency key for a request.
Inputs:
- task: the original user task (normalized: trimmed + lowercased).
- trigger: "error" | "hallucination" | "low_quality" | "manual".
- attempt: the attempt content (full output, trimmed).
Two triggers with identical normalized inputs produce the same ID — caller relies on this for dedupe. Very short inputs still hash deterministically; no length minimum.
type LLMCaller ¶
LLMCaller mirrors quality.LessonLLM but is re-declared here so the package stays importable without pulling the full quality surface in test paths. Runner.NewProcessor adapts a quality.LessonLLM into this shape; see queue_runner.go.
type LessonJob ¶
type LessonJob struct {
ID JobID
Request quality.LessonRequest
EnqueuedAt time.Time
NextAttemptAt time.Time // scheduling hint; workers honor back-off
Attempts int
LastError string // human-readable last failure, for DLQ inspection
}
LessonJob is the unit of work the queue processes.
Invariants:
- ID is stable across retries (dedupe key).
- Request is immutable after Enqueue — retries reuse it verbatim.
- Attempts tracks how many processing tries have happened, including the current one. 0 before first dequeue, N after Nth failure.
- NextAttemptAt is the earliest time the worker should pick it up. Workers skip jobs whose NextAttemptAt is in the future.
type Metrics ¶
type Metrics struct {
// Enqueue counters, labeled by outcome ("accepted"|"rejected_full"|
// "deduped"|"dropped_oldest").
EnqueueTotal *prometheus.CounterVec
// QueueDepth is the current in-memory queue length. Gauge.
QueueDepth prometheus.Gauge
// ProcessingDuration histograms time from dequeue to outcome.
ProcessingDuration *prometheus.HistogramVec // labels: outcome
// AttemptsTotal counts processing attempts, labeled by outcome.
AttemptsTotal *prometheus.CounterVec
// RetryTotal counts scheduled retries labeled by attempt number.
RetryTotal *prometheus.CounterVec
// DLQSize is the current DLQ length.
DLQSize prometheus.Gauge
// WALCorruption counts torn-write / bad-CRC records detected on
// read. A non-zero value is worth paging on.
WALCorruption prometheus.Counter
// WALSegments gauges the number of active WAL segments on disk.
WALSegments prometheus.Gauge
// StaleDiscarded counts entries dropped at drain time because
// they exceeded StaleAfter.
StaleDiscarded prometheus.Counter
// PersistFailures counts persist callback errors (separate from
// LLM errors — we want to distinguish fs/store vs provider).
PersistFailures prometheus.Counter
}
Metrics holds the Prometheus collectors for the lesson queue subsystem. All metrics use the "chatcli" namespace and "lessonq" subsystem so they group cleanly in dashboards.
func GetMetrics ¶
func GetMetrics() *Metrics
GetMetrics returns the process-wide singleton, registering the collectors on first call. Safe for concurrent use.
type OverflowPolicy ¶
type OverflowPolicy int
OverflowPolicy governs what happens when the bounded queue is full at Enqueue time. Block is the enterprise default (WAL already has the record, caller blocks briefly waiting for a slot); DropOldest is available for throughput-sensitive deployments.
const ( // OverflowBlock waits up to EnqueueTimeout for a slot, then errors. OverflowBlock OverflowPolicy = iota // OverflowDropOldest evicts the oldest in-memory job (WAL-backed so // it'll be picked up on the next drain) to make room. Used for // latency-sensitive callers that never want to block. OverflowDropOldest )
type PersistFn ¶
PersistFn writes a materialized lesson into long-term memory. Mirrors quality.PersistLessonFunc.
type ProcessOutcome ¶
type ProcessOutcome int
ProcessOutcome classifies the result of a single processing attempt. Workers map this to either an ACK (success), a reschedule, or a DLQ move. Mapping lives in Runner.handleOutcome.
const ( // OutcomeSuccess — lesson generated + persisted. ACK and drop. OutcomeSuccess ProcessOutcome = iota // OutcomeSkipped — LLM declared "no actionable lesson". ACK and // drop (this is a valid terminal state, not an error). OutcomeSkipped // OutcomeTransient — processing failed with a retryable error // (LLM 429/503, fs temp error). Reschedule with back-off. OutcomeTransient // OutcomePermanent — processing failed with an unrecoverable error // (parser failure, config error). Move to DLQ immediately. OutcomePermanent )
func (ProcessOutcome) String ¶
func (o ProcessOutcome) String() string
String makes outcomes log-friendly.
type ProcessResult ¶
type ProcessResult struct {
Outcome ProcessOutcome
Err error
}
ProcessResult carries the outcome of a single processing attempt back from the worker function to the Runner. Err is populated for OutcomeTransient and OutcomePermanent; ignored otherwise.
type Processor ¶
type Processor func(ctx context.Context, job LessonJob) ProcessResult
Processor is the callback the Runner invokes per-job. Implementations perform the LLM call, parse the response, persist the lesson, and return a ProcessResult classifying what happened.
The ctx passed here is bounded by a per-job budget (configurable), distinct from the caller's ctx that fired the trigger — reflexion outlives the turn by design.
func NewProcessor ¶
func NewProcessor(llm quality.LessonLLM, persist quality.PersistLessonFunc, metrics *Metrics, logger *zap.Logger) Processor
NewProcessor builds a Processor that generates a lesson via llm and persists it via persist. Either being nil returns a Processor that always classifies jobs as permanent (so they flush to DLQ fast without looping).
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue is a bounded, priority-scheduled (by NextAttemptAt) job queue. Safe for concurrent use.
func NewQueue ¶
func NewQueue(capacity int, policy OverflowPolicy, blockTimeout time.Duration, metrics *Metrics) *Queue
NewQueue builds a queue with the given capacity and overflow policy. capacity ≤ 0 disables the bound (unlimited); used mainly in tests.
func (*Queue) Close ¶
func (q *Queue) Close()
Close signals shutdown. Dequeue-ers wake up and return ctx.Err().
func (*Queue) Dequeue ¶
Dequeue blocks until a job is ready (NextAttemptAt ≤ now), the queue is closed, or ctx is canceled. When a job comes up, the entry is removed from the active set (it's in-flight) and returned with an Ack callback. The caller MUST invoke ack — otherwise the entry is lost from the in-memory view (WAL still has it, so it'd resurface on reboot, but that's a degraded path).
Ack semantics are encoded in ProcessResult: Success/Skipped → WAL delete; Transient → WAL update + re-enqueue with new NextAttemptAt; Permanent → WAL move to DLQ (caller's responsibility, Ack just drops from the active queue).
func (*Queue) Enqueue ¶
Enqueue adds a job. Returns ErrDuplicate if the JobID is already known (call site should treat as success — idempotent re-submit). Returns ErrQueueFull on OverflowBlock timeout; never on DropOldest.
type RetryPolicy ¶
type RetryPolicy struct {
InitialDelay time.Duration // first retry waits this long
MaxDelay time.Duration // cap on the exponential ramp
Multiplier float64 // typically 2.0
JitterFraction float64 // 0.0–0.5; fraction of delay added as uniform jitter
MaxAttempts int // total attempts (1 = no retries)
}
RetryPolicy governs back-off between transient-failure attempts. All fields have safe defaults in DefaultRetryPolicy().
func DefaultRetryPolicy ¶
func DefaultRetryPolicy() RetryPolicy
DefaultRetryPolicy returns production-ready defaults: 5 attempts, 1s → 5min ramp, 20% jitter.
func (RetryPolicy) NextDelay ¶
NextDelay returns the backoff for the Nth attempt (1-indexed). Attempt 1 maps to InitialDelay (with jitter); each subsequent attempt multiplies by Multiplier, capped at MaxDelay.
The rng parameter is injectable so tests can make jitter deterministic. Pass nil to use the global rand.
func (RetryPolicy) ShouldRetry ¶
func (p RetryPolicy) ShouldRetry(attempts int) bool
ShouldRetry reports whether another attempt is allowed given the total attempts already made (the attempt that just failed counts).
type Runner ¶
type Runner struct {
// contains filtered or unexported fields
}
Runner is the composed durable-queue engine.
func NewRunner ¶
func NewRunner(cfg RunnerConfig, logger *zap.Logger) (*Runner, error)
NewRunner builds a Runner and opens its WAL + DLQ. Does NOT start workers — call Start + Replay separately.
func (*Runner) DLQReplay ¶
DLQReplay moves a DLQ entry back to the active queue (for /reflect retry <id>). Resets Attempts to 0 so the retry policy starts fresh.
func (*Runner) DrainAndShutdown ¶
DrainAndShutdown signals workers to stop, waits up to timeout for in-flight jobs to finish, closes the queue and WAL. Any jobs still queued are left in the WAL and will be replayed on next boot.
func (*Runner) Enqueue ¶
Enqueue accepts a LessonRequest: derives its JobID, writes the WAL record durably, and hands the job to the in-memory queue. Returns nil on accept (including the dedup case — idempotent).
func (*Runner) PendingSnapshot ¶
PendingSnapshot returns a copy of queued jobs (for debugging / /reflect listing — keeps internals private).
func (*Runner) QueueDepth ¶
QueueDepth returns current in-memory queue depth.
type RunnerConfig ¶
type RunnerConfig struct {
// BaseDir is where the WAL and DLQ subdirs live. Typically
// <workspaceDir>/.chatcli/reflexion.
BaseDir string
// Workers is the number of goroutines concurrently processing
// jobs. Default 2 — lesson gen is I/O-bound on the LLM call, so
// parallelism beyond 2-3 buys little.
Workers int
// QueueCapacity is the max in-memory queue depth. Enqueue honors
// OverflowPolicy when full. Default 1000.
QueueCapacity int
// OverflowPolicy: Block | DropOldest. Default Block.
OverflowPolicy OverflowPolicy
// EnqueueBlockTimeout is how long Enqueue will wait on a full
// queue before returning ErrQueueFull. Default 5s.
EnqueueBlockTimeout time.Duration
// Retry controls per-job back-off + max attempts.
Retry RetryPolicy
// PerJobTimeout bounds a single Processor invocation. Default
// 2 minutes — lesson generation is a short LLM call but we
// accept slow providers.
PerJobTimeout time.Duration
// StaleAfter discards Replay entries older than this at drain
// time. Default 7 days — lessons about stale task contexts are
// usually not useful. Set to 0 to disable.
StaleAfter time.Duration
}
RunnerConfig bundles all the knobs the Runner needs. Defaults (DefaultRunnerConfig) give a production-safe baseline.
func DefaultRunnerConfig ¶
func DefaultRunnerConfig() RunnerConfig
DefaultRunnerConfig returns production defaults. BaseDir is empty and must be set by the caller.
type WAL ¶
type WAL struct {
// contains filtered or unexported fields
}
WAL is the append-only log for lesson jobs. A single WAL instance owns one directory and is safe for concurrent Append/Ack/List.
func NewWAL ¶
NewWAL opens (and creates if needed) a WAL in dir. metrics may be nil for tests; logger nil is upgraded to a no-op.
func (*WAL) Ack ¶
Ack removes the record for id. Missing records return nil (idempotent — tests and retries may ACK the same id twice).
func (*WAL) Append ¶
Append writes a job record durably and returns nil iff the record is safely on disk. Concurrency-safe. See AppendNew for a variant that signals whether the record was already present.
func (*WAL) AppendNew ¶
AppendNew is like Append but returns (true, nil) when a new record was created and (false, nil) when an idempotent no-op was taken (record already present). Callers that need to distinguish first- write from retry (e.g. the Runner, which only pushes new records onto the in-memory queue) use this variant to avoid double- processing jobs that are already in flight.
func (*WAL) Close ¶
func (w *WAL) Close()
Close marks the WAL closed. Subsequent Appends fail with ErrWALClosed. Existing files are untouched — reopen the WAL with NewWAL pointing at the same dir to resume.
func (*WAL) Count ¶
Count returns how many valid .wal files live in the dir without parsing them. Used by metrics to keep a gauge accurate without re- decoding every record.
func (*WAL) Dir ¶
Dir returns the directory backing this WAL. Useful for tests and for the DLQ (which shares the same layout with a different dir).
func (*WAL) List ¶
List scans the WAL directory and returns every valid record. Corrupt records are logged (and incremented against WALCorruption) but do not abort the scan — one bad record must never lock out the rest.
Returned jobs are sorted by EnqueuedAt ascending so drain processes oldest first.
Source Files