Documentation
¶
Overview ¶
Package dlqfix closes the loop around dead-lettered agent work: every dead-lettered agent task gets ONE autopsy by a second agent, and the autopsy's verdict is executed mechanically — "fixed" rescues the original with a fresh attempt budget, "wontfix" dismisses it with the recorded reason. The agent only contributes the diagnosis; everything downstream (disposition, dedup, loop guarding) is mechanical.
Loop safety is structural: autopsies are only ever minted for dead "agent" tasks (a dead autopsy can never mint another autopsy), each dead task is autopsied at most once (the dlqfix:<task-id> dedup key, forever), and every mint pass runs under the pool's budget guard, which counts autopsy tasks like every other enqueue.
Index ¶
Constants ¶
const ( // ConsumerKey is the sweeper's identity in the watermarks table: the // persisted cursor shared by every sweeper over the same database, so // tasks that dead-lettered while no sweeper was running still get // their autopsy when the next pool starts. ConsumerKey = "dlqfix-sweeper" // DismissedBySweeper names the sweeper in a dismissed task's // task.cancelled fact detail (operators dismiss as "operator"). DismissedBySweeper = "dlqfix-sweeper" )
Variables ¶
This section is empty.
Functions ¶
func DedupKey ¶
DedupKey is the dedup identity of the autopsy of one dead task: exactly one autopsy per dead task, forever. A rescued task that dies again does NOT get a second autopsy — the second death is deliberately a human surface (the first verdict and the fresh failure evidence are both in the journal).
Types ¶
type SweepStats ¶
type SweepStats struct {
// Facts is the number of new facts consumed.
Facts int
// FixesEnqueued counts fresh autopsy tasks this sweep created.
FixesEnqueued int
// FixesKnown counts dead tasks whose autopsy task already existed
// (dedup hit — the autopsy ran, or is pending, elsewhere).
FixesKnown int
// Rescued counts dead tasks re-queued by a "fixed" verdict.
Rescued int
// Dismissed counts dead tasks cancelled by a "wontfix" verdict.
Dismissed int
// Skipped counts facts that could not yield work (vanished records,
// foreign payload shapes, dispositions already executed elsewhere).
Skipped int
}
SweepStats summarizes one sweep pass.
type Sweeper ¶
type Sweeper struct {
// contains filtered or unexported fields
}
Sweeper turns the journal's dead-letter and autopsy-completion facts into DLQ dispositions. It is safe for concurrent use (a mutex serializes sweeps; ticks and the --once drain watcher may call it from different goroutines).
func NewSweeper ¶
NewSweeper returns a sweeper over store. The cursor resumes from the persisted checkpoint — deaths recorded while no sweeper was running are autopsied on the next start. A first run bootstraps at the journal head: dead letters that predate the sweeper are not replayed (same semantics as the review sweeper; rewind with `tq watermarks set dlqfix-sweeper SEQ`).
func (*Sweeper) Sweep ¶
func (s *Sweeper) Sweep(ctx context.Context) (SweepStats, error)
Sweep consumes new facts since the last pass and enqueues autopsy (and executes disposition) work. Idempotent by dedup and transition guards: calling it twice never duplicates work — a replayed page (crash between consumption and checkpoint) re-hits dedup keys and idempotent store transitions instead of minting duplicates or double-disposing. The cursor checkpoints after each page; a failed checkpoint stops the sweep — facts are never consumed past an unpersisted cursor.
type SweeperConfig ¶
type SweeperConfig struct {
// Model, when set, is written into every autopsy payload. Empty (the
// default) leaves model choice to each repo's crush config.
Model string
// Log receives one summary line per mint/disposition. nil logs nothing.
Log *slog.Logger
// PageSize bounds one fact-stream page; 0 selects the default.
PageSize int
}
SweeperConfig controls one Sweeper.
Source Files