Documentation
¶
Overview ¶
Package harvest turns repository backlogs into queue tasks.
It is the ingestion half of a self-managing agent pool: every repo's TODO_LIST.md is the shared, human-readable work backlog; the harvester repeatedly scans it and enqueues one agent task per open item, paced to at most one in-flight task per repo. Agents (see executor.AgentExecutor) do the item and mark it done in the same file, which closes the loop: done items are never re-enqueued, edited items become new work.
Index ¶
- Constants
- Variables
- func DiscoverRepos(dir, todoFile string) ([]string, error)
- func DiscoverReposDaemon(ctx context.Context, addr, projectsDir, todoFile string) ([]string, error)
- func DiscoverReposFor(ctx context.Context, addr, projectsDir, todoFile string, log *slog.Logger) ([]string, error)
- func ItemKey(repoName, text string) string
- func KeywordBump(text string) int
- func MarkerPriority(level int) int
- func ReadImportance(repo string) (int, error)
- func RepriMutable(current int) bool
- func SplitMarker(text string) (int, string, bool)
- type Config
- type Drift
- type DriftKind
- type DriftResult
- type Enqueued
- type Harvester
- type Item
- type PayloadItem
- type PrioritySource
- type PruneResult
- type PruneWhy
- type PrunedTask
- type RepriChange
- type ResolveInput
- type Result
- type ScanFailure
- type Skipped
- type WatchConfig
- type Watcher
Constants ¶
const CatchupKeyPrefix = "catchup:"
CatchupKeyPrefix marks a catch-up task's dedup key. A catch-up task asks an agent to tick a checkbox whose work is already done; the prefix keeps it distinct from the original item's key, so it can never mask the drift it exists to close.
const DefaultCatchupPrompt = `` /* 982-byte string literal not displayed */
DefaultCatchupPrompt is the contract for catch-up tasks: close the loop in the file, change nothing else.
const DefaultDiscoveryTimeout = 5 * time.Second
DefaultDiscoveryTimeout bounds one daemon discovery RPC: an unreachable or wedged daemon must never stall a harvest tick (the local scan is the fallback and costs milliseconds).
const DefaultImportance = 50
DefaultImportance is the effective importance of a repo without a .config/metadata.yaml (project-meta file contract, ADR-0015 §7).
const DefaultMaxPerTick = 10
DefaultMaxPerTick bounds how many new agent tasks one harvest run may enqueue across all repos — the cost throttle for unattended pools.
const DefaultPromptTemplate = `` /* 1223-byte string literal not displayed */
DefaultPromptTemplate is the agent contract handed to Crush for each item. Placeholders: {{REPO_ABS}}, {{REPO}}, {{HEADING}}, {{ITEM}}, and {{TASK_ID}} (substituted at EXECUTION time by the agent executor — the queue task ID does not exist when the harvester renders this template; it exists so commits can carry a Task-Queue-ID footer for git log ↔ tq facts cross-reference).
const DefaultTodoFile = "TODO_LIST.md"
DefaultTodoFile is the backlog file scanned in each repository.
const DefaultType = "agent"
DefaultType is the task type harvest enqueues: the headless agent executor (executor.AgentExecutor, registered as "agent" by the tq CLI).
const MaxMarkerLevel = 4
MaxMarkerLevel is the highest valid marker level (P4).
const ReasonScanFailed = "scan failed"
ReasonScanFailed prefixes every repo-level scan error in Skipped.Reason (the detail after the colon is the underlying error).
Variables ¶
var ErrNoRepos = errors.New("harvest: no repos and no projects dir configured")
ErrNoRepos is returned when neither --repos nor a --projects-dir is configured: there is nothing to harvest. One sentinel (previously the same message was inlined in Run and Audit independently, so callers could not match it).
Functions ¶
func DiscoverRepos ¶
DiscoverRepos returns the depth-1 subdirectories of dir that contain the todo file, sorted by name.
func DiscoverReposDaemon ¶ added in v0.2.0
DiscoverReposDaemon enumerates candidate repos on a project-discovery-daemon (POST /v1/discover over the HTTP endpoint at addr, usually a unix socket path such as /run/project-discovery/daemon.sock) and maps the result to harvestable repos — absolute project paths that contain the todo file, deduplicated and sorted, exactly the contract DiscoverRepos upholds for the local scan. Addr forms: /path/to/sock and unix:///path/to/sock dial a unix socket, anything else is treated as a host:port TCP address (what httptest servers and networked daemons use).
func DiscoverReposFor ¶ added in v0.2.0
func DiscoverReposFor(ctx context.Context, addr, projectsDir, todoFile string, log *slog.Logger) ([]string, error)
DiscoverReposFor resolves the candidate repos for one harvest tick: with a daemon addr it asks the daemon and, when that fails (socket unreachable, non-200, bad payload), logs ONE warning and falls back to the local scan — daemon mode is additive and must never fail the tick. An empty addr is the plain scan (the zero-external-services default).
func ItemKey ¶
ItemKey derives the stable dedup key for an item: the hash of repo name and item text. Item text is whitespace-collapsed so reflowing a line does not create duplicate work.
func KeywordBump ¶ added in v0.3.0
KeywordBump reports the strongest keyword bump for an item text, 0 when none matches. Case-insensitive.
func MarkerPriority ¶ added in v0.3.0
MarkerPriority returns the backlog priority of a marker level — P1 through P4. 0 ("no marker") maps to 0.
func ReadImportance ¶ added in v0.3.0
ReadImportance reads a repo's importance from its .config/metadata.yaml (the project-meta flat file contract). Absent file: DefaultImportance. Present but malformed (unparseable importance, or out of 0–100): an error the caller reports as a repo skip reason — an importance the owner DID set must never silently degrade to a default.
This is a deliberate strict-subset reader: project-meta is proprietary (ADR-0015 §7), so tq parses the two bytes it needs instead of requiring the module. Only a top-level `importance:` key counts; indented keys belong to nested mappings and are ignored.
func RepriMutable ¶ added in v0.3.0
RepriMutable reports whether a re-resolution pass (tq reprioritize, the prioritize sweeper) may rewrite a task currently stored at current: hot (100–149) and machine (150+) priorities are explicit human/flag acts and are never overwritten by automated re-resolution — the band protection half of the ADR-0015 §3 precedence.
func SplitMarker ¶ added in v0.3.0
SplitMarker splits a trailing priority marker off an item line, returning the marker level (1–4) and the marker-free text. ok is false when the text carries no marker — including a marker-only line (there is no item left to prioritize).
Types ¶
type Config ¶
type Config struct {
// ProjectsDir is scanned (depth 1) for repos containing TodoFile.
ProjectsDir string
// DiscoveryAddr points at a project-discovery-daemon endpoint (unix
// socket path, unix:// prefixed, or host:port) used INSTEAD of the
// depth-1 ProjectsDir scan to enumerate candidate repos. Additive by
// contract: an unreachable daemon logs one warning per tick and the
// scan takes over. Empty keeps the zero-external-services default.
DiscoveryAddr string
// Log receives the daemon-discovery degradation warning; nil discards.
Log *slog.Logger
// Repos lists explicit repo directories; when set, ProjectsDir is ignored.
Repos []string
// Type is the task type enqueued (must match a registered executor).
Type string
// TodoFile is the backlog file name inside each repo.
TodoFile string
// MaxPerTick caps new enqueues per run across all repos.
MaxPerTick int
// MaxAttempts is the attempt budget for harvested tasks (0 = default).
MaxAttempts int
// Priority for harvested tasks.
Priority int
// SameSessionPriority promotes items whose text references /tmp paths
// to this priority at enqueue time ("hot": work the same session, the
// files will not survive). Zero disables the promotion.
SameSessionPriority int
// UseImportance resolves harvested priorities from the repo's
// .config/metadata.yaml importance (default 50) plus keyword bumps,
// clamped to the backlog band (ADR-0015 §3). Markers and hot promotion
// apply regardless of this flag; a malformed metadata file skips the
// repo's items with a reason instead of guessing.
UseImportance bool
// MaxPendingPerRepo caps how many PENDING tasks one repo may hold in
// the queue (0 = unlimited): the queue is the WORKING SET, TODO_LIST.md
// is the warehouse — prompts stay fresh and AI cost stays O(working
// set). Admission resumes as claims drain the slot; running tasks do
// not count (the repo-busy rule owns those).
MaxPendingPerRepo int
// PromptTemplate overrides DefaultPromptTemplate.
PromptTemplate string
// Model overrides the crush model ("provider/model") in every harvested
// agent payload. Empty = the agent binary's default model.
Model string
// RepoIntervals sets a minimum gap between new enqueues per repo (by
// repo name): after any enqueue for a repo, further items wait until
// the interval has passed. Empty or zero entries = only the default
// one-per-tick pacing applies.
RepoIntervals map[string]time.Duration
// RepoTimeouts sets a per-repo agent-task timeout ladder (by repo
// name), pinned into each harvested payload's TimeoutMinutes: big
// repos get long ceilings, quick ones stay tight. Repos without an
// entry keep the executor's 30-minute default.
RepoTimeouts map[string]time.Duration
// DLQBackoff pauses harvesting of a repo whose recent work all went to
// the dead-letter queue (dead tasks present, none completed, newest
// dead within the window): the repo is poisoned until a human fixes or
// rescues it. Zero disables the guard. Default 0 (set it, e.g. 30m).
DLQBackoff time.Duration
// RequireClean passes the clean-tree policy through to agent payloads:
// nil = executor default (require a clean git tree), false lets agents
// run on dirty repos (scratch/fixtures only), true forces the check.
RequireClean *bool
// DryRun reports what a real run would enqueue, without writing to the
// queue. Result.Enqueued entries then carry an empty TaskID.
DryRun bool
}
Config controls one harvester.
type DriftKind ¶ added in v0.2.0
type DriftKind string
DriftKind names which side of the file-vs-queue contract broke.
const ( // DriftStaleOpen: the queue says the work completed, the checkbox is // still open — the loop was never closed in the file (or was reverted). DriftStaleOpen DriftKind = "stale-open" // DriftStaleDone: the checkbox is ticked but the task never completed // (pending, running or dead) — someone did the work by hand or ticked // prematurely; the queue task is now pointless. Report-only: cancelling // human-visible work is an operator decision (`tq cancel`). DriftStaleDone DriftKind = "stale-done" )
type DriftResult ¶ added in v0.2.0
type DriftResult struct {
Repos int
// StaleOpen lists completed-work-open-checkbox drift; with DryRun unset,
// each entry here also got a catch-up task enqueued (see Enqueued).
StaleOpen []Drift
// StaleDone lists ticked-checkbox-unfinished-task drift (report-only).
StaleDone []Drift
// Enqueued lists the catch-up tasks actually created this pass.
Enqueued []Enqueued
// ScanFailures lists repos that could not be read during the audit;
// the audit keeps going past them.
ScanFailures []ScanFailure
}
DriftResult summarizes one audit pass over all repos.
type Enqueued ¶
type Enqueued struct {
Item Item
TaskID task.ID
Fresh bool // true when this run actually created the task
Hot bool // text referenced /tmp: enqueued at SameSessionPriority
}
Enqueued records a task created (or already present) for an item.
type Harvester ¶
type Harvester struct {
// contains filtered or unexported fields
}
Harvester scans repos and enqueues agent tasks for open todo items.
func (*Harvester) Audit ¶ added in v0.2.0
func (h *Harvester) Audit(ctx context.Context) (DriftResult, error)
Audit compares every repo's checkbox states against the queue's terminal task states and repairs one direction of drift: when a task COMPLETED but its checkbox is still open, a catch-up task (dedup-keyed CatchupKeyPrefix + item key) is enqueued to close the loop in the file. The store's unique dedup index makes that enqueue-once, so a repeating audit never piles up duplicate catch-ups even if the first one also fails to tick the box. The opposite direction (ticked checkbox, unfinished task) is reported only. DryRun reports everything without enqueuing.
func (*Harvester) PruneStale ¶ added in v0.2.0
func (h *Harvester) PruneStale(ctx context.Context) (PruneResult, error)
PruneStale cancels pending queue tasks whose TODO_LIST item is now `[x]` (dedup-key match), or whose item text is gone from the file entirely, so a pool relaunch never inherits zombies: harvesting while no pool runs creates pending tasks, and ticking — or deleting — an item later never withdraws them (21:40 report §d1/§e1 — six stale tasks sat 5–24 h). Only PENDING tasks are cancelled; running ones are reported (a cooperative stop mid-execution is an operator decision), dead ones are terminal. Only tasks carrying a harvest dedup key in their payload are ever touched — external work is invisible to the sweep. DryRun reports without cancelling.
func (*Harvester) Reprioritize ¶ added in v0.3.0
Reprioritize re-resolves the priorities of PENDING tasks from the CURRENT truth in the todo files (markers) and, with UseImportance, each repo's metadata importance — so an owner's marker edit or importance change reaches tasks already sitting in the queue (ADR-0015 §5).
Rules: only PENDING tasks are touched; hot/machine tasks are protected (RepriMutable); same-value writes are no-ops (the store appends no fact); dryRun reports without writing. Per-repo scan failures are returned in failures, never as errors — a broken repo must not block the others.
type Item ¶
type Item struct {
Repo string // absolute repo path
RepoName string // project name (repo dir base name)
Heading string // nearest markdown heading above the item
Text string // the checkbox text, trimmed, priority marker stripped
Key string // stable dedup key: hash of repo name + MARKER-STRIPPED text
Done bool // true when the checkbox is ticked ([x])
// MarkerLevel is the parsed trailing `— P[1-4]` marker (1–4); 0 = none.
// Stripped from Text before hashing so editing a marker never forks a
// task (ADR-0015 §2). omitempty keeps legacy JSON consumers and golden
// fixtures byte-stable for unmarked items.
MarkerLevel int `json:"markerLevel,omitempty"`
}
Item is one checkbox found in a repo's todo file.
func ParseRepo ¶
ParseRepo extracts the open checkbox items from a repo's todo file.
Semantics: a line matching `- [ ]`/`* [ ]` (leading whitespace allowed) is an item; `- [x]`/`- [X]` are done and ignored (ParseRepoAll returns them with Done set). The nearest heading above the item (any level, text without the #s) is its Heading. Lines inside fenced code blocks are ignored, so TODO examples in documentation are never harvested. The Key changes when the item text changes, so editing an item re-arms it even if a previous task for the old wording exists.
func ParseRepoAll ¶ added in v0.2.0
ParseRepoAll returns every checkbox item (open and done) from a repo's todo file; Item.Done distinguishes them. It backs the drift auditor, which compares BOTH checkbox states against terminal task states.
type PayloadItem ¶ added in v0.3.0
type PayloadItem struct {
// Repo is the payload's AgentPayload.Repo — the resolution-ready
// reference the agent executor itself uses.
Repo string
// Text is the marker-stripped item text.
Text string
// Key is the item's dedup key (ItemKey derivation) — the score-cache
// key.
Key string
// MarkerLevel is 0 (no marker) or 1-4 when the item carried a
// `— P[1-4]` marker at harvest time.
MarkerLevel int
}
PayloadItem is the harvested backlog-item identity carried by a task payload: what was harvested, from where, under which dedup key.
func PayloadItemOf ¶ added in v0.3.0
func PayloadItemOf(t task.Task) (PayloadItem, bool)
PayloadItemOf reads one task's payload back into its harvested item identity. ok is false for non-harvest payloads (foreign agent tasks, review/status/dlqfix mints): only harvest-minted tasks carry the "dedup" key.
type PrioritySource ¶ added in v0.3.0
type PrioritySource string
PrioritySource names what produced an effective priority (ADR-0015 §3).
const ( // PrioritySourceHot: same-session promotion. PrioritySourceHot PrioritySource = "hot" // PrioritySourceMarker: trailing `— P[1-4]` human marker. PrioritySourceMarker PrioritySource = "marker" // PrioritySourceAI: cached priority_scores verdict. PrioritySourceAI PrioritySource = "ai" // PrioritySourceKeyword: importance + keyword bump. PrioritySourceKeyword PrioritySource = "keyword" // PrioritySourceImportance: bare repo importance. PrioritySourceImportance PrioritySource = "importance" // PrioritySourceDefault: flat config priority (legacy behavior). PrioritySourceDefault PrioritySource = "default" )
func ResolvePriority ¶ added in v0.3.0
func ResolvePriority(input ResolveInput) (int, PrioritySource)
ResolvePriority applies the precedence ladder (ADR-0015 §3) and returns the effective priority with the source that won. Backlog computations clamp to [0, queue.BacklogMax]; only HotPriority may cross upward.
type PruneResult ¶ added in v0.2.0
type PruneResult struct {
Repos int
// Cancelled lists pending tasks withdrawn because their TODO_LIST item
// is now ticked — the zombies a pool relaunch would otherwise execute —
// or because the item text is gone from the file entirely.
Cancelled []PrunedTask
// Running lists stale-item tasks currently executing: a cooperative
// stop is an operator decision (tq cancel), not a sweep default.
Running []PrunedTask
// Dead lists stale-item tasks already dead-lettered: terminal, listed
// because the dedup key still suppresses re-enqueue if the item is
// ever un-ticked with unchanged text.
Dead []PrunedTask
// ScanFailures lists repos that could not be read; the prune keeps
// going past them (same contract as Audit).
ScanFailures []ScanFailure
}
PruneResult summarizes one stale-task prune pass over all repos.
type PruneWhy ¶ added in v0.2.0
type PruneWhy string
PruneWhy records which stale rule withdrew (or reported) a task.
type PrunedTask ¶ added in v0.2.0
PrunedTask pairs a withdrawn/declined task with the item text that doomed it and the rule that applied. For absent items the text is gone by definition; Item carries the repo and the dedup key instead.
type RepriChange ¶ added in v0.3.0
type RepriChange struct {
TaskID task.ID `json:"task_id"`
ItemText string `json:"item_text"`
OldPriority int `json:"old_priority"`
NewPriority int `json:"new_priority"`
Source PrioritySource `json:"source"`
}
RepriChange records one would-be or applied priority change from a Reprioritize pass.
type ResolveInput ¶ added in v0.3.0
type ResolveInput struct {
Text string
MarkerLevel int
HotPriority int // --same-session-priority; <= 0 disables hot
FlatPriority int // --priority fallback when nothing else applies
Importance int // repo importance, already defaulted to 50
ImportanceEnabled bool // importance + keyword composite on
AIScore *int // cached AI score; nil = no verdict yet
}
ResolveInput carries every signal the precedence resolver considers. Text is the MARKER-STRIPPED item text; MarkerLevel is 0 (absent) or 1–4.
type ScanFailure ¶ added in v0.2.0
ScanFailure records one repo the audit could not scan.
type Skipped ¶
Skipped records an item deliberately not enqueued, and why. Reasons are stable strings meant for humans and tests.
type WatchConfig ¶ added in v0.2.0
type WatchConfig struct {
// Addr is the daemon endpoint, same forms as Config.DiscoveryAddr
// (unix socket path, unix:// prefixed, or host:port).
Addr string
// ProjectsDir scopes triggers to repos under this directory; events for
// projects elsewhere on the daemon never fire a tick. Ignored when
// Repos is set.
ProjectsDir string
// Repos, when set, restricts triggers to exactly these repos (the
// pool's --repos mode) and overrides the ProjectsDir scope.
Repos []string
// RepoIntervals debounces triggers per repo, keyed by repo dir base
// name (the same key the Harvester paces enqueues by): after a trigger
// for a repo, further watch events for it wait out the repo's gap.
// Repos without an entry have no explicit gap — the coalescing trigger
// channel and the harvester's own pacing still apply.
RepoIntervals map[string]time.Duration
// Log receives connection and drop warnings; nil discards.
Log *slog.Logger
// InitialBackoff is the first reconnect delay after a dropped stream;
// it doubles up to MaxBackoff. Zero uses the default (500ms).
InitialBackoff time.Duration
// MaxBackoff caps the reconnect backoff ladder. Zero uses the default
// (30s).
MaxBackoff time.Duration
}
WatchConfig controls the watch-driven harvest trigger.
type Watcher ¶ added in v0.2.0
type Watcher struct {
// contains filtered or unexported fields
}
Watcher subscribes to the daemon's GET /v1/watch SSE stream and converts project events into coalesced harvest-tick triggers. It is purely additive: a dead or missing watch stream never stalls the pool — the interval tick stays the fallback heartbeat, and the trigger channel simply stays silent.
func NewWatcher ¶ added in v0.2.0
func NewWatcher(cfg WatchConfig) *Watcher
NewWatcher creates a Watcher. Signals are delivered on a coalescing channel of capacity one: while a tick runs, further events collapse into at most one pending tick.
func (*Watcher) Run ¶ added in v0.2.0
Run subscribes to the watch stream and reconnects with exponential backoff until ctx is cancelled. It never returns an error for a dead daemon: each failed connect or dropped stream logs one warning and retries — degradation to interval-only harvesting, never a stalled pool.
func (*Watcher) Triggers ¶ added in v0.2.0
func (w *Watcher) Triggers() <-chan struct{}
Triggers yields harvest-tick signals. The channel is never closed; when the daemon is unreachable it simply stays quiet (the interval tick owns progress). A nil channel (watch off) blocks its select case forever.