sift/docs/plans/p6/02-runner-core.md

P6 task 02: the batch runner's core in internal/batch (no workers yet)

Branch: p6 (git switch p6; git status --short must be empty, otherwise stop) Commit subject: batch: runner core — types, Submit with idempotency and caps, Status, Touch, Cancel

Goal

A new package internal/batch. This task writes the whole public surface, and the part of it that needs no goroutines:

• Submit (idempotency, caps, storing the job);
• Status, Touch;
• Cancel for a job no worker holds.

Task 03 adds the workers. Until then every job stays queued: New starts nothing, and Close has nothing to stop.

The runner knows nothing about HTTP, the guard, or the real fetch pipeline. The per-target work is an injected FetchFunc (unused until task 03), and persistence is the *dcache.DB from task 01.

Files

• Copy: docs/plans/p6/_files/batch/runner_core_test.go → internal/batch/runner_core_test.go
• Create: internal/batch/batch.go: every type, variable and function in Interfaces below. Task 03 adds worker.go.
• Modify: docs/implementer-log.md

Work in small steps, in files. First write batch.go with the types and function signatures, and make it compile (go vet ./internal/batch/). Then fill in one function at a time, running the tests as you go. Do not write the implementation out in your reply before putting it in a file.

What already exists (do not re-read these packages; this is their API)

// internal/dcache (task 01). Every time argument is unix seconds.
type BatchJob struct{ ID, Caller, CallerClass, IDemKey, Status, Error string; TargetsTotal, TargetsDone int; CreatedAt, Lease, NBytes int64 }
type BatchTarget struct{ JobID string; Idx int; URL, Options, Status string; Result, Error, Hint *string }
func (d *DB) CreateBatchJob(j BatchJob) error
func (d *DB) BatchJob(id string) (BatchJob, bool)                          // a pure read
func (d *DB) BatchJobByIDem(caller, idemKey string, now int64, ttl time.Duration) (string, bool) // queued/running/done/partial only
func (d *DB) LiveBatchJobs(caller string) int                             // queued + running
func (d *DB) LiveBatchJobsTotal() int
func (d *DB) PutBatchTargets(jobID string, targets []BatchTarget) error
func (d *DB) BatchTargets(jobID string) ([]BatchTarget, bool)             // in idx order
func (d *DB) UpdateBatchTarget(jobID string, idx int, now int64, status string, result, errMsg, hint *string) error
func (d *DB) SetBatchJobStatus(id, status, errMsg string, now int64) error
func (d *DB) TouchBatchJob(id string, now int64) error
func (d *DB) SweepBatch(now int64, ttl time.Duration) int
func (d *DB) EvictBatchOldest(maxBytes int64) int
const ClassFleet, ClassRestricted = "fleet", "restricted"

// internal/auth
type Caller struct{ Name, Workspace, KeySHA256 string }

Interfaces (exact: the given test compiles against them)

package batch

type Options struct {
	Mode     string `json:"mode,omitempty"`
	Question string `json:"question,omitempty"`
	Pages    string `json:"pages,omitempty"`
	Grep     string `json:"grep,omitempty"`
	JQ       string `json:"jq,omitempty"`
	Context  *bool  `json:"context,omitempty"` // nil = /v1/fetch's default (on when there is a question)
	Recover  bool   `json:"recover,omitempty"`
}

type Target struct {
	URL     string  `json:"url"`
	Options Options `json:"options"`
}

// Result is a finished target: the /v1/fetch envelope's content, and every other envelope field
// in Meta. It is stored JSON-encoded in the target's result column.
type Result struct {
	Content string         `json:"content"`
	Meta    map[string]any `json:"meta,omitempty"`
}

// TargetError is a failure with an agent-facing hint. Status is the HTTP status the same
// failure gets from /v1/fetch; the runner ignores it (the metrics hook reads it). Any other
// error is stored with its Error() text and no hint.
type TargetError struct {
	Status    int
	Msg, Hint string
}

func (e *TargetError) Error() string { return e.Msg }

// FetchFunc does one target for the caller that submitted the job.
type FetchFunc func(ctx context.Context, c *auth.Caller, t Target) (Result, error)

type Job struct {
	ID           string
	Caller       string
	Status       string // queued | running | done | partial | failed | cancelled
	Error        string
	TargetsTotal int
	TargetsDone  int
	CreatedAt    int64 // unix seconds
}

type TargetStatus struct {
	Idx    int
	URL    string
	Status string // pending | running | done | failed
	Result *Result
	Error  string
	Hint   string
}

// Hooks observe the runner for metrics. Any nil field is skipped.
type Hooks struct {
	Job        func(outcome string)                              // created, reused, done, partial, failed, cancelled
	Target     func(err error, d time.Duration, resultBytes int) // once per stored target; err nil on success
	JobSeconds func(d time.Duration)                             // submit → terminal, once per job
	Depth      func(live int)                                    // live jobs, after every change
	Inflight   func(n int)                                       // targets being fetched, after every change
}

type Config struct {
	TTL           time.Duration // 0 → 24h
	TargetsPerJob int           // 0 → 32
	JobsPerCaller int           // 0 → 1
	JobsTotal     int           // 0 → 8
	Workers       int           // 0 → 1; values above 2 are clamped to 2
	MaxBytes      int64         // 0 → 64 MiB
	RetryWait     time.Duration // 0 → 2s: the pause before a busy target is tried again
	RetryFor      time.Duration // 0 → 10m: how long one target may keep coming back busy
	Hooks         Hooks
}

var (
	ErrNoTargets      = errors.New("no targets")
	ErrTooManyTargets = errors.New("too many targets")
	ErrCallerBusy     = errors.New("busy")
	ErrBusy           = errors.New("jobs busy")
	ErrRetryLater     = errors.New("retry later") // a FetchFunc wraps this when the target was only busy
)

func New(db *dcache.DB, fetch FetchFunc, cfg Config, log *slog.Logger) *Runner // starts the workers
func (r *Runner) Submit(ctx context.Context, c *auth.Caller, callerClass string, targets []Target) (id string, reused bool, err error)
func (r *Runner) Cancel(id string) bool
func (r *Runner) Status(id string) (Job, []TargetStatus, bool) // a pure read
func (r *Runner) Touch(id string)                              // bumps the lease to now
func (r *Runner) Close()                                       // stops the workers; waits for an in-flight fetch to return

Rules

1. Submit, in this order:
   • len(targets) == 0 → ErrNoTargets.
   • Idempotency. idem_key = hex SHA-256 of c.Name + "\x00" + json.Marshal(targets). If BatchJobByIDem(c.Name, key, now, TTL) finds a job, return its id with reused = true: no new job, no caps, Hooks.Job("reused").
   • Caps: more than TargetsPerJob → ErrTooManyTargets; LiveBatchJobs(c.Name) >= JobsPerCaller → ErrCallerBusy; LiveBatchJobsTotal() >= JobsTotal → ErrBusy.
   • Store the job queued (random 16-byte hex id, CreatedAt = Lease = now) and its targets (pending, Options = the JSON of t.Options), before returning. Keep c in memory for this job: jobs never resume after a restart, so the caller is never persisted.
   • Hooks.Job("created"); SweepBatch(now, TTL). (Task 03 also wakes a worker here.)
2. New stores its arguments and applies the Config defaults. It starts no goroutine (task 03 starts the workers there). Close is safe to call more than once.
3. Cancel(id) returns false for an unknown id or a job already terminal (done, partial, failed, cancelled). Otherwise it sets the job cancelled at once (its pending targets stay pending) and calls Hooks.Job("cancelled"). Keep the rest of Cancel simple; task 03 adds the in-flight target case.
4. Status(id) maps the store rows; Result is decoded from the result column. It writes nothing. Touch(id) is TouchBatchJob(id, now); the API calls it after an owner's read.
5. Library code never panics; a store error is logged (Warn, with the job id, never a URL) and the job goes on. Log lines carry the job id, counts and statuses, never a URL or host.

Facts

• runner_core_test.go holds the helpers that task 03's test also uses (newRunner, waitStatus, urls, and the callers alice, bob, carol).
• Without workers, a submitted job stays queued, so it counts as live for the caps.

Steps

• 1. git switch p6; copy the given test. 2. See it fail (does not compile).
• 3. batch.go: types and signatures first, compile; then Submit, Status, Touch, Cancel.
• 4. gofmt -w; go test -race -count=3 ./internal/batch/ → ok. 5. make gate.
• 6. Row p6/02-runner-core; commit.

git add internal/batch docs/implementer-log.md
git commit -F .state/commit-msg.txt

Done when

• The given test passes under -race -count=3; make gate ok; the given file is byte-identical.