Documentation
¶
Overview ¶
Package prioritize keeps the score cache warm: when unscored backlog items sit in a repo's working set, ONE batch-scorer task (executor type "prioritize") is minted per repo; when that task completes, its verdicts are cached (priority_scores) and applied to the matching PENDING tasks as task.reprioritized facts. The agent only contributes scores; everything downstream (batching, dedup, precedence, protection) is mechanical.
Triggers are journal-driven (cursor "prioritize-sweeper", head-bootstrapped like every sweeper): a harvest enqueue means new unscored work; a scorer completion means verdicts to apply — and, right after applying, a re-check for items that entered the working set while the batch was in flight. The first sweep (BootMint) also mints for pre-cursor pending items, so enabling the sweeper on a pool with a standing backlog scores it without waiting for a fresh enqueue.
Loop safety is structural: batch tasks are type "prioritize", never "agent", so they never trigger mints themselves; a batch's dedup key is the hash of its exact key set, so an unchanged working set never re-mints (a dead batch is retried only when the set changes — the DLQ is the human surface until then); and every mint pass runs under the pool's budget guard like every other enqueue.
Index ¶
Constants ¶
const ( // ConsumerKey is the sweeper's identity in the watermarks table: the // persisted cursor shared by every sweeper instance over the same // database, rewindable with `tq watermarks set prioritize-sweeper SEQ` // (replay is idempotent: cache upserts and same-value priority updates // write nothing new). ConsumerKey = "prioritize-sweeper" // SourceScorer stamps cached verdicts (queue.PriorityScore.Source): // the batch scorer's identity. When SweeperConfig.Model is set the // source gains a ":<model>" suffix (see scorerSource), so a verdict // scored by a different model is treated as stale and re-scored — a // provider retraining behind a fixed name, or a model swap, must not // rank fresh work against a stale model's verdict (turnstone lesson 14). SourceScorer = "ai:batch-scorer" // PrioritySourceAI is the task.reprioritized source label for applied // verdicts (ADR-0015 §3 ladder: ai). PrioritySourceAI = "ai" // DefaultScoreTTL is the default age at which a cached verdict goes // stale (SweeperConfig.ScoreTTL): a month-old score of an unchanged // item may no longer describe the repo, so the item re-joins the next // batch instead of the verdict living forever. DefaultScoreTTL = 30 * 24 * time.Hour )
Variables ¶
This section is empty.
Functions ¶
func DedupKey ¶
DedupKey is the dedup identity of one repo's batch over one exact key set: "prioritize:<repo>:<hash>". A set that changes (new item, item reworded and re-keyed) mints a fresh batch; an unchanged set never mints twice — even after the first batch died, whose recovery then rides the next set change (the DLQ is the human surface until then).
Types ¶
type SweepStats ¶
type SweepStats struct {
// Facts is the number of new facts consumed.
Facts int
// BatchesEnqueued counts fresh scorer tasks this sweep created.
BatchesEnqueued int
// BatchesKnown counts mints whose batch task already existed (dedup
// hit — the batch ran, or is in flight, elsewhere).
BatchesKnown int
// VerdictsCached counts verdicts written to the score cache.
VerdictsCached int
// TasksReprioritized counts PENDING tasks whose priority changed by an
// applied verdict.
TasksReprioritized int
// Skipped counts facts that could not yield work (vanished records,
// foreign payload shapes, unparsable results).
Skipped int
// CachePruned counts stale cache entries the prune pass evicted:
// verdicts whose item key no longer belongs to any PENDING backlog
// task (the item was reworded and re-keyed, checked off, or its task
// reached a terminal state).
CachePruned int
}
SweepStats summarizes one sweep pass.
type Sweeper ¶
type Sweeper struct {
// contains filtered or unexported fields
}
Sweeper turns the journal's enqueue and scorer-completion facts into batch-scorer mints and score applications. It is safe for concurrent use (the mutex serializes the first-sweep boot mint; the cursor serializes the page pump).
func NewSweeper ¶
NewSweeper returns a sweeper over store. The cursor resumes from the persisted checkpoint; a first run bootstraps at the journal head (same semantics as the review and dlqfix sweepers).
func (*Sweeper) Sweep ¶
func (s *Sweeper) Sweep(ctx context.Context) (SweepStats, error)
Sweep consumes new facts since the last pass: harvest enqueues trigger a mint check for their repo, scorer completions apply their verdicts. Idempotent by dedup keys, cache upserts and same-value transition guards — a replayed page (crash between consumption and checkpoint) re-hits dedup instead of duplicating work.
type SweeperConfig ¶
type SweeperConfig struct {
// Model, when set, overrides the scorer model in every batch payload.
// Empty (the recommended default) leaves model+effort to each repo's
// crush config.
Model string
// Yolo marks the scorer autonomous (same contract as agent payloads).
Yolo bool
// AllowDirty lets the scorer start on a dirty tree. Default false: the
// scorer reads the repo for context and should see committed state.
AllowDirty bool
// BootMint makes the FIRST sweep also mint for the pre-cursor working
// set (items already pending when the sweeper was enabled), instead of
// waiting for the next enqueue to trigger.
BootMint bool
// Log receives one summary line per mint/apply. nil logs nothing.
Log *slog.Logger
// PageSize bounds one fact-stream page; 0 selects the default.
PageSize int
// ScoreTTL ages cached verdicts: an item whose verdict is older than
// this re-joins the next batch and is re-scored (the cache upsert
// overwrites in place). Zero selects DefaultScoreTTL; a negative
// value disables aging — verdicts never go stale and only the prune
// pass evicts entries.
ScoreTTL time.Duration
}
SweeperConfig controls one Sweeper.
Source Files