Merge P5c: POST /v1/extract and sift extract (PDFs already on a host)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -33,14 +33,18 @@ with `blocked: …`.
|
||||
sift serve [-listen addr] [-config /etc/sift/config.toml] [-keys /var/lib/sift/keys.toml] [-deny-prefix CIDR]...
|
||||
sift fetch <url> [--question q] [--mode clean|raw|digest] [--jq filter] [--pages a-b] [--grep re]
|
||||
[--no-context] [--recover] [--json]
|
||||
sift extract <file.pdf> [--pages a-b] [--grep re] [--question q] [--mode clean|digest] [--json]
|
||||
sift search "<query>" [--k n] [--json]
|
||||
sift keys add --caller name [--workspace ws] [--out file] | list | revoke --caller name
|
||||
sift version
|
||||
```
|
||||
|
||||
`sift fetch` and `sift search` are clients: they read `url` and `key_file` from `client.toml`
|
||||
(`$SIFT_CLIENT`, else `$HERMES_HOME/sift/client.toml`, else `~/.config/sift/client.toml`); the key
|
||||
file must be mode 0600. The agent-facing guide is [`skill/SKILL.md`](skill/SKILL.md).
|
||||
`sift fetch`, `sift extract` and `sift search` are clients: they read `url` and `key_file` from
|
||||
`client.toml` (`$SIFT_CLIENT`, else `$HERMES_HOME/sift/client.toml`, else `~/.config/sift/client.toml`);
|
||||
the key file must be mode 0600. `sift extract` uploads a local PDF to `/v1/extract` and prints the
|
||||
answer exactly as `sift fetch` does for a PDF URL — `--pages`, `--grep` and `--question` behave the
|
||||
same, and a file over 100 MiB is refused before the upload. The agent-facing guide is
|
||||
[`skill/SKILL.md`](skill/SKILL.md).
|
||||
|
||||
## Configuration
|
||||
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
package main
|
||||
|
||||
import "testing"
|
||||
|
||||
// P5c owner review: a client base URL with a path prefix keeps it, as fetch and search do.
|
||||
func TestExtractURLKeepsBasePath(t *testing.T) {
|
||||
got := extractURL("https://sift.example:8443/prefix/", extractOpts{File: "a.pdf", Pages: "2"})
|
||||
if want := "https://sift.example:8443/prefix/v1/extract?pages=2"; got != want {
|
||||
t.Fatalf("extractURL = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
package main
|
||||
|
||||
// P5c task 02: `sift extract <file>` uploads a PDF to /v1/extract.
|
||||
|
||||
import (
|
||||
"net/url"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestParseExtractOpts(t *testing.T) {
|
||||
o, err := parseExtractOpts([]string{"thesis.pdf", "--pages", "41-60", "--grep", "attention head"})
|
||||
if err != nil || o.File != "thesis.pdf" || o.Pages != "41-60" || o.Grep != "attention head" {
|
||||
t.Errorf("%+v %v", o, err)
|
||||
}
|
||||
o, err = parseExtractOpts([]string{"--question", "what is the method?", "--json", "paper.pdf"})
|
||||
if err != nil || o.File != "paper.pdf" || o.Question != "what is the method?" || !o.JSON {
|
||||
t.Errorf("flags first: %+v %v", o, err)
|
||||
}
|
||||
if _, err := parseExtractOpts(nil); err == nil {
|
||||
t.Error("no file: want an error")
|
||||
}
|
||||
if _, err := parseExtractOpts([]string{"a.pdf", "b.pdf"}); err == nil {
|
||||
t.Error("two files: want an error")
|
||||
}
|
||||
if !strings.Contains(extractUsage, "sift extract <file.pdf>") {
|
||||
t.Errorf("usage %q", extractUsage)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractURL(t *testing.T) {
|
||||
u, err := url.Parse(extractURL("https://sift.example:8443", extractOpts{File: "x.pdf", Pages: "2-3", Grep: "a b&c"}))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if u.Path != "/v1/extract" || u.Query().Get("pages") != "2-3" || u.Query().Get("grep") != "a b&c" {
|
||||
t.Errorf("url %s", u)
|
||||
}
|
||||
if u.Query().Has("question") || u.Query().Has("mode") {
|
||||
t.Errorf("unset options must not be sent: %s", u)
|
||||
}
|
||||
u, _ = url.Parse(extractURL("https://sift.example:8443", extractOpts{File: "x.pdf", Question: "why?"}))
|
||||
if u.Query().Get("question") != "why?" || u.Query().Get("mode") != "digest" {
|
||||
t.Errorf("question url %s (mode must be digest with a question)", u)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,156 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// maxExtractUpload mirrors the server's /v1/extract cap (a file over 100 MiB is
|
||||
// refused on upload with 413), so reject it before uploading.
|
||||
const maxExtractUpload = 100 << 20
|
||||
|
||||
type extractOpts struct {
|
||||
File, Pages, Grep, Question, Mode string
|
||||
JSON bool
|
||||
}
|
||||
|
||||
const extractUsage = "usage: sift extract <file.pdf> [--pages a-b] [--grep RE] [--question Q] [--mode clean|digest] [--json]"
|
||||
|
||||
// parseExtractOpts parses `sift extract` args into extractOpts. It knows which
|
||||
// flags take a value; the PDF path is the one positional.
|
||||
func parseExtractOpts(args []string) (extractOpts, error) {
|
||||
flagArgs, pos, err := splitAroundPositional(args, extractValueFlags)
|
||||
if err != nil {
|
||||
return extractOpts{}, err
|
||||
}
|
||||
fs := flag.NewFlagSet("extract", flag.ContinueOnError)
|
||||
fs.SetOutput(io.Discard)
|
||||
modeF := fs.String("mode", "", "clean|digest (default digest if --question)")
|
||||
pagesF := fs.String("pages", "", "only these pages of a PDF (a-b or a)")
|
||||
grepF := fs.String("grep", "", "only lines matching RE in a PDF")
|
||||
questionF := fs.String("question", "", "focus the digest")
|
||||
asJSONF := fs.Bool("json", false, "print full /v1/extract envelope")
|
||||
if err := fs.Parse(flagArgs); err != nil {
|
||||
return extractOpts{}, err
|
||||
}
|
||||
if fs.NArg() != 0 {
|
||||
return extractOpts{}, fmt.Errorf("%s", extractUsage)
|
||||
}
|
||||
if len(pos) != 1 {
|
||||
return extractOpts{}, fmt.Errorf("%s", extractUsage)
|
||||
}
|
||||
return extractOpts{
|
||||
File: pos[0],
|
||||
Mode: *modeF,
|
||||
Pages: *pagesF,
|
||||
Grep: *grepF,
|
||||
Question: *questionF,
|
||||
JSON: *asJSONF,
|
||||
}, nil
|
||||
}
|
||||
|
||||
var extractValueFlags = map[string]bool{
|
||||
"mode": true, "pages": true, "grep": true, "question": true,
|
||||
}
|
||||
|
||||
// extractURL builds the /v1/extract URL from base plus only the options that are
|
||||
// set; mode defaults to digest when a question is given and no mode was.
|
||||
func extractURL(base string, o extractOpts) string {
|
||||
v := url.Values{}
|
||||
if o.Pages != "" {
|
||||
v.Set("pages", o.Pages)
|
||||
}
|
||||
if o.Grep != "" {
|
||||
v.Set("grep", o.Grep)
|
||||
}
|
||||
if o.Question != "" {
|
||||
v.Set("question", o.Question)
|
||||
}
|
||||
mode := o.Mode
|
||||
if mode == "" && o.Question != "" {
|
||||
mode = "digest"
|
||||
}
|
||||
if mode != "" {
|
||||
v.Set("mode", mode)
|
||||
}
|
||||
u := strings.TrimRight(base, "/") + "/v1/extract"
|
||||
if q := v.Encode(); q != "" {
|
||||
u += "?" + q
|
||||
}
|
||||
return u
|
||||
}
|
||||
|
||||
func extractMain() {
|
||||
o, err := parseExtractOpts(os.Args[1:])
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(2)
|
||||
}
|
||||
base, key, err := loadClient()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(5)
|
||||
}
|
||||
f, err := os.Open(o.File)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(5)
|
||||
}
|
||||
defer f.Close()
|
||||
st, err := f.Stat()
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(5)
|
||||
}
|
||||
if st.Size() > maxExtractUpload {
|
||||
fmt.Fprintf(os.Stderr, "extract: %s is %d bytes, over the %d MB limit\n", o.File, st.Size(), maxExtractUpload>>20)
|
||||
os.Exit(3)
|
||||
}
|
||||
req, err := http.NewRequest(http.MethodPost, extractURL(base, o), f)
|
||||
if err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(5)
|
||||
}
|
||||
req.ContentLength = st.Size()
|
||||
req.Header.Set("Authorization", "Bearer "+key)
|
||||
req.Header.Set("Content-Type", "application/pdf")
|
||||
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)
|
||||
hint, _ := env["hint"].(string)
|
||||
if hint != "" {
|
||||
fmt.Fprintf(os.Stderr, "%s\nhint: %s\n", msg, hint)
|
||||
} else {
|
||||
fmt.Fprintln(os.Stderr, msg)
|
||||
}
|
||||
os.Exit(exitFor(resp.StatusCode, msg))
|
||||
}
|
||||
if o.JSON {
|
||||
enc := json.NewEncoder(os.Stdout)
|
||||
enc.SetIndent("", " ")
|
||||
_ = enc.Encode(env)
|
||||
return
|
||||
}
|
||||
if c, ok := env["content"].(string); ok {
|
||||
fmt.Print(c)
|
||||
if !strings.HasSuffix(c, "\n") {
|
||||
fmt.Println()
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -22,6 +22,8 @@ func main() {
|
||||
serveMain()
|
||||
case "fetch":
|
||||
fetchMain()
|
||||
case "extract":
|
||||
extractMain()
|
||||
case "search":
|
||||
searchMain()
|
||||
case "keys":
|
||||
@@ -45,6 +47,7 @@ func usage() {
|
||||
Commands:
|
||||
sift serve [flags] run the service
|
||||
sift fetch <url> fetch via the service
|
||||
sift extract <file.pdf> extract a local PDF via the service
|
||||
sift search "<query>" search via the service
|
||||
sift keys add|list|revoke
|
||||
sift version
|
||||
|
||||
@@ -15,6 +15,8 @@ One row per task session that ends in a commit. Newest last.
|
||||
| p5b/02-retrieval | 2026-09-26 | done | 1 | gate: ok | none | Cached chunk embeddings + BM25-first retrieval. `digest.go`: `EmbedSyncMax=48`, `MaxDocChunks=5000`; `Result` gains `Retrieval string` and `Embedded int`; `selectText` now returns `(body, used, total, mode, embedded, err)` and `Run` sets both; `PutDigest` skips storing any `mode=="bm25"` answer (provisional, not cached). `retrieval.go`: `hybridOrBM25` — no cache, `len(chunks)<=48` embeds question+chunks inline and scores hybrid; `>48` scores BM25 and queues; cache present embeds the question and only the uncached chunks inline and scores hybrid on the union; both embed paths return hybrid, both bm25 paths return `bm25`. `background.go`: `bgWorker` with `bgOnce sync.Once` (creates the worker, never starts a goroutine) + `w.started` under `w.mu` (starts the goroutine once), dedup by `contentHash|callerClass`, full-queue drops the doc without marking seen so a later question re-queues, `acquireGate` retries once after 1 s on refusal, `WaitBackground` polls every 20 ms for true idle. `retrieval_test.go` copied byte-identical from the plan. All 12 digest tests pass under `-race -count=3`; `make gate` ok. |
|
||||
| p5a/07-api-cache-pages | 2026-09-26 | done | 1 | gate: ok | none | Implemented the PDF document cache with revalidation in `doFetch`: `pdfFromCache` (miss / fresh within `PDFTTL` / stale → `FetchIf` with 304→`TouchPDF`+answer-fresh / 200→return response with ok=false and no second GET) and `storePDF` after `extractPDF` (`ContentHash=res.SHA256`, ETag/Last-Modified from headers, FetchedAt=now, Meta=JSON of `doc.Meta`, Pages=`doc.Text`). `pages`/`grep` passed into `pdfdoc.Options` in `renderPDF`; `ErrBadPages`/`ErrBadGrep`→400 with the error's own text, non-PDF→400 "pages and grep apply to PDFs only" before the doc block. Question limit in digest mode: `n=len(digest.Split(pdfdoc.Full(doc), ChunkTokens))`, `n>PDFMaxQuestionChunks`→422 "document too long for question (n chunks); use grep or pages" before any digest call. Envelope `cached` set true for cache answers via the synthetic cache res's empty body (`len(res.Body)==0`); the one `doFetch` change beyond the rule-1 call was reusing that 200 response instead of a second `Fetch`. Added `fmt` import to api.go. Given file byte-identical; make gate ok. |
|
||||
| p5b/03-api | 2026-09-26 | done | 1 | gate: ok | none | The digest now answers long documents itself (task 02), so the API's refusal limit is only a safety cap: `DefaultPDFMaxQuestionChunks` 200→5000 in `internal/api/pdf.go` (the 422 message text unchanged, still "document too long for question (N chunks); use grep or pages"). In `v1fetch` (`internal/api/api.go`) the digest-meta map gains `"retrieval": dr.Retrieval` and `"embedded": dr.Embedded` next to `chunks_used`/`chunks_total`, only when the digest ran without a fallback. Copied the two given tests byte-identical from the plan — the replacement `pdf_test.go` is now the protected one and asserts `DefaultPDFMaxQuestionChunks == 5000`; `pdf_retrieval_test.go` asserts the first long-doc question returns `retrieval=bm25`, `embedded=0`. README PDF section: a long-doc question is answered first from BM25 (`retrieval: "bm25"`) while its chunks embed in the background and hybrid on later questions, plus the 5000-chunk cap. One sentence added to `skill/SKILL.md`. All api tests pass under `-race -count=3`; make gate ok; given files byte-identical. |
|
||||
| p5c/01-extract-api | 2026-09-26 | done | 1 | gate: ok | none | `POST /v1/extract` answers a PDF posted in the body like `/v1/fetch` answers the same PDF by URL, nothing fetched. `internal/api/extract.go`: auth via `caller`; `io.ReadAll` of the body (a `*http.MaxBytesError`→413 "too large (>N MB)", empty→400); `clean.Kind` must be "pdf" else 415; query `pages`/`grep`/`question`/`mode` (digest when a question is set, else clean; other→400). Content-addressed cache: `hash="sha256:"+HashBody(body)`, `key=PDFKey("upload:"+hash, callerClass)`; a `GetPDF` hit rebuilds the doc via `pdfFromEntry` with `cached:true` + `PDFCacheHit` and no child (bypasses the TTL/revalidation that `pdfFromCache` does, so identical content is extracted once); a miss builds a synthetic `fetch.Resource{URL:"upload",...}` and runs `extractPDF`+`storePDF`. Bad pages/grep→400 via `writeFetchErr`, the chunk limit→the same 422 as `v1fetch`, extraction errors→`pdfError` via `writeFetchErr`. Envelope mirrors the PDF fetch path (`content`/`mode`/`untrusted`/`warnings`/`sizes`/`timings`/`pdf`/`digest`); `source` is `{"kind":"upload","hash","bytes","cached","fetched_at"}` with no `url` key; no Honcho context for uploads; one log line "extract" with caller/bytes/pdf_pages/mode/cached. `internal/api/api.go` scoped to the rule-1 changes: `DefaultPDFMaxUpload` const (100 MiB), `PDFMaxUpload int64` Server field, the `POST /v1/extract` route in `New`, and `ServeHTTP` (per-path upload cap instead of the 64 KiB JSON limit, and `/v1/extract` added to the `requestDeadline` switch). The `pdfMaxUpload()` field-or-default resolver lives in extract.go. Given test byte-identical; all api tests pass under `-race -count=3`; make gate ok. |
|
||||
| p5c/02-extract-cli | 2026-09-26 | done | 1 | gate: ok | none | `sift extract <file.pdf>` uploads a local PDF to `/v1/extract` and prints the answer exactly as `sift fetch` does. `cmd/sift/extractcmd.go`: `extractOpts` + `parseExtractOpts` (reuses `splitAroundPositional` with `extractValueFlags` mode/pages/grep/question, one positional file, `--json`), `extractURL` builds `base+"/v1/extract"` with `url.Values` set only for the options that are set and `mode=digest` when a question is given and no mode — values are URL-encoded so a grep like `a b&c` survives round-trip; `extractMain` reuses `loadClient`, `stat`s the file and refuses >100 MiB before uploading (exit 3, matching the server's 413→exitFor), then `http.NewRequest` with the `*os.File` as the streaming body and `ContentLength` set (no second in-memory copy), `Content-Type: application/pdf` + bearer, and the same error/exitFor path as fetch; success prints `content` (or the `--json` envelope). `main.go`: `case "extract"` and one help line. README: extract line in the CLI block plus a sentence; SKILL.md: a PDF-on-disk bullet in the PDFs section. Given test byte-identical; all cmd/sift tests pass under `-race -count=3`; make gate ok; `bin/sift extract` prints the usage line. |
|
||||
|
||||
## P5a review (owner, 2026-09-26)
|
||||
|
||||
|
||||
+17
-2
@@ -38,6 +38,10 @@ const (
|
||||
maxReqBody = 64 << 10
|
||||
writeTimeout = 240 * time.Second // P5a: a PDF may take 120 s to download + 20 s queued + 60 s to extract
|
||||
requestDeadline = 210 * time.Second // P5a: the PDF path's worst case (see writeTimeout); other paths keep their own shorter timeouts
|
||||
|
||||
// DefaultPDFMaxUpload is the largest body POST /v1/extract accepts: an uploaded
|
||||
// PDF, addressed by its hash and extracted once per content.
|
||||
DefaultPDFMaxUpload = 100 << 20
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
@@ -55,6 +59,7 @@ type Server struct {
|
||||
PDFPool *pdfpool.Pool
|
||||
PDFTTL time.Duration
|
||||
PDFMaxQuestionChunks int
|
||||
PDFMaxUpload int64 // 0 = DefaultPDFMaxUpload; the /v1/extract upload cap
|
||||
Search *search.Engine
|
||||
Ctx *fleetctx.Client
|
||||
// PermitRestricted is context.permit_restricted: those callers never
|
||||
@@ -69,6 +74,10 @@ type Server struct {
|
||||
reqN, blockN, scrapeN atomic.Uint64
|
||||
|
||||
pdfPoolOnce sync.Once
|
||||
|
||||
// uploads bounds how many /v1/extract bodies (up to PDFMaxUpload each) are read into
|
||||
// memory at once, ahead of the PDF pool's own limits.
|
||||
uploads chan struct{}
|
||||
}
|
||||
|
||||
func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log *slog.Logger) *Server {
|
||||
@@ -85,6 +94,7 @@ func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log
|
||||
OutCap: 200_000,
|
||||
global: make(chan struct{}, 8),
|
||||
per: map[string]chan struct{}{},
|
||||
uploads: make(chan struct{}, maxUploads),
|
||||
}
|
||||
s.Prom = metrics.New(s)
|
||||
s.Mux.HandleFunc("GET /healthz", s.healthz)
|
||||
@@ -93,6 +103,7 @@ func New(version string, g *guard.Guard, f *fetch.Fetcher, keys *auth.Store, log
|
||||
s.Mux.HandleFunc("GET /metrics.json", s.metrics)
|
||||
s.Mux.HandleFunc("POST /v2/scrape", s.scrape)
|
||||
s.Mux.HandleFunc("POST /v1/fetch", s.v1fetch)
|
||||
s.Mux.HandleFunc("POST /v1/extract", s.v1extract)
|
||||
s.Mux.HandleFunc("POST /v1/search", s.v1search)
|
||||
s.Mux.HandleFunc("POST /v2/search", s.v2search)
|
||||
return s
|
||||
@@ -111,9 +122,13 @@ func (s *Server) HTTPServer(addr string) *http.Server {
|
||||
}
|
||||
|
||||
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
||||
r.Body = http.MaxBytesReader(w, r.Body, maxReqBody)
|
||||
var limit int64 = maxReqBody
|
||||
if r.URL.Path == "/v1/extract" {
|
||||
limit = int64(s.pdfMaxUpload())
|
||||
}
|
||||
r.Body = http.MaxBytesReader(w, r.Body, limit)
|
||||
switch r.URL.Path {
|
||||
case "/v1/fetch", "/v2/scrape", "/v1/search", "/v2/search":
|
||||
case "/v1/fetch", "/v2/scrape", "/v1/search", "/v2/search", "/v1/extract":
|
||||
ctx, cancel := context.WithTimeout(r.Context(), requestDeadline)
|
||||
defer cancel()
|
||||
r = r.WithContext(ctx)
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"git.wntrmute.dev/kyle/sift/internal/clean"
|
||||
"git.wntrmute.dev/kyle/sift/internal/dcache"
|
||||
"git.wntrmute.dev/kyle/sift/internal/digest"
|
||||
"git.wntrmute.dev/kyle/sift/internal/fetch"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdfdoc"
|
||||
"git.wntrmute.dev/kyle/sift/internal/pdftext"
|
||||
)
|
||||
|
||||
// v1extract answers a PDF the caller POSTs in the request body, exactly as /v1/fetch
|
||||
// answers the same PDF by URL. Nothing is fetched: the bytes are hashed, the hash is the
|
||||
// cache key, and identical content is extracted once (content-addressed, so no TTL and no
|
||||
// revalidation).
|
||||
func (s *Server) v1extract(w http.ResponseWriter, r *http.Request) {
|
||||
s.reqN.Add(1)
|
||||
c := s.caller(w, r)
|
||||
if c == nil {
|
||||
return
|
||||
}
|
||||
|
||||
select {
|
||||
case s.uploads <- struct{}{}:
|
||||
defer func() { <-s.uploads }()
|
||||
case <-time.After(uploadWait):
|
||||
w.Header().Set("Retry-After", "10")
|
||||
writeErr(w, 429, "busy", "")
|
||||
return
|
||||
case <-r.Context().Done():
|
||||
return
|
||||
}
|
||||
|
||||
body, rerr := io.ReadAll(r.Body)
|
||||
if rerr != nil {
|
||||
var me *http.MaxBytesError
|
||||
if errors.As(rerr, &me) {
|
||||
writeErr(w, 413, fmt.Sprintf("too large (>%d MB)", s.pdfMaxUpload()>>20), "")
|
||||
return
|
||||
}
|
||||
writeErr(w, 400, "empty body", "")
|
||||
return
|
||||
}
|
||||
if len(body) == 0 {
|
||||
writeErr(w, 400, "empty body", "")
|
||||
return
|
||||
}
|
||||
if clean.Kind(r.Header.Get("Content-Type"), body) != "pdf" {
|
||||
writeErr(w, 415, "extract accepts PDFs only", "")
|
||||
return
|
||||
}
|
||||
|
||||
q := r.URL.Query()
|
||||
pages := q.Get("pages")
|
||||
grep := q.Get("grep")
|
||||
question := q.Get("question")
|
||||
mode := q.Get("mode")
|
||||
if mode == "" {
|
||||
if question != "" {
|
||||
mode = "digest"
|
||||
} else {
|
||||
mode = "clean"
|
||||
}
|
||||
}
|
||||
if mode != "clean" && mode != "digest" {
|
||||
writeErr(w, 400, "invalid mode", "")
|
||||
return
|
||||
}
|
||||
|
||||
hash := "sha256:" + HashBody(body)
|
||||
key := dcache.PDFKey("upload:"+hash, s.callerClass(c))
|
||||
|
||||
t0 := time.Now()
|
||||
var res *fetch.Resource
|
||||
var doc *pdftext.Doc
|
||||
cached := false
|
||||
if cache := s.cache(); cache != nil {
|
||||
if e, ok := cache.GetPDF(key); ok {
|
||||
res, doc = s.pdfFromEntry("upload", e)
|
||||
cached = true
|
||||
s.Prom.PDFCacheHit()
|
||||
}
|
||||
}
|
||||
if res == nil {
|
||||
res = &fetch.Resource{URL: "upload", FinalURL: "upload", Status: 200, ContentType: "application/pdf", Body: body, Hash: hash, FetchedAt: t0}
|
||||
var eerr error
|
||||
doc, eerr = s.extractPDF(r.Context(), c, res)
|
||||
if eerr != nil {
|
||||
s.writeFetchErr(w, c, "", eerr)
|
||||
return
|
||||
}
|
||||
s.storePDF(key, res, doc)
|
||||
}
|
||||
fetchMS := time.Since(t0).Milliseconds()
|
||||
|
||||
rp, err := s.renderPDF(doc, pdfdoc.Options{Pages: pages, Grep: grep, OutCap: s.OutCap})
|
||||
if err != nil {
|
||||
if errors.Is(err, pdfdoc.ErrBadPages) || errors.Is(err, pdfdoc.ErrBadGrep) {
|
||||
s.writeFetchErr(w, c, "", &fetch.Error{Status: 400, Msg: err.Error()})
|
||||
return
|
||||
}
|
||||
s.writeFetchErr(w, c, "", err)
|
||||
return
|
||||
}
|
||||
full := pdfdoc.Full(doc)
|
||||
docLen := utf8.RuneCountInString(full)
|
||||
content := rp.Content
|
||||
if mode == "digest" {
|
||||
content = full
|
||||
}
|
||||
|
||||
warn := []string{}
|
||||
var digMeta map[string]any
|
||||
var digestMS int64
|
||||
if mode == "digest" {
|
||||
n := len(digest.Split(full, digest.ChunkTokens))
|
||||
limit := s.PDFMaxQuestionChunks
|
||||
if limit == 0 {
|
||||
limit = DefaultPDFMaxQuestionChunks
|
||||
}
|
||||
if n > limit {
|
||||
s.writeFetchErr(w, c, "", &fetch.Error{Status: 422, Msg: fmt.Sprintf("document too long for question (%d chunks); use grep or pages", n)})
|
||||
return
|
||||
}
|
||||
if s.Digest == nil {
|
||||
warn = append(warn, "digest fell back to clean: unset")
|
||||
mode = "clean"
|
||||
content, warn = capDigestFallback(content, warn)
|
||||
digMeta = map[string]any{"fallback": "unset"}
|
||||
} else {
|
||||
dr := s.Digest.Run(r.Context(), content, question, res.FinalURL, s.callerClass(c))
|
||||
digestMS = dr.DigestMS
|
||||
if !dr.Cached {
|
||||
s.Prom.ObserveDigest(time.Duration(dr.DigestMS) * time.Millisecond)
|
||||
}
|
||||
if dr.Fallback != "" {
|
||||
warn = append(warn, "digest fell back to clean: "+dr.Fallback)
|
||||
mode = "clean"
|
||||
content, warn = capDigestFallback(content, warn)
|
||||
digMeta = map[string]any{"fallback": dr.Fallback, "model": dr.Model}
|
||||
} else {
|
||||
content = dr.Text
|
||||
digMeta = map[string]any{
|
||||
"model": dr.Model,
|
||||
"chunks_used": dr.ChunksUsed,
|
||||
"chunks_total": dr.ChunksTotal,
|
||||
"retrieval": dr.Retrieval,
|
||||
"embedded": dr.Embedded,
|
||||
"fallback": false,
|
||||
"cached": dr.Cached,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pdfMap := map[string]any{
|
||||
"pages": doc.Pages,
|
||||
"title": doc.Title,
|
||||
"outline": doc.Outline,
|
||||
"extracted": len(doc.Text),
|
||||
"first": rp.First,
|
||||
"last": rp.Last,
|
||||
"map": rp.Map,
|
||||
"grep_hits": rp.GrepHits,
|
||||
"note": rp.Note,
|
||||
}
|
||||
|
||||
s.Log.Info("extract", "caller", c.Name, "bytes", len(body), "pdf_pages", doc.Pages, "mode", mode, "cached", cached)
|
||||
|
||||
src := map[string]any{
|
||||
"kind": "upload",
|
||||
"hash": hash,
|
||||
"bytes": len(body),
|
||||
"cached": cached,
|
||||
"fetched_at": res.FetchedAt.Format(time.RFC3339),
|
||||
}
|
||||
out := map[string]any{
|
||||
"content": content,
|
||||
"mode": mode,
|
||||
"untrusted": true,
|
||||
"source": src,
|
||||
"warnings": warn,
|
||||
"sizes": map[string]int{
|
||||
"raw_tokens": docLen / 4,
|
||||
"clean_tokens": digest.EstTokens(full),
|
||||
"returned_tokens": digest.EstTokens(content),
|
||||
},
|
||||
"timings": map[string]int64{"fetch_ms": fetchMS, "digest_ms": digestMS},
|
||||
}
|
||||
if pdfMap != nil {
|
||||
out["pdf"] = pdfMap
|
||||
}
|
||||
if digMeta != nil {
|
||||
out["digest"] = digMeta
|
||||
}
|
||||
writeJSON(w, 200, out)
|
||||
}
|
||||
|
||||
// maxUploads is how many uploads may be in memory at once, and uploadWait how long a further
|
||||
// one waits for a slot before a 429: at 100 MiB each, a few are all a 1 GiB service can hold
|
||||
// beside the extraction children.
|
||||
const (
|
||||
maxUploads = 2
|
||||
uploadWait = 5 * time.Second
|
||||
)
|
||||
|
||||
// pdfMaxUpload is the /v1/extract upload cap: the configured value, or the default when unset.
|
||||
func (s *Server) pdfMaxUpload() int64 {
|
||||
if s.PDFMaxUpload > 0 {
|
||||
return s.PDFMaxUpload
|
||||
}
|
||||
return DefaultPDFMaxUpload
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
package api
|
||||
|
||||
// P5c owner review: regression tests for /v1/extract.
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Uploads are buffered whole (up to 100 MiB) before the PDF pool sees them, so the number read
|
||||
// at once is bounded: a third concurrent upload waits, then is refused busy.
|
||||
func TestExtractUploadsBounded(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
for i := 0; i < cap(r.s.uploads); i++ {
|
||||
r.s.uploads <- struct{}{}
|
||||
}
|
||||
start := time.Now()
|
||||
code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf")
|
||||
if code != http.StatusTooManyRequests {
|
||||
t.Fatalf("upload with every slot taken: %d %v, want 429", code, out)
|
||||
}
|
||||
if d := time.Since(start); d < uploadWait-100*time.Millisecond {
|
||||
t.Fatalf("refused after %v, want a wait of about %v first", d, uploadWait)
|
||||
}
|
||||
<-r.s.uploads
|
||||
if code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf"); code != 200 {
|
||||
t.Fatalf("upload with a free slot: %d %v", code, out)
|
||||
}
|
||||
}
|
||||
|
||||
// The envelope's warnings are a list, never null, as in /v1/fetch.
|
||||
func TestExtractWarningsIsAList(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf")
|
||||
if code != 200 {
|
||||
t.Fatalf("%d %v", code, out)
|
||||
}
|
||||
if _, ok := out["warnings"].([]any); !ok {
|
||||
t.Fatalf("warnings = %#v, want a list", out["warnings"])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
package api
|
||||
|
||||
// P5c task 01: POST /v1/extract takes a PDF's bytes (a file already on the caller's host), and
|
||||
// answers exactly like /v1/fetch does for a PDF URL. The cache key is the content hash.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func (r *pdfRig) extract(t *testing.T, key, query string, body []byte, ct string) (int, map[string]any) {
|
||||
t.Helper()
|
||||
req := httptest.NewRequest(http.MethodPost, "/v1/extract"+query, bytes.NewReader(body))
|
||||
if ct != "" {
|
||||
req.Header.Set("Content-Type", ct)
|
||||
}
|
||||
if key != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+key)
|
||||
}
|
||||
rec := httptest.NewRecorder()
|
||||
r.s.ServeHTTP(rec, req)
|
||||
var out map[string]any
|
||||
_ = json.Unmarshal(rec.Body.Bytes(), &out)
|
||||
return rec.Code, out
|
||||
}
|
||||
|
||||
func TestExtractUpload(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
pdf := threePDF()
|
||||
code, out := r.extract(t, r.fleet, "", pdf, "application/pdf")
|
||||
if code != 200 {
|
||||
t.Fatalf("%d %v", code, out)
|
||||
}
|
||||
content, _ := out["content"].(string)
|
||||
if !strings.Contains(content, "--- page 2 ---") || !strings.Contains(content, "quick brown fox") {
|
||||
t.Errorf("content %q", content)
|
||||
}
|
||||
if p := pdfInfo(t, out); p["pages"] != 3.0 || p["title"] != "Three Pages" {
|
||||
t.Errorf("pdf %v", p)
|
||||
}
|
||||
src, _ := out["source"].(map[string]any)
|
||||
if src["kind"] != "upload" || src["bytes"] != float64(len(pdf)) || !strings.HasPrefix(fmt.Sprint(src["hash"]), "sha256:") || src["cached"] != false {
|
||||
t.Errorf("source %v", src)
|
||||
}
|
||||
if _, hasURL := src["url"]; hasURL {
|
||||
t.Error("an upload has no URL; source must not invent one")
|
||||
}
|
||||
// The same bytes again: from the cache, no child.
|
||||
code, out = r.extract(t, r.fleet, "?pages=2", pdf, "application/octet-stream")
|
||||
if code != 200 || r.runs() != 1 {
|
||||
t.Fatalf("second upload: %d, %d extractions; want 200 and 1", code, r.runs())
|
||||
}
|
||||
if src, _ := out["source"].(map[string]any); src["cached"] != true {
|
||||
t.Errorf("second upload source %v, want cached", src)
|
||||
}
|
||||
if c, _ := out["content"].(string); strings.Contains(c, "--- page 1 ---") || !strings.Contains(c, "--- page 2 ---") {
|
||||
t.Errorf("pages=2: %q", c)
|
||||
}
|
||||
code, out = r.extract(t, r.fleet, "?grep=quick+brown", pdf, "application/pdf")
|
||||
if c, _ := out["content"].(string); code != 200 || !strings.Contains(c, "p. 2: The quick brown fox") {
|
||||
t.Errorf("grep: %d %q", code, c)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRefusals(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
if code, _ := r.extract(t, "", "", threePDF(), "application/pdf"); code != 401 {
|
||||
t.Errorf("no key: %d, want 401", code)
|
||||
}
|
||||
if code, out := r.extract(t, r.fleet, "", []byte("<html>not a pdf</html>"), "text/html"); code != 415 {
|
||||
t.Errorf("html body: %d %v, want 415", code, out)
|
||||
}
|
||||
if code, _ := r.extract(t, r.fleet, "", nil, "application/pdf"); code != 400 {
|
||||
t.Errorf("empty body: %d, want 400", code)
|
||||
}
|
||||
if code, _ := r.extract(t, r.fleet, "?pages=0-1", threePDF(), "application/pdf"); code != 400 {
|
||||
t.Errorf("bad pages: %d, want 400", code)
|
||||
}
|
||||
r.s.PDFMaxUpload = 1000
|
||||
if code, out := r.extract(t, r.fleet, "", threePDF(), "application/pdf"); code != 413 {
|
||||
t.Errorf("over PDFMaxUpload: %d %v, want 413", code, out)
|
||||
}
|
||||
if DefaultPDFMaxUpload != 100<<20 {
|
||||
t.Errorf("DefaultPDFMaxUpload = %d, want 100 MiB", DefaultPDFMaxUpload)
|
||||
}
|
||||
// The ordinary JSON endpoints keep their small request limit.
|
||||
big := bytes.Repeat([]byte("x"), 128<<10)
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodPost, "/v1/fetch", bytes.NewReader(append([]byte(`{"url":"`), append(big, []byte(`"}`)...)...)))
|
||||
req.Header.Set("Authorization", "Bearer "+r.fleet)
|
||||
r.s.ServeHTTP(rec, req)
|
||||
if rec.Code == 200 {
|
||||
t.Error("a 128 KiB /v1/fetch body was accepted; the JSON limit must still apply there")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRestrictedPartition(t *testing.T) {
|
||||
r := newPDFRig(t)
|
||||
pdf := threePDF()
|
||||
if code, _ := r.extract(t, r.gs, "", pdf, "application/pdf"); code != 200 {
|
||||
t.Fatalf("restricted upload: %d", code)
|
||||
}
|
||||
r.logs.mu.Lock()
|
||||
for _, rec := range r.logs.recs {
|
||||
var b strings.Builder
|
||||
b.WriteString(rec.Message)
|
||||
rec.Attrs(func(a slog.Attr) bool { fmt.Fprintf(&b, " %s=%v", a.Key, a.Value); return true })
|
||||
if strings.Contains(b.String(), "Three Pages") || strings.Contains(b.String(), "quick brown") {
|
||||
t.Errorf("an upload's log line carries document text: %q", b.String())
|
||||
}
|
||||
}
|
||||
r.logs.mu.Unlock()
|
||||
if code, _ := r.extract(t, r.fleet, "", pdf, "application/pdf"); code != 200 || r.runs() != 2 {
|
||||
t.Errorf("fleet after restricted: %d, %d extractions; want 200 and 2 (separate partitions)", code, r.runs())
|
||||
}
|
||||
}
|
||||
@@ -60,6 +60,9 @@ for blocked pages), `--json` (full envelope: content, source, context, sizes, wa
|
||||
PDFs work: `sift fetch` extracts them and returns the text. A short one comes back whole; a
|
||||
long one comes back as a map — an outline with page numbers, then the first pages. Then:
|
||||
|
||||
- A PDF on disk (an attachment, a download): `sift extract <file.pdf>` uploads it to `/v1/extract`
|
||||
and prints the answer as `sift fetch` does, with the same `--pages`/`--grep`/`--question`.
|
||||
Never run `pdftotext` yourself.
|
||||
- `sift fetch URL --pages 41-60` for a page range, `--grep 'pattern'` for matching lines, or
|
||||
`--question "…"` for a digest that cites each fact as `(p. N)`; `--question` works on long
|
||||
PDFs, and the first answer may be `bm25` (lexical) — ask again later for a sharper one.
|
||||
|
||||
Reference in New Issue
Block a user