From 7c40531891c7a30b3cadad37f2681fc57fc2f114 Mon Sep 17 00:00:00 2001 From: Kyle Isom Date: Sat, 26 Sep 2026 19:37:45 -0700 Subject: [PATCH] api: POST /v1/extract answers an uploaded PDF like a fetched one (content-addressed cache) Add POST /v1/extract: auth the caller, read the PDF body, hash it, and answer it on the PDF fetch path exactly as /v1/fetch answers the same PDF by URL. The hash is the content-addressed cache key ("upload:"+hash under the caller class), so identical content is extracted once with no TTL or revalidation. The api.go diff is scoped to the rule-1 changes: the DefaultPDFMaxUpload const, the PDFMaxUpload Server field, the route, and the two ServeHTTP lines. Co-Authored-By: ornith-1.5-35b-a3b --- docs/implementer-log.md | 1 + internal/api/api.go | 14 ++- internal/api/extract.go | 201 +++++++++++++++++++++++++++++++++++ internal/api/extract_test.go | 122 +++++++++++++++++++++ 4 files changed, 336 insertions(+), 2 deletions(-) create mode 100644 internal/api/extract.go create mode 100644 internal/api/extract_test.go diff --git a/docs/implementer-log.md b/docs/implementer-log.md index f6fcb3b..c6fe6d0 100644 --- a/docs/implementer-log.md +++ b/docs/implementer-log.md @@ -15,6 +15,7 @@ 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. | ## P5a review (owner, 2026-09-26) diff --git a/internal/api/api.go b/internal/api/api.go index a9cfe46..a29338b 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -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 @@ -93,6 +98,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 +117,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) diff --git a/internal/api/extract.go b/internal/api/extract.go new file mode 100644 index 0000000..19864d3 --- /dev/null +++ b/internal/api/extract.go @@ -0,0 +1,201 @@ +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 + } + + 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 + } + + var 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) +} + +// 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 +} diff --git a/internal/api/extract_test.go b/internal/api/extract_test.go new file mode 100644 index 0000000..196166a --- /dev/null +++ b/internal/api/extract_test.go @@ -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("not a pdf"), "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()) + } +}