From be48ecb550ca4a4be5cb7de2e5e4cdc53a117c59 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 23:13:48 -0700 Subject: [PATCH] cli: sift batch submit|status|get|cancel; README and skill Add `sift batch` so agents reach the batch endpoint through the CLI like `fetch` and `search`. `batchcmd.go` has `parseBatchArgs` (verb plus the unchanged args after it; submit takes one arg, status a job id optionally preceded by `--results`, get a job id plus a non-negative-integer idx, cancel a job id, else a usage error), `batchMain` dispatch, one small function per verb, and `batchDo` (bearer header, decode the envelope, print it on stdout, on non-2xx the error to stderr and `exitFor`). `submitMain` reads the file or stdin and validates locally: a JSON object with a non-empty `targets` array, or a bare array of targets or URL strings wrapped into `{"targets": [...]}`. README gains a "Batch" section (endpoints, serial-per-job/one-worker/ round-robin model, caps, idempotency, lease/TTL/restart, summary-vs-results, redaction, metrics); SKILL.md gains a "Batch" section. Co-Authored-By: ornith-1.5-35b-a3b --- README.md | 43 ++++++++- cmd/sift/batch_test.go | 46 +++++++++ cmd/sift/batchcmd.go | 200 ++++++++++++++++++++++++++++++++++++++++ cmd/sift/main.go | 3 + docs/implementer-log.md | 1 + skill/SKILL.md | 27 ++++++ 6 files changed, 317 insertions(+), 3 deletions(-) create mode 100644 cmd/sift/batch_test.go create mode 100644 cmd/sift/batchcmd.go diff --git a/README.md b/README.md index 5ffb2c4..b11d079 100644 --- a/README.md +++ b/README.md @@ -34,9 +34,10 @@ sift serve [-listen addr] [-config /etc/sift/config.toml] [-keys /var/lib/sift/ sift fetch [--question q] [--mode clean|raw|digest] [--jq filter] [--pages a-b] [--grep re] [--no-context] [--recover] [--json] sift extract [--pages a-b] [--grep re] [--question q] [--mode clean|digest] [--json] -sift search "" [--k n] [--json] -sift keys add --caller name [--workspace ws] [--out file] | list | revoke --caller name -sift version + sift search "" [--k n] [--json] + sift batch submit|status|get|cancel + sift keys add --caller name [--workspace ws] [--out file] | list | revoke --caller name + sift version ``` `sift fetch`, `sift extract` and `sift search` are clients: they read `url` and `key_file` from @@ -46,6 +47,42 @@ answer exactly as `sift fetch` does for a PDF URL — `--pages`, `--grep` and `- same, and a file over 100 MiB is refused before the upload. The agent-facing guide is [`skill/SKILL.md`](skill/SKILL.md). +## Batch + +`sift batch` runs several fetches as one job, so an agent can read a handful of URLs later instead +of a loop of `sift fetch`. The four endpoints: + +| Endpoint | What | +|---|---| +| `POST /v1/batch` | `{"targets":[…]}` → `202 {"success","job_id","status","targets","reused"}` | +| `GET /v1/batch/{id}` | summary; `?results=1` inlines each done target's `/v1/fetch` envelope | +| `GET /v1/batch/{id}/{idx}` | one target's result, as `sift fetch --json` | +| `DELETE /v1/batch/{id}` | cancel a queued job → `200 {"success","status"}` | + +`sift batch submit ` reads a JSON object with a non-empty `targets` array locally, +or a bare array of targets or URL strings (each string becomes `{"url": s}`), then posts it. `status` +prints the summary (or the body with `?results=1`); `get ` prints one result; `cancel` +stops a queued job. + +A job runs **serially** (target `i+1` never starts before `i` is stored) with **one worker by +default** (see `[batch]` config) and the runner rotates jobs **round-robin** while several are live. +Caps: **32 targets** per job, **1 live job per caller**, **8 live jobs** in all. The posted body is +capped at **256 KiB**. A target that is blocked, times out, or errors is recorded and the job +continues; a job whose targets are all done is `done`, all failed `failed`, mixed `partial`. + +**Idempotency**: the same list from the same caller returns the same job while it is live or already +has results (`reused: true`), so a retry does not start a second job. Each job carries a **lease** +that a status read renews; a job whose lease has expired for more than the **24 h TTL** is swept +along with its targets. On restart, jobs in flight when the server died are marked failed so a retry +re-submits them. + +A `status` read is a **summary** (per-target status, never the content); `?results=1` inlines the +full envelope for each done target, so it can be large — prefer `get ` for one result. +Restricted callers see the same endpoints; their results are **redacted** the same way a restricted +`sift fetch` is. Batch counters are `sift_batch_jobs_total{outcome}`, +`sift_batch_targets_total{outcome}`, `sift_batch_queue_depth`, `sift_batch_targets_inflight`, +`sift_batch_job_seconds`, `sift_batch_target_seconds` and `sift_batch_result_bytes`. + ## Configuration `/etc/sift/config.toml` (`$SIFT_CONFIG`); every key is shown in diff --git a/cmd/sift/batch_test.go b/cmd/sift/batch_test.go new file mode 100644 index 0000000..7315e11 --- /dev/null +++ b/cmd/sift/batch_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "slices" + "testing" +) + +func TestParseBatchArgs(t *testing.T) { + cases := []struct { + args []string + verb string + rest []string + ok bool + }{ + {[]string{"submit", "targets.json"}, "submit", []string{"targets.json"}, true}, + {[]string{"submit", "-"}, "submit", []string{"-"}, true}, + {[]string{"status", "job-123"}, "status", []string{"job-123"}, true}, + {[]string{"status", "--results", "job-123"}, "status", []string{"--results", "job-123"}, true}, + {[]string{"get", "job-123", "4"}, "get", []string{"job-123", "4"}, true}, + {[]string{"cancel", "job-123"}, "cancel", []string{"job-123"}, true}, + {[]string{}, "", nil, false}, + {[]string{"submit"}, "", nil, false}, + {[]string{"status"}, "", nil, false}, + {[]string{"status", "a", "b"}, "", nil, false}, + {[]string{"status", "--results"}, "", nil, false}, + {[]string{"get", "job-123"}, "", nil, false}, + {[]string{"get", "job-123", "x"}, "", nil, false}, + {[]string{"get", "job-123", "-1"}, "", nil, false}, + {[]string{"cancel"}, "", nil, false}, + {[]string{"bogus", "x"}, "", nil, false}, + } + for _, c := range cases { + verb, rest, err := parseBatchArgs(c.args) + if c.ok { + if err != nil { + t.Errorf("%v: unexpected error %v", c.args, err) + continue + } + if verb != c.verb || !slices.Equal(rest, c.rest) { + t.Errorf("%v: got verb=%q rest=%q", c.args, verb, rest) + } + } else if err == nil { + t.Errorf("%v: want an error", c.args) + } + } +} diff --git a/cmd/sift/batchcmd.go b/cmd/sift/batchcmd.go new file mode 100644 index 0000000..ef2986b --- /dev/null +++ b/cmd/sift/batchcmd.go @@ -0,0 +1,200 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strconv" +) + +const batchUsage = `usage: sift batch submit|status|get|cancel + sift batch submit + sift batch status [--results] + sift batch get + sift batch cancel ` + +// parseBatchArgs validates a `sift batch` command line (without "batch"). verb is +// the subcommand; rest is the arguments after it, unchanged. +func parseBatchArgs(args []string) (verb string, rest []string, err error) { + if len(args) == 0 { + return "", nil, fmt.Errorf("%s", batchUsage) + } + verb, rest = args[0], args[1:] + switch verb { + case "submit": + if len(rest) != 1 { + return "", nil, fmt.Errorf("%s", batchUsage) + } + case "status": + switch len(rest) { + case 1: + if rest[0] == "--results" { + return "", nil, fmt.Errorf("%s", batchUsage) + } + case 2: + if rest[0] != "--results" { + return "", nil, fmt.Errorf("%s", batchUsage) + } + default: + return "", nil, fmt.Errorf("%s", batchUsage) + } + case "get": + if len(rest) != 2 { + return "", nil, fmt.Errorf("%s", batchUsage) + } + idx, e := strconv.Atoi(rest[1]) + if e != nil || idx < 0 { + return "", nil, fmt.Errorf("%s", batchUsage) + } + case "cancel": + if len(rest) != 1 { + return "", nil, fmt.Errorf("%s", batchUsage) + } + default: + return "", nil, fmt.Errorf("%s", batchUsage) + } + return verb, rest, nil +} + +func batchMain() { + verb, rest, err := parseBatchArgs(os.Args[1:]) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + switch verb { + case "submit": + submitMain(rest) + case "status": + statusMain(rest) + case "get": + getMain(rest) + case "cancel": + cancelMain(rest) + } +} + +// submitMain reads the targets file (or stdin when the argument is "-") and +// validates it locally before calling the server: a JSON object with a +// non-empty targets array, or a bare array of targets or URL strings wrapped +// into {"targets": [...]}. +func submitMain(rest []string) { + var r io.Reader = os.Stdin + if rest[0] != "-" { + f, err := os.Open(rest[0]) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(5) + } + defer f.Close() + r = f + } + raw, err := io.ReadAll(r) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(5) + } + body, err := buildTargetsBody(raw) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + batchDo(http.MethodPost, "/v1/batch", bytes.NewReader(body)) +} + +// buildTargetsBody accepts the three accepted shapes and returns the +// {"targets": [...]} body to send. A bare array of URL strings gets each string +// wrapped as {"url": s}; bare target objects pass through. +func buildTargetsBody(raw []byte) ([]byte, error) { + var probe any + if err := json.Unmarshal(raw, &probe); err != nil { + return nil, fmt.Errorf("targets: not valid JSON: %w", err) + } + switch p := probe.(type) { + case map[string]any: + arr, ok := p["targets"].([]any) + if !ok || len(arr) == 0 { + return nil, fmt.Errorf("targets: a JSON object needs a non-empty targets array") + } + return raw, nil + case []any: + if len(p) == 0 { + return nil, fmt.Errorf("targets: an empty array") + } + out := make([]any, 0, len(p)) + for _, el := range p { + switch e := el.(type) { + case map[string]any: + out = append(out, e) + case string: + out = append(out, map[string]any{"url": e}) + default: + return nil, fmt.Errorf("targets: array elements must be URLs or objects") + } + } + return json.Marshal(map[string]any{"targets": out}) + default: + return nil, fmt.Errorf("targets: expected a JSON object or array") + } +} + +func statusMain(rest []string) { + id := rest[len(rest)-1] + path := "/v1/batch/" + url.PathEscape(id) + if len(rest) == 2 && rest[0] == "--results" { + path += "?results=1" + } + batchDo(http.MethodGet, path, nil) +} + +func getMain(rest []string) { + path := "/v1/batch/" + url.PathEscape(rest[0]) + "/" + url.PathEscape(rest[1]) + batchDo(http.MethodGet, path, nil) +} + +func cancelMain(rest []string) { + batchDo(http.MethodDelete, "/v1/batch/"+url.PathEscape(rest[0]), nil) +} + +// batchDo calls base+path with the bearer header, decodes the JSON envelope, +// and prints it on stdout. On a non-2xx it writes the error to stderr and exits +// with exitFor, as fetch does. +func batchDo(method, path string, body io.Reader) { + base, key, err := loadClient() + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(5) + } + req, err := http.NewRequest(method, base+path, body) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(5) + } + req.Header.Set("Authorization", "Bearer "+key) + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(5) + } + defer resp.Body.Close() + var env map[string]any + if err := json.NewDecoder(resp.Body).Decode(&env); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(5) + } + if resp.StatusCode != 200 { + msg, _ := env["error"].(string) + fmt.Fprintln(os.Stderr, msg) + os.Exit(exitFor(resp.StatusCode, msg)) + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + _ = enc.Encode(env) +} diff --git a/cmd/sift/main.go b/cmd/sift/main.go index b42c8fd..a9c83c0 100644 --- a/cmd/sift/main.go +++ b/cmd/sift/main.go @@ -26,6 +26,8 @@ func main() { extractMain() case "search": searchMain() + case "batch": + batchMain() case "keys": keysMain() case "pdftext": @@ -49,6 +51,7 @@ Commands: sift fetch fetch via the service sift extract extract a local PDF via the service sift search "" search via the service + sift batch submit|status|get|cancel run or watch a batch of fetches sift keys add|list|revoke sift version `, version) diff --git a/docs/implementer-log.md b/docs/implementer-log.md index a2823b7..498ca05 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -25,6 +25,7 @@ One row per task session that ends in a commit. Newest last. | p6/06-batch-wiring | 2026-09-26 | done | 1 | gate: ok | none | New `internal/api/batch.go`: `NewBatchRunner` overwrites `cfg.Hooks` so Job→ObserveBatchJob, Target→ObserveBatchTarget(batchOutcome)+ObserveBatchTargetSeconds (+ObserveBatchResultBytes only when err==nil), and JobSeconds/Depth/Inflight→the matching `s.Prom` methods; the runner's FetchFunc is `batchFetch`, which tags ctx with an unexported `batchCtxKey`, runs under `requestDeadline`, and maps fetchTarget's errors (a *fetch.Error 429/503 → `fmt.Errorf("%s: %w", fe.Msg, batch.ErrRetryLater)` so the runner retries; any other *fetch.Error → TargetError with 502 rewritten to 500 "internal error" and `s.blockN.Add(1)` for 403; anything else → TargetError 500 logged Error with `s.logErr`). `batchOutcome` classifies nil→ok, ErrRetryLater-wrapped→busy, then by TargetError.Status (403 blocked, 429 busy, 504 timeout, 415 unsupported, 422 whose Msg starts "document too long"→too_long, other 4xx→upstream_4xx, 5xx→upstream_5xx, any other error→upstream_5xx). Added `Batch *batch.Runner` Server field (nil → the /v1/batch endpoints answer 503 "batch disabled"). `pdf.go` extractPDF charges the PDF pool to `c.Name+"/batch"` when the ctx carries the batch mark, single-flight key unchanged so a batch and an interactive extract of the same PDF still share one child; the only new log line carries caller and host="-" (never a URL/host). All api tests pass under -race -count=3; make gate ok; given file byte-identical. | | p6/07-api-endpoints | 2026-09-26 | done | 1 | gate: ok | none | The four /v1/batch handlers in new `internal/api/batch.go`: `batchSubmit` (POST) decodes `{"targets":[{url,mode,question,pages,grep,jq,context,recover}]}` flat like `/v1/fetch`, 400 "invalid batch" on bad json/empty list/empty url/bad mode (only empty,clean,raw,digest), then `PreCheck` per target → 422 "blocked: target : " with no job, then `Submit` mapping ErrTooManyTargets→422 "too many targets (max 32)", ErrCallerBusy→429+Retry-After 30, ErrBusy→429+Retry-After 60, ErrNoTargets→400, and 202 {success,job_id,status,targets,reused} (status "queued" new, current status reused). `batchStatus` (GET) and `batchResult` (GET {id}/{idx}) and `batchCancel` (DELETE) share `ownership` (unknown or not `c.Name` → 404, never 403; nil batch → 503); status is a summary with no results unless `?results=1` inlines the envelope (Meta+content) on done targets, idx validated 0..targets_total-1 else 404, cancel of a terminal job → 409 "job already finished" else 200 {success,status:cancelled}. `api.go` scoped to the rule-2 changes: `batchMaxBody` (256 KiB) const, the four routes in `New`, and `ServeHTTP` applying the per-path cap once (`/v1/batch`→256 KiB, `/v1/extract`→pdfMaxUpload, else 64 KiB) and collapsing every batch path to a fixed metrics label via `batchObservePath` (`/v1/batch` or `/v1/batch/{id}`, never a job id). Each handler logs one `batch` line with caller/job_id/targets/status-or-reused and never a URL/hostname/idem_key. Given test byte-identical; all api tests pass under -race -count=3; make gate ok. | | p6/08-serve-config | 2026-09-26 | done | 1 | gate: ok | none | `[batch]` config in `internal/config`: the `Batch` type (disabled/targets_per_job/jobs_per_caller/jobs_total/workers/ttl/max_bytes) and `File.Batch`, plus `ParseSize` (trims spaces, accepts only binary KiB/MiB/GiB suffixes or a plain byte count, zero/negative/other → def). serve.go opens the dcache store when batch is on (`!cfg.Batch.Disabled`) alongside dixie/search — same path, same call, Warn on failure — then, before building the runner, reaps in-flight jobs a restart killed via `MarkInFlightFailed("restart", now)` with one `ObserveBatchJob("restart")` per reaped job and a `batch reaped` info log when n>0; builds `srv.Batch` from the config (TTL via `ParseDuration`, MaxBytes via `ParseSize`, runner defaults fill the zeros) only when enabled and the store opened; closes `srv.Batch` in both shutdown branches when non-nil. Added a commented `[batch]` block to dist/sift.toml.example. Given test byte-identical; all config and cmd tests pass under -race -count=3. | +| p6/09-cli-docs | 2026-09-26 | done | 1 | gate: ok | none | `cmd/sift/batchcmd.go`: `parseBatchArgs` (verb + unchanged `rest`; submit takes one arg, status one job id optionally preceded by `--results`, get a job id plus a non-negative-integer idx, cancel one job id, else a usage error), `batchMain` dispatch, one small function per verb, and `batchDo` (bearer header, decode the JSON envelope, print it indented on stdout, on non-2xx print `env["error"]` to stderr and `exitFor`). `submitMain` reads the file or stdin and `buildTargetsBody` accepts a JSON object with a non-empty `targets` array, a bare array of target objects, or a bare array of URL strings (each wrapped `{"url": s}`), validating locally before the call. `main.go`: one `case "batch"` and one help line. README "Batch" section (four endpoints, serial-per-job/one-worker/round-robin model, 32 targets/1-per-caller/8-total caps, 256 KiB body, idempotency, lease/TTL/restart, summary-vs-results, restricted redaction, `sift_batch_*` metrics); SKILL.md "Batch" section. Given test byte-identical; all cmd/sift tests pass under -race -count=3. | ## P5a review (owner, 2026-09-26) diff --git a/skill/SKILL.md b/skill/SKILL.md index 1b3c381..77d5391 100644 --- a/skill/SKILL.md +++ b/skill/SKILL.md @@ -55,6 +55,33 @@ Useful flags: `--mode digest|clean|raw` (default `digest` when `--question` is g else `clean`), `--no-context` (skip Honcho lookup), `--recover` (server-side archive ladder for blocked pages), `--json` (full envelope: content, source, context, sizes, warnings). +## Batch + +When you will read the content of several URLs later (a shortlist to compare), submit them once +instead of a loop of `sift fetch`: + +```bash +# one JSON object, or a bare array of URLs +sift batch submit targets.json +# ["https://a", "https://b"] also works; each string becomes {"url": ...} +``` + +Then poll and read: + +```bash +sift batch status # summary: per-target status, no content +sift batch status --results # everything at once — can be large, avoid unless needed +sift batch get # one target's result, as `sift fetch --json` +sift batch cancel # stop a queued job +``` + +- Submit once; a retry of the same list from this caller returns the same job (`reused`), it does + not start a second. +- Don't use `--results` unless you need every result at once — a batch of 32 can be several + megabytes. Read each piece with `sift batch get `. +- A batch is slow on purpose: it runs one target at a time per job. For one URL you need now, use + `sift fetch`. + ## PDFs PDFs work: `sift fetch` extracts them and returns the text. A short one comes back whole; a